3D printer: what is inside

A networked 3D printer. Its firmware owns heating, motion and thermal-runaway protection; HEART on the print server notices, explains and tells a person. HEART runs on the printer's host computer (a Raspberry Pi with OctoPrint or Klipper, or the printer's own Linux board), never in the firmware.

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)

Room  ──►  Printer + firmware  ──(read only)──►  Heart  ──►  local HEART API      (127.0.0.1)
drafts,     jobs, PID heating,                    watches     └►  HEART Hub    (outbound only)
Wi-Fi       thermal protection

Run it yourself, and register it on HEART Hub

1. Download README.md, printer3dSimulator.py, test_e2e.py.
2. Run python3 printer3dSimulator.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: 3D printer.

The code: printer3dSimulator.py

750 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/printer3d-simulator/v1.2611.2
"""printer3dSimulator.py: a simulated 3D printer with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. It drives no real printer, heater or motor. Its numbers are plausible, not a model of any
real machine, and nothing here is a safety design. See README.md beside this file, and the case study
guides/3D-Printer.md.

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

  * a simulated ROOM: room temperature with day and night, drafts (a window, a door, an air conditioner) that cool
    the bed, and a home Wi-Fi link that drops now and then, each on its own random timer;
  * a simulated PRINTER and its FIRMWARE: print jobs of random length are queued now and then; the firmware heats
    the hotend and the bed with a PID-like controller (with a little overshoot), prints layer by layer, uses
    filament from a spool, pauses when the filament runs out, and has its own thermal-runaway protection that cuts
    the heaters by itself. Stepper drivers warm up as the motors work;
  * a HEART that only WATCHES the printer. It never heats, moves or pauses anything: nothing in this file lets it.
    It 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.

Run it:

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

Scenarios (--scenario, comma-separated): stepper, stuck, wifi, layer, runout, heater, runaway, range, all. Each
fault starts at a set time, lasts a while, and is then "fixed by a person", so you can watch an alert open,
escalate and resolve. With any scenario, two staged print jobs are queued so the faults have a print to happen to.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts.
Temperatures, progress, filament and every other measurement stay local (internal), and job names (file names can
be personal) are private.
"""
import argparse
import hashlib
import json
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.printer-3d"
NAME = "3D Printer Simulator (demo)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/printer3d-simulator/v1.2611.2"
PULSE_SEC = 30
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
SCENARIOS = ("stepper", "stuck", "wifi", "layer", "runout", "heater", "runaway", "range")

# The firmware's own limits. They live in the firmware and protect the printer without HEART.
RUNAWAY_HYSTERESIS, RUNAWAY_PERIOD = 4.0, 40.0   # reading below target by this many degrees for this long: cut heaters
WATCH_PERIOD, WATCH_RISE = 20.0, 2.0              # while heating, the reading must rise this much in this time
MAXTEMP = 275.0                                   # the firmware halts above this hotend reading
# The HEART's thresholds: when to tell a person. Illustrations; a real product takes them from its own design.
PLAUSIBLE = (-10.0, 300.0)                        # hotend readings outside this are physically impossible here
HEAT_SLOW_SEC = 120.0                             # normal hotend heat-up is 60 to 100 s
STUCK_SEC = 30.0                                  # a live thermistor reading never sits exactly still this long
OVERSHOOT_LIMIT = 15.0                            # degrees past the hotend target
DRIVER_HOT = 80.0                                 # stepper driver heatsink, Cel

JOB_NAMES = ["benchy", "phone-stand", "cable-clips-x10", "replacement-knob", "lithophane-family-photo", "planter",
             "keycap-set", "drawer-handle", "name-tag-for-emma", "dentist-appointment-reminder-holder", "vase"]
MATERIALS = {"PLA": (210, 60, 100), "PETG": (240, 80, 50)}   # hotend, bed, part fan %


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 Room:
    """The simulated room around the printer: temperature, drafts and the home Wi-Fi."""

    def __init__(self, rng):
        self.rng = rng
        self.temp = rng.uniform(19, 24)
        self.t_temp = Timer(rng, 60, 300)
        self.draft = Episode(rng, (1800, 7200), (300, 1800))        # a window, a door, an air conditioner
        self.wifi_ep = Episode(rng, (2 * 3600, 6 * 3600), (60, 600))  # the router restarts, the signal fades
        self.drafty = False

    def step(self, t, wall):
        hour = time.localtime(wall).tm_hour
        night = hour >= 23 or hour < 6
        self.drafty = self.draft.active(t)
        if self.t_temp.due(t):
            target = (18.0 if night else 22.0) - (2.5 if self.drafty else 0.0)
            self.temp += (target - self.temp) * 0.3 + self.rng.gauss(0, 0.1)


class PID:
    """The firmware's temperature controller: full power far from the target, PID close to it (like most firmware).
    The small integrator wind-up near the target gives the usual few degrees of overshoot."""

    def __init__(self, kp, ki, kd, band, i_max):
        self.kp, self.ki, self.kd, self.band, self.i_max = kp, ki, kd, band, i_max
        self.i, self.last = 0.0, None

    def reset(self):
        self.i, self.last = 0.0, None

    def power(self, target, reading, h):
        if target <= 0:
            self.reset()
            return 0.0
        err = target - reading
        if err > self.band:                     # far below: full power, no integration
            self.i, self.last = 0.0, reading
            return 1.0
        d = 0.0 if self.last is None else (reading - self.last) / h
        self.last = reading
        self.i = min(self.i_max, max(0.0, self.i + self.ki * err * h))
        return min(1.0, max(0.0, self.kp * err + self.i - self.kd * d))


class Printer:
    """The simulated printer and its firmware. The HEART reads these values. Nothing in this file lets it change them."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        room = 21.0
        self.block = self.sense = self.hotend = room          # heater block, thermistor, reading (Cel)
        self.bed_actual = self.bed = room
        self.hotend_target = self.bed_target = 0.0
        self.power_h = self.power_b = 0.0
        self.pid_h = PID(0.08, 0.012, 0.0, 10.0, 0.7)
        self.pid_b = PID(0.30, 0.006, 0.0, 3.0, 0.6)
        self.fan, self.driver_temp = 0.0, room + 5
        self.filament_g = 800.0 if scenarios else rng.uniform(300, 1000)
        self.state, self.job, self.queue, self.history, self.events = "idle", None, [], [], []
        self.next_job = 1000.0 if scenarios else rng.uniform(60, 900)
        self.staged_jobs = [(1.0, ("gift-box-for-mum", "PLA", 180, 1800.0, 60.0))] if scenarios else []
        self.heat_started = self.watch_at = self.watch_ref = self.runaway_for = 0.0
        self.reached = self.runout = self.from_below = False
        self.refill_at = self.reset_at = None
        self.halted_reason = None
        self.bed_still, self.last_bed, self.frozen_bed = 0.0, None, None
        self.shift_ep = Episode(rng, (6 * 3600, 24 * 3600), (30, 120))  # a belt slips, a nozzle hits a curled edge
        self.shift_layer, self.stalls, self.motor_current, self.peak_current = None, 0, 0.0, 0.0
        self.wifi_down = self.driver_fan_failed = self.weak_heater = self.slipped = self.open_circuit = False
        self.staged_runout_done = False
        self.t = 0.0

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

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

    # -- jobs

    def new_job(self, name, material, layers, print_sec, grams):
        h, b, fan = MATERIALS[material]
        return {"name": f"{name}.gcode", "material": material, "hotend": float(h + self.rng.choice((0, 0, 5))),
                "bed": float(b), "fan": fan, "layers": layers, "printSec": print_sec, "grams": grams, "done": 0.0,
                "queuedAt": utc(), "startedAt": None, "endedAt": None, "outcome": "queued"}

    def end_job(self, outcome):
        j = self.job
        j.update(outcome=outcome, endedAt=utc())
        self.history = (self.history + [j])[-50:]
        self.event(f"job.{outcome}")
        self.job, self.hotend_target, self.bed_target, self.fan = None, 0.0, 0.0, 0.0
        self.shift_layer = None

    def halt(self, reason):
        """The firmware's thermal protection: heaters off, motors off, job aborted. HEART only reports it."""
        self.state, self.halted_reason = "halted", reason
        self.reset_at = self.t + self.rng.uniform(600, 3600)      # until a person checks the printer and resets it
        self.power_h = self.power_b = 0.0
        self.pid_h.reset()
        self.pid_b.reset()
        if self.job:
            self.end_job("aborted")
        self.hotend_target = self.bed_target = 0.0
        self.event("firmware.halted", reason=reason)

    @property
    def progress(self):
        return 100.0 * self.job["done"] / self.job["printSec"] if self.job else 0.0

    @property
    def layer(self):
        if not self.job or self.state not in ("printing", "paused"):
            return 0
        return min(self.job["layers"], 1 + int(self.job["done"] / self.job["printSec"] * self.job["layers"]))

    @property
    def heat_secs(self):
        return self.t - self.heat_started if self.state == "heating" else 0.0

    # -- one step of simulated time

    def step(self, t, dt, room):
        self.t = t
        # Staged faults, for the demo. Each lasts a while and is then "fixed by a person".
        self.weak_heater = self.staged("heater", t, 0, 200)
        self.driver_fan_failed = self.staged("stepper", t, 240, 120)
        stuck = self.staged("stuck", t, 300, 90)
        self.wifi_down = room.wifi_ep.active(t) or self.staged("wifi", t, 330, 180)
        shifting = self.staged("layer", t, 420, 60) or (self.state == "printing" and self.shift_ep.active(t))
        slipped = self.staged("runaway", t, 700, 120)
        if self.slipped and not slipped:
            self.sense = self.block          # a person has pushed the thermistor back into the heater block
        self.slipped = slipped
        self.open_circuit = self.staged("range", t, 830, 60)
        if "runout" in self.sc and t >= 520 and not self.staged_runout_done and self.state == "printing":
            self.staged_runout_done, self.filament_g = True, 0.0

        # Jobs are queued now and then (and two staged ones with --scenario).
        while self.staged_jobs and t >= self.staged_jobs[0][0]:
            self.queue.append(self.new_job(*self.staged_jobs.pop(0)[1]))
        if t >= self.next_job:
            self.next_job = t + self.rng.uniform(1200, 5400)
            if len(self.queue) < 3:
                self.queue.append(self.new_job(self.rng.choice(JOB_NAMES), self.rng.choice(("PLA", "PLA", "PETG")),
                                               self.rng.randint(40, 400), self.rng.uniform(600, 7200),
                                               round(self.rng.uniform(5, 120), 1)))

        # A person fixes what the firmware stopped for: a new spool, or a reset after a thermal shutdown.
        if self.runout and self.refill_at is not None and t >= self.refill_at:
            self.filament_g, self.runout, self.refill_at = round(self.rng.uniform(750, 1000)), False, None
            self.state = "printing"
            self.event("job.resumed")
        if self.state == "halted" and self.reset_at is not None and t >= self.reset_at:
            self.state, self.halted_reason, self.reset_at = "idle", None, None
            self.event("firmware.reset")

        # Start the next job, if the firmware lets it: a filament is loaded and the hotend reading is sane.
        if self.state == "idle" and self.queue and self.filament_g > 0 and PLAUSIBLE[0] <= self.hotend <= PLAUSIBLE[1]:
            self.job = self.queue.pop(0)
            self.job.update(startedAt=utc(), outcome="printing")
            self.hotend_target, self.bed_target = self.job["hotend"], self.job["bed"]
            self.state, self.heat_started, self.reached = "heating", t, False
            self.watch_at, self.watch_ref = t + WATCH_PERIOD, self.hotend
            self.from_below = self.hotend < self.hotend_target      # overshoot only counts when heating up to it
            self.event("job.started")

        # Physics and firmware, in half-second steps.
        h, n = 0.5, max(1, int(round(dt / 0.5)))
        h = dt / n
        for _ in range(n):
            self.substep(h, room, stuck, shifting)

    def substep(self, h, room, stuck, shifting):
        rng, t = self.rng, self.t
        # Firmware: controllers (stopped when halted, or when the hotend reading is impossible).
        sane = PLAUSIBLE[0] <= self.hotend <= PLAUSIBLE[1]
        if self.state == "halted" or not sane:
            self.power_h = self.power_b = 0.0
        else:
            self.power_h = self.pid_h.power(self.hotend_target, self.hotend, h)
            self.power_b = self.pid_b.power(self.bed_target, self.bed, h)
        # Hotend: a 40 W cartridge in an aluminium block, cooled by the room and the part fan.
        gain = 0.45 if self.weak_heater else 1.0
        self.block += h * (2.7 * gain * self.power_h - 0.0045 * (self.block - room.temp) * (1 + 0.8 * self.fan / 100))
        if self.slipped:   # the thermistor has slipped out of the block: it reads half way to the room
            self.sense += h / 8 * (0.5 * self.block + 0.5 * room.temp - self.sense)
        else:
            self.sense += h / 3 * (self.block - self.sense)
        self.hotend = round(rng.uniform(-39.5, -37.5), 2) if self.open_circuit else round(self.sense + rng.gauss(0, 0.12), 2)
        # Bed: a 200 W plate, cooled more in a draft.
        loss = 0.005 * (1.6 if room.drafty else 1.0)
        self.bed_actual += h * (0.5 * self.power_b - loss * (self.bed_actual - room.temp))
        if stuck:
            self.frozen_bed = self.bed if self.frozen_bed is None else self.frozen_bed
            bed = self.frozen_bed
        else:
            self.frozen_bed = None
            bed = round(self.bed_actual + rng.gauss(0, 0.08), 2)
        self.bed_still = self.bed_still + h if bed == self.last_bed else 0.0
        self.bed, self.last_bed = bed, bed
        # Stepper drivers warm with motor work; their small fan keeps them cool.
        moving = self.state == "printing"
        target = room.temp + 8 + (14 if moving else 0) + (55 if self.driver_fan_failed and moving else 0)
        self.driver_temp += h / 40 * (target - self.driver_temp)
        # Motor current, and the stall detector the drivers report: a skipped step shows as a spike.
        self.motor_current = (0.8 + rng.gauss(0, 0.02)) if moving else 0.1
        if moving and shifting:
            if rng.random() < 0.2 * h:
                self.stalls += 1
                self.motor_current = rng.uniform(1.4, 1.8)
                self.peak_current = max(self.peak_current, self.motor_current)
                if self.shift_layer is None:
                    self.shift_layer = self.layer
                    self.event("motion.stall", axis="X", layer=self.layer)
        elif not shifting:
            self.shift_layer, self.stalls, self.peak_current = None, 0, 0.0

        # Firmware protection, which works on its own: MAXTEMP, heating watch, thermal runaway.
        if self.state != "halted" and sane and self.hotend > MAXTEMP:
            return self.halt("maxtemp")
        if self.state == "heating":
            if self.hotend >= self.hotend_target - RUNAWAY_HYSTERESIS:
                self.reached = True
            elif t >= self.watch_at:
                if self.hotend < self.watch_ref + WATCH_RISE:
                    return self.halt("heating-failed")
                self.watch_at, self.watch_ref = t + WATCH_PERIOD, self.hotend
            if self.reached and self.hotend >= self.hotend_target - 2 and self.bed >= self.bed_target - 1.5:
                self.state = "printing"
                self.event("job.printing")
        if self.state in ("printing", "paused") and self.reached and self.hotend_target > 0:
            low = self.hotend < self.hotend_target - RUNAWAY_HYSTERESIS
            self.runaway_for = self.runaway_for + h if low else 0.0
            if self.runaway_for >= RUNAWAY_PERIOD:
                self.runaway_for = 0.0
                self.halt("thermal-runaway")
                if "runaway" in self.sc and t < 820:
                    self.reset_at = 820.0       # the staged fault: a person reseats the thermistor and resets
                return

        # Printing: layers, progress, filament.
        if self.state == "printing" and self.job:
            j = self.job
            j["done"] += h
            self.fan = 0.0 if self.layer <= 2 else j["fan"] + rng.uniform(-1, 1)
            self.filament_g = max(0.0, self.filament_g - j["grams"] * h / j["printSec"])
            if self.filament_g <= 0:
                # The firmware's run-out sensor pauses the job and parks the head; the hotend is kept hot.
                self.state, self.runout = "paused", True
                self.refill_at = 640.0 if self.staged_runout_done and t < 640 else t + rng.uniform(600, 2400)
                self.event("job.paused", reason="filament-runout")
            elif j["done"] >= j["printSec"]:
                self.end_job("finished")
                self.state = "idle"


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

RULES = [
    # rule, code, severity, condition, summary. Two rules may share a code: each keeps its own alert.
    ("thermal-shutdown", "system.main-function-lost", "critical", lambda p: p.state == "halted",
     "The firmware has shut the heaters down"),
    ("runout", "system.main-function-lost", "error", lambda p: p.runout,
     "Filament has run out: the print is paused"),
    ("slow-heating", "system.main-function-degraded", "warning",
     lambda p: p.heat_secs > HEAT_SLOW_SEC and p.hotend < p.hotend_target - 5,
     "The hotend is heating far more slowly than normal"),
    ("bed-stuck", "sensors.stuck", "error", lambda p: p.bed_target > 0 and p.bed_still >= STUCK_SEC,
     "The bed temperature reading has stopped changing"),
    ("hotend-range", "sensors.out-of-range", "error", lambda p: not PLAUSIBLE[0] <= p.hotend <= PLAUSIBLE[1],
     "The hotend temperature reading is impossible"),
    ("layer-shift", "actuators.position-mismatch", "error", lambda p: p.shift_layer is not None,
     "A layer shift was detected"),
    ("overshoot", "performance.overshoot", "warning",
     lambda p: p.from_below and p.hotend_target > 0 and p.hotend - p.hotend_target > OVERSHOOT_LIMIT,
     "The hotend went well past its target"),
    ("driver-hot", "performance.too-hot", "warning", lambda p: p.driver_temp > DRIVER_HOT,
     "A stepper driver is too hot"),
    ("wifi", "connectivity.link-lost", "warning", lambda p: p.wifi_down, "The printer's Wi-Fi link is down"),
]

