Home EV charger: what is inside

A 7.4 kW home wallbox and the cars that use it. When a car charges shows when people are home, so every session stays private; HEART changes nothing. HEART runs on the charger's communication board, beside its certified safety controller and never in it.

Specifications it follows

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

World  ──►  Charger  ──(read only)──►  Heart  ──►  local HEART API      (127.0.0.1)
cars, grid,  the certified              watches     └►  HEART Hub    (outbound only)
weather,     controller and its
network      protections

Run it yourself, and register it on HEART Hub

1. Download README.md, evChargerSimulator.py, test_e2e.py.
2. Run python3 evChargerSimulator.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: Home EV charger.

The code: evChargerSimulator.py

673 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/ev-charger-simulator/v1.2611.2
"""evChargerSimulator.py: a simulated home EV charger with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. It is not a charger, controls no real hardware, and must never be used to run, test or
certify one. Its numbers are plausible, not engineering data. See README.md beside this file, and the case study
guides/EV-Charger.md.

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

  * a simulated WORLD around one HOME charger (one tethered connector, 32 A single-phase, about 7.4 kW): cars
    that arrive on random timers, mostly in the evening, each with its own battery size, state of charge and
    charging curve, and leave again; the outdoor temperature; the grid voltage with dips and the odd outage; the
    home network that carries the charger's link to its backend;
  * a simulated CHARGER (the certified controller): it offers current to the car, closes and opens its contactor,
    counts energy with its meter, and protects itself: a residual current device (RCD) that trips on an earth
    leak, over-temperature derating of the connector, under-voltage pausing, a weld check on the contactor;
  * a HEART that only WATCHES the charger. It never changes charging: nothing in this file lets it. 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.

Run it:

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

Scenarios (--scenario, comma-separated): meter, hot, grid, link, rcd, contactor, weld, all. Each fault starts at a
set time, lasts a while, and is then "fixed by the electrician", so you can watch an alert open, escalate and
resolve. With any scenario, a demo car plugs in at once and charges through the run, until the weld scenario
sends it away.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. When a
car charges, how much and for how long stay local (private): they show when people are at home. Card IDs and
vehicle IDs are never part of HEART at all: this file does not even simulate them.
"""
import argparse
import hashlib
import json
import os
import random
import secrets
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SYSTEM_ID = "org.heartaispec.demo.ev-charger"
NAME = "EV Charger Simulator (demo, not a real charger)"
VERSION = "1.0.0"
FIRMWARE = "4.2.1"                        # the simulated charger's own firmware, not the HEART's version
TEMPLATE = "heart/embeddedstandard/ev-charger-simulator/v1.2611.2"
PULSE_SEC = 30
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

# Ratings of the simulated charger, set by its installer. Only the charger's certified controller enforces them.
CABLE_A = 32                              # the tethered cable's rating: no current above it, ever
SUPPLY_A = 32                             # the circuit breaker the installer fitted
MIN_A = 6                                 # IEC 61851: below 6 A a car is not offered charging at all
V_NOMINAL = 230
V_OK = (207, 253)                         # +-10 %: outside this, the supply is unstable
V_PAUSE = (195, 260)                      # the charger's own protection pauses charging outside this
DERATE_C, RESUME_C, STOP_C = 70.0, 60.0, 90.0   # connector temperature: derate above, back to full below, stop above


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


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


# ------------------------------------------------------------------------------------------------ the simulation

class Timer:
    """A value that changes on its own random timer (in simulated seconds)."""

    def __init__(self, rng, lo, hi):
        self.rng, self.lo, self.hi, self.at = rng, lo, hi, 0.0

    def due(self, t):
        if t < self.at:
            return False
        self.at = t + self.rng.uniform(self.lo, self.hi)
        return True


class Episode:
    """Something that happens now and then, for a while: next start after a random gap, lasting a random time."""

    def __init__(self, rng, gap, length):
        self.rng, self.gap, self.length = rng, gap, length
        self.start = rng.uniform(*gap)
        self.end = 0.0

    def active(self, t):
        if t >= self.start and self.end <= self.start:
            self.end = self.start + self.rng.uniform(*self.length)
        if self.end > self.start and t >= self.end:
            self.start = t + self.rng.uniform(*self.gap)
        return self.start <= t < self.end


