Birdhouse camera: what is inside

A solar nest box camera. Two Playbooks the owner adopted move its zoom, inside the owner's envelope; the infrared light changes only at the button. HEART runs on the camera itself: an ESP32-S3 camera board in the nest box, with a push button outside the box.

Specifications it follows

• Part 2: Autonomy Levels (A1, Minimum Tending)
• Part 2: Autonomy Levels (A4, Fixes known problems: Playbooks inside an envelope)
• 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)
• Signal Tables (every signal declared once)

Garden  ──►  Camera  ──(reads)──►  Heart  ──►  local HEART API      (127.0.0.1)
sun, birds    the product            watches     └►  HEART Hub    (outbound only)
nest life     vision, battery,       Playbooks ──(gate: allowlist + envelope)──► camera.zoom
              card, Wi-Fi, IR        owner ──(button at the box)──► camera.ir_level

Run it yourself, and register it on HEART Hub

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

The code: birdhouseSimulator.py

1031 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/birdhouse-simulator/v1.2611.2
"""birdhouseSimulator.py: a simulated birdhouse camera with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. There is no camera, no box and no bird: everything is simulated, and no picture is ever made.
Its numbers are plausible, not measured. See README.md beside this file, and the case study guides/Birdhouse-Camera.md.

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

  * a simulated GARDEN: birds visit the box on random timers (more at dawn and dusk, none at night), sometimes
    several at once; over simulated days a pair builds a nest, lays 1 to 6 eggs, and the chicks hatch and fledge;
  * a simulated CAMERA (the product): it takes pictures when a bird comes in, its own on-chip vision counts the
    birds in view and the eggs it sees (with a confidence), and it has a solar-charged battery, a box temperature,
    an SD card, a Wi-Fi link, an infrared (IR) light for the night and a zoom;
  * a HEART that WATCHES the camera: 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;
  * two adopted Playbooks that CHANGE ONE THING: the zoom. "3 or more birds in view for 10 s: zoom out one step" and
    "eggs seen clearly with at most one bird in view: zoom in one step, towards 3.0". The zoom is not physical (it
    changes the picture, nothing in the box), and it is on the owner's allowlist for initiator rule with an envelope
    (1.0 to 4.0, at most 1.0 per step, 12 a day, 300 s apart). Every change is an Action record; a change outside
    the envelope is refused, never clipped. This is a demonstration of the mechanism, not a claim of Autonomy A4.
  * the owner may set the zoom at once with the owner key. The IR light is PHYSICAL (light shining on living birds):
    a change waits until a person at the box presses the button (touch button.press in the working directory);
    an approval over the network with only the key is refused with 403 presence-required.

Run it:

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

Scenarios (--scenario, comma-separated): storage, battery, heat, camera, wifi, crowd, eggs, all. Each fault starts at
a set time, lasts a while, and is then fixed by the owner, so you can watch an alert open, escalate and resolve.
crowd brings flocks of birds and eggs lays a clutch at once, so the zoom Playbooks visibly fire.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. Readings,
bird counts, eggs, the visit timeline and the zoom stay local. Pictures do not exist here, and in a real camera they
never pass through HEART. The location of the box is not stored anywhere.
"""
import argparse
import hashlib
import json
import math
import os
import random
import secrets
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SYSTEM_ID = "org.heartaispec.demo.birdhouse"
NAME = "Birdhouse Camera Simulator (demo)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/birdhouse-simulator/v1.2611.2"
PULSE_SEC = 30
PROMISED_LEVEL = 4                        # the level its control offers: max 4 is mode act (API Standard 13.4)
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
DAY = 86400.0

# The product's own limits. They live in the camera's firmware; HEART reads them, and cannot widen them.
ZOOM_RANGE = (1.0, 4.0)                   # optical-plus-digital zoom the camera can do
IR_MAX = 60                               # the brightest the firmware ever lets the IR light shine, %
IR_GUARD_TEMP = 35.0                      # interlock: the IR light (it warms the box) is never raised above this, Cel
BATTERY_LOW, TOO_HOT, CARD_LOW = 20.0, 40.0, 10.0   # healthy ranges, as in the case study's signal table

# The owner's allowlist entry for the zoom Playbooks (Part 6, Section 2.3; API Standard 6.3).
ZOOM_ENVELOPE = {"arguments": {"zoom": {"min": 1.0, "max": 4.0, "maxStep": 1.0}}, "maxPerDay": 12, "cooldownSec": 300}
RULE_HOLD_SEC = 10                        # a Playbook's condition must hold this long before it fires

SCENARIOS = ("storage", "battery", "heat", "camera", "wifi", "crowd", "eggs")
# When each staged event starts and how long it lasts, in simulated seconds.
STAGED = {"storage": [(60, 300)], "eggs": [(150, None)], "crowd": [(300, 150), (780, 150)], "heat": [(420, 180)],
          "wifi": [(660, 240)], "camera": [(960, 180)], "battery": [(0, 1150)]}


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 canonical(v):
    """RFC 8785 for the simple values used here: sorted keys, no spaces, integral numbers without a fraction."""
    if isinstance(v, dict):
        return "{" + ",".join(json.dumps(k) + ":" + canonical(v[k]) for k in sorted(v)) + "}"
    if isinstance(v, list):
        return "[" + ",".join(canonical(x) for x in v) + "]"
    if isinstance(v, float) and v.is_integer():
        return str(int(v))
    return json.dumps(v, ensure_ascii=False)


# ------------------------------------------------------------------------------------------------ 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):
    """True while a staged event of this scenario is running."""
    return name in scenarios and any(s <= t and (n is None or t < s + n) for s, n in STAGED[name])


def staged_ended(scenarios, name, t):
    return name in scenarios and any(n is not None and t >= s + n for s, n in STAGED[name])


class Garden:
    """The simulated world: the hour of the day, the sun, and the birds. Nothing here is a product value."""

    STAGES = ("empty", "building", "eggs", "incubating", "chicks", "fledged")

    def __init__(self, rng, scenarios, clock0):
        self.rng, self.sc, self.clock0 = rng, scenarios, clock0
        self.visits = []                    # [end time, birds] of the visits now at the box
        self.next_visit = rng.uniform(5, 60)
        self.stage, self.stage_until = "empty", rng.uniform(1, 3) * DAY
        self.clutch = self.eggs = self.chicks = 0
        self.sitting = Episode(rng, (600, 1800), (900, 3000))      # a parent sitting on the eggs, by day
        self.cloudy = Episode(rng, (3 * DAY, 9 * DAY), (DAY / 2, 2 * DAY))
        self.arrivals = 0                   # visits started so far: the camera takes pictures on each
        self.crowd_on = set()
        self.eggs_jumped = False
        self._t = 0.0

    def hour(self, t):
        lt = time.localtime(self.clock0 + t)
        return lt.tm_hour + lt.tm_min / 60 + lt.tm_sec / 3600

    def sun(self, t):
        h = self.hour(t)
        cloud = 0.0 if staged(self.sc, "battery", t) else 0.25 if self.cloudy.active(t) else 1.0
        return cloud * max(0.0, math.sin(math.pi * (h - 6) / 14)) if 6 <= h <= 20 else 0.0

    def part_of_day(self, t):
        h = self.hour(t)
        return "night" if h < 5 or h >= 21.5 else "dawn" if h < 8.5 else "dusk" if h >= 18 else "day"

    def step(self, t, dt):
        # The nest's life, over simulated days. The eggs scenario jumps straight to a fresh clutch.
        if staged(self.sc, "eggs", t) and not self.eggs_jumped:
            self.eggs_jumped = True
            self.stage, self.stage_until = "eggs", t + 2 * DAY
            self.clutch = self.eggs = self.rng.randint(1, 6)
            log(f"garden: (staged) the pair has laid {self.eggs} egg{'s' if self.eggs > 1 else ''}")
        if self.stage == "eggs" and self.eggs < self.clutch and t >= self.stage_until - (self.clutch - self.eggs) * DAY:
            self.eggs += 1                   # one egg a day until the clutch is complete
        if t >= self.stage_until:
            i = self.STAGES.index(self.stage)
            self.stage = self.STAGES[(i + 1) % len(self.STAGES)]
            if self.stage == "building":
                self.stage_until = t + self.rng.uniform(4, 7) * DAY
            elif self.stage == "eggs":
                self.clutch, self.eggs = self.rng.randint(1, 6), 1
                self.stage_until = t + self.clutch * DAY
            elif self.stage == "incubating":
                self.stage_until = t + self.rng.uniform(12, 14) * DAY
            elif self.stage == "chicks":
                self.chicks, self.eggs = self.eggs, 0
                self.stage_until = t + self.rng.uniform(15, 19) * DAY
            elif self.stage == "fledged":
                self.chicks, self.stage_until = 0, t + DAY
            else:
                self.clutch, self.stage_until = 0, t + self.rng.uniform(1, 3) * DAY
            log(f"garden: the nest is now {self.stage}")

        # Visits, on random timers: busiest at dawn and dusk, none at night, more while there is a nest to tend.
        self.visits = [v for v in self.visits if v[0] > t]
        part = self.part_of_day(t)
        if t >= self.next_visit:
            if part != "night":
                n = self.rng.choices([1, 2, 3, 4, 5], [70, 18, 6, 4, 2])[0]
                self.visits.append([t + self.rng.uniform(8, 90), n])
                self.arrivals += 1
            gap = {"dawn": 90, "dusk": 120, "day": 360, "night": 600}[part]
            gap /= {"building": 2, "chicks": 3}.get(self.stage, 1)
            self.next_visit = t + self.rng.expovariate(1 / gap)
        for i, (s, n) in enumerate(STAGED["crowd"]):
            if staged(self.sc, "crowd", t) and s <= t < s + n and i not in self.crowd_on:
                self.crowd_on.add(i)
                size = self.rng.randint(4, 6)
                self.visits.append([s + n, size])
                self.arrivals += 1
                log(f"garden: (staged) a flock of {size} birds lands on the box")

    @property
    def birds(self):
        return sum(n for _, n in self.visits) + (1 if self.parent_in_box else 0)

    @property
    def parent_in_box(self):
        return self.stage == "incubating" and (self.part_of_day(self._t) == "night" or self.sitting.active(self._t))

    def at(self, t):
        self._t = t
        return self


