Feat/beat envelope transform

BITACORA.md

@@ -7,3 +7,4 @@
 - 2026-07-19: Event-demo and sparse scenes now route HarMoCAP gains through Shaper `harmonic_envelope` rather than the passive `harmonic_gain` parameter. The safety profile resets those envelopes to zero. The rehearsal assertion now follows the declared capability rather than a historical capability name.
 - 2026-07-19: The full hardware-free rehearsal passed as `t45-20260719T092417Z`: focused HarMoCAP partials `1..5`, scene hot-swap, panic release/rearm, and non-silent finite audio all passed. The live-stack launcher accepts `--harmocap-checkpoint <path>` for an explicit local model override without altering HarMoCAP's promoted model configuration.
 - 2026-07-19: Live camera → HarMoCAP → Weaver → Shaper/R24 control was audibly confirmed by the user using explicit `yolo26m-pose.pt`. The camera process later hit an Ultralytics CUDA/ReID illegal-memory-access failure; it is recorded as a HarMoCAP GPU stability issue, not as a successful soak run.
+- 2026-07-20: add `beat_envelope` transform (rising-edge trigger → decaying gain envelope, tau auto-scaled from inter-beat interval) for the Latido heartbeat pulse — turns the ECG beat into a breathing pulse instead of a one-frame gate flash. Compiler validation + range propagation `[floor, peak]` + stateful runtime; 8 tests; docs. Full suite 65 passed. Branch `feat/beat-envelope-transform`.

docs/CORE_DESIGN.md

@@ -99,9 +99,10 @@ Transform execution is ordered and uses finite numbers only:
 | `phase_accumulator` | Integrate an angular velocity (deg/s) into a running phase wrapped to `[0, wrap_deg)` (default `360`). Optional `max_rate` clamps `|velocity|`; `max_dt_ms` (default `100`) clamps the per-evaluation step so a gap on resume cannot jump. Stateful. |
 | `slew_limiter` | Chase a continuous target at bounded rate (`max_rate` units/s). `max_dt_ms` clamps the integration step so a network gap cannot jump the parameter. First sample snaps to target (cold start). Output static range equals the incoming range. Stateful. |
 | `derivative` | Causal trailing difference of a position/feature → signed velocity: `(x[t]-x[t-1])/dt`. `window_ms > 0` (declared min Δt), `max_abs > 0` clamps `|d/dt|` and defines static range `[-max_abs, +max_abs]`, `max_dt_ms >= 0` clamps `dt`. First sample emits `0`. Stateful. |
+| `beat_envelope` | Rising-edge trigger → decaying gain envelope: on each edge (input crossing `threshold`, default `0.5`) the output snaps to `peak` (default `1`) and relaxes toward `floor` (default `0`). The time constant is `tau_ms`, or auto-scaled from the measured inter-beat interval by `tau_ratio` (default `0.3`); `min_interval_ms` (default `250`) is a refractory guard. Output bounded to `[floor, peak]`. Stateful. |
 
 `scale_range`, `curve`, `combine` and `gate` are memoryless; `smoothing`,
-`phase_accumulator`, `slew_limiter` and `derivative` are **stateful** — they hold
+`phase_accumulator`, `slew_limiter`, `derivative` and `beat_envelope` are **stateful** — they hold
 per-route, per-position state in `RouteRuntime` and derive their time step from
 the engine's `now_us` deltas (the same monotonic clock used everywhere, so
 replay/resume stays deterministic — never wall-clock). See

src/harmonic_weaver/engine/compiler.py

@@ -193,6 +193,7 @@ class RouteRuntime:
     slew_at_us: dict[int, int] = field(default_factory=dict)
     derivative_values: dict[int, float] = field(default_factory=dict)
     derivative_at_us: dict[int, int] = field(default_factory=dict)
+    beat_state: dict[int, dict] = field(default_factory=dict)
     last_usable_output: float | None = None
     last_usable_at_us: int | None = None
     invalid_reset_sent: bool = False
@@ -421,6 +422,7 @@ def compile_route(
             "phase_accumulator",
             "slew_limiter",
             "derivative",
+            "beat_envelope",
         }:
             raise validation(f"{tpath}.type is invalid")
         if kind == "combine":
