brewpi/tasks/sud.py

import asyncio
import bisect
from tasks import ATask
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."""
		if self.sud.state in (SudState.IDLE, SudState.DONE):
			return
		if self.heater.connected and self.stirrer.connected:
			return
		self.sud.stop()
		asyncio.create_task(self.send({'Error': 'Heater/Stirrer disconnected - brew stopped.'}))
		asyncio.create_task(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
			# 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)