jensandClaude Sonnet 4.6
47fb09aa06
Callbacks were registered inside the async on_process() coroutine, meaning they weren't active until the event loop started ticking. Moving them to __init__ ensures they're wired up at construction time. Also removes remaining debug prints from the affected task files. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tqxrk8uj4M3w3d3eXm3xK8
91 lines
2.6 KiB
Python
Executable File
91 lines
2.6 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.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 actor(self, y):
|
|
self.power_actor = max(0, self.device.get_power_max() * y)
|
|
|
|
def on_changed_active(self, is_activated):
|
|
asyncio.create_task(self.send({'Activate': int(is_activated)}))
|
|
|
|
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 '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()}}})
|
|
|
|
pulse_period_s = 10
|
|
pulse_period_count = pulse_period_s/self.interval
|
|
|
|
with self.device.open():
|
|
while True:
|
|
# Combine power set from different sources
|
|
power_soll = max(self.power_soll, self.power_actor)
|
|
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)
|
|
|
|
|
|
|