SOFI Ticker Heartbeat Implementation Plan¶

"""SOFI Ticker heartbeat (§41b) — structural tests.

The bulb (bulb_2) plays an event-driven lub-dub on sharp 5-min moves, layered on
the steady tint. Renderer logic lives in script.sofi_ticker_render; the beat in
script.sofi_heartbeat. §41 keeps the /5 resting tick; §41b owns the sensor trigger.
Bulb is Matter, snap-and-hold: transition: is a no-op, ~3 cmd/s ceiling.
"""
import json

from conftest import load_yaml

BULB = "light.louis_office_sengled_a19_bulb_2"


def _by_id(automations):
    return {a.get("id"): a for a in automations}


def _dump(obj):
    return json.dumps(obj)


def _iter_actions(node):
    """Yield every action dict, descending into sequence/repeat/choose/if/then/else/default."""
    if isinstance(node, dict):
        if "action" in node:
            yield node
        rpt = node.get("repeat")
        if isinstance(rpt, dict):
            yield from _iter_actions(rpt.get("sequence"))
        for key in ("sequence", "then", "else", "default", "actions"):
            yield from _iter_actions(node.get(key))
        for branch in node.get("choose", []) or []:
            yield from _iter_actions(branch.get("sequence"))
    elif isinstance(node, list):
        for item in node:
            yield from _iter_actions(item)


# ---- Task 1: helpers ----
def test_mutex_helper_defined(configuration):
    assert "sofi_heartbeat_active" in (configuration.get("input_boolean") or {})


def test_threshold_helper_defined(configuration):
    num = (configuration.get("input_number") or {}).get("sofi_heartbeat_threshold")
    assert num, "input_number.sofi_heartbeat_threshold must exist"
    assert "initial" not in num, "no initial: — restore_state persistence (repo convention)"
    assert num.get("min", 99) <= 0.5 <= num.get("max", 0)


def test_new_entities_allowlisted(entity_allowlist):
    for e in (
        "input_boolean.sofi_heartbeat_active",
        "input_number.sofi_heartbeat_threshold",
        "script.sofi_ticker_render",
        "script.sofi_heartbeat",
    ):
        assert e in entity_allowlist, f"{e} must be in entities_allow.txt until post-deploy snapshot"

  # Mutex for the SOFI Ticker heartbeat (§41b). ON while a lub-dub burst is playing so
  # §41's /5 resting-tint tick stands down and doesn't stomp the beat mid-flight. Cleared
  # by script.sofi_heartbeat before it settles the bulb. Not dashboard-facing.
  sofi_heartbeat_active:
    name: "SOFI Heartbeat Active"
    icon: mdi:heart-pulse

  # Trip point (abs % move over one 5-min quote poll) for the SOFI Ticker heartbeat (§41b).
  # |delta| >= this → a lub-dub burst in the move's direction. Dashboard-tunable; no `initial:`
  # so your value persists across restarts (restore_state) — starts at the 0.1 min on first
  # boot, set to 0.5 (the design default) once. Bigger moves add pairs (>=1.0 → 3, >=1.5 → 4).
  sofi_heartbeat_threshold:
    name: "SOFI Heartbeat Threshold"
    icon: mdi:heart-pulse
    min: 0.1
    max: 5
    step: 0.1
    unit_of_measurement: "%"
    mode: box

# SOFI Ticker heartbeat (§41b) — helpers + scripts created on the next HA restart (they don't
# exist in the registry until the config deploys). Fold into entities.txt via `make snapshot`
# post-deploy, then remove these four lines.
# Spec: docs/superpowers/specs/2026-07-22-sofi-heartbeat-design.md
input_boolean.sofi_heartbeat_active
input_number.sofi_heartbeat_threshold
script.sofi_ticker_render
script.sofi_heartbeat

git add configuration.yaml tests/test_sofi_heartbeat.py tests/fixtures/entities_allow.txt
git commit -m "feat(sofi): add heartbeat mutex + threshold helpers"

# ---- Task 2: render script ----
def test_render_script_exists(scripts):
    assert "sofi_ticker_render" in scripts


def test_render_script_tints_and_offs_bulb(scripts):
    body = _dump(scripts["sofi_ticker_render"])
    assert BULB in body, "render must target the SOFI bulb"
    assert "light.turn_on" in body and "light.turn_off" in body, \
        "render must both tint (market hours) and turn off (outside hours)"
    assert "sensor.sofi_change_pct" in body, "render tints by the change-pct sensor"

# SOFI Ticker — resting appearance (single source). Tints bulb_2 by SOFI's % change vs
# previous close during market hours (green up / red down, deeper+brighter for bigger moves,
# pale when flat); off outside 9:30-16:00 ET weekdays. The Sengled is Matter and occasionally
# times out ("Peer no longer responding"), so each write retries up to 3× until it reports on
# within ~8% of target. Called by automations.yaml §41 (/5 tick, start, presence-return) and by
# script.sofi_heartbeat (to settle after a burst). transition: is a no-op on this bulb.
sofi_ticker_render:
  alias: "SOFI Ticker Render"
  mode: restart
  sequence:
    - choose:
        - conditions:
            - condition: time
              after: "09:30:00"
              before: "16:00:00"
              weekday: [mon, tue, wed, thu, fri]
            - condition: template
              value_template: "{{ has_value('sensor.sofi_change_pct') }}"
          sequence:
            - variables:
                pct: "{{ states('sensor.sofi_change_pct') | float(0) }}"
                target_hs: >-
                  {{ [40, 12] if (pct | abs) < 0.1 else [ (120 if pct >= 0 else 0), 100 ] }}
                target_bri_pct: >-
                  {{ 8 if (pct | abs) < 0.1 else (10 + ([pct | abs, 4] | min) / 4 * 55) | round(0) | int }}
                target_bri_255: "{{ ((target_bri_pct | float) * 255 / 100) | round(0) | int }}"
            - repeat:
                count: 3
                sequence:
                  - action: light.turn_on
                    target:
                      entity_id: light.louis_office_sengled_a19_bulb_2
                    data:
                      hs_color: "{{ target_hs }}"
                      brightness_pct: "{{ target_bri_pct }}"
                    continue_on_error: true
                  - if:
                      - condition: template
                        value_template: >-
                          {{ is_state('light.louis_office_sengled_a19_bulb_2','on')
                             and (((state_attr('light.louis_office_sengled_a19_bulb_2','brightness') | int(0))
                                   - target_bri_255) | abs) < 20 }}
                    then:
                      - stop: "SOFI bulb applied"
                  - delay: "00:00:01"
      default:
        - repeat:
            count: 3
            sequence:
              - action: light.turn_off
                target:
                  entity_id: light.louis_office_sengled_a19_bulb_2
                continue_on_error: true
              - if:
                  - condition: state
                    entity_id: light.louis_office_sengled_a19_bulb_2
                    state: "off"
                then:
                  - stop: "SOFI bulb off"
              - delay: "00:00:01"

git add scripts.yaml tests/test_sofi_heartbeat.py
git commit -m "feat(sofi): extract §41 tint logic into script.sofi_ticker_render"

# ---- Task 3: heartbeat script ----
def test_heartbeat_script_exists_with_fields(scripts):
    hb = scripts.get("sofi_heartbeat")
    assert hb, "script.sofi_heartbeat must exist"
    assert set(hb.get("fields", {})) >= {"pairs", "hue"}, "needs pairs + hue fields"


def test_heartbeat_uses_snaps_not_transition(scripts):
    """Every light write is a transition:0 snap (Matter bulb ignores transition duration)."""
    hb = scripts["sofi_heartbeat"]
    body = _dump(hb)
    assert BULB in body
    writes = [n for n in _iter_actions(hb["sequence"]) if n.get("action") == "light.turn_on"]
    assert writes, "heartbeat must write the bulb"
    assert all(w.get("data", {}).get("transition") == 0 for w in writes), \
        "every beat write must be transition:0 (transition duration is a no-op on this bulb)"