HALT_WHY = {
    "thermal-runaway": "its thermal-runaway protection: the hotend reading stayed below target while the heater ran at full power. "
                       "The thermistor may have come loose from the heater block",
    "heating-failed": "its heating watch: the hotend did not warm up as it should. Check the heater cartridge and its wiring",
    "maxtemp": "its maximum-temperature limit",
}

EXPLAIN = {
    "thermal-shutdown": lambda p: f"The printer's firmware cut the heaters by itself ({HALT_WHY.get(p.halted_reason, p.halted_reason)}). "
                                  "The job was aborted. HEART only reports this: check the hotend before resetting the printer.",
    "runout": lambda p: "The run-out sensor stopped the print and the firmware parked the head, keeping the nozzle hot. "
                        "Load a new spool and resume the job at the printer.",
    "slow-heating": lambda p: f"Heating for {p.heat_secs:.0f} s and the hotend is at {p.hotend:.0f} of {p.hotend_target:.0f} Cel "
                              "(normally about 80 s). A weak heater cartridge, a loose connection or a missing silicone sock.",
    "bed-stuck": lambda p: f"The bed reading has sat at exactly {p.bed:.2f} Cel for {p.bed_still:.0f} s. A live thermistor always "
                           "jitters a little. The firmware still sees a temperature at target, so it will not notice. Check the bed "
                           "thermistor and its cable.",
    "hotend-range": lambda p: f"The hotend reads {p.hotend:.1f} Cel, which is impossible in this room: an open or shorted thermistor. "
                              "The firmware will not heat until it reads sensibly. Check the thermistor connector.",
    "layer-shift": lambda p: f"The X stepper driver reported {p.stalls} stalls with current spikes up to {p.peak_current:.1f} A "
                             f"around layer {p.shift_layer}: the head likely skipped steps. Check the print, the belts and pulleys.",
    "overshoot": lambda p: f"The hotend reads {p.hotend:.0f} Cel against a target of {p.hotend_target:.0f} Cel: more than "
                           f"{OVERSHOOT_LIMIT:.0f} degrees over. The PID tuning may need to be redone.",
    "driver-hot": lambda p: f"The stepper driver heatsink is at {p.driver_temp:.0f} Cel (limit {DRIVER_HOT:.0f}). Its fan may have "
                            "stopped. Hot drivers cut their current and can skip steps.",
    "wifi": lambda p: "Reports cannot be sent until the link is back. The printer keeps printing and its firmware keeps protecting it.",
}


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

    def __init__(self, printer, room):
        self.p, self.room = printer, room
        self.lock = threading.Lock()
        self.sequence, self.status, self.alerts, self.since, self.open = 0, "unknown", [], {}, {}
        self.owner_key = secrets.token_urlsafe(18)

    def tend(self, t):
        with self.lock:
            self.sequence += 1
            for rule, code, severity, cond, summary in RULES:
                hit = cond(self.p)
                # A condition must hold for 5 simulated seconds before it is raised: one odd reading is not a fault.
                prev = self.since.get(rule)
                self.since[rule] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[rule] >= 5
                open_ = self.open.get(rule)
                if held and open_ is None:
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": summary, "explanation": EXPLAIN[rule](self.p)}
                    self.alerts = (self.alerts + [a])[-100:]
                    self.open[rule] = a
                    self.escalate(a)
                elif not hit and open_ is not None:
                    open_.update(status="resolved", resolvedAt=utc())
                    del self.open[rule]
                    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 printer's screen and buzzer, and the owner's phone. 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": self.p.state,
                                       "lastEvent": self.p.events[-1]["name"] if self.p.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, j = self.p, self.p.job
        items = [  # name, value, unit, privacy
            ("printer_state", p.state, None, "public"),
            ("hotend_temp", p.hotend, "Cel", "internal"),
            ("hotend_target", p.hotend_target, "Cel", "internal"),
            ("bed_temp", p.bed, "Cel", "internal"),
            ("bed_target", p.bed_target, "Cel", "internal"),
            ("fan_speed", round(p.fan), "%", "internal"),
            ("job_progress", round(p.progress, 1), "%", "internal"),
            ("layer", p.layer, None, "internal"),
            ("layers_total", j["layers"] if j else 0, None, "internal"),
            ("print_time", round(j["done"]) if j else 0, "s", "internal"),
            ("filament_remaining", round(p.filament_g), "g", "internal"),
            ("stepper_driver_temp", round(p.driver_temp, 1), "Cel", "internal"),
            ("room_temp", round(self.room.temp, 1), "Cel", "internal"),
            ("wifi_link", "down" if p.wifi_down else "up", None, "internal"),
            ("job_name", j["name"] if j else "", None, "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 3D printer
    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-printer3d-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, printer, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated Wi-Fi is down: 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 printer.wifi_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/jobs":
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Print jobs are private (file names can be personal): send the owner key")
                with heart.lock:
                    p = heart.p
                    jobs = {"current": p.job, "queued": list(p.queue), "history": list(reversed(p.history)),
                            "events": list(reversed(p.events[-50:]))}
                return self.send(200, jobs, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties or /jobs")

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "This HEART only watches. Heating, motion, pausing and thermal protection belong to the printer's "
                         "firmware. A later version could offer one guarded command (see spec/examples/custom/"
                         "gcode-printer-set_nozzle.json); this one offers none.")

        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 3D printer 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="printer3d-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43307, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help=",".join(SCENARIOS) + " 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 - set(SCENARIOS)
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")

    rng = random.Random(args.seed)
    room, printer = Room(rng), Printer(rng, scenarios)
    heart = Heart(printer, room)
    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}
  A DEMONSTRATION ONLY: no real printer. HEART only watches: the firmware heats, moves and protects the printer.
  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, printer, 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
            room.step(t, wall0 + t)
            with heart.lock:
                printer.step(t, dt, room)
            heart.tend(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                p = printer
                log(f"{p.state:8} · hotend {p.hotend:6.1f}/{p.hotend_target:3.0f} · bed {p.bed:5.1f}/{p.bed_target:2.0f} Cel · "
                    f"fan {p.fan:3.0f}% · {p.progress:5.1f}% layer {p.layer:3} · filament {p.filament_g:4.0f} g · "
                    f"driver {p.driver_temp:4.1f} Cel · wifi {'down' if p.wifi_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