Dog tracking collar: what is inside

A GPS collar on a dog. Where the dog is never leaves for the monitor site: only the collar's own health does. HEART runs split: faults are noticed on the collar's microcontroller; the API and monitor reports run on the home base or the maker's server.

Specifications it follows

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

Dog  ──►  Collar  ──(reads)──►  Heart  ──►  local HEART API      (127.0.0.1)
day,       GNSS, uplink,          watches     └►  HEART Hub    (outbound only: status and fault codes)
walks,     accelerometer,         owner ──(key)──► collar.lost_mode           (not physical: at once)
wanders    temperature, battery,  owner ──(passkey; here button.press)──► collar.led, collar.buzzer  (physical)
           safe zone ──► private events for the owner's app (never HEART Hub)

Run it yourself, and register it on HEART Hub

1. Download README.md, dogCollarSimulator.py, test_e2e.py.
2. Run python3 dogCollarSimulator.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: Dog tracking collar.

The code: dogCollarSimulator.py

1328 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/dog-collar-simulator/v1.2611.2
"""dogCollarSimulator.py: a simulated dog tracking collar with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. There is no collar and no dog: everything is simulated. Every position is in metres east and
north of a made-up home origin, never a real place. Its numbers are plausible, not measured. See README.md beside
this file, and the case study guides/Dog-Tracking-Collar.md.

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

  * a simulated DOG'S DAY: it sleeps at night, rests and plays in the garden, goes for walks with its owner, and
    sometimes slips out of the gate and wanders off alone. Its position is a random walk in and around the home
    "safe zone" (a circle of 60 m round the home origin);
  * a simulated COLLAR (the product): GNSS (satellite positioning) with a fix quality, an LTE-M uplink (or the home
    base by Bluetooth when it is near the house), an accelerometer that tells resting, walking and running apart, a
    temperature sensor, a battery that drains faster with GNSS, the uplink and lost mode, an LED and a buzzer. The
    collar's own firmware checks the safe zone and tells the owner's app, privately, when the dog leaves it;
  * a HEART that WATCHES the collar's health: 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. Leaving the safe zone is a fact about the dog, not a fault of the collar: it is a PRIVATE
    event for the owner, never an alert, and it never reaches HEART Monitor;
  * the owner's changes, through the local API with the owner key. LOST MODE (GNSS fix and a position report every
    30 s instead of every 10 minutes) is not physical: it changes reporting, nothing the dog feels, so it runs at
    once, and its Action shows the battery it costs. The LED and the BUZZER act on a living animal, so they are
    PHYSICAL: a change waits for a person's proof of presence. In a real collar that is the owner confirming on
    their phone with a passkey; in this simulator it is a file, touch button.press in the working directory. An
    approval over the network with only the key is refused with 403 presence-required, and the collar's own guard
    refuses a buzzer louder than 50 %, longer than 5 s, or sooner than 60 s after the last one (409 guard-failed);
  * SLEEP until a stated time (POST /heart/v1/lifecycle/sleep, durationSec or wakeAt) and restart. Asleep, the
    collar closes its HTTP port and does no sensor work; its timer or its button (touch button.press) wakes it. It
    refuses to sleep (409 conflict) while lost mode is on, the dog is outside the safe zone, a critical alert is
    open, or it has warned of heat: asleep it could not say where the dog is.

Run it:

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

Scenarios (--scenario, comma-separated): lowbattery, wander, link, off, hotcar, stuck, nofix, all. Each starts at a
set time, lasts a while, and is then put right by the owner, so you can watch an alert open, escalate and resolve.
wander raises no alert at all: the dog leaving the safe zone is a private event for the owner only.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. The
position, the track, the safe-zone events, the activity, lost mode and every reading stay local (internal or
private).
"""
import argparse
import calendar
import hashlib
import json
import math
import os
import random
import re
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.dog-collar"
NAME = "Dog Tracking Collar Simulator (demo)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/dog-collar-simulator/v1.2611.2"
PULSE_SEC = 30
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
DAY = 86400.0

# The simulated place. Metres east and north of a made-up home origin: there is no latitude or longitude anywhere.
SAFE_ZONE_M = 60.0                        # the owner's safe zone: a circle round home
HOME_BASE_M = 30.0                        # the home base's Bluetooth reaches about this far
GARDEN_M = 25.0                           # the garden the dog plays in
CAR_PARK = (1380.0, -910.0)               # where the owner parks for the shops (made up)

# The collar's own settings and limits. They live in its firmware; HEART reads them and cannot widen them.
BATTERY_MAH = 900.0
REPORT_NORMAL, REPORT_LOST = 600, 30      # position report to the owner's app, s
GNSS_AWAY = 60                            # away from home: a GNSS fix every 60 s for the safe-zone check
LED_MODES = ("off", "steady", "blink")
MAX_VOLUME, MAX_BUZZ_SEC = 100, 30        # what the buzzer hardware can do
SAFE_VOLUME, SAFE_BUZZ_SEC, BUZZ_REST_SEC = 50, 5, 60   # the welfare guard: never louder, longer or more often

# Healthy ranges, as in the case study's fault table.
BATTERY_LOW, BATTERY_EMPTY, TOO_HOT = 15.0, 3.0, 45.0
NOFIX_TRIES, NOFIX_SEC = 3, 150           # failed fixes in a row, and seconds without a position, while away
STUCK_SEC, STUCK_MOVED_M = 60, 40.0       # accelerometer frozen this long while GNSS says it moved this far
OFF_STILL_SEC, OFF_COOL = 180, 26.0       # completely still (no breathing) this long, and cooler than a worn collar

SCENARIOS = ("lowbattery", "wander", "link", "off", "hotcar", "stuck", "nofix")
# When each staged event starts and how long it lasts, in simulated seconds. stuck and nofix share one walk.
STAGED = {"lowbattery": [(0, 600)], "wander": [(30, 180)], "link": [(60, 240)], "off": [(280, 250)],
          "hotcar": [(540, 240)], "stuck": [(800, 200)], "nofix": [(1000, 240)]}
FORCED_WALK = (800, 460)                  # the walk in the woods that stuck and nofix happen on

# Lifecycle (API Standard 13.2 rule 3, D29): the collar can sleep until a stated time, and restart. wake is not
# offered: in deep sleep the collar cannot answer HTTP, so only its timer or its button can wake it (rule 3.3).
STANDARD_OPS = ("restart", "shutdown", "sleep", "wake", "drain", "reload", "power-cycle", "safe-mode", "maintenance-on",
                "maintenance-off", "reset-config", "reset-network", "clear-data", "factory-reset", "decommission")
