jensandClaude Sonnet 4.6
6d481d7764
Adds HeaterTask.shutdown() which switches to open-loop at power 0, disables TC, and broadcasts ClosedLoop=false to clients so the browser checkbox unchecks immediately. SudTask fires it via a new set_on_end() callback whenever state reaches DONE or IDLE, giving both a completed schedule and a manual Stop the same safe behaviour: stirrer and heater both off, no heater left running unattended. Re-enabling after shutdown: check the closed-loop checkbox (sends ClosedLoop=true → TC re-enabled) or adjust the power slider directly; the next brew's Play press already sends ClosedLoop=true automatically. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SH2D4LRuCnhLSzoYerAfJX
113 lines
3.4 KiB
Python
Executable File
113 lines
3.4 KiB
Python
Executable File
import asyncio
|
|
from tasks import ATask
|
|
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
|
|
device.set_on_changed('power_eff', ChangedInteger(self.on_changed_power).set)
|
|
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 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 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 '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({'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
|
|
|
|
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)
|
|
|
|
self.device.process()
|
|
await asyncio.sleep(self.interval)
|
|
|
|
|
|
|