class Camera:
    """The simulated product. Its firmware owns the camera, the pictures, the IR light and its limits. HEART reads
    these values, and may change only the zoom and the IR light, through capabilities that pass the gate."""

    def __init__(self, rng, scenarios, garden):
        self.rng, self.sc, self.g = rng, scenarios, garden
        self.battery = 23.0 if "battery" in scenarios else rng.uniform(60, 90)
        self.box_temp = 12.0 + rng.uniform(-2, 2)
        self.sd_free = rng.uniform(55, 80)
        self.rssi, self.wifi_down = -62.0, False
        self.zoom, self.ir_level = 1.0, 30
        self.camera_ok = True
        self.pictures_today, self.day_of_pictures, self.last_picture = 0, None, None
        self.seen_arrivals = 0
        self.birds_seen, self.eggs_seen, self.egg_confidence = 0, 0, 0.0
        self.t_rssi, self.t_vision = Timer(rng, 5, 20), Timer(rng, 1, 2)
        self.wifi_ep = Episode(rng, (6 * 3600, 18 * 3600), (120, 900))   # the router reboots, or leaves fall on the antenna
        self.visits_log = []                  # private visit timeline, newest last
        self.t = 0.0

    @property
    def ir_on(self):
        return self.g.part_of_day(self.t) == "night"

    def step(self, t, dt):
        self.t, g = t, self.g.at(t)
        hour = g.hour(t)
        # Staged faults, for the demo. Each lasts a while and is then fixed by the owner.
        self.camera_ok = not staged(self.sc, "camera", t)
        self.wifi_down = self.wifi_ep.active(t) or staged(self.sc, "wifi", t)

        # Battery: the camera and Wi-Fi stay awake, the IR light adds at night, the solar panel charges by day.
        fast = 40.0 if staged(self.sc, "battery", t) else 1.0             # a cloudy week, squeezed into minutes
        drain = 0.00025 + (0.0001 * self.ir_level / IR_MAX if self.ir_on else 0.0)
        self.battery += (0.0009 * g.sun(t) - drain * fast) * dt
        if "battery" in self.sc and getattr(self, "_battery_fixed", False) is False and staged_ended(self.sc, "battery", t):
            self._battery_fixed = True
            self.battery = 80.0
            log("garden: (staged) the sun is back, and the owner charged the battery")
        self.battery = min(100.0, max(0.0, self.battery))

        # Temperature in the box: the air, the sun on the roof, and the birds inside.
        outside = 12 + 7 * math.sin(math.pi * (hour - 9) / 12)
        target = outside + 14 * g.sun(t) + 0.4 * g.birds
        tau = 40.0 if staged(self.sc, "heat", t) else 120.0
        if staged(self.sc, "heat", t):
            target = 46.0                                                # a heatwave, and the box in full sun
        self.box_temp += (target - self.box_temp) * min(1.0, dt / tau)

        # The SD card: pictures fill it; the firmware deletes the oldest pictures the app has already collected.
        if staged(self.sc, "storage", t):
            s, n = STAGED["storage"][0]
            self.sd_free = max(0.0, 9.0 - 9.0 * (t - s) / 150)            # the app stopped collecting pictures
            self._card_swapped = False
        elif "storage" in self.sc and staged_ended(self.sc, "storage", t) and not getattr(self, "_card_swapped", True):
            self._card_swapped = True
            self.sd_free = 95.0
            log("garden: (staged) the owner fitted an empty card")
        elif not self.wifi_down:
            self.sd_free = min(85.0, self.sd_free + 0.0004 * dt)

        if self.t_rssi.due(t):
            self.rssi = min(-40.0, max(-85.0, self.rssi + (-62 - self.rssi) * 0.3 + self.rng.gauss(0, 2)))

        # Pictures: a burst each time a bird comes in, if the camera works and the card has room. They are counted
        # here and nothing more: no picture exists in this file, and none would ever pass through HEART.
        day = time.localtime(g.clock0 + t).tm_yday
        if day != self.day_of_pictures:
            self.day_of_pictures, self.pictures_today = day, 0
        while self.seen_arrivals < g.arrivals:
            self.seen_arrivals += 1
            n = self.rng.randint(1, 3) if self.camera_ok and self.sd_free > 0.5 else 0
            self.pictures_today += n
            self.sd_free = max(0.0, self.sd_free - 0.02 * n)
            if n:
                self.last_picture = t
            self.visits_log = (self.visits_log + [{"at": utc(), "simClock": time.strftime("%H:%M", time.localtime(g.clock0 + t)),
                                                   "birds": g.birds, "pictures": n}])[-200:]

        # The camera's own on-chip vision: how many birds are in view, how many eggs it sees, and how sure it is.
        if self.t_vision.due(t):
            if not self.camera_ok:
                self.birds_seen, self.eggs_seen, self.egg_confidence = None, None, 0.0
            else:
                fits = max(3, round(8 / self.zoom))                    # zoomed in, fewer birds fit in the picture
                self.birds_seen = min(g.birds, fits)
                eggs = g.eggs if g.stage in ("eggs", "incubating") else 0
                base = {0: 0.93, 1: 0.84, 2: 0.55}.get(min(g.birds, 3), 0.35)    # birds hide the nest cup
                conf = base + 0.02 * (self.zoom - 1) + self.rng.gauss(0, 0.015)
                if self.ir_on:
                    conf *= 0.92 if self.ir_level >= 15 else 0.3
                if g.parent_in_box:
                    eggs = 0                                           # a sitting parent covers the eggs
                self.eggs_seen = eggs
                self.egg_confidence = round(min(0.99, max(0.0, conf)), 2) if eggs else 0.0

    def main_job_problem(self):
        """CHANGE for a product: None while the camera does its main job (pictures when a bird comes in)."""
        if not self.camera_ok:
            return "The camera stopped answering: no picture is taken when a bird comes in."
        if self.sd_free <= 0.5:
            return "The card is full: pictures cannot be saved."
        return None

    # The two settings HEART may change, each behind the product's own guards.
    def set_zoom(self, z):
        self.zoom = z

    def ir_guard(self, level):
        """The interlock, in the product: the IR light warms the box, so it is never raised while the box is hot."""
        if level > self.ir_level and self.box_temp >= IR_GUARD_TEMP:
            return f"The box is {self.box_temp:.1f} Cel: the IR light is not raised above {IR_GUARD_TEMP:.0f} Cel."
        return None

    def set_ir(self, level):
        self.ir_level = level


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

