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
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.

		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):
		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]:
				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/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)