Water pump station: what is inside

A town's pump station: a river intake, two pumps and a PLC that holds every interlock. HEART reads the PLC's Modbus map, read only, and tells a person. HEART runs on an edge gateway in the station's control panel, next to the PLC. Never in the PLC, never in the safety relays.

Specifications it follows

• Part 2: Autonomy Levels (A1, Minimum Tending)
• Part 3: Information Model (messages, fault codes)
• Part 4: Connections (ports, the API, the fallback port)
• Part 6: Human Control and Safety (the gate, proof of presence, staying out of the safety chain)
• Signal Tables (every signal declared once)
• HEART Hub profile (claim codes, reports, Flatline)
• API Standard (the Valve, Section 4; the gate, Section 6.3)

World  ──►  Station (plant + PLC)  ──(Modbus map, read only)──►  Heart  ──►  local HEART API   (127.0.0.1)
river, town   pumps, interlocks, duty                             watches     └►  HEART Hub  (outbound only)

Run it yourself, and register it on HEART Hub

1. Download README.md, pumpStationSimulator.py, test_e2e.py.
2. Run python3 pumpStationSimulator.py --scenario all. It prints a device ID and a claim code.
3. Sign in, open Register a device, enter them, and watch your own copy on My devices.

Now do it for your own device and monitor it here: https://heartcortex.ai/monitor. Start from a template, follow the HEART specifications, then register your device with its claim code.

Full instructions: README · case study: Water pump station.

The code: pumpStationSimulator.py

737 lines of Python, standard library only. The live page above replays the same design in the site's own code; this file is the version you run.

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Template: heart/embeddedstandard/pump-station-simulator/v1.2611.2
"""pumpStationSimulator.py: a simulated water pump station with a HEART beside its PLC, reporting to HEART Monitor.

A DEMONSTRATION ONLY. It is not a control system, simulates no real station or utility, and must never be connected
to real plant. Its numbers are plausible, not engineering data. See README.md beside this file, and the case study
guides/Water-Pump-Station.md.

What it does, in one process and with only the Python standard library:

  * a simulated WORLD: a river intake that fills the station's sump, and a town whose water demand follows a daily
    curve (low at night, peaks at breakfast and in the evening), drawing from a reservoir on the hill;
  * a simulated STATION run by a simulated PLC: two pumps (duty and standby) that lift water from the sump to the
    reservoir. The PLC starts and stops them on reservoir level, rotates the duty pump, and holds its own interlocks:
    dry-run protection from a low-level float switch, a high discharge pressure trip, and a motor thermal trip. Its
    values sit in a Modbus register map (0-based addresses), as a real PLC's would;
  * a HEART that only WATCHES the station, as an edge gateway beside the PLC would. It reads the PLC's registers
    (here from the simulation, as if over Modbus TCP; it has no write function at all), notices problems, raises
    alerts with standard fault codes, explains them, tells a person, serves the HEART API on this computer, and
    reports public health (status and fault codes only) to a HEART Monitor site.

The PLC runs the station and its interlocks. HEART is out of the safety chain: if it crashes, the pumps carry on.

Run it:

  python3 pumpStationSimulator.py                       # report to https://heartcortex.ai
  python3 pumpStationSimulator.py --scenario all        # also stage each fault, one after another
  python3 pumpStationSimulator.py --no-monitor          # local only: http://127.0.0.1:43305/heart/v1/pulse
  python3 pumpStationSimulator.py --monitor-url http://127.0.0.1:8787   # a local monitor site (heart-monitor/test/serve.mjs)

At first start it prints a device ID and a claim code. Register them at https://heartcortex.ai/register, signed in
with your email, then watch the device at https://heartcortex.ai/device/<ID>. The ID, claim code and key are kept
in pump-station-monitor.json (mode 600), so the same simulated station comes back after a restart.

Scenarios (--scenario, comma-separated): stuck, link, cavitation, hot, trip, power, all. Each fault starts at a set
time, lasts a while, and is then "fixed by the maintenance crew", so you can watch an alert open, escalate and
resolve.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. Levels,
flows, pressures, motor readings and the PLC's event log stay local (internal). Nothing here is personal.
"""
import argparse
import hashlib
import json
import math
import os
import random
import secrets
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SYSTEM_ID = "org.heartaispec.demo.pump-station"
NAME = "Water Pump Station Simulator (demo, not a control system)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/pump-station-simulator/v1.2611.2"
PULSE_SEC = 30
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