RULES = [
    # code, severity, condition, summary
    ("system.main-function-lost", "critical", lambda c: c.main_job_problem() is not None, "The camera cannot take pictures"),
    ("power.battery-low", "warning", lambda c: c.battery < BATTERY_LOW, "The battery is low"),
    ("performance.too-hot", "warning", lambda c: c.box_temp > TOO_HOT, "It is too hot in the box"),
    ("storage.nearly-full", "warning", lambda c: 0.5 < c.sd_free < CARD_LOW, "The card is nearly full"),
    ("storage.full", "error", lambda c: c.sd_free <= 0.5, "The card is full"),
    ("connectivity.link-lost", "warning", lambda c: c.wifi_down, "The Wi-Fi link is lost"),
]

EXPLAIN = {
    "system.main-function-lost": lambda c: c.main_job_problem() or "",
    "power.battery-low": lambda c: f"Battery {c.battery:.1f} %, below {BATTERY_LOW:.0f} %. Too little sun to charge: clean the panel or charge the battery.",
    "performance.too-hot": lambda c: f"{c.box_temp:.1f} Cel in the box, above {TOO_HOT:.0f} Cel. Shade the box; the IR light will not be raised.",
    "storage.nearly-full": lambda c: f"{c.sd_free:.1f} % of the card is free. Let the app collect the pictures, or fit a new card.",
    "storage.full": lambda c: "No room on the card: new pictures are lost. Fit a new card.",
    "connectivity.link-lost": lambda c: "Pictures wait on the card and the monitor site hears nothing until Wi-Fi is back. The camera keeps working.",
}


def playbooks(adopted_at):
    """The two Playbooks the owner adopted (origin owner, status adopted). They are plain rules: no model, no network."""
    return [
        {"id": "zoom-out-for-a-crowd", "enabled": True, "origin": "owner", "status": "adopted", "adoptedAt": adopted_at,
         "adoptedBy": "owner", "adoptedPhysical": False,
         "description": "When 3 or more birds have been in view for 10 s, zoom out one step, so they all fit in the picture.",
         "when": {"signal": f"{SYSTEM_ID}.birds_in_view", "feature": "min", "op": ">=", "value": 3, "windowSec": RULE_HOLD_SEC},
         "then": {"capability": "camera.zoom", "arguments": {"step": -1.0}}},
        {"id": "zoom-in-on-the-eggs", "enabled": True, "origin": "owner", "status": "adopted", "adoptedAt": adopted_at,
         "adoptedBy": "owner", "adoptedPhysical": False,
         "description": "When the camera has seen eggs with confidence 0.8 or more, and at most one bird, for 10 s, zoom in one step, towards 3.0.",
         "when": {"signal": f"{SYSTEM_ID}.egg_view", "feature": "min", "op": ">=", "value": 1, "windowSec": RULE_HOLD_SEC},
         "then": {"capability": "camera.zoom", "arguments": {"step": 1.0, "toward": 3.0}}},
    ]