class Car:
    """One visiting car. Its battery and charging curve are its own; the charger only sees how much it draws."""

    def __init__(self, rng, t, capacity=None, soc=None, max_a=None, stay=None):
        self.capacity = capacity or rng.choice([40, 58, 64, 77, 82, 100])       # kWh
        self.soc = soc if soc is not None else rng.uniform(0.1, 0.7)            # state of charge, 0 to 1
        self.target = rng.choice([0.8, 0.9, 1.0])
        self.max_a = max_a or rng.choice([16, 32, 32])                          # the car's on-board charger
        self.leave_at = t + (stay or rng.uniform(1.5 * 3600, 13 * 3600))

    def wants(self):
        """The charging curve: full current up to 80 %, then tapering towards the target."""
        if self.soc >= self.target:
            return 0.0
        if self.soc < 0.8:
            return float(self.max_a)
        return max(MIN_A, self.max_a * (1 - 0.8 * (self.soc - 0.8) / 0.2))

    def charge(self, amps, volts, dt):
        self.soc = min(1.0, self.soc + volts * amps * dt * 0.92 / 3.6e6 / self.capacity)


class World:
    """Around the charger: the cars, the weather, the grid and the home network. Private: it never leaves here."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.car, self.ambient, self.voltage = None, 16.0, 231.0
        self.t_arrive, self.t_amb, self.t_volt = Timer(rng, 120, 600), Timer(rng, 60, 300), Timer(rng, 2, 8)
        self.dip = Episode(rng, (3600, 4 * 3600), (10, 90))                   # a neighbour's heat pump, a fault nearby
        self.outage = Episode(rng, (3 * 86400, 9 * 86400), (60, 1800))       # a power cut
        self.net = Episode(rng, (6 * 3600, 24 * 3600), (60, 600))            # the home router restarts
        self.leak = Episode(rng, (4 * 86400, 12 * 86400), (1200, 5400))      # an earth leak, until someone resets
        self.demo_car = bool(scenarios)

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

    def step(self, t, wall):
        hour = time.localtime(wall).tm_hour
        if self.t_amb.due(t):
            target = 8 + 10 * (1 - abs(hour - 15) / 12)                       # warmest mid-afternoon
            self.ambient += (target - self.ambient) * 0.2 + self.rng.gauss(0, 0.3)
        if self.t_volt.due(t):
            self.voltage = V_NOMINAL + 2 + self.rng.gauss(0, 1.5)
            if self.dip.active(t):
                self.voltage = self.rng.uniform(196, 205)
            if self.staged("grid", t, 480, 60):                                # first unstable ...
                self.voltage = self.rng.choice([self.rng.uniform(186, 203), self.rng.uniform(255, 259)])
        if self.outage.active(t) or self.staged("grid", t, 540, 60):          # ... then lost
            self.voltage = 0.0
        elif self.voltage == 0.0:
            self.voltage = V_NOMINAL
        # Cars: the demo car arrives at once and leaves when the weld scenario starts. Others come on random timers,
        # mostly in the evening, and leave after a few hours or the next morning.
        if self.demo_car and self.car is None and 20 <= t < 1080:
            self.car = Car(self.rng, t, capacity=82, soc=0.12, max_a=32, stay=1080 - t)
            self.demo_car = False
        if self.car and t >= self.car.leave_at:
            self.car = None
        if self.car is None and self.t_arrive.due(t) and not (self.sc and t < 1200):
            chance = 0.35 if 17 <= hour < 22 else 0.02 if hour >= 23 or hour < 6 else 0.06
            if self.rng.random() < chance:
                self.car = Car(self.rng, t)


class Charger:
    """The simulated charger and its certified controller. The HEART reads these values. Nothing in this file lets
    the HEART change them: current, start and stop, and every protection stay here (and with the backend)."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.state, self.current, self.offered, self.temp = "available", 0.0, 0.0, 16.0
        self.meter, self.session_count, self.sessions = rng.uniform(800, 4000), 0, []   # kWh register; private log
        self.command_closed = self.closed = False
        self.rcd_tripped = self.welded = self.derating = self.link_down = False
        self.cycles = rng.randint(400, 3000)                                  # contactor operations so far
        self.session, self.car_was = None, None
        self.reason = ""                                                      # why it is faulted, in plain words

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

    def step(self, t, dt, world):
        car, v = world.car, world.voltage
        hot_pin = self.staged("hot", t, 220, 240)                            # a worn, high-resistance contact
        meter_stuck = self.staged("meter", t, 60, 140)
        not_closing = self.staged("contactor", t, 960, 100)                  # a burnt coil
        self.link_down = world.net.active(t) or self.staged("link", t, 620, 180)

        # The RCD trips on an earth leak (in the car, the cable or the wiring) and stays tripped until a person resets
        # it at the charger. Random life: the owner resets it after a while. Staged: the electrician, at the end.
        if (world.leak.active(t) and self.current > 0) or self.staged("rcd", t, 820, 30):
            self.rcd_tripped = True
        if self.rcd_tripped and not world.leak.active(t) and not self.staged("rcd", t, 820, 120):
            self.rcd_tripped = False
            log("charger: RCD reset by a person at the charger, after checking the car and cable")
        # The weld: when the demo car leaves at the start of the weld scenario, the contactor does not open.
        if self.staged("weld", t, 1080, 30) and self.closed:
            self.welded = True
        if self.welded and not self.staged("weld", t, 1080, 120):
            self.welded, self.closed = False, False
            log("charger: contactor replaced by the electrician")

        # Sessions: a car plugs in or leaves. No card ID and no vehicle ID: HEART never sees one, so none is simulated.
        if car is not self.car_was:
            if self.session:
                self.session.update(end=utc(), kWh=round(self.meter - self.session.pop("_m0"), 2))
                self.sessions = (self.sessions + [self.session])[-50:]
                self.session = None
            if car:
                self.session_count += 1
                self.session = {"start": utc(), "_m0": self.meter, "peakA": 0.0}
            self.car_was = car

        # The controller: how much current may be offered, and whether the contactor should be closed.
        if self.temp > DERATE_C:
            self.derating = True
        elif self.temp < RESUME_C:
            self.derating = False
        limit = min(CABLE_A, SUPPLY_A, 16 if self.derating else CABLE_A)
        if self.temp >= STOP_C or not V_PAUSE[0] <= v <= V_PAUSE[1]:
            limit = 0
        if self.rcd_tripped or self.welded:
            self.state, self.offered, self.command_closed = "faulted", 0.0, False
            self.reason = "rcd" if self.rcd_tripped else "weld"
        elif v == 0:
            self.state, self.offered, self.command_closed, self.reason = "unavailable", 0.0, False, ""
        elif car is None:
            self.state, self.offered, self.command_closed, self.reason = "available", 0.0, False, ""
        elif car.wants() == 0:
            self.state, self.offered, self.command_closed, self.reason = "suspended-ev", float(limit), False, ""
        elif limit < MIN_A:
            self.state, self.offered, self.command_closed, self.reason = "suspended-evse", 0.0, False, ""
        else:
            self.offered, self.command_closed = float(limit), True
            self.state = "charging" if self.closed or not not_closing else "preparing"
            self.reason = ""
        # The contactor follows its command, unless its coil is burnt (it will not close) or it is welded (it will
        # not open). Its mirror contact tells the controller where it really is.
        was = self.closed
        if not self.welded:
            self.closed = self.command_closed and v > 0 and not not_closing
        if self.closed and not was:
            self.cycles += 1
        if self.closed and car and v > 0 and self.offered >= MIN_A:
            self.current = min(self.offered, car.wants()) + self.rng.gauss(0, 0.1)
            car.charge(self.current, v, dt)
        else:
            self.current = 0.0
        self.current = max(0.0, self.current)
        if not meter_stuck:
            self.meter += v * self.current * dt / 3.6e6
        if self.session:
            self.session["peakA"] = round(max(self.session["peakA"], self.current), 1)
        # The connector warms with current squared and with the weather.
        rise = (240 if hot_pin else 22) * (self.current / CABLE_A) ** 2
        self.temp += (world.ambient + rise - self.temp) * min(1.0, dt / 90)
        self.voltage = v

    @property
    def power_kw(self):
        return self.voltage * self.current / 1000

    @property
    def contactor_health(self):
        if self.welded:
            return "welded"
        if self.command_closed and not self.closed:
            return "not-closing"
        return "ok"


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

