jensandClaude Sonnet 4.6
57b86bc075
Previously, the dynamic forecast's projected remainder was fully
recomputed on every step change, and a step requiring user
confirmation was simulated with zero added delay ("count it as
instant for estimation purposes"). Both undercut the actual goal of
the forecast: an honest comparison between real control behavior and
a simulation-based prediction. A forecast that keeps re-anchoring
itself to match reality isn't a useful baseline to compare reality
against, and assuming zero confirmation delay quietly understates the
schedule.
components/sud_forecast.py: SudForecastEstimator.estimate() now stops
simulating at the first step requiring user confirmation instead of
auto-confirming, returning the final SudState alongside the data so
the caller knows whether it stopped there or actually finished.
tasks/sud.py: SudTask now accumulates the forecast across piecewise
segments (forecast_t/forecast_theta). send_forecast() computes only
the first segment, at Load/Save. The real Confirm handler triggers
_continue_forecast_after_confirm(), which computes the next segment
anchored at the real elapsed time and real current temperature -
bridging the unforecastable wait with a flat segment rather than
guessing at its length - and sends the updated, stitched Forecast
(now carrying a Finished flag). The old per-step-change
RemainingForecast recompute is gone entirely, along with
remaining_schedule()/send_remaining_forecast().
client/brewpi_gui.py: SudForecastPlot redesigned around this - the
dashed line is the fixed, piecewise forecast (locked axes, untouched
except when a genuinely new segment arrives via show_forecast());
the solid line is purely the actual measured trace
(show_dynamic(), now only touching that). extend_forecast_while_
waiting() repeats the forecast's last value while the real Sud sits
in WAIT_USER, so the dashed line doesn't just stop, then snaps to the
new segment the moment the real confirmation appends one.
Verified via a raw-protocol test (segment boundaries, the Finished
flag, and real-time stitching across a confirm delay) and visually in
the GUI (locked dashed forecast, "X min so far total" title while
incomplete, solid/dashed comparison rendering with realistic
convergence/divergence).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LhiQe64F74uHV8jzuoSa5K
109 lines
4.5 KiB
Python
109 lines
4.5 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, final_state): t/theta are parallel lists of
|
|
elapsed simulated seconds and temperature for one piecewise
|
|
segment of the schedule, starting from doc['steps'][0] - either
|
|
all the way to the end (final_state is SudState.DONE), or, if a
|
|
step requires user confirmation, only up to and including that
|
|
step's hold (final_state is SudState.WAIT_USER). A real user's
|
|
confirm time can't be forecast, so the simulation simply stops
|
|
there instead of guessing zero delay or otherwise - the caller is
|
|
expected to compute the next segment for real once the user
|
|
actually confirms (see tasks/sud.py's SudTask), anchored at
|
|
whatever the real elapsed time and temperature are by then,
|
|
rather than this estimate continuing to guess at it.
|
|
|
|
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], SudState.DONE
|
|
|
|
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())
|
|
# Seed the target at start_theta - this tc is a fresh, throwaway
|
|
# instance (unlike the real run's persistent one), so without this
|
|
# its theta_soll_set defaults to 0 until a step pushes its own.
|
|
# Steps without their own 'temperature' (common now that ramping
|
|
# isn't gated by a 'ramp' key - see components/sud.py) rely on
|
|
# inheriting whatever target was already running, which for a
|
|
# schedule starting mid-brew (the dynamic remaining forecast) is
|
|
# start_theta, not 0 - without this, such a schedule's first step
|
|
# would have the simulated controller chase 0 degrees indefinitely,
|
|
# hitting MAX_TICKS and producing a needlessly huge result.
|
|
tc.set_theta_soll(start_theta)
|
|
|
|
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 not in (SudState.DONE, SudState.WAIT_USER) 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()
|
|
sud.tick(self.dt)
|
|
|
|
t.append(t[-1] + self.dt)
|
|
theta.append(pot.get_temperature())
|
|
ticks += 1
|
|
|
|
return t, theta, sud.state
|