brewpi/components/pid/temp_controller_fsm.py

import enum
from components.pid.pid import Pid


DEFAULT_THRESHOLDS = {
    "HoldHeat": 1.0,
    # HoldCool used to be 0.1 ("eagerly give up and coast" made sense back
    # when COOL meant nothing more than going idle - it's an active state
    # with its own PID now, so a threshold this tight relative to a real
    # heater's discrete power steps/sensor noise causes the system to
    # chatter in and out of it every tick, resetting both pid_hold's and
    # pid_cool's integrators each time and never letting either actually
    # converge. Symmetric with the others instead.
    "HoldCool": 1.0,
    "HeatHold": 1.0,
    "HeatCool": 1.0,
    "CoolHold": 1.0,
    "CoolHeat": 1.0,
}


class States(enum.Enum):
    INIT = -1,
    IDLE = 0,
    HEAT = 1,
    HOLD = 2,
    COOL = 3


class TempControllerFsm:

    def __init__(self, dt):
        self.pid_hold = Pid(dt)
        self.pid_heat = Pid(dt)
        # Separate gains for ramping down (negative diff) - the actuator
        # (e.g. a heat-only Pot/heater) is responsible for clamping the
        # resulting negative power to whatever it's actually capable of;
        # the controller itself no longer assumes "can't cool" == "must go
        # idle".
        self.pid_cool = Pid(dt)
        self.thresholds = None
        self.state = States.INIT
        self.is_startup = True
        # Master on/off switch: while disabled, the FSM is held in IDLE and
        # the controller tries not to drive the heater at all (output
        # forced to 0) - off by default, enabled either by the user
        # (manual mode) or by SudTask for the duration of a run.
        self.enabled = False

    def on_state_entered(self, state):
        pass

    def set_enabled(self, value):
        self.enabled = value

    def is_holding(self):
        """Whether the FSM currently considers theta_ist close enough to
        theta_soll_set to no longer be actively heating/cooling toward
        it - the single source of truth for "is a ramp toward the
        current target done" (see tasks/sud.py's SudTask)."""
        return self.state == States.HOLD

    def process_fsm(self, diff):
        state_next = self.state
        if self.state == States.INIT:
            # Wait for a real sensor reading before acting on anything,
            # regardless of enabled - avoids reacting to the bogus
            # theta_ist=0 default.
            if not self.is_startup:
                state_next = States.IDLE
        elif not self.enabled:
            state_next = States.IDLE
        elif self.state == States.IDLE:
            # Just (re-)enabled - resolve straight to HEAT/COOL/HOLD against
            # the real gap, the same threshold check the HOLD branch below
            # uses, rather than landing in HOLD and waiting for the next
            # tick to correct it: is_holding() is read synchronously within
            # this very same call chain (tasks/sud.py's SudTask.on_step_
            # changed() pushes the new step's setpoint via set_theta_soll(),
            # which calls this method directly), so an unconditional HOLD
            # here gets mistaken for "ramp already reached" and can finish
            # a freshly (re-)started step instantly - see SudTask.
            # on_process()'s is_holding() check. pid_heat/pid_cool were
            # frozen (see process_pid()) and possibly stale for as long as
            # we were disabled - start whichever one matters clean rather
            # than resuming wherever it last left off.
            self.pid_hold.reset()
            self.pid_heat.reset()
            self.pid_cool.reset()
            if diff >= self.thresholds['HoldHeat']:
                state_next = States.HEAT
            elif diff <= -self.thresholds['HoldCool']:
                state_next = States.COOL
            else:
                state_next = States.HOLD
        elif self.state == States.HOLD:
            if diff >= self.thresholds['HoldHeat']:
                state_next = States.HEAT
                self.pid_heat.reset()
            elif diff <= -self.thresholds['HoldCool']:
                state_next = States.COOL
                self.pid_cool.reset()
        elif self.state == States.HEAT:
            if diff <= -self.thresholds['HeatCool']:
                state_next = States.COOL
                self.pid_cool.reset()
            elif diff <= self.thresholds['HeatHold']:
                state_next = States.HOLD
                self.pid_hold.reset()
        elif self.state == States.COOL:
            if diff >= self.thresholds['CoolHeat']:
                state_next = States.HEAT
                self.pid_heat.reset()
            elif diff >= -self.thresholds['CoolHold']:
                state_next = States.HOLD
                self.pid_hold.reset()
                # pid_heat was frozen during COOL (see process_pid()) -
                # resume it clean rather than from whatever it last held
                # before COOL took over, which by now may be a stale fit
                # for a completely different part of the curve.
                self.pid_heat.reset()

        if state_next != self.state:
            self.state = state_next
            self.on_state_entered(state_next)