class Heart:
    """Minimum Tending (notice, change status, raise, explain, escalate), plus a small Tending engine for the zoom
    Playbooks and the gate for the owner's changes. No network or model is needed for any of it."""

    def __init__(self, camera):
        self.cam = camera
        self.lock = threading.RLock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)
        self.rules = playbooks(utc())
        self.control = {"max": PROMISED_LEVEL, "current": PROMISED_LEVEL, "mode": "act", "approvalTimeoutSec": 600, "remoteSitePhysical": False,
                        "allowlist": [{"target": "camera.zoom", "initiators": ["rule"], "envelope": ZOOM_ENVELOPE}]}
        self.actions = []                     # every Action, newest last: kept locally, shown to the owner
        self.rule_since, self.rule_fired, self.rule_rearmed = {}, {}, {}
        self.auto_zoom_times = []             # simulated times of automatic zoom changes that ran
        self.t = 0.0

    # ---- noticing

    def tend(self, t):
        with self.lock:
            self.t = t
            self.sequence += 1
            for code, severity, cond, summary in RULES:
                hit = cond(self.cam)
                # A condition must hold for 5 simulated seconds before it is raised: one odd reading is not a fault.
                prev = self.since.get(code)
                self.since[code] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[code] >= 5
                open_ = next((a for a in self.alerts if a["code"] == code and a["status"] != "resolved"), None)
                if held and open_ is None:
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": summary, "explanation": EXPLAIN[code](self.cam)}
                    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.run_playbooks(t)
            for a in self.actions:
                if a["status"] == "awaiting-approval" and a["expiresAt"] < utc():
                    a.update(status="expired", updatedAt=utc())
                    log(f"HEART action     {a.get('capability') or a.get('operation')} expired: nobody pressed the button")

    def escalate(self, a):
        """CHANGE for a product: blink the LED outside the box, notify the owner's app. Here: the console."""
        a["status"] = "escalated"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}")

    # ---- the Tending engine: the zoom Playbooks

    def signal(self, name):
        c = self.cam
        if name == "birds_in_view":
            return c.birds_seen
        if name == "egg_view":   # derived: eggs seen with confidence >= 0.8, and at most one bird in view
            return 1 if c.eggs_seen and c.egg_confidence >= 0.8 and c.birds_seen is not None and c.birds_seen <= 1 else 0
        return None

    def run_playbooks(self, t):
        for r in self.rules:
            if not (r["enabled"] and r["status"] == "adopted"):
                continue
            w = r["when"]
            v = self.signal(w["signal"].rsplit(".", 1)[1])
            hit = v is not None and v >= w["value"]
            prev = self.rule_since.get(r["id"])
            self.rule_since[r["id"]] = (t if prev is None else prev) if hit else None
            if not hit:
                self.rule_rearmed[r["id"]] = True
                continue
            if t - self.rule_since[r["id"]] < w["windowSec"]:
                continue
            # A rule fires once each time its condition comes to hold. While the condition goes on holding, it may
            # fire again only after cooldownSec, so a lasting crowd does not ask every second.
            if not self.rule_rearmed.get(r["id"], True) and t - self.rule_fired.get(r["id"], -1e12) < ZOOM_ENVELOPE["cooldownSec"]:
                continue
            args = r["then"]["arguments"]
            cur = self.cam.zoom
            if "toward" in args:
                if cur >= args["toward"]:
                    continue                  # already there: nothing to do, and no Action
                new = min(cur + args["step"], args["toward"])
            else:
                new = cur + args["step"]
            if self.control["mode"] != "act":
                continue                      # stopped: no rule acts (it would be refused with autonomy-stopped)
            self.rule_fired[r["id"]], self.rule_rearmed[r["id"]] = t, False
            self.rule_zoom(r, cur, round(new, 2), t)

    def allowed(self, target, kind, args, cur, t):
        """The allowlist check of API Standard 6.3. Returns None, or why the change is refused."""
        entry = next((e for e in self.control["allowlist"] if e["target"] == target and kind in e["initiators"]), None)
        if entry is None:
            return f"{target} is not on the owner's allowlist for initiator {kind}"
        env = entry["envelope"]
        for name, value in args.items():
            lim = env["arguments"].get(name)
            if lim is None:
                return f"the envelope does not name {name}, so it must not change automatically"
            if not lim["min"] <= value <= lim["max"]:
                return f"{name} {value} is outside the envelope {lim['min']} to {lim['max']}"
            if "maxStep" in lim and abs(value - cur) > lim["maxStep"] + 1e-9:
                return f"a step of {abs(value - cur):.2f} is more than maxStep {lim['maxStep']}"
        recent = [x for x in self.auto_zoom_times if t - x < DAY]
        if len(recent) >= env["maxPerDay"]:
            return f"{len(recent)} automatic changes in the last 24 hours already: maxPerDay is {env['maxPerDay']}"
        if recent and t - recent[-1] < env["cooldownSec"]:
            return f"only {t - recent[-1]:.0f} s since the last automatic change: cooldownSec is {env['cooldownSec']}"
        return None

    def new_action(self, initiator, args, physical, capability=None, operation=None, risk="write"):
        aid = str(uuid.uuid4())
        op = capability or operation
        a = {"id": aid, "status": "running", "risk": risk, "createdAt": utc(), "updatedAt": utc(),
             "initiator": initiator, "arguments": args, "physical": physical,
             "digest": hashlib.sha256(canonical({"system": SYSTEM_ID, "id": aid, "operation": op, "arguments": args,
                                                 "physical": physical}).encode()).hexdigest()}
        a["capability" if capability else "operation"] = op
        self.actions = (self.actions + [a])[-500:]
        return a

    def rule_zoom(self, r, cur, new, t):
        a = self.new_action({"kind": "rule", "id": r["id"]}, {"zoom": new}, False, capability="camera.zoom")
        why = self.allowed("camera.zoom", "rule", {"zoom": new}, cur, t)
        if why:
            # Refused, not clipped: the zoom stays where it is, and the record says why.
            a.update(status="denied", updatedAt=utc(), error=f"Refused by the gate: {why}.",
                     explanation=r["description"], result={"zoomBefore": cur, "proposed": new, "simSec": round(t)})
            log(f"HEART playbook   {r['id']}: zoom {cur:.1f} -> {new:.1f} REFUSED: {why}")
            return
        self.cam.set_zoom(new)
        self.auto_zoom_times.append(t)
        a.update(status="succeeded", updatedAt=utc(), explanation=r["description"],
                 result={"zoomBefore": cur, "zoomAfter": new, "simSec": round(t)})
        # Tell a person afterwards (Part 2, A4 rule 3): what it noticed, what it did.
        log(f"HEART playbook   {r['id']}: zoom {cur:.1f} -> {new:.1f}. {r['description']}")

    # ---- the owner's changes

    def owner_zoom(self, z):
        with self.lock:
            cur = self.cam.zoom
            a = self.new_action({"kind": "human", "id": "owner"}, {"zoom": z}, False, capability="camera.zoom")
            self.cam.set_zoom(z)
            a.update(status="succeeded", updatedAt=utc(), result={"zoomBefore": cur, "zoomAfter": z, "simSec": round(self.t)})
            log(f"HEART owner      zoom {cur:.1f} -> {z:.1f}")
            return a

    def wait_for_person(self, a, what):
        for b in self.actions:
            if b is not a and b["status"] == "awaiting-approval":
                b.update(status="cancelled", updatedAt=utc(), error="A newer request replaced it.")
        a.update(status="awaiting-approval", expiresAt=utc(time.time() + self.control["approvalTimeoutSec"]),
                 explanation=f"Waiting for a person at the box: check the code {a['digest'][:6].upper()} on the camera's "
                             "light, then press the button (in this simulator: touch button.press).")
        log(f"HEART waiting    {what}: press the button at the box to approve. Code {a['digest'][:6].upper()}  "
            f"(touch button.press)")

    def owner_ir(self, level):
        with self.lock:
            a = self.new_action({"kind": "human", "id": "owner"}, {"level": level}, True, capability="camera.ir_level")
            self.wait_for_person(a, f"IR light {self.cam.ir_level} % -> {level} % (physical)")
            return a

    def set_max(self, m):
        """The ceiling a person set. The current level is the lower of it and the promised level; the mode follows."""
        cur = min(m, PROMISED_LEVEL)
        self.control.update(max=m, current=cur, mode="observe" if cur <= 2 else "suggest" if cur == 3 else "act")

    def owner_autonomy(self, m):
        with self.lock:
            if m <= 2:
                self.stop("the owner lowered autonomy")
                self.set_max(min(m, self.control["max"]))
                return None
            if m <= self.control["max"]:
                self.set_max(m)
                return None
            a = self.new_action({"kind": "human", "id": "owner"}, {"max": m}, False,
                                operation="PUT /heart/v1/tending/autonomy", risk="destructive")
            self.wait_for_person(a, "resume the Playbooks (raising autonomy)")
            return a

    def stop(self, why):
        with self.lock:
            self.set_max(min(self.control["max"], 2))
            self.control.update(stoppedAt=utc(), reason=why)
            log(f"HEART STOP       {why}: no Playbook acts until a person resumes them at the box")

    def button_press(self):
        """A short press on the box's button: the proof that a person is there (approves the newest waiting Action)."""
        with self.lock:
            a = next((x for x in reversed(self.actions) if x["status"] == "awaiting-approval"), None)
            if a is None:
                log("button pressed: nothing is waiting")
                return
            a.update(status="approved", updatedAt=utc(), decision={"decision": "approve", "at": utc(), "by": "person at the box",
                                                                   "presenceMethod": "device-button"})
            if a.get("capability") == "camera.ir_level":
                level, cur = a["arguments"]["level"], self.cam.ir_level
                why = self.cam.ir_guard(level)        # guards still apply after approval
                if why:
                    a.update(status="failed", updatedAt=utc(), error=f"guard-failed: {why}")
                    log(f"HEART action     IR light not changed: {why}")
                    return
                self.cam.set_ir(level)
                a.update(status="succeeded", updatedAt=utc(), result={"levelBefore": cur, "levelAfter": level})
                log(f"HEART action     IR light {cur} % -> {level} %, approved by a person at the box")
            else:
                self.set_max(a["arguments"]["max"])
                self.control.pop("stoppedAt", None)
                self.control.pop("reason", None)
                a.update(status="succeeded", updatedAt=utc())
                log("HEART action     Playbooks resumed, approved by a person at the box")

    def decide(self, aid, body):
        with self.lock:
            a = next((x for x in self.actions if x["id"] == aid), None)
            if a is None:
                return 404, "not-found", "No such Action"
            if a["status"] != "awaiting-approval":
                return 409, "conflict", f"The Action is {a['status']}, not awaiting approval"
            d = body.get("decision")
            if d in ("deny", "cancel"):
                a.update(status="denied" if d == "deny" else "cancelled", updatedAt=utc(),
                         decision={"decision": d, "at": utc(), "by": "owner"})
                log(f"HEART action     {a.get('capability') or a.get('operation')}: {d} by the owner")
                return 200, None, a
            if d != "approve":
                return 400, "invalid-request", "decision must be approve, deny or cancel"
            if "presence" not in body:
                return 403, "presence-required", ("A person must be present: press the button at the box. "
                                                  "The owner key alone cannot approve a waiting change.")
            return 403, "presence-invalid", "This camera accepts only its own button as proof of presence."

    # ---- views

    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": "watching" if self.cam.camera_ok else "camera-failed"}
                d["autonomy"] = {"max": self.control["max"], "current": self.control["current"]}
                if self.control["current"] < PROMISED_LEVEL:
                    d["autonomy"]["limitedBy"] = "stopped"
        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):
        c, g = self.cam, self.cam.g
        items = [  # name, value, unit, privacy, writable
            ("battery_pct", round(c.battery, 1), "%", "public", False),
            ("box_temp", round(c.box_temp, 1), "Cel", "internal", False),
            ("sd_free_pct", round(c.sd_free, 1), "%", "internal", False),
            ("wifi_rssi", None if c.wifi_down else round(c.rssi), "dB[mW]", "internal", False),
            ("pictures_today", c.pictures_today, "1", "internal", False),
            ("ir_level", c.ir_level, "%", "internal", True),
            ("zoom", c.zoom, "1", "internal", True),
            ("solar_charging", g.sun(c.t) > 0.05, None, "internal", False),
            ("nest_stage", g.stage, None, "internal", False),
            ("eggs_seen", c.eggs_seen, "1", "internal", False),
            ("egg_confidence", c.egg_confidence, "1", "internal", False),
            ("birds_in_view", c.birds_seen, "1", "private", False),
            ("seconds_since_last_picture", None if c.last_picture is None else round(c.t - c.last_picture), "s", "private", False),
        ]
        out = []
        for name, value, unit, privacy, writable in items:
            if privacy != "public" and not owner:
                continue
            d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": value, "writable": writable, "time": utc()}
            if unit:
                d["unit"] = unit
            out.append(self.envelope("heart.property", d, privacy))
        return out


