brewpi/components/pid/temp_controller_base.py

from components import APid
from components.pid.temp_controller_fsm import TempControllerFsm, States, DEFAULT_THRESHOLDS


class TempControllerBase(TempControllerFsm, APid):

    def __init__(self, dt):
        APid.__init__(self)
        TempControllerFsm.__init__(self, dt)
        self.dt = dt
        self.theta_ist_set = 0
        self.theta_soll_set = 0
        self.heatrate_ist_set = 0
        self.heatrate_soll_set = 1.0
        self.heatrate_soll = 1.0
        self.theta_ist = 0
        self.heatrate_ist = 0
        # None until set_params() is called - mirrors Pid's own
        # params/set_params() (components/pid/pid.py), whose process()
        # already relies on the same "None means not configured yet".
        self.params = None
        self.y = -1
        # Heat-rate pre-filter state (option A) - None until first process() tick
        self.last_theta_ist = None
        self.theta_ist_filtered = None
        self.beta = 0.05

    def set_params(self, params):
        self.params = params
        self.thresholds = {**DEFAULT_THRESHOLDS, **params.get('Thresholds', {})}
        self.beta = params.get('beta', 0.05)
        self.pid_hold.set_params(params['Hold'])
        self.pid_heat.set_params(params['Heat'])
        self.pid_cool.set_params(params['Cool'])

    def _compute_heatrate(self, theta_ist):
        """Pre-filter theta_ist, then differentiate and low-pass to get heatrate_ist.

        Filtering before differentiating (option A) keeps noise that comes in on
        theta_ist from being amplified by the derivative — stirrer-induced
        temperature fluctuations would otherwise produce K/min rate noise of the
        same order as a typical rate_soll, directly corrupting the heat PID error."""
        if self.theta_ist_filtered is None:
            self.theta_ist_filtered = theta_ist
        else:
            self.theta_ist_filtered = (1 - self.beta) * self.theta_ist_filtered + self.beta * theta_ist
        heatrate = 0.0 if self.last_theta_ist is None else \
            (self.theta_ist_filtered - self.last_theta_ist) / self.dt * 60
        alpha = 0.1
        self.heatrate_ist = (1 - alpha) * self.heatrate_ist + alpha * heatrate
        self.last_theta_ist = self.theta_ist_filtered

    def _require_params(self):
        """Called by each subclass's process() before doing anything else -
        without it, a missing set_params() call would otherwise only
        surface as an unhelpful "'NoneType' object is not subscriptable"
        buried inside Pid.process() (components/pid/pid.py), once
        process_pid() below gets to it."""
        if self.params is None:
            raise RuntimeError(
                "{}.process(): PID gains not set - call set_params() first".format(type(self).__name__))

    def is_configured(self):
        """Whether set_params() has been called - lets a caller (e.g.
        TcTask's own process loop) check upfront and skip processing
        gracefully while not yet configured, rather than relying on
        process() raising. Overridden by TempController(Smith) to also
        require its internal model's plant params/ambient."""
        return self.params is not None

    def post_pid(self):
        pass

    def set_ambient_temperature(self, theta_amb):
        pass

    def set_model_plant_params(self, model_params):
        pass

    def set_model_power(self, power):
        pass

    def set_theta_ist(self, value):
        self.theta_ist_set = value
        if self.is_startup:
            self.is_startup = False

    def get_theta_ist(self):
        return self.theta_ist

    def get_theta_ist_set(self):
        """The raw sensor reading, as last pushed via set_theta_ist() -
        unlike get_theta_ist() (self.theta_ist), this is updated the
        instant a reading comes in, regardless of whether process() has
        ever actually run (e.g. before set_params()/a model-based
        controller's plant params have been configured - see SudTask.
        send_forecast(), which needs a real "current temperature" at the
        moment a Sud is loaded, before this controller may have ticked
        even once)."""
        return self.theta_ist_set

    def set_heatrate_ist(self, value):
        self.heatrate_ist_set = value

    def get_heatrate_ist(self):
        return self.heatrate_ist

    def set_theta_soll(self, value):
        self.theta_soll_set = value
        # Recompute the FSM right away against the new target, rather than
        # waiting for the next process() tick - otherwise self.state can
        # still read HOLD from the previous target for up to one tick
        # after a much-further-away one is pushed, which would make
        # is_holding() report "reached" instantly instead of once the gap
        # has actually closed.
        self.process_fsm(self.theta_soll_set - self.theta_ist)

    def get_theta_soll(self):
        return self.theta_soll

    def get_theta_soll_set(self):
        return self.theta_soll_set

    def set_heatrate_soll(self, value):
        self.heatrate_soll_set = value

    def get_heatrate_soll(self):
        return self.heatrate_soll

    def get_heatrate_soll_set(self):
        return self.heatrate_soll_set

    def process_pid(self, theta_err, hold_scale=1.0):
        self.pid_hold.process(theta_err, -self.theta_ist, hold_scale)
        # In HOLD state, clamp pid_hold's output to [0, 1]: a small temperature
        # overshoot makes pid_hold.y go negative, which would invert heatrate_soll
        # and drive pid_heat's power to 0, causing a limit cycle (power on →
        # overshoot → power off → coast down → repeat). Clamping to 0 lets the
        # outer loop reduce the inner setpoint to zero but no further.
        pid_hold_y = self.pid_hold.get_y()
        if self.state == States.HOLD:
            pid_hold_y = max(0.0, pid_hold_y)
        self.heatrate_soll = self.heatrate_soll_set * pid_hold_y
        heatrate_err = self.heatrate_soll - self.heatrate_ist

        # Only the PID actually driving y is advanced - otherwise the
        # inactive one (e.g. pid_heat while COOL has pid_cool driving)
        # would keep silently integrating against a heatrate_err that
        # isn't actually under its control, building a stale windup that
        # causes a discontinuity in y the moment it takes back over.
        if self.state == States.IDLE:
            self.y = 0
        elif self.state == States.COOL:
            self.pid_cool.process(heatrate_err, -self.heatrate_ist)
            self.y = self.pid_cool.get_y()
        else:
            self.pid_heat.process(heatrate_err, -self.heatrate_ist)
            self.y = self.pid_heat.get_y()

        self.post_pid()

    def get_power(self):
        return self.y