Hot water tank model¶

This model demonstrates how Plugboard can be used to connect physics-based models. We'll build a component to simulate the temperature of an insulated hot-water tank, along with another to control the tank's heating element.

In [ ]:

import typing as _t

from plugboard.connector import AsyncioConnector
from plugboard.component import Component
from plugboard.component import IOController as IO
from plugboard.schemas import ComponentArgsDict, ConnectorSpec
from plugboard.process import LocalProcess
from plugboard.library import FileWriter

import typing as _t from plugboard.connector import AsyncioConnector from plugboard.component import Component from plugboard.component import IOController as IO from plugboard.schemas import ComponentArgsDict, ConnectorSpec from plugboard.process import LocalProcess from plugboard.library import FileWriter

In [ ]:

WATER_SPECIFIC_HEAT = 4186  # J/(kg°C)
WATER_DENSITY = 1.0  # kg/L

WATER_SPECIFIC_HEAT = 4186 # J/(kg°C) WATER_DENSITY = 1.0 # kg/L

In [ ]:

class HotWaterTank(Component):
    """This component represents an insulated hot water tank with an on/off heating element."""

    io = IO(inputs=["heating_element", "prev_temperature"], outputs=["temperature"])

    def __init__(
        self,
        volume: float,
        heater_power: float,
        insulation_r: float,
        ambient_temp: float,
        delta_t: float = 60,
        **kwargs: _t.Unpack[ComponentArgsDict],
    ) -> None:
        super().__init__(**kwargs)
        # Set the initial running total to 0
        self._heater_power_watts = heater_power * 1000  # W
        self._water_mass = volume * WATER_DENSITY  # kg
        self._heat_capacity = self._water_mass * WATER_SPECIFIC_HEAT  # J/°C
        self._delta_t = delta_t  # s
        self._insulation_r = insulation_r  # °C/W
        self._ambient_temp = ambient_temp  # °C

    async def step(self) -> None:
        # Apply heater power to the water
        self.temperature = self.prev_temperature
        if self.heating_element:
            self.temperature += self._heater_power_watts * self._delta_t / self._heat_capacity
        # Apply heat loss to the environment
        heat_loss = (
            (self.prev_temperature - self._ambient_temp) / self._insulation_r * self._delta_t
        )
        self.temperature -= heat_loss / self._heat_capacity

class HotWaterTank(Component): """This component represents an insulated hot water tank with an on/off heating element.""" io = IO(inputs=["heating_element", "prev_temperature"], outputs=["temperature"]) def __init__( self, volume: float, heater_power: float, insulation_r: float, ambient_temp: float, delta_t: float = 60, **kwargs: _t.Unpack[ComponentArgsDict], ) -> None: super().__init__(**kwargs) # Set the initial running total to 0 self._heater_power_watts = heater_power * 1000 # W self._water_mass = volume * WATER_DENSITY # kg self._heat_capacity = self._water_mass * WATER_SPECIFIC_HEAT # J/°C self._delta_t = delta_t # s self._insulation_r = insulation_r # °C/W self._ambient_temp = ambient_temp # °C async def step(self) -> None: # Apply heater power to the water self.temperature = self.prev_temperature if self.heating_element: self.temperature += self._heater_power_watts * self._delta_t / self._heat_capacity # Apply heat loss to the environment heat_loss = ( (self.prev_temperature - self._ambient_temp) / self._insulation_r * self._delta_t ) self.temperature -= heat_loss / self._heat_capacity

In [ ]:

class ThermostatController(Component):
    """This component represents a thermostat with hysteresis."""

    io = IO(inputs=["setpoint", "temperature"], outputs=["heating_element"])

    def __init__(self, hysteresis: float, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._hysteresis = hysteresis

    async def step(self) -> None:
        if self.temperature < self.setpoint - self._hysteresis:
            self.heating_element = True
        elif self.temperature > self.setpoint + self._hysteresis:
            self.heating_element = False

class ThermostatController(Component): """This component represents a thermostat with hysteresis.""" io = IO(inputs=["setpoint", "temperature"], outputs=["heating_element"]) def __init__(self, hysteresis: float, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: super().__init__(**kwargs) self._hysteresis = hysteresis async def step(self) -> None: if self.temperature < self.setpoint - self._hysteresis: self.heating_element = True elif self.temperature > self.setpoint + self._hysteresis: self.heating_element = False

We'll use a Constant value to represent the setpoint and trigger the rest of the model.

In [ ]:

class Constant(Component):
    """This component represents a constant value."""

    io = IO(outputs=["value"])

    def __init__(
        self, value: float, n_iterations: int, **kwargs: _t.Unpack[ComponentArgsDict]
    ) -> None:
        super().__init__(**kwargs)
        self.value = value
        self._remaining_iterations = n_iterations

    async def step(self) -> None:
        self._remaining_iterations -= 1
        if self._remaining_iterations <= 0:
            await self.io.close()

class Constant(Component): """This component represents a constant value.""" io = IO(outputs=["value"]) def __init__( self, value: float, n_iterations: int, **kwargs: _t.Unpack[ComponentArgsDict] ) -> None: super().__init__(**kwargs) self.value = value self._remaining_iterations = n_iterations async def step(self) -> None: self._remaining_iterations -= 1 if self._remaining_iterations <= 0: await self.io.close()

In [ ]:

setpoint = Constant(name="setpoint", value=60, n_iterations=24 * 60)
tank = HotWaterTank(
    name="tank",
    initial_values={"prev_temperature": [58], "heating_element": [False]},
    volume=150,
    heater_power=1.1,
    insulation_r=0.9,
    ambient_temp=20,
)
thermostat = ThermostatController(name="controller", hysteresis=1)
save = FileWriter(
    name="save", path="temperature.csv", field_names=["heater", "temperature", "setpoint"]
)

setpoint = Constant(name="setpoint", value=60, n_iterations=24 * 60) tank = HotWaterTank( name="tank", initial_values={"prev_temperature": [58], "heating_element": [False]}, volume=150, heater_power=1.1, insulation_r=0.9, ambient_temp=20, ) thermostat = ThermostatController(name="controller", hysteresis=1) save = FileWriter( name="save", path="temperature.csv", field_names=["heater", "temperature", "setpoint"] )

Now connect the components together in a LocalProcess.

In [ ]:

process = LocalProcess(
    components=[setpoint, tank, thermostat, save],
    connectors=[
        # Connect setpoint to controller
        AsyncioConnector(
            spec=ConnectorSpec(source="setpoint.value", target="controller.setpoint"),
        ),
        # Connect controller to tank
        AsyncioConnector(
            spec=ConnectorSpec(source="controller.heating_element", target="tank.heating_element"),
        ),
        # Connect tank to controller
        AsyncioConnector(
            spec=ConnectorSpec(source="tank.temperature", target="controller.temperature"),
        ),
        # Connect tank to itself to save the previous temperature
        AsyncioConnector(
            spec=ConnectorSpec(source="tank.temperature", target="tank.prev_temperature"),
        ),
        # Connect tank, controller and setpoint to save
        AsyncioConnector(
            spec=ConnectorSpec(source="tank.temperature", target="save.temperature"),
        ),
        AsyncioConnector(
            spec=ConnectorSpec(source="controller.heating_element", target="save.heater"),
        ),
        AsyncioConnector(
            spec=ConnectorSpec(source="setpoint.value", target="save.setpoint"),
        ),
    ],
)

process = LocalProcess( components=[setpoint, tank, thermostat, save], connectors=[ # Connect setpoint to controller AsyncioConnector( spec=ConnectorSpec(source="setpoint.value", target="controller.setpoint"), ), # Connect controller to tank AsyncioConnector( spec=ConnectorSpec(source="controller.heating_element", target="tank.heating_element"), ), # Connect tank to controller AsyncioConnector( spec=ConnectorSpec(source="tank.temperature", target="controller.temperature"), ), # Connect tank to itself to save the previous temperature AsyncioConnector( spec=ConnectorSpec(source="tank.temperature", target="tank.prev_temperature"), ), # Connect tank, controller and setpoint to save AsyncioConnector( spec=ConnectorSpec(source="tank.temperature", target="save.temperature"), ), AsyncioConnector( spec=ConnectorSpec(source="controller.heating_element", target="save.heater"), ), AsyncioConnector( spec=ConnectorSpec(source="setpoint.value", target="save.setpoint"), ), ], )

Now we can initialise and run the simulation.

In [ ]:

async with process:
    await process.run()

async with process: await process.run()

Finally check we have the output data saved in temperature.csv.

In [ ]:

try:
    import pandas as pd

    fig = pd.read_csv("temperature.csv").plot(
        backend="plotly",
        y=["temperature", "setpoint"],
        title="Temperature vs. Setpoint",
        labels={"index": "Time (min)", "value": "Temperature (°C)"},
    )
except (ImportError, ValueError):
    print("Please install plotly to run this cell.")
    fig = None
fig

try: import pandas as pd fig = pd.read_csv("temperature.csv").plot( backend="plotly", y=["temperature", "setpoint"], title="Temperature vs. Setpoint", labels={"index": "Time (min)", "value": "Temperature (°C)"}, ) except (ImportError, ValueError): print("Please install plotly to run this cell.") fig = None fig