#!/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/. 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()