def test_heartbeat_sets_then_clears_mutex_before_settle(scripts):
    """Mutex ON, beats, mutex OFF, THEN settle render — clear must precede the render
    so the settling render isn't blocked by the very mutex it's recovering from."""
    seq = scripts["sofi_heartbeat"]["sequence"]
    flat = _dump(seq)
    assert "sofi_heartbeat_active" in flat and "sofi_ticker_render" in flat
    on_i = flat.index("input_boolean.turn_on")
    off_i = flat.rindex("input_boolean.turn_off")
    render_i = flat.index("sofi_ticker_render")
    assert on_i < off_i < render_i, "order must be: mutex on → ... → mutex off → render"

# SOFI Ticker — lub-dub heartbeat on a sharp move (§41b calls this). Plays `pairs` two-pulse
# beats in `hue` (0 red = down / 120 green = up), then settles the bulb via sofi_ticker_render.
# The bulb is Matter snap-and-hold: transition: is a no-op, so every level is a transition:0
# snap; ~275ms Matter latency provides most inter-snap spacing. Mutex sofi_heartbeat_active
# gates §41's resting tick out while a burst plays; cleared BEFORE the settle render so the
# render isn't blocked by it. mode: restart — a newer, bigger move preempts an in-flight burst.
sofi_heartbeat:
  alias: "SOFI Heartbeat"
  mode: restart
  fields:
    pairs:
      description: "Number of lub-dub pairs (2/3/4 by move size)"
      example: 2
    hue:
      description: "Beat hue: 0 = red (down), 120 = green (up)"
      example: 120
  sequence:
    - action: input_boolean.turn_on
      target:
        entity_id: input_boolean.sofi_heartbeat_active
    - repeat:
        count: "{{ pairs | int(2) }}"
        sequence:
          # lub
          - action: light.turn_on
            target: {entity_id: light.louis_office_sengled_a19_bulb_2}
            data: {hs_color: "{{ [hue | int, 100] }}", brightness_pct: 100, transition: 0}
          - action: light.turn_on
            target: {entity_id: light.louis_office_sengled_a19_bulb_2}
            data: {hs_color: "{{ [hue | int, 100] }}", brightness_pct: 30, transition: 0}
          # dub
          - action: light.turn_on
            target: {entity_id: light.louis_office_sengled_a19_bulb_2}
            data: {hs_color: "{{ [hue | int, 100] }}", brightness_pct: 85, transition: 0}
          - action: light.turn_on
            target: {entity_id: light.louis_office_sengled_a19_bulb_2}
            data: {hs_color: "{{ [hue | int, 100] }}", brightness_pct: 20, transition: 0}
          - action: light.turn_on
            target: {entity_id: light.louis_office_sengled_a19_bulb_2}
            data: {hs_color: "{{ [hue | int, 100] }}", brightness_pct: 8, transition: 0}
          - delay: {milliseconds: 700}
    - action: input_boolean.turn_off
      target:
        entity_id: input_boolean.sofi_heartbeat_active
    - action: script.sofi_ticker_render

git add scripts.yaml tests/test_sofi_heartbeat.py
git commit -m "feat(sofi): add script.sofi_heartbeat lub-dub burst"

# ---- Task 4: §41 trimmed ----
def test_s41_no_sensor_trigger(automations):
    a = _by_id(automations)["sofi_ticker_color"]
    trig_entities = _dump(a["triggers"])
    assert "sensor.sofi_change_pct" not in trig_entities, \
        "§41 must NOT trigger on the price sensor — §41b owns that"
    # still has the /5 resting tick
    assert any(t.get("trigger") == "time_pattern" for t in a["triggers"])


def test_s41_has_mutex_condition(automations):
    a = _by_id(automations)["sofi_ticker_color"]
    conds = _dump(a["conditions"])
    assert "sofi_heartbeat_active" in conds and "off" in conds, \
        "§41 render tick must stand down while a burst plays"
    assert "office_occupied" in conds, "keep the presence gate"


def test_s41_calls_render_script(automations):
    a = _by_id(automations)["sofi_ticker_color"]
    assert "sofi_ticker_render" in _dump(a["actions"]), "§41 delegates to the render script"

- alias: "SOFI Ticker — Stock Color"
  id: sofi_ticker_color
  description: >
    Keeps the SOFI Ticker bulb's RESTING tint fresh (green up / red down vs previous close,
    deeper+brighter for bigger moves, off outside market hours) by calling script.sofi_ticker_render.
    Runs on a /5 tick, HA start, and office return. Sharp intraday MOVES are handled separately by
    §41b (heartbeat) — this automation no longer triggers on the price sensor. Stands down while a
    heartbeat burst is playing (sofi_heartbeat_active) so the tick can't stomp the beat mid-flight.
  triggers:
    - trigger: time_pattern
      minutes: "/5"
    - trigger: homeassistant
      event: start
    - trigger: state
      entity_id: input_boolean.office_occupied
      to: "on"
  conditions:
    - condition: state
      entity_id: input_boolean.office_occupied
      state: "on"
    - condition: state
      entity_id: input_boolean.sofi_heartbeat_active
      state: "off"
  actions:
    - action: script.sofi_ticker_render
  mode: single

git add automations.yaml tests/test_sofi_heartbeat.py
git commit -m "refactor(sofi): §41 delegates resting tint to render script, drops sensor trigger"

# ---- Task 5: §41b heartbeat automation ----
def test_s41b_exists_and_triggers_on_sensor(automations):
    a = _by_id(automations).get("sofi_ticker_heartbeat")
    assert a, "§41b (sofi_ticker_heartbeat) must exist"
    assert "sensor.sofi_change_pct" in _dump(a["triggers"])


def test_s41b_presence_gated(automations):
    a = _by_id(automations)["sofi_ticker_heartbeat"]
    assert "office_occupied" in _dump(a["conditions"])


def test_s41b_has_freshness_guard(automations):
    """The 09:30 open gap / restart guard: from_state must be recent (600s)."""
    a = _by_id(automations)["sofi_ticker_heartbeat"]
    body = _dump(a["actions"])
    assert "last_changed" in body and "600" in body, \
        "must guard against comparing across a stale previous reading"


def test_s41b_beat_mapping_and_dispatch(automations):
    a = _by_id(automations)["sofi_ticker_heartbeat"]
    body = _dump(a["actions"])
    for tier in ("1.5", "1.0", "sofi_heartbeat_threshold"):
        assert tier in body, f"beat mapping must reference {tier}"
    assert "script.sofi_heartbeat" in body, "big move → heartbeat"
    assert "script.sofi_ticker_render" in body, "small move → render only"
    assert "unknown" in body and "unavailable" in body, "must skip non-numeric states"

# 41b. SOFI Ticker — Heartbeat on Sharp Move (spec 2026-07-22). When a 5-min quote poll moves
# the price >= the threshold, play a lub-dub burst in the MOVE's direction (green up / red down),
# then settle to the resting tint. Beat count scales with size. Guards the 09:30 open gap / HA
# restart by requiring the previous reading to be recent (< 600s). Delta comes straight from the
# state trigger — no helper needed to remember the last value. Sub-threshold changes just re-render
# (this automation now owns the sensor-change render; §41 keeps only the /5 tick).
- alias: "SOFI Ticker — Heartbeat on Sharp Move"
  id: sofi_ticker_heartbeat
  triggers:
    - trigger: state
      entity_id: sensor.sofi_change_pct
  conditions:
    - condition: state
      entity_id: input_boolean.office_occupied
      state: "on"
  actions:
    - variables:
        from_s: "{{ trigger.from_state.state if trigger.from_state else 'unknown' }}"
        to_s: "{{ trigger.to_state.state if trigger.to_state else 'unknown' }}"
        # Freshness: previous reading changed within 10 min. Kills the 09:30 open-gap false beat
        # (first poll vs yesterday's close) and the same bug after an HA restart / stalled sensor.
        fresh: >-
          {{ trigger.from_state is not none
             and (now() - trigger.from_state.last_changed).total_seconds() < 600 }}
        valid: >-
          {{ fresh
             and from_s not in ['unknown', 'unavailable']
             and to_s not in ['unknown', 'unavailable'] }}
        delta: "{{ ((to_s | float(0)) - (from_s | float(0))) if valid else 0 }}"
        adelta: "{{ delta | abs }}"
        threshold: "{{ states('input_number.sofi_heartbeat_threshold') | float(0.5) }}"
        pairs: >-
          {{ 4 if adelta >= 1.5 else 3 if adelta >= 1.0 else 2 if adelta >= threshold else 0 }}
        hue: "{{ 120 if delta >= 0 else 0 }}"
    - choose:
        - conditions:
            - condition: template
              value_template: "{{ (pairs | int) > 0 }}"
          sequence:
            - action: script.sofi_heartbeat
              data:
                pairs: "{{ pairs | int }}"
                hue: "{{ hue | int }}"
      default:
        - action: script.sofi_ticker_render
  mode: single

git add automations.yaml tests/test_sofi_heartbeat.py
git commit -m "feat(sofi): §41b heartbeat lub-dub on sharp intraday moves"

git add scripts.yaml automations.yaml
git commit -m "docs(sofi): correct bulb transport Zigbee→Matter in code comments"

 **Heartbeat on sharp moves (#sofi-heartbeat, 2026-07-22):** on a 5-min quote poll where |Δ%| ≥ `input_number.sofi_heartbeat_threshold` (default 0.5, dashboard-tunable), the bulb plays a **lub-dub burst in the MOVE's direction** (green up / red down — independent of the day's resting tint), 2/3/4 pairs for ≥0.5/1.0/1.5%, then settles. The resting tint moved out of §41 into **`script.sofi_ticker_render`** (single source); §41 now only keeps it fresh (/5 tick, start, office-return) and stands down via `input_boolean.sofi_heartbeat_active` while a burst plays; the sensor-change trigger moved to **§41b** (`automation.sofi_ticker_heartbeat`), which computes Δ from the state trigger (freshness-guarded < 600s to kill the 09:30 open-gap false beat) and calls **`script.sofi_heartbeat(pairs, hue)`**. **Hardware facts (measured):** the bulb is **Matter** (not Zigbee — old comment was wrong), `transition:` is a **no-op** (fixed ~1.4%/s ramp, so every beat is a `transition: 0` snap), and it tops out at **~3 commands/sec**. Design: `docs/superpowers/specs/2026-07-22-sofi-heartbeat-design.md`.

git add CLAUDE.md docs/automations.md docs/entities.md
git commit -m "docs(sofi): document heartbeat (§41b) in manual, entities, CLAUDE.md"

# Confirm HA came back up
ssh haos "ha core logs --lines 20"
# Set the threshold to the design default 0.5 once (first boot starts at the 0.1 min):
#   Dashboard, or: input_number.set_value entity=input_number.sofi_heartbeat_threshold value=0.5
# Fold the 4 new entities into the registry snapshot, then remove them from the allowlist:
make snapshot
# Edit tests/fixtures/entities_allow.txt — delete the 4 SOFI heartbeat lines
git add tests/fixtures/entities.txt tests/fixtures/entities_allow.txt
git commit -m "chore(sofi): snapshot heartbeat entities, drop from allowlist"