CAPABILITIES = [
    {"name": "camera.zoom", "title": "Zoom", "risk": "write", "privacy": "internal", "physical": False,
     "callers": ["human", "rule"], "reversible": True, "idempotent": True,
     "description": "Set the camera's zoom, 1.0 to 4.0. It changes the picture, nothing in the box. Rules may run it "
                    "inside the owner's allowlist envelope.",
     "inputSchema": {"type": "object", "properties": {"zoom": {"type": "number", "minimum": 1, "maximum": 4}},
                     "required": ["zoom"], "additionalProperties": False}},
    {"name": "camera.ir_level", "title": "Infrared light", "risk": "write", "privacy": "internal", "physical": True,
     "callers": ["human"], "reversible": True,
     "description": "Set how bright the IR light shines at night, 0 to 60 %. Physical: light on living birds. Waits for "
                    "a person to press the button at the box. Never raised while the box is 35 Cel or hotter.",
     "inputSchema": {"type": "object", "properties": {"level": {"type": "integer", "minimum": 0, "maximum": IR_MAX}},
                     "required": ["level"], "additionalProperties": False}},
]


# ------------------------------------------------------------------------------------------------ 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 birdhouse camera
    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-birdhouse-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, cam, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated Wi-Fi link is lost: 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 cam.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", extra=None):
            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")
            for k, v in (extra or {}).items():
                self.send_header(k, v)
            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 body(self):
            try:
                n = int(self.headers.get("Content-Length") or 0)
                b = json.loads(self.rfile.read(n) or b"{}") if n else {}
                return b if isinstance(b, dict) else None
            except ValueError:
                return None

        def action(self, a, status=200):
            with heart.lock:
                msg = heart.envelope("heart.action", json.loads(json.dumps(a)), "internal")
            extra = {"Location": f"/heart/v1/actions/{a['id']}"} if status == 202 else None
            self.send(status, msg, extra=extra)

        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]
                d = {"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": "private", "action": "internal",
                                "rule": "internal", "control": "internal"},
                     "autonomy": {"level": 1}, "template": TEMPLATE}
                if owner:
                    d["capabilities"] = CAPABILITIES     # internal: shown only to a caller with that clearance
                return self.send(200, heart.envelope("heart.card", d))
            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 not owner and (path.split("/") + ["", "", "", ""])[3] in ("tending", "actions", "visits") and path.startswith("/heart/v1/"):
                return self.problem(403, "insufficient-clearance", "Playbooks, Actions and visits are for the owner: send the owner key")
            with heart.lock:
                if path == "/heart/v1/tending/rules":
                    return self.send(200, {"items": [heart.envelope("heart.rule", dict(r), "internal") for r in heart.rules]},
                                     "application/json")
                if path in ("/heart/v1/tending/autonomy", "/heart/v1/tending/allowlist"):
                    return self.send(200, heart.envelope("heart.control", json.loads(json.dumps(heart.control)), "internal"))
                if path == "/heart/v1/actions":
                    return self.send(200, {"items": [heart.envelope("heart.action", json.loads(json.dumps(a)), "internal")
                                                     for a in reversed(heart.actions)]}, "application/json")
                if path.startswith("/heart/v1/actions/"):
                    a = next((x for x in heart.actions if x["id"] == path.rsplit("/", 1)[1]), None)
                    return self.action(a) if a else self.problem(404, "not-found", "No such Action")
                if path == "/heart/v1/visits":
                    return self.send(200, {"items": list(reversed(heart.cam.visits_log))}, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties, /tending/rules, /actions or /visits")

        def do_POST(self):
            path = self.path.split("?")[0].rstrip("/")
            if not self.owner():
                return self.problem(401, "unauthenticated", "Changes need the owner key")
            body = self.body()
            if body is None:
                return self.problem(400, "invalid-request", "The body must be a JSON object")
            if path == "/heart/v1/capabilities/camera.zoom/invoke":
                z = (body.get("arguments") or {}).get("zoom")
                if not isinstance(z, (int, float)) or isinstance(z, bool):
                    return self.problem(400, "invalid-request", "arguments.zoom must be a number")
                if not ZOOM_RANGE[0] <= z <= ZOOM_RANGE[1]:
                    return self.problem(422, "out-of-range", f"zoom must be {ZOOM_RANGE[0]} to {ZOOM_RANGE[1]}")
                return self.action(heart.owner_zoom(float(z)))
            if path == "/heart/v1/capabilities/camera.ir_level/invoke":
                lv = (body.get("arguments") or {}).get("level")
                if not isinstance(lv, int) or isinstance(lv, bool):
                    return self.problem(400, "invalid-request", "arguments.level must be a whole number")
                if not 0 <= lv <= IR_MAX:
                    return self.problem(422, "out-of-range", f"level must be 0 to {IR_MAX} %, the most the firmware allows")
                with heart.lock:
                    why = heart.cam.ir_guard(lv)
                if why:
                    return self.problem(409, "guard-failed", why)
                return self.action(heart.owner_ir(lv), 202)
            if path.startswith("/heart/v1/actions/") and path.endswith("/decision"):
                status, code, out = heart.decide(path.split("/")[4], body)
                return self.action(out) if status == 200 else self.problem(status, code, out)
            if path == "/heart/v1/tending/stop":
                heart.stop("the owner pressed Stop")
                with heart.lock:
                    return self.send(200, heart.envelope("heart.control", json.loads(json.dumps(heart.control)), "internal"))
            self.refuse()

        def do_PUT(self):
            path = self.path.split("?")[0].rstrip("/")
            if path == "/heart/v1/tending/autonomy":
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Changes need the owner key")
                body = self.body() or {}
                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 self.problem(400, "invalid-request", f"This camera offers a max of 0 to 2 (Stop) or {PROMISED_LEVEL}: "
                                        f'send {{"max": 2}} or {{"max": {PROMISED_LEVEL}}}')
                if "presence" in body and m > heart.control["max"]:
                    return self.problem(403, "presence-invalid", "This camera accepts only its own button as proof of presence.")
                a = heart.owner_autonomy(m)
                with heart.lock:
                    return self.action(a, 202) if a else self.send(200, heart.envelope("heart.control", json.loads(json.dumps(heart.control)), "internal"))
            self.refuse()

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "Only the zoom and the IR light can be changed (POST /heart/v1/capabilities/{name}/invoke). "
                         "Playbooks and the allowlist are adopted at the box, not over the network.")

        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 birdhouse camera 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="birdhouse-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43280, 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("--clock", default="", help="start the simulated clock at this local time, HH:MM (default: now)")
    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))}")
    clock0 = time.time()
    if args.clock:
        try:
            hh, mm = (int(x) for x in args.clock.split(":"))
            lt = time.localtime()
            clock0 = time.mktime((lt.tm_year, lt.tm_mon, lt.tm_mday, hh, mm, 0, 0, 0, -1))
        except ValueError:
            ap.error("--clock takes HH:MM")

    rng = random.Random(args.seed)
    garden = Garden(rng, scenarios, clock0)
    cam = Camera(rng, scenarios, garden)
    heart = Heart(cam)
    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}
  DEMONSTRATION ONLY. Simulated garden, birds and camera; no picture is ever made. HEART watches, and two adopted
  Playbooks may change the zoom inside the owner's envelope. The IR light changes only with a person at the box.
  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, cam, args.monitor_url, args.state_file, stop), daemon=True).start()

    t, wall0, last_print = 0.0, time.time(), 0.0
    sub = max(1.0, args.speed / 50)           # simulated seconds per step: fine enough for 10 s holds
    try:
        while not stop.is_set():
            left = args.speed
            while left > 1e-9:
                dt = min(sub, left)
                left -= dt
                t += dt
                with heart.lock:
                    garden.step(t, dt)
                    cam.step(t, dt)
                heart.tend(t)
            for name, act in (("button.press", heart.button_press), ("button.hold", lambda: heart.stop("Stop: the button was held"))):
                if os.path.exists(name):
                    os.unlink(name)
                    act()
            if time.time() - last_print >= 15:
                last_print = time.time()
                g = garden.at(t)
                eggs = f"{cam.eggs_seen} seen @{cam.egg_confidence:.2f}" if cam.eggs_seen else "none seen"
                log(f"{time.strftime('%H:%M', time.localtime(clock0 + t))} {g.part_of_day(t)} · birds {cam.birds_seen if cam.birds_seen is not None else '?'} · "
                    f"nest {g.stage} (eggs {eggs}) · zoom {cam.zoom:.1f} · IR {cam.ir_level} % {'on' if cam.ir_on else 'off'} · "
                    f"battery {cam.battery:4.1f} %{' charging' if g.sun(t) > 0.05 else ''} · box {cam.box_temp:4.1f} Cel · "
                    f"card {cam.sd_free:4.1f} % free · Wi-Fi {'down' if cam.wifi_down else f'{cam.rssi:.0f} dBm'} · "
                    f"{cam.pictures_today} pictures today · 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 camera stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo