jens/brewpi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: embed dt and startup plant params in server log; open-loop replay_sim

Browse Source 
server/brewpi.py: pass physics DT to ServerLogTask (not wall-clock DT_TASK)
so logs carry the correct integration timestep; log startup plant params as
the first PlantParams event so replay_sim can reconstruct them even when no
Sud is ever loaded (previously fell back to hardcoded defaults, causing ~3°C
temperature offset in replay).

tasks/server_log.py: accept and embed dt in log JSON; switch
log_plant_params() to wall-clock monotonic time (same base as sample t),
since the elapsed argument was simulated time and not comparable.

utils/replay_sim.py: switch to open-loop replay (plant driven by logged
power_eff, not sim controller output); use PidFactory so Smith vs Normal
controller type is honoured; set Td=0 in replay plant for Smith mode (logged
temp_ist is already Smith-corrected); feed Smith predictor's internal model
with real power; apply embedded PlantParams mid-run; prefer embedded dt over
sample-based fallback; clamp sim power ≥ 0; replace meaningless sim-power
trace in power subplot with power_set vs power_eff from the log.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jens
2026-07-01 00:31:46 +02:00
co-authored by Claude Sonnet 4.6
parent 4a1c969ad0
commit 57c23f293a
3 changed files with 72 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/brewpi.py
+4 -1
View File
@@ -110,6 +110,7 @@ if __name__ == '__main__':
# the kettle at startup (e.g. strike water pre-filled). A loaded sud's
# own apply_plant_params() will override this on the first step.
startup_water = config.get('Pot', {}).get('water_mass', 0)
startup_params = None
if startup_water > 0:
startup_params = sud.derive_plant_params(0, startup_water)
pot.set_plant_params(startup_params)
@@ -131,8 +132,10 @@ if __name__ == '__main__':
# Continuous server-session log - records from startup to shutdown
# regardless of Sud state; written to logs/log_{date_time}.json on exit.
server_log_task = ServerLogTask(tc, heater, heater_task, DT_TASK, args.logdir, config=config)
server_log_task = ServerLogTask(tc, heater, heater_task, DT_TASK, args.logdir, config=config, dt=DT)
sud_task.set_on_plant_params(server_log_task.log_plant_params)
if startup_params is not None:
server_log_task.log_plant_params(0, startup_params)
taskmgr.add(server_log_task)
# Assign data flow
tasks/server_log.py
+5 -2
View File
@@ -16,20 +16,21 @@ 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):
def __init__(self, tc: APid, heater: AHeater, heater_task, interval, path='./logs', config=None, dt=None):
ATask.__init__(self, interval)
self.tc = tc
self.heater = heater
self.heater_task = heater_task
self.path = path
self.config = config
self.dt = dt
self._samples = []
self._param_events = []
self._t0 = time.monotonic()
self._run_id = time.strftime("%Y%m%dT%H%M%S", time.localtime())
def log_plant_params(self, elapsed, params):
self._param_events.append({'t': elapsed, 'params': params})
self._param_events.append({'t': time.monotonic() - self._t0, 'params': params})
async def on_process(self):
while True:
@@ -53,6 +54,8 @@ class ServerLogTask(ATask):
filename = 'log_{}.json'.format(self._run_id)
path = os.path.join(self.path, filename)
log_data = {'Name': ''}
if self.dt is not None:
log_data['dt'] = self.dt
if self.config is not None:
log_data['Config'] = self.config
if self._param_events:
utils/replay_sim.py
+63 -29
View File
@@ -41,7 +41,7 @@ from matplotlib.ticker import AutoMinorLocator
sys.path.insert(0, str(Path(__file__).parent.parent))
from components.pid.temp_controller import TempController
from components.pid import PidFactory
from components.plant.pot import Pot
@@ -70,37 +70,65 @@ def _infer_heatrate_soll_set(samples):
return max(candidates, default=1.0) or 1.0
def run_replay(samples, pid_params, heatrate_soll_set, plant_params, ambient, max_power, heater_efficiency, dt):
"""Closed-loop simulation: controller drives plant, plant feeds temp back.
def run_replay(samples, pid_type, pid_params, heatrate_soll_set, plant_params, param_events,
ambient, dt):
"""Open-loop plant replay with parallel controller observation.
The plant is seeded at the real starting temperature from the log so the
simulation begins at a realistic operating point. Setpoints follow the
real run's temp_soll schedule sample by sample."""
tc = TempController(dt)
The plant is driven by the real logged power_eff each tick so that
sim_temp_ist tracks real_temp_ist when the plant model is accurate.
The controller runs in parallel (observing sim_temp) to produce a
comparable power signal — useful to spot gain issues — but its output
does NOT feed back into the plant.
param_events is a list of {t, params} dicts (from the log's PlantParams
section) applied at their logged wall-clock t so the simulated plant
tracks the same M/C changes (grain added, water boiling off) as the
real run."""
tc = PidFactory.create(pid_type, dt)
tc.set_params(pid_params)
tc.set_ambient_temperature(ambient)
tc.set_model_plant_params(plant_params)
tc.set_enabled(True)
# The logged temp_ist is Smith-corrected (transport delay already removed
# by the predictor), so the replay plant must not add another Td on top.
# For Normal mode the raw sensor reading is logged so keep Td as-is.
replay_plant_params = dict(plant_params, Td=0) if pid_type == 'Smith' else plant_params
plant = Pot(dt)
plant.set_plant_params(plant_params)
plant.set_plant_params(replay_plant_params)
plant.set_ambient_temperature(ambient)
plant.initial(samples[0]['temp_ist'])
# Walk param_events in order, applying each when the sample t passes it.
param_queue = list(param_events)
param_idx = 0
out = []
for s in samples:
# Apply any plant param changes that fall at or before this sample.
while param_idx < len(param_queue) and param_queue[param_idx]['t'] <= s['t']:
p = param_queue[param_idx]['params']
plant.set_plant_params(dict(p, Td=0) if pid_type == 'Smith' else p)
tc.set_model_plant_params(p)
param_idx += 1
plant.process()
sim_temp = plant.get_temperature()
# Controller observes sim_temp but its output doesn't drive the plant.
tc.set_theta_ist(sim_temp)
tc.set_theta_soll(s['temp_soll'])
tc.set_heatrate_soll(heatrate_soll_set)
tc.process()
power_watts = max(0.0, tc.get_power() * max_power * heater_efficiency)
plant.set_power(power_watts)
# Drive plant and Smith model with the real logged power.
real_power_watts = s['power_eff']
tc.set_model_power(real_power_watts)
plant.set_power(real_power_watts)
out.append({
'sim_temp_ist': sim_temp,
'power': tc.get_power(),
'power': max(0.0, tc.get_power()),
'state': tc.state.name,
'heatrate_ist': tc.get_heatrate_ist(),
'heatrate_soll': tc.get_heatrate_soll(),
@@ -140,11 +168,13 @@ def plot_replay(samples, replay, name, max_power, params_label):
ax_rate.set_ylabel('Rate [K/min]')
ax_rate.legend(loc='upper left', fontsize='small')
# Power: log (normalised) vs sim controller output
log_norm = [s['power_set'] / max_power for s in samples]
replay_pwr = [r['power'] for r in replay]
ax_power.plot(t, log_norm, '-b', linewidth=1, label='power (real, norm.)')
ax_power.plot(t, replay_pwr, '-', color='C1', linewidth=1, label='power (sim)')
# Power: power_set (what the controller commanded) and power_eff (what
# the heater actually delivered, quantised to its discrete steps).
# The gap between the two is the heater's rounding/quantisation.
pwr_set = [s['power_set'] / max_power for s in samples]
pwr_eff = [s['power_eff'] / max_power for s in samples]
ax_power.plot(t, pwr_set, '-b', linewidth=1, label='power_set (norm.)')
ax_power.plot(t, pwr_eff, '-', color='C1', linewidth=1, alpha=0.7, label='power_eff (norm.)')
ax_power.set_ylabel('Power [0–1]')
ax_power.set_xlabel('t [s]')
ax_power.legend(loc='upper left', fontsize='small')
@@ -180,8 +210,6 @@ def main():
parser.add_argument('--plant-C', type=float, default=None, metavar='J/kgK', help='Specific heat [J/(kg·K)]')
parser.add_argument('--plant-L', type=float, default=None, metavar='W/kgK', help='Heat loss coeff [W/(kg·K)]')
parser.add_argument('--plant-Td', type=float, default=None, metavar='s', help='Transport delay [s]')
parser.add_argument('--heater-efficiency', type=float, default=1.0, metavar='0-1',
help='Fraction of rated heater power delivered to the plant (default: %(default)s)')
for section, prefix in (('Hold', 'hold'), ('Heat', 'heat'), ('Cool', 'cool')):
for gain in ('kp', 'ki', 'kd', 'kt'):
@@ -224,13 +252,17 @@ def main():
'Td': args.plant_Td if args.plant_Td is not None else (log_plant['Td'] if log_plant else 30.0),
}
samples = log['Samples']
name = log.get('Name', log_path.stem)
pid_params = _apply_gain_overrides(config['TempCtrl'], args)
dt = (samples[1]['t'] - samples[0]['t']) if len(samples) > 1 else 1.0
max_power = _infer_max_power(samples)
rate_soll = args.rate_soll if args.rate_soll is not None else _infer_heatrate_soll_set(samples)
ambient = args.ambient if args.ambient is not None else config.get('ambient_temperature', 20.0)
samples = log['Samples']
param_events = log.get('PlantParams', [])
name = log.get('Name', log_path.stem)
pid_type = config['TempCtrl'].get('pid_type', 'Normal')
pid_params = _apply_gain_overrides(config['TempCtrl'], args)
# 'dt' is the physics timestep the controller/plant were created with —
# distinct from the sample interval (wall-clock) in server logs.
dt = log.get('dt') or ((samples[1]['t'] - samples[0]['t']) if len(samples) > 1 else 1.0)
max_power = _infer_max_power(samples)
rate_soll = args.rate_soll if args.rate_soll is not None else _infer_heatrate_soll_set(samples)
ambient = args.ambient if args.ambient is not None else config.get('ambient_temperature', 20.0)
any_override = any(
getattr(args, '{}_{}'.format(p, g)) is not None
@@ -240,11 +272,12 @@ def main():
config_source = 'log' if log.get('Config') else 'file'
params_label = 'candidate' if any_override else config_source
print('Log: {} ({} samples, dt={:.1f} s)'.format(name, len(samples), dt))
print('Log: {} ({} samples, dt={} s)'.format(name, len(samples), round(dt, 4)))
print('Power: max={:.0f} W'.format(max_power))
print('Rate: heatrate_soll_set={:.2f} K/min'.format(rate_soll))
print('Plant: M={M} kg C={C:.0f} J/kgK L={L} W/kgK Td={Td} s ambient={amb} °C heater_efficiency={eff}'.format(
amb=ambient, eff=args.heater_efficiency, **plant_params))
print('Plant: M={M} kg C={C:.0f} J/kgK L={L} W/kgK Td={Td} s ambient={amb} °C'.format(
amb=ambient, **plant_params))
print('Controller: {}'.format(pid_type))
print('Params: {} ({})'.format('overridden' if any_override else 'from {}'.format(config_source),
'log' if log_plant else 'defaults'))
print()
@@ -252,7 +285,8 @@ def main():
p = pid_params[section]
print(' {}: kp={kp} ki={ki} kd={kd} kt={kt}'.format(section, **p))
replay = run_replay(samples, pid_params, rate_soll, plant_params, ambient, max_power, args.heater_efficiency, dt)
replay = run_replay(samples, pid_type, pid_params, rate_soll, plant_params, param_events,
ambient, dt)
plot_replay(samples, replay, name, max_power, params_label)
plt.show()