@@ -484,6 +486,23 @@ def compile_route(
             max_abs = positive(transform.get("max_abs"), f"{tpath}.max_abs")
             nonnegative(transform.get("max_dt_ms"), f"{tpath}.max_dt_ms")
             current_range = (-max_abs, max_abs)
+        elif kind == "beat_envelope":
+            # Rising-edge trigger -> decaying gain envelope: on each edge the
+            # output snaps to `peak` and relaxes toward `floor` with a time
+            # constant (auto-scaled from the measured inter-beat interval, or a
+            # fixed `tau_ms`). Output is bounded to [floor, peak].
+            peak = finite(transform.get("peak", 1.0), f"{tpath}.peak")
+            floor_value = finite(transform.get("floor", 0.0), f"{tpath}.floor")
+            if floor_value >= peak:
+                raise validation(f"{tpath}.floor must be less than {tpath}.peak")
+            finite(transform.get("threshold", 0.5), f"{tpath}.threshold")
+            if "tau_ratio" in transform:
+                positive(transform["tau_ratio"], f"{tpath}.tau_ratio")
+            if "tau_ms" in transform:
+                positive(transform["tau_ms"], f"{tpath}.tau_ms")
+            if "min_interval_ms" in transform:
+                nonnegative(transform["min_interval_ms"], f"{tpath}.min_interval_ms")
+            current_range = (floor_value, peak)
     validity_policy = validate_validity(raw["validity"], f"{path}.validity")
     definition = copy.deepcopy(dict(raw))
     definition["validity"] = validity_policy
@@ -667,6 +686,40 @@ def evaluate_route(
             runtime.derivative_values[transform_index] = current
             runtime.derivative_at_us[transform_index] = now_us
             current = out
+        elif kind == "beat_envelope":
+            assert isinstance(current, float)
+            peak = float(transform.get("peak", 1.0))
+            floor_value = float(transform.get("floor", 0.0))
+            threshold = float(transform.get("threshold", 0.5))
+            tau_ratio = float(transform.get("tau_ratio", 0.3))
+            fixed_tau_ms = transform.get("tau_ms")
+            min_interval_us = int(float(transform.get("min_interval_ms", 250.0)) * 1000)
+            state = runtime.beat_state.get(transform_index)
+            if state is None:
+                state = {
+                    "value": floor_value, "beat_us": None, "eval_us": None,
+                    "last_in": 0.0,
+                    "tau_ms": float(fixed_tau_ms) if fixed_tau_ms is not None else 250.0,
+                }
+            # Decay toward floor over the time since the last evaluation.
+            if state["eval_us"] is not None:
+                dt_s = max(0.0, (now_us - state["eval_us"]) / 1_000_000.0)
+                tau_s = max(1e-3, state["tau_ms"] / 1000.0)
+                state["value"] = floor_value + (state["value"] - floor_value) * math.exp(-dt_s / tau_s)
+            # Rising edge -> fire the pulse (with a refractory guard).
+            fired = state["last_in"] < threshold <= current
+            if fired and (state["beat_us"] is None or now_us - state["beat_us"] >= min_interval_us):
+                if fixed_tau_ms is not None:
+                    state["tau_ms"] = float(fixed_tau_ms)
+                elif state["beat_us"] is not None:
+                    interval_ms = (now_us - state["beat_us"]) / 1000.0
+                    state["tau_ms"] = max(1.0, tau_ratio * interval_ms)
+                state["value"] = peak
+                state["beat_us"] = now_us
+            state["last_in"] = current
+            state["eval_us"] = now_us
+            runtime.beat_state[transform_index] = state
+            current = state["value"]
     if isinstance(current, list) or not math.isfinite(current):
         return None, "suppress"
     runtime.last_usable_output = current

tests/test_beat_envelope.py

@@ -0,0 +1,121 @@
+"""beat_envelope transform: rising-edge trigger -> decaying gain envelope.
+
+On each rising edge the output snaps to `peak` and relaxes toward `floor` with
+a time constant (fixed `tau_ms`, or auto-scaled from the measured inter-beat
+interval by `tau_ratio`). Turns the ECG beat trigger into a pulse that breathes
+instead of a one-frame flash.
+"""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from harmonic_weaver.engine.compiler import (
+    RouteRuntime,
+    compile_route,
+    destination_key,
+    evaluate_route,
+)
+from harmonic_weaver.engine.model import OBSERVED, ValueEnvelope
+
+from engine_fixtures import instrument_manifest
+
+INSTRUMENT = instrument_manifest()
+MANIFESTS = {"synth": INSTRUMENT}
+CHANNELS = {"sensor.beat": (0.0, 1.0)}
+TOL = 1e-6
+
+
+def _destination() -> dict:
+    return {"instrument_id": "synth", "capability": "voice_gain",
+            "bindings": {"N": 0}, "argument": "gain"}
+
+
+def _compile(transform: dict):
+    route = {
+        "route_id": "beat-to-gain",
+        "route_version": 1,
+        "enabled": True,
+        "inputs": [{"channel": "sensor.beat"}],
+        "transforms": [transform],
+        "destination": _destination(),
+        "validity": {"held": "accept", "min_confidence": 0.0, "invalid": "suppress"},
+    }
+    return compile_route(route, CHANNELS, MANIFESTS, {destination_key(_destination()): 0.0},
+                         "scene.routes[0]")
+
+
+def _beat(value: float, now_us: int) -> dict:
+    return {"sensor.beat": ValueEnvelope(value, OBSERVED, 1.0, now_us, now_us)}
+
+
+def test_compiles_with_floor_peak_range():
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.2})
+    assert compiled.static_range == (0.2, 1.0)
+
+
+def test_rejects_floor_ge_peak():
+    with pytest.raises(Exception):
+        _compile({"type": "beat_envelope", "peak": 0.5, "floor": 0.5})
+
+
+def test_rising_edge_fires_to_peak():
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.2, "tau_ms": 100.0})
+    rt = RouteRuntime()
+    v0, _ = evaluate_route(compiled, rt, _beat(0.0, 0), 0)          # rest at floor
+    assert abs(v0 - 0.2) < TOL
+    v1, _ = evaluate_route(compiled, rt, _beat(1.0, 10_000), 10_000)  # edge -> peak
+    assert abs(v1 - 1.0) < TOL
+
+
+def test_decays_toward_floor_by_tau():
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.2, "tau_ms": 100.0})
+    rt = RouteRuntime()
+    evaluate_route(compiled, rt, _beat(1.0, 0), 0)                  # fire at t=0 -> 1.0
+    v, _ = evaluate_route(compiled, rt, _beat(0.0, 100_000), 100_000)  # +1 tau (0.1 s)
+    expected = 0.2 + (1.0 - 0.2) * math.exp(-1.0)                   # ~0.494
+    assert abs(v - expected) < 1e-4
+
+
+def test_no_refire_while_held_high():
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.0, "tau_ms": 50.0})
+    rt = RouteRuntime()
+    evaluate_route(compiled, rt, _beat(1.0, 0), 0)                  # fire
+    a, _ = evaluate_route(compiled, rt, _beat(1.0, 50_000), 50_000)  # still high -> no refire, decays
+    b, _ = evaluate_route(compiled, rt, _beat(1.0, 100_000), 100_000)
+    assert a < 1.0 and b < a                                        # monotonic decay, never re-peaks
+
+
+def test_refractory_ignores_too_close_beats():
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.0,
+                         "tau_ms": 100.0, "min_interval_ms": 300.0})
+    rt = RouteRuntime()
+    evaluate_route(compiled, rt, _beat(1.0, 0), 0)                  # beat 1 -> peak
+    evaluate_route(compiled, rt, _beat(0.0, 50_000), 50_000)        # drop low
+    v, _ = evaluate_route(compiled, rt, _beat(1.0, 100_000), 100_000)  # beat 2 at +100ms < 300ms
+    assert v < 1.0                                                  # refused -> not re-peaked
+
+
+def test_tau_autoscales_from_interval():
+    # No fixed tau_ms: tau = tau_ratio * measured interval. Two beats 1 s apart
+    # -> tau = 0.3 * 1000 ms = 300 ms. One tau after the 2nd beat -> ~1/e above floor.
+    compiled = _compile({"type": "beat_envelope", "peak": 1.0, "floor": 0.0, "tau_ratio": 0.3})
+    rt = RouteRuntime()
+    evaluate_route(compiled, rt, _beat(1.0, 0), 0)                  # beat 1
+    evaluate_route(compiled, rt, _beat(0.0, 500_000), 500_000)
+    evaluate_route(compiled, rt, _beat(1.0, 1_000_000), 1_000_000)  # beat 2, interval 1 s -> tau 300 ms
+    v, _ = evaluate_route(compiled, rt, _beat(0.0, 1_300_000), 1_300_000)  # +300 ms = 1 tau
+    assert abs(v - math.exp(-1.0)) < 1e-3                           # peak 1, floor 0 -> ~0.368
+
+
+def test_output_stays_within_floor_peak():
+    compiled = _compile({"type": "beat_envelope", "peak": 0.9, "floor": 0.3, "tau_ms": 40.0})
+    rt = RouteRuntime()
+    vals = []
+    seq = [(1.0, 0), (0.0, 20_000), (0.0, 60_000), (1.0, 200_000), (0.0, 260_000)]
+    for x, t in seq:
+        v, _ = evaluate_route(compiled, rt, _beat(x, t), t)
+        vals.append(v)
+    assert all(0.3 - TOL <= v <= 0.9 + TOL for v in vals)