jensandClaude Sonnet 5
6d1429daca
Both clients' status line only showed pot mass/water/energy once a Sud schedule was loaded and running, leaving it blank/minimal at connect time and while idle. Show it as early as possible instead, falling back to config.json's Pot section (mass/water_mass/volumen) when no schedule is loaded yet; step description/remaining-time still only appear once a schedule is actually active. web/app.js: updateStatusLine() now mirrors updatePotVisualization()'s existing sudEmpty ? potConfig... : sud... pattern, and adds a Volume figure that was previously only used for the pot-fill visualization. client/brewpi_gui.py: same treatment, plus new sud_volumen tracking (the desktop GUI had no volume concept at all before - added by reading the doc's own pot.volumen with a config fallback, since Sud._parse_data() doesn't return it). The System message's 'Pot' key now also refreshes the status label immediately so it doesn't wait for some unrelated message to trigger a repaint. tasks/sud.py: energy was only accumulated outside IDLE/DONE, so a manually-driven heater with no Sud loaded/started never counted toward the total shown in either client. Now counts in every state except DONE - IDLE banks into a stable index -1 that's never shown per-step (Progress tab only covers the schedule's own indices) but is included in the status line's summed total. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M2ierBoxW3v7nUbDw3M2pE
670 lines
32 KiB
Python
670 lines
32 KiB
Python
import asyncio
|
|
import bisect
|
|
from tasks import ATask, fire_and_forget
|
|
from ws.message import MsgIo
|
|
from utils.value import ChangedFloat
|
|
from components import APid, AStirrer, AHeater
|
|
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, heater: AHeater, pot: APlant, dt, interval, msg_handler: MsgIo,
|
|
forecast_estimator=None):
|
|
ATask.__init__(self, interval)
|
|
self.sud = sud
|
|
self.tc = tc
|
|
self.stirrer = stirrer
|
|
self.heater = heater
|
|
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)
|
|
# Saved on pause so resume can restore the effective setpoint even
|
|
# when the current step has temperature=None (inherits from a prior step).
|
|
self._paused_temp_soll = None
|
|
self._on_end = None
|
|
self._on_start = None
|
|
self._on_plant_params = None
|
|
self._on_reanchor = None
|
|
msg_handler.set_recv_handler(self.recv)
|
|
|
|
def set_on_plant_params(self, callback):
|
|
"""Register a callback invoked with (params) whenever
|
|
apply_plant_params() fires, so external observers (e.g.
|
|
SudLogTask/ServerLogTask) can record each change. Deliberately
|
|
doesn't pass elapsed - Sud.elapsed is meaningful for a SudLogTask
|
|
(its own Samples' 't' timeline) but not a ServerLogTask (a
|
|
separate timeline since server startup); each logger stamps its
|
|
own _elapsed() itself (see ServerLogTask.log_plant_params())."""
|
|
self._on_plant_params = callback
|
|
|
|
def set_on_reanchor(self, callback):
|
|
"""Register a callback invoked with (elapsed, index, theta_ist)
|
|
every time _reanchor_forecast() fires (see on_step_changed()) -
|
|
lets an external observer (SudLogTask) record the exact anchor
|
|
points a live client's forecast got corrected at, so an offline
|
|
re-simulation (utils/analyze_log.py) can reproduce the same
|
|
splice-per-transition forecast instead of just the single,
|
|
uncorrected cold-start one."""
|
|
self._on_reanchor = callback
|
|
|
|
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)
|
|
if self._on_plant_params:
|
|
self._on_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.
|
|
if self._on_reanchor:
|
|
# Read synchronously, right here - the same values
|
|
# _reanchor_forecast() itself will read moments later, off
|
|
# the same unyielded call stack, before anything else can
|
|
# mutate them.
|
|
self._on_reanchor(self.sud.elapsed, self.sud.index, self.tc.get_theta_ist())
|
|
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 set_on_end(self, callback):
|
|
self._on_end = callback
|
|
|
|
def set_on_start(self, callback):
|
|
self._on_start = callback
|
|
|
|
def check_connections(self):
|
|
"""Called whenever the heater's or stirrer's connection state
|
|
changes (wired up in server/brewpi.py via HeaterTask/StirrerTask's
|
|
set_on_connected_changed()) - force-stops a run already in progress
|
|
if either has dropped, since continuing to run a schedule with a
|
|
dead actuator is misleading. Reuses Sud.stop() (already a no-op
|
|
outside RAMPING/HOLDING/WAIT_USER/PAUSED), the same path a manual
|
|
Stop press takes, so all the usual shutdown plumbing (heater
|
|
shutdown, sud log stop_run - see SudTask.set_on_end()'s wiring in
|
|
server/brewpi.py) fires exactly as it would for Stop.
|
|
|
|
No-op if there's no running event loop - happens when a
|
|
Connectable's connected attribute changes as a side effect of the
|
|
server's own final best-effort hardware-release cleanup
|
|
(server/brewpi.py's `finally:` block), which runs after the loop
|
|
has already stopped. self.sud.stop() below cascades into Sud's own
|
|
'state' observable (on_state_changed -> heater_task.shutdown()/
|
|
sud_log_task.stop_run()), which isn't itself guarded against a
|
|
stopped loop - the whole server is already tearing down at that
|
|
point regardless, so there's nothing useful to stop/notify."""
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
return
|
|
if self.sud.state in (SudState.IDLE, SudState.DONE):
|
|
return
|
|
if self.heater.connected and self.stirrer.connected:
|
|
return
|
|
self.sud.stop()
|
|
fire_and_forget(self.send({'Error': 'Heater/Stirrer disconnected - brew stopped.'}))
|
|
fire_and_forget(self.send({'Error': 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)
|
|
if self._on_end:
|
|
self._on_end()
|
|
|
|
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
|
|
schedule = self.sud.schedule
|
|
index = self.sud.index
|
|
# Drop the now-stale tail (everything beyond right now) - it's
|
|
# about to be replaced by a freshly anchored simulation. Cut at
|
|
# forecast_step_starts[index], the old forecast's own (possibly
|
|
# very wrong) belief of where step `index` begins - not at
|
|
# real_elapsed directly: a step whose real timing blew way past
|
|
# what its zero-delay guess assumed (a long WAIT_USER confirm,
|
|
# above all - see SudForecastEstimator.estimate()'s docstring)
|
|
# leaves the old forecast's *entire* speculative remainder sitting
|
|
# at timestamps still numerically less than real_elapsed, so
|
|
# bisecting against real_elapsed itself would find nothing to
|
|
# discard and just tack the fresh, correct simulation on after it
|
|
# - a doubled-back, self-overlapping curve instead of a clean cut.
|
|
# forecast_step_starts[index] doesn't have this problem: it lives
|
|
# in the same (speculative) coordinate space as forecast_t itself,
|
|
# so the cut lands in the right place regardless of how far real
|
|
# and speculated time have diverged by now.
|
|
cut = bisect.bisect_right(self.forecast_t, self.forecast_step_starts.get(index, 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.
|
|
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]:
|
|
# Refuse to (re)start a brew if either actuator it depends
|
|
# on isn't actually connected - continuing to "run" a
|
|
# schedule with a dead heater/stirrer is misleading. Mirrors
|
|
# the 'Load'-while-running refusal below: send the Error then
|
|
# immediately clear it, since the dispatcher has no concept of
|
|
# a one-shot event (see that branch's own comment).
|
|
missing = [name for name, device in (('heater', self.heater), ('stirrer', self.stirrer)) if not device.connected]
|
|
if missing:
|
|
await self.send({'Error': f"Cannot start - {' and '.join(missing)} not connected."})
|
|
await self.send({'Error': None})
|
|
continue
|
|
# 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 self._on_start:
|
|
self._on_start()
|
|
if fresh_start:
|
|
asyncio.create_task(self.send_forecast(self.sud.save()))
|
|
else:
|
|
# Resume from PAUSED: re-apply current step's schedule
|
|
# parameters so any manual TC/stirrer changes made while
|
|
# paused are overwritten by the schedule on play.
|
|
index = self.sud.index
|
|
if index is not None and 0 <= index < len(self.sud.schedule):
|
|
step = self.sud.schedule[index]
|
|
ramp = step.get('ramp')
|
|
hold = step.get('hold')
|
|
ramping = self.sud.state == SudState.RAMPING
|
|
# Use step's own temperature if set; fall back to
|
|
# the value saved at pause for steps that inherit
|
|
# their target from a prior step (temperature=None).
|
|
target_temp = step.get('temperature')
|
|
if target_temp is None:
|
|
target_temp = self._paused_temp_soll
|
|
if target_temp is not None:
|
|
self.tc.set_theta_soll(target_temp)
|
|
if ramping and ramp:
|
|
self.tc.set_heatrate_soll(ramp['rate'])
|
|
phase = ramp if ramping else hold
|
|
if phase is not None:
|
|
self.apply_stirrer(phase)
|
|
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._paused_temp_soll = self.tc.theta_soll_set
|
|
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
|
|
# state except DONE, IDLE included - a manually-driven heater
|
|
# (no schedule loaded/started yet, e.g. pre-heating strike
|
|
# water by hand) still draws real power and should show up in
|
|
# the same total; self.sud.index is a stable -1 while IDLE
|
|
# (see components/sud.py's _reset_run_state()), so this banks
|
|
# into energy_by_step[-1] same as any other step once a real
|
|
# Start moves the index on - never shown per-step (Progress
|
|
# tab's step-plates only cover the schedule's own indices) but
|
|
# still included in the status line's summed total.
|
|
if self.sud.state != 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)
|