jensandClaude Sonnet 4.6
33f75c22d2
Makes it easier to follow which schedule step is currently active when watching the demo run. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CgR9tPaSzFkAwRAUyeaaCD
128 lines
3.5 KiB
Python
128 lines
3.5 KiB
Python
import json
|
|
from enum import Enum
|
|
from utils.value import AttributeChange
|
|
|
|
# Specific heat capacities [J/(kg*K)] used to derive a single lumped
|
|
# (mass, specific heat) pair for the pot's contents from pot_mass/
|
|
# grain_mass/water_mass. Approximate, typical values.
|
|
SPECIFIC_HEAT_WATER = 4190
|
|
SPECIFIC_HEAT_GRAIN = 1800
|
|
SPECIFIC_HEAT_BY_MATERIAL = {
|
|
"Edelstahl 18/10": 500,
|
|
}
|
|
|
|
|
|
class SudState(Enum):
|
|
IDLE = 0
|
|
RAMPING = 1
|
|
HOLDING = 2
|
|
WAIT_USER = 3
|
|
DONE = 4
|
|
|
|
|
|
def _merge_defaults(default, override):
|
|
"""Deep-merges override onto default, recursing into nested dicts so
|
|
values missing from override fall back to default's at every level."""
|
|
merged = dict(default)
|
|
for key, value in override.items():
|
|
if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
|
|
merged[key] = _merge_defaults(merged[key], value)
|
|
else:
|
|
merged[key] = value
|
|
return merged
|
|
|
|
|
|
def _build_step(number, defaults, raw_step):
|
|
"""Fills in a schedule step with default.step's values. A step is a
|
|
ramp or a hold depending on which of those keys it specifies; the other
|
|
is left out rather than synthesized from defaults."""
|
|
step = {'number': number}
|
|
for key in ('descr', 'user_message', 'user_wait_for_continue'):
|
|
step[key] = raw_step.get(key, defaults.get(key))
|
|
for key in ('ramp', 'hold'):
|
|
if key in raw_step:
|
|
step[key] = _merge_defaults(defaults.get(key, {}), raw_step[key])
|
|
return step
|
|
|
|
|
|
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', '')
|
|
|
|
step_defaults = data.get('default', {}).get('step', {})
|
|
self.schedule = [_build_step(i + 1, step_defaults, raw) for i, raw in enumerate(data['steps'])]
|
|
|
|
self.pot_mass = data.get('pot_mass', 0)
|
|
self.pot_material = data.get('pot_material')
|
|
self.grain_mass = data.get('grain_mass', 0)
|
|
self.water_mass = data.get('water_mass', 0)
|
|
|
|
self.index = -1
|
|
self.hold_remaining = 0.0
|
|
self.state = SudState.IDLE
|
|
self.step = None
|
|
self.user_message = None
|
|
|
|
def derive_plant_params(self):
|
|
"""Lumped (mass, specific heat) lifted from pot_mass/pot_material/
|
|
grain_mass/water_mass, for use as Pot's "M"/"C" params."""
|
|
c_pot = SPECIFIC_HEAT_BY_MATERIAL.get(self.pot_material, SPECIFIC_HEAT_WATER)
|
|
mass = self.water_mass + self.grain_mass + self.pot_mass
|
|
capacitance = (self.water_mass * SPECIFIC_HEAT_WATER
|
|
+ self.grain_mass * SPECIFIC_HEAT_GRAIN
|
|
+ self.pot_mass * c_pot)
|
|
|
|
return {
|
|
'M': mass,
|
|
'C': capacitance / mass if mass > 0 else SPECIFIC_HEAT_WATER,
|
|
}
|
|
|
|
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._finish_step()
|
|
|
|
def tick(self, dt):
|
|
if self.state == SudState.HOLDING:
|
|
self.hold_remaining -= dt
|
|
if self.hold_remaining <= 0:
|
|
self._finish_step()
|
|
|
|
def _advance(self):
|
|
self.index += 1
|
|
if self.index >= len(self.schedule):
|
|
self.state = SudState.DONE
|
|
self.user_message = None
|
|
self.step = None
|
|
return
|
|
|
|
next_step = self.schedule[self.index]
|
|
if 'ramp' in next_step:
|
|
self.state = SudState.RAMPING
|
|
else:
|
|
self.hold_remaining = next_step['hold'].get('duration', 0)
|
|
self.state = SudState.HOLDING
|
|
|
|
self.user_message = next_step.get('user_message')
|
|
self.step = next_step
|
|
|
|
def _finish_step(self):
|
|
if self.step.get('user_wait_for_continue', False):
|
|
self.state = SudState.WAIT_USER
|
|
else:
|
|
self._advance()
|