Pacemaker: what is inside

An implant and its home transmitter. HEART watches pacing and never touches it. HEART runs on the bedside home transmitter, never in the implant.

Specifications it follows

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

Body  ──►  Pacemaker  ──(read only)──►  Heart  ──►  local HEART API      (127.0.0.1)
patient     the therapy                  watches     └►  HEART Hub    (outbound only)

Run it yourself, and register it on HEART Hub

1. Download README.md, pacemakerSimulator.py, test_e2e.py.
2. Run python3 pacemakerSimulator.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: Pacemaker.

The code: pacemakerSimulator.py

531 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/pacemaker-simulator/v1.2611.2
"""pacemakerSimulator.py: a simulated pacemaker with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. It is not a medical device, simulates no real patient, and must never be used for anything
medical. Its numbers are plausible, not clinical. See README.md beside this file, and the case study guides/Pacemaker.md.

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

  * a simulated BODY: the patient's own heart rate (with day and night, exercise, slow-heart pauses and atrial
    fibrillation episodes), movement and temperature, each changing on its own random timer;
  * a simulated PACEMAKER (the therapy): it senses each beat and paces when the heart is too slow, with rate
    response from movement, and has a battery, a lead, a capture threshold and a home transmitter link that age,
    drift and sometimes fail;
  * a HEART that only WATCHES the pacemaker. It never changes pacing: nothing in this file lets it. It notices
    problems, raises alerts with standard fault codes, tells a person, serves the HEART API on this computer, and
    reports public health (status and fault codes only) to a HEART Monitor site.

Run it:

  python3 pacemakerSimulator.py                       # report to https://heartcortex.ai
  python3 pacemakerSimulator.py --scenario all        # also stage each fault, one after another, a few minutes apart
  python3 pacemakerSimulator.py --no-monitor          # local only: http://127.0.0.1:43278/heart/v1/pulse
  python3 pacemakerSimulator.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 pacemaker-monitor.json (mode 600), so the same simulated device comes back after a restart.

Scenarios (--scenario, comma-separated): noise, link, capture, lead, battery, all. Each fault starts at a set time,
lasts a while, and is then "fixed at the clinic", so you can watch an alert open, escalate and resolve. battery
starts the battery near its replacement point and ages it fast.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. Heart
rate, rhythm episodes, temperature and every device measurement stay local (internal or private).
"""
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.pacemaker"
NAME = "Pacemaker Simulator (demo, not a medical device)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/pacemaker-simulator/v1.2611.2"
PULSE_SEC = 30
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

# Programmed settings of the simulated pacemaker. Only a clinician's programmer may change them, never the HEART.
LOWER_RATE, UPPER_RATE = 60, 130          # beats per minute
OUTPUT_V = 2.5                            # pacing output, volts
IMPEDANCE_OK = (200, 2000)                # ohms: outside this, the lead is suspect
ERI_MONTHS, EOL_MONTHS = 3.0, 0.5         # elective replacement, and end of life


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))


# ------------------------------------------------------------------------------------------------ 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 Body:
    """The simulated patient. Private data: it never leaves this computer."""

    def __init__(self, rng):
        self.rng = rng
        self.activity, self.rate, self.temp = 0.1, 72.0, 36.9
        self.t_activity, self.t_rate, self.t_temp = Timer(rng, 3, 15), Timer(rng, 1, 3), Timer(rng, 60, 300)
        self.exercise = Episode(rng, (900, 3600), (300, 1200))         # a walk
        self.pause = Episode(rng, (600, 2700), (20, 150))              # the patient's own heart slows: pacing takes over
        self.af = Episode(rng, (3600, 4 * 3600), (120, 1200))          # atrial fibrillation, irregular and fast
        self.fever = Episode(rng, (12 * 3600, 48 * 3600), (1800, 7200))

    def step(self, t, wall):
        hour = time.localtime(wall).tm_hour
        night = hour >= 23 or hour < 6
        if self.t_activity.due(t):
            target = 0.6 if self.exercise.active(t) else (0.02 if night else 0.12)
            self.activity = min(1.0, max(0.0, self.activity + (target - self.activity) * 0.4 + self.rng.gauss(0, 0.04)))
        if self.t_rate.due(t):
            base = (58 if night else 70) + 60 * self.activity
            if self.pause.active(t):
                base = self.rng.uniform(30, 40)
            if self.af.active(t):
                base = self.rng.uniform(100, 160)
            self.rate += (base - self.rate) * 0.5 + self.rng.gauss(0, 1.5)
        if self.t_temp.due(t):
            target = 38.4 if self.fever.active(t) else 36.9 + 0.3 * self.activity
            self.temp += (target - self.temp) * 0.3 + self.rng.gauss(0, 0.05)


