Smart thermostat: what is inside

A thermostat, its hub and a gas boiler. Every change to the heating waits for a person to press the thermostat's own button, the mock AI's suggestions too. HEART runs on the hub at the boiler, beside the control firmware. The wall unit's button is the proof that a person is in the house.

Specifications it follows

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

House ──► Thermostat (firmware: control loop, limits) ──► Boiler (its own firmware)
             ▲  │ read                                       │ read
   apply()   │  ▼                                             ▼
   (clamps)  └── Heart ◄─── MockAI (suggests only)
                  │  watches, gates changes, tells a person
                  ├─► local HEART API    (127.0.0.1; owner key; button for physical changes)
                  └─► HEART Hub      (outbound only; approvals only with the flags)

Run it yourself, and register it on HEART Hub

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

The code: thermostatSimulator.py

1175 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/thermostat-simulator/v1.2611.2
"""thermostatSimulator.py: a simulated smart thermostat with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. It controls no real heating, simulates no real home, and must never be connected to a boiler or
heat pump. Its numbers are plausible, not engineered, and the house's thermal behaviour is sped up so that heating
cycles take minutes, not hours. See README.md beside this file, and the case study guides/Thermostat.md.

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

  * a simulated HOUSE: outdoor temperature on a day and night cycle with cold and mild spells, the room's thermal
    mass, radiators that warm and cool, sunshine, windows opened now and then, and people coming and going;
  * a simulated THERMOSTAT and BOILER (the product and its firmware): a wall unit with a battery-powered room sensor
    and a button, a Wi-Fi hub, and a gas boiler it switches on and off. The firmware runs the control loop and keeps
    its own limits, which HEART never overrides: frost protection at 5 C, a 28 C maximum, and anti-short-cycle timers;
  * a HEART beside the firmware. It WATCHES: 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. It also takes CHANGES through the gate: the owner may change the setpoint, the night setpoint and the
    mode with the owner key, but heating acts on the physical world, so every change waits for a press of the
    thermostat's own button (here: touch button.press in the working directory). A clearly labelled MOCK AI suggests
    energy savings as Actions that also wait for a person's yes, and never run alone. HEART is not in the control loop.

Run it:

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

Scenarios (--scenario, comma-separated): sun, link, stuck, range, boiler, cycling, battery, ai, all. Each fault starts
at a set time, lasts a while, and is then "fixed by the installer", so you can watch an alert open, escalate and
resolve. battery starts the sensor's batteries nearly flat and ages them fast. ai makes the mock AI suggest early.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts (and, only
with --monitor-approvals, the title and change of a waiting Action, so the owner can approve it on the site).
Temperatures, the boiler's state, occupancy, windows and the schedule stay local (internal or private).
"""
import argparse
import base64
import calendar
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.thermostat"
NAME = "Smart Thermostat Simulator (demo, controls no real heating)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/thermostat-simulator/v1.2611.2"
PULSE_SEC = 30
PROMISED_LEVEL = 3                 # the level its control offers: max 3 is mode suggest; it never acts on its own
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

# The firmware's own limits. HEART never overrides them: every change passes through Thermostat.apply(), which clamps.
FROST_C, MAX_C = 5.0, 28.0                # frost protection, in every mode (also "off"); highest setpoint
MIN_ON_SEC, MIN_OFF_SEC = 60, 90          # anti-short-cycle: the burner runs and rests at least this long
FLAME_CUT_SEC = 20                        # with too little water flow, the boiler's own overheat stat cuts the flame
HYSTERESIS = 0.4                          # heat below target - 0.4, stop above target + 0.4
ECO_OFFSET = 3.0                          # eco mode heats to setpoint - 3
NIGHT_HOURS = (23, 6)                     # the night schedule (private: it shows when people sleep)
SENSOR_POSSIBLE = (-10.0, 50.0)           # a room reading outside this is impossible

# What the owner may ask for through HEART: narrower than the firmware's limits.
SETPOINT_RANGE = (10.0, 28.0)             # outside: 422 out-of-range
MODES = ("heat", "eco", "off")
MIN_INTERVAL_SEC = 300                    # guard: one change of the same setting per 5 simulated minutes
OVERSHOOT_C = 1.5                         # performance.overshoot: room above target by more than this
SHORT_CYCLE_STARTS, SHORT_CYCLE_WINDOW = 4, 600   # actuators.short-cycling: this many burner starts in 10 minutes
BATTERY_LOW_MONTHS = 1.0

# Staged faults, in simulated seconds: (start, length). At --speed 5 they all fit into about 4.5 minutes.
STAGES = {
    "sun":     (30, 120),     # strong sun through a south window: performance.overshoot
    "link":    (60, 200),     # the hub loses Wi-Fi: connectivity.link-lost, and no reports
    "stuck":   (280, 180),    # the room sensor freezes: sensors.stuck
    "range":   (470, 80),     # the room sensor reads -40 C: sensors.out-of-range
    "boiler":  (560, 400),    # the boiler does not fire when called (and a window is open): system.main-function-lost
    "cycling": (980, 400),     # air in the radiators, low flow: the flame cuts again and again: actuators.short-cycling
}
SCENARIOS = set(STAGES) | {"battery", "ai"}
AI_FIRST_SEC, AI_AGAIN_SEC = 300, 100     # with scenario ai: when the mock AI first suggests, and how soon again


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


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


def num(v):
    """A number as RFC 8785 writes it: 18.0 becomes 18."""
    return int(v) if isinstance(v, float) and v.is_integer() else v


def short(name):
    """A property name without the system ID in front."""
    return name[len(SYSTEM_ID) + 1:] if name.startswith(SYSTEM_ID + ".") else name