LIFECYCLE_OPS = ("restart", "sleep")
WAKE_SOURCES = ["timer", "button"]
LIFE_FIELDS = {"operation", "mode", "target", "timeoutSec", "drainSec", "delaySec", "depth", "wakeAt", "durationSec", "reason"}
MAX_SLEEP_SEC = 604800                    # at most 7 days of simulated time
SLEEP_MA = 0.012                          # deep sleep: only the real-time clock and the button's wake-up line
GO_DOWN_SEC, WAKE_UP_SEC, RESTART_SEC = 0.5, 3, 3   # real seconds: let the reply out; start again; a restart


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)


def dist(p, q=(0.0, 0.0)):
    return math.hypot(p[0] - q[0], p[1] - q[1])


# ------------------------------------------------------------------------------------------------ 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 < s + n for s, n in STAGED[name])


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


class Dog:
    """The simulated world: the hour, the weather, the dog and its owner. Nothing here is a product value."""

    MOVE = {"sleep": 0.012, "rest": 0.03, "car": 0.025, "walk": 0.35, "wander": 0.45, "return": 0.4}

    def __init__(self, rng, scenarios, clock0):
        self.rng, self.sc, self.clock0 = rng, scenarios, clock0
        self.pos, self.heading = (3.0, -2.0), rng.uniform(0, 2 * math.pi)
        self.mode, self.move = "rest", 0.03
        self.walk_ep = Episode(rng, (3 * 3600, 7 * 3600), (1200, 2700))
        self.play_ep = Episode(rng, (1800, 5400), (300, 900))
        self.wander_ep = Episode(rng, (3 * DAY, 10 * DAY), (300, 1500))      # the gate left open
        self.woods_ep = Episode(rng, (2 * 3600, 8 * 3600), (120, 600))       # thick trees on a walk: a poor sky
        self.walk = None                     # (start, length, radius, angle) of the walk now going on
        self.car_on = False

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

    def outside_temp(self, t):
        return 12 + 7 * math.sin(math.pi * (self.hour(t) - 9) / 12)

    def ambient(self, t, pos):
        """The air round the collar: indoors near the house, the car when shut in it, else outside."""
        if self.mode == "car":
            return 62.0                       # a parked car in the sun, windows shut
        if dist(pos) < 8:
            return 21.0
        return self.outside_temp(t)

    @property
    def owner_nearby(self):
        return self.mode in ("walk", "return")

    @property
    def poor_sky(self):
        return self.mode in ("walk", "wander") and self.woods_ep.active(self._t)

    def choose(self, t):
        h = self.hour(t)
        if staged(self.sc, "hotcar", t):
            return "car"
        quiet = bool(self.sc & {"wander", "off", "hotcar", "stuck", "nofix"})   # staged days: no random outings
        if staged(self.sc, "wander", t) or (not quiet and self.wander_ep.active(t) and 7 <= h < 21):
            return "wander"
        if self.mode in ("wander", "return") and dist(self.pos) > 8:
            return "return"                   # the owner goes and fetches the dog
        fw = FORCED_WALK
        if ("stuck" in self.sc or "nofix" in self.sc) and fw[0] <= t < fw[0] + fw[1]:
            return "walk"
        if self.walk and t < self.walk[0] + self.walk[1]:
            return "walk"
        if h >= 22 or h < 6.5:
            return "sleep"
        if self.walk_ep.active(t) and 7 <= h < 21 and not quiet:
            return "walk"
        if self.play_ep.active(t):
            return "play"
        return "rest"

    def step(self, t, dt):
        self._t = t
        mode = self.choose(t)
        if mode != self.mode:
            if mode == "walk":
                fw = FORCED_WALK
                start, length = (fw[0], fw[1]) if fw[0] <= t < fw[0] + fw[1] and (
                    "stuck" in self.sc or "nofix" in self.sc) else (t, self.rng.uniform(1200, 2700))
                speed = self.rng.uniform(1.1, 1.4)
                self.walk = (start, length, speed * length / (2 * math.pi), self.rng.uniform(0, 2 * math.pi))
                if "stuck" in self.sc and start == fw[0]:
                    log("world: (staged) the owner takes the dog for a walk in the woods")
            if mode == "car":
                self.pos = CAR_PARK
                log("world: (staged) the owner drives to the shops, and leaves the dog in the car in the sun")
            if self.mode == "car":
                self.pos = (3.0, -2.0)
                log("world: (staged) the owner is back, takes the dog out of the car, and drives home")
            if mode == "wander":
                log("world: (staged) the gate is open: the dog slips out and wanders off alone")
            if mode == "return" and self.mode == "wander":
                log("world: the owner goes after the dog and brings it home")
            if self.mode == "walk":
                self.walk = None
            self.mode = mode
        x, y = self.pos
        r = self.rng
        if mode in ("sleep", "rest", "car"):
            if mode != "car":
                x, y = x + (3.0 - x) * min(1.0, dt / 30) + r.gauss(0, 0.05), y + (-2.0 - y) * min(1.0, dt / 30) + r.gauss(0, 0.05)
            self.move = self.MOVE[mode]
        elif mode == "play":
            self.heading += r.gauss(0, 0.8)
            v = r.choice((0.5, 1.5, 3.0))
            x, y = x + v * dt * math.cos(self.heading), y + v * dt * math.sin(self.heading)
            if math.hypot(x, y) > GARDEN_M:
                self.heading += math.pi
                x, y = x * GARDEN_M / math.hypot(x, y), y * GARDEN_M / math.hypot(x, y)
            self.move = 0.25 + 0.3 * v
        elif mode == "walk":
            start, length, radius, a0 = self.walk
            ang = 2 * math.pi * min(1.0, (t - start) / length)
            cx, cy = radius * math.cos(a0), radius * math.sin(a0)
            x = cx + radius * math.cos(a0 + math.pi + ang) + r.gauss(0, 1.5)
            y = cy + radius * math.sin(a0 + math.pi + ang) + r.gauss(0, 1.5)
            self.move = self.MOVE["walk"] * (1.6 if r.random() < 0.15 else 1.0)
        elif mode == "wander":
            out = math.atan2(y, x) if math.hypot(x, y) > 1 else self.heading
            self.heading += r.gauss(0, 0.3) + 0.2 * math.sin(out - self.heading)
            v = r.uniform(0.8, 1.6)
            x, y = x + v * dt * math.cos(self.heading), y + v * dt * math.sin(self.heading)
            self.move = self.MOVE["wander"]
        elif mode == "return":
            d = math.hypot(x, y)
            step = min(d, 1.6 * dt)
            x, y = x - step * x / d, y - step * y / d
            self.move = self.MOVE["return"]
        self.pos = (x, y)