class Pacemaker:
    """The simulated therapy. The HEART reads these values. Nothing in this file lets the HEART change them."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.months = 3.06 if "battery" in scenarios else rng.uniform(80, 110)
        self.ageing = 4000.0 if "battery" in scenarios else 1.0        # the battery scenario ages it fast
        self.impedance, self.threshold, self.r_wave = rng.uniform(480, 620), rng.uniform(0.6, 1.0), rng.uniform(7, 11)
        self.t_imp, self.t_thr, self.t_rw = Timer(rng, 30, 120), Timer(rng, 60, 300), Timer(rng, 10, 60)
        self.noise_ep = Episode(rng, (1800, 5400), (20, 60))            # electrical noise near a magnet or a motor
        self.link_ep = Episode(rng, (2 * 3600, 6 * 3600), (120, 600))  # the bedside transmitter loses the implant
        self.paced_fraction, self.rate, self.paced, self.captured = 0.0, 70.0, False, True
        self.noise = self.link_down = False
        self.events = []                                              # private rhythm log, newest last

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

    def step(self, t, dt, body):
        # Staged faults, for the demo. Each lasts a while and is then "fixed at the clinic".
        lead_broken = self.staged("lead", t, 600, 240)
        dislodged = self.staged("capture", t, 900, 180)
        self.noise = self.noise_ep.active(t) or self.staged("noise", t, 60, 45)
        self.link_down = self.link_ep.active(t) or self.staged("link", t, 180, 360)

        if self.t_imp.due(t):
            self.impedance = min(1400, max(300, self.impedance + self.rng.gauss(0, 8)))
        if self.t_thr.due(t):
            self.threshold = min(1.6, max(0.4, self.threshold + self.rng.gauss(0, 0.04)))
        if self.t_rw.due(t):
            self.r_wave = min(15, max(3, self.r_wave + self.rng.gauss(0, 0.4)))
        impedance = self.rng.uniform(2600, 3400) if lead_broken else self.impedance
        threshold = self.rng.uniform(3.0, 3.8) if dislodged else self.threshold

        # One beat decision per step: pace when the heart is slower than the rate the patient's movement needs.
        sensor_rate = LOWER_RATE + (UPPER_RATE - LOWER_RATE) * min(1.0, body.activity * 1.2)
        own = body.rate
        if dislodged:
            own = min(own, 42)     # the demo pairs the dislodged lead with a slow-heart pause, so pacing is needed
        needed = own < sensor_rate and not self.noise          # noise is taken for beats: pacing is held back
        self.captured = (not needed) or (OUTPUT_V > threshold and IMPEDANCE_OK[0] <= impedance <= IMPEDANCE_OK[1])
        self.paced = needed
        self.rate = (sensor_rate if self.captured else own) if needed else own
        k = min(1.0, dt / 600)                                 # paced fraction over about 10 minutes
        self.paced_fraction += ((1.0 if needed else 0.0) - self.paced_fraction) * k
        drain = (1 + 0.5 * self.paced_fraction) * self.ageing  # pacing uses the battery
        self.months = max(0.0, self.months - dt * drain / (30.44 * 86400))
        self.impedance_now, self.threshold_now = impedance, threshold
        if body.af.active(t) and not (self.events and self.events[-1]["name"] == "af.started"):
            self.events.append({"time": utc(), "name": "af.started", "rate": round(own)})
        if not body.af.active(t) and self.events and self.events[-1]["name"] == "af.started":
            self.events.append({"time": utc(), "name": "af.ended"})
        self.events = self.events[-200:]

    @property
    def voltage(self):
        # Lithium-iodine cells hold their voltage, then fall towards the end.
        return round(2.45 + 0.35 * (1 - math.exp(-self.months / 6)), 3)


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

RULES = [
    # code, severity, condition, summary
    ("system.main-function-lost", "critical", lambda p: p.paced and not p.captured,
     "Pacing is not capturing the heart"),
    ("system.main-function-degraded", "warning", lambda p: p.captured and p.threshold_now > OUTPUT_V / 2,
     "The pacing safety margin is low"),
    ("sensors.out-of-range", "error", lambda p: not IMPEDANCE_OK[0] <= p.impedance_now <= IMPEDANCE_OK[1],
     "Lead impedance is out of range"),
    ("sensors.noisy", "warning", lambda p: p.noise, "Electrical noise is being sensed as heartbeats"),
    ("power.battery-low", "error", lambda p: p.months <= ERI_MONTHS, "The battery has reached its replacement point"),
    ("power.battery-failing", "critical", lambda p: p.months <= EOL_MONTHS, "The battery is at end of life"),
    ("connectivity.link-lost", "warning", lambda p: p.link_down, "The home transmitter cannot reach the implant"),
]

EXPLAIN = {
    "system.main-function-lost": lambda p: f"Pacing at {OUTPUT_V} V, but the capture threshold is {p.threshold_now:.1f} V or the lead is not carrying the pulse. Contact the clinic now.",
    "system.main-function-degraded": lambda p: f"Capture threshold {p.threshold_now:.2f} V against an output of {OUTPUT_V} V: less than a 2:1 safety margin.",
    "sensors.out-of-range": lambda p: f"Lead impedance {p.impedance_now:.0f} ohms, outside {IMPEDANCE_OK[0]} to {IMPEDANCE_OK[1]} ohms: possible lead fracture or insulation break.",
    "sensors.noisy": lambda p: "Noise may stop pacing when it is needed. Move away from strong magnets, motors or welding equipment.",
    "power.battery-low": lambda p: f"About {p.months:.1f} months left ({p.voltage} V). Book a replacement.",
    "power.battery-failing": lambda p: f"About {p.months * 30.44:.0f} days left ({p.voltage} V). Replacement is urgent.",
    "connectivity.link-lost": lambda p: "Readings cannot be sent until the transmitter is back in range. The implant keeps working.",
}


class Heart:
    """Minimum Tending: notice, change status, raise, explain, escalate. No network or model is needed for any of it."""

    def __init__(self, pacemaker, body):
        self.pm, self.body = pacemaker, body
        self.lock = threading.Lock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)

    def tend(self, t):
        with self.lock:
            self.sequence += 1
            for code, severity, cond, summary in RULES:
                hit = cond(self.pm)
                # 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](self.pm)}
                    self.alerts = (self.alerts + [a])[-100:]
                    self.escalate(a)
                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):
        """CHANGE for a product: the clinic's own route. Here: the console."""
        a["status"] = "escalated"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}")

    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:
                d["reconnaissance"] = {"state": "pacing" if self.pm.paced else "sensing",
                                       "lastEvent": self.pm.events[-1]["name"] if self.pm.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):
        p, b = self.pm, self.body
        items = [  # name, value, unit, privacy
            ("battery_status", "eol" if p.months <= EOL_MONTHS else "eri" if p.months <= ERI_MONTHS else "ok", None, "public"),
            ("battery_months", round(p.months, 2), "mo", "internal"),
            ("battery_voltage", p.voltage, "V", "internal"),
            ("lead_impedance", round(p.impedance_now), "Ohm", "internal"),
            ("capture_threshold", round(p.threshold_now, 2), "V", "internal"),
            ("r_wave", round(p.r_wave, 1), "mV", "internal"),
            ("paced_percent", round(100 * p.paced_fraction, 1), "%", "internal"),
            ("transmitter_link", "down" if p.link_down else "up", None, "internal"),
            ("heart_rate", round(p.rate), "/min", "private"),
            ("own_heart_rate", round(b.rate), "/min", "private"),
            ("activity", round(b.activity, 2), "1", "private"),
            ("temperature", round(b.temp, 1), "Cel", "private"),
        ]
        out = []
        for name, value, unit, privacy in items:
            if privacy != "public" and not owner:
                continue
            d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": value, "writable": False, "time": utc()}
            if unit:
                d["unit"] = unit
            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 pacemaker
    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-pacemaker-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, pm, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated home transmitter has lost the implant: 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 pm.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": "private"},
                    "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/rhythm":
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Rhythm episodes are private: send the owner key")
                return self.send(200, {"items": list(reversed(heart.pm.events))}, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties or /rhythm")

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "This HEART only watches. Pacing is programmed by a clinician with a programmer, never through HEART.")

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

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


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

INSTANCE = "sim"


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated pacemaker 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="pacemaker-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43278, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help="noise,link,capture,lead,battery 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 = {"noise", "link", "capture", "lead", "battery"} if args.scenario == "all" else \
        {s for s in args.scenario.split(",") if s}
    unknown = scenarios - {"noise", "link", "capture", "lead", "battery"}
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")

    rng = random.Random(args.seed)
    body, pm = Body(rng), Pacemaker(rng, scenarios)
    heart = Heart(pm, body)
    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 MEDICAL DEVICE. Simulated patient and implant. HEART only watches: it never changes pacing.
  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, pm, 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
            body.step(t, wall0 + t)
            pm.step(t, dt, body)
            heart.tend(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                log(f"rate {pm.rate:5.0f}/min ({('paced' if pm.captured else 'NO CAPTURE') if pm.paced else 'own'}) · paced {100 * pm.paced_fraction:4.1f}% · "
                    f"lead {pm.impedance_now:5.0f} ohm · threshold {pm.threshold_now:.2f} V · battery {pm.months:6.2f} mo · "
                    f"link {'down' if pm.link_down else 'up'} · 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 device stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo