Files
brewpi/scripts/demos/sud/demo_sud.py
T
jensandClaude Sonnet 4.6 b33f89a569 Adapt Sud/SudTask/demo to the new flat heat/hold schedule schema
sude/sud_0010.json moved from a list of combined ramp+hold+wait
"Rasten" with global stirrer constants to a flat "schedule" list of
explicit {"type": "heat"|"hold", ...} steps, each carrying its own
stirrer_speed/stirrer_time_on/stirrer_time_off and an optional
user_wait_for_continue (+ user_message).

- components/sud.py: Sud now tracks the current `step` instead of
  `rast`; "heat" steps enter RAMPING (advanced externally via
  temp_reached()), "hold" steps enter HOLDING and count down
  `duration` minutes (0 if omitted). Either step type can pause in
  WAIT_USER via user_wait_for_continue, surfaced through the new
  user_message attribute.
- tasks/sud.py: SudTask only pushes theta_soll/heatrate_soll on
  "heat" steps, converts each step's stirrer_time_on/off into a
  duty_cycle/cycle_time on every step change (replacing the old
  state-based RAMPING/HOLDING stirrer switching), and broadcasts
  user_message changes. Stirring is now fully schedule-driven instead
  of implicitly stopped on WAIT_USER/DONE (DONE still stops it, since
  there's no step left to read settings from).
- scripts/demos/sud/demo_sud.py: mirrors the same step-driven wiring.
- README's Mash schedules section rewritten for the new schema.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-19 19:09:43 +02:00

181 lines
4.7 KiB
Python

import numpy as np
from matplotlib.pyplot import plot, step, figure, subplot, grid, show, legend, yticks
from components.sud import Sud, SudState
from components.pid.temp_controller_smith import TempController
from components.plant.pot import Pot
from components.actor.stirrersim import StirrerSim
from components.actor.heater_sim import HeaterSim
# Mirrors tasks/sud.py's SudTask, but driven synchronously for a demo run.
TEMP_REACHED_TOLERANCE = 0.2
# A real user would click "Confirm" in the GUI; auto-confirm after this many
# simulated seconds so the demo can run to completion unattended.
WAIT_USER_AUTO_CONFIRM_S = 30.0
if __name__ == '__main__':
dt = 1.0
theta_amb = 20
ctrl_params = {
"Hold": {
"kp": 0.4,
"ki": 0.0,
"kd": 0.0,
"kt": 0.0
},
"Heat": {
"kp": 0.08,
"ki": 0.008,
"kd": 0.0,
"kt": 1.5
}
}
plant_params = {
"C" : 4190,
"M" : 20,
"L" : 0.2,
"Td" : 30
}
sud = Sud("sude/sud_0010.json")
ctrl = TempController(dt, ctrl_params, plant_params, theta_amb)
plant = Pot(dt, plant_params, theta_amb)
stirrer = StirrerSim(dt)
heater = HeaterSim({})
# Mirror brewpi.py's data flow: the controller's internal Smith-predictor
# model sees the (continuous) requested power, the real plant sees the
# heater's actual (discretized) output power.
heater.set_on_changed('power_set', ctrl.set_model_power)
heater.set_on_changed('power_eff', plant.set_power)
def apply_stirrer(step):
speed = step.get('stirrer_speed', 0)
on = step.get('stirrer_time_on', 0)
off = step.get('stirrer_time_off', 0)
cycle = on + off
if cycle > 0:
duty = on / cycle
else:
cycle = 1.0
duty = 1.0 if speed > 0 else 0.0
stirrer.set_cycle_time(cycle)
stirrer.set_duty_cycle(duty)
stirrer.set_speed(speed)
def on_step_changed(step):
if step is not None:
if step['type'] == 'heat':
ctrl.set_theta_soll(step['temp'])
ctrl.set_heatrate_soll(step['rate'])
apply_stirrer(step)
def on_state_changed(state):
if state == SudState.DONE:
stirrer.set_duty_cycle(1.0)
stirrer.set_speed(0)
sud.set_on_changed('step', on_step_changed)
sud.set_on_changed('state', on_state_changed)
sud.start()
_t = []
_temp_ist = []
_temp_soll = []
_heatrate_ist = []
_heatrate_soll = []
_sud_state = []
_step_index = []
_stirrer_speed = []
_stirrer_duty = []
_heater_power_set = []
_heater_power_eff = []
wait_user_elapsed = 0.0
t = 0.0
steps = 0
max_steps = 30000
while sud.state != SudState.DONE and steps < max_steps:
plant.process()
ctrl.set_theta_ist(plant.get_temperature())
ctrl.process()
y = ctrl.get_power()
heater.set_power(max(0, 3500*y))
heater.process()
stirrer.process()
if sud.state == SudState.RAMPING:
if abs(ctrl.get_theta_ist() - ctrl.get_theta_soll_set()) < TEMP_REACHED_TOLERANCE:
sud.temp_reached()
if sud.state == SudState.WAIT_USER:
wait_user_elapsed += dt
if wait_user_elapsed >= WAIT_USER_AUTO_CONFIRM_S:
sud.confirm()
wait_user_elapsed = 0.0
else:
wait_user_elapsed = 0.0
sud.tick(dt)
_t.append(t)
_temp_ist.append(ctrl.get_theta_ist())
_temp_soll.append(ctrl.get_theta_soll_set())
_heatrate_ist.append(ctrl.get_heatrate_ist())
_heatrate_soll.append(ctrl.get_heatrate_soll())
_sud_state.append(sud.state.value)
_step_index.append(sud.index)
_stirrer_speed.append(stirrer.speed * stirrer.isOn)
_stirrer_duty.append(stirrer.dutyCycle * 100)
_heater_power_set.append(heater.power_set)
_heater_power_eff.append(heater.power_eff)
t += dt
steps += 1
if sud.state != SudState.DONE:
print("Sud did not finish within {} steps (stuck in {})".format(max_steps, sud.state))
print("Sud finished after {:.0f} s ({:.1f} min), final state = {}".format(t, t/60.0, sud.state))
_t_min = np.array(_t) / 60.0
figure(1)
subplot(3, 1, 1)
plot(_t_min, _temp_ist, '-b', _t_min, _temp_soll, '-r', linewidth=1)
legend(["theta_ist", "theta_soll"])
grid(True)
subplot(3, 1, 2)
plot(_t_min, _heatrate_ist, '-b', _t_min, _heatrate_soll, '-r', linewidth=1)
legend(["heatrate_ist", "heatrate_soll"])
grid(True)
subplot(3, 1, 3)
plot(_t_min, _heater_power_set, '-r', _t_min, _heater_power_eff, '-b', linewidth=1)
legend(["heater power_set", "heater power_eff"])
grid(True)
figure(2)
subplot(3, 1, 1)
step(_t_min, _sud_state, where='post', color='m')
yticks([0, 1, 2, 3, 4], ["IDLE", "RAMPING", "HOLDING", "WAIT_USER", "DONE"])
legend(["Sud state"])
grid(True)
subplot(3, 1, 2)
step(_t_min, _step_index, where='post', color='k')
legend(["Step index"])
grid(True)
subplot(3, 1, 3)
plot(_t_min, _stirrer_speed, '-g', _t_min, _stirrer_duty, '-c', linewidth=1)
legend(["stirrer speed (effective)", "stirrer duty cycle [%]"])
grid(True)
show()
print("End of program")