def canonical(obj):
    """RFC 8785 JSON for the simple objects used here: strings, booleans, short numbers and objects of them."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()


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


def staged(scenarios, name, t):
    start, length = STAGES[name]
    return name in scenarios and start <= t < start + length


class House:
    """The simulated home and its weather. Occupancy and windows are private: they show when people are home."""

    def __init__(self, rng, scenarios, rhythm):
        self.rng, self.sc, self.rhythm = rng, scenarios, rhythm
        self.room, self.outdoor, self.radiator = 19.6, 5.0, 0.0
        self.t_out, self.t_people = Timer(rng, 60, 300), Timer(rng, 120, 600)
        self.cold = Episode(rng, (8 * 3600, 30 * 3600), (3 * 3600, 12 * 3600))   # a cold snap
        self.mild = Episode(rng, (8 * 3600, 30 * 3600), (3 * 3600, 12 * 3600))   # a mild, damp spell
        self.sunny = Episode(rng, (3 * 3600, 9 * 3600), (1800, 3 * 3600))
        self.window_ep = Episode(rng, (2 * 3600, 6 * 3600), (120, 600))
        self.outing = Episode(rng, (3 * 3600, 8 * 3600), (1800, 3 * 3600))
        self.people, self.occupied, self.window, self.sun = 2, True, False, 0.0

    def step(self, t, dt, wall, firing):
        lt = time.localtime(wall)
        hour = lt.tm_hour + lt.tm_min / 60
        night = hour >= NIGHT_HOURS[0] or hour < NIGHT_HOURS[1]
        if self.t_out.due(t):
            base = 5 + 3 * math.sin(2 * math.pi * (hour - 9) / 24)
            base += -9 if self.cold.active(t) else 6 if self.mild.active(t) else 0
            self.outdoor += (base - self.outdoor) * 0.3 + self.rng.gauss(0, 0.2)
        if self.t_people.due(t):
            workday = lt.tm_wday < 5 and 8.5 <= hour < 17.5
            people = 0 if (workday and self.rng.random() < 0.9) or self.outing.active(t) else 2
            if (people > 0) != (self.people > 0):
                self.rhythm("occupancy.arrived" if people else "occupancy.left", "domain", "private",
                            "Someone came home" if people else "The house is empty")
            self.people = people
            self.occupied = people > 0 and not night and self.rng.random() < 0.8   # the living room
        # The demo pairs the boiler fault with a window opened for two minutes, so that heat is soon needed.
        window = (self.window_ep.active(t) and self.people > 0) or (staged(self.sc, "boiler", t) and t < STAGES["boiler"][0] + 120)
        if window != self.window:
            self.rhythm("window.opened" if window else "window.closed", "domain", "private",
                        "A window was opened" if window else "The window was closed")
            self.window = window
        if staged(self.sc, "sun", t):
            self.sun = 0.035
        else:
            self.sun = 0.006 if self.sunny.active(t) and 9 <= hour < 16 else 0.0
        # The room: heat lost to outside, gained from radiators, sun and people. Sped up: tau 30 minutes, not hours.
        tau = 900 if self.window else 1800
        self.radiator += ((1.0 if firing else 0.0) - self.radiator) * min(1.0, dt / 40)
        gain = 0.016 * self.radiator + self.sun + 0.0002 * (self.people if self.occupied else 0)
        self.room += ((self.outdoor - self.room) / tau + gain) * dt


class Boiler:
    """The simulated gas boiler and its own firmware. It fires when the thermostat calls, within its own timers."""

    def __init__(self):
        self.firing, self.since, self.last_fired, self.flow = False, -1e9, -1e9, 30.0
        self.starts, self.lockout = [], False

    def step(self, t, dt, demand, no_ignition, low_flow):
        if self.firing:
            self.last_fired = t
            on_for = t - self.since
            if no_ignition or (low_flow and on_for >= FLAME_CUT_SEC) or (not demand and on_for >= MIN_ON_SEC):
                self.firing, self.since = False, t
        elif demand and t - self.since >= MIN_OFF_SEC:
            if no_ignition:
                self.lockout = True                       # tries, and fails to light
            else:
                self.lockout, self.firing, self.since, self.last_fired = False, True, t, t
                self.starts.append(t)
        if not no_ignition:
            self.lockout = False
        self.starts = [s for s in self.starts if s > t - SHORT_CYCLE_WINDOW]
        target = 70.0 if self.firing else 25.0
        self.flow += (target - self.flow) * min(1.0, dt / 60)


class Thermostat:
    """The simulated product: wall unit (sensor, battery, button, display), Wi-Fi hub, and the firmware's control loop.
    The HEART reads these values, and may only change settings through apply(), which keeps the firmware's limits."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.setpoint, self.mode, self.night_setpoint = 20.0, "heat", None
        self.raw = self.reading = 19.6
        self.frozen, self.prev_raw, self.unchanged_since = None, None, 0.0
        self.call, self.call_since, self.target_now, self.last_lowered = False, 0.0, 20.0, -1e9
        self.months = 1.05 if "battery" in scenarios else rng.uniform(8, 14)       # two AA cells in the wall unit
        self.ageing = 1500.0 if "battery" in scenarios else 1.0
        self.link_ep = Episode(rng, (3 * 3600, 9 * 3600), (60, 600))              # the router restarts, and so on
        self.wifi_up, self.last_change = True, {}

    def target(self, wall):
        lt = time.localtime(wall)
        night = lt.tm_hour >= NIGHT_HOURS[0] or lt.tm_hour < NIGHT_HOURS[1]
        if self.mode == "off":
            t = FROST_C
        elif self.mode == "eco":
            t = self.setpoint - ECO_OFFSET
        else:
            t = self.night_setpoint if night and self.night_setpoint is not None else self.setpoint
        return min(MAX_C, max(FROST_C, t))        # the firmware's limits, whatever was asked

    @property
    def possible(self):
        return SENSOR_POSSIBLE[0] <= self.raw <= SENSOR_POSSIBLE[1]

    def quality(self, t):
        if not self.possible:
            return "bad"
        return "uncertain" if t - self.unchanged_since >= 90 else "good"

    @property
    def voltage(self):
        return round(2.2 + 0.8 * (1 - math.exp(-self.months / 2.5)), 2)      # two alkaline cells, 3.0 V new

    def step(self, t, dt, wall, house, boiler):
        # The room sensor, as the wall unit reads it.
        if staged(self.sc, "stuck", t):
            if self.frozen is None:
                self.frozen = self.raw
            self.raw = self.frozen
        else:
            self.frozen = None
            self.raw = -40.0 if staged(self.sc, "range", t) else round(house.room + self.rng.gauss(0, 0.02), 2)
        if self.raw != self.prev_raw:
            self.prev_raw, self.unchanged_since = self.raw, t
        if self.possible:
            self.reading = self.raw                  # an impossible reading is ignored: the last good one is kept
        # The firmware's control loop. HEART is not in it.
        tgt = self.target(wall)
        if tgt < self.target_now - 0.05:
            self.last_lowered = t
        self.target_now = tgt
        call = self.call
        if self.reading < tgt - HYSTERESIS or self.reading < FROST_C:
            call = True
        elif self.reading > tgt + HYSTERESIS:
            call = False
        if call and not self.call:
            self.call_since = t
        self.call = call
        self.months = max(0.0, self.months - dt * self.ageing / (30.44 * 86400))
        self.wifi_up = not (self.link_ep.active(t) or staged(self.sc, "link", t))

    def guard(self, name, t):
        """The product's guards, checked when a change is asked for and again just before it is applied, whoever asks
        (API Standard 14.4). Returns None, or why it refuses (409 guard-failed)."""
        if self.quality(t) != "good":
            return "Interlock: the room sensor's reading is not good right now, so the thermostat cannot control to a new value"
        last = self.last_change.get(name)
        if last is not None and t - last < MIN_INTERVAL_SEC:
            return f"minIntervalSec: {name} changed {t - last:.0f} s ago; one change per {MIN_INTERVAL_SEC} s"
        return None

    def apply(self, name, value, t):
        """The only way to change a setting. The firmware has the last word: its limits clamp whatever was asked."""
        if name == "setpoint":
            self.setpoint = min(MAX_C, max(FROST_C, float(value)))
        elif name == "night_setpoint":
            self.night_setpoint = None if value is None else min(MAX_C, max(FROST_C, float(value)))
        elif name == "mode":
            self.mode = value
        self.last_change[name] = t


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