# The PLC's settings. They live in the PLC; only the utility's engineers change them, with the PLC's own tools.
START_LEVEL, STOP_LEVEL = 3.8, 4.8       # reservoir level, m: start the duty pump below, stop it above
ASSIST_LEVEL = 2.0                        # reservoir level, m: below this the standby pump assists
FLOAT_LOW, FLOAT_RESET = 0.6, 1.0         # sump level, m: the hardwired float switch stops the pumps (dry run)
HIGH_PRESSURE_TRIP = 8.0                  # bar: the PLC trips the station and locks it out until a local reset
THERMAL_TRIP = 120.0                      # Cel, motor winding: the PLC trips that pump
RESTART_DELAY = 20                        # s after mains returns, before the PLC restarts a pump
RESERVOIR_AREA, SUMP_AREA = 1500.0, 40.0  # m2
RESERVOIR_LOW = 1.0                       # m: below this the town's pressure suffers

# What HEART expects, from the pumps' commissioning test. Its own thresholds, not the PLC's.
EXPECTED_FLOW = {0: 0.0, 1: 300.0, 2: 520.0}   # m3/h with 0, 1 or 2 pumps running
VIBRATION_ALARM = 7.1                     # mm/s RMS (ISO 10816-3 zone boundary for a machine of this size)
HOT_WARNING = 100.0                       # Cel: well below the PLC's trip, so a person hears first
STUCK_SEC = 90                            # s with not one change in a live analogue reading


def log(msg):
    print(time.strftime("%H:%M:%S ") + msg, file=sys.stderr, flush=True)


def utc(t=None):
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))


def toward(value, target, dt, tau):
    """First-order lag: move value toward target with time constant tau (stable for any dt)."""
    return value + (target - value) * min(1.0, dt / tau)


# ------------------------------------------------------------------------------------------------ the simulation

class Timer:
    """A value that changes on its own random timer (in simulated seconds)."""

    def __init__(self, rng, lo, hi):
        self.rng, self.lo, self.hi, self.at = rng, lo, hi, 0.0

    def due(self, t):
        if t < self.at:
            return False
        self.at = t + self.rng.uniform(self.lo, self.hi)
        return True


class Episode:
    """Something that happens now and then, for a while: next start after a random gap, lasting a random time."""

    def __init__(self, rng, gap, length):
        self.rng, self.gap, self.length = rng, gap, length
        self.start = rng.uniform(*gap)
        self.end = 0.0

    def active(self, t):
        if t >= self.start and self.end <= self.start:
            self.end = self.start + self.rng.uniform(*self.length)
        if self.end > self.start and t >= self.end:
            self.start = t + self.rng.uniform(*self.gap)
        return self.start <= t < self.end


class World:
    """The river intake and the town. The station cannot change either."""

    def __init__(self, rng):
        self.rng = rng
        self.river, self.factor, self.demand = 4.5, 1.0, 150.0      # river head, m; demand, m3/h
        self.t_river, self.t_factor = Timer(rng, 300, 1800), Timer(rng, 30, 120)

    def step(self, t, wall):
        lt = time.localtime(wall)
        h = lt.tm_hour + lt.tm_min / 60
        # A town's daily curve: about 80 m3/h at night, a breakfast peak and a longer evening peak.
        curve = 80 + 170 * math.exp(-((h - 7.5) / 1.4) ** 2) + 140 * math.exp(-((h - 19.0) / 2.0) ** 2) \
            + 60 * (6 < h < 22)
        if self.t_factor.due(t):
            self.factor = min(1.2, max(0.8, self.factor + self.rng.gauss(0, 0.03)))
        if self.t_river.due(t):
            self.river = min(4.8, max(4.2, self.river + self.rng.gauss(0, 0.05)))
        self.demand = curve * self.factor


class Pump:
    def __init__(self, rng, n):
        self.n, self.rng = n, rng
        self.running, self.thermal_trip = False, False
        self.hours = rng.uniform(8000, 21000)
        self.temp, self.vib, self.current, self.flow = (68.0 if n == 1 else 26.0), 0.1, 0.0, 0.0


