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 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, step): """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 step's 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()) and once immediately on Load (see recv()), 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(step.get('grain_mass', 0), step.get('water_mass', 0)) 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) 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. 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() t, theta, final_state, confirm_points = await loop.run_in_executor(None, self.forecast_estimator.estimate, doc) 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): await self.send({'Forecast': { 'T': self.forecast_t, 'Theta': self.forecast_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, '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]: self.sud.start() 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 (and its first step's expected grain/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.schedule[0]) 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)