RULES = [
    # code, severity, condition, summary
    ("system.main-function-lost", "critical",
     lambda th, b, t: th.call and t - max(th.call_since, b.last_fired) >= 120,
     "The boiler does not heat when the thermostat calls for heat"),
    ("actuators.short-cycling", "warning", lambda th, b, t: len(b.starts) >= SHORT_CYCLE_STARTS,
     "The boiler switches on and off far too often"),
    ("sensors.stuck", "error", lambda th, b, t: th.possible and t - th.unchanged_since >= 90,
     "The room temperature reading has stopped changing"),
    ("sensors.out-of-range", "error", lambda th, b, t: not th.possible, "The room sensor reads an impossible temperature"),
    ("performance.overshoot", "warning",
     lambda th, b, t: th.mode != "off" and th.possible and th.reading > th.target_now + OVERSHOOT_C
     and t - th.last_lowered >= 1800,
     "The room is much warmer than its target"),
    ("power.battery-low", "warning", lambda th, b, t: th.months <= BATTERY_LOW_MONTHS,
     "The thermostat's batteries are low"),
    ("connectivity.link-lost", "warning", lambda th, b, t: not th.wifi_up, "The thermostat has lost its Wi-Fi link"),
]

# What a person reads. A3 asks for a fix where the maker lists one; for these faults HEART has none it may make.
EXPLAIN = {
    "system.main-function-lost": lambda th, b: f"Called for heat for over 2 minutes, but the burner has not fired"
    f"{' (the boiler reports a lockout)' if b.lockout else ''}. The house will get cold. Check the boiler's display and "
    "gas supply, or call the installer. No fix HEART can make: it never resets a boiler.",
    "actuators.short-cycling": lambda th, b: f"{len(b.starts)} burner starts in 10 minutes, each flame lasting under a minute. "
    "This wears the boiler and wastes gas. Bleed the radiators and check the pump, or call the installer. "
    "No fix HEART can make: the anti-short-cycle timers are the firmware's.",
    "sensors.stuck": lambda th, b: f"The room reading has been exactly {th.raw:.2f} C for over 90 s: a real room never holds "
    "still to 0.01 C that long. The thermostat is controlling to a frozen value. Change the batteries or restart the wall unit.",
    "sensors.out-of-range": lambda th, b: f"The room sensor reads {th.raw:.1f} C. The firmware ignores it and keeps the last "
    f"good reading ({th.reading:.1f} C). The sensor needs replacing.",
    "performance.overshoot": lambda th, b: f"Room {th.reading:.1f} C against a target of {th.target_now:.1f} C. The heating is "
    "off, so the heat comes from elsewhere (strong sun, another heater). Close the blinds or open a window: HEART cannot cool.",
    "power.battery-low": lambda th, b: f"About {th.months * 30.44:.0f} days left ({th.voltage} V). Replace both AA cells.",
    "connectivity.link-lost": lambda th, b: "The thermostat goes on heating as set, but cannot report until the Wi-Fi is "
    "back. Check the router.",
}

CAUSE = {  # A2-style cause and confidence, for the owner
    "system.main-function-lost": ("No ignition or a boiler lockout", "medium"),
    "actuators.short-cycling": ("The flame cuts out after seconds: likely low water flow (air in the radiators, a stuck pump)", "medium"),
    "sensors.stuck": ("The wall unit's sensor or its radio has frozen", "high"),
    "sensors.out-of-range": ("A broken room sensor", "high"),
    "performance.overshoot": ("Heat from outside the heating system, most likely sun", "medium"),
    "power.battery-low": ("The wall unit's AA cells are nearly flat", "high"),
    "connectivity.link-lost": ("The hub cannot reach the Wi-Fi router", "high"),
}