def lost_summary(c, h):
    return {"rcd": "The residual current device has tripped: charging is stopped",
            "weld": "The contactor is welded: the charger has locked itself out"}.get(
        c.reason, "The contactor does not close: the car cannot charge")


RULES = [
    # code, severity, condition, summary (text, or a function of the charger and the HEART)
    ("system.main-function-lost", "critical",
     lambda c, h: c.rcd_tripped or c.welded or c.contactor_health == "not-closing", lost_summary),
    ("actuators.not-responding", "error", lambda c, h: c.contactor_health != "ok",
     lambda c, h: "The contactor does not follow its command"),
    ("performance.too-hot", "warning", lambda c, h: c.derating or c.temp >= STOP_C,
     "The connector is too hot: charging current is reduced"),
    ("power.supply-lost", "error", lambda c, h: c.voltage == 0, "Grid supply is lost"),
    ("power.supply-unstable", "warning", lambda c, h: c.voltage > 0 and not V_OK[0] <= c.voltage <= V_OK[1],
     "Grid voltage is outside its normal range"),
    ("connectivity.link-lost", "warning", lambda c, h: c.link_down, "The charger cannot reach its backend"),
    ("sensors.stuck", "error", lambda c, h: h.meter_still >= 60, "The energy meter has stopped counting"),
]

EXPLAIN = {
    "system.main-function-lost": lambda c, h: {
        "rcd": "Earth-leakage protection tripped (a residual current in the car, the cable or the wiring). The charger "
               "is safe and will stay off. A qualified person must inspect the car and cable and reset it at the "
               "charger. HEART never resets it, and nothing can do so remotely.",
        "weld": f"The contactor stayed closed after the car left (its mirror contact reads closed, command open). The "
                f"charger has isolated itself and refuses to charge. Call the installer: the contactor must be replaced "
                f"({c.cycles} operations so far).",
    }.get(c.reason, f"The car asked for charge and the controller closed the contactor, but it stayed open: no current "
                    f"flows. A burnt coil or a failed relay driver is likely. Call the installer."),
    "actuators.not-responding": lambda c, h: f"Command {'closed' if c.command_closed else 'open'}, mirror contact "
                                             f"{'closed' if c.closed else 'open'} ({c.contactor_health}). The charger's own "
                                             f"protection handles it; this is for the installer.",
    "performance.too-hot": lambda c, h: f"Connector at {c.temp:.0f} °C (derates above {DERATE_C:.0f} °C, stops at "
                                        f"{STOP_C:.0f} °C): the charger has cut the current to 16 A. A worn or dirty "
                                        f"connector pin is the usual cause. Have it inspected.",
    "power.supply-lost": lambda c, h: "No grid voltage at the charger. Check the circuit breaker and whether there "
                                      "is a power cut. HEART reports this on its small backup supply.",
    "power.supply-unstable": lambda c, h: f"{c.voltage:.0f} V, outside {V_OK[0]} to {V_OK[1]} V. The charger pauses "
                                          f"itself below {V_PAUSE[0]} V or above {V_PAUSE[1]} V. If it happens often, "
                                          f"tell the electricity network operator.",
    "connectivity.link-lost": lambda c, h: "The charger's network link is down: its backend (over OCPP) and HEART "
                                           "Monitor cannot be reached. Charging carries on locally.",
    "sensors.stuck": lambda c, h: f"Current is flowing ({c.current:.0f} A), but the energy register has not moved "
                                  f"for {h.meter_still:.0f} s. Charging is safe, but billing and energy records are "
                                  f"wrong until the meter is fixed.",
}


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

    def __init__(self, charger, world):
        self.c, self.world = charger, world
        self.lock = threading.Lock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)
        self.meter_seen, self.meter_since, self.meter_still = None, 0.0, 0.0

    def tend(self, t):
        with self.lock:
            self.sequence += 1
            # The HEART's own check on the meter: while current flows, the energy register must move.
            if self.c.current <= 1 or self.c.meter != self.meter_seen:
                self.meter_seen, self.meter_since = self.c.meter, t
            self.meter_still = t - self.meter_since
            for code, severity, cond, summary in RULES:
                hit = cond(self.c, self)
                # A condition must hold for 5 simulated seconds before it is raised: one odd reading is not a fault.
                prev = self.since.get(code)
                self.since[code] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[code] >= 5
                open_ = next((a for a in self.alerts if a["code"] == code and a["status"] != "resolved"), None)
                if held and open_ is None:
                    text = summary(self.c, self) if callable(summary) else summary
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": text, "explanation": EXPLAIN[code](self.c, self)}
                    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"

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

    def envelope(self, kind, data, privacy="public"):
        return {"specversion": "1.0", "id": str(uuid.uuid4()), "source": f"/heart/{SYSTEM_ID}/{INSTANCE}", "type": kind,
                "time": utc(), "datacontenttype": "application/json", "heartversion": "1.0", "heartprivacy": privacy,
                "data": data}

    def in_service(self):
        return self.c.state not in ("faulted", "unavailable")

    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:
                # internal, so no connector state here: whether a car is plugged in is private
                d["reconnaissance"] = {"state": "in-service" if self.in_service() else "out-of-service"}
                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, w, s = self.c, self.world, self.c.session
        items = [  # name, value, unit, privacy
            ("charger_state", "in-service" if self.in_service() else "out-of-service", None, "public"),
            ("firmware_version", FIRMWARE, None, "internal"),
            ("grid_voltage", round(c.voltage, 1), "V", "internal"),
            ("rcd_state", "tripped" if c.rcd_tripped else "ok", None, "internal"),
            ("contactor_health", c.contactor_health, None, "internal"),
            ("contactor_cycles", c.cycles, "1", "internal"),
            ("network_link", "down" if c.link_down else "up", None, "internal"),
            ("session_count", c.session_count, "1", "internal"),
            ("ambient_temperature", round(w.ambient, 1), "Cel", "internal"),
            ("connector_state", c.state, None, "private"),
            ("current", round(c.current, 1), "A", "private"),
            ("current_offered", round(c.offered, 1), "A", "private"),
            ("power", round(c.power_kw, 2), "kW", "private"),
            ("session_energy", round(c.meter - s["_m0"], 2) if s else 0.0, "kW.h", "private"),
            ("meter_register", round(c.meter, 2), "kW.h", "private"),
            ("connector_temperature", round(c.temp, 1), "Cel", "private"),
        ]
        out = []
        for name, value, unit, privacy in items:
            if privacy != "public" and not owner:
                continue
            d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": value, "writable": False, "time": utc()}
            if unit:
                d["unit"] = unit
            out.append(self.envelope("heart.property", d, privacy))
        return out

    def session_log(self):
        """Private: when a car charged and how much. Times and energy only: no card ID, no vehicle ID."""
        with self.lock:
            done = list(reversed(self.c.sessions))
            now = self.c.session
            if now:
                s = {k: v for k, v in now.items() if not k.startswith("_")}
                done.insert(0, {**s, "end": None, "kWh": round(self.c.meter - now["_m0"], 2)})
            return done


# ------------------------------------------------------------------------------------------------ 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 EV charger
    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-ev-charger-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, charger, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated network link is down: then the site shows a Flatline after three missed intervals."""
    host = url.split("://", 1)[-1].split("/")[0].split(":")[0]
    if not url.startswith("https://") and host not in ("127.0.0.1", "localhost"):
        log("HEART Monitor: refusing a site without https")
        return
    ident, phase, next_at, last, sent, backoff, claimed = identity(path, url), "hello", 0.0, 0.0, None, 0, None
    show_claim(ident, url)
    while not stop.is_set():
        with heart.lock:
            seen = (heart.status, tuple((a["code"], a["status"]) for a in heart.alerts[-16:]))
        due = time.time() >= next_at or (phase == "report" and seen != sent and time.time() - last >= 10)
        if not due or (phase == "report" and charger.link_down):
            stop.wait(1)
            continue
        try:
            if phase == "hello":
                status, reply = post(url, f"/v1/devices/{ident['id']}/hello", {
                    "claimHash": hashlib.sha256(f"heart-claim:{ident['claimCode']}".encode()).hexdigest(),
                    "keyHash": hashlib.sha256(ident["key"].encode()).hexdigest(), "template": TEMPLATE})
                if status in (200, 201):
                    phase, next_at = "report", 0
                    log(f"HEART Monitor: {url} knows this device")
                elif status == 409:
                    ident = identity(path, url, new=True)
                    show_claim(ident, url)
                else:
                    log(f"HEART Monitor: hello answered {status}; trying again in 10 minutes")
                    next_at = time.time() + 600
                continue
            last, sent = time.time(), seen
            status, reply = post(url, f"/v1/devices/{ident['id']}/report",
                                 {"pulse": heart.pulse(), "alerts": heart.public_alerts()}, ident["key"])
            backoff = 0
            if status in (200, 202) and reply.get("claimed") != claimed:
                claimed = reply.get("claimed")
                log("HEART Monitor: registered, reports are now stored" if claimed else
                    "HEART Monitor: not registered yet, nothing is stored (see the claim code above)")
            if status in (200, 202, 429):
                next_at = time.time() + int(reply.get("nextSec") or PULSE_SEC)
            elif status == 404:
                phase = "hello"
            elif status == 401:
                log("HEART Monitor: the site refused this device's key. Reporting has stopped.")
                return
            else:
                log(f"HEART Monitor: report answered {status} {reply.get('detail', '')}; not retrying it unchanged")
                next_at = time.time() + 600
        except (OSError, ValueError) as e:
            backoff = min(backoff * 2 or 30, 600)
            log(f"HEART Monitor: offline ({e.__class__.__name__}); trying again in {backoff} s. Noticing goes on.")
            next_at = time.time() + backoff


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

def handler(heart):
    class H(BaseHTTPRequestHandler):
        def send(self, status, body, ctype="application/cloudevents+json"):
            raw = json.dumps(body, indent=1).encode()
            self.send_response(status)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(raw)))
            self.send_header("Heart-Version", "1.0")
            self.end_headers()
            self.wfile.write(raw)

        def problem(self, status, code, detail):
            self.send(status, {"type": f"https://heartcortex.ai/problems/{code}", "title": code.replace("-", " "),
                               "status": status, "detail": detail, "code": code}, "application/problem+json")

        def owner(self):
            return secrets.compare_digest(self.headers.get("Authorization", ""), f"Bearer {heart.owner_key}")

        def do_GET(self):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            if path == "/.well-known/heart-card.json":
                port = self.server.server_address[1]
                return self.send(200, heart.envelope("heart.card", {
                    "system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "level": "basic",
                    "bindings": ["http-server"], "endpoints": {"api": f"http://127.0.0.1:{port}/heart/v1"},
                    "pulse": {"intervalSec": PULSE_SEC}, "modules": ["properties"],
                    "shares": {"pulse": "public", "arrhythmia": "public", "property": "private"},
                    "autonomy": {"level": 1}, "template": TEMPLATE}))
            if path == "/heart/v1/pulse":
                return self.send(200, heart.pulse(owner))
            if path == "/heart/v1/arrhythmias":
                if owner:
                    with heart.lock:
                        return self.send(200, {"items": [heart.envelope("heart.arrhythmia", dict(a), "internal")
                                                         for a in reversed(heart.alerts)]}, "application/json")
                return self.send(200, {"items": heart.public_alerts(100)}, "application/json")
            if path == "/heart/v1/properties":
                return self.send(200, {"items": heart.properties(owner)}, "application/json")
            if path == "/heart/v1/sessions":
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Charging sessions are private: send the owner key")
                return self.send(200, {"items": heart.session_log()}, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties or /sessions")

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "This HEART only watches. Charging current, start and stop and every protection stay with the "
                         "charger's certified controller and its operator's backend, never with HEART.")

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

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


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

INSTANCE = "sim"
SCENARIOS = {"meter", "hot", "grid", "link", "rcd", "contactor", "weld"}


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated home EV charger 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="ev-charger-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43306, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help="meter,hot,grid,link,rcd,contactor,weld or all")
    ap.add_argument("--speed", type=float, default=1.0, help="simulated seconds per real second (default 1)")
    ap.add_argument("--seed", type=int, default=None, help="random seed, to repeat a run")
    ap.add_argument("--duration", type=float, default=0, help="stop after this many real seconds (0: run until Ctrl-C)")
    args = ap.parse_args()
    scenarios = set(SCENARIOS) if args.scenario == "all" else {s for s in args.scenario.split(",") if s}
    unknown = scenarios - SCENARIOS
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")

    rng = random.Random(args.seed)
    world, charger = World(rng, scenarios), Charger(rng, scenarios)
    heart = Heart(charger, world)
    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}
  NOT A REAL CHARGER. Simulated cars, grid and charger. HEART only watches: it never changes charging.
  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, charger, args.monitor_url, args.state_file, stop), daemon=True).start()

    t, wall0, last_print = 0.0, time.time(), 0.0
    try:
        while not stop.is_set():
            dt = args.speed
            t += dt
            world.step(t, wall0 + t)
            charger.step(t, dt, world)
            heart.tend(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                car = f"car {100 * world.car.soc:3.0f}%" if world.car else "no car  "
                log(f"{car} · {charger.state:14} {charger.current:4.1f} A {charger.power_kw:4.2f} kW · "
                    f"grid {charger.voltage:3.0f} V · connector {charger.temp:4.1f} °C · "
                    f"RCD {'TRIPPED' if charger.rcd_tripped else 'ok'} · contactor {charger.contactor_health} · "
                    f"link {'down' if charger.link_down else 'up'} · HEART {heart.status}")
            if args.duration and time.time() - wall0 >= args.duration:
                break
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    stop.set()
    log("stopped. The monitor site will show a Flatline if this device stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo