From 9d0e9e19ef8c5d776c46f5dc0d96164115c8f32a Mon Sep 17 00:00:00 2001 From: "dispatch-bot[bot]" <3106185+dispatch-bot[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 08:13:44 -0400 Subject: [PATCH 1/9] docs: design spec for ZAP lighting controls (issue #12) Co-Authored-By: Claude Fable 5 --- .../2026-07-13-lighting-controls-design.md | 163 ++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-13-lighting-controls-design.md diff --git a/docs/superpowers/specs/2026-07-13-lighting-controls-design.md b/docs/superpowers/specs/2026-07-13-lighting-controls-design.md new file mode 100644 index 0000000..4eeeee2 --- /dev/null +++ b/docs/superpowers/specs/2026-07-13-lighting-controls-design.md @@ -0,0 +1,163 @@ +# Design: ZAP Lighting Controls for the RF Server and Remote PWA + +**Issue:** [#12](https://github.com/tclancy/radiofrequency/issues/12) (research: [#10](https://github.com/tclancy/radiofrequency/issues/10)) +**Date:** 2026-07-13 +**Status:** Approved by Tom (conversation, 2026-07-13) + +## Goal + +Control the four Etekcity ZAP outlet-switched lamps in the living room (remote +positions 2–5) from the existing NodeMCU RF server and the 22-parsons-remote +PWA, the same way the ceiling fans work today. + +| Position | Lamp | +|----------|---------| +| 2 | window | +| 3 | couch | +| 4 | speaker | +| 5 | chairs | + +UI gets individual on/off per lamp plus **All On / All Off**. + +## Background and key decisions + +### The protocol (from issue #10) + +The ZAP 5LX remote uses an HS2260A-R4 encoder (PT2260 family) at 433.92 MHz: +OOK with 12 tri-state symbols (0/1/F) per frame — 8 address + 4 data — sync +gap after each frame, repeated while the button is held. The outlets are +learning-code receivers paired to *our* remote. + +### Decision: take the knowledge, not the library + +We will **not** vendor [cpetrescu/ZAP-remote](https://github.com/cpetrescu/ZAP-remote). +Its hardcoded codes belong to the author's remote — PT2260 addresses are baked +into each remote's encoder, and our outlets learned ours. The library and the +[vmallet ZAP 3L teardown](https://vmallet.com/2020/07/etekcity-zap-3l-remote-power-outlet-teardown-and-analysis/) +serve as protocol references only. Codes come from an RTL-SDR capture of our +actual remote. + +### Decision: generic pulse-train transmit (Option A) + +The firmware's current `POST /transmit` assumes pulse-*distance* modulation +(fixed HIGH burst, data in the LOW gap). PT2260 is pulse-*width* modulation +(data in the HIGH burst length; two pulses per symbol) — inexpressible in the +current schema. + +Rather than add a per-protocol mode flag (rejected: firmware update per future +protocol family) or hardcode ZAP routes like the legacy `/fan` endpoints +(rejected: violates the generic-firmware architecture rule), the firmware +gains one universal capability: transmit an explicit pulse train. + +``` +POST /transmit (second accepted body shape) +{"pulses": [[high_us, low_us], ...], "repeat_count": N} +``` + +All encoding intelligence stays in Python, derived from the YAML profile. +This shape can express any OOK protocol, so the firmware never needs another +protocol update. The legacy `{bits, timing}` shape and `/fan/*` GET routes +remain untouched — fans keep working throughout. + +## Components + +### 1. Capture & decode (Tom-in-the-loop, ~15 min) + +- Record all 8 buttons (positions 2–5 × on/off) at 433.92 MHz with + `rtl_433 -A` (pulse analyzer); PT2260 is among the best-supported OOK + protocols, so no manual Audacity work is expected. +- Captures saved as `captures/zap_remote_pos{2-5}_{on,off}.*`. +- Comparing the 8 frames yields the address/data split and the measured + `short_us` / `long_us` / sync timings. Decoded results documented in + `PROTOCOL.md`. + +### 2. Device profile — `devices/zap_lights.yaml` + +```yaml +frequency_mhz: 433.92 +encoding: PT2260 # tri-state OOK PWM +timing: + short_us: # α-derived short segment + long_us: # 3α long segment + sync_gap_us: # long quiet gap ending each frame + repeat_count: 6 # remote repeats while held; 5–6 is plenty +units: + window: { position: 2, ... } # exact code fields finalized after capture + couch: { position: 3, ... } + speaker: { position: 4, ... } + chairs: { position: 5, ... } +commands: + "on": ... + "off": ... +``` + +The exact split between per-unit and per-command tri-state fields is +finalized from the capture (expected: shared remote address + button/state in +the data nibble, but the frames are the source of truth). + +### 3. Python encoder — `src/device.py` + +- New pure function: profile + unit + command → pulse train + `[[high_us, low_us], ...]` including the sync symbol. +- New payload builder for the `pulses` body shape with the same validation + spirit as `build_transmit_payload`. +- Small, composable, unit-tested (known tri-state code → known waveform). + +### 4. Firmware — `firmware/src/main.cpp` + +- `handleTransmit` accepts the new `pulses` body shape alongside the legacy + one (presence of the `pulses` key selects the path). +- Validation: µs values clamped 1..100000, max 256 pulse pairs, + `repeat_count` 1..100. Watchdog fed between repetitions, as today. +- No device knowledge added; `/fan/*` and legacy `/transmit` untouched. + +### 5. CLI + +`python cli.py send zap_lights window on` — profile loading detects the +PT2260 encoding and routes through the pulse-train encoder/payload. + +### 6. Derived web bundle — `scripts/export_web_devices.py` + +Generates `devices.json` (button → ready-to-POST payload) from +`devices/*.yaml`, so the PWA never duplicates codes or timings. One documented +regen command for now; automation hook later if the manual step annoys us +(**open item**, see below). + +### 7. homelab repo (separate PR, companion issue) + +- 22-parsons-remote PWA adds a **Lights card**: window / couch / speaker / + chairs with on/off each, plus All On / All Off. "All" fires the four codes + sequentially from the client — each transmit is sub-second, no firmware + queueing. +- `app.js` loads the generated `devices.json` and POSTs to `/api/transmit`; + one new Caddy `reverse_proxy` route for the POST path. +- Fans stay on their existing GET pattern for now. + +## Testing + +1. **Unit tests first** (failing-test-as-spec): tri-state encoder waveform, + payload validation edge cases. +2. **Bench proof**: capture the NodeMCU's own transmission with the RTL-SDR + and diff against the remote's capture — the same technique that validated + the fans (`captures/nodemcu_main_light.wav`). +3. **End-to-end**: lamps switch from the CLI first, then from the PWA on a + phone. + +## Out of scope / future + +- Extracting the remote PWA into its own app (Tom's instinct, 2026-07-13 — + "a problem for another day"). +- Migrating the fan buttons off the legacy GET endpoints onto `/transmit`. +- Home Assistant exposure of the lights. +- Cross-repo automation for regenerating `devices.json` (manual documented + command in v1). + +## Open items + +- Regen ergonomics for `devices.json` across two repos — revisit after v1. + +## Deliverables + +1. This spec, committed on `claude/12-lighting-controls`. +2. Implementation plan posted to issue #12 (radiofrequency work). +3. Companion issue in tclancy/homelab for the PWA/Caddy half, linked from #12. From 79085fcf9a467a14a8a15fff7e63c7a01bc24f07 Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 08:22:35 -0400 Subject: [PATCH 2/9] docs: apply reviewer fixes to lighting spec (WDT budget, sync placement, PWA/Caddy details) Co-Authored-By: Claude Fable 5 --- .../2026-07-13-lighting-controls-design.md | 69 ++++++++++++++----- 1 file changed, 52 insertions(+), 17 deletions(-) diff --git a/docs/superpowers/specs/2026-07-13-lighting-controls-design.md b/docs/superpowers/specs/2026-07-13-lighting-controls-design.md index 4eeeee2..96db7b4 100644 --- a/docs/superpowers/specs/2026-07-13-lighting-controls-design.md +++ b/docs/superpowers/specs/2026-07-13-lighting-controls-design.md @@ -24,9 +24,14 @@ UI gets individual on/off per lamp plus **All On / All Off**. ### The protocol (from issue #10) The ZAP 5LX remote uses an HS2260A-R4 encoder (PT2260 family) at 433.92 MHz: -OOK with 12 tri-state symbols (0/1/F) per frame — 8 address + 4 data — sync -gap after each frame, repeated while the button is held. The outlets are -learning-code receivers paired to *our* remote. +OOK, 12 symbols per frame — 8 tri-state (0/1/F) address symbols + 4 binary +data bits — followed by a sync symbol, repeated while the button is held. The +outlets are learning-code receivers paired to *our* remote. + +Note: the 5LX has 10 buttons (5 on + 5 off) against a 4-bit data nibble, so +buttons may assert multiple data lines or borrow address-area symbols. The +capture is the source of truth — the implementation must not bake in a +nibble-only assumption for button/state. ### Decision: take the knowledge, not the library @@ -63,10 +68,15 @@ remain untouched — fans keep working throughout. ### 1. Capture & decode (Tom-in-the-loop, ~15 min) -- Record all 8 buttons (positions 2–5 × on/off) at 433.92 MHz with - `rtl_433 -A` (pulse analyzer); PT2260 is among the best-supported OOK - protocols, so no manual Audacity work is expected. -- Captures saved as `captures/zap_remote_pos{2-5}_{on,off}.*`. +- Record all 8 buttons (positions 2–5 × on/off) at 433.92 MHz: + `rtl_433 -f 433.92M -A -S unknown` — the pulse analyzer prints decoded + frames live and `-S unknown` writes native `.cu8` sample files (the fan-era + WAVs needed format conversion before rtl_433 could read them; `.cu8` avoids + that). +- Fallback ladder if `-A` doesn't decode cleanly (the fan project's WBFM + mis-timing history says have one): URH → triq.org/pdv pulse visualizer → + Audacity. +- Captures saved as `captures/zap_remote_pos{2-5}_{on,off}.cu8`. - Comparing the 8 frames yields the address/data split and the measured `short_us` / `long_us` / sync timings. Decoded results documented in `PROTOCOL.md`. @@ -74,7 +84,7 @@ remain untouched — fans keep working throughout. ### 2. Device profile — `devices/zap_lights.yaml` ```yaml -frequency_mhz: 433.92 +frequency_mhz: 433.92 # documentation-only: the MX-FS-03V TX is SAW-locked encoding: PT2260 # tri-state OOK PWM timing: short_us: # α-derived short segment @@ -98,9 +108,14 @@ the data nibble, but the frames are the source of truth). ### 3. Python encoder — `src/device.py` - New pure function: profile + unit + command → pulse train - `[[high_us, low_us], ...]` including the sync symbol. -- New payload builder for the `pulses` body shape with the same validation - spirit as `build_transmit_payload`. + `[[high_us, low_us], ...]`. The sync pair is the **last** element of the + train (PT2260 transmits sync after the 12 data symbols); since every pair + ends LOW, repeats are contiguous valid codewords with the sync gap doubling + as the inter-frame gap, and the pin is left LOW. +- New payload builder for the `pulses` body shape enforcing the **same + numeric limits as the firmware** (µs values 1..100,000, ≤256 pairs, + repeat_count 1..100, total-duration budget below) so bad payloads fail + locally with a readable message instead of a NodeMCU 400. - Small, composable, unit-tested (known tri-state code → known waveform). ### 4. Firmware — `firmware/src/main.cpp` @@ -108,7 +123,13 @@ the data nibble, but the frames are the source of truth). - `handleTransmit` accepts the new `pulses` body shape alongside the legacy one (presence of the `pulses` key selects the path). - Validation: µs values clamped 1..100000, max 256 pulse pairs, - `repeat_count` 1..100. Watchdog fed between repetitions, as today. + `repeat_count` 1..100, **and a total-duration budget**: + `repeat_count × Σ(high_us + low_us)` must be ≤ 5 s. Without the budget, + worst-case limits allow a 51 s busy-wait inside a single repetition — + `delayMicroseconds()` blocks and the ESP8266 soft watchdog fires at + ~3.2 s, hard-resetting the chip mid-transmit. (A ZAP transmit is ~140 ms; + a hypothetical fan migration at 20 repeats is ~1.3 s; 5 s is generous.) +- Watchdog fed **inside the pulse-pair loop**, not just between repetitions. - No device knowledge added; `/fan/*` and legacy `/transmit` untouched. ### 5. CLI @@ -126,11 +147,22 @@ regen command for now; automation hook later if the manual step annoys us ### 7. homelab repo (separate PR, companion issue) - 22-parsons-remote PWA adds a **Lights card**: window / couch / speaker / - chairs with on/off each, plus All On / All Off. "All" fires the four codes - sequentially from the client — each transmit is sub-second, no firmware - queueing. -- `app.js` loads the generated `devices.json` and POSTs to `/api/transmit`; - one new Caddy `reverse_proxy` route for the POST path. + chairs with on/off each, plus All On / All Off. +- **"All" must await the four POSTs serially, not `Promise.all`.** During a + transmit the ESP8266 is busy-waiting — `server.handleClient()` isn't + running — so concurrent requests stall or time out against the + single-threaded server. On a mid-sequence failure: surface the error via + the existing toast; each transmit is idempotent, so retry is safe. +- `app.js` loads the generated `devices.json` and POSTs to `/api/transmit`. + New Caddy route follows the existing pattern **including the prefix + strip** (`handle /api/transmit` + `uri strip_prefix /api`). Note the + template now lives at + `ansible/roles/products/templates/22-parsons-remote-Caddyfile.j2` + (moved from `roles/docker-services/` since PR #92). +- Service worker: `sw.js` is already network-first with cache fallback and + skips `/api/*`, so updated `app.js` reaches online clients without ceremony. + Add `devices.json` to the precache `ASSETS` list and bump the `CACHE` + version so first-load-offline behavior includes it. - Fans stay on their existing GET pattern for now. ## Testing @@ -155,6 +187,9 @@ regen command for now; automation hook later if the manual step annoys us ## Open items - Regen ergonomics for `devices.json` across two repos — revisit after v1. + Cheap interim guard: CI in this repo regenerates and diffs against a + committed copy, catching YAML/JSON drift without solving cross-repo + automation. ## Deliverables From 0ce3c891d44becfb8da07ebdbeb45f3166d54588 Mon Sep 17 00:00:00 2001 From: "dispatch-bot[bot]" <3106185+dispatch-bot[bot]@users.noreply.github.com> Date: Mon, 13 Jul 2026 08:27:24 -0400 Subject: [PATCH 3/9] docs: implementation plan for ZAP lighting controls (issue #12) Co-Authored-By: Claude Fable 5 --- .../plans/2026-07-13-zap-lighting-controls.md | 778 ++++++++++++++++++ 1 file changed, 778 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-13-zap-lighting-controls.md diff --git a/docs/superpowers/plans/2026-07-13-zap-lighting-controls.md b/docs/superpowers/plans/2026-07-13-zap-lighting-controls.md new file mode 100644 index 0000000..a313e18 --- /dev/null +++ b/docs/superpowers/plans/2026-07-13-zap-lighting-controls.md @@ -0,0 +1,778 @@ +# ZAP Lighting Controls Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Control the four Etekcity ZAP outlet lamps (window, couch, speaker, chairs) from the NodeMCU RF server via a generic pulse-train transmit path. + +**Architecture:** All PT2260 protocol encoding lives in Python, derived from a YAML device profile; the firmware gains one universal capability — transmit an explicit `[[high_µs, low_µs], ...]` pulse train — and never learns another protocol. Legacy `{bits, timing}` transmit and `/fan/*` routes are untouched. + +**Tech Stack:** Python 3.13 + uv + pytest (host side), Arduino/PlatformIO ESP8266 + ArduinoJson 7 (firmware), rtl_433 + RTL-SDR (capture/verification). + +**Spec:** `docs/superpowers/specs/2026-07-13-lighting-controls-design.md` — read it before starting. + +## Global Constraints + +- Work on branch `claude/12-lighting-controls`. Commit after every task; pre-commit hooks must pass. +- Python: always `uv run pytest ...`, never bare pytest/pip. +- TDD: every Python task writes the failing test first and shows it failing. +- Firmware validation limits (spec §4, copy exactly): µs values 1..100000; 1..256 pulse pairs; `repeat_count` 1..100; total-duration budget `repeat_count × Σ(high_us + low_us) ≤ 5,000,000 µs`. Watchdog fed inside the pulse-pair loop. +- Python payload validation mirrors those numbers exactly (spec §3). +- PT2260 sync pair is the LAST element of the pulse train; every pair ends LOW. +- Tri-state code strings use only the characters `0`, `1`, `F`. +- Tasks 6–7 need Tom at the keyboard (button presses, flashing, lamp checks). Everything in Tasks 1–5 runs without hardware. + +## Deviation from the spec's YAML sketch (intentional) + +The spec sketches `units` + shared `commands` and hedges "the frames are the source of truth." This plan stores the **full 12-symbol code per unit per command** (`units..codes.on/off`) because the 5LX has 10 buttons against a 4-bit data nibble and the address/data split may not be clean. `resolve_code()` (Task 1) is the single place that knows this schema — if the capture shows a clean factoring we can refactor later without touching the encoder. + +--- + +### Task 1: PT2260 waveform encoder + code resolution + +**Files:** +- Modify: `src/device.py` +- Test: `tests/test_device.py` + +**Interfaces:** +- Consumes: `DeviceProfile` (existing dataclass in `src/device.py`). +- Produces: + - `pt2260_pulses(code: str, timing: dict) -> list[tuple[int, int]]` — timing needs keys `short_us`, `long_us`, `sync_gap_us`. + - `resolve_code(profile: DeviceProfile, unit: str, command: str) -> str` + - `DeviceProfile.load` tolerates a profile with no top-level `commands` key (PT2260 profiles keep codes under units). + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_device.py`: + +```python +# --- PT2260 pulse-train encoding --- + +from src.device import pt2260_pulses, resolve_code + +PT2260_TIMING = {"short_us": 180, "long_us": 540, "sync_gap_us": 5580, "repeat_count": 6} + + +def _pt2260_profile(): + return DeviceProfile( + frequency_mhz=433.92, + encoding="PT2260", + timing=PT2260_TIMING, + commands={}, + units={ + "window": {"position": 2, "codes": {"on": "0F1F0F0F1010", "off": "0F1F0F0F1001"}}, + }, + ) + + +def test_pt2260_symbol_waveforms(): + # '0' = 2x (short-high, long-low); '1' = 2x (long-high, short-low); + # 'F' = (short-high, long-low) then (long-high, short-low); sync pair last. + assert pt2260_pulses("01F", PT2260_TIMING) == [ + (180, 540), (180, 540), # 0 + (540, 180), (540, 180), # 1 + (180, 540), (540, 180), # F + (180, 5580), # sync + ] + + +def test_pt2260_full_frame_is_25_pairs(): + # 12 symbols x 2 pulses + 1 sync pair + assert len(pt2260_pulses("0F1F0F0F1010", PT2260_TIMING)) == 25 + + +def test_pt2260_rejects_invalid_symbol(): + with pytest.raises(ValueError, match="symbol"): + pt2260_pulses("01X", PT2260_TIMING) + + +def test_resolve_code_looks_up_unit_command(): + profile = _pt2260_profile() + assert resolve_code(profile, "window", "on") == "0F1F0F0F1010" + + +def test_resolve_code_unknown_unit_raises_keyerror(): + with pytest.raises(KeyError): + resolve_code(_pt2260_profile(), "basement", "on") + + +def test_profile_load_tolerates_missing_commands(tmp_path): + yaml_text = ( + "frequency_mhz: 433.92\n" + "encoding: PT2260\n" + "timing: {short_us: 180, long_us: 540, sync_gap_us: 5580, repeat_count: 6}\n" + "units:\n" + " window: {position: 2, codes: {'on': '0F1F0F0F1010', 'off': '0F1F0F0F1001'}}\n" + ) + p = tmp_path / "zap.yaml" + p.write_text(yaml_text) + profile = DeviceProfile.load(str(p)) + assert profile.commands == {} + assert profile.units["window"]["codes"]["off"] == "0F1F0F0F1001" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_device.py -v -k "pt2260 or resolve_code or missing_commands"` +Expected: FAIL — `ImportError: cannot import name 'pt2260_pulses'` + +- [ ] **Step 3: Implement in `src/device.py`** + +Change one line in `DeviceProfile.load`: + +```python + commands=data.get("commands", {}), +``` + +Append: + +```python +# PT2260 tri-state symbols → two (level-duration) pulse halves each. +# Sync is a short HIGH followed by the long inter-frame gap; it is +# appended LAST so repeats are contiguous valid codewords and the +# TX pin is always left LOW. +_PT2260_SYMBOLS = { + "0": (("short", "long"), ("short", "long")), + "1": (("long", "short"), ("long", "short")), + "F": (("short", "long"), ("long", "short")), +} + + +def pt2260_pulses(code: str, timing: dict) -> list[tuple[int, int]]: + """Encode a PT2260 tri-state code string as [(high_us, low_us), ...].""" + duration = {"short": timing["short_us"], "long": timing["long_us"]} + pulses: list[tuple[int, int]] = [] + for symbol in code: + if symbol not in _PT2260_SYMBOLS: + raise ValueError(f"invalid PT2260 symbol {symbol!r} (want 0/1/F)") + for high, low in _PT2260_SYMBOLS[symbol]: + pulses.append((duration[high], duration[low])) + pulses.append((timing["short_us"], timing["sync_gap_us"])) + return pulses + + +def resolve_code(profile: DeviceProfile, unit: str, command: str) -> str: + """Full tri-state code for a unit+command. + + PT2260 profiles keep whole codes per unit (no address/command split — + see the design spec); this is the only function that knows that. + """ + return profile.units[unit]["codes"][command] # KeyError on unknown names +``` + +- [ ] **Step 4: Run the full suite** + +Run: `uv run pytest tests/ -v` +Expected: all PASS (new tests plus every pre-existing fan test). + +- [ ] **Step 5: Commit** + +```bash +git add src/device.py tests/test_device.py +git commit -m "feat(device): PT2260 tri-state pulse-train encoder" +``` + +--- + +### Task 2: Pulse payload builder with firmware-mirrored validation + +**Files:** +- Modify: `src/device.py` +- Test: `tests/test_device.py` + +**Interfaces:** +- Consumes: pulse lists from `pt2260_pulses` (Task 1). +- Produces: `build_pulses_payload(pulses: list[tuple[int, int]], repeat_count: int) -> dict` returning `{"pulses": [[h, l], ...], "repeat_count": n}`; module constants `MAX_PULSE_PAIRS = 256`, `MAX_PULSE_US = 100_000`, `MAX_TOTAL_US = 5_000_000`. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_device.py`: + +```python +# --- pulses payload validation (mirrors firmware limits exactly) --- + +from src.device import build_pulses_payload + + +def test_pulses_payload_shape(): + payload = build_pulses_payload([(180, 540), (180, 5580)], repeat_count=6) + assert payload == {"pulses": [[180, 540], [180, 5580]], "repeat_count": 6} + + +def test_pulses_payload_rejects_empty(): + with pytest.raises(ValueError, match="at least one"): + build_pulses_payload([], repeat_count=6) + + +def test_pulses_payload_rejects_too_many_pairs(): + with pytest.raises(ValueError, match="256"): + build_pulses_payload([(10, 10)] * 257, repeat_count=1) + + +def test_pulses_payload_rejects_out_of_range_us(): + with pytest.raises(ValueError, match="1..100000"): + build_pulses_payload([(0, 540)], repeat_count=6) + with pytest.raises(ValueError, match="1..100000"): + build_pulses_payload([(180, 100_001)], repeat_count=6) + + +def test_pulses_payload_rejects_bad_repeat_count(): + for bad in (0, 101): + with pytest.raises(ValueError, match="repeat_count"): + build_pulses_payload([(180, 540)], repeat_count=bad) + + +def test_pulses_payload_rejects_over_duration_budget(): + # 256 pairs x 200ms x 100 reps = 5120s >> 5s budget. The firmware + # hard-resets on payloads like this (soft WDT ~3.2s); fail locally. + with pytest.raises(ValueError, match="budget"): + build_pulses_payload([(100_000, 100_000)] * 256, repeat_count=100) + + +def test_zap_frame_fits_budget_comfortably(): + pulses = pt2260_pulses("0F1F0F0F1010", PT2260_TIMING) + payload = build_pulses_payload(pulses, repeat_count=6) + total_us = sum(h + l for h, l in pulses) * payload["repeat_count"] + assert total_us < 5_000_000 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_device.py -v -k pulses_payload` +Expected: FAIL — `ImportError: cannot import name 'build_pulses_payload'` + +- [ ] **Step 3: Implement in `src/device.py`** + +```python +# Limits mirror firmware/src/main.cpp exactly so bad payloads fail here +# with a readable message instead of a NodeMCU 400 (or a watchdog reset). +MAX_PULSE_PAIRS = 256 +MAX_PULSE_US = 100_000 +MAX_TOTAL_US = 5_000_000 # repeat_count x sum(high+low) busy-waits the ESP8266 + + +def build_pulses_payload(pulses: list[tuple[int, int]], repeat_count: int) -> dict: + """JSON-ready body for POST /transmit (pulse-train shape).""" + if not pulses: + raise ValueError("pulses must contain at least one (high_us, low_us) pair") + if len(pulses) > MAX_PULSE_PAIRS: + raise ValueError(f"too many pulse pairs ({len(pulses)}), max {MAX_PULSE_PAIRS}") + if not 1 <= repeat_count <= 100: + raise ValueError(f"repeat_count must be 1..100, got {repeat_count}") + total_us = 0 + for high_us, low_us in pulses: + for value in (high_us, low_us): + if not 1 <= value <= MAX_PULSE_US: + raise ValueError(f"pulse durations must be 1..{MAX_PULSE_US} µs, got {value}") + total_us += high_us + low_us + if total_us * repeat_count > MAX_TOTAL_US: + raise ValueError( + f"transmission exceeds duration budget: {total_us * repeat_count} µs " + f"> {MAX_TOTAL_US} µs (would busy-wait the ESP8266 into a watchdog reset)" + ) + return {"pulses": [[h, l] for h, l in pulses], "repeat_count": repeat_count} +``` + +- [ ] **Step 4: Run the full suite** + +Run: `uv run pytest tests/ -v` +Expected: all PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/device.py tests/test_device.py +git commit -m "feat(device): pulses payload builder with firmware-mirrored limits" +``` + +--- + +### Task 3: Firmware — pulse-train body shape on POST /transmit + +**Files:** +- Modify: `firmware/src/main.cpp` (handler starts at `handleTransmit()`, ~line 107) + +**Interfaces:** +- Consumes: JSON body `{"pulses": [[high_us, low_us], ...], "repeat_count": N}` (exactly what Task 2 emits). +- Produces: HTTP 200 `OK pairs x reps`; legacy `{bits, timing}` shape and all `/fan/*` routes unchanged. + +No host-side unit test exists for firmware; the "test" is a clean compile (Step 3) and the Task 7 bench capture. Do not refactor the legacy paths. + +- [ ] **Step 1: Add the pulse-train transmitter after `transmitGeneric` (~line 101)** + +```cpp +// Fully generic transmit: an explicit train of (high_us, low_us) pairs. +// This is the last protocol-shaped capability the firmware should ever +// need — PDM, PWM, tri-state etc. are all just pulse trains to the pin. +// Every pair ends LOW, so the pin is always left LOW. +void transmitPulses(JsonArrayConst pairs, int repeat_count) { + for (int r = 0; r < repeat_count; r++) { + for (JsonArrayConst pair : pairs) { + digitalWrite(TX_PIN, HIGH); + delayMicroseconds(pair[0].as()); + digitalWrite(TX_PIN, LOW); + delayMicroseconds(pair[1].as()); + // Fed per pair, not per repetition: delayMicroseconds() is a + // busy-wait and the soft WDT fires at ~3.2 s. + ESP.wdtFeed(); + } + } +} +``` + +- [ ] **Step 2: Route and validate the new body shape in `handleTransmit()`** + +Insert immediately after the `deserializeJson` error check (after the `if (err) {...}` block), before the `const char *bits = ...` line: + +```cpp + // New body shape: {"pulses": [[high_us, low_us], ...], "repeat_count": N} + if (doc["pulses"].is()) { + JsonArrayConst pairs = doc["pulses"].as(); + size_t n = pairs.size(); + if (n == 0 || n > 256) { + server.send(400, "text/plain", "pulses must be 1..256 pairs\n"); + return; + } + int repeat_count = doc["repeat_count"] | 0; + if (repeat_count < 1 || repeat_count > 100) { + server.send(400, "text/plain", "repeat_count must be 1..100\n"); + return; + } + uint64_t total_us = 0; + for (JsonArrayConst pair : pairs) { + if (pair.size() != 2) { + server.send(400, "text/plain", "each pulse must be [high_us, low_us]\n"); + return; + } + uint32_t high_us = pair[0] | 0u; + uint32_t low_us = pair[1] | 0u; + if (high_us == 0 || high_us > 100000 || low_us == 0 || low_us > 100000) { + server.send(400, "text/plain", "pulse durations must be 1..100000 us\n"); + return; + } + total_us += high_us + low_us; + } + // Budget: worst-case limits would otherwise allow a 51 s busy-wait + // in a single repetition — far past the ~3.2 s soft watchdog. + if (total_us * (uint64_t)repeat_count > 5000000ULL) { + server.send(400, "text/plain", "transmission exceeds 5 s duration budget\n"); + return; + } + transmitPulses(pairs, repeat_count); + String reply = String("OK ") + n + " pairs x " + repeat_count + " reps\n"; + server.send(200, "text/plain", reply); + return; + } +``` + +- [ ] **Step 3: Compile** + +Run: `cd firmware && pio run` +Expected: `SUCCESS` (no upload yet — flashing happens in Task 7 with Tom present). +If `pio` is not on PATH: `export PATH="$HOME/.local/bin:/opt/homebrew/bin:$PATH"` first; it may also live at `~/.platformio/penv/bin/pio`. + +- [ ] **Step 4: Update the endpoint banner in `setup()`** + +Change the existing line: + +```cpp + Serial.println(" POST http://ceilingfans.local/transmit (generic bits + timing)"); +``` + +to: + +```cpp + Serial.println(" POST http://ceilingfans.local/transmit (bits+timing, or pulses+repeat_count)"); +``` + +- [ ] **Step 5: Re-compile, then commit** + +Run: `cd firmware && pio run` — Expected: `SUCCESS`. + +```bash +git add firmware/src/main.cpp +git commit -m "feat(firmware): generic pulse-train shape on POST /transmit" +``` + +--- + +### Task 4: Route profiles through one payload chooser; wire the CLI + +**Files:** +- Modify: `src/device.py`, `cli.py` (in `send`, the `build_packet`/`build_transmit_payload` pair, ~lines 60-69) +- Test: `tests/test_device.py` + +**Interfaces:** +- Consumes: everything from Tasks 1–2. +- Produces: `build_payload_for(profile: DeviceProfile, unit: str, command: str) -> dict` — returns the pulses payload for `encoding == "PT2260"`, the legacy bits payload otherwise. `cli.py send` calls only this. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_device.py`: + +```python +# --- payload routing by encoding --- + +from src.device import build_payload_for + + +def test_payload_for_pt2260_profile_is_pulse_shaped(): + payload = build_payload_for(_pt2260_profile(), "window", "on") + assert set(payload) == {"pulses", "repeat_count"} + assert payload["repeat_count"] == 6 + assert len(payload["pulses"]) == 25 + assert payload["pulses"][-1] == [180, 5580] # sync pair last + + +def test_payload_for_fan_profile_unchanged(profile): + payload = build_payload_for(profile, "main", "light") + assert set(payload) == {"bits", "timing"} + assert payload == build_transmit_payload( + profile, bits=build_packet(profile, unit="main", command="light") + ) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/test_device.py -v -k payload_for` +Expected: FAIL — `ImportError: cannot import name 'build_payload_for'` + +- [ ] **Step 3: Implement in `src/device.py`** + +```python +def build_payload_for(profile: DeviceProfile, unit: str, command: str) -> dict: + """Build the right POST /transmit body for this profile's encoding.""" + if profile.encoding == "PT2260": + code = resolve_code(profile, unit, command) + pulses = pt2260_pulses(code, profile.timing) + return build_pulses_payload(pulses, profile.timing["repeat_count"]) + bits = build_packet(profile, unit=unit, command=command) + return build_transmit_payload(profile, bits=bits) +``` + +- [ ] **Step 4: Use it in `cli.py`** + +In `send`, replace the two-step `bits = build_packet(...)` / `payload = build_transmit_payload(profile, bits=bits)` sequence with: + +```python + payload = build_payload_for(profile, unit=unit, command=command) +``` + +and update the import line to pull `build_payload_for` from `src.device` (keep existing imports that `raw` still uses; drop any that become unused — ruff will flag them). + +- [ ] **Step 5: Run the full suite (covers tests/test_cli.py regressions)** + +Run: `uv run pytest tests/ -v` +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/device.py cli.py tests/test_device.py +git commit -m "feat(cli): route PT2260 profiles through pulse-train payloads" +``` + +--- + +### Task 5: Derived web bundle — scripts/export_web_devices.py + +**Files:** +- Create: `scripts/export_web_devices.py` +- Test: `tests/test_export_web_devices.py` + +**Interfaces:** +- Consumes: `DeviceProfile.load`, `build_payload_for` (Task 4). +- Produces: `build_bundle(device_dir: Path) -> dict` and a `__main__` that prints JSON to stdout. Bundle shape (what the homelab PWA will consume — keep stable): + +```json +{ + "lights": [ + { + "unit": "window", + "label": "Window", + "position": 2, + "commands": { + "on": {"pulses": [[180, 540], "..."], "repeat_count": 6}, + "off": {"pulses": [[180, 540], "..."], "repeat_count": 6} + } + } + ] +} +``` + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_export_web_devices.py`: + +```python +import json +from pathlib import Path + +from scripts.export_web_devices import build_bundle + +ZAP_YAML = """\ +frequency_mhz: 433.92 +encoding: PT2260 +timing: {short_us: 180, long_us: 540, sync_gap_us: 5580, repeat_count: 6} +units: + couch: {position: 3, codes: {'on': '0F1F0F0F1100', 'off': '0F1F0F0F0011'}} + window: {position: 2, codes: {'on': '0F1F0F0F1010', 'off': '0F1F0F0F1001'}} +""" + +FAN_YAML_PATH = Path("devices/sofa_king_fan.yaml") + + +def test_bundle_exports_pt2260_units_sorted_by_position(tmp_path): + (tmp_path / "zap_lights.yaml").write_text(ZAP_YAML) + # Non-PT2260 profiles are skipped (fans stay on their GET endpoints). + (tmp_path / "fan.yaml").write_text(FAN_YAML_PATH.read_text()) + + bundle = build_bundle(tmp_path) + + assert [u["unit"] for u in bundle["lights"]] == ["window", "couch"] + window = bundle["lights"][0] + assert window["label"] == "Window" + assert window["position"] == 2 + assert window["commands"]["on"]["repeat_count"] == 6 + assert len(window["commands"]["on"]["pulses"]) == 25 + json.dumps(bundle) # must be JSON-serializable as-is +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/test_export_web_devices.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'scripts.export_web_devices'` +(If `scripts/` lacks an `__init__.py` and the import fails for that reason instead, create an empty `scripts/__init__.py` — same as `tests/`.) + +- [ ] **Step 3: Implement `scripts/export_web_devices.py`** + +```python +"""Derive the PWA's devices.json from the YAML device profiles. + +The homelab 22-parsons-remote PWA ships this file so it never duplicates +codes or timings — devices/*.yaml stays the single source of truth. + +Usage: + uv run python scripts/export_web_devices.py > devices.json +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +from src.device import DeviceProfile, build_payload_for + + +def build_bundle(device_dir: Path) -> dict: + """Bundle every PT2260 profile's units into PWA-ready payloads.""" + lights = [] + for path in sorted(device_dir.glob("*.yaml")): + profile = DeviceProfile.load(str(path)) + if profile.encoding != "PT2260": + continue # fans et al. stay on their existing endpoints + for unit_name, unit in profile.units.items(): + lights.append( + { + "unit": unit_name, + "label": unit_name.capitalize(), + "position": unit["position"], + "commands": { + command: build_payload_for(profile, unit_name, command) + for command in unit["codes"] + }, + } + ) + lights.sort(key=lambda entry: entry["position"]) + return {"lights": lights} + + +if __name__ == "__main__": + json.dump(build_bundle(Path("devices")), sys.stdout, indent=2) + sys.stdout.write("\n") +``` + +- [ ] **Step 4: Run the full suite** + +Run: `uv run pytest tests/ -v` +Expected: all PASS. + +- [ ] **Step 5: Commit** + +```bash +git add scripts/export_web_devices.py tests/test_export_web_devices.py +git commit -m "feat(scripts): derive PWA devices.json from YAML profiles" +``` + +--- + +### Task 6: Capture the remote and write the real device profile — TOM REQUIRED + +**Files:** +- Create: `devices/zap_lights.yaml`, `captures/zap_remote_pos{2-5}_{on,off}.cu8` +- Modify: `PROTOCOL.md` (append a ZAP section) +- Test: `tests/test_device.py` + +This is the physical gate. Tom presses buttons; the agent runs commands and records results. + +- [ ] **Step 1: Capture all 8 buttons** + +For each of positions 2–5, ON then OFF (8 recordings): + +```bash +export PATH="$HOME/.local/bin:/opt/homebrew/bin:$PATH" +rtl_433 -f 433.92M -A -S unknown +# Tom presses ONE button for ~1 second, then Ctrl-C. +# -A prints the pulse analysis live; -S unknown writes g###_433.92M_250k.cu8 +# into the current directory. Rename each to captures/zap_remote_pos_.cu8 +``` + +Record for each button: the tri-state code (rtl_433 -A prints PWM pulse widths and often the decoded row), measured short pulse µs, long pulse µs, and sync gap µs. +Fallback ladder if `-A` output is ambiguous: URH (`brew install urh`) → upload the .cu8 to triq.org/pdv → Audacity on an AM-demod recording (the fan workflow). + +- [ ] **Step 2: Sanity-check the 8 codes against each other** + +Expected pattern per issue #10 research: a shared 8-symbol prefix (remote address) with per-button variation in the tail. If the split is NOT clean, nothing changes — the YAML stores full codes anyway. Note whatever structure appears in PROTOCOL.md. + +- [ ] **Step 3: Write `devices/zap_lights.yaml` with the measured values** + +Template — replace every `MEASURED` with numbers from Step 1 and every `CODE` with the captured 12-symbol strings: + +```yaml +# Etekcity ZAP 5LX outlet remote — RF Device Profile +# Encoding: PT2260-family tri-state OOK PWM (HS2260A-R4 encoder) +# Codes captured from OUR living-room remote on 2026-XX-XX — see PROTOCOL.md. +# Positions 2-5 drive the living room lamps; outlets are learning-code +# receivers paired to this remote. + +frequency_mhz: 433.92 # documentation-only: the MX-FS-03V TX is SAW-locked here +encoding: PT2260 + +timing: + short_us: MEASURED # 1-alpha segment + long_us: MEASURED # 3-alpha segment (~3x short_us) + sync_gap_us: MEASURED # long LOW after the sync pulse (~31x short_us) + repeat_count: 6 # remote repeats while held; receiver needs a few clean ones + +# Full 12-symbol tri-state code per button (no address/command factoring — +# the capture is the source of truth; see the design spec deviation note). +units: + window: + position: 2 + codes: {"on": "CODE", "off": "CODE"} + couch: + position: 3 + codes: {"on": "CODE", "off": "CODE"} + speaker: + position: 4 + codes: {"on": "CODE", "off": "CODE"} + chairs: + position: 5 + codes: {"on": "CODE", "off": "CODE"} +``` + +- [ ] **Step 4: Write the failing test, then make it pass with the real file** + +Append to `tests/test_device.py`: + +```python +# --- real ZAP profile --- + +ZAP_PROFILE_PATH = "devices/zap_lights.yaml" + + +@pytest.fixture +def zap_profile(): + return DeviceProfile.load(ZAP_PROFILE_PATH) + + +def test_zap_profile_has_all_four_lamps(zap_profile): + assert set(zap_profile.units) == {"window", "couch", "speaker", "chairs"} + + +def test_zap_all_eight_buttons_encode(zap_profile): + for unit, spec in zap_profile.units.items(): + for command in ("on", "off"): + payload = build_payload_for(zap_profile, unit, command) + assert len(payload["pulses"]) == 25, f"{unit}/{command}" + assert payload["repeat_count"] == zap_profile.timing["repeat_count"] + + +def test_zap_codes_are_twelve_tristate_symbols(zap_profile): + for unit, spec in zap_profile.units.items(): + for command, code in spec["codes"].items(): + assert len(code) == 12, f"{unit}/{command}" + assert set(code).issubset({"0", "1", "F"}), f"{unit}/{command}" +``` + +Run: `uv run pytest tests/test_device.py -v -k zap` — Expected: PASS (fails only if the YAML is malformed, which is the point). + +- [ ] **Step 5: Document in PROTOCOL.md** + +Append a `## Etekcity ZAP 5LX (lighting outlets)` section: measured timings, the 8 codes in a table (position / on / off), observed address/data structure, capture filenames, and the rtl_433 command used. + +- [ ] **Step 6: Commit** + +```bash +git add devices/zap_lights.yaml captures/zap_remote_*.cu8 PROTOCOL.md tests/test_device.py +git commit -m "feat(devices): ZAP 5LX lighting profile from live capture" +``` + +(If the .cu8 files are large, check `git ls-files captures/` first — the fan-era WAVs are committed, so captures belong in git per repo convention.) + +--- + +### Task 7: Flash, bench-verify, and switch real lamps — TOM REQUIRED + +**Files:** none created (verification task; findings go in PROTOCOL.md if timings need adjustment). + +- [ ] **Step 1: Flash the firmware** + +```bash +cd firmware && pio run -t upload +# NodeMCU on USB; if the port isn't auto-found: pio device list +``` + +- [ ] **Step 2: Bench proof — capture the NodeMCU's own transmission** + +Terminal A: `rtl_433 -f 433.92M -A -S unknown` +Terminal B: `uv run python cli.py send zap_lights window on` +Compare the analyzer output against the remote's Step-1 capture for the same button: same code, pulse widths within ~10%. This is the same technique that validated the fans (`captures/nodemcu_main_light.wav`). If widths drift, adjust `timing:` in the YAML (never the firmware) and re-send. + +- [ ] **Step 3: End-to-end — all 8 buttons against real lamps** + +```bash +for unit in window couch speaker chairs; do + uv run python cli.py send zap_lights $unit on; sleep 2 + uv run python cli.py send zap_lights $unit off; sleep 2 +done +``` + +Tom confirms each lamp switches. Any failures: check the outlet is paired (side button re-learns), then re-check Step 2's timing diff. + +- [ ] **Step 4: Regenerate and eyeball the web bundle** + +```bash +uv run python scripts/export_web_devices.py | head -30 +``` + +Expected: all four lamps present, sorted by position 2→5. + +- [ ] **Step 5: Update project docs and commit** + +Update `CLAUDE.md` "Project Status": lights decoded + controllable via CLI; homelab PWA work tracked in the companion issue. + +```bash +git add CLAUDE.md PROTOCOL.md devices/zap_lights.yaml +git commit -m "docs: ZAP lighting verified end-to-end" +``` + +--- + +## Done means + +- `uv run pytest tests/` green. +- `pio run` compiles clean. +- All four lamps switch from `cli.py` (Task 7 Step 3 witnessed by Tom). +- `export_web_devices.py` emits the bundle the homelab companion issue consumes. +- Fans still work (spot-check one `/fan/1/light` GET after flashing). From 24caba09775a6c019a8e50e413d0a2a1d9bd67b6 Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 08:39:13 -0400 Subject: [PATCH 4/9] feat(device): PT2260 tri-state pulse-train encoder Co-Authored-By: Claude Fable 5 --- src/device.py | 35 ++++++++++++++++++- tests/test_device.py | 82 +++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 115 insertions(+), 2 deletions(-) diff --git a/src/device.py b/src/device.py index 13c6403..8220ca0 100644 --- a/src/device.py +++ b/src/device.py @@ -19,7 +19,7 @@ def load(cls, path: str) -> "DeviceProfile": frequency_mhz=data["frequency_mhz"], encoding=data["encoding"], timing=data["timing"], - commands=data["commands"], + commands=data.get("commands", {}), units=data["units"], ) @@ -55,3 +55,36 @@ def build_transmit_payload(profile: DeviceProfile, bits: str) -> dict: raise ValueError("bits must contain only '0' and '1'") timing = {k: profile.timing[k] for k in _TIMING_KEYS} return {"bits": bits, "timing": timing} + + +# PT2260 tri-state symbols → two (level-duration) pulse halves each. +# Sync is a short HIGH followed by the long inter-frame gap; it is +# appended LAST so repeats are contiguous valid codewords and the +# TX pin is always left LOW. +_PT2260_SYMBOLS = { + "0": (("short", "long"), ("short", "long")), + "1": (("long", "short"), ("long", "short")), + "F": (("short", "long"), ("long", "short")), +} + + +def pt2260_pulses(code: str, timing: dict) -> list[tuple[int, int]]: + """Encode a PT2260 tri-state code string as [(high_us, low_us), ...].""" + duration = {"short": timing["short_us"], "long": timing["long_us"]} + pulses: list[tuple[int, int]] = [] + for symbol in code: + if symbol not in _PT2260_SYMBOLS: + raise ValueError(f"invalid PT2260 symbol {symbol!r} (want 0/1/F)") + for high, low in _PT2260_SYMBOLS[symbol]: + pulses.append((duration[high], duration[low])) + pulses.append((timing["short_us"], timing["sync_gap_us"])) + return pulses + + +def resolve_code(profile: DeviceProfile, unit: str, command: str) -> str: + """Full tri-state code for a unit+command. + + PT2260 profiles keep whole codes per unit (no address/command split — + see the design spec); this is the only function that knows that. + """ + return profile.units[unit]["codes"][command] # KeyError on unknown names diff --git a/tests/test_device.py b/tests/test_device.py index 484d67a..4490e48 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -1,6 +1,12 @@ import pytest -from src.device import DeviceProfile, build_packet, build_transmit_payload +from src.device import ( + DeviceProfile, + build_packet, + build_transmit_payload, + pt2260_pulses, + resolve_code, +) PROFILE_PATH = "devices/sofa_king_fan.yaml" @@ -126,3 +132,77 @@ def test_build_transmit_payload_rejects_bad_bits(profile): build_transmit_payload(profile, bits="") with pytest.raises(ValueError): build_transmit_payload(profile, bits="0102") + + +# --- PT2260 pulse-train encoding --- + +PT2260_TIMING = { + "short_us": 180, + "long_us": 540, + "sync_gap_us": 5580, + "repeat_count": 6, +} + + +def _pt2260_profile(): + return DeviceProfile( + frequency_mhz=433.92, + encoding="PT2260", + timing=PT2260_TIMING, + commands={}, + units={ + "window": { + "position": 2, + "codes": {"on": "0F1F0F0F1010", "off": "0F1F0F0F1001"}, + }, + }, + ) + + +def test_pt2260_symbol_waveforms(): + # '0' = 2x (short-high, long-low); '1' = 2x (long-high, short-low); + # 'F' = (short-high, long-low) then (long-high, short-low); sync pair last. + assert pt2260_pulses("01F", PT2260_TIMING) == [ + (180, 540), + (180, 540), # 0 + (540, 180), + (540, 180), # 1 + (180, 540), + (540, 180), # F + (180, 5580), # sync + ] + + +def test_pt2260_full_frame_is_25_pairs(): + # 12 symbols x 2 pulses + 1 sync pair + assert len(pt2260_pulses("0F1F0F0F1010", PT2260_TIMING)) == 25 + + +def test_pt2260_rejects_invalid_symbol(): + with pytest.raises(ValueError, match="symbol"): + pt2260_pulses("01X", PT2260_TIMING) + + +def test_resolve_code_looks_up_unit_command(): + profile = _pt2260_profile() + assert resolve_code(profile, "window", "on") == "0F1F0F0F1010" + + +def test_resolve_code_unknown_unit_raises_keyerror(): + with pytest.raises(KeyError): + resolve_code(_pt2260_profile(), "basement", "on") + + +def test_profile_load_tolerates_missing_commands(tmp_path): + yaml_text = ( + "frequency_mhz: 433.92\n" + "encoding: PT2260\n" + "timing: {short_us: 180, long_us: 540, sync_gap_us: 5580, repeat_count: 6}\n" + "units:\n" + " window: {position: 2, codes: {'on': '0F1F0F0F1010', 'off': '0F1F0F0F1001'}}\n" + ) + p = tmp_path / "zap.yaml" + p.write_text(yaml_text) + profile = DeviceProfile.load(str(p)) + assert profile.commands == {} + assert profile.units["window"]["codes"]["off"] == "0F1F0F0F1001" From 015ea18e7e6e4aeadd8c0ab489a5c68a071390de Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 08:44:16 -0400 Subject: [PATCH 5/9] feat(device): pulses payload builder with firmware-mirrored limits Co-Authored-By: Claude Fable 5 --- src/device.py | 34 ++++++++++++++++++++++++++++++++ tests/test_device.py | 46 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) diff --git a/src/device.py b/src/device.py index 8220ca0..cef85ef 100644 --- a/src/device.py +++ b/src/device.py @@ -88,3 +88,37 @@ def resolve_code(profile: DeviceProfile, unit: str, command: str) -> str: see the design spec); this is the only function that knows that. """ return profile.units[unit]["codes"][command] # KeyError on unknown names + + +# Limits mirror firmware/src/main.cpp exactly so bad payloads fail here +# with a readable message instead of a NodeMCU 400 (or a watchdog reset). +MAX_PULSE_PAIRS = 256 +MAX_PULSE_US = 100_000 +MAX_TOTAL_US = 5_000_000 # repeat_count x sum(high+low) busy-waits the ESP8266 + + +def build_pulses_payload(pulses: list[tuple[int, int]], repeat_count: int) -> dict: + """JSON-ready body for POST /transmit (pulse-train shape).""" + if not pulses: + raise ValueError("pulses must contain at least one (high_us, low_us) pair") + if len(pulses) > MAX_PULSE_PAIRS: + raise ValueError(f"too many pulse pairs ({len(pulses)}), max {MAX_PULSE_PAIRS}") + if not 1 <= repeat_count <= 100: + raise ValueError(f"repeat_count must be 1..100, got {repeat_count}") + total_us = 0 + for high_us, low_us in pulses: + for value in (high_us, low_us): + if not 1 <= value <= MAX_PULSE_US: + raise ValueError( + f"pulse durations must be 1..{MAX_PULSE_US} µs, got {value}" + ) + total_us += high_us + low_us + if total_us * repeat_count > MAX_TOTAL_US: + raise ValueError( + f"transmission exceeds duration budget: {total_us * repeat_count} µs " + f"> {MAX_TOTAL_US} µs (would busy-wait the ESP8266 into a watchdog reset)" + ) + return { + "pulses": [[high, low] for high, low in pulses], + "repeat_count": repeat_count, + } diff --git a/tests/test_device.py b/tests/test_device.py index 4490e48..f0af5f0 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -3,6 +3,7 @@ from src.device import ( DeviceProfile, build_packet, + build_pulses_payload, build_transmit_payload, pt2260_pulses, resolve_code, @@ -206,3 +207,48 @@ def test_profile_load_tolerates_missing_commands(tmp_path): profile = DeviceProfile.load(str(p)) assert profile.commands == {} assert profile.units["window"]["codes"]["off"] == "0F1F0F0F1001" + + +# --- pulses payload validation (mirrors firmware limits exactly) --- + + +def test_pulses_payload_shape(): + payload = build_pulses_payload([(180, 540), (180, 5580)], repeat_count=6) + assert payload == {"pulses": [[180, 540], [180, 5580]], "repeat_count": 6} + + +def test_pulses_payload_rejects_empty(): + with pytest.raises(ValueError, match="at least one"): + build_pulses_payload([], repeat_count=6) + + +def test_pulses_payload_rejects_too_many_pairs(): + with pytest.raises(ValueError, match="256"): + build_pulses_payload([(10, 10)] * 257, repeat_count=1) + + +def test_pulses_payload_rejects_out_of_range_us(): + with pytest.raises(ValueError, match="1..100000"): + build_pulses_payload([(0, 540)], repeat_count=6) + with pytest.raises(ValueError, match="1..100000"): + build_pulses_payload([(180, 100_001)], repeat_count=6) + + +def test_pulses_payload_rejects_bad_repeat_count(): + for bad in (0, 101): + with pytest.raises(ValueError, match="repeat_count"): + build_pulses_payload([(180, 540)], repeat_count=bad) + + +def test_pulses_payload_rejects_over_duration_budget(): + # 256 pairs x 200ms x 100 reps = 5120s >> 5s budget. The firmware + # hard-resets on payloads like this (soft WDT ~3.2s); fail locally. + with pytest.raises(ValueError, match="budget"): + build_pulses_payload([(100_000, 100_000)] * 256, repeat_count=100) + + +def test_zap_frame_fits_budget_comfortably(): + pulses = pt2260_pulses("0F1F0F0F1010", PT2260_TIMING) + payload = build_pulses_payload(pulses, repeat_count=6) + total_us = sum(high + low for high, low in pulses) * payload["repeat_count"] + assert total_us < 5_000_000 From 14cc2221703a82a24b525b4c41b87c2d9746ab46 Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 08:50:04 -0400 Subject: [PATCH 6/9] feat(firmware): generic pulse-train shape on POST /transmit Co-Authored-By: Claude Fable 5 --- firmware/src/main.cpp | 59 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/firmware/src/main.cpp b/firmware/src/main.cpp index abf1c93..b508ad8 100644 --- a/firmware/src/main.cpp +++ b/firmware/src/main.cpp @@ -100,6 +100,24 @@ void transmitGeneric(const char *bits, } } +// Fully generic transmit: an explicit train of (high_us, low_us) pairs. +// This is the last protocol-shaped capability the firmware should ever +// need — PDM, PWM, tri-state etc. are all just pulse trains to the pin. +// Every pair ends LOW, so the pin is always left LOW. +void transmitPulses(JsonArrayConst pairs, int repeat_count) { + for (int r = 0; r < repeat_count; r++) { + for (JsonArrayConst pair : pairs) { + digitalWrite(TX_PIN, HIGH); + delayMicroseconds(pair[0].as()); + digitalWrite(TX_PIN, LOW); + delayMicroseconds(pair[1].as()); + // Fed per pair, not per repetition: delayMicroseconds() is a + // busy-wait and the soft WDT fires at ~3.2 s. + ESP.wdtFeed(); + } + } +} + // ─── HTTP HANDLERS ─────────────────────────────────────────────────────────── // POST /transmit — body is JSON: // {"bits": "010...", "timing": {"sync_us":N, "sync_gap_us":N, @@ -121,6 +139,45 @@ void handleTransmit() { return; } + // New body shape: {"pulses": [[high_us, low_us], ...], "repeat_count": N} + if (doc["pulses"].is()) { + JsonArrayConst pairs = doc["pulses"].as(); + size_t n = pairs.size(); + if (n == 0 || n > 256) { + server.send(400, "text/plain", "pulses must be 1..256 pairs\n"); + return; + } + int repeat_count = doc["repeat_count"] | 0; + if (repeat_count < 1 || repeat_count > 100) { + server.send(400, "text/plain", "repeat_count must be 1..100\n"); + return; + } + uint64_t total_us = 0; + for (JsonArrayConst pair : pairs) { + if (pair.size() != 2) { + server.send(400, "text/plain", "each pulse must be [high_us, low_us]\n"); + return; + } + uint32_t high_us = pair[0] | 0u; + uint32_t low_us = pair[1] | 0u; + if (high_us == 0 || high_us > 100000 || low_us == 0 || low_us > 100000) { + server.send(400, "text/plain", "pulse durations must be 1..100000 us\n"); + return; + } + total_us += high_us + low_us; + } + // Budget: worst-case limits would otherwise allow a 51 s busy-wait + // in a single repetition — far past the ~3.2 s soft watchdog. + if (total_us * (uint64_t)repeat_count > 5000000ULL) { + server.send(400, "text/plain", "transmission exceeds 5 s duration budget\n"); + return; + } + transmitPulses(pairs, repeat_count); + String reply = String("OK ") + n + " pairs x " + repeat_count + " reps\n"; + server.send(200, "text/plain", reply); + return; + } + const char *bits = doc["bits"] | (const char *)nullptr; if (!bits) { server.send(400, "text/plain", "missing 'bits'\n"); @@ -262,7 +319,7 @@ void setup() { server.begin(); Serial.println("HTTP server ready"); Serial.println("Endpoints:"); - Serial.println(" POST http://ceilingfans.local/transmit (generic bits + timing)"); + Serial.println(" POST http://ceilingfans.local/transmit (bits+timing, or pulses+repeat_count)"); Serial.println(" GET http://ceilingfans.local/fan/{1,2}/{light,off,speed1,speed2,speed3}"); } From 674a5dc5944cf6e7d65087d00a6c42fe2d3e53b3 Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 08:56:43 -0400 Subject: [PATCH 7/9] feat(cli): route PT2260 profiles through pulse-train payloads Co-Authored-By: Claude Fable 5 --- cli.py | 7 +++---- src/device.py | 10 ++++++++++ tests/test_device.py | 20 ++++++++++++++++++++ 3 files changed, 33 insertions(+), 4 deletions(-) diff --git a/cli.py b/cli.py index 152e420..6bad2f7 100644 --- a/cli.py +++ b/cli.py @@ -11,7 +11,7 @@ import click import httpx -from src.device import DeviceProfile, build_packet, build_transmit_payload +from src.device import DeviceProfile, build_payload_for, build_transmit_payload DEVICES_DIR = "devices" @@ -64,10 +64,9 @@ def send(device: str, unit: str, command: str, host: str, port: int) -> None: ) sys.exit(1) - bits = build_packet(profile, unit=unit, command=command) - payload = build_transmit_payload(profile, bits=bits) + payload = build_payload_for(profile, unit=unit, command=command) _post_transmit(host, port, payload) - click.echo(f"OK {command} → {device}/{unit} [{bits}]") + click.echo(f"OK {command} → {device}/{unit}") @cli.command() diff --git a/src/device.py b/src/device.py index cef85ef..006297a 100644 --- a/src/device.py +++ b/src/device.py @@ -122,3 +122,13 @@ def build_pulses_payload(pulses: list[tuple[int, int]], repeat_count: int) -> di "pulses": [[high, low] for high, low in pulses], "repeat_count": repeat_count, } + + +def build_payload_for(profile: DeviceProfile, unit: str, command: str) -> dict: + """Build the right POST /transmit body for this profile's encoding.""" + if profile.encoding == "PT2260": + code = resolve_code(profile, unit, command) + pulses = pt2260_pulses(code, profile.timing) + return build_pulses_payload(pulses, profile.timing["repeat_count"]) + bits = build_packet(profile, unit=unit, command=command) + return build_transmit_payload(profile, bits=bits) diff --git a/tests/test_device.py b/tests/test_device.py index f0af5f0..aa16f82 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -3,6 +3,7 @@ from src.device import ( DeviceProfile, build_packet, + build_payload_for, build_pulses_payload, build_transmit_payload, pt2260_pulses, @@ -252,3 +253,22 @@ def test_zap_frame_fits_budget_comfortably(): payload = build_pulses_payload(pulses, repeat_count=6) total_us = sum(high + low for high, low in pulses) * payload["repeat_count"] assert total_us < 5_000_000 + + +# --- payload routing by encoding --- + + +def test_payload_for_pt2260_profile_is_pulse_shaped(): + payload = build_payload_for(_pt2260_profile(), "window", "on") + assert set(payload) == {"pulses", "repeat_count"} + assert payload["repeat_count"] == 6 + assert len(payload["pulses"]) == 25 + assert payload["pulses"][-1] == [180, 5580] # sync pair last + + +def test_payload_for_fan_profile_unchanged(profile): + payload = build_payload_for(profile, "main", "light") + assert set(payload) == {"bits", "timing"} + assert payload == build_transmit_payload( + profile, bits=build_packet(profile, unit="main", command="light") + ) From 59a299153b62697bc6d0f9a80adad7c4fab581c7 Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 09:01:51 -0400 Subject: [PATCH 8/9] feat(scripts): derive PWA devices.json from YAML profiles Co-Authored-By: Claude Fable 5 --- scripts/__init__.py | 0 scripts/export_web_devices.py | 47 ++++++++++++++++++++++++++++++++ tests/test_export_web_devices.py | 31 +++++++++++++++++++++ 3 files changed, 78 insertions(+) create mode 100644 scripts/__init__.py create mode 100644 scripts/export_web_devices.py create mode 100644 tests/test_export_web_devices.py diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/export_web_devices.py b/scripts/export_web_devices.py new file mode 100644 index 0000000..bb9d7eb --- /dev/null +++ b/scripts/export_web_devices.py @@ -0,0 +1,47 @@ +"""Derive the PWA's devices.json from the YAML device profiles. + +The homelab 22-parsons-remote PWA ships this file so it never duplicates +codes or timings — devices/*.yaml stays the single source of truth. + +Usage: + uv run python scripts/export_web_devices.py > devices.json +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +# Ensure the project root is in sys.path so we can import src +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.device import DeviceProfile, build_payload_for + + +def build_bundle(device_dir: Path) -> dict: + """Bundle every PT2260 profile's units into PWA-ready payloads.""" + lights = [] + for path in sorted(device_dir.glob("*.yaml")): + profile = DeviceProfile.load(str(path)) + if profile.encoding != "PT2260": + continue # fans et al. stay on their existing endpoints + for unit_name, unit in profile.units.items(): + lights.append( + { + "unit": unit_name, + "label": unit_name.capitalize(), + "position": unit["position"], + "commands": { + command: build_payload_for(profile, unit_name, command) + for command in unit["codes"] + }, + } + ) + lights.sort(key=lambda entry: entry["position"]) + return {"lights": lights} + + +if __name__ == "__main__": + json.dump(build_bundle(Path("devices")), sys.stdout, indent=2) + sys.stdout.write("\n") diff --git a/tests/test_export_web_devices.py b/tests/test_export_web_devices.py new file mode 100644 index 0000000..777f02c --- /dev/null +++ b/tests/test_export_web_devices.py @@ -0,0 +1,31 @@ +import json +from pathlib import Path + +from scripts.export_web_devices import build_bundle + +ZAP_YAML = """\ +frequency_mhz: 433.92 +encoding: PT2260 +timing: {short_us: 180, long_us: 540, sync_gap_us: 5580, repeat_count: 6} +units: + couch: {position: 3, codes: {'on': '0F1F0F0F1100', 'off': '0F1F0F0F0011'}} + window: {position: 2, codes: {'on': '0F1F0F0F1010', 'off': '0F1F0F0F1001'}} +""" + +FAN_YAML_PATH = Path("devices/sofa_king_fan.yaml") + + +def test_bundle_exports_pt2260_units_sorted_by_position(tmp_path): + (tmp_path / "zap_lights.yaml").write_text(ZAP_YAML) + # Non-PT2260 profiles are skipped (fans stay on their GET endpoints). + (tmp_path / "fan.yaml").write_text(FAN_YAML_PATH.read_text()) + + bundle = build_bundle(tmp_path) + + assert [u["unit"] for u in bundle["lights"]] == ["window", "couch"] + window = bundle["lights"][0] + assert window["label"] == "Window" + assert window["position"] == 2 + assert window["commands"]["on"]["repeat_count"] == 6 + assert len(window["commands"]["on"]["pulses"]) == 25 + json.dumps(bundle) # must be JSON-serializable as-is From 9bf91de434e1be232603421df4d42eafd4aaa96f Mon Sep 17 00:00:00 2001 From: Tom Clancy Date: Mon, 13 Jul 2026 09:10:52 -0400 Subject: [PATCH 9/9] =?UTF-8?q?fix(cli):=20validate=20commands=20per=20enc?= =?UTF-8?q?oding=20=E2=80=94=20PT2260=20codes=20live=20under=20units?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- cli.py | 12 +++++++++--- src/device.py | 7 +++++++ tests/test_cli.py | 46 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_device.py | 9 +++++++++ 4 files changed, 71 insertions(+), 3 deletions(-) diff --git a/cli.py b/cli.py index 6bad2f7..5248b3b 100644 --- a/cli.py +++ b/cli.py @@ -11,7 +11,12 @@ import click import httpx -from src.device import DeviceProfile, build_payload_for, build_transmit_payload +from src.device import ( + DeviceProfile, + available_commands, + build_payload_for, + build_transmit_payload, +) DEVICES_DIR = "devices" @@ -57,8 +62,9 @@ def send(device: str, unit: str, command: str, host: str, port: int) -> None: click.echo(f"Error: unknown unit '{unit}'. Available: {available}", err=True) sys.exit(1) - if command not in profile.commands: - available = ", ".join(sorted(profile.commands)) + commands = available_commands(profile, unit) + if command not in commands: + available = ", ".join(sorted(commands)) click.echo( f"Error: unknown command '{command}'. Available: {available}", err=True ) diff --git a/src/device.py b/src/device.py index 006297a..c6e6d84 100644 --- a/src/device.py +++ b/src/device.py @@ -90,6 +90,13 @@ def resolve_code(profile: DeviceProfile, unit: str, command: str) -> str: return profile.units[unit]["codes"][command] # KeyError on unknown names +def available_commands(profile: DeviceProfile, unit: str) -> set[str]: + """Command names valid for this unit under the profile's encoding.""" + if profile.encoding == "PT2260": + return set(profile.units[unit].get("codes", {})) + return set(profile.commands) + + # Limits mirror firmware/src/main.cpp exactly so bad payloads fail here # with a readable message instead of a NodeMCU 400 (or a watchdog reset). MAX_PULSE_PAIRS = 256 diff --git a/tests/test_cli.py b/tests/test_cli.py index a77cc2d..0bc98c2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,6 +4,29 @@ from click.testing import CliRunner from cli import cli +from src.device import DeviceProfile + +PT2260_TIMING = { + "short_us": 180, + "long_us": 540, + "sync_gap_us": 5580, + "repeat_count": 6, +} + + +def _pt2260_profile(): + return DeviceProfile( + frequency_mhz=433.92, + encoding="PT2260", + timing=PT2260_TIMING, + commands={}, + units={ + "window": { + "position": 2, + "codes": {"on": "0F1F0F0F1010", "off": "0F1F0F0F1001"}, + }, + }, + ) @pytest.fixture @@ -37,6 +60,29 @@ def test_send_posts_to_transmit_endpoint(runner): assert payload["timing"]["pulse_us"] == 560 +def test_send_posts_pulses_for_pt2260_profile(runner): + with ( + patch("cli.httpx.post") as mock_post, + patch("cli._load_profile", return_value=_pt2260_profile()), + ): + mock_post.return_value = MagicMock(status_code=200, text="OK") + mock_post.return_value.raise_for_status = MagicMock() + + result = runner.invoke( + cli, + ["send", "zap_lights", "window", "on", "--host", "1.2.3.4"], + ) + + assert result.exit_code == 0, result.output + assert mock_post.call_count == 1 + (url,) = mock_post.call_args.args + assert url == "http://1.2.3.4:80/transmit" + payload = mock_post.call_args.kwargs["json"] + assert set(payload) == {"pulses", "repeat_count"} + assert payload["repeat_count"] == 6 + assert all(len(pair) == 2 for pair in payload["pulses"]) + + def test_send_rejects_unknown_unit(runner): with patch("cli.httpx.post") as mock_post: result = runner.invoke( diff --git a/tests/test_device.py b/tests/test_device.py index aa16f82..6a0158c 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -2,6 +2,7 @@ from src.device import ( DeviceProfile, + available_commands, build_packet, build_payload_for, build_pulses_payload, @@ -195,6 +196,14 @@ def test_resolve_code_unknown_unit_raises_keyerror(): resolve_code(_pt2260_profile(), "basement", "on") +def test_available_commands_pt2260_reads_unit_codes(): + assert available_commands(_pt2260_profile(), "window") == {"on", "off"} + + +def test_available_commands_legacy_reads_profile_commands(profile): + assert available_commands(profile, "main") == set(profile.commands) + + def test_profile_load_tolerates_missing_commands(tmp_path): yaml_text = ( "frequency_mhz: 433.92\n"