class Collar:
    """The simulated product. Its firmware owns GNSS, the uplink, the safe zone, the owner's app, the LED and the
    buzzer with their limits. HEART reads these values, and may change only lost mode, the LED and the buzzer,
    through capabilities that pass the gate."""

    def __init__(self, rng, scenarios, dog):
        self.rng, self.sc, self.dog = rng, scenarios, dog
        self.mah = BATTERY_MAH * ((15.6 if "lowbattery" in scenarios else rng.uniform(55, 90)) / 100)
        self.temp = 29.0
        self.worn, self.off_at = True, (14.0, 9.0)
        self.pos, self.pos_t, self.pos_source = dog.pos, 0.0, "home-base"
        self.fixes = []                        # private track: (t, east, north), newest last
        self.gnss_state, self.sats, self.hdop, self.fails, self.next_fix = "off", 0, None, 0, 0.0
        self.link_down, self.rssi, self.link_ep, self.t_rssi = False, -92.0, Episode(rng, (DAY, 4 * DAY), (120, 900)), Timer(rng, 5, 20)
        self.lost_mode, self.next_report, self.reports, self.queued = False, 0.0, 0, 0
        self.led, self.buzz_until, self.last_buzz = "off", -1.0, -1e9
        self.accel, self.accel_same_since, self.still_since = (0.0, 0.0, 1.0), None, None
        self.inside = True
        self.events, self.seq = [], 0         # private safe-zone and heat events (the firmware's own), newest last
        self.heat_warned = False
        self.t = 0.0

    # ---- what the collar measures

    @property
    def battery(self):
        return max(0.0, 100 * self.mah / BATTERY_MAH)

    @property
    def home_base(self):
        return dist(self.dog.pos if self.worn else self.off_at) <= HOME_BASE_M

    def current_ma(self, lost=None, away=None):
        """Average current in a mode: what the battery estimate uses. GNSS and the uplink dominate."""
        lost = self.lost_mode if lost is None else lost
        away = (not self.home_base) if away is None else away
        ma = 0.35                                              # microcontroller, accelerometer, Bluetooth
        if lost:
            ma += 30.0 + 12.0                                  # GNSS kept tracking, LTE-M kept connected
        else:
            ma += 2.5 if away else 0.0                         # a warm-start fix every 60 s while away
            ma += 0.7 if away else 0.15                        # a report every 10 min: LTE-M away, Bluetooth at home
        ma += {"off": 0.0, "steady": 5.0, "blink": 2.0}[self.led]
        return ma

    def hours_left(self, lost=None, away=None):
        return round(self.mah / self.current_ma(lost, away), 1)

    @property
    def accel_dynamic(self):
        x, y, z = self.accel
        return math.sqrt(x * x + y * y + (z - 1) * (z - 1))

    @property
    def activity(self):
        a = self.accel_dynamic
        return "resting" if a < 0.1 else "walking" if a < 0.6 else "running"

    def record(self, name, severity, summary, **attrs):
        self.seq += 1
        self.events = (self.events + [{"sequence": self.seq, "name": name, "category": "domain", "severity": severity,
                                       "summary": summary, "attributes": {"at": utc(), **attrs}}])[-200:]
        to = "waiting for the uplink, then to the owner's app" if self.link_down else "to the owner's app"
        log(f"PRIVATE  {name}: {summary} ({to} only; never to HEART Monitor)")

    def new_position(self, p, source, t):
        self.pos, self.pos_t, self.pos_source = p, t, source
        inside = dist(p) <= SAFE_ZONE_M
        if inside != self.inside:
            self.inside = inside
            d = round(dist(p))
            alone = not self.dog.owner_nearby and self.dog.mode != "car"
            if not inside:
                self.record("zone.left", "warning" if alone else "info",
                            f"The dog has left the safe zone{' alone' if alone else ' with its owner'}, {d} m from home."
                            + (" Turn on lost mode to follow it every 30 s." if alone and not self.lost_mode else ""),
                            withOwner=not alone, distanceM=d, eastM=round(p[0], 1), northM=round(p[1], 1))
            else:
                self.record("zone.returned", "info", "The dog is back inside the safe zone.", distanceM=d)

    def step(self, t, dt):
        self.t, dog = t, self.dog
        # Staged faults, for the demo. Each lasts a while and is then put right by the owner.
        was_worn = self.worn
        self.worn = not staged(self.sc, "off", t)
        if was_worn and not self.worn:
            self.off_at = (14.0, 9.0)                # by the hedge, in the garden
            log("world: (staged) the collar's buckle opens in the garden: the collar lies on the grass")
        if not was_worn and self.worn:
            log("world: (staged) the owner finds the collar and puts it back on the dog")
        where = dog.pos if self.worn else self.off_at
        self.link_down = staged(self.sc, "link", t) or (not self.sc and self.link_ep.active(t))
        stuck = staged(self.sc, "stuck", t)
        if "stuck" in self.sc and staged_ended(self.sc, "stuck", t) and not getattr(self, "_restarted", False):
            self._restarted = True
            log("world: (staged) the owner restarts the collar: the accelerometer reads again")

        # The accelerometer: gravity plus the dog's movement (breathing, even asleep). Lying off the dog: only
        # sensor noise. Stuck: the very same reading, again and again.
        if not stuck:
            m = dog.move * max(0.3, self.rng.gauss(1, 0.2)) if self.worn else 0.0012
            a = (self.rng.gauss(0, m), self.rng.gauss(0, m), 1 + self.rng.gauss(0, m))
            self.accel_same_since = None
        else:
            a = (0.0213, -0.3374, 1.1162)            # frozen mid-stride
            self.accel_same_since = t if self.accel_same_since is None else self.accel_same_since
        self.accel = a
        if self.accel_dynamic < 0.005:
            self.still_since = t if self.still_since is None else self.still_since
        else:
            self.still_since = None

        # Temperature: the air, the dog's warmth when worn, the battery's own heat in lost mode.
        target = dog.ambient(t, where) if not self.worn else 0.45 * dog.ambient(t, where) + 0.55 * 37.5
        if dog.mode == "car" and self.worn:
            target = 0.8 * 62.0 + 0.2 * 39.0
        target += 1.5 if self.lost_mode else 0.0
        self.temp += (target - self.temp) * min(1.0, dt / (60.0 if dog.mode == "car" else 90.0))

        # Position: from the home base while near the house, from GNSS away from it, or every 30 s in lost mode.
        if self.home_base and not self.lost_mode:
            self.gnss_state, self.sats, self.hdop, self.fails = "off", 0, None, 0
            if t - self.pos_t >= 5:
                self.new_position(where, "home-base", t)
            self.next_fix = t                  # leaving the home base: the first fix is taken at once
        elif t >= self.next_fix:
            self.next_fix = t + (REPORT_LOST if self.lost_mode else GNSS_AWAY)
            nofix = staged(self.sc, "nofix", t)
            ok = not nofix and self.rng.random() < (0.25 if dog.poor_sky else 0.97)
            if ok:
                self.sats, self.hdop = self.rng.randint(6, 12), round(self.rng.uniform(0.8, 1.9), 1)
                err = 3 * self.hdop
                p = (where[0] + self.rng.gauss(0, err), where[1] + self.rng.gauss(0, err))
                self.gnss_state, self.fails = "fix", 0
                self.fixes = (self.fixes + [(t, round(p[0], 1), round(p[1], 1), utc())])[-500:]
                self.new_position(p, "gnss", t)
            else:
                self.sats, self.hdop, self.gnss_state = self.rng.randint(0, 3), None, "searching"
                self.fails += 1

        # Reports to the owner's app: every 10 minutes, or every 30 s in lost mode. Held while the uplink is down.
        if t >= self.next_report:
            self.next_report = t + (REPORT_LOST if self.lost_mode else REPORT_NORMAL)
            if self.link_down:
                self.queued += 1
            else:
                self.reports += 1 + self.queued
                self.queued = 0
        if self.t_rssi.due(t):
            self.rssi = min(-70.0, max(-115.0, self.rssi + (-92 - self.rssi) * 0.3 + self.rng.gauss(0, 2)))

        # The collar's own private warning for the dog: hot, and not with its owner. Not a HEART alert.
        hot = self.worn and self.temp > 40 and not dog.owner_nearby and not self.home_base
        if hot and not self.heat_warned:
            self.record("heat.risk", "critical", f"The collar is {self.temp:.0f} Cel and the dog is not with you. "
                        "Is it shut in a car? Check on the dog now.", collarTempCel=round(self.temp, 1))
        self.heat_warned = hot if hot else (self.heat_warned and self.temp > 35)

        # Battery: the average current of what is on, plus an old cell losing charge by itself (lowbattery).
        drain = self.current_ma() * dt / 3600
        if self.buzz_until > t:
            drain += 30.0 * min(dt, self.buzz_until - t) / 3600
        if staged(self.sc, "lowbattery", t):
            drain += 0.00006 * BATTERY_MAH * dt
        self.mah = max(0.0, self.mah - drain)
        if "lowbattery" in self.sc and staged_ended(self.sc, "lowbattery", t) and not getattr(self, "_charged", False):
            self._charged = True
            self.mah = 0.96 * BATTERY_MAH
            log("world: (staged) the owner charges the collar: battery 96 %")

    def doze(self, t, dt):
        """Deep sleep (or a restart): no GNSS, no uplink, no accelerometer, no safe-zone check, no reports, no heat
        warning. Only the clock runs, and the battery drains at the sleep current (and by itself, if it is old)."""
        self.t = t
        self.gnss_state, self.sats, self.hdop = "off", 0, None
        drain = SLEEP_MA * dt / 3600
        if staged(self.sc, "lowbattery", t):
            drain += 0.00006 * BATTERY_MAH * dt
        self.mah = max(0.0, self.mah - drain)

    # ---- the checks the HEART reads

    def no_fix(self):
        return not self.home_base and self.fails >= NOFIX_TRIES and self.t - self.pos_t >= NOFIX_SEC

    def accel_stuck(self):
        if self.accel_same_since is None or self.t - self.accel_same_since < STUCK_SEC:
            return False
        moved = [f for f in self.fixes if f[0] >= self.accel_same_since]
        return len(moved) >= 2 and dist(moved[0][1:3], moved[-1][1:3]) >= STUCK_MOVED_M

    def maybe_off(self):
        return self.still_since is not None and self.t - self.still_since >= OFF_STILL_SEC and self.temp < OFF_COOL

    def main_job_problem(self):
        """CHANGE for a product: None while the collar does its main job (it can tell the owner where the dog is)."""
        if self.battery <= BATTERY_EMPTY:
            return "The battery is nearly empty: the collar will switch off and stop reporting where the dog is."
        if self.no_fix() and self.link_down:
            return "No GNSS fix and no uplink, away from home: the owner cannot be told where the dog is."
        return None

    # The three things HEART may change, each behind the product's own limits.
    def set_lost_mode(self, on):
        self.lost_mode = on
        self.next_report = self.next_fix = self.t     # the first report and fix at once

    def set_led(self, mode):
        self.led = mode

    def buzz_guard(self, volume, sec):
        """The welfare guard, in the product: a buzzer on a dog's neck is never loud, long or frequent."""
        if volume > SAFE_VOLUME:
            return f"A volume of {volume} % is above the safe {SAFE_VOLUME} % for a dog's ears."
        if sec > SAFE_BUZZ_SEC:
            return f"{sec} s is longer than the safe {SAFE_BUZZ_SEC} s."
        if self.t - self.last_buzz < BUZZ_REST_SEC:
            return f"The buzzer sounded {self.t - self.last_buzz:.0f} s ago: it rests at least {BUZZ_REST_SEC} s."
        return None

    def buzz(self, volume, sec):
        self.last_buzz, self.buzz_until = self.t, self.t + sec


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

RULES = [
    # code, severity, condition, summary
    ("system.main-function-lost", "critical", lambda c: c.main_job_problem() is not None,
     "The collar cannot tell the owner where the dog is"),
    ("system.main-function-degraded", "warning", lambda c: c.maybe_off(), "The collar may have slipped off the dog"),
    ("power.battery-low", "warning", lambda c: c.battery < BATTERY_LOW, "The battery is low"),
    ("performance.too-hot", "error", lambda c: c.temp > TOO_HOT, "The collar is too hot"),
    ("sensors.stuck", "warning", lambda c: c.accel_stuck(), "The accelerometer reading is stuck"),
    ("sensors.no-reading", "warning", lambda c: c.no_fix(), "GNSS has had no fix for a long time"),
    ("connectivity.link-lost", "warning", lambda c: c.link_down, "The collar's uplink is lost"),
]

EXPLAIN = {
    "system.main-function-lost": lambda c: c.main_job_problem() or "",
    "system.main-function-degraded": lambda c: (
        f"No movement at all, not even breathing, for {c.t - (c.still_since or c.t):.0f} s, and the collar is "
        f"{c.temp:.0f} Cel, cooler than on a dog. It may be lying somewhere without the dog. HEART cannot be sure: "
        "check that the collar is on the dog."),
    "power.battery-low": lambda c: f"Battery {c.battery:.1f} %, below {BATTERY_LOW:.0f} %: about {c.hours_left():.0f} hours "
                                   "left as it is now. Charge it soon.",
    "performance.too-hot": lambda c: f"The collar is {c.temp:.1f} Cel, above {TOO_HOT:.0f} Cel: too hot for its battery. "
                                     "The collar has warned the owner privately about the dog.",
    "sensors.stuck": lambda c: f"The accelerometer has given exactly the same reading for {c.t - c.accel_same_since:.0f} s, "
                               "though GNSS shows the collar moving. Activity cannot be trusted. Restart the collar; "
                               "if it stays stuck, it needs service.",
    "sensors.no-reading": lambda c: f"{c.fails} GNSS fixes failed in a row, {c.t - c.pos_t:.0f} s without a position. "
                                    "Thick trees or a building block the sky; the collar keeps trying.",
    "connectivity.link-lost": lambda c: "Neither LTE-M nor the home base answers. Reports wait on the collar and the "
                                        "monitor site hears nothing until the link is back. The collar keeps working.",
}


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

    def __init__(self, collar):
        self.c = collar
        self.lock = threading.RLock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)
        self.actions = []                     # every Action, newest last: kept locally, shown to the owner
        self.approval_timeout = 600
        self.t = 0.0
        self.speed = 1.0                      # simulated seconds per real second (set by main)
        self.phase, self.phase_since = "running", utc()
        self.life = None                      # the lifecycle Action under way: {"a", "op", "go", "wake", "went", ...}

    # ---- noticing

    def tend(self, t):
        with self.lock:
            self.t = t
            self.sequence += 1
            for code, severity, cond, summary in RULES:
                hit = cond(self.c)
                # 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.c)}
                    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"
            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['capability']} expired: nobody confirmed it")

    def escalate(self, a):
        """CHANGE for a product: a notification in the owner's app. Here: the console. Not the LED or the buzzer:
        those act on the dog, and HEART never uses them by itself."""
        a["status"] = "escalated"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}")

    # ---- the owner's changes

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

    def owner_lost_mode(self, on):
        """Not physical: it changes how often the collar reports, nothing the dog feels. Runs at once."""
        with self.lock:
            c = self.c
            before = {"lostMode": c.lost_mode, "reportIntervalSec": REPORT_LOST if c.lost_mode else REPORT_NORMAL,
                      "batteryHoursLeft": c.hours_left()}
            a = self.new_action("collar.lost_mode", {"on": on}, False)
            c.set_lost_mode(on)
            after = {"lostMode": on, "reportIntervalSec": REPORT_LOST if on else REPORT_NORMAL,
                     "batteryHoursLeft": c.hours_left(lost=on), "currentMilliamps": round(c.current_ma(lost=on), 2)}
            a.update(status="succeeded", updatedAt=utc(), result={"before": before, "after": after},
                     explanation=f"Lost mode {'on' if on else 'off'}: GNSS and reports every "
                                 f"{after['reportIntervalSec']} s. Battery about {before['batteryHoursLeft']:.0f} h -> "
                                 f"{after['batteryHoursLeft']:.0f} h.")
            log(f"HEART owner      {a['explanation']}")
            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.")
        code = a["digest"][:6].upper()
        a.update(status="awaiting-approval", expiresAt=utc(time.time() + self.approval_timeout),
                 explanation=f"Physical: it acts on the dog. Waiting for the owner to confirm code {code} with their "
                             "passkey on the phone (in this simulator: touch button.press).")
        log(f"HEART waiting    {what}: confirm code {code} with the passkey on the phone (simulator: touch button.press)")

    def owner_physical(self, capability, args, what):
        with self.lock:
            a = self.new_action(capability, args, True)
            self.wait_for_person(a, what)
            return a

    def button_press(self):
        """The simulator's stand-in for the owner's passkey: 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
            # device-button, because that is what this simulator really checks: a file, not a passkey.
            a.update(status="approved", updatedAt=utc(), decision={"decision": "approve", "at": utc(), "by": "owner",
                                                                   "presenceMethod": "device-button"})
            c = self.c
            if a["capability"] == "collar.led":
                before = c.led
                c.set_led(a["arguments"]["mode"])
                a.update(status="succeeded", updatedAt=utc(), result={"ledBefore": before, "ledAfter": c.led})
                log(f"HEART action     LED {before} -> {c.led}, confirmed by the owner")
            else:
                v, s = a["arguments"]["volume"], a["arguments"]["durationSec"]
                why = c.buzz_guard(v, s)                  # guards still apply after approval
                if why:
                    a.update(status="failed", updatedAt=utc(), error=f"guard-failed: {why}")
                    log(f"HEART action     buzzer not sounded: {why}")
                    return
                c.buzz(v, s)
                a.update(status="succeeded", updatedAt=utc(), result={"volume": v, "durationSec": s})
                log(f"HEART action     buzzer {v} % for {s} s, confirmed by the owner")

    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['capability']}: {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", ("The LED and the buzzer act on the dog: the owner must confirm with "
                                                  "their passkey. The owner key alone cannot approve a waiting change.")
            return 403, "presence-invalid", "This simulator accepts only its own button file as proof of presence."

    # ---- lifecycle: sleep until a stated time, and restart (API Standard 13.2 rule 3)

    def sleep_refusal(self):
        """The collar's safety function is telling the owner where the dog is (and warning of heat). Asleep, it can do
        neither, so it stays awake while that matters. None when sleeping is safe."""
        c = self.c
        crit = [a["code"] for a in self.alerts if a["status"] != "resolved" and a["severity"] == "critical"]
        if crit:
            return (f"A critical alert is open ({crit[0]}). Asleep, the collar could not tell the owner where the dog "
                    "is, and nobody would hear about the fault. Put it right first.")
        if c.lost_mode:
            return "Lost mode is on: the owner is looking for the dog. Turn lost mode off first."
        if not c.inside:
            return ("The dog is outside the safe zone. Asleep, the collar would stop following it and could not say "
                    "where it is. It can sleep once the dog is back.")
        if c.heat_warned:
            return "The collar has warned that the dog may be too hot. It stays awake until that is over."
        return None

    def lifecycle_request(self, op, body):
        """Checks a request. Returns (status, code, detail) for a refusal, or (request, real seconds asleep)."""
        def bad(detail):
            return 400, "invalid-request", detail
        extra = set(body) - LIFE_FIELDS
        if extra:
            return bad(f"Unknown fields: {', '.join(sorted(extra))}")
        if body.get("operation", op) != op:
            return bad("operation in the body must equal the one in the path")
        for k, lo, hi in (("timeoutSec", 1, 86400), ("drainSec", 0, 86400), ("delaySec", 0, 2592000),
                          ("durationSec", 1, MAX_SLEEP_SEC)):
            v = body.get(k, lo)
            if not (isinstance(v, int) and not isinstance(v, bool) and lo <= v <= hi):
                return bad(f"{k} must be a whole number of seconds from {lo} to {hi}")
        if body.get("delaySec", 0) > 0:
            return bad("delaySec needs the schedules module, which this collar does not offer. Ask at the time you want.")
        if body.get("drainSec", 0) > 0:
            return bad("The collar has no running work to drain")
        if not isinstance(body.get("reason", ""), str) or len(body.get("reason", "")) > 500:
            return bad("reason must be text, at most 500 characters")
        if op == "restart":
            if any(k in body for k in ("depth", "wakeAt", "durationSec")):
                return bad("depth, wakeAt and durationSec are for sleep")
            if body.get("mode", "graceful") not in ("graceful", "force"):
                return bad("mode is graceful or force")
            if body.get("target", "system") not in ("system", "heart"):
                return bad("target is system or heart: the collar has no separate components")
            if body.get("mode") == "force":
                return 403, "presence-required", ("A force restart is destructive: it needs the owner's proof of "
                                                  "presence at the collar. Over the network the collar restarts only "
                                                  "gracefully: leave mode out.")
            return body, RESTART_SEC
        if "mode" in body or "target" in body:
            return bad("mode and target are for restart")
        if body.get("depth", "deep") != "deep":
            return bad("This collar sleeps only deeply (depth deep): it has no light sleep that could still answer")
        if ("durationSec" in body) == ("wakeAt" in body):
            return bad(f"Give durationSec (1 to {MAX_SLEEP_SEC} simulated seconds) or wakeAt, not both and not neither")
        if "durationSec" in body:
            return body, body["durationSec"] / self.speed
        try:
            w = body["wakeAt"]
            at = calendar.timegm(time.strptime(re.sub(r"\.\d+", "", w).replace("+00:00", "Z"), "%Y-%m-%dT%H:%M:%SZ"))
        except (TypeError, ValueError):
            return bad("wakeAt must be a UTC time such as 2026-11-12T22:00:00Z")
        real = at - time.time()
        if real <= 0:
            return bad("wakeAt must be in the future")
        if real * self.speed > MAX_SLEEP_SEC:
            return bad(f"wakeAt must be at most {MAX_SLEEP_SEC} simulated seconds (7 days) ahead")
        return body, real

    def lifecycle_post(self, op, body):
        """POST /heart/v1/lifecycle/{op} with the owner key. Write risk, not physical, a human: it runs at once
        (API Standard 6.3). Returns (status, (code, detail)) for a refusal, or (202, Action)."""
        with self.lock:
            checked = self.lifecycle_request(op, body)
            if isinstance(checked[0], int):
                return checked[0], checked[1:]
            request, real = checked
            if self.life:
                return 409, ("conflict", "A restart or sleep is already under way")
            why = self.sleep_refusal() if op == "sleep" else None
            if why:
                log(f"HEART lifecycle  sleep refused: {why}")
                return 409, ("conflict", why)
            aid, operation, now = str(uuid.uuid4()), f"POST /heart/v1/lifecycle/{op}", time.time()
            go = now + GO_DOWN_SEC
            wake = now + real if "wakeAt" in request else go + real
            back = utc(math.ceil(max(go, wake) + WAKE_UP_SEC))
            a = {"id": aid, "operation": operation, "status": "running", "risk": "write", "physical": False,
                 "createdAt": utc(), "updatedAt": utc(), "initiator": {"kind": "human", "id": "owner"},
                 "arguments": request, "cancellable": False, "nextReachableAt": back,
                 "digest": hashlib.sha256(canonical({"system": SYSTEM_ID, "id": aid, "operation": operation,
                                                     "arguments": request, "physical": False}).encode()).hexdigest()}
            if op == "sleep":
                a["explanation"] = (f"Deep sleep for {real * self.speed:.0f} simulated s ({real:.0f} s at {self.speed:g}x): "
                                    "no GNSS, no safe-zone check, no reports to the owner's app, no HEART, no HEART "
                                    f"Monitor. It cannot answer until it wakes, by its timer (back by {back}) or by its "
                                    "button.")
            else:
                a["explanation"] = f"A graceful restart: the collar is unreachable for about {RESTART_SEC} s (back by {back})."
            self.actions = (self.actions + [a])[-500:]
            self.life = {"a": a, "op": op, "go": go, "wake": wake, "by": None}
            log(f"HEART lifecycle  {op} accepted: {a['explanation']}")
            return 202, a

    def lifecycle_view(self, owner=False):
        with self.lock:
            d = {"phase": self.phase}
            if owner:
                ops = [] if self.life else [op for op in LIFECYCLE_OPS if op != "sleep" or not self.sleep_refusal()]
                d.update(allowed=ops, since=self.phase_since, wakeSources=list(WAKE_SOURCES))
        return self.envelope("heart.lifecycle", d, "internal" if owner else "public")

    def button_wake(self):
        """The collar's button, pressed while it sleeps: it wakes at once."""
        with self.lock:
            if self.life and self.phase == "sleeping":
                self.life["by"] = "button"
                self.life["wake"] = time.time()

    def lifecycle_tick(self, api):
        """Called often by the main loop: carries out an accepted sleep or restart, and wakes when it is time."""
        with self.lock:
            life = self.life
            if not life:
                return
            now, c, a = time.time(), self.c, life["a"]
            if self.phase == "running":
                if now < life["go"]:
                    return
                for b in self.actions:               # nothing can wait for a person while the collar is down
                    if b["status"] == "awaiting-approval":
                        b.update(status="cancelled", updatedAt=utc(), error=f"The collar went to {life['op']} first.")
                self.phase = "sleeping" if life["op"] == "sleep" else "starting"
                self.phase_since, life["went"], life["led"] = utc(), now, c.led
                c.set_led("off")
                a["updatedAt"] = utc()
                if life["op"] == "sleep":
                    log(f"HEART lifecycle  asleep (deep): it answers nothing until {a['nextReachableAt']}, or until "
                        "touch button.press")
                else:
                    log("HEART lifecycle  restarting: it answers nothing for a few seconds")
                down = True
            elif now >= life["wake"]:
                by = life["by"] or "timer"
                slept = (now - life["went"]) * self.speed     # simulated seconds; the world's clock steps once a second
                c.set_led(life["led"])
                if life["op"] == "sleep":
                    result = {"note": "Slept, and woke again", "wokeBy": by, "wokeAt": utc(),
                              "sleptSimulatedSec": round(slept)}
                    log(f"HEART lifecycle  awake again, woken by its {by} after {slept:.0f} simulated s")
                else:
                    result = {"note": "Restarted, and running again"}
                    log("HEART lifecycle  restarted, and running again")
                a.update(status="succeeded", updatedAt=utc(), result=result)
                self.phase, self.phase_since, self.life = "running", utc(), None
                down = False
            else:
                return
        if api:
            api.stop() if down else api.start()

    # ---- 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": "offline" if self.c.link_down else "tracking"}
                d["autonomy"] = {"current": 1}
        return self.envelope("heart.pulse", d, "internal" if owner else "public")

    def public_alerts(self, n=16):
        with self.lock:
            keep = ("id", "code", "severity", "status", "detectedAt", "resolvedAt")
            return [self.envelope("heart.arrhythmia", {k: a[k] for k in keep if k in a}) for a in reversed(self.alerts[-n:])]

    def properties(self, owner=False):
        c = self.c
        with self.lock:
            items = [  # name, value, unit, privacy, writable
                ("battery_pct", round(c.battery, 1), "%", "public", False),
                ("battery_status", "empty" if c.battery <= BATTERY_EMPTY else "low" if c.battery < BATTERY_LOW else "ok",
                 None, "public", False),
                ("battery_hours_left", c.hours_left(), "h", "internal", False),
                ("collar_temp", round(c.temp, 1), "Cel", "internal", False),
                ("gnss_state", c.gnss_state, None, "internal", False),
                ("gnss_satellites", c.sats, "1", "internal", False),
                ("gnss_hdop", c.hdop, "1", "internal", False),
                ("uplink", "down" if c.link_down else "home-base" if c.home_base else "lte-m", None, "internal", False),
                ("uplink_rssi", None if c.link_down or c.home_base else round(c.rssi), "dB[mW]", "internal", False),
                ("report_interval_s", REPORT_LOST if c.lost_mode else REPORT_NORMAL, "s", "internal", False),
                ("reports_queued", c.queued, "1", "internal", False),
                ("accel_dynamic", round(c.accel_dynamic, 3), "[g]", "internal", False),
                ("led_mode", c.led, None, "internal", True),
                ("lost_mode", c.lost_mode, None, "private", True),
                ("activity", c.activity, None, "private", False),
                ("position_east_m", round(c.pos[0], 1), "m", "private", False),
                ("position_north_m", round(c.pos[1], 1), "m", "private", False),
                ("position_source", c.pos_source, None, "private", False),
                ("position_age_s", round(c.t - c.pos_t), "s", "private", False),
                ("distance_from_home_m", round(dist(c.pos)), "m", "private", False),
                ("in_safe_zone", c.inside, None, "private", False),
                ("home_base_in_range", c.home_base, None, "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

    def rhythm(self):
        with self.lock:
            return [self.envelope("heart.rhythm", json.loads(json.dumps(e)), "private") for e in reversed(self.c.events)]


CAPABILITIES = [
    {"name": "collar.lost_mode", "title": "Lost mode", "risk": "write", "privacy": "internal", "physical": False,
     "callers": ["human"], "reversible": True, "idempotent": True,
     "description": "Turn lost mode on or off: a GNSS fix and a position report every 30 s instead of every 10 minutes. "
                    "Not physical: it changes reporting, nothing the dog feels. It costs battery: about a day instead of weeks.",
     "inputSchema": {"type": "object", "properties": {"on": {"type": "boolean"}}, "required": ["on"],
                     "additionalProperties": False}},
    {"name": "collar.led", "title": "LED", "risk": "write", "privacy": "internal", "physical": True,
     "callers": ["human"], "reversible": True, "idempotent": True,
     "description": "Set the collar's LED: off, steady or blink, to see the dog in the dark. Physical: light at a living "
                    "animal's neck. Waits for the owner to confirm with their passkey.",
     "inputSchema": {"type": "object", "properties": {"mode": {"enum": list(LED_MODES)}}, "required": ["mode"],
                     "additionalProperties": False}},
    {"name": "collar.buzzer", "title": "Buzzer", "risk": "write", "privacy": "internal", "physical": True,
     "callers": ["human"], "reversible": False,
     "description": f"Sound the buzzer, to call the dog or find the collar. Physical: sound at a dog's ear. Waits for the "
                    f"owner to confirm with their passkey. The collar's guard refuses more than {SAFE_VOLUME} %, more "
                    f"than {SAFE_BUZZ_SEC} s, or sooner than {BUZZ_REST_SEC} s after the last time.",
     "inputSchema": {"type": "object", "properties": {"volume": {"type": "integer", "minimum": 0, "maximum": MAX_VOLUME},
                                                      "durationSec": {"type": "integer", "minimum": 1, "maximum": MAX_BUZZ_SEC}},
                     "required": ["volume", "durationSec"], "additionalProperties": False},
     "guards": {"minIntervalSec": BUZZ_REST_SEC}},
]


# ------------------------------------------------------------------------------------------------ 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 dog collar
    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-dog-collar-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, collar, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated uplink is lost: then the site shows a Flatline after three missed intervals. Only the public Pulse and
    public alerts are sent: never a position, a safe-zone event or a reading."""
    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 collar.link_down or heart.phase != "running":   # asleep or restarting: silent
            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", "lifecycle"],
                     "shares": {"pulse": "public", "arrhythmia": "public", "property": "private", "rhythm": "private",
                                "action": "internal", "lifecycle": "public"},
                     "autonomy": {"level": 1}, "template": TEMPLATE}
                if owner:
                    d["capabilities"] = CAPABILITIES     # internal: shown only to a caller with that clearance
                    d["lifecycle"] = {"operations": list(LIFECYCLE_OPS), "wakeSources": list(WAKE_SOURCES)}
                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/lifecycle":
                return self.send(200, heart.lifecycle_view(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 in ("/heart/v1/rhythm", "/heart/v1/track", "/heart/v1/actions") or path.startswith("/heart/v1/actions/"):
                if not owner:
                    return self.problem(403, "insufficient-clearance",
                                        "Where the dog is and has been, and the owner's changes, are private: send the owner key")
                if path == "/heart/v1/rhythm":
                    return self.send(200, {"items": heart.rhythm()}, "application/json")
                with heart.lock:
                    if path == "/heart/v1/track":
                        return self.send(200, {"items": [{"at": at, "eastM": e, "northM": n}
                                                         for _, e, n, at in reversed(heart.c.fixes)]}, "application/json")
                    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")
                    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")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties, /rhythm, /track or /actions")

        def do_POST(self):
            path = self.path.split("?")[0].rstrip("/")
            parts = path.split("/")
            if len(parts) == 5 and parts[1:4] == ["heart", "v1", "lifecycle"]:
                return self.lifecycle(parts[4])
            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")
            args = body.get("arguments") or {}
            if path == "/heart/v1/capabilities/collar.lost_mode/invoke":
                on = args.get("on")
                if not isinstance(on, bool):
                    return self.problem(400, "invalid-request", "arguments.on must be true or false")
                return self.action(heart.owner_lost_mode(on))
            if path == "/heart/v1/capabilities/collar.led/invoke":
                mode = args.get("mode")
                if mode not in LED_MODES:
                    return self.problem(400, "invalid-request", "arguments.mode must be off, steady or blink")
                with heart.lock:
                    before = heart.c.led
                return self.action(heart.owner_physical("collar.led", {"mode": mode}, f"LED {before} -> {mode} (physical)"), 202)
            if path == "/heart/v1/capabilities/collar.buzzer/invoke":
                v, s = args.get("volume"), args.get("durationSec")
                if not all(isinstance(x, int) and not isinstance(x, bool) for x in (v, s)):
                    return self.problem(400, "invalid-request", "arguments.volume and arguments.durationSec must be whole numbers")
                if not (0 <= v <= MAX_VOLUME and 1 <= s <= MAX_BUZZ_SEC):
                    return self.problem(422, "out-of-range", f"volume 0 to {MAX_VOLUME} %, durationSec 1 to {MAX_BUZZ_SEC}")
                with heart.lock:
                    why = heart.c.buzz_guard(v, s)
                if why:
                    return self.problem(409, "guard-failed", why)
                return self.action(heart.owner_physical("collar.buzzer", {"volume": v, "durationSec": s},
                                                        f"buzzer {v} % for {s} s (physical)"), 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)
            self.refuse()

        def lifecycle(self, op):
            if op not in STANDARD_OPS and not re.fullmatch(r"x-[a-z0-9][a-z0-9-]*", op):
                return self.problem(404, "not-found", "No such lifecycle operation")
            if not self.owner():
                return self.problem(401, "unauthenticated", "Lifecycle operations need the owner key")
            if op not in LIFECYCLE_OPS:
                why = ("A collar in deep sleep cannot answer HTTP, so wake is not offered: it wakes by its timer or its "
                       "button." if op == "wake" else f"This collar does not offer {op}. It offers restart and sleep.")
                return self.problem(501, "not-implemented", why)
            body = self.body()
            if body is None:
                return self.problem(400, "invalid-request", "The body must be a JSON object")
            status, out = heart.lifecycle_post(op, body)
            return self.action(out, 202) if status == 202 else self.problem(status, *out)

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "Only lost mode, the LED and the buzzer can be changed (POST /heart/v1/capabilities/{name}/invoke), "
                         "and the collar can sleep or restart (POST /heart/v1/lifecycle/{operation}). "
                         "The safe zone, GNSS, the uplink and everything else are set in the owner's app, not through HEART.")

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

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


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

INSTANCE = "sim"


class Api:
    """The local HEART API server. Stopped while the collar sleeps or restarts: its port is closed, so a caller's
    connection is refused, as with a real collar whose radio is off."""

    def __init__(self, heart, port):
        self.heart = heart
        try:
            self.srv = ThreadingHTTPServer(("127.0.0.1", port), handler(heart))
        except OSError:
            self.srv = ThreadingHTTPServer(("127.0.0.1", 7685), handler(heart))   # the standard fallback port
        self.port = self.srv.server_address[1]
        self.serve()

    def serve(self):
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()

    def stop(self):
        self.srv.shutdown()
        self.srv.server_close()

    def start(self):
        for _ in range(20):
            try:
                self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), handler(self.heart))
                return self.serve()
            except OSError:
                time.sleep(0.25)
        log(f"HEART API: could not listen on port {self.port} again")


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated dog tracking collar 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="dog-collar-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43281, 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)
    dog = Dog(rng, scenarios, clock0)
    collar = Collar(rng, scenarios, dog)
    heart = Heart(collar)
    heart.speed = args.speed
    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 dog, owner and collar; positions are metres from a made-up home, never a real place.
  HEART watches the collar's health. Where the dog is stays private: it never reaches HEART Monitor.
  Lost mode runs at once with the owner key; the LED and the buzzer act on the dog and wait for the owner's passkey.
  Scenarios: {', '.join(sorted(scenarios)) or 'none (random life only)'} · speed {args.speed}x
""", file=sys.stderr, flush=True)

    stop = threading.Event()
    api = None
    if args.port:
        api = Api(heart, args.port)
        log(f"HEART API on http://127.0.0.1:{api.port}/heart/v1/pulse  (owner key for private views: {heart.owner_key})")
    if not args.no_monitor:
        threading.Thread(target=monitor, args=(heart, collar, 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 5 s holds
    try:
        while not stop.is_set():
            left = args.speed
            while left > 1e-9:
                dt = min(sub, left)
                left -= dt
                t += dt
                awake = heart.phase == "running"     # asleep or restarting: the world goes on, the collar does not
                with heart.lock:
                    dog.step(t, dt)
                    collar.step(t, dt) if awake else collar.doze(t, dt)
                if awake:
                    heart.tend(t)
            if os.path.exists("button.press"):
                os.unlink("button.press")
                heart.button_wake() if heart.phase == "sleeping" else heart.button_press()
            if time.time() - last_print >= 15 and heart.phase != "running":
                last_print = time.time()
                log(f"{time.strftime('%H:%M', time.localtime(clock0 + t))} dog {dog.mode} · collar "
                    f"{'ASLEEP' if heart.phase == 'sleeping' else 'RESTARTING'} (deep sleep: "
                    f"no GNSS, no safe-zone check, no reports, no HEART) · battery {collar.battery:4.1f} %")
            elif time.time() - last_print >= 15:
                last_print = time.time()
                c = collar
                fix = f"GNSS {c.gnss_state}" + (f" ({c.sats} sats)" if c.gnss_state != "off" else "")
                log(f"{time.strftime('%H:%M', time.localtime(clock0 + t))} dog {dog.mode} · collar {'on' if c.worn else 'OFF'} · "
                    f"{c.activity} · {'in' if c.inside else 'OUT OF'} safe zone · {fix} · "
                    f"uplink {'down' if c.link_down else 'home base' if c.home_base else f'LTE-M {c.rssi:.0f} dBm'} · "
                    f"lost mode {'ON' if c.lost_mode else 'off'} · battery {c.battery:4.1f} % (~{c.hours_left():.0f} h) · "
                    f"{c.temp:4.1f} Cel · HEART {heart.status}")
            if args.duration and time.time() - wall0 >= args.duration:
                break
            for _ in range(5):                     # the lifecycle needs a finer clock than the world
                heart.lifecycle_tick(api)
                time.sleep(0.2)
    except KeyboardInterrupt:
        pass
    stop.set()
    log("stopped. The monitor site will show a Flatline if this collar stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo