brewpi/tasks/heater.py

import asyncio
from tasks import ATask, fire_and_forget
from components import AHeater
from ws.message import MsgIo
from utils.value import ChangedInteger


class HeaterTask(ATask):
	def __init__(self, device: AHeater, interval, msg_handler: MsgIo):
		ATask.__init__(self, interval)

		self.device = device
		self.interval = interval
		self.msg_handler = msg_handler
		msg_handler.set_recv_handler(self.recv)
		self.power_soll = 0
		self.power_actor = 0
		self.closed_loop = True
		self._on_closed_loop_changed = None
		self.pulse_counter = 0
		self._on_connected_changed = None
		device.set_on_changed('power_eff', ChangedInteger(self.on_changed_power).set)
		device.set_on_changed('connected', self.on_connected_changed)
		self.power_set_changed = ChangedInteger(self.on_changed_power_set).set

	def set_on_closed_loop_changed(self, callback):
		self._on_closed_loop_changed = callback

	def set_on_connected_changed(self, callback):
		"""Registers a callback(connected: bool) invoked whenever the
		device's connection state changes - lets server/brewpi.py wire this
		into SudTask's start-gating/mid-brew auto-stop."""
		self._on_connected_changed = callback

	def on_connected_changed(self, value):
		fire_and_forget(self.send({'Connected': value}))
		if self._on_connected_changed:
			self._on_connected_changed(value)

	def shutdown(self):
		"""Called when the brew ends (DONE/IDLE). Switches to open-loop at
		power 0 and disables TC, mirroring what a Stop press should do."""
		self.closed_loop = False
		self.power_soll = 0
		if self._on_closed_loop_changed:
			self._on_closed_loop_changed(False)
		asyncio.create_task(self.send({'ClosedLoop': False}))
		self.power_set_changed(0)

	def start_closed_loop(self):
		"""Called when a Sud starts. Re-enables closed-loop so the TC drives
		the heater regardless of which client started the brew."""
		self.closed_loop = True
		if self._on_closed_loop_changed:
			self._on_closed_loop_changed(True)
		asyncio.create_task(self.send({'ClosedLoop': True}))

	def actor(self, y):
		self.power_actor = max(0, self.device.get_power_max() * y)

	def on_changed_power(self, power):
		asyncio.create_task(self.send({'Power': power}))

	def on_changed_power_set(self, power):
		asyncio.create_task(self.send({'PowerSet': power}))

	async def recv(self, data):
		for pair in data.items():
			if pair[0] == 'ClosedLoop':
				self.closed_loop = pair[1]
				if self.closed_loop:
					self.power_soll = 0
				if self._on_closed_loop_changed:
					self._on_closed_loop_changed(self.closed_loop)
				await self.send({'ClosedLoop': self.closed_loop})
			elif pair[0] == 'Connect':
				if self.device.connect():
					self.device.activate(True)
			elif pair[0] == 'Disconnect':
				self.device.activate(False)
				self.device.disconnect()
			elif 'Power' in pair[0]:
				self.power_soll = pair[1]

	async def send(self, data):
		await self.msg_handler.send(data)

	def get_next_smaller_power(self, power):
		result = self.device.get_power_min()
		p_list = [p for p in self.device.get_powers() if p <= power]
		if len(p_list) > 0:
			result = p_list[-1]
		return result

	def get_next_greater_power(self, power):
		result = self.device.get_power_min()
		p_list = [p for p in self.device.get_powers() if p > power]
		if len(p_list) > 0:
			result = p_list[0]
		return result

	async def on_process(self):
		await self.send({'Capabilities': {'Power': {'Min': 0, 'Max': self.device.get_power_max()}}})
		await self.send({'Simulated': self.device.simulated})
		await self.send({'ClosedLoop': self.closed_loop})
		if self._on_closed_loop_changed:
			self._on_closed_loop_changed(self.closed_loop)

		pulse_period_s = 10
		pulse_period_count = pulse_period_s/self.interval

		# Auto-connect on startup - replaces what used to happen implicitly
		# in the device's own constructor; a missing/unplugged device just
		# stays disconnected instead of crashing the server (see
		# components/actor/heater_hendi.py's connect()).
		self.device.connect()
		# Explicit initial push, same as Capabilities/Simulated/ClosedLoop
		# above - a simulated device's connect() is a no-op that never
		# reassigns self.connected (see components/connectable.py), so its
		# set_on_changed('connected', ...) observer would otherwise never
		# fire and a freshly-subscribed client would never learn it's
		# connected.
		await self.send({'Connected': self.device.connected})

		# Outer retry loop - belt-and-suspenders around the tick loop's own
		# per-call guards below. A comm failure a device-level guard didn't
		# anticipate (e.g. a raw pyserial/termios exception escaping from
		# somewhere other than device.process(), the one call already
		# wrapped below) must never be allowed to fall out of this
		# coroutine entirely: nothing restarts a dead ATask, so a single
		# uncaught exception here would permanently stop this device from
		# ever being process()'d again - even after a later, otherwise-
		# successful reconnect. See the incident this was added for: a
		# mid-brew unplug raised termios.error from deep inside
		# HendiCtrl.disconnect() itself (fixed there too), which escaped
		# all the way out of this loop and killed it for the rest of the
		# process's life.
		while True:
			try:
				with self.device.open():
					while True:
						# Closed-loop: TC has full control. Open-loop: direct manual power only.
						power_soll = self.power_actor if self.closed_loop else self.power_soll
						self.power_set_changed(power_soll)

						# Calculate duty cycle
						power_low = self.get_next_smaller_power(power_soll)
						power_high = self.get_next_greater_power(power_soll)
						power_step = power_high - power_low
						power_diff = power_soll - power_low
						duty = power_diff / power_step

						on_count = pulse_period_count*duty

						self.pulse_counter += 1
						if self.pulse_counter >= pulse_period_count:
							self.pulse_counter = 0

						if power_soll == 0 or self.pulse_counter >= on_count:
							self.device.set_power(power_low)
						elif self.pulse_counter < on_count:
							self.device.set_power(power_high)

						try:
							self.device.process()
						except Exception as e:
							print(f"HeaterTask: comm error, marking disconnected: {e}")
							self.device.disconnect()
						await asyncio.sleep(self.interval)
			except Exception as e:
				print(f"HeaterTask: unexpected error, recovering: {e}")
				try:
					self.device.disconnect()
				except Exception:
					pass
				await asyncio.sleep(self.interval)