brewpi/tasks/sud.py

import asyncio
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

# How close theta_ist needs to be to theta_soll_set to count as "reached".
TEMP_REACHED_TOLERANCE = 0.2


class SudTask(ATask):
	def __init__(self, sud: Sud, tc: APid, stirrer: AStirrer, pot: APlant, dt, interval, msg_handler: MsgIo):
		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
		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
		# A step may carry both 'ramp' and 'hold' (ramp to temp, then hold) -
		# 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 and ramp is not None else hold

		# A ramp step whose target is below the current temperature can
		# only be reached by passive cooling - force the heater off for
		# its whole duration (decided once, here, rather than re-checked
		# every tick) instead of relying on the controller's own FSM to
		# notice and idle out.
		# bool(...): get_theta_ist() can be a numpy.float64 (Pot's transport
		# delay line is a numpy array internally) - the '<' comparison would
		# then yield numpy.bool_, which json.dumps() can't serialize.
		cooling = bool(ramping and ramp is not None and ramp['temp'] < self.tc.get_theta_ist() - TEMP_REACHED_TOLERANCE)
		self.tc.set_cooling(cooling)

		if step is not None:
			self.apply_plant_params(step)
			if ramping and ramp is not None:
				self.tc.set_theta_soll(ramp['temp'])
				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 and ramp is not None else 'hold' if hold is not None else None,
			'Descr': step.get('descr') if step else None,
			'Temp': ramp.get('temp') if ramp 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,
			'Cooling': cooling,
		}}))

	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)

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

	async def recv(self, data):
		for pair in data.items():
			if 'Start' in pair[0]:
				self.sud.start()
			elif 'Confirm' in pair[0]:
				self.sud.confirm()
			elif 'Pause' in pair[0]:
				self.sud.pause()
			elif 'Stop' in pair[0]:
				self.sud.stop()
			elif 'Save' in pair[0]:
				await self.send({'Json': self.sud.save()})
			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]})

	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)

		asyncio.create_task(self.send({'Name': self.sud.name, 'Description': self.sud.description}))

		while True:
			if self.sud.state == SudState.RAMPING:
				# Compare against the controller's own live setpoint/measurement
				# rather than its state machine, which can still read HOLD from
				# the previous step for one tick after a new target is pushed.
				if abs(self.tc.get_theta_ist() - self.tc.get_theta_soll_set()) < TEMP_REACHED_TOLERANCE:
					self.sud.temp_reached()
			self.sud.tick(self.dt)
			await asyncio.sleep(self.interval)