jensandClaude Sonnet 4.6
2bdd3203cf
Previously, whether a schedule step ramped at all was decided purely by the presence of a 'ramp' key (components/sud.py's _advance()) - a hold-only step jumped straight into its hold countdown, assuming the plant was already at temperature. Now every step ramps toward 'temperature' first, for as long as the controller's own gap-tracking FSM (TempControllerBase, already distinguishing HEAT/COOL/HOLD) says the gap actually warrants it - via the new is_holding() (on APid and TempControllerBase), which replaces a separate, redundant TEMP_REACHED_TOLERANCE constant duplicated across tasks/sud.py, components/sud_forecast.py, and scripts/demos/sud/demo_sud.py. set_theta_soll() now recomputes the FSM eagerly so is_holding() can't read stale HOLD for a tick after a much-further-away target is pushed. components/sud.py's _build_step() always synthesizes a 'ramp' block from default.step.ramp (so 'rate' is always available), but leaves 'temperature' undefaulted - every real sude/*.json's default.step.temperature is inert template filler, and defaulting it would send hold-only steps chasing 0 degrees. SudTask/ SudForecastEstimator/the demo only push a new theta_soll/heatrate_soll when a step actually specifies its own temperature; otherwise the controller keeps whatever the previous step left running. Also fixes a related crash this surfaced: SudTask.remaining_schedule()'s synthetic mid-hold step dropped 'ramp' to signal "don't re-ramp" - now that every step always carries a 'ramp' dict, dropping it left 'rate' missing the moment the synthetic step was re-resolved by SudForecastEstimator. Drops 'temperature' instead, which is what actually signals "no new target" under the new model. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LhiQe64F74uHV8jzuoSa5K
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
from components.plant import Pot
|
|
from components.pid import PidFactory
|
|
from components.sud import Sud, SudState
|
|
|
|
# Safety cap so a schedule whose target a step can never actually reach
|
|
# (e.g. a "hold" colder than ambient with no active cooling) can't hang the
|
|
# simulation forever - the estimate is simply cut off there.
|
|
MAX_TICKS = 200000
|
|
|
|
|
|
class SudForecastEstimator:
|
|
"""Predicts how long a Sud schedule will actually take by simulating it
|
|
with the same machinery (and params) the real server's brewpi.py wires
|
|
up - a fresh Pot and temperature controller of the configured pid_type,
|
|
driven through the schedule exactly as tasks/sud.py's SudTask would.
|
|
|
|
This is deliberately independent of wall-clock/asyncio time: it just
|
|
iterates dt-sized ticks as fast as the CPU allows (a multi-hour brew
|
|
simulates in well under a second), so it can be run synchronously
|
|
whenever a client needs an estimate - the naive "abs(delta)/rate" model
|
|
the GUI used to compute itself has no way to see the real PID cascade's
|
|
spin-up/settling lag, which is exactly why its estimate drifted so far
|
|
from reality (see README.md's "Forecast vs. actual duration")."""
|
|
|
|
def __init__(self, dt, theta_amb, plant_params, pid_type, tempctrl_params, heater_max_power):
|
|
self.dt = dt
|
|
self.theta_amb = theta_amb
|
|
self.plant_params = plant_params
|
|
self.pid_type = pid_type
|
|
self.tempctrl_params = tempctrl_params
|
|
self.heater_max_power = heater_max_power
|
|
|
|
def set_ambient_temperature(self, theta_amb):
|
|
self.theta_amb = theta_amb
|
|
|
|
def estimate(self, doc, start_theta=None):
|
|
"""Returns (t, theta): parallel lists of elapsed simulated seconds
|
|
and temperature, one point per tick, for the given sud.json
|
|
document. start_theta defaults to the configured ambient
|
|
temperature - i.e. a cold start, same as the GUI's static
|
|
estimate."""
|
|
if start_theta is None:
|
|
start_theta = self.theta_amb
|
|
|
|
sud = Sud()
|
|
if not sud.load(doc) or not sud.schedule:
|
|
return [0.0], [start_theta]
|
|
|
|
pot = Pot(self.dt, self.plant_params, self.theta_amb)
|
|
pot.initial(start_theta)
|
|
tc = PidFactory.create(self.pid_type, self.dt, self.tempctrl_params, self.plant_params, theta_amb=self.theta_amb)
|
|
tc.set_enabled(True)
|
|
tc.set_theta_ist(pot.get_temperature())
|
|
|
|
def on_step_changed(step):
|
|
if step is None:
|
|
return
|
|
params = sud.derive_plant_params(step.get('grain_mass', 0), step.get('water_mass', 0))
|
|
pot.set_thermal_params(params['M'], params['C'])
|
|
if hasattr(tc, 'set_model_params'):
|
|
tc.set_model_params(params['M'], params['C'])
|
|
if sud.state == SudState.RAMPING and step['temperature'] is not None:
|
|
tc.set_theta_soll(step['temperature'])
|
|
tc.set_heatrate_soll(step['ramp']['rate'])
|
|
sud.set_on_changed('step', on_step_changed)
|
|
|
|
t = [0.0]
|
|
theta = [pot.get_temperature()]
|
|
|
|
sud.start()
|
|
ticks = 0
|
|
while sud.state != SudState.DONE and ticks < MAX_TICKS:
|
|
pot.process()
|
|
tc.set_theta_ist(pot.get_temperature())
|
|
tc.process()
|
|
pot.set_power(max(0, self.heater_max_power * tc.get_power()))
|
|
|
|
if sud.state == SudState.RAMPING:
|
|
if tc.is_holding():
|
|
sud.temp_reached()
|
|
elif sud.state == SudState.WAIT_USER:
|
|
# A real user's confirm time isn't predictable - count it
|
|
# as instant for estimation purposes.
|
|
sud.confirm()
|
|
sud.tick(self.dt)
|
|
|
|
t.append(t[-1] + self.dt)
|
|
theta.append(pot.get_temperature())
|
|
ticks += 1
|
|
|
|
return t, theta
|