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()/ # _continue_forecast_after_confirm()) - 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. 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 user-confirmation step's zero-delay assumption sits # in the forecast timeline above, keyed by the schedule's # (absolute) step index - see SudForecastEstimator.estimate()'s # confirm_points. Consulted by _continue_forecast_after_confirm() # to know where to cut the optimistic guess loose and splice in a # freshly anchored simulation once that confirmation actually # happens for real. self.forecast_confirm_marks = {} 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) 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) - _continue_forecast_after_confirm() 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 real confirmations actually happen - see _continue_forecast_after_confirm().""" if self.forecast_estimator is None: return # 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, confirm_points = await loop.run_in_executor( None, self.forecast_estimator.estimate, doc, start_theta) self.forecast_t = t self.forecast_theta = theta self.forecast_finished = (final_state == SudState.DONE) self.forecast_confirm_marks = dict(confirm_points) 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, }}) async def _continue_forecast_after_confirm(self, confirmed_index, confirmed_target): """Corrects the optimistic, zero-delay guess send_forecast() (or a previous call to this method) made at confirmed_index's user-confirmation step, now that the confirmation has actually happened for real - see SudForecastEstimator.estimate()'s docstring for why that assumption can't just be trusted as-is. Truncates the forecast right back to that point and splices in a freshly anchored simulation of the rest of the schedule, anchored at the real elapsed time (self.sud.elapsed) - so an actual delay shows up honestly as a gap rather than as the assumed zero - and the real current temperature (more trustworthy than whatever the now-superseded guess predicted it would be by now). confirmed_target is the real controller's theta_soll_set at the moment of confirmation (captured by recv() *before* calling Sud.confirm(), since that synchronously pushes the next step's own target onto it) - used to fill the real-world wait itself (see below). No-op if confirmed_index was never part of a computed forecast in the first place (e.g. the estimator isn't configured).""" if self.forecast_estimator is None: return mark_t = self.forecast_confirm_marks.pop(confirmed_index, None) if mark_t is None: return # Drop the now-stale tail (everything beyond the confirmation # point) - both the curve itself and any confirm marks that fell # within it, since they're about to be replaced by fresh ones. cut = bisect.bisect_right(self.forecast_t, mark_t) self.forecast_t = self.forecast_t[:cut] self.forecast_theta = self.forecast_theta[:cut] self.forecast_confirm_marks = {idx: t for idx, t in self.forecast_confirm_marks.items() if t <= mark_t} schedule = self.sud.schedule index = self.sud.index if not (0 <= index < len(schedule)): self.forecast_finished = True 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() real_elapsed = self.sud.elapsed loop = asyncio.get_event_loop() t, theta, final_state, confirm_points = await loop.run_in_executor(None, self.forecast_estimator.estimate, doc, start_theta) # Bridge the gap between the confirmation point and the real # elapsed time it actually happened at - the real controller # stays enabled and actively holding through WAIT_USER, so # confirmed_target (its setpoint at the time) is what it was # actually converging toward during the wait, not whatever # (possibly still mid-ramp) value the forecast happened to log # the instant WAIT_USER tripped. if self.forecast_t and self.forecast_t[-1] < real_elapsed: self.forecast_t.append(real_elapsed) self.forecast_theta.append(confirmed_target) self.forecast_t.extend(real_elapsed + seconds for seconds in t) self.forecast_theta.extend(theta) # confirm_points' 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. self.forecast_confirm_marks.update( (index + local_index, real_elapsed + local_t) for local_index, local_t in confirm_points) self.forecast_finished = (final_state == SudState.DONE) 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]: confirmed_index = self.sud.index confirmed_target = self.tc.get_theta_soll_set() self.sud.confirm() asyncio.create_task(self._continue_forecast_after_confirm(confirmed_index, confirmed_target)) 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)