jensandClaude Sonnet 4.6
b33f89a569
sude/sud_0010.json moved from a list of combined ramp+hold+wait
"Rasten" with global stirrer constants to a flat "schedule" list of
explicit {"type": "heat"|"hold", ...} steps, each carrying its own
stirrer_speed/stirrer_time_on/stirrer_time_off and an optional
user_wait_for_continue (+ user_message).
- components/sud.py: Sud now tracks the current `step` instead of
`rast`; "heat" steps enter RAMPING (advanced externally via
temp_reached()), "hold" steps enter HOLDING and count down
`duration` minutes (0 if omitted). Either step type can pause in
WAIT_USER via user_wait_for_continue, surfaced through the new
user_message attribute.
- tasks/sud.py: SudTask only pushes theta_soll/heatrate_soll on
"heat" steps, converts each step's stirrer_time_on/off into a
duty_cycle/cycle_time on every step change (replacing the old
state-based RAMPING/HOLDING stirrer switching), and broadcasts
user_message changes. Stirring is now fully schedule-driven instead
of implicitly stopped on WAIT_USER/DONE (DONE still stops it, since
there's no step left to read settings from).
- scripts/demos/sud/demo_sud.py: mirrors the same step-driven wiring.
- README's Mash schedules section rewritten for the new schema.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
import asyncio
|
|
from tasks import ATask
|
|
from ws.message import MsgIo
|
|
from utils.value import ChangedFloat
|
|
from components import APid, AStirrer
|
|
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, interval, msg_handler: MsgIo):
|
|
ATask.__init__(self, interval)
|
|
self.sud = sud
|
|
self.tc = tc
|
|
self.stirrer = stirrer
|
|
self.msg_handler = msg_handler
|
|
msg_handler.set_recv_handler(self.recv)
|
|
|
|
def apply_stirrer(self, step):
|
|
speed = step.get('stirrer_speed', 0)
|
|
on = step.get('stirrer_time_on', 0)
|
|
off = step.get('stirrer_time_off', 0)
|
|
cycle = on + off
|
|
if cycle > 0:
|
|
duty = on / cycle
|
|
else:
|
|
cycle = 1.0
|
|
duty = 1.0 if speed > 0 else 0.0
|
|
|
|
self.stirrer.set_cycle_time(cycle)
|
|
self.stirrer.set_duty_cycle(duty)
|
|
self.stirrer.set_speed(speed)
|
|
|
|
def on_step_changed(self, step):
|
|
if step is not None:
|
|
if step['type'] == 'heat':
|
|
self.tc.set_theta_soll(step['temp'])
|
|
self.tc.set_heatrate_soll(step['rate'])
|
|
self.apply_stirrer(step)
|
|
|
|
asyncio.create_task(self.send({'Step': {
|
|
'Index': self.sud.index,
|
|
'Type': step['type'] if step else None,
|
|
'Descr': step.get('descr') if step else None,
|
|
'Temp': step.get('temp') if step else None,
|
|
'Rate': step.get('rate') if step else None,
|
|
'Duration': step.get('duration') if step 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 == SudState.DONE:
|
|
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()
|
|
|
|
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.interval)
|
|
await asyncio.sleep(self.interval)
|