A Python PID Controller

# Create the PID controller object with specific gains.
controller = PidController(kp, ki, kd)
controller.set_target(target)

# A basic control loop: measure the system error and produce a control signal.
while True:
    t = system.get_time()
    state = system.get_state()
    resp = controller.next(t, state)
    system.control(resp)

class PidController():
    """A classical PID controller which maintains state between calls.

    This class is intended to be integrated into an external control loop. It
    remembers enough of the state history to compute integral and derivative
    terms, and produces a combined control signal with the given gains.
    """

    def __init__(self,
                 kp: float,
                 ki: float,
                 kd: float,
                 target: float,
                 initial_state: float,
                 t_0: int) -> None:
        """Create a PID controller with the specified gains and initial state.

        Parameters
        ----------
        kp, ki, kd    : The PID control gains.
        target        : The desired system state, also called a "setpoint".
        initial_state : The starting state of the system.
        t_0           : The starting time.
        """
        # Gains for the proportional, integral, and derivative terms.
        self._kp: float = kp
        self._ki: float = ki
        self._kd: float = kd

        # The target state which the controller tries to maintain.
        self._target: float = target

        # Tracks the integrated error over time. This starts at 0 as no time has passed.
        self._accumulated_error: float = 0.0
        # Tracks the previous sample's error to compute derivative term.
        self._last_error: float = initial_state - target
        # Tracks the previous sample time point for computing the d_t value used in I and D terms.
        self._last_t: int = t_0

def next(self, t: int, state: float) -> float:
    """Incorporate a sample of the state at time t and produce a control value.

    Because the controller is stateful, calls to this method should be
    monotonic - that is, subsequent calls should not go backwards in time.

    Parameters
    ----------
    t     : The time at which the sample was taken.
    state : The system state at time t.
    """
    error = state - self._target
    d_t = (t - self._last_t)
    p = self._proportional(error)
    i = self._integral(d_t, error)
    d = self._derivative(d_t, error)
    self._last_t = t
    self._last_error = error
    return p + i + d

def _proportional(self, error: float) -> float:
    return self._kp * error


def _integral(self, d_t: float, error: float) -> float:
    # The constant part of the error.
    base_error = min(error, self._last_error) * d_t
    # Adjust by adding a little triangle on the constant part.
    error_adj = abs(error - self._last_error) * d_t / 2.0
    self._accumulated_error += base_error + error_adj
    return self._ki * self._accumulated_error


def _derivative(self, d_t: float, error: float) -> float:
    d_e = (error - self._last_error)
    if d_t > 0:
        return self._kd * (d_e / d_t)
    else:
        return 0

class ThermalMass():

    def __init__(self, mass: float, cp: float, start_temp: float):
        """Create a thermal mass with the given properties.

        Parameters
        ----------
        mass       : The body's mass in kilograms.
        cp         : The body's specific heat capacity in J/K/kg.
        start_temp : The body's starting temperature, in Kelvin.
        """
        self._mass = mass
        self._cp = cp
        self._temp = start_temp


    def current_temperature(self) -> float:
        """Get the current temperature of the body."""
        return self._temp


    def update(self, heat_in: float, heat_out: float):
        """Update the body state given a certain heat input and output.

        Parameters
        ----------
        heat_in  : The amount of heat added to the body in Joules.
        heat_out : The amount of heat removed from the body in Joules.
        """
        d_q = heat_in - heat_out
        # We find the change in temperature by taking the change in heat and scaling it by the
        # body's heat capacity.
        self._temp += d_q / (self._mass * self._cp)

from typing import Callable, Iterable, Tuple, List

class ThermalSystem():

    def __init__(self,
                 mass: ThermalMass,
                 in_flow: Callable[[int], float],
                 out_flow: Callable[[int, float], float]):
        """Create a simple thermal system with the given mass and flow functions.

        Parameters
        ----------
        mass     : A thermal mass to use as the system's body of interest.
        in_flow  : The heat flow rate (in J/s) into the body as a function of time.
        out_flow : The heat flow rate (in J/s) out of the body as a function of time and
                   temperature.
        """
        self._mass = mass
        self._in_flow = in_flow
        self._out_flow = out_flow


    def simulate(self, timesteps: Iterable[int]) -> List[Tuple[int, float, float, float]]:
        """Simulate the system behavior over time, and give a state trace.

        This function runs the system state forward using the given timestep
        sequence. At each time point it computes input and output heat flows,
        updates the thermal mass, and records a datapoint for the output trace.
        The trace consists of a sequence of tuples, one for each time point,
        containing the time value, the temperature, and the instantaneous heat
        flow in and out values at that time.

        Parameters
        ----------
        timesteps : A monotonically increasing sequence of timepoints.
        """
        trace = []
        it = timesteps.__iter__()
        last_t = it.__next__()
        temp = self._mass.current_temperature()
        flow_in = self._in_flow(last_t)
        flow_out = self._out_flow(last_t, temp)

        trace.append((last_t, temp, flow_in, flow_out))

        for t in it:
            dt = t - last_t
            last_t = t

            heat_in = flow_in * dt
            heat_out = flow_out * dt

            self._mass.update(heat_in, heat_out)

            temp = self._mass.current_temperature()
            flow_in = self._in_flow(t)
            flow_out = self._out_flow(t, temp)
            trace.append((t, temp, flow_in, flow_out))

        return trace

import fixture
import thermal

# Our mass is approximately equivalent to a liter of water at room temperature.
mass = thermal.ThermalMass(1.0, 4000, 0)
in_flow = lambda t: 100.0 # 1 kJ/s
out_flow = lambda t, s: 100.0 # 1 kJ/s
system = thermal.ThermalSystem(mass, in_flow, out_flow)
f = fixture.GraphicalFixture("equilibrium", system)
f.simulate(range(0, 100))

import fixture
import thermal

# Our mass is approximately equivalent to a liter of water at room temperature.
mass = thermal.ThermalMass(1.0, 4000, 0)
in_flow = lambda t: 200.0 # 1 kJ/s
out_flow = lambda t, s: 5.0 # 0.5 kJ/s
system = thermal.ThermalSystem(mass, in_flow, out_flow)
f = fixture.GraphicalFixture("runaway", system)
f.simulate(range(0, 100))

import fixture
import thermal
import random

random.seed(a=1613837227)

# Our mass is approximately equivalent to a liter of water at room temperature.
mass = thermal.ThermalMass(1.0, 4000, 0)
in_flow = lambda t: 100 * random.random()
out_flow = lambda t, s: 100 * random.random()
system = thermal.ThermalSystem(mass, in_flow, out_flow)
f = fixture.GraphicalFixture("random", system)
f.simulate(range(0, 100), y_scales=(20, 1, 20))