fix: make sim_warp_factor real-hardware-safe and sud logs forecast-reproducible
sim_warp_factor was scaling DT_TASK unconditionally, including against real hardware, and was leaking into recorded sample timestamps via wall-clock time - both violate its intended meaning (a reciprocal scale on task wait time only). Clamp it to 1.0 whenever Heater.type isn't "sim", and track elapsed time via tick-counted accumulators (Sud.elapsed for SudLogTask, a matching one for ServerLogTask) instead of time.monotonic(). Sud/server logs now also carry HeaterPowers, the effective SimWarpFactor, the raw Doc, a correct Name, and ForecastAnchors (one entry per real _reanchor_forecast() trigger) - enough for utils/analyze_log.py to rebuild a SudForecastEstimator and replay the exact same splice-per-transition forecast correction offline that a live client sees, replacing the old (always-dead) forecast_*.json sidecar lookup. Two bugs found and fixed along the way: - ServerLogTask's periodic wall-clock checkpoint had no way to know a SudLogTask run had just ended, so it would immediately re-write the just-completed log with PlantParams/ForecastAnchors cleared back to empty. Gated the checkpoint on _should_sample(). - _reanchor_forecast() truncated the old forecast against real elapsed time, which can end up numerically less than the old forecast's own (very wrong) speculative timestamps once a WAIT_USER confirmation runs long - leaving a "ghost" segment instead of a clean cut. Truncate against forecast_step_starts[index] instead, which lives in the same coordinate space as the forecast being cut. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JFDeoTKBDkXRhPfnyDEoBt
This commit is contained in:
+50
-8
@@ -16,7 +16,7 @@ class ServerLogTask(ATask):
|
||||
Sud state — useful for verifying controller and heater behaviour
|
||||
outside of a scheduled brew."""
|
||||
|
||||
def __init__(self, tc: APid, heater: AHeater, heater_task, interval, path='./logs', config=None, dt=None, log_interval=None):
|
||||
def __init__(self, tc: APid, heater: AHeater, heater_task, interval, path='./logs', config=None, dt=None, log_interval=None, sim_warp_factor=None):
|
||||
ATask.__init__(self, interval)
|
||||
self.tc = tc
|
||||
self.heater = heater
|
||||
@@ -25,14 +25,15 @@ class ServerLogTask(ATask):
|
||||
self.config = config
|
||||
self.dt = dt
|
||||
self.log_interval = log_interval
|
||||
self.sim_warp_factor = sim_warp_factor
|
||||
self._samples = []
|
||||
self._param_events = []
|
||||
self._t0 = time.monotonic()
|
||||
self._elapsed_total = 0.0
|
||||
self._run_id = time.strftime("%Y%m%dT%H%M%S", time.localtime())
|
||||
self._last_write = self._t0
|
||||
self._last_write = time.monotonic()
|
||||
|
||||
def log_plant_params(self, elapsed, params):
|
||||
self._param_events.append({'t': time.monotonic() - self._t0, 'params': params})
|
||||
self._param_events.append({'t': elapsed, 'params': params})
|
||||
|
||||
async def on_process(self):
|
||||
while True:
|
||||
@@ -40,20 +41,42 @@ class ServerLogTask(ATask):
|
||||
self._samples.append(self._take_sample())
|
||||
# Periodically checkpoints to disk (same file, overwritten whole
|
||||
# each time - see write()) so a hard crash/power loss only loses
|
||||
# up to log_interval seconds, not the whole session/run.
|
||||
# up to log_interval seconds, not the whole session/run. This
|
||||
# checkpoint cadence is deliberately real (wall-clock) time,
|
||||
# unlike _elapsed() below - it's bounding real data-loss risk,
|
||||
# not tracking brew progress.
|
||||
now = time.monotonic()
|
||||
if self.log_interval and now - self._last_write >= self.log_interval:
|
||||
# Gated on _should_sample(): for SudLogTask, a completed run's
|
||||
# stop_run() already wrote the correct final state and cleared
|
||||
# _param_events for the *next* run - an unguarded checkpoint
|
||||
# here would otherwise fire on the very next iteration (once
|
||||
# log_interval real seconds have passed since the last one,
|
||||
# regardless of the run having ended in between) and clobber
|
||||
# that already-correct file with _samples unchanged but
|
||||
# _param_events now empty.
|
||||
if self.log_interval and self._should_sample() and now - self._last_write >= self.log_interval:
|
||||
self.write()
|
||||
self._last_write = now
|
||||
await asyncio.sleep(self.interval)
|
||||
self._elapsed_total += self.dt
|
||||
|
||||
def _should_sample(self):
|
||||
return True
|
||||
|
||||
def _elapsed(self):
|
||||
"""Simulated seconds elapsed since this task started sampling -
|
||||
tick-counted (advances by self.dt once per on_process() iteration),
|
||||
never derived from wall-clock time: under sim_warp_factor, real and
|
||||
simulated time diverge (see components/sud.py's Sud.elapsed, and
|
||||
the README's "Forecast vs. actual duration" section, for the same
|
||||
reasoning), so a wall-clock 't' would desync from the forecast's
|
||||
own tick-counted timeline it's meant to be compared against."""
|
||||
return self._elapsed_total
|
||||
|
||||
def _take_sample(self):
|
||||
power_set = max(self.heater_task.power_soll, self.heater_task.power_actor)
|
||||
return {
|
||||
't': time.monotonic() - self._t0,
|
||||
't': self._elapsed(),
|
||||
'timestamp': time.time(),
|
||||
'temp_ist': self.tc.theta_ist,
|
||||
'temp_soll': self.tc.theta_soll_set,
|
||||
@@ -66,19 +89,38 @@ class ServerLogTask(ATask):
|
||||
def _filename(self):
|
||||
return 'log_{}.json'.format(self._run_id)
|
||||
|
||||
def _extra_log_data(self):
|
||||
"""Hook for subclasses (SudLogTask) to contribute additional keys
|
||||
to write()'s output - see there."""
|
||||
return {}
|
||||
|
||||
def _log_name(self):
|
||||
"""Hook for subclasses (SudLogTask) - a server-session log covers
|
||||
the whole process lifetime rather than one named run, so it has
|
||||
none of its own."""
|
||||
return ''
|
||||
|
||||
def write(self):
|
||||
if not self._samples:
|
||||
return
|
||||
os.makedirs(self.path, exist_ok=True)
|
||||
filename = self._filename()
|
||||
path = os.path.join(self.path, filename)
|
||||
log_data = {'Name': ''}
|
||||
log_data = {'Name': self._log_name()}
|
||||
if self.dt is not None:
|
||||
log_data['dt'] = self.dt
|
||||
if self.sim_warp_factor is not None:
|
||||
log_data['SimWarpFactor'] = self.sim_warp_factor
|
||||
if self.config is not None:
|
||||
log_data['Config'] = self.config
|
||||
# Queried live from the connected device rather than reconstructed
|
||||
# offline (utils/analyze_log.py has no hardware to ask) - needed to
|
||||
# reproduce this run's SudForecastEstimator (see components/
|
||||
# sud_forecast.py) for a post-hoc forecast-vs-actual comparison.
|
||||
log_data['HeaterPowers'] = self.heater.get_powers()
|
||||
if self._param_events:
|
||||
log_data['PlantParams'] = self._param_events
|
||||
log_data.update(self._extra_log_data())
|
||||
log_data['Samples'] = self._samples
|
||||
with open(path, 'w') as f:
|
||||
json.dump(log_data, f, indent='\t')
|
||||
|
||||
+35
-4
@@ -113,6 +113,7 @@ class SudTask(ATask):
|
||||
self._on_end = None
|
||||
self._on_start = None
|
||||
self._on_plant_params = None
|
||||
self._on_reanchor = None
|
||||
msg_handler.set_recv_handler(self.recv)
|
||||
|
||||
def set_on_plant_params(self, callback):
|
||||
@@ -120,6 +121,16 @@ class SudTask(ATask):
|
||||
so external observers (e.g. SudLogTask) can record each change."""
|
||||
self._on_plant_params = callback
|
||||
|
||||
def set_on_reanchor(self, callback):
|
||||
"""Register a callback invoked with (elapsed, index, theta_ist)
|
||||
every time _reanchor_forecast() fires (see on_step_changed()) -
|
||||
lets an external observer (SudLogTask) record the exact anchor
|
||||
points a live client's forecast got corrected at, so an offline
|
||||
re-simulation (utils/analyze_log.py) can reproduce the same
|
||||
splice-per-transition forecast instead of just the single,
|
||||
uncorrected cold-start one."""
|
||||
self._on_reanchor = callback
|
||||
|
||||
def apply_plant_params(self, grain_mass, water_mass):
|
||||
"""Keeps the real plant's and the controller's internal model's
|
||||
plant params in sync with this Sud's own doc - L/Td come straight
|
||||
@@ -215,6 +226,12 @@ class SudTask(ATask):
|
||||
# (a malt fill-in's actual cooldown, a longer/shorter ramp than
|
||||
# modeled, ...) at the next opportunity, not just at the next
|
||||
# user confirmation.
|
||||
if self._on_reanchor:
|
||||
# Read synchronously, right here - the same values
|
||||
# _reanchor_forecast() itself will read moments later, off
|
||||
# the same unyielded call stack, before anything else can
|
||||
# mutate them.
|
||||
self._on_reanchor(self.sud.elapsed, self.sud.index, self.tc.get_theta_ist())
|
||||
asyncio.create_task(self._reanchor_forecast())
|
||||
|
||||
asyncio.create_task(self.send({'Step': {
|
||||
@@ -373,16 +390,30 @@ class SudTask(ATask):
|
||||
self._forecast_generation += 1
|
||||
generation = self._forecast_generation
|
||||
real_elapsed = self.sud.elapsed
|
||||
schedule = self.sud.schedule
|
||||
index = self.sud.index
|
||||
# Drop the now-stale tail (everything beyond right now) - it's
|
||||
# about to be replaced by a freshly anchored simulation.
|
||||
cut = bisect.bisect_right(self.forecast_t, real_elapsed)
|
||||
# about to be replaced by a freshly anchored simulation. Cut at
|
||||
# forecast_step_starts[index], the old forecast's own (possibly
|
||||
# very wrong) belief of where step `index` begins - not at
|
||||
# real_elapsed directly: a step whose real timing blew way past
|
||||
# what its zero-delay guess assumed (a long WAIT_USER confirm,
|
||||
# above all - see SudForecastEstimator.estimate()'s docstring)
|
||||
# leaves the old forecast's *entire* speculative remainder sitting
|
||||
# at timestamps still numerically less than real_elapsed, so
|
||||
# bisecting against real_elapsed itself would find nothing to
|
||||
# discard and just tack the fresh, correct simulation on after it
|
||||
# - a doubled-back, self-overlapping curve instead of a clean cut.
|
||||
# forecast_step_starts[index] doesn't have this problem: it lives
|
||||
# in the same (speculative) coordinate space as forecast_t itself,
|
||||
# so the cut lands in the right place regardless of how far real
|
||||
# and speculated time have diverged by now.
|
||||
cut = bisect.bisect_right(self.forecast_t, self.forecast_step_starts.get(index, real_elapsed))
|
||||
forecast_t = self.forecast_t[:cut]
|
||||
forecast_theta = self.forecast_theta[:cut]
|
||||
# Steps already passed (< index) have their real, now-immutable
|
||||
# start time; anything from index on is about to be resimulated
|
||||
# fresh below and must not keep a stale prediction around.
|
||||
schedule = self.sud.schedule
|
||||
index = self.sud.index
|
||||
forecast_step_starts = {i: tt for i, tt in self.forecast_step_starts.items() if i < index}
|
||||
|
||||
if not (0 <= index < len(schedule)):
|
||||
|
||||
+40
-4
@@ -14,10 +14,12 @@ class SudLogTask(ServerLogTask):
|
||||
logs/log_{date_time}_{sud_name}.log. A Pause/resume doesn't start a new
|
||||
file - only a fresh Play (from IDLE/DONE) does."""
|
||||
|
||||
def __init__(self, tc: APid, heater: AHeater, heater_task, sud, interval, path='./logs', config=None, dt=None, log_interval=None):
|
||||
ServerLogTask.__init__(self, tc, heater, heater_task, interval, path, config, dt, log_interval)
|
||||
def __init__(self, tc: APid, heater: AHeater, heater_task, sud, interval, path='./logs', config=None, dt=None, log_interval=None, sim_warp_factor=None):
|
||||
ServerLogTask.__init__(self, tc, heater, heater_task, interval, path, config, dt, log_interval, sim_warp_factor)
|
||||
self.sud = sud
|
||||
self._active = False
|
||||
self._doc = None
|
||||
self._reanchors = []
|
||||
|
||||
def _should_sample(self):
|
||||
return self._active
|
||||
@@ -26,13 +28,46 @@ class SudLogTask(ServerLogTask):
|
||||
name_part = _safe_filename_part(self.sud.name or 'Sud')
|
||||
return 'log_{}_{}.log'.format(self._run_id, name_part)
|
||||
|
||||
def _log_name(self):
|
||||
return self.sud.name
|
||||
|
||||
def _elapsed(self):
|
||||
# The Sud's own tick-counted ground truth (see components/sud.py's
|
||||
# Sud.elapsed) rather than the base class's own tick counter -
|
||||
# already resets to 0 on every fresh Play (Sud.start()), so samples
|
||||
# line up with SudForecastEstimator's t=0 the same way Sud.elapsed
|
||||
# does everywhere else it's consumed (e.g. SudTask's forecast
|
||||
# re-anchoring).
|
||||
return self.sud.elapsed
|
||||
|
||||
def log_reanchor(self, elapsed, index, theta_ist):
|
||||
# One entry per real _reanchor_forecast() trigger (see SudTask.
|
||||
# set_on_reanchor()) - lets utils/analyze_log.py replay the exact
|
||||
# same splice-per-transition forecast correction a live client
|
||||
# would have seen, instead of just the single, uncorrected
|
||||
# cold-start estimate() run - see build_forecast() there.
|
||||
self._reanchors.append({'t': elapsed, 'index': index, 'theta_ist': theta_ist})
|
||||
|
||||
def _extra_log_data(self):
|
||||
# The raw sud.json doc this run was playing, exactly as loaded (see
|
||||
# Sud.save()) - lets utils/analyze_log.py re-simulate this run's own
|
||||
# SudForecastEstimator forecast for comparison against Samples,
|
||||
# without depending on the sude/*.json file it came from still
|
||||
# existing or being unchanged.
|
||||
data = {}
|
||||
if self._doc is not None:
|
||||
data['Doc'] = self._doc
|
||||
if self._reanchors:
|
||||
data['ForecastAnchors'] = self._reanchors
|
||||
return data
|
||||
|
||||
def start_run(self):
|
||||
if self._active:
|
||||
return
|
||||
self._samples = []
|
||||
self._t0 = time.monotonic()
|
||||
self._last_write = self._t0
|
||||
self._last_write = time.monotonic()
|
||||
self._run_id = time.strftime("%Y%m%dT%H%M%S", time.localtime())
|
||||
self._doc = self.sud.save()
|
||||
self._active = True
|
||||
|
||||
def stop_run(self):
|
||||
@@ -41,3 +76,4 @@ class SudLogTask(ServerLogTask):
|
||||
self._active = False
|
||||
self.write()
|
||||
self._param_events = []
|
||||
self._reanchors = []
|
||||
|
||||
Reference in New Issue
Block a user