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