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


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
		lumped (M, C) in sync with the step's grain_mass/water_mass, since
		those vary over the course of a brew (malt going in, water boiling
		off) - mirrors demo_sud.py's apply_plant_params()."""
		params = self.sud.derive_plant_params(step.get('grain_mass', 0), step.get('water_mass', 0))
		self.pot.set_thermal_params(params['M'], params['C'])
		if hasattr(self.tc, 'set_model_params'):
			self.tc.set_model_params(params['M'], params['C'])

	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]})
					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)