class Station:
    """The simulated plant and its PLC. The PLC's logic and interlocks live here, and nowhere else. The HEART reads
    its register map through read_input_registers() and read_discrete_inputs(); nothing lets the HEART write."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.sump, self.reservoir = 3.0, rng.uniform(3.2, 3.7)
        self.pumps = [Pump(rng, 1), Pump(rng, 2)]
        self.duty, self.call, self.hp_trip, self.dry_run, self.was_idle = 0, False, False, False, False
        self.mains, self.mains_back_at, self.ups = True, -1e9, 100.0
        self.pressure, self.flow, self.outflow, self.voltage = 3.6, 0.0, 150.0, 400.0
        self.sump_reading, self.link_down, self.valve_closed = self.sump, False, False
        self.hot_pump = None
        self.t_v = Timer(rng, 2, 10)
        self.link_ep = Episode(rng, (3 * 3600, 8 * 3600), (60, 600))        # the telemetry router drops out
        self.mains_ep = Episode(rng, (12 * 3600, 48 * 3600), (20, 300))     # a grid fault
        self.events = []                                                     # the PLC's event log, newest last
        self.regs = {"input": [0] * 20, "discrete": [0] * 8}

    def staged(self, name, t, start, length):
        return name in self.sc and start <= t < start + length

    def event(self, name, **kw):
        self.events.append({"time": utc(), "name": name, **kw})
        self.events = self.events[-200:]

    def step(self, t, dt, world):
        # Staged faults, for the demo. Each lasts a while and is then "fixed by the maintenance crew".
        stuck = self.staged("stuck", t, 60, 180)
        self.link_down = self.link_ep.active(t) or self.staged("link", t, 270, 300)
        cavitation = self.staged("cavitation", t, 600, 150)
        fan_blocked = self.staged("hot", t, 780, 240)           # one motor's cooling fan is blocked
        if fan_blocked and self.hot_pump is None:
            self.hot_pump = self.pumps[self.duty]
        if "hot" in self.sc and t - dt < 1020 <= t:
            for p in self.pumps:                                 # the crew clears the fan and resets the trip
                if p.thermal_trip:
                    p.thermal_trip = False
                    self.event("thermal-trip.reset", pump=p.n)
        self.valve_closed = self.staged("trip", t, 1050, 120)
        if "trip" in self.sc and t - dt < 1170 <= t and self.hp_trip:
            self.hp_trip = False                                 # an operator opens the valve and resets locally
            self.event("high-pressure-trip.reset")
        mains = not (self.mains_ep.active(t) or self.staged("power", t, 1200, 120))

        # --- the PLC scan: interlocks first, then duty and standby
        if mains != self.mains:
            self.mains = mains
            self.event("mains.restored" if mains else "mains.lost")
            if mains:
                self.mains_back_at = t
        if self.sump < FLOAT_LOW and not self.dry_run:
            self.dry_run = True
            self.event("dry-run.stop")
        elif self.dry_run and self.sump > FLOAT_RESET:
            self.dry_run = False
            self.event("dry-run.cleared")
        if self.pressure > HIGH_PRESSURE_TRIP and not self.hp_trip:
            self.hp_trip = True
            self.event("high-pressure-trip", bar=round(self.pressure, 2))
        for p in self.pumps:
            if p.temp > THERMAL_TRIP and not p.thermal_trip:
                p.thermal_trip = True
                self.event("thermal-trip", pump=p.n, temp=round(p.temp, 1))
        if self.reservoir < START_LEVEL:
            self.call = True
        elif self.reservoir > STOP_LEVEL:
            self.call = False
        ready = self.mains and t - self.mains_back_at >= RESTART_DELAY and not self.hp_trip and not self.dry_run
        avail = [ready and not p.thermal_trip for p in self.pumps]
        if not avail[self.duty] and avail[1 - self.duty]:
            self.duty = 1 - self.duty
            self.event("duty.changed", pump=self.duty + 1, reason="duty pump unavailable")
        need = 0 if not self.call else 2 if self.reservoir < ASSIST_LEVEL else 1
        if need and not any(p.running for p in self.pumps) and self.was_idle and avail[1 - self.duty]:
            self.duty = 1 - self.duty                            # duty rotation: the other pump leads each new cycle
            self.event("duty.changed", pump=self.duty + 1, reason="rotation")
        self.was_idle = need == 0
        order = [self.duty, 1 - self.duty]
        want = [False, False]
        for i in order[:need]:
            want[i] = avail[i]
        if need == 1 and not avail[self.duty] and avail[1 - self.duty]:
            want[1 - self.duty] = True
        for i, p in enumerate(self.pumps):
            if want[i] != p.running:
                p.running = want[i]
                self.event("pump.started" if p.running else "pump.stopped", pump=p.n)

        # --- hydraulics and motors
        running = [p for p in self.pumps if p.running]
        n = len(running)
        q = 0.0 if self.valve_closed else EXPECTED_FLOW[n] * (self.rng.uniform(0.95, 1.03))
        if cavitation and n:
            q *= self.rng.uniform(0.55, 0.65)
        static = 3.2 + 0.098 * self.reservoir
        if n == 0:
            press = static
        elif self.valve_closed:
            press = 8.4 + 0.2 * n                              # pumping against a shut valve: shut-off head
        else:
            press = static + 1.6 * (q / 400) ** 2
            if cavitation:
                press -= self.rng.uniform(0.1, 0.3)
        self.pressure = toward(self.pressure, press + self.rng.gauss(0, 0.03), dt, 3)
        self.flow = q
        for p in self.pumps:
            share = q / n if p.running and n else 0.0
            if p.running:
                amps = 32 + 20 * min(1.3, share / 300)            # a centrifugal pump draws more with more flow
                if cavitation:
                    amps += self.rng.gauss(0, 3)
                p.current = amps + self.rng.gauss(0, 0.6)
                vib = self.rng.uniform(8.0, 11.0) if cavitation else self.rng.uniform(1.8, 2.6)
                p.vib = toward(p.vib, vib, dt, 10)
                blocked = fan_blocked and p is self.hot_pump
                target = 150.0 if blocked else 25 + 50 * p.current / 52
                p.temp = toward(p.temp, target, dt, 150 if blocked else 600)
                p.hours += dt / 3600
            else:
                p.current, p.vib = 0.0, toward(p.vib, 0.1, dt, 10)
                p.temp = toward(p.temp, 25.0, dt, 400)
            p.flow = share
        inflow = max(0.0, 200 * (world.river - self.sump))
        self.sump = min(4.0, max(0.0, self.sump + (inflow - q) / 3600 * dt / SUMP_AREA))
        self.outflow = world.demand if self.reservoir > 0.2 else world.demand * self.reservoir / 0.2
        self.reservoir = min(6.0, max(0.0, self.reservoir + (q - self.outflow) / 3600 * dt / RESERVOIR_AREA))
        if self.t_v.due(t):
            self.voltage = self.rng.gauss(400, 3)
        self.ups = min(100.0, self.ups + dt / 60) if self.mains else max(0.0, self.ups - dt / 30)
        if not stuck:
            self.sump_reading = self.sump + self.rng.gauss(0, 0.003)   # a live transmitter always flickers
        self.write_registers()

    def write_registers(self):
        """The PLC's Modbus map (0-based protocol addresses, as on the wire). See the README's signal table."""
        p1, p2 = self.pumps
        ir = self.regs["input"]
        u16 = lambda v: max(0, min(65535, int(round(v))))
        ir[0] = u16(self.sump_reading * 1000)           # sump level, mm
        ir[1] = u16(self.reservoir * 1000)              # reservoir level, mm
        ir[2] = u16(self.flow * 10)                     # station flow, 0.1 m3/h
        ir[3] = u16(self.pressure * 100)                # discharge pressure, 0.01 bar
        ir[4], ir[5] = u16(p1.current * 10), u16(p2.current * 10)      # motor current, 0.1 A
        ir[6], ir[7] = u16(p1.temp * 10), u16(p2.temp * 10)            # motor winding temperature, 0.1 Cel
        ir[8], ir[9] = u16(p1.vib * 100), u16(p2.vib * 100)            # vibration, 0.01 mm/s RMS
        for base, p in ((10, p1), (12, p2)):                            # run hours, 0.1 h, 32 bits, high word first
            v = int(p.hours * 10)
            ir[base], ir[base + 1] = (v >> 16) & 0xFFFF, v & 0xFFFF
        ir[14] = u16((self.voltage if self.mains else 0) * 10)          # mains voltage, 0.1 V
        ir[15] = (p1.thermal_trip << 0 | p2.thermal_trip << 1 | self.hp_trip << 2 | self.dry_run << 3
                  | (not self.mains) << 4 | self.link_down << 5 | (self.duty == 1) << 6 | self.call << 7)
        ir[16] = u16(self.outflow * 10)                 # outflow to the town, 0.1 m3/h
        ir[17] = u16(self.ups)                          # UPS charge, %
        di = self.regs["discrete"]
        di[0], di[1] = int(p1.running), int(p2.running)
        di[2] = int(self.sump < FLOAT_LOW)              # the hardwired low-level float switch
        di[3] = int(not self.mains)                     # UPS on battery
        di[4] = int(not self.link_down)                 # telemetry link to the control room healthy

    def read_input_registers(self, address, count):     # Modbus function 4
        return list(self.regs["input"][address:address + count])

    def read_discrete_inputs(self, address, count):     # Modbus function 2
        return list(self.regs["discrete"][address:address + count])


# ------------------------------------------------------------------------------------------------ the HEART

class Readings:
    """One poll of the PLC, scaled into engineering units. CHANGE for a product: a Modbus TCP client (read only)."""

    def __init__(self, plc, heart):
        ir, di = plc.read_input_registers(0, 18), plc.read_discrete_inputs(0, 5)
        self.sump, self.reservoir, self.flow, self.pressure = ir[0] / 1000, ir[1] / 1000, ir[2] / 10, ir[3] / 100
        self.current, self.temp, self.vib = [ir[4] / 10, ir[5] / 10], [ir[6] / 10, ir[7] / 10], [ir[8] / 100, ir[9] / 100]
        self.hours = [((ir[10] << 16) | ir[11]) / 10, ((ir[12] << 16) | ir[13]) / 10]
        self.voltage, self.outflow, self.ups, word = ir[14] / 10, ir[16] / 10, ir[17], ir[15]
        self.thermal_trip = [bool(word & 1), bool(word & 2)]
        self.hp_trip, self.dry_run, self.mains_lost = bool(word & 4), bool(word & 8), bool(word & 16)
        self.duty, self.call = 2 if word & 64 else 1, bool(word & 128)
        self.running = [bool(di[0]), bool(di[1])]
        self.float_low, self.on_ups, self.link_up = bool(di[2]), bool(di[3]), bool(di[4])
        self.n = sum(self.running)
        self.available = sum(1 for i in (0, 1) if not self.thermal_trip[i]) \
            if not (self.hp_trip or self.dry_run or self.mains_lost) else 0
        self.sump_unchanged = heart.unchanged_for("sump", ir[0])
        self.state = "tripped" if self.hp_trip or self.dry_run else "no-power" if self.mains_lost else \
            "pumping" if self.n else "idle"

    @property
    def hours_left(self):
        return max(0.0, (self.reservoir - RESERVOIR_LOW) * RESERVOIR_AREA / max(self.outflow, 1))

    def why_lost(self):
        if self.mains_lost:
            return "Mains supply is lost"
        if self.hp_trip:
            return "The PLC has tripped the station on high discharge pressure and locked both pumps out"
        if self.dry_run:
            return "The low-level float switch has stopped both pumps (dry-run protection)"
        if self.available == 0:
            return "Both pumps are tripped"
        return "The PLC is calling for water but no pump is running"

    def hottest(self):
        i = 0 if self.temp[0] >= self.temp[1] else 1
        return i + 1, self.temp[i]


RULES = [
    # code, severity, condition, summary
    ("system.main-function-lost", "critical", lambda r: r.available == 0 or (r.call and r.n == 0),
     "The station cannot pump water"),
    ("system.main-function-degraded", "warning",
     lambda r: r.n > 0 and max(r.vib[i] for i in (0, 1) if r.running[i]) > VIBRATION_ALARM
     and r.flow < 0.75 * EXPECTED_FLOW[r.n], "Pumping is degraded: high vibration and low flow"),
    ("sensors.stuck", "warning", lambda r: r.sump_unchanged >= STUCK_SEC, "The sump level transmitter is stuck"),
    ("performance.too-hot", "warning", lambda r: max(r.temp) >= HOT_WARNING, "A pump motor is running hot"),
    ("power.supply-lost", "error", lambda r: r.mains_lost or r.on_ups, "Mains supply to the station is lost"),
    ("connectivity.link-lost", "warning", lambda r: not r.link_up, "The telemetry link to the control room is down"),
]

EXPLAIN = {
    "system.main-function-lost": lambda r: f"{r.why_lost()}. The town is drawing {r.outflow:.0f} m3/h; the reservoir is "
    f"at {r.reservoir:.2f} m, about {r.hours_left:.1f} h of supply above its low level. Send the on-call operator.",
    "system.main-function-degraded": lambda r: f"Flow {r.flow:.0f} m3/h against {EXPECTED_FLOW[r.n]:.0f} expected, vibration "
    f"{max(r.vib):.1f} mm/s (alarm {VIBRATION_ALARM}). Likely cavitation: check the suction strainer, the suction valve "
    f"and the sump level ({r.sump:.2f} m).",
    "sensors.stuck": lambda r: f"The sump level has read {r.sump:.3f} m for {r.sump_unchanged:.0f} s while water moves. "
    "Do not trust it. Dry-run protection is not affected: it comes from the PLC's own hardwired float switch.",
    "performance.too-hot": lambda r: "Pump {} winding at {:.0f} Cel; the PLC trips it at {:.0f} Cel. Check the motor's "
    "cooling fan and air path.".format(*r.hottest(), THERMAL_TRIP),
    "power.supply-lost": lambda r: f"The PLC and this gateway are on the UPS ({r.ups} %). The pumps cannot run until mains "
    "returns; the PLC will restart them itself.",
    "connectivity.link-lost": lambda r: "The control room cannot see the station and reports cannot be sent. The PLC keeps "
    "running the pumps on its own; alerts wait here until the link returns.",
}


class Heart:
    """Minimum Tending: notice, change status, raise, explain, escalate. No network or model is needed for any of it.
    It holds a Station only to read its registers: read_input_registers() and read_discrete_inputs()."""

    def __init__(self, plc):
        self.plc = plc
        self.lock = threading.Lock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.last_raw, self.t = {}, 0.0
        self.r = None
        self.owner_key = secrets.token_urlsafe(18)

    def unchanged_for(self, name, raw):
        prev = self.last_raw.get(name)
        if prev is None or prev[0] != raw:
            self.last_raw[name] = (raw, self.t)
            return 0.0
        return self.t - prev[1]

    def tend(self, t):
        with self.lock:
            self.t = t
            self.r = r = Readings(self.plc, self)
            self.sequence += 1
            for code, severity, cond, summary in RULES:
                hit = cond(r)
                # A condition must hold for 5 simulated seconds before it is raised: one odd reading is not a fault.
                prev = self.since.get(code)
                self.since[code] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[code] >= 5
                open_ = next((a for a in self.alerts if a["code"] == code and a["status"] != "resolved"), None)
                if held and open_ is None:
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": summary, "explanation": EXPLAIN[code](r)}
                    self.alerts = (self.alerts + [a])[-100:]
                    self.escalate(a, r)
                elif not hit and open_ is not None:
                    open_.update(status="resolved", resolvedAt=utc())
                    log(f"HEART resolved   {code}: {open_['summary'].lower()}")
            sev = {a["severity"] for a in self.alerts if a["status"] != "resolved"}
            self.status = "failing" if "critical" in sev else "degraded" if sev else "healthy"

    def escalate(self, a, r):
        """CHANGE for a product: the utility's own route (a SCADA alarm, a page to the on-call operator) and the
        station's local alarm lamp, which needs no network. Here: the console."""
        a["status"] = "escalated"
        held = "" if r.link_up else " [link down: local alarm lamp on, page held until the link returns]"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}{held}")

    def envelope(self, kind, data, privacy="public"):
        return {"specversion": "1.0", "id": str(uuid.uuid4()), "source": f"/heart/{SYSTEM_ID}/{INSTANCE}", "type": kind,
                "time": utc(), "datacontenttype": "application/json", "heartversion": "1.0", "heartprivacy": privacy,
                "data": data}

    def pulse(self, owner=False):
        with self.lock:
            d = {"system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "status": self.status,
                 "sequence": self.sequence, "intervalSec": PULSE_SEC}
            if owner and self.r:
                d["reconnaissance"] = {"state": self.r.state,
                                       "lastEvent": self.plc.events[-1]["name"] if self.plc.events else "none"}
                d["autonomy"] = {"current": 1}
        return self.envelope("heart.pulse", d, "internal" if owner else "public")

    def public_alerts(self, n=16):
        with self.lock:
            keep = ("id", "code", "severity", "status", "detectedAt", "resolvedAt")
            return [self.envelope("heart.arrhythmia", {k: a[k] for k in keep if k in a}) for a in reversed(self.alerts[-n:])]

    def properties(self, owner=False):
        with self.lock:
            r = self.r
        if r is None or not owner:
            return []          # every value of this station is internal: critical infrastructure says little in public
        stuck = r.sump_unchanged >= STUCK_SEC
        items = [  # name, value, unit, privacy, quality
            ("station_state", r.state, None, "internal", None),
            ("sump_level", round(r.sump, 3), "m", "internal", "bad" if stuck else "good"),
            ("low_level_float", "low" if r.float_low else "ok", None, "internal", None),
            ("reservoir_level", round(r.reservoir, 3), "m", "internal", None),
            ("flow", round(r.flow, 1), "m3/h", "internal", None),
            ("outflow_to_town", round(r.outflow, 1), "m3/h", "internal", None),
            ("discharge_pressure", round(r.pressure, 2), "bar", "internal", None),
            ("duty_pump", r.duty, None, "internal", None),
            ("mains_voltage", round(r.voltage, 1), "V", "internal", None),
            ("ups_charge", r.ups, "%", "internal", None),
            ("ups_on_battery", r.on_ups, None, "internal", None),
            ("telemetry_link", "up" if r.link_up else "down", None, "internal", None),
        ]
        for i in (0, 1):
            n = i + 1
            items += [
                (f"pump{n}_running", r.running[i], None, "internal", None),
                (f"pump{n}_thermal_trip", r.thermal_trip[i], None, "internal", None),
                (f"pump{n}_motor_current", round(r.current[i], 1), "A", "internal", None),
                (f"pump{n}_motor_temperature", round(r.temp[i], 1), "Cel", "internal", None),
                (f"pump{n}_vibration", round(r.vib[i], 2), "mm/s", "internal", None),
                (f"pump{n}_run_hours", round(r.hours[i], 1), "h", "internal", None),
            ]
        out = []
        for name, value, unit, privacy, quality in items:
            d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": value, "writable": False, "time": utc()}
            if unit:
                d["unit"] = unit
            if quality:
                d["quality"] = quality
            out.append(self.envelope("heart.property", d, privacy))
        return out


# ------------------------------------------------------------------------------------------------ HEART Monitor

def identity(path, url, new=False):
    """Made once from a cryptographic random source, never from anything public."""
    if not new and os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    ident = {"id": secrets.token_hex(8), "claimCode": "".join(secrets.choice(CROCKFORD) for _ in range(20)),
             "key": secrets.token_urlsafe(32)}
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(ident, f)
    return ident


def show_claim(ident, url):
    code = "-".join(ident["claimCode"][i:i + 5] for i in range(0, 20, 5))
    print(f"""
  HEART Monitor: register this simulated pump station
    1. Sign in at {url}/login with your email (a 6-digit code is sent to you).
    2. Open {url}/register?id={ident['id']}
    3. Enter the claim code  {code}
    4. Watch it at {url}/device/{ident['id']}  (private to you until you share it)
""", file=sys.stderr, flush=True)


def post(url, path, body, key=None):
    req = urllib.request.Request(url.rstrip("/") + path, data=json.dumps(body).encode(), method="POST",
                                 headers={"Content-Type": "application/json", "User-Agent": "heart-pump-station-simulator/1.0",
                                          **({"Authorization": f"Bearer {key}"} if key else {})})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except ValueError:
            return e.code, {}


def monitor(heart, plc, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    station's telemetry link is down (the reports go out through the same router): then the site shows a Flatline
    after three missed intervals."""
    host = url.split("://", 1)[-1].split("/")[0].split(":")[0]
    if not url.startswith("https://") and host not in ("127.0.0.1", "localhost"):
        log("HEART Monitor: refusing a site without https")
        return
    ident, phase, next_at, last, sent, backoff, claimed = identity(path, url), "hello", 0.0, 0.0, None, 0, None
    show_claim(ident, url)
    while not stop.is_set():
        with heart.lock:
            seen = (heart.status, tuple((a["code"], a["status"]) for a in heart.alerts[-16:]))
        due = time.time() >= next_at or (phase == "report" and seen != sent and time.time() - last >= 10)
        if not due or (phase == "report" and plc.link_down):
            stop.wait(1)
            continue
        try:
            if phase == "hello":
                status, reply = post(url, f"/v1/devices/{ident['id']}/hello", {
                    "claimHash": hashlib.sha256(f"heart-claim:{ident['claimCode']}".encode()).hexdigest(),
                    "keyHash": hashlib.sha256(ident["key"].encode()).hexdigest(), "template": TEMPLATE})
                if status in (200, 201):
                    phase, next_at = "report", 0
                    log(f"HEART Monitor: {url} knows this device")
                elif status == 409:
                    ident = identity(path, url, new=True)
                    show_claim(ident, url)
                else:
                    log(f"HEART Monitor: hello answered {status}; trying again in 10 minutes")
                    next_at = time.time() + 600
                continue
            last, sent = time.time(), seen
            status, reply = post(url, f"/v1/devices/{ident['id']}/report",
                                 {"pulse": heart.pulse(), "alerts": heart.public_alerts()}, ident["key"])
            backoff = 0
            if status in (200, 202) and reply.get("claimed") != claimed:
                claimed = reply.get("claimed")
                log("HEART Monitor: registered, reports are now stored" if claimed else
                    "HEART Monitor: not registered yet, nothing is stored (see the claim code above)")
            if status in (200, 202, 429):
                next_at = time.time() + int(reply.get("nextSec") or PULSE_SEC)
            elif status == 404:
                phase = "hello"
            elif status == 401:
                log("HEART Monitor: the site refused this device's key. Reporting has stopped.")
                return
            else:
                log(f"HEART Monitor: report answered {status} {reply.get('detail', '')}; not retrying it unchanged")
                next_at = time.time() + 600
        except (OSError, ValueError) as e:
            backoff = min(backoff * 2 or 30, 600)
            log(f"HEART Monitor: offline ({e.__class__.__name__}); trying again in {backoff} s. Noticing goes on.")
            next_at = time.time() + backoff


# ------------------------------------------------------------------------------------------------ the HEART API

def handler(heart):
    class H(BaseHTTPRequestHandler):
        def send(self, status, body, ctype="application/cloudevents+json"):
            raw = json.dumps(body, indent=1).encode()
            self.send_response(status)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(raw)))
            self.send_header("Heart-Version", "1.0")
            self.end_headers()
            self.wfile.write(raw)

        def problem(self, status, code, detail):
            self.send(status, {"type": f"https://heartcortex.ai/problems/{code}", "title": code.replace("-", " "),
                               "status": status, "detail": detail, "code": code}, "application/problem+json")

        def owner(self):
            return secrets.compare_digest(self.headers.get("Authorization", ""), f"Bearer {heart.owner_key}")

        def do_GET(self):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            if path == "/.well-known/heart-card.json":
                port = self.server.server_address[1]
                return self.send(200, heart.envelope("heart.card", {
                    "system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "level": "basic",
                    "bindings": ["http-server"], "endpoints": {"api": f"http://127.0.0.1:{port}/heart/v1"},
                    "pulse": {"intervalSec": PULSE_SEC}, "modules": ["properties"],
                    "shares": {"pulse": "public", "arrhythmia": "public", "property": "internal"},
                    "autonomy": {"level": 1}, "template": TEMPLATE}))
            if path == "/heart/v1/pulse":
                return self.send(200, heart.pulse(owner))
            if path == "/heart/v1/arrhythmias":
                if owner:
                    with heart.lock:
                        return self.send(200, {"items": [heart.envelope("heart.arrhythmia", dict(a), "internal")
                                                         for a in reversed(heart.alerts)]}, "application/json")
                return self.send(200, {"items": heart.public_alerts(100)}, "application/json")
            if path == "/heart/v1/properties":
                return self.send(200, {"items": heart.properties(owner)}, "application/json")
            if path == "/heart/v1/plc-events":
                if not owner:
                    return self.problem(403, "insufficient-clearance", "The PLC's event log is internal: send the owner key")
                with heart.lock:
                    return self.send(200, {"items": list(reversed(heart.plc.events))}, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties or /plc-events")

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "This HEART only watches. The PLC runs the pumps and its interlocks; HEART changes nothing.")

        do_PUT = do_POST = do_PATCH = do_DELETE = lambda self: self.refuse()

        def log_message(self, *a):
            pass
    return H


# ------------------------------------------------------------------------------------------------ main

INSTANCE = "sim"
SCENARIOS = {"stuck", "link", "cavitation", "hot", "trip", "power"}


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated water pump station with a HEART, reporting to HEART Monitor. Demo only.")
    ap.add_argument("--monitor-url", default=os.environ.get("HEART_MONITOR_URL", "https://heartcortex.ai"))
    ap.add_argument("--no-monitor", action="store_true", help="do not report to a monitor site")
    ap.add_argument("--state-file", default="pump-station-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43305, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help="stuck,link,cavitation,hot,trip,power or all")
    ap.add_argument("--speed", type=float, default=1.0, help="simulated seconds per real second (default 1)")
    ap.add_argument("--seed", type=int, default=None, help="random seed, to repeat a run")
    ap.add_argument("--duration", type=float, default=0, help="stop after this many real seconds (0: run until Ctrl-C)")
    args = ap.parse_args()
    scenarios = set(SCENARIOS) if args.scenario == "all" else {s for s in args.scenario.split(",") if s}
    unknown = scenarios - SCENARIOS
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")

    rng = random.Random(args.seed)
    world, plc = World(rng), Station(rng, scenarios)
    heart = Heart(plc)
    if not args.no_monitor and os.path.exists(args.state_file):
        with open(args.state_file) as f:
            INSTANCE = json.load(f).get("id", INSTANCE)
    elif not args.no_monitor:
        INSTANCE = identity(args.state_file, args.monitor_url)["id"]

    print(f"""
  {NAME}
  NOT A CONTROL SYSTEM. Simulated town, river, pumps and PLC. HEART only watches: the PLC runs the station.
  Scenarios: {', '.join(sorted(scenarios)) or 'none (random life only)'} · speed {args.speed}x
""", file=sys.stderr, flush=True)

    stop = threading.Event()
    if args.port:
        try:
            srv = ThreadingHTTPServer(("127.0.0.1", args.port), handler(heart))
        except OSError:
            srv = ThreadingHTTPServer(("127.0.0.1", 7685), handler(heart))   # the standard fallback port
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        p = srv.server_address[1]
        log(f"HEART API on http://127.0.0.1:{p}/heart/v1/pulse  (owner key for private views: {heart.owner_key})")
    if not args.no_monitor:
        threading.Thread(target=monitor, args=(heart, plc, args.monitor_url, args.state_file, stop), daemon=True).start()

    t, wall0, last_print = 0.0, time.time(), 0.0
    try:
        while not stop.is_set():
            dt = args.speed
            t += dt
            world.step(t, wall0 + t)
            plc.step(t, dt, world)
            heart.tend(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                r = heart.r
                pumps = " ".join(f"P{i + 1} {'RUN' if r.running[i] else 'TRIP' if r.thermal_trip[i] else 'off'}"
                                 for i in (0, 1))
                log(f"sump {r.sump:4.2f} m · reservoir {r.reservoir:4.2f} m · {pumps} (duty P{r.duty}) · flow "
                    f"{r.flow:4.0f} m3/h · town {r.outflow:4.0f} m3/h · {r.pressure:4.2f} bar · motor "
                    f"{max(r.temp):4.0f} Cel · vib {max(r.vib):4.1f} mm/s · mains {'OFF' if r.mains_lost else 'on'} · "
                    f"link {'up' if r.link_up else 'down'} · HEART {heart.status}")
            if args.duration and time.time() - wall0 >= args.duration:
                break
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    stop.set()
    log("stopped. The monitor site will show a Flatline if this station stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo