jensandClaude Sonnet 4.6
7e131df4ca
brewpi.py had a dead -m/--model CLI arg and sude/*.json schedules were pure data with no code to drive them, despite the README documenting them as if they were already wired up. Add components/sud.py's Sud, which loads a sude/*.json schedule and steps through its Rasten (ramp -> hold -> optional wait-for-user -> next rest), and tasks/sud.py's SudTask, which drives the temperature controller's theta_soll/heatrate_soll from the current rest and switches the stirrer between continuous (ramping) and intermittent (holding, via stirrSpeedRast/stirrDutyRast/stirrCycleTime) operation. Exposes progress on a new "Sud" WebSocket channel and accepts Start/Confirm commands. Enabled via a new optional top-level "sud" config key (see config.json.templ/.sim). "Reached target" is detected via abs(theta_ist - theta_soll_set) < tolerance rather than tc.state == HOLD, since the latter can still read HOLD left over from the previous rest for one tick after a new target is pushed (tc.state only updates on the controller's own, independently-scheduled process() tick). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
71 lines
1.5 KiB
Python
71 lines
1.5 KiB
Python
import json
|
|
from enum import Enum
|
|
from utils.value import AttributeChange
|
|
|
|
|
|
class SudState(Enum):
|
|
IDLE = 0
|
|
RAMPING = 1
|
|
HOLDING = 2
|
|
WAIT_USER = 3
|
|
DONE = 4
|
|
|
|
|
|
class Sud(AttributeChange):
|
|
def __init__(self, path):
|
|
AttributeChange.__init__(self)
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
|
|
self.name = data.get('Name', '')
|
|
self.description = data.get('Description', '')
|
|
self.rasten = data['Rasten']
|
|
|
|
self.stirr_speed_heat = data.get('stirrSpeedHeat', 0)
|
|
self.stirr_speed_rast = data.get('stirrSpeedRast', 0)
|
|
self.stirr_duty_rast = data.get('stirrDutyRast', 1.0)
|
|
self.stirr_cycle_time = data.get('stirrCycleTime', 1.0)
|
|
|
|
self.index = -1
|
|
self.hold_remaining = 0.0
|
|
self.state = SudState.IDLE
|
|
self.rast = None
|
|
|
|
def start(self):
|
|
if self.state != SudState.IDLE:
|
|
return
|
|
self.index = -1
|
|
self._advance()
|
|
|
|
def confirm(self):
|
|
if self.state == SudState.WAIT_USER:
|
|
self._advance()
|
|
|
|
def temp_reached(self):
|
|
if self.state == SudState.RAMPING:
|
|
self.state = SudState.HOLDING
|
|
|
|
def tick(self, dt):
|
|
if self.state == SudState.HOLDING:
|
|
self.hold_remaining -= dt
|
|
if self.hold_remaining <= 0:
|
|
self._finish_rast()
|
|
|
|
def _advance(self):
|
|
self.index += 1
|
|
if self.index >= len(self.rasten):
|
|
self.state = SudState.DONE
|
|
self.rast = None
|
|
return
|
|
|
|
next_rast = self.rasten[self.index]
|
|
self.hold_remaining = next_rast['time'] * 60.0
|
|
self.state = SudState.RAMPING
|
|
self.rast = next_rast
|
|
|
|
def _finish_rast(self):
|
|
if self.rast.get('waitForUser', False):
|
|
self.state = SudState.WAIT_USER
|
|
else:
|
|
self._advance()
|