jens/brewpi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
e49af8a347afb9800efb08dfaf5ccef2d4de263c
brewpi/tasks/sud.py
T
jensandClaude Sonnet 4.6 e49af8a347 Add a Progress tab: one step plate per schedule step, stacked 
A glance at the Automatic tab's forecast plot couldn't show where the
brew actually stands within its own schedule - just a curve. The new
Progress tab stacks one StepPlate per step instead: an LED for its
status (off/done/active-ramping/active-holding), its resolved target
temp, masses and ramp rate, plus, for whichever step is currently
active, live actual temp and stirrer state (frozen at their last
value once that step finishes) and a forecast-based remaining-time
countdown.

The countdown needs to know where each step actually begins in the
forecast's timeline, which the server didn't track before -
SudForecastEstimator.estimate() now returns per-step start times
alongside t/theta, and SudTask threads them through send_forecast()/
_reanchor_forecast() the same way (including the same generation-guard
against the concurrent-call race) as a new 'StepStarts' field on the
'Forecast' message.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019qvu5giu7gvRCyEWzf2Vpx
2026-06-24 17:15:14 +02:00

425 lines
19 KiB
Python
Raw Blame History

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
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)
if hasattr(self.tc, 'set_model_plant_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):
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:
self.apply_plant_params(step.get('grain_mass', 0), step.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)
# 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}))
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,
'pot_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)
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})
await self.send_forecast(doc)
elif 'Load' in pair[0]:
if self.sud.load(pair[1]):
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()
self.sud.tick(self.dt)
await asyncio.sleep(self.interval)
Reference in New Issue View Git Blame Copy Permalink