class Heart:
    """Minimum Tending (notice, change status, raise, explain, escalate), and the gate for changes (Part 6)."""

    def __init__(self, thermo, boiler, house, args):
        self.th, self.b, self.house, self.args = thermo, boiler, house, args
        self.lock = threading.RLock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)
        self.actions, self.events, self.rhythm_seq = [], [], 0
        self.max, self.stopped_at, self.t = PROMISED_LEVEL, None, 0.0     # max: the autonomy ceiling a person set
        self.last_press = 0.0

    @property
    def current(self):
        """The current level: the lower of the ceiling and the promised level (API Standard 13.4)."""
        return min(self.max, PROMISED_LEVEL)

    @property
    def mode(self):
        """Never set: it follows from the current level."""
        return "observe" if self.current <= 2 else "suggest" if self.current == 3 else "act"

    # ---- noticing

    def tend(self, t):
        with self.lock:
            self.t = t
            self.sequence += 1
            for code, severity, cond, summary in RULES:
                hit = cond(self.th, self.b, t)
                # 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:
                    text, conf = CAUSE[code]
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": summary, "explanation": EXPLAIN[code](self.th, self.b),
                         "cause": {"text": text, "confidence": conf}}
                    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"
            self.expire()

    def escalate(self, a):
        """CHANGE for a product: the owner's app, and the thermostat's own display. Here: the console."""
        a["status"] = "escalated"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}")

    def rhythm(self, name, category, privacy, summary, **attributes):
        with self.lock:
            self.rhythm_seq += 1
            e = {"sequence": self.rhythm_seq, "name": name, "category": category, "summary": summary,
                 "_time": utc(), "_privacy": privacy}
            if attributes:
                e["attributes"] = attributes
            self.events = (self.events + [e])[-300:]

    # ---- the gate (Part 6, Section 2.3; API Standard 6.3)

    def waiting(self):
        return next((a for a in self.actions if a["status"] == "awaiting-approval"), None)

    def new_action(self, kind, operation, arguments, physical, risk, title, change, explanation):
        """One Action waits for the button at a time, so one press can never approve the wrong thing. A person's newer
        request replaces a waiting one; the mock AI waits its turn."""
        with self.lock:
            w = self.waiting()
            if w and kind == "ai":
                return None
            if w:
                self.finish(w, "cancelled", "Replaced by a newer request")
            now = time.time()
            a = {"id": str(uuid.uuid4()), "operation": operation, "status": "awaiting-approval", "risk": risk,
                 "physical": physical, "createdAt": utc(now), "updatedAt": utc(now),
                 "expiresAt": utc(now + self.args.approval_timeout),
                 "initiator": {"kind": "human", "id": "owner"} if kind == "human" else
                 {"kind": "ai", "id": "mock-ai", "onBehalfOf": "owner"},
                 "arguments": arguments, "explanation": explanation,
                 "_expires": now + self.args.approval_timeout, "_created": now, "_title": title, "_change": change}
            a["digest"] = hashlib.sha256(canonical({"system": SYSTEM_ID, "id": a["id"], "operation": operation,
                                                    "arguments": arguments, "physical": physical})).hexdigest()
            self.actions = (self.actions + [a])[-50:]
            self.rhythm("tending.action.requested", "tending", "internal", f"{title}: {change}, asked by {kind}",
                        action=a["id"])
            log(f"THERMOSTAT DISPLAY  {title}: {change}?  code {a['digest'][:6].upper()}  "
                f"(press the button within {self.args.approval_timeout} s: touch button.press)")
            return a

    def finish(self, a, status, error=None):
        a["status"], a["updatedAt"] = status, utc()
        if error:
            a["error"] = error
        self.rhythm(f"tending.action.{status}", "tending", "internal",
                    f"{a['_title']}: {a['_change']}: {status}{'. ' + error if error else ''}", action=a["id"])
        if status in ("expired", "cancelled", "denied", "failed"):
            log(f"HEART action {status}: {a['_title']} ({a['_change']}){'. ' + error if error else ''}")

    def decide(self, a, decision, method=None):
        """approve (with a proof of presence already checked), deny or cancel."""
        with self.lock:
            if a["status"] != "awaiting-approval":
                return False
            a["decision"] = {"decision": decision, "at": utc(), "by": "owner"}
            if decision != "approve":
                self.finish(a, "denied" if decision == "deny" else "cancelled")
                return True
            a["decision"]["presenceMethod"] = method
            a["status"] = "approved"
            self.run(a)
            return True

    def run(self, a):
        a["status"] = "running"
        args = a["arguments"]
        if a["operation"] == "PUT /heart/v1/tending/autonomy":
            self.max, self.stopped_at = args["max"], None
            a["result"] = {"max": self.max, "current": self.current, "mode": self.mode}
            self.finish(a, "succeeded")
            log("HEART resumed: mode suggest. The mock AI may suggest again; every suggestion still waits for a yes.")
            return
        name = args["name"].rsplit(".", 1)[1]
        why = self.th.guard(name, self.t)        # guards again, just before sending, whoever asked
        if why:
            self.finish(a, "failed", f"guard-failed: {why}")
            return
        self.th.apply(name, args["value"], self.t)
        a["result"] = {"name": args["name"], "value": num(getattr(self.th, name))}
        self.finish(a, "succeeded")
        log(f"HEART applied {a['_title']}: {a['_change']} (approved by {a['decision']['presenceMethod']})")

    def expire(self):
        for a in self.actions:
            if a["status"] == "awaiting-approval" and time.time() > a["_expires"]:
                self.finish(a, "expired", "Nobody said yes in time. Nothing changed.")

    def button(self, cwd="."):
        """The thermostat's own button: a short press approves the one waiting Action; holding it stops HEART."""
        if os.path.exists(os.path.join(cwd, "button.hold")):
            os.remove(os.path.join(cwd, "button.hold"))
            self.stop("the button was held")
        p = os.path.join(cwd, "button.press")
        if os.path.exists(p):
            os.remove(p)
            with self.lock:
                a = self.waiting()
                if a:
                    self.decide(a, "approve", "device-button")
                else:
                    log("THERMOSTAT button pressed: nothing is waiting")

    def stop(self, why):
        """Stop needs no approval and no internet (Part 6, Section 6)."""
        with self.lock:
            self.max, self.stopped_at = min(self.max, 2), utc()
            for a in self.actions:
                if a["status"] == "awaiting-approval":
                    self.finish(a, "cancelled", "Stopped")
            self.rhythm("tending.stopped", "tending", "internal", f"Stopped: {why}")
            log(f"HEART stopped ({why}): mode observe. It still notices; nothing automatic is suggested until a person resumes it.")

    # ---- what it serves

    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:
                state = "off" if self.th.mode == "off" else "heating" if self.b.firing else "calling" if self.th.call else "idle"
                d["reconnaissance"] = {"state": state, "lastEvent": self.events[-1]["name"] if self.events else "none"}
                d["autonomy"] = {"max": self.max, "current": self.current}
                if self.current < PROMISED_LEVEL:
                    d["autonomy"]["limitedBy"] = "stopped"
                done = [a for a in self.actions if a["status"] != "awaiting-approval"]
                d["tending"] = {"openIssues": sum(a["status"] != "resolved" for a in self.alerts),
                                "lastAction": done[-1]["id"] if done else None}
        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 action_msg(self, a):
        return self.envelope("heart.action", {k: v for k, v in a.items() if not k.startswith("_")}, "private")

    def control(self, owner=False):
        d = {"max": self.max, "current": self.current, "mode": self.mode}
        if self.stopped_at:
            d["stoppedAt"] = self.stopped_at
        if owner:
            d.update(approvalTimeoutSec=self.args.approval_timeout, remoteSitePhysical=self.args.monitor_approvals_physical)
        return self.envelope("heart.control", d, "internal" if owner else "public")

    def rhythm_msgs(self):
        with self.lock:
            return [self.envelope("heart.rhythm", {k: v for k, v in e.items() if not k.startswith("_")}, e["_privacy"])
                    for e in reversed(self.events)]

    PROPS = {  # name: (unit, privacy, writable)
        "sensor_battery_status": (None, "public", False),
        "room_temperature": ("Cel", "internal", False),
        "sensor_quality": (None, "internal", False),
        "setpoint": ("Cel", "internal", True),
        "mode": (None, "internal", True),
        "target_temperature": ("Cel", "internal", False),
        "call_for_heat": (None, "internal", False),
        "boiler_firing": (None, "internal", False),
        "boiler_lockout": (None, "internal", False),
        "flow_temperature": ("Cel", "internal", False),
        "burner_starts_10min": ("1", "internal", False),
        "outdoor_temperature": ("Cel", "internal", False),
        "sensor_battery_voltage": ("V", "internal", False),
        "wifi_link": (None, "internal", False),
        "frost_protection": ("Cel", "internal", False),
        "max_setpoint": ("Cel", "internal", False),
        "night_setpoint": ("Cel", "private", True),
        "night_hours": (None, "private", False),
        "occupied": (None, "private", False),
        "people_home": (None, "private", False),
        "window_open": (None, "private", False),
    }

    def value(self, name):
        th, b, h = self.th, self.b, self.house
        return {
            "sensor_battery_status": "low" if th.months <= BATTERY_LOW_MONTHS else "ok",
            "room_temperature": round(th.reading, 2), "sensor_quality": th.quality(self.t),
            "setpoint": num(th.setpoint), "mode": th.mode, "target_temperature": num(round(th.target_now, 1)),
            "call_for_heat": th.call, "boiler_firing": b.firing, "boiler_lockout": b.lockout,
            "flow_temperature": round(b.flow, 1), "burner_starts_10min": len(b.starts),
            "outdoor_temperature": round(h.outdoor, 1), "sensor_battery_voltage": th.voltage,
            "wifi_link": "up" if th.wifi_up else "down", "frost_protection": num(FROST_C), "max_setpoint": num(MAX_C),
            "night_setpoint": None if th.night_setpoint is None else num(th.night_setpoint),
            "night_hours": f"{NIGHT_HOURS[0]:02d}:00-{NIGHT_HOURS[1]:02d}:00",
            "occupied": h.occupied, "people_home": h.people > 0, "window_open": h.window,
        }[name]

    def prop(self, name):
        unit, privacy, writable = self.PROPS[name]
        d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": self.value(name), "writable": writable, "time": utc()}
        if unit:
            d["unit"] = unit
        if name == "room_temperature":
            d["quality"] = self.th.quality(self.t)
        return self.envelope("heart.property", d, privacy)

    def properties(self, owner=False):
        with self.lock:
            return [self.prop(n) for n, (_u, p, _w) in self.PROPS.items() if owner or p == "public"]


class MockAI:
    """MOCK AI. A stand-in for a model, so the demo needs no network and no provider: a few plain rules over the
    private occupancy log. Whatever it suggests is an `ai`-initiated, physical Action, so the gate treats it as
    destructive and it always waits for a person's yes (a press of the thermostat's button, or, if the owner allowed
    it, a signed approval from the monitor site). It never applies anything itself."""

    def __init__(self, heart, scenarios):
        self.h, self.next_at = heart, AI_FIRST_SEC if "ai" in scenarios else 1800
        self.again = AI_AGAIN_SEC if "ai" in scenarios else 6 * 3600
        self.mine = None

    def step(self, t):
        h = self.h
        with h.lock:
            if self.mine and self.mine["status"] in ("awaiting-approval", "approved", "running"):
                return
            if self.mine and self.mine["status"] != "succeeded" and self.next_at <= t:
                self.next_at = t + self.again       # said no, or did not answer: ask again much later
                self.mine = None
                return
            if t < self.next_at or h.mode != "suggest" or h.th.mode != "heat":
                return
            if h.th.night_setpoint is not None and h.th.night_setpoint <= 18:
                return
            now = h.th.night_setpoint if h.th.night_setpoint is not None else h.th.setpoint
            name = f"{SYSTEM_ID}.night_setpoint"
            a = h.new_action("ai", f"PUT /heart/v1/properties/{name}", {"name": name, "value": 18}, True, "destructive",
                             "Night temperature", f"{num(now)} to 18 °C at night",
                             "MOCK AI suggestion (plain rules, not a model): the living room has been empty "
                             f"{NIGHT_HOURS[0]:02d}:00-{NIGHT_HOURS[1]:02d}:00 on recent nights, so heating it to "
                             f"{num(now)} °C then is wasted. Lower the night temperature to 18 °C. This changes only the "
                             "night setpoint: the day setpoint, the mode, frost protection and the boiler's limits stay "
                             "as they are. It waits for your yes: press the thermostat's button.")
            if a:
                self.mine, self.next_at = a, t + self.again
                h.rhythm("ai.suggestion.made", "ai", "private", "The mock AI suggested a lower night temperature",
                         action=a["id"])


# ------------------------------------------------------------------------------------------------ Ed25519 (RFC 8032)
# Copied from heart-templates/templates/python/heart.py: verifies the monitor site's signed decisions.

_P = 2 ** 255 - 19
_L = 2 ** 252 + 27742317777372353535851937790883648493
_D = -121665 * pow(121666, _P - 2, _P) % _P
_SQRT_M1 = pow(2, (_P - 1) // 4, _P)


def _x_from_y(y, sign):
    xx = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P
    x = pow(xx, (_P + 3) // 8, _P)
    if (x * x - xx) % _P:
        x = x * _SQRT_M1 % _P
    if (x * x - xx) % _P or (x == 0 and sign):
        return None
    return _P - x if x & 1 != sign else x


def _point(b):
    y = int.from_bytes(b, "little")
    sign, y = y >> 255, y & ((1 << 255) - 1)
    x = _x_from_y(y, sign) if y < _P else None
    return None if x is None else (x, y, 1, x * y % _P)


def _add(p, q):
    a = (p[1] - p[0]) * (q[1] - q[0]) % _P
    b = (p[1] + p[0]) * (q[1] + q[0]) % _P
    c, d = 2 * p[3] * q[3] * _D % _P, 2 * p[2] * q[2] % _P
    e, f, g, h = b - a, d - c, d + c, b + a
    return (e * f % _P, g * h % _P, f * g % _P, e * h % _P)


def _times(n, p):
    q = (0, 1, 1, 0)
    while n:
        if n & 1:
            q = _add(q, p)
        p, n = _add(p, p), n >> 1
    return q


_BY = 4 * pow(5, _P - 2, _P) % _P
_BASE = (_x_from_y(_BY, 0), _BY, 1, _x_from_y(_BY, 0) * _BY % _P)


def ed25519_verify(public_key, message, signature):
    """True only if `signature` (64 bytes) is a valid Ed25519 signature of `message` by `public_key` (32 bytes)."""
    if len(public_key) != 32 or len(signature) != 64:
        return False
    a, r, s = _point(public_key), _point(signature[:32]), int.from_bytes(signature[32:], "little")
    if a is None or r is None or s >= _L:
        return False
    h = int.from_bytes(hashlib.sha512(signature[:32] + public_key + message).digest(), "little") % _L
    left, right = _times(s, _BASE), _add(r, _times(h, a))
    return (left[0] * right[2] - right[0] * left[2]) % _P == 0 and (left[1] * right[2] - right[1] * left[2]) % _P == 0


def b64url(text):
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


DECISION_MAX_AGE_SEC = 300


def decision_ok(d, site_key):
    """The remote-site checks of API Standard 6.3: the pinned site key signed exactly these four fields, and recently."""
    try:
        fields = {k: d[k] for k in ("actionId", "at", "decision", "digest")}
        if not all(isinstance(v, str) for v in fields.values()) or not site_key:
            return False
        if not ed25519_verify(b64url(site_key), canonical(fields), b64url(d["signature"])):
            return False
        age = time.time() - calendar.timegm(time.strptime(fields["at"], "%Y-%m-%dT%H:%M:%SZ"))
        return -60 <= age <= DECISION_MAX_AGE_SEC
    except (KeyError, TypeError, ValueError):
        return False


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

def save_identity(path, ident):
    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)


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)}
    save_identity(path, ident)
    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 thermostat
    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-thermostat-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 offered(heart, args):
    """Waiting Actions the owner may approve on the monitor site: only with --monitor-approvals, physical ones only
    with --monitor-approvals-physical too, and never a resume (raising autonomy needs a person at the device)."""
    if not args.monitor_approvals:
        return []
    with heart.lock:
        return [a for a in heart.actions if a["status"] == "awaiting-approval" and a["operation"] != "PUT /heart/v1/tending/autonomy"
                and (not a["physical"] or args.monitor_approvals_physical)][:4]


def decisions(heart, reply, site_key, args):
    """A decision counts only for a change this device offered, with the digest the owner saw, signed by the site key
    pinned at hello, and at most 300 seconds old. The device's own gate, guards and limits still apply."""
    for d in reply.get("decisions") or []:
        if not decision_ok(d, site_key):
            continue
        with heart.lock:
            a = next((x for x in offered(heart, args) if x["id"] == d.get("actionId")), None)
            if a and d.get("digest") == a["digest"] and d.get("decision") in ("approve", "deny"):
                log(f"HEART Monitor: the owner said {d['decision']} on the site, signature checked")
                heart.decide(a, d["decision"], "remote-site")


def monitor(heart, th, args, 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. It takes no commands from
    the site: only a yes or no to a change this device itself offered (--monitor-approvals)."""
    url, path = args.monitor_url, args.state_file
    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:]),
                    tuple(a["id"] for a in offered(heart, args)))
        due = time.time() >= next_at or (phase == "report" and seen != sent and time.time() - last >= 10)
        if not due or (phase == "report" and not th.wifi_up):
            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
                    if reply.get("siteKey") and not ident.get("siteKey"):     # pinned once, at the first hello
                        ident["siteKey"] = reply["siteKey"]
                        save_identity(path, ident)
                    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
            body = {"pulse": heart.pulse(), "alerts": heart.public_alerts()}
            asks = [{"actionId": a["id"], "digest": a["digest"], "title": a["_title"][:120], "change": a["_change"][:200],
                     "physical": a["physical"], "expiresAt": a["expiresAt"]} for a in offered(heart, args)]
            if asks:
                body["approvals"] = asks
            status, reply = post(url, f"/v1/devices/{ident['id']}/report", body, ident["key"])
            backoff = 0
            if status == 200 and args.monitor_approvals:
                decisions(heart, reply, ident.get("siteKey"), args)
            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):
    th = heart.th
    done = {}   # Idempotency-Key results: (key, method, path, body) -> (status, reply)

    class H(BaseHTTPRequestHandler):
        def send(self, status, body, ctype="application/cloudevents+json"):
            raw = json.dumps(body, indent=1, ensure_ascii=False).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", "control"],
                    "shares": {"pulse": "public", "arrhythmia": "public", "property": "public", "control": "public",
                               "action": "private", "rhythm": "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.startswith("/heart/v1/properties/"):
                name = short(path.rsplit("/", 1)[1])
                if name in heart.PROPS and (owner or heart.PROPS[name][1] == "public"):
                    with heart.lock:
                        return self.send(200, heart.prop(name))
                return self.problem(404, "not-found", "No such property")       # never says that a private one exists
            if path == "/heart/v1/tending/autonomy":
                return self.send(200, heart.control(owner))
            if path in ("/heart/v1/actions", "/heart/v1/rhythm") or path.startswith("/heart/v1/actions/"):
                if not owner:
                    return self.problem(401, "unauthenticated", "Actions and the Rhythm are private: send the owner key")
                with heart.lock:
                    if path == "/heart/v1/rhythm":
                        return self.send(200, {"items": heart.rhythm_msgs()}, "application/json")
                    if path == "/heart/v1/actions":
                        return self.send(200, {"items": [heart.action_msg(a) for a in reversed(heart.actions)]},
                                         "application/json")
                    a = next((a for a in heart.actions if a["id"] == path.rsplit("/", 1)[1]), None)
                    return self.send(200, heart.action_msg(a)) if a else self.problem(404, "not-found", "No such action")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties, /actions, /tending/autonomy or /rhythm")

        def body(self):
            try:
                n = int(self.headers.get("Content-Length") or 0)
                raw = self.rfile.read(n) if n else b""
                return raw, (json.loads(raw) if raw else {})
            except ValueError:
                return b"", None

        def change(self, method):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            raw, body = self.body()
            if not owner:
                return self.problem(401, "unauthenticated", "Changes need the owner key")
            if not isinstance(body, dict):
                return self.problem(400, "invalid-request", "Send a JSON object")
            key = self.headers.get("Idempotency-Key")
            if key:          # a repeat with the same key and body gets the first answer, and runs nothing twice
                k = (key, method, path, raw)
                if k in done:
                    return self.send(*done[k])
                if any(x[:3] == k[:3] for x in done):
                    return self.problem(422, "idempotency-conflict", "That Idempotency-Key was used with another body")
            status, reply = self.route(method, path, body)
            if key and status < 500:
                done[(key, method, path, raw)] = (status, reply) if status < 400 else (status, reply, "application/problem+json")
            ctype = "application/problem+json" if status >= 400 else "application/cloudevents+json"
            self.send(status, reply, ctype)

        def route(self, method, path, body):
            def problem(status, code, detail):
                return status, {"type": f"https://heartcortex.ai/problems/{code}", "title": code.replace("-", " "),
                                "status": status, "detail": detail, "code": code}

            if "presence" in body:
                return problem(403, "presence-invalid", "This thermostat accepts presence only as a press of its own "
                               "button (and, if allowed, a signed approval from its monitor site), never through the API")
            if method == "PUT" and path.startswith("/heart/v1/properties/"):
                name = short(path.rsplit("/", 1)[1])
                if name not in heart.PROPS:
                    return problem(404, "not-found", "No such property")
                if not heart.PROPS[name][2]:
                    return problem(409, "conflict", f"{name} is read-only. The firmware's limits "
                                   "(frost protection, maximum, anti-short-cycle) and the boiler can never be changed through HEART.")
                if "value" not in body:
                    return problem(400, "invalid-request", 'Send {"value": ...}')
                v = body["value"]
                if name == "mode":
                    if v not in MODES:
                        return problem(422, "out-of-range", "mode is heat, eco or off")
                    title, change = "Mode", f"{th.mode} to {v}"
                else:
                    if not (v is None and name == "night_setpoint") and (isinstance(v, bool) or not isinstance(v, (int, float))
                                                                         or not math.isfinite(v)):
                        return problem(400, "invalid-request", "The value must be a number of degrees Celsius")
                    if v is not None and not SETPOINT_RANGE[0] <= v <= SETPOINT_RANGE[1]:
                        return problem(422, "out-of-range", f"{name} must be {num(SETPOINT_RANGE[0])} to "
                                       f"{num(SETPOINT_RANGE[1])} °C. Frost protection ({num(FROST_C)} °C) stays on whatever you set.")
                    v = None if v is None else num(round(float(v), 1))
                    old = getattr(th, name)
                    title = "Setpoint" if name == "setpoint" else "Night temperature"
                    change = f"{'none' if old is None else num(old)} to {'none' if v is None else str(v) + ' °C'}"
                with heart.lock:
                    why = th.guard(name, heart.t)
                    if why:
                        heart.rhythm("tending.guard.refused", "tending", "internal", why)
                        return problem(409, "guard-failed", why)
                    full = f"{SYSTEM_ID}.{name}"
                    # Heating acts on the physical world: a person's own change needs proof of presence (Part 6, 2.3).
                    a = heart.new_action("human", f"PUT /heart/v1/properties/{full}", {"name": full, "value": v}, True,
                                         "write", title, change, "")
                    a["explanation"] = (f"Press the thermostat's button within {heart.args.approval_timeout} s to approve. "
                                        f"Its display shows code {a['digest'][:6].upper()}.")
                    return 202, heart.action_msg(a)
            if method == "POST" and path.startswith("/heart/v1/actions/") and path.endswith("/decision"):
                aid = path.split("/")[4]
                with heart.lock:
                    a = next((x for x in heart.actions if x["id"] == aid), None)
                    if not a:
                        return problem(404, "not-found", "No such action")
                    if a["status"] != "awaiting-approval":
                        return problem(409, "conflict", f"This action is no longer waiting: it is {a['status']}")
                    d = body.get("decision")
                    if d == "approve":
                        return problem(403, "presence-required", "Approve by pressing the button on the thermostat. "
                                       "A key alone, or an AI that holds it, can never approve a physical change.")
                    if d not in ("deny", "cancel"):
                        return problem(400, "invalid-request", "decision: approve, deny or cancel")
                    heart.decide(a, d)
                    return 200, heart.action_msg(a)
            if method == "POST" and path == "/heart/v1/tending/stop":
                heart.stop("the owner asked")
                return 200, heart.control(True)
            if method == "PUT" and path == "/heart/v1/tending/autonomy":
                m = body.get("max")
                if (not isinstance(m, int) or isinstance(m, bool) or not (0 <= m <= 2 or m == PROMISED_LEVEL)
                        or set(body) - {"max", "reason", "actionId", "presence"}):
                    return problem(400, "invalid-request", f"This thermostat offers a max of 0 to 2 (Stop) or {PROMISED_LEVEL} "
                                   "(suggest): it never acts on its own")
                if m <= 2:
                    heart.stop("the owner lowered autonomy")
                    heart.max = min(m, heart.max)
                    return 200, heart.control(True)
                if m <= heart.max:
                    return 200, heart.control(True)
                if "presence" in body:
                    return problem(403, "presence-invalid", "Approve by pressing the button on the thermostat.")
                # Raising autonomy is a change to authority: destructive, so a person presses the button (Part 6, 4.1).
                a = heart.new_action("human", "PUT /heart/v1/tending/autonomy", {"max": m}, False, "destructive",
                                     "Resume suggestions", f"autonomy ceiling {heart.max} to {m}",
                                     "Press the thermostat's button to let the mock AI suggest again.")
                return 202, heart.action_msg(a)
            return problem(404, "not-found", "Changes: PUT /heart/v1/properties/{name}, POST /heart/v1/actions/{id}/decision, "
                           "POST /heart/v1/tending/stop, PUT /heart/v1/tending/autonomy")

        def do_PUT(self):
            self.change("PUT")

        def do_POST(self):
            self.change("POST")

        def do_PATCH(self):
            self.body()
            self.problem(403, "capability-not-allowed", "Use PUT /heart/v1/properties/{name}")

        do_DELETE = do_PATCH

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


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

INSTANCE = "sim"


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated smart thermostat 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("--monitor-approvals", action="store_true",
                    help="offer waiting changes to the owner on the monitor site (signed decisions, method remote-site)")
    ap.add_argument("--monitor-approvals-physical", action="store_true",
                    help="also offer physical changes there (remoteSitePhysical). Off: they need the thermostat's button")
    ap.add_argument("--approval-timeout", type=int, default=900, help="real seconds a change waits for a yes (at least 60)")
    ap.add_argument("--state-file", default="thermostat-monitor.json", help="the device ID, claim code, key and site key")
    ap.add_argument("--port", type=int, default=43278, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help=f"{','.join(sorted(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 - SCENARIOS
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")
    if args.approval_timeout < 60:
        ap.error("--approval-timeout is at least 60 seconds (approvalTimeoutSec)")
    if args.monitor_approvals_physical and not args.monitor_approvals:
        ap.error("--monitor-approvals-physical needs --monitor-approvals")

    rng = random.Random(args.seed)
    th, boiler = Thermostat(rng, scenarios), Boiler()
    heart = Heart(th, boiler, None, args)
    house = House(rng, scenarios, heart.rhythm)
    heart.house = house
    ai = MockAI(heart, scenarios)
    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. Simulated house, thermostat and boiler. HEART is not in the control loop and never overrides the
  firmware's limits. Every change waits for a press of the thermostat's button: touch button.press here.
  Scenarios: {', '.join(sorted(scenarios)) or 'none (random life only)'} · speed {args.speed}x · approvals on the monitor site: \
{'on, physical too' if args.monitor_approvals_physical else 'on, not physical' if args.monitor_approvals else 'off'}
""", 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, th, args, stop), daemon=True).start()

    t, wall0, last_print = 0.0, time.time(), 0.0
    try:
        while not stop.is_set():
            n = max(1, math.ceil(args.speed / 5))         # physics in steps of at most 5 simulated seconds
            for _ in range(n):
                dt = args.speed / n
                t += dt
                with heart.lock:
                    house.step(t, dt, wall0 + t, boiler.firing)
                    th.step(t, dt, wall0 + t, house, boiler)
                    boiler.step(t, dt, th.call, staged(scenarios, "boiler", t), staged(scenarios, "cycling", t))
            heart.tend(t)
            heart.button()
            ai.step(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                log(f"room {th.reading:5.1f} C · target {th.target_now:4.1f} ({th.mode}) · "
                    f"{'CALL' if th.call else 'idle'} · boiler {'ON ' if boiler.firing else 'off'} flow {boiler.flow:4.1f} C "
                    f"starts {len(boiler.starts)} · outdoor {house.outdoor:5.1f} C · battery {th.voltage} V · "
                    f"wifi {'up' if th.wifi_up else 'down'} · HEART {heart.status}")
            if args.duration and time.time() - wall0 >= args.duration:
                break
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    stop.set()
    log("stopped. The monitor site will show a Flatline if this device stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo