jensandClaude Sonnet 4.6
e2e3baab08
Top-level pot_mass/pot_material/L/Td/grain_mass/water_mass and their per-step overrides are now grouped under a 'pot' sub-object, matching the structure already applied to sud_0010.json's parent level. Updated all sude/*.json files and the parsing/consuming code in components/sud.py, tasks/sud.py, components/sud_forecast.py, and scripts/demos/sud/demo_sud.py. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T52JH848ojhXXHn1bAzdC3
546 lines
26 KiB
Python
546 lines
26 KiB
Python
import asyncio
|
|
import bisect
|
|
from tasks import ATask
|
|
from ws.message import MsgIo
|
|
from utils.value import ChangedFloat
|
|
from components import APid, AStirrer
|
|
from components.plant import APlant
|
|
from components.sud import Sud, SudState
|
|
|
|
# Upper bound on how many (t, theta) points the forecast is thinned to
|
|
# before going out over the wire (see _send_forecast()) - a fine enough
|
|
# dt over a multi-hour brew can otherwise produce a single JSON message
|
|
# of several MB, large enough to exceed the websockets library's default
|
|
# 1 MiB max_size and get the connection closed outright (code 1009). The
|
|
# GUI's forecast plot is a few hundred pixels wide, so this many points
|
|
# is already far more resolution than it can show; self.forecast_t/
|
|
# forecast_theta themselves stay at full simulated resolution - this
|
|
# only thins the copy actually sent to clients.
|
|
MAX_FORECAST_POINTS = 1000
|
|
|
|
|
|
def _downsample(t, theta, max_points=MAX_FORECAST_POINTS):
|
|
"""Returns (t, theta) thinned to at most max_points entries by simple
|
|
decimation, always keeping the first and last point - losing a few
|
|
intermediate samples doesn't matter for a plot this size, but losing
|
|
the endpoints would visibly truncate the curve or its final value."""
|
|
n = len(t)
|
|
if n <= max_points:
|
|
return t, theta
|
|
step = -(-n // max_points) # ceil(n / max_points)
|
|
t_ds = t[::step]
|
|
theta_ds = theta[::step]
|
|
if t_ds[-1] != t[-1]:
|
|
t_ds = t_ds + [t[-1]]
|
|
theta_ds = theta_ds + [theta[-1]]
|
|
return t_ds, theta_ds
|
|
|
|
|
|
class SudTask(ATask):
|
|
def __init__(self, sud: Sud, tc: APid, stirrer: AStirrer, pot: APlant, dt, interval, msg_handler: MsgIo,
|
|
forecast_estimator=None):
|
|
ATask.__init__(self, interval)
|
|
self.sud = sud
|
|
self.tc = tc
|
|
self.stirrer = stirrer
|
|
self.pot = pot
|
|
# Simulated seconds per tick, vs. interval's wall-clock seconds per
|
|
# tick - same dt/interval split Pot/TempController/Stirrer already
|
|
# use internally. Their warp-induced speedup falls out naturally
|
|
# from ticking real physics at simulated dt; Sud has no physics of
|
|
# its own, so hold_remaining must be ticked by dt explicitly to get
|
|
# the same speedup instead of running in real time.
|
|
self.dt = dt
|
|
self.msg_handler = msg_handler
|
|
# Predicts a schedule's actual duration by simulating it with the
|
|
# same plant/controller machinery this server uses for real - see
|
|
# components/sud_forecast.py. Optional only so tests/demos that
|
|
# build a SudTask without one still work; the server always passes
|
|
# one.
|
|
self.forecast_estimator = forecast_estimator
|
|
# The forecast as last computed/corrected (see send_forecast()/
|
|
# _reanchor_forecast()) - T in simulated seconds, Theta in degrees,
|
|
# parallel lists, both monotonically growing as corrections get
|
|
# spliced in. Covers the whole schedule from the very first Load,
|
|
# including through steps requiring user confirmation - those are
|
|
# simulated as a zero-delay auto-confirm (see components/
|
|
# sud_forecast.py's SudForecastEstimator.estimate()) rather than
|
|
# left unforecast, until _reanchor_forecast() corrects it for real.
|
|
self.forecast_t = []
|
|
self.forecast_theta = []
|
|
# Whether the forecast above actually reaches the schedule's real
|
|
# end (sent as 'Finished') - False only in the pathological case
|
|
# of a step whose target can never be reached (see
|
|
# components/sud_forecast.py's MAX_TICKS).
|
|
self.forecast_finished = True
|
|
# Where each schedule step begins, in the same absolute timeline as
|
|
# forecast_t - keyed by the schedule's own (absolute) step index,
|
|
# rebased the same way as forecast_t/forecast_theta themselves on
|
|
# every send_forecast()/_reanchor_forecast() call (see either's own
|
|
# comment). Lets a client (the GUI's Progress tab) show each step's
|
|
# predicted total/remaining duration without re-deriving it itself.
|
|
self.forecast_step_starts = {}
|
|
# Bumped by every call to send_forecast()/_reanchor_forecast() -
|
|
# see either's own comment for why: it lets a call that's still
|
|
# awaiting its worker-thread simulation tell, once it resumes,
|
|
# whether a newer call has since started and already committed a
|
|
# fresher result - if so, it discards its own rather than
|
|
# corrupting forecast_t/forecast_theta with stale data.
|
|
self._forecast_generation = 0
|
|
# Energy consumption per step (Wh), integrated from the heater's
|
|
# own live effective power - see pot.get_power() (set from
|
|
# heater.power_eff - server/brewpi.py wires that up) - rather
|
|
# than recomputed from the forecast like the timing fields above,
|
|
# since energy actually used can't be predicted in advance, only
|
|
# measured as it happens. energy_step_accum_j is the *current*
|
|
# step's running total, in Joules (integrated every tick in
|
|
# on_process() - converted to Wh only at the message boundary,
|
|
# see _on_energy_changed()); energy_by_step holds each *finished*
|
|
# step's final Wh total, keyed by its (absolute) schedule index -
|
|
# both reset on every Load (see recv()) since neither make sense
|
|
# carried over to a different schedule. _energy_index is simply
|
|
# which index the running accumulator currently belongs to, so
|
|
# on_step_changed() can tell a genuine new step (a different
|
|
# index) from its own ramp->hold phase switch (same index, must
|
|
# not reset the accumulator mid-step).
|
|
self.energy_step_accum_j = 0.0
|
|
self.energy_by_step = {}
|
|
self._energy_index = None
|
|
self._energy_changed = ChangedFloat(self._on_energy_changed, prec=2)
|
|
msg_handler.set_recv_handler(self.recv)
|
|
|
|
def apply_plant_params(self, grain_mass, water_mass):
|
|
"""Keeps the real plant's and the controller's internal model's
|
|
plant params in sync with this Sud's own doc - L/Td come straight
|
|
from it (constant for the whole brew), while M/C also fold in the
|
|
given grain_mass/water_mass, which vary over the brew's course
|
|
(malt going in, water boiling off) - mirrors demo_sud.py's
|
|
apply_plant_params(). Called both on every real step transition
|
|
(via on_step_changed(), with that step's own grain_mass/water_mass)
|
|
and once immediately on Load (see recv(), with the doc's own
|
|
initial Sud.grain_mass/water_mass - no need to parse the first
|
|
step out of the schedule for that), so the controller's behavior
|
|
already matches the expected plant as soon as a Sud is loaded,
|
|
not just once a run actually starts."""
|
|
params = self.sud.derive_plant_params(grain_mass, water_mass)
|
|
self.pot.set_plant_params(params)
|
|
self.tc.set_model_plant_params(params)
|
|
|
|
def apply_stirrer(self, phase):
|
|
stirrer_cfg = phase.get('stirrer', {})
|
|
speed = stirrer_cfg.get('speed', 0)
|
|
interval_time = stirrer_cfg.get('interval_time', 0)
|
|
on_ratio = stirrer_cfg.get('on_ratio', 1.0)
|
|
if interval_time > 0:
|
|
self.stirrer.set_cycle_time(interval_time)
|
|
self.stirrer.set_duty_cycle(on_ratio)
|
|
else:
|
|
self.stirrer.set_cycle_time(1.0)
|
|
self.stirrer.set_duty_cycle(1.0 if speed > 0 else 0.0)
|
|
self.stirrer.set_speed(speed)
|
|
|
|
def on_step_changed(self, step):
|
|
# A genuinely new step (this index differs from whichever one the
|
|
# running accumulator currently belongs to - including the
|
|
# transition to DONE, step=None, self.sud.index past the last
|
|
# real one) banks the just-finished step's total and starts a
|
|
# fresh one; the ramp->hold phase switch within the *same* step
|
|
# re-fires this callback too (see below) but must not reset
|
|
# mid-step, hence comparing the index itself rather than reacting
|
|
# to every call.
|
|
if self.sud.index != self._energy_index:
|
|
if self._energy_index is not None:
|
|
self.energy_by_step[self._energy_index] = self.energy_step_accum_j / 3600.0
|
|
self.energy_step_accum_j = 0.0
|
|
self._energy_index = self.sud.index
|
|
ramp = step.get('ramp') if step else None
|
|
hold = step.get('hold') if step else None
|
|
# Every step ramps to 'temperature' first, then optionally holds -
|
|
# self.sud.state tells us which phase is currently active; it's
|
|
# already up to date by the time this callback fires, both on a
|
|
# full step transition and on the ramp->hold phase switch within
|
|
# one step (components/sud.py's temp_reached()).
|
|
ramping = self.sud.state == SudState.RAMPING
|
|
phase = ramp if ramping else hold
|
|
|
|
if step is not None:
|
|
pot = step.get('pot', {})
|
|
self.apply_plant_params(pot.get('grain_mass', 0), pot.get('water_mass', 0))
|
|
if ramping and step['temperature'] is not None:
|
|
self.tc.set_theta_soll(step['temperature'])
|
|
self.tc.set_heatrate_soll(ramp['rate'])
|
|
self.apply_stirrer(phase)
|
|
# Records the moment this step actually began *synchronously* -
|
|
# only on the genuine first entry (ramping, per Sud._advance()
|
|
# unconditionally setting state to RAMPING first - see this
|
|
# method's own comment above), not the same step's later ramp->
|
|
# hold switch, so this can't itself get overwritten by that
|
|
# switch's slightly later timestamp. Unconditional, not
|
|
# setdefault: an *earlier* reanchor's own forward-looking
|
|
# simulation may already have written a prediction for this
|
|
# same index (it simulates every remaining step from its own
|
|
# anchor point, real ones included) - that guess must be
|
|
# replaced now that the real thing has actually happened,
|
|
# never left to linger as if it still were one.
|
|
#
|
|
# _reanchor_forecast() recomputes this same entry too, but
|
|
# asynchronously, so it can be (and routinely is, e.g. on the
|
|
# very next step boundary arriving before its own simulation
|
|
# finishes) superseded and discarded before ever committing -
|
|
# see its own comment. Without this synchronous copy, a later
|
|
# reanchor's "everything before my own index is real and
|
|
# immutable" filter would then preserve whatever *that*
|
|
# discarded call's predecessor had left behind instead - a
|
|
# stale prediction, sometimes even one that hasn't happened
|
|
# yet by the schedule's real current position.
|
|
if ramping:
|
|
self.forecast_step_starts[self.sud.index] = self.sud.elapsed
|
|
# Every real step boundary (full step change or ramp->hold
|
|
# within one) is a trustworthy checkpoint to correct the
|
|
# forecast against - see _reanchor_forecast(). Catches drift
|
|
# from anything the original simulation couldn't have known
|
|
# (a malt fill-in's actual cooldown, a longer/shorter ramp than
|
|
# modeled, ...) at the next opportunity, not just at the next
|
|
# user confirmation.
|
|
asyncio.create_task(self._reanchor_forecast())
|
|
|
|
asyncio.create_task(self.send({'Step': {
|
|
'Index': self.sud.index,
|
|
'Type': 'ramp' if ramping else 'hold' if hold is not None else None,
|
|
'Descr': step.get('descr') if step else None,
|
|
'Temp': step.get('temperature') if step else None,
|
|
'Rate': ramp.get('rate') if ramp else None,
|
|
'Duration': hold.get('duration') if (hold is not None and not ramping) else None,
|
|
'WaitForUser': step.get('user_wait_for_continue', False) if step else None,
|
|
}}))
|
|
|
|
def on_state_changed(self, value):
|
|
asyncio.create_task(self.send({'State': str(value)}))
|
|
|
|
if value in (SudState.DONE, SudState.IDLE):
|
|
self.stirrer.set_duty_cycle(1.0)
|
|
self.stirrer.set_speed(0)
|
|
# A finished/stopped run no longer owns the controller - hand
|
|
# control back to manual mode (off by default there too).
|
|
self.tc.set_enabled(False)
|
|
else:
|
|
# Any other state (RAMPING/HOLDING/WAIT_USER/PAUSED) means a
|
|
# run is in progress and needs the controller actively driving
|
|
# the heater.
|
|
self.tc.set_enabled(True)
|
|
|
|
def on_user_message_changed(self, value):
|
|
asyncio.create_task(self.send({'UserMessage': value}))
|
|
|
|
def on_hold_remaining_changed(self, value):
|
|
asyncio.create_task(self.send({'HoldRemaining': value}))
|
|
|
|
def on_elapsed_changed(self, value):
|
|
asyncio.create_task(self.send({'Elapsed': value}))
|
|
|
|
def _on_energy_changed(self, value):
|
|
"""value is the currently active step's running total (Wh) -
|
|
throttled to once per self._energy_changed's rounding step (see
|
|
__init__), same pattern as on_hold_remaining_changed()/
|
|
on_elapsed_changed() above, just driven manually from on_process()
|
|
each tick rather than via Sud's own AttributeChange (energy isn't
|
|
one of Sud's own attributes). energy_by_step (every *finished*
|
|
step's own final Wh total) rides along on every such push rather
|
|
than only on change - it's a handful of entries at most, and
|
|
piggybacking means the client never has to reconcile two
|
|
differently-timed messages to know "the rest of the totals plus
|
|
what's happening right now"."""
|
|
asyncio.create_task(self.send({'Energy': {
|
|
'StepEnergy': sorted(self.energy_by_step.items()),
|
|
'Current': value,
|
|
}}))
|
|
|
|
async def send_forecast(self, doc):
|
|
"""Computes and sends the full forecast for doc, start to finish -
|
|
including through every step requiring user confirmation, which
|
|
is modeled as a zero-delay auto-confirm rather than left
|
|
unforecast (see components/sud_forecast.py's
|
|
SudForecastEstimator.estimate()) - so the whole schedule's
|
|
projected curve is visible right away instead of stopping at the
|
|
first one. Always a fresh start: discards whatever forecast was
|
|
accumulated for the previously loaded schedule.
|
|
|
|
Anchored at the real current temperature
|
|
(self.tc.get_theta_ist_set()), not a cold start at ambient - the
|
|
pot may already be warm (a previous run, or manual heating) at
|
|
the moment this Sud is loaded, and a forecast that assumes
|
|
ambient regardless would reach every target later than it
|
|
actually will, never lining up with the actual trace even at
|
|
t=0. Deliberately the raw sensor reading (theta_ist_set), not
|
|
the controller's own get_theta_ist() (theta_ist) -
|
|
_reanchor_forecast() can trust that one because a real run has
|
|
been actively ticking for a while by the time it runs, but this
|
|
is called right after Load, before this controller may have
|
|
processed even a single tick yet (e.g. its plant params/model
|
|
only just got configured - see SudTask.recv()), so theta_ist
|
|
itself could still be sitting at its never-updated __init__
|
|
default.
|
|
|
|
Those zero-delay assumptions get corrected piecewise as the
|
|
schedule actually reaches each step boundary - see
|
|
_reanchor_forecast()."""
|
|
if self.forecast_estimator is None:
|
|
return
|
|
# Both this and _reanchor_forecast() can end up running
|
|
# concurrently - e.g. a fresh Start triggers this explicitly *and*
|
|
# (via Sud.start() synchronously firing on_step_changed() for the
|
|
# first step) a _reanchor_forecast() of its own; a step whose hold
|
|
# duration is already 0 can likewise advance twice within a single
|
|
# tick, firing on_step_changed() twice back to back. Each such call
|
|
# awaits a worker-thread simulation, so without this guard whichever
|
|
# one resumes second would blindly splice its own tail onto
|
|
# whatever the other already finished writing, producing a
|
|
# spurious connecting line across the plot. Bumping/checking this
|
|
# generation counter across the await ensures only the very latest
|
|
# call's result is ever committed - any older one discards itself.
|
|
self._forecast_generation += 1
|
|
generation = self._forecast_generation
|
|
# Runs the simulation in a worker thread - it's CPU-bound and can
|
|
# take a couple hundred ms for a long schedule, which would
|
|
# otherwise stall every other task (heater, sensor, ...) for that
|
|
# whole window.
|
|
loop = asyncio.get_event_loop()
|
|
start_theta = self.tc.get_theta_ist_set()
|
|
t, theta, final_state, step_starts = await loop.run_in_executor(
|
|
None, self.forecast_estimator.estimate, doc, start_theta)
|
|
if generation != self._forecast_generation:
|
|
return
|
|
self.forecast_t = t
|
|
self.forecast_theta = theta
|
|
self.forecast_finished = (final_state == SudState.DONE)
|
|
self.forecast_step_starts = step_starts
|
|
await self._send_forecast()
|
|
|
|
async def _send_forecast(self):
|
|
t, theta = _downsample(self.forecast_t, self.forecast_theta)
|
|
await self.send({'Forecast': {
|
|
'T': t,
|
|
'Theta': theta,
|
|
'Finished': self.forecast_finished,
|
|
# Sorted [index, t] pairs rather than a {index: t} object - JSON
|
|
# object keys are always strings, which would force every
|
|
# consumer to int() them back; a plain sorted list sidesteps
|
|
# that and is just as easy to look up from (the GUI's Progress
|
|
# tab only ever needs it index-aligned with its own step list).
|
|
'StepStarts': sorted(self.forecast_step_starts.items()),
|
|
}})
|
|
|
|
async def _reanchor_forecast(self):
|
|
"""Corrects the optimistic, zero-delay guesses baked into the
|
|
forecast (see SudForecastEstimator.estimate()'s docstring) now
|
|
that the schedule has actually reached a real step boundary -
|
|
called from on_step_changed() on every transition, whether it's
|
|
a user confirmation or fully automatic (e.g. a ramp reaching its
|
|
target, or a hold's duration running out). Truncates the forecast
|
|
back to right now and splices in a freshly anchored simulation of
|
|
the rest of the schedule, anchored at the real elapsed time
|
|
(self.sud.elapsed) and the real current temperature
|
|
(self.tc.get_theta_ist()) - so any divergence the original
|
|
simulation couldn't have predicted (a malt fill-in's actual
|
|
cooldown, a longer/shorter ramp than modeled, ...) gets corrected
|
|
at the next opportunity instead of leaving the forecast stuck
|
|
showing what was once guessed.
|
|
|
|
No-op if the estimator isn't configured.
|
|
|
|
Guards against the same concurrent-call race send_forecast() does
|
|
(see its own comment) by working off local copies of the forecast
|
|
lists throughout, only ever committing them to self.forecast_t/
|
|
forecast_theta right at the end, and only if this is still the
|
|
latest call by then - an older call resuming after a newer one
|
|
has already committed must discard its own (now-stale) result
|
|
rather than splice it onto what the newer call already wrote."""
|
|
if self.forecast_estimator is None:
|
|
return
|
|
self._forecast_generation += 1
|
|
generation = self._forecast_generation
|
|
real_elapsed = self.sud.elapsed
|
|
# Drop the now-stale tail (everything beyond right now) - it's
|
|
# about to be replaced by a freshly anchored simulation.
|
|
cut = bisect.bisect_right(self.forecast_t, real_elapsed)
|
|
forecast_t = self.forecast_t[:cut]
|
|
forecast_theta = self.forecast_theta[:cut]
|
|
# Steps already passed (< index) have their real, now-immutable
|
|
# start time; anything from index on is about to be resimulated
|
|
# fresh below and must not keep a stale prediction around.
|
|
schedule = self.sud.schedule
|
|
index = self.sud.index
|
|
forecast_step_starts = {i: tt for i, tt in self.forecast_step_starts.items() if i < index}
|
|
|
|
if not (0 <= index < len(schedule)):
|
|
if generation != self._forecast_generation:
|
|
return
|
|
self.forecast_t = forecast_t
|
|
self.forecast_theta = forecast_theta
|
|
self.forecast_finished = True
|
|
self.forecast_step_starts = forecast_step_starts
|
|
await self._send_forecast()
|
|
return
|
|
doc = {
|
|
'Name': self.sud.name,
|
|
'Description': self.sud.description,
|
|
'pot': {
|
|
'mass': self.sud.pot_mass,
|
|
'material': self.sud.pot_material,
|
|
'L': self.sud.L,
|
|
'Td': self.sud.Td,
|
|
'grain_mass': self.sud.grain_mass,
|
|
'water_mass': self.sud.water_mass,
|
|
},
|
|
'steps': schedule[index:],
|
|
}
|
|
start_theta = self.tc.get_theta_ist()
|
|
loop = asyncio.get_event_loop()
|
|
t, theta, final_state, step_starts = await loop.run_in_executor(None, self.forecast_estimator.estimate, doc, start_theta)
|
|
if generation != self._forecast_generation:
|
|
return
|
|
# Bridge any gap between the last surviving old point and right
|
|
# now (e.g. the old forecast's timeline had already drifted
|
|
# behind real_elapsed) with the same real measurement the fresh
|
|
# simulation below starts from, so the spliced curve doesn't
|
|
# visibly jump.
|
|
if forecast_t and forecast_t[-1] < real_elapsed:
|
|
forecast_t.append(real_elapsed)
|
|
forecast_theta.append(start_theta)
|
|
forecast_t.extend(real_elapsed + seconds for seconds in t)
|
|
forecast_theta.extend(theta)
|
|
# step_starts' indices/times are relative to this sub-schedule
|
|
# (starting fresh at doc['steps'][0]) - rebase both onto the real
|
|
# schedule's absolute indices and the master forecast timeline,
|
|
# same as t/theta above.
|
|
forecast_step_starts.update(
|
|
(index + local_index, real_elapsed + local_t) for local_index, local_t in step_starts.items())
|
|
self.forecast_t = forecast_t
|
|
self.forecast_theta = forecast_theta
|
|
self.forecast_finished = (final_state == SudState.DONE)
|
|
self.forecast_step_starts = forecast_step_starts
|
|
await self._send_forecast()
|
|
|
|
async def recv(self, data):
|
|
for pair in data.items():
|
|
if 'Start' in pair[0]:
|
|
# A fresh start (not a resume from Pause, which keeps
|
|
# whatever forecast the run already established) re-
|
|
# anchors the forecast to the real temperature right now
|
|
# - that's the actual "t=0" the about-to-start actual
|
|
# trace will be plotted from, which may no longer match
|
|
# whatever temperature existed back at Load (time passed,
|
|
# possibly manual heating in between).
|
|
fresh_start = self.sud.state in (SudState.IDLE, SudState.DONE)
|
|
if fresh_start:
|
|
# Belongs to the run that just ended (Stop, or running
|
|
# the schedule through to DONE), not the one about to
|
|
# begin - a Pause->resume (fresh_start False) keeps it,
|
|
# same as the forecast above.
|
|
self.energy_step_accum_j = 0.0
|
|
self.energy_by_step = {}
|
|
self._energy_index = None
|
|
self.sud.start()
|
|
if fresh_start:
|
|
asyncio.create_task(self.send_forecast(self.sud.save()))
|
|
elif 'Confirm' in pair[0]:
|
|
# Sud.confirm() synchronously fires on_step_changed() for
|
|
# the now-current step, which schedules the forecast
|
|
# reanchor itself - see _reanchor_forecast().
|
|
self.sud.confirm()
|
|
elif 'Pause' in pair[0]:
|
|
self.sud.pause()
|
|
elif 'Stop' in pair[0]:
|
|
self.sud.stop()
|
|
elif 'Save' in pair[0]:
|
|
doc = self.sud.save()
|
|
await self.send({'Json': doc})
|
|
# A run already in progress (e.g. a client connecting mid-
|
|
# brew, which always fires this on connect - see client/
|
|
# brewpi_gui.py's Window.connect()) must NOT get send_
|
|
# forecast()'s cold, from-step-0 simulation here: it knows
|
|
# nothing of the real current step/elapsed/temperature, so
|
|
# it would silently overwrite the forecast/forecast_step_
|
|
# starts that's been accurately, continuously maintained
|
|
# by _reanchor_forecast() all along with a context-free
|
|
# "starting fresh right now" guess - the exact bug that
|
|
# made a freshly-connected client highlight/countdown the
|
|
# wrong step. Just re-send what's already there instead;
|
|
# only a genuinely not-yet-started schedule (IDLE/DONE)
|
|
# has nothing accurate yet to preserve, so it alone still
|
|
# gets the real, full computation.
|
|
if self.sud.state in (SudState.IDLE, SudState.DONE):
|
|
await self.send_forecast(doc)
|
|
else:
|
|
await self._send_forecast()
|
|
elif 'Load' in pair[0]:
|
|
if self.sud.load(pair[1]):
|
|
# Energy consumption belongs to a specific schedule's
|
|
# run, same as forecast_step_starts' timings - neither
|
|
# means anything carried over to a different one.
|
|
self.energy_step_accum_j = 0.0
|
|
self.energy_by_step = {}
|
|
self._energy_index = None
|
|
await self.send({'Name': self.sud.name, 'Description': self.sud.description})
|
|
await self.send({'Json': pair[1]})
|
|
# on_step_changed() only re-applies plant params once a
|
|
# real step starts (Start) - apply them right away too,
|
|
# so the controller already matches this Sud's own pot/
|
|
# L/Td/initial grain_mass/water_mass from the moment
|
|
# it's loaded, rather than whatever the previously
|
|
# loaded Sud (or the generic startup baseline - see
|
|
# server/brewpi.py) left behind.
|
|
if self.sud.schedule:
|
|
self.apply_plant_params(self.sud.grain_mass, self.sud.water_mass)
|
|
await self.send_forecast(pair[1])
|
|
else:
|
|
# Sud.load() refuses while a run is in progress (state
|
|
# not IDLE/DONE) - tell the client why instead of
|
|
# silently dropping the request. The client has no
|
|
# business pre-emptively guessing this itself from its
|
|
# own (replicated, laggy) view of the state.
|
|
#
|
|
# Immediately cleared back to None - unlike every other
|
|
# field here, this is a one-shot event, not state. The
|
|
# dispatcher has no concept of "don't persist this into
|
|
# global_state" (see ws/user.py's update()), so without
|
|
# clearing it, any client connecting later - even one
|
|
# that never touched Load - would get this stale error
|
|
# replayed on connect, with nothing it just did to
|
|
# explain why.
|
|
await self.send({'Error': 'Cannot load a new schedule while a run is in progress - stop it first.'})
|
|
await self.send({'Error': None})
|
|
|
|
async def send(self, data):
|
|
await self.msg_handler.send(data)
|
|
|
|
async def on_process(self):
|
|
print("{}: Started with interval {} s".format(self.msg_handler.get_key(), self.interval))
|
|
|
|
self.sud.set_on_changed('step', self.on_step_changed)
|
|
self.sud.set_on_changed('state', self.on_state_changed)
|
|
self.sud.set_on_changed('user_message', self.on_user_message_changed)
|
|
self.sud.set_on_changed('hold_remaining', ChangedFloat(self.on_hold_remaining_changed, prec=0).set)
|
|
self.sud.set_on_changed('elapsed', ChangedFloat(self.on_elapsed_changed, prec=1).set)
|
|
|
|
asyncio.create_task(self.send({'Name': self.sud.name, 'Description': self.sud.description}))
|
|
|
|
while True:
|
|
if self.sud.state == SudState.RAMPING and self.tc.is_holding():
|
|
self.sud.temp_reached()
|
|
# Energy actually drawn this tick - pot.get_power() reflects
|
|
# the heater's own live effective power (see server/brewpi.
|
|
# py's heater.set_on_changed("power_eff", ...pot.set_power)
|
|
# wiring), in Watts; dt is this tick's simulated seconds, so
|
|
# the product is Joules, accumulated for whichever step is
|
|
# current (see on_step_changed()). Counted through every
|
|
# non-idle state, WAIT_USER/PAUSED included - the controller
|
|
# stays enabled (see on_state_changed()) and may well still
|
|
# be actively holding, drawing real power, even though the
|
|
# schedule itself isn't progressing.
|
|
if self.sud.state not in (SudState.IDLE, SudState.DONE):
|
|
self.energy_step_accum_j += self.pot.get_power() * self.dt
|
|
self._energy_changed.set(self.energy_step_accum_j / 3600.0)
|
|
self.sud.tick(self.dt)
|
|
await asyncio.sleep(self.interval)
|