Skip to content

Using the core in Python

Scenarios and domain libraries compile down to the core: operators wired into a graph, and an engine that steps it. The core can also be used directly, for plants that no template describes or for experiments that drive the engine step by step.

Building a graph

import math
from homeostat.core import Engine, Event, Graph
from homeostat.core.operators import FOPDT, OU, PID, Lag, Schedule, Sum

g = Graph()
g.add(Schedule("L.sp", value=50.0))
g.add(Schedule("L.mode", value=1.0))                  # AUTO
g.add(Schedule("L.man", value=math.nan))              # no manual output yet
g.add(
    PID("L.pid", Kc=0.8, Ti=5.0),
    inputs={"pv": "m", "sp": "L.sp", "mode": "L.mode", "man_out": "L.man"},
    outputs={"out": "L.out"},
    meta={"loop": "L"},
)
g.add(FOPDT("plant", K=1.0, tau=5.0, theta=1.0), inputs={"u": "L.out"})
g.add(OU("d", std=2.0, tau=60.0))                     # a disturbance
g.add(Sum("y", weights=[1.0, 1.0]), inputs={"in0": "plant", "in1": "d"})
g.add(Lag("m", tau=1.0), inputs={"u": "y"})           # the instrument the controller reads
g.describe("y", unit="m3/h", doc="true flow")
  • add(operator, inputs, outputs, meta) adds an operator. Its output signal has the operator's name unless outputs renames it; inputs maps each input port to a signal. meta={"loop": "L"} names the loop a controller belongs to.
  • describe(signal, unit, doc) adds a unit and a description to the signal table.
  • modulate(path, signal, "scale" | "shift") makes a parameter follow a signal: its effective value is its base value times every scale signal, plus every shift signal. Degradations use it, for example fouling scales a heating gain.

The operator reference lists every operator with its equations, ports and parameters.

Compiling

compiled = g.compile(1.0, promote=["plant.K"])   # dt = 1 s; plant.K will change during the run
compiled.loops     # [Loop(name='L', controller='L.pid', regulated='y', measured='m', reference='L.sp', ...)]
compiled.roles     # {'y': [{'role': 'regulated', 'loop': 'L'}], 'd': [{'role': 'exogenous', ...}], ...}

Compiling checks the graph and fails with every problem at once: two writers for one signal, a missing input, an unknown signal (with a suggestion), an algebraic loop (with its cycle), a controller that reads a true value instead of a measurement. It then finds the control loops and every signal's roles, and orders the operators for the step.

promote lists parameters that events will change. batch sets the number of lanes, and lane_values gives parameters a different value per lane.

Running

engine = Engine(compiled, seed=1)
engine.schedule(Event(at=600, target="L.sp", value=60.0), Event(at=1200, target="plant.K", value=0.8))
engine.initialize("steady")
engine.run(1800)
run = engine.result()

The engine is resumable: run(until) advances the clock to until seconds, and more events can be scheduled between runs.

engine.schedule(Event(at=2400, target="L.mode", value=0.0), Event(at=2400, target="L.man", value=40.0))
engine.run(3000)
run = engine.result()   # everything recorded since the start

initialize takes "steady", "none" or {"burn_in": seconds}, once, before the first run. fork() copies an engine, random streams included, so two copies can continue differently from the same point.

An Event has a time at in seconds, a target, a value, an action (set, shift, scale, or reset to return an operator to its initial state), a profile (step or ramp over over seconds), a label (planned or fault:<name>), and optionally the lanes it applies to. Events are checked when they are scheduled.

From a scenario to the core

homeostat.prepare(scenario) returns the pieces a scenario compiles to: prepared.graph, prepared.compiled, prepared.events and the domain library's prepared.instances. prepared.engine() gives an engine with the scenario's events scheduled, not yet initialized, and prepared.run(engine) runs it for the scenario's duration.