From af6852445ab1600e3dddeee0804b4f6f67470dd1 Mon Sep 17 00:00:00 2001 From: orren5 Date: Fri, 10 Jul 2026 17:58:18 +0300 Subject: [PATCH 01/25] design doc --- documentation/Receiver Design.md | 481 +++++++++++++++++++++++++++++++ 1 file changed, 481 insertions(+) create mode 100644 documentation/Receiver Design.md diff --git a/documentation/Receiver Design.md b/documentation/Receiver Design.md new file mode 100644 index 0000000..7c0f65f --- /dev/null +++ b/documentation/Receiver Design.md @@ -0,0 +1,481 @@ +# Design: RTS Receiver — Track Physical Remote Presses + +Status: **Draft / proposal** +Target: Pi-Somfy v3.2+ + +## 1 Motivation + +Pi-Somfy transmits Somfy RTS frames and estimates shutter position from motor +travel times (`durationDown` / `durationUp`). This works well as long as +**every** command goes through Pi-Somfy. The moment someone presses a button on +a physical Somfy remote, the blind moves but Pi-Somfy (and therefore Home +Assistant) never learns about it — the tracked position is wrong until the next +full up/down through the software. + +RTS is a one-way broadcast protocol: remotes transmit, motors listen, nobody +acknowledges. But that also means anyone tuned to 433.42 MHz can hear the +remotes. This design adds an **RF receiver** so Pi-Somfy hears physical remote +presses and runs the *same* position estimation it already uses for its own +commands. + +``` +Physical remote press + → RF receiver (433.42 MHz OOK, data pin on RXGPIO) + → edge timestamps (pigpio callbacks on Pi 1–4 / lgpio alerts on Pi 5) + → RTS decoder (sync detect → manchester → de-XOR → checksum) + → frame {remote address, button, rolling code} + → filters: self-echo, repeated frames + → [PhysicalRemotes] mapping: address → shutterId(s) + → existing position simulation (rise/lower/stop math) + → setPosition() → existing callbacks → MQTT → Home Assistant +``` + +## 2 Goals / Non-goals + +**Goals** + +1. Detect UP / DOWN / STOP(MY) presses from physical RTS remotes and update the + tracked position of the mapped shutter(s). +2. Position changes propagate to MQTT / Home Assistant through the existing + callback path with no HA-side changes. +3. Simple pairing flow to map a physical remote (channel) to one or more + shutters. +4. Work on Pi 1–5, bare Raspbian install and the Home Assistant add-on alike. +5. Do not disturb the existing TX path in any way. + +**Non-goals (v1)** + +- Decoding encrypted Somfy io-homecontrol devices (different protocol entirely). +- Tilt/long-press handling (future work, see §10). +- Replacing the TX hardware with the receiver's transceiver (future work). + +## 3 Protocol background + +The decoder is the exact inverse of `Shutter.sendCommand()` in +`operateShutters.py`, which is the authoritative in-repo reference for the +frame layout. On air, one button press is: + +| Element | Timing | +|---|---| +| Wake-up pulse | 9 415 µs high, 89 565 µs low (first frame only) | +| Hardware sync | 2 560 µs high + 2 560 µs low, ×2 (first frame) or ×7 (repeats) | +| Software sync | 4 550 µs high + 640 µs low | +| Payload | 56 bits, manchester: `1` = low→high, `0` = high→low, 640 µs half-symbol | +| Inter-frame gap | 30 415 µs silence | + +Payload after de-obfuscation (`plain[i] = recv[i] XOR recv[i-1]`, i = 6…1): + +| Byte | Content | +|---|---| +| 0 | "Encryption key", 0xA0–0xAF | +| 1 | Button in high nibble (0x1 My/Stop, 0x2 Up, 0x4 Down, 0x8 Prog), 4-bit checksum in low nibble | +| 2–3 | Rolling code, big endian — increments on every press | +| 4–6 | Remote address (24 bit) — **unique per remote channel**; a 5-channel Telis appears as 5 addresses | + +Checksum check: XOR of all 14 nibbles of the de-obfuscated frame must equal 0. + +Every press is transmitted as one frame plus several repeats **with the same +rolling code**, so `(address, rollingCode)` uniquely identifies one press and +is a perfect de-duplication key. A held button keeps repeating frames +(same code) — the repeat count distinguishes short from long press. + +## 4 Hardware + +### 4.1 Why the frequency matters (again) + +Somfy RTS uses 433.**42** MHz; generic modules ship tuned to 433.**92** MHz. +For the transmitter the fix was swapping the 3-pin SAW resonator. **The same +trick does not exist for receivers:** + +- The cheap kit receiver (XY-MK-5V style, super-regenerative) sets its + frequency with an LC tank — there is no resonator to swap. It will hear + Somfy remotes only at very short range and is extremely noisy. Not viable. +- Superheterodyne receivers (RXB6/RXB8) use a local-oscillator crystal at + ~1/64 of the receive frequency; a 433.42 MHz cut (~6.73 MHz) is not a + commodity part. Running one unmodified (centered 500 kHz off) loses most of + its sensitivity. + +### 4.2 Recommended: CC1101 transceiver module (~$3) + +The CC1101 is tuned **in software**: we write its frequency registers once at +startup and set it to OOK receive with *asynchronous serial output*, after +which its `GDO0`/`GDO2` pin behaves exactly like the data pin of a dumb +receiver — demodulated 0/1 that we timestamp with GPIO edge callbacks, matching +the project's existing GPIO style. No soldering, no rare parts, exact +433.42 MHz, 3.3 V native (Pi-safe). + +Wiring (SPI is only used for one-time configuration; see §5.2): + +| CC1101 pin | Signal | Default GPIO | Physical pin (Pi 4) | Note | +|---|---|---|---|---| +| VCC | 3.3 V supply | — | 17 (or 1) | **never 5 V** | +| GND | ground | — | 39 | keeps the whole harness in one corner (34 also works) | +| SCK | SPI clock | GPIO 21 (`RXSpiSCK`) | 40 | bit-banged, any free GPIO | +| MOSI (SI) | SPI data → radio | GPIO 20 (`RXSpiMOSI`) | 38 | | +| MISO (SO) | SPI data → Pi | GPIO 19 (`RXSpiMISO`) | 35 | required — read-back verification (§5.1) | +| CSN | chip select | GPIO 16 (`RXSpiCSN`) | 36 | | +| GDO0 | demodulated data out | GPIO 26 (`RXGPIO`) | 37 | the receiver's actual data pin | +| GDO2 | — | not connected | — | | +| ANT | antenna | — | — | 17 cm solid-core wire, **required** — see §4.4 | + +The defaults deliberately cluster every signal in the bottom corner of the +40-pin header (physical pins 35–40, plus 3.3 V from pin 17), far from the +existing transmitter on GPIO 4 (physical pin 7). All pins are configurable +(§5.3); module silkscreens vary between CC1101 board revisions, so always +match by label, not by position (MOSI may be printed `SI`, MISO `SO`). + +``` + ┌──────┬──────┐ + 3.3V (VCC) │ 17 │ 18 │ + │ .. │ .. │ + │ 33 │ 34 │ + GPIO19 (MISO) │ 35 │ 36 │ GPIO16 (CSN) + GPIO26 (GDO0) │ 37 │ 38 │ GPIO20 (MOSI) + GND │ 39 │ 40 │ GPIO21 (SCK) + └──────┴──────┘ +``` + +### 4.3 Alternative: RXB6/RXB8 superheterodyne + +For purists who accept reduced range or manage to source a 433.42 crystal: +power at **3.3 V** (the data pin follows VCC; 5 V would damage the Pi), data +pin → `RXGPIO`. Works with the same software; only §5.2 (CC1101 init) is +skipped. Not the recommended path. + +### 4.4 Antenna + +Today's antenna-less transmitter reaches the whole house because the *blind +motors* have good factory antennas — the weak TX signal is compensated by good +ears on the receiving end. For the new receive direction the roles flip: the +Pi must hear a handheld remote pressed 15–20 m and several walls away, and a +receiver without an antenna has terrible ears. The 17 cm quarter-wave wire +(same as the README describes for TX) is mandatory on the receiver; many +CC1101 modules ship with a coil antenna or SMA connector. + +## 5 Software design + +### 5.1 New module: `receiver.py` + +A `Receiver(threading.Thread)` class following the existing service pattern +(`MQTT`, `Alexa`): constructed with `kwargs = {log, shutter, config}`, a +`shutdown_flag`, started from `operateShutters.ProcessCommand`. Enabled when +`RXGPIO` is present in `[General]` — no new CLI flag. + +Internal components: + +- **`CC1101` init helper** — bit-banged SPI using the project's existing GPIO + libraries (pigpio's built-in `bb_spi_*` functions on Pi 1–4, plain `lgpio` + writes on Pi 5) writing the ~50 configuration registers: 433.42 MHz carrier, + OOK/ASK, no packet engine, async serial mode routing demodulated data to + GDO0. Runs once at startup; speed is irrelevant, so software SPI is fine and + avoids requiring the hardware-SPI overlay (important for the HA add-on, §7). + The MISO line is not optional even for this one-time setup: init must prove + the radio is really there and configured — read `PARTNUM`/`VERSION`, check + the status byte returned with every transfer, read back each written + register, and abort startup loudly on any mismatch (a mis-wired SPI + otherwise degrades silently into a deaf receiver). Register values: see + Appendix A. Skipped when `RXType = raw` (plain receiver wired to `RXGPIO`). +- **Edge source** — mirrors the TX path's library split (selected by the + existing `IS_PI5` flag), so the receiver runs on the exact stack the project + already ships: + - *Pi 1–4:* `pigpio` edge callbacks (`pi.callback(RXGPIO, EITHER_EDGE)`) on + the **same pigpiod daemon the TX path already runs** — no extra footprint. + pigpiod timestamps every edge daemon-side in µs ticks, so Python + scheduling jitter does not affect decoding accuracy. + `pi.set_glitch_filter(RXGPIO, 150)` drops sub-150 µs noise glitches inside + the daemon before they ever reach Python (the shortest real pulse is + 640 µs). + - *Pi 5:* `lgpio.gpio_claim_alert` + callback with kernel timestamps (ns), + `lgpio.gpio_set_debounce_micros(…, 150)` as the glitch filter — the same + library the TX path's Pi 5 branch uses. + + A thin `EdgeSource` wrapper normalises both backends to a stream of + `(level, timestamp_µs)` events, keeping the decoder itself library-free and + unit-testable. +- **Decoder** — a small state machine fed `(level, timestamp)` events: + 1. Hunt for ≥2 hardware-sync pairs (2 560 µs ± 30 %). + 2. Expect software sync (4 550 µs high ± 30 %, then 640 µs low). + 3. Collect manchester transitions: durations classify as one half-symbol + (640 µs ± 35 %) or two (1 280 µs ± 35 %). The machine fails fast: the + first out-of-tolerance duration aborts straight back to sync hunt + (re-examining the offending edge as a candidate new sync), and a + whole-frame watchdog (~90 ms, longer than any legal frame) catches + stalls — noise must never leave the decoder waiting for its 56th bit. + Emit 56 bits. + 4. De-obfuscate, verify checksum, extract `{address, button, rollingCode}`. + + The decoder is a pure function of an edge-timestamp stream — fully unit + testable off-Pi with synthetic or recorded streams (§9). +- **Press filter**: + - *Self-echo:* frames whose address matches a key in `config.Shutters` are + the Pi's own transmissions (their state is already updated by the TX path) + → ignored. Additionally the Receiver pauses decoding while + `Shutter.sendCommand` holds its lock, so TX energy doesn't feed garbage + into the state machine. The receiver sits centimetres from the + transmitter and is fully RF-saturated during TX, so when the + transmitting flag clears the Receiver must also discard any queued edge + events and reset the decoder to sync hunt — trailing saturation + artifacts must not corrupt the first real frame heard afterwards. + - *De-dup:* remember `(address, rollingCode)` with a ~3 s TTL; the frame + repeats of a single press collapse into one press event. Repeat count is + retained on the event for future long-press features. + - *Unknown addresses:* counted and kept in a small ring buffer for the + learning UI (§5.5); logged at INFO (`Unknown remote 0x14A2C7 pressed UP`). + +### 5.2 `Shutter` refactor: share the position simulation + +Today `rise`/`lower`/`stop` interleave "send RF" with "update the position +model". Extract the model updates into internal methods so the RX path can +invoke them without transmitting: + +| New method | Extracted from | Behaviour | +|---|---|---| +| `_simulateUp(shutterId)` | `rise()` | `registerCommand('up')` + `waitAndSetFinalPosition(…, 100)` thread | +| `_simulateDown(shutterId)` | `lower()` | `registerCommand('down')` + `waitAndSetFinalPosition(…, 0)` thread | +| `_simulateStop(shutterId)` | `stop()` | elapsed-time position math incl. the intermediate-("my"-)position fallback | + +`_simulateStop` inherits the full MY-button ping-pong the motors implement: +MY while stationary travels toward the stored MY position (up if below, down if +above), MY mid-travel stops and estimates the reached position, and a further +MY resumes travel toward MY. This falls out of the existing +`lastCommandDirection` / fallback logic in `stop()` — no new code, but two +consequences for physical-remote tracking: + +- **`[ShutterIntermediatePositions]` becomes effectively mandatory** for + tracked shutters. When it is unset the model assumes a stationary MY press + "stays put", while the real motor travels to its stored favourite — with + physical remotes (where MY is the most-used button) this would be the main + source of position drift. The pairing UI (§5.5) should prompt for it. +- The M1 refactor should switch `stop()`'s elapsed-time math from + `int(round(...))` seconds with a `> 0` guard to float seconds: a MY press + within ~0.5 s of a movement command currently misclassifies as a stationary + go-to-MY press, and quick double-presses are far more likely on a physical + remote than via the network path. + +`rise()` becomes `sendCommand(…, buttonUp); _simulateUp(…)` — a pure refactor, +no behaviour change for the TX path. The Receiver calls +`shutter.recordExternalCommand(shutterId, button)`, which dispatches to the +same `_simulate*` methods. Interruption handling (a press arriving while a +previous movement simulation is still counting down) already works via the +`lastCommandTime` check in `waitAndSetFinalPosition`. + +Because `_simulate*` ends in `setPosition()`, the existing callback chain +(`mqtt.set_state` → position + open/closed/stopped topics) fires for physical +presses with **zero MQTT/HA changes**. + +### 5.3 Configuration + +```ini +[General] +# (Optional) GPIO where the RF receiver's data pin is connected. +# Presence of this key enables the receiver. +RXGPIO = 26 +# Receiver type: cc1101 (default, configured via bit-banged SPI) or raw +RXType = cc1101 +# CC1101 bit-banged SPI pins (only used when RXType = cc1101) +RXSpiSCK = 21 +RXSpiMOSI = 20 +RXSpiMISO = 19 +RXSpiCSN = 16 + +# Maps a physical remote (channel) address to the shutter(s) it controls. +# One press updates all listed shutters (group channels list several ids). +[PhysicalRemotes] +0x14A2C7 = 0x279620 +0x14A2C8 = 0x279620, 0x279621 +``` + +`MyConfig.LoadConfig` gains parsing for these keys, mirroring the existing +`[Shutters]` handling. Addresses are normalised to the same `0x%06X` string +form used as shutter ids so self-echo comparison and mapping lookups are plain +dict operations. + +### 5.4 Movement state (opening/closing) for HA + +Today `opening`/`closing` are only published from the MQTT command handler +(`mqtt.receiveMessageFromMQTT`), so physical presses would jump straight from +one resting state to another. Fix the asymmetry at the source: add a second +callback list to `Shutter` (`registerMovementCallBack`), invoked from +`_simulateUp/_simulateDown/_simulateStop` with `opening` / `closing` / +`stopped`. MQTT registers for it and drops its own inline `_publish_state` +calls. Both the software and physical paths then report movement identically. + +### 5.5 Learning mode (pairing UX) + +Users don't know their remotes' addresses, so pairing is "press a button, then +claim what was heard": + +- **v1 (config-file):** unknown presses are logged; the user copies the + address into `[PhysicalRemotes]`. +- **v2 (web UI):** a "Remotes" page backed by two endpoints — + `GET /cmd/getUnheardRemotes` returns the ring buffer of recently heard + unknown addresses `{address, lastButton, count, secondsAgo}`; + `POST /cmd/assignRemote` writes the mapping to `[PhysicalRemotes]`. The flow + mirrors the existing shutter-programming UI: open page → press the physical + remote → the address appears → tick the shutter(s) it controls → save. + +### 5.6 Position persistence across restarts + +Today positions live only in RAM (`Shutter.shutterStateList`); nothing writes +them to disk. After a reboot every shutter re-initialises to 0 and MQTT's +`on_connect` publishes 0/"closed" for all shutters — overwriting even the +retained topics HA still had. The model only re-anchors on the next full +up/down. This predates the receiver, but once physical presses are tracked the +position becomes trustworthy enough that losing it on every reboot is the +weakest link. So M1 adds persistence: + +- New config section `[ShutterPositions]`, written through the existing + atomic `MyConfig.WriteValue` — the same mechanism that already rewrites the + config on **every** command for rolling codes, so SD-card write load stays + in the same order of magnitude. +- Written only when a position *settles* (end of `waitAndSetFinalPosition`, + `stop()`, and the partial-move completions) — not on transient + opening/closing states. +- Loaded in `MyConfig.LoadConfig` and used to seed `shutterStateList` at + startup, so MQTT's reconnect publish reports the last known position + instead of 0. +- In the HA add-on the config already lives in `/data/operateShutters.conf` + (persistent volume), so restored positions survive add-on restarts, HA + updates and host reboots with no packaging change. + +Residual, accepted gap: movements made **while the Pi is off** (physical +remote during downtime or a power outage) are invisible to any design — the +restored position is a best guess until the next full up/down re-anchors the +model at 0/100. The receiver shrinks this window from "any physical press, +ever" to "physical presses during downtime only". + +## 6 What stays untouched + +- The TX path (`sendCommand`, waveforms, pigpio/lgpio TX split, rolling-code + persistence). +- The MQTT topic scheme, HA discovery payloads and the HA custom component. +- Scheduler, Alexa, web UI (until the v2 learning page). + +## 7 Proof of concept — standalone, before touching this codebase + +The POC validates hardware, frequency, range and the decoder **with zero +coupling to Pi-Somfy's code**, packaged the same way the project already +ships: as a Home Assistant add-on. It deliberately uses **the same library +stack and build recipe as the production add-on** — pigpio on Pi 1–4 / lgpio +on Pi 5, built from source exactly as in +`Home Assistant/addon/pi_somfy/Dockerfile`, with the same access grants +(`gpio: true`, `SYS_RAWIO`, `/dev/mem`, `/dev/gpiochip*`) — so what the POC +proves is the configuration the integrated feature will actually run, not a +lookalike: + +``` +addons/rts_sniffer_poc/ +├── config.yaml # same grants as the pi_somfy add-on, no ingress +├── Dockerfile # same base + pigpio/lgpio source builds as pi_somfy +├── run.sh # starts pigpiod on Pi 1–4, exactly like pi_somfy +└── sniffer.py # CC1101 init (bb_spi) + edge decoder + test TX + logging +``` + +- `sniffer.py` is self-contained (~350 lines): configure the CC1101 via + pigpio `bb_spi_*`, register edge callbacks on the data pin, decode, and log + every frame (`0x14A2C7 UP code=1337 repeats=4`) to the add-on log. + Optionally publish to MQTT topic `somfy_sniffer/event` for visibility in HA. +- **Built-in loopback transmitter:** the sniffer embeds the frame/waveform + generation from `sendCommand` (copied, not imported) and can transmit a + test frame from a dummy address on the TX GPIO every N seconds — TX and RX + through the **one shared pigpiod**, which is precisely how the integrated + feature will run on Pi 1–4. No physical remote needed for the first test. +- Installed as a **local add-on** (copy the folder to `/addons` via the + Samba/SSH add-on, then Add-on Store → ⋮ → Check for updates). +- **Stop the Pi-Somfy add-on while the POC runs** (Pi 1–4): each container + would start its own pigpiod, and two daemons contending for DMA/`/dev/mem` + is not supported. The built-in test transmitter keeps the loopback test + available regardless. The restriction disappears after integration — one + process, one daemon, both directions. +- Bit-banged SPI means **no HAOS host changes** — no `dtparam=spi=on` edit of + `config.txt`, no reboot. + +**POC success criteria** + +1. Loopback: ≥95 % of the built-in test transmissions decoded with correct + address/button/rolling code. +2. Range: every physical remote press from the farthest room is decoded + (each press repeats its frame several times, so catching any one repeat + counts). +3. Noise: zero checksum-valid false positives over 24 h of idle listening. +4. Load: sniffer CPU < 5 % on a Pi 4 in a normal RF environment. + +Only after the POC passes do we start the integration milestones — and +`sniffer.py`'s decoder moves into `receiver.py` nearly verbatim. + +## 8 Milestones + +| # | Deliverable | Depends on | +|---|---|---| +| M0 | POC sniffer add-on (§7) + decoder unit tests | CC1101 hardware | +| M1 | `receiver.py`, `Shutter` `_simulate*` refactor, `[PhysicalRemotes]`, self-echo + de-dup filters, config-file pairing, position persistence (§5.6) | M0 | +| M2 | Movement-state callback (§5.4), web UI learning page (§5.5), README hardware chapter, add-on options (`rx_gpio_pin`, SPI pins) | M1 | +| M3 | Nice-to-haves: HA event entities per physical remote (any RTS remote as automation trigger), long-press/tilt, Somfy sun/wind sensors (Soliris/Eolis speak RTS too) | M2 | + +## 9 Testing + +- **Decoder unit tests** (run anywhere, incl. CI/Windows like the existing + platform stubs): feed synthetic edge streams generated from the *same* pulse + tables `sendCommand` uses, plus recorded real-remote captures; assert + decoded frames, checksum rejection, glitch tolerance, truncated-frame reset. +- **TX→RX loopback** on-device: Pi-Somfy transmits (known frame), receiver + decodes; run in a soak loop. +- **Simulation-equivalence tests:** for each button sequence, assert + `recordExternalCommand` leaves the position model in the same state as the + equivalent `rise`/`lower`/`stop` call (minus the RF side effect). +- **Manual matrix:** short press up/down/stop, stop-while-moving, + stop-while-stationary (my-position fallback), the MY ping-pong sequence + (stationary MY → travels toward stored MY; second MY mid-travel → stops; + third MY → continues to MY — verify in both directions), group channel, + presses during a software-initiated movement, presses on unmapped remotes. + +## 10 Risks & mitigations + +| Risk | Mitigation | +|---|---| +| Frequency offset kills range | CC1101 tuned to exactly 433.42 MHz; POC range test before integration | +| 5 V receiver data pin damages Pi GPIO | CC1101 is 3.3 V native; RXB6 alternative documented as 3.3 V-powered only | +| RF noise floods the edge callback | 150 µs kernel debounce, cheap state-machine reset, checksum, address filter; POC criterion #4 | +| Receiver hears the Pi's own TX | Address self-echo filter + decode pause while `sendCommand` holds its lock | +| STOP while stationary moves to stored "my" position | Already modelled by the existing intermediate-position fallback in `stop()` — physical presses inherit it, incl. the MY ping-pong (see §5.2). Requires `[ShutterIntermediatePositions]` to match the motor's stored MY | +| Physical 5 s MY long-press reprograms the motor's stored MY, silently invalidating `[ShutterIntermediatePositions]` | Document in README; M3 detects it (high MY repeat count while the model says stationary), logs a warning and raises an HA notification that the configured intermediate position may have diverged | +| POC add-on and Pi-Somfy add-on each start a pigpiod (DMA/`/dev/mem` contention) | Never run both at once; the POC's built-in test transmitter covers loopback without Pi-Somfy. Not an issue after integration: one process, one daemon for TX+RX | +| Edge-timestamp jitter under load | Timestamps come from pigpiod (µs ticks, Pi 1–4) or the kernel (lgpio, Pi 5), not from Python; symbol tolerance ±35 % (±224 µs) vs typical jitter of tens of µs; loopback soak validates | +| Positions lost on reboot (all shutters report "closed" to HA) | Persist settled positions to `[ShutterPositions]` and restore at startup (§5.6); blinds moved while the Pi is off remain a best guess until the next full up/down | + +## 11 Future work + +- Use the CC1101 for TX as well (it is a transceiver): retires the + soldered-resonator transmitter and the pigpio waveform path entirely. +- Long-press detection (repeat count) → venetian tilt steps. +- Rolling-code plausibility tracking per physical remote to flag stuck/replayed + frames. +- Decode RTS sensors (Soliris sun/wind) as HA sensor entities. + +## Appendix A — CC1101 configuration notes (finalised in M0) + +The exact register map is settled during M0, anchored in this order: the +CC1101 datasheet formulas, TI SmartRF Studio output for 433.42 MHz OOK, and +proven open-source Somfy implementations (ESPSomfy-RTS is the reference). +Third-party register dumps — including AI-suggested ones — must be re-derived +against the datasheet before use. Cautionary example from review: a suggested +`MDMCFG4 = 0xC7` annotated as "~325 kHz bandwidth" actually computes to +≈102 kHz (`BW = 26 MHz / (8 × (4+CHANBW_M) × 2^CHANBW_E)` with E=3, M=0) — +narrow, i.e. the opposite of its stated intent of catching drifted remotes. + +Requirements the final map must satisfy (26 MHz crystal assumed): + +| Concern | Register(s) | Requirement | +|---|---|---| +| Carrier | `FREQ2/1/0` | 433.42 MHz exactly: `FREQ = round(433.42 MHz × 2^16 / 26 MHz)` | +| Modulation | `MDMCFG2` | ASK/OOK; sync-word detection disabled (raw stream) | +| Serial output | `PKTCTRL0`, `IOCFG0` | asynchronous serial mode; GDO0 routes demodulated data (0x0D) | +| RX bandwidth | `MDMCFG4` (high nibble) | wide enough for aged, drifting handheld remotes — target ~200–325 kHz; narrow only if the noise floor forces it | +| Data-rate filter | `MDMCFG4` (low nibble), `MDMCFG3` | matched to the 640 µs half-symbol stream (~1.6 kBaud chip rate) | +| OOK demod behaviour | `AGCCTRL2..0` | AGC/decision thresholds from a proven Somfy OOK profile | +| Comms sanity | `PARTNUM`, `VERSION` | read at startup; abort if SPI read-back fails (§5.1) | + +The empirical tuning loop is the M0 loopback transmitter plus a real remote +at increasing distances: adjust bandwidth and AGC until success criteria +§7 (1)–(4) pass. Register values judged "working" without that loop are not +accepted. From 877c22f8404f8ddb2952463f2acff6f700a28a44 Mon Sep 17 00:00:00 2001 From: orren5 Date: Fri, 24 Jul 2026 16:47:38 +0300 Subject: [PATCH 02/25] working poc --- .gitignore | 6 +- addons/rts_sniffer_poc/Dockerfile | 43 + addons/rts_sniffer_poc/README.md | 74 ++ addons/rts_sniffer_poc/build.yaml | 3 + addons/rts_sniffer_poc/config.yaml | 48 ++ addons/rts_sniffer_poc/patch_pigpiod.py | 39 + addons/rts_sniffer_poc/run.sh | 74 ++ addons/rts_sniffer_poc/sniffer.py | 887 ++++++++++++++++++++ addons/rts_sniffer_poc/test_sniffer.py | 245 ++++++ addons/rts_sniffer_poc/translations/en.yaml | 42 + documentation/Receiver Design.md | 21 +- 11 files changed, 1465 insertions(+), 17 deletions(-) create mode 100644 addons/rts_sniffer_poc/Dockerfile create mode 100644 addons/rts_sniffer_poc/README.md create mode 100644 addons/rts_sniffer_poc/build.yaml create mode 100644 addons/rts_sniffer_poc/config.yaml create mode 100644 addons/rts_sniffer_poc/patch_pigpiod.py create mode 100644 addons/rts_sniffer_poc/run.sh create mode 100644 addons/rts_sniffer_poc/sniffer.py create mode 100644 addons/rts_sniffer_poc/test_sniffer.py create mode 100644 addons/rts_sniffer_poc/translations/en.yaml diff --git a/.gitignore b/.gitignore index 39672af..595d462 100644 --- a/.gitignore +++ b/.gitignore @@ -1,9 +1,11 @@ /.venv -/__pycache__ +__pycache__ # User config (contains GPS coordinates, shutter names, MQTT credentials) operateShutters.conf # Log files (contain personal data and occupancy patterns) *.log -*.log.* \ No newline at end of file +*.log.* + +.vscode \ No newline at end of file diff --git a/addons/rts_sniffer_poc/Dockerfile b/addons/rts_sniffer_poc/Dockerfile new file mode 100644 index 0000000..bcd4424 --- /dev/null +++ b/addons/rts_sniffer_poc/Dockerfile @@ -0,0 +1,43 @@ +ARG BUILD_FROM +FROM ${BUILD_FROM} + +# Same base and library builds as the pi_somfy add-on (design doc §7): +# lgpio from source (for Pi 5) and pigpiod from source (for Pi 1/2/3/4). +COPY patch_pigpiod.py /tmp/patch_pigpiod.py +RUN apt-get update && apt-get install -y --no-install-recommends \ + python3 \ + python3-pip \ + python3-dev \ + procps \ + git \ + make \ + gcc \ + g++ \ + libc6-dev \ + swig \ + && cd /tmp \ + && git clone --depth 1 https://github.com/joan2937/lg.git \ + && cd lg \ + && make \ + && make install \ + && cd /tmp/lg/PY_LGPIO \ + && pip3 install --no-cache-dir --break-system-packages . \ + && cd /tmp \ + && git clone --depth 1 https://github.com/joan2937/pigpio.git \ + && python3 /tmp/patch_pigpiod.py /tmp/pigpio/pigpiod.c \ + && cd pigpio \ + && make \ + && make install \ + && cd / \ + && rm -rf /tmp/lg /tmp/pigpio /tmp/patch_pigpiod.py \ + && apt-get purge -y --auto-remove make gcc g++ libc6-dev swig python3-dev \ + && rm -rf /var/lib/apt/lists/* + +# paho-mqtt is optional at runtime (only used when mqtt_host is configured) +RUN pip3 install --no-cache-dir --break-system-packages paho-mqtt + +COPY sniffer.py /sniffer.py +COPY run.sh /run.sh +RUN chmod a+x /run.sh + +CMD ["/run.sh"] diff --git a/addons/rts_sniffer_poc/README.md b/addons/rts_sniffer_poc/README.md new file mode 100644 index 0000000..fb0bc67 --- /dev/null +++ b/addons/rts_sniffer_poc/README.md @@ -0,0 +1,74 @@ +# RTS Sniffer — proof of concept + +Standalone Home Assistant add-on that listens for Somfy RTS remote presses on +433.42 MHz and logs every decoded press: + +``` +0x14A2C7 UP code=1337 repeats=4 +``` + +This is **M0** of the receiver design (see +`documentation/Receiver Design.md` §7): it validates hardware, frequency, +range and the decoder with zero coupling to Pi-Somfy's code. Only after the +POC passes its success criteria does the decoder move into `receiver.py` for +the integrated feature. + +## Hardware + +A CC1101 transceiver module (~$3) tuned in software to exactly 433.42 MHz. +SPI is bit-banged on ordinary GPIOs and only used once at startup, so no +host `config.txt` changes or reboots are needed. Match module pins by +silkscreen **label**, not position (MOSI may be printed `SI`, MISO `SO`): + +| CC1101 pin | Signal | Default GPIO | Physical pin (Pi 4) | +|---|---|---|---| +| VCC | 3.3 V supply | — | 17 (or 1) — **never 5 V** | +| GND | ground | — | 39 | +| SCK | SPI clock | GPIO 21 | 40 | +| MOSI (SI) | SPI data → radio | GPIO 20 | 38 | +| MISO (SO) | SPI data → Pi | GPIO 19 | 35 | +| CSN | chip select | GPIO 16 | 36 | +| GDO0 | demodulated data out | GPIO 26 | 37 | +| GDO2 | — | not connected | — | +| ANT | antenna | — | 17 cm solid-core wire, **required** | + +## Install (local add-on) + +1. Copy this folder to `/addons/rts_sniffer_poc` on the HAOS host (via the + Samba or SSH add-on). +2. Settings → Add-ons → Add-on Store → ⋮ → *Check for updates*, then install + "RTS Sniffer (POC)". +3. **Stop the Pi-Somfy add-on first** (Pi 1–4): each container starts its own + pigpiod, and two daemons contending for DMA/`/dev/mem` is not supported. +4. Start the add-on and watch the log while pressing a physical remote. + +## Loopback test (no physical remote, no CC1101 required for TX) + +Set `test_tx_interval` to e.g. `30`: the add-on transmits a test frame from +dummy address `0xDEC0DE` through the existing 433.42 MHz **transmitter** +(GPIO 4 by default) every 30 s and verifies its own receiver decodes it, +logging a running success rate. TX and RX share the one pigpiod, which is +exactly how the integrated feature will run. + +## POC success criteria (design doc §7) + +1. Loopback: ≥95 % of test transmissions decoded with correct + address/button/rolling code. +2. Range: every physical remote press from the farthest room is decoded. +3. Noise: zero checksum-valid false positives over 24 h of idle listening. +4. Load: sniffer CPU < 5 % on a Pi 4 (watch the periodic `Status:` log line). + +## Decoder unit tests (run anywhere, no Pi needed) + +``` +python3 -m unittest discover addons/rts_sniffer_poc +``` + +## Notes + +- The CC1101 register map derivations live in `sniffer.py` + (`CC1101_RX_CONFIG`); AGC/TEST analog values are the SmartRF OOK starting + point and are finalised in the M0 tuning loop per Appendix A of the design. +- Startup is deliberately loud about SPI problems: VERSION is read first and + every register write is read back — a mis-wired SPI otherwise degrades + silently into a deaf receiver. diff --git a/addons/rts_sniffer_poc/build.yaml b/addons/rts_sniffer_poc/build.yaml new file mode 100644 index 0000000..cb6749b --- /dev/null +++ b/addons/rts_sniffer_poc/build.yaml @@ -0,0 +1,3 @@ +build_from: + aarch64: ghcr.io/home-assistant/aarch64-base-debian:bookworm + armv7: ghcr.io/home-assistant/armv7-base-debian:bookworm diff --git a/addons/rts_sniffer_poc/config.yaml b/addons/rts_sniffer_poc/config.yaml new file mode 100644 index 0000000..ef48c83 --- /dev/null +++ b/addons/rts_sniffer_poc/config.yaml @@ -0,0 +1,48 @@ +name: "RTS Sniffer (POC)" +description: "Proof-of-concept Somfy RTS receiver: decodes physical remote presses via a CC1101 at 433.42 MHz and logs them" +version: "0.1.0" +slug: "rts_sniffer_poc" +url: "https://github.com/Nickduino/Pi-Somfy" +arch: + - aarch64 + - armv7 +boot: manual +startup: application +init: false +full_access: true +# Same hardware access grants as the pi_somfy add-on, no ingress (§7). +# Stop the Pi-Somfy add-on while this POC runs on Pi 1-4: each container +# starts its own pigpiod and two daemons contending for DMA//dev/mem is +# not supported. Use the built-in loopback transmitter instead. +gpio: true +privileged: + - SYS_RAWIO +devices: + - /dev/mem + - /dev/vcio + - /dev/gpiochip0 + - /dev/gpiochip4 +options: + rx_gpio_pin: 26 + spi_sck: 21 + spi_mosi: 20 + spi_miso: 19 + spi_csn: 16 + test_tx_interval: 0 + tx_gpio_pin: 4 + mqtt_host: "" + mqtt_port: 1883 + mqtt_user: "" + mqtt_password: "" +schema: + rx_gpio_pin: "int(0,27)" + spi_sck: "int(0,27)" + spi_mosi: "int(0,27)" + spi_miso: "int(0,27)" + spi_csn: "int(0,27)" + test_tx_interval: "int(0,3600)" + tx_gpio_pin: "int(0,27)" + mqtt_host: "str?" + mqtt_port: "port" + mqtt_user: "str?" + mqtt_password: "password?" diff --git a/addons/rts_sniffer_poc/patch_pigpiod.py b/addons/rts_sniffer_poc/patch_pigpiod.py new file mode 100644 index 0000000..abba040 --- /dev/null +++ b/addons/rts_sniffer_poc/patch_pigpiod.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +"""Patch pigpiod's unconditional /dev/pigerr error-FIFO setup (build-time only). + +pigpiod.c's main() always does unlink()+mkfifo()+chmod()+freopen() on +/dev/pigerr, with no command-line flag to disable it (verified against +joan2937/pigpio master — -f only gates the /dev/pigpio *command* FIFO, not +this error-output companion). In a container where /dev is read-only except +for the specific device nodes this add-on's config.yaml requests, mkfifo() +silently fails and the following chmod() aborts the whole daemon. + +This add-on only talks to pigpiod over its TCP socket interface, so the +error FIFO is unneeded — replace its setup with a direct assignment to the +process's own stderr, which the container already captures as add-on logs. +""" +import re +import sys + +path = sys.argv[1] +with open(path) as f: + src = f.read() + +pattern = re.compile( + r"/\*\s*create pipe for error reporting\s*\*/.*?" + r"errFifo\s*=\s*freopen\(PI_ERRFIFO,\s*\"w\+\",\s*stderr\);", + re.DOTALL) + +patched, count = pattern.subn( + "/* patched: container /dev is read-only, skip the error FIFO */\n" + " errFifo = stderr;", + src, count=1) + +if count != 1: + sys.exit("patch_pigpiod.py: expected pigpiod.c pattern not found — " + "pigpio source may have changed upstream, update the patch") + +with open(path, "w") as f: + f.write(patched) + +print("pigpiod.c patched: error FIFO disabled (using stderr directly)") diff --git a/addons/rts_sniffer_poc/run.sh b/addons/rts_sniffer_poc/run.sh new file mode 100644 index 0000000..508a764 --- /dev/null +++ b/addons/rts_sniffer_poc/run.sh @@ -0,0 +1,74 @@ +#!/usr/bin/with-contenv bashio + +RX_GPIO_PIN=$(bashio::config 'rx_gpio_pin') +SPI_SCK=$(bashio::config 'spi_sck') +SPI_MOSI=$(bashio::config 'spi_mosi') +SPI_MISO=$(bashio::config 'spi_miso') +SPI_CSN=$(bashio::config 'spi_csn') +TEST_TX_INTERVAL=$(bashio::config 'test_tx_interval') +TX_GPIO_PIN=$(bashio::config 'tx_gpio_pin') +MQTT_HOST=$(bashio::config 'mqtt_host') +MQTT_PORT=$(bashio::config 'mqtt_port') +MQTT_USER=$(bashio::config 'mqtt_user') +MQTT_PASSWORD=$(bashio::config 'mqtt_password') + +bashio::log.info "RTS Sniffer POC starting..." +bashio::log.info "RX: GPIO ${RX_GPIO_PIN} (CC1101)" + +# Detect Pi model — Pi 5 uses lgpio (no daemon), older Pis use pigpio (needs +# pigpiod). Same detection as the pi_somfy add-on: /proc/device-tree/model may +# not be accessible inside the container; fall back to /dev/gpiochip4 (RP1 +# chip, Pi 5 only) or the CPU revision code. +IS_PI5=false +PI_MODEL="unknown" +if [ -f /proc/device-tree/model ]; then + PI_MODEL=$(tr -d '\0' < /proc/device-tree/model) +fi +bashio::log.info "Detected board: ${PI_MODEL}" +bashio::log.info "Available gpiochip devices: $(ls /dev/gpiochip* 2>/dev/null || echo 'none')" + +if echo "${PI_MODEL}" | grep -q "Pi 5"; then + IS_PI5=true +elif [ -e /dev/gpiochip4 ]; then + bashio::log.info "/dev/gpiochip4 found — assuming Pi 5" + IS_PI5=true +elif grep -q "^Revision.*[[:space:]].*[cd]0[34]17" /proc/cpuinfo 2>/dev/null; then + bashio::log.info "Pi 5 CPU revision detected in /proc/cpuinfo" + IS_PI5=true +fi + +if [ "${IS_PI5}" = true ]; then + bashio::log.info "Pi 5 detected — using lgpio (no pigpiod needed)" +else + bashio::log.info "Starting pigpiod..." + # -f disables the legacy pipe/FIFO command interface (/dev/pigpio): the + # sniffer only ever talks to pigpiod over its TCP socket interface anyway. + # Deliberately NOT passing -m (disable alerts): alerts are the mechanism + # pi.callback() uses to deliver edge notifications, which this receiver + # needs — unlike operateShutters.py's TX-only pigpiod startup, which + # never needed alerts and passes -m. + pigpiod -l -f + sleep 1 + + if ! pgrep -x pigpiod > /dev/null; then + bashio::log.error "Failed to start pigpiod!" + bashio::log.error "Is the Pi-Somfy add-on still running? Two pigpiod daemons cannot share DMA//dev/mem — stop Pi-Somfy while the POC runs." + exit 1 + fi + + bashio::log.info "pigpiod started successfully" +fi + +ARGS="--rx-gpio ${RX_GPIO_PIN}" +ARGS="${ARGS} --spi-sck ${SPI_SCK} --spi-mosi ${SPI_MOSI} --spi-miso ${SPI_MISO} --spi-csn ${SPI_CSN}" +ARGS="${ARGS} --test-tx-interval ${TEST_TX_INTERVAL} --tx-gpio ${TX_GPIO_PIN}" +if bashio::var.has_value "${MQTT_HOST}"; then + ARGS="${ARGS} --mqtt-host ${MQTT_HOST} --mqtt-port ${MQTT_PORT}" + if bashio::var.has_value "${MQTT_USER}"; then + ARGS="${ARGS} --mqtt-user ${MQTT_USER} --mqtt-password ${MQTT_PASSWORD}" + fi +fi + +bashio::log.info "Starting sniffer..." +# shellcheck disable=SC2086 +exec python3 /sniffer.py ${ARGS} diff --git a/addons/rts_sniffer_poc/sniffer.py b/addons/rts_sniffer_poc/sniffer.py new file mode 100644 index 0000000..5173368 --- /dev/null +++ b/addons/rts_sniffer_poc/sniffer.py @@ -0,0 +1,887 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +"""Standalone Somfy RTS sniffer — proof of concept (design doc §7). + +Validates hardware, frequency, range and the RTS decoder with zero coupling to +Pi-Somfy's code: + + CC1101 @ 433.42 MHz OOK, async serial out on GDO0 + -> GPIO edge timestamps (pigpio callbacks on Pi 1-4, lgpio alerts on Pi 5) + -> RTSDecoder (sync detect -> manchester -> de-XOR -> checksum) + -> press log: "0x14A2C7 UP code=1337 repeats=4" + +The frame/waveform generation is copied (not imported) from +Shutter.sendCommand in operateShutters.py, which is the authoritative +reference for the frame layout — it doubles as the built-in loopback +transmitter so the RX chain can be tested without a physical remote. + +The decoder is a pure function of a (level, timestamp_us) edge stream and has +no GPIO dependencies: run the unit tests anywhere with + python3 -m unittest discover addons/rts_sniffer_poc +""" + +import argparse +import collections +import json +import logging +import os +import signal +import sys +import threading +import time + +# GPIO / MQTT libraries are only needed on the Pi. Import lazily so the +# decoder stays unit-testable on any dev machine (design doc §9). +try: + import pigpio +except ImportError: + pigpio = None +try: + import lgpio +except ImportError: + lgpio = None +try: + import paho.mqtt.client as paho_mqtt +except ImportError: + paho_mqtt = None + + +# ── Pi model detection (copied from operateShutters.py — POC is standalone) ── +# Pi 5 uses the RP1 southbridge chip which is incompatible with pigpio. +IS_PI5 = False +LGPIO_CHIP = 4 # gpiochip number for lgpio (Pi 5): 4 on older kernels, 0 on newer +if sys.platform.startswith("linux"): + try: + with open('/proc/device-tree/model', 'r') as f: + _model = f.read() + if 'Pi 5' in _model: + IS_PI5 = True + except (FileNotFoundError, PermissionError): + pass + if not IS_PI5 and os.path.exists('/dev/gpiochip4'): + IS_PI5 = True + if not IS_PI5: + try: + with open('/proc/cpuinfo', 'r') as f: + for line in f: + if line.startswith('Revision') and any(rev in line for rev in ['c04170', 'd04170', 'c04171', 'd04171']): + IS_PI5 = True + break + except (FileNotFoundError, PermissionError): + pass + + +# ── RTS protocol constants (must match Shutter.sendCommand exactly, §3) ───── +WAKEUP_HIGH_US = 9415 +WAKEUP_LOW_US = 89565 +HW_SYNC_HALF_US = 2560 # one half of a hardware-sync pair +SW_SYNC_HIGH_US = 4550 +HALF_SYMBOL_US = 640 # manchester half-symbol +INTER_FRAME_GAP_US = 30415 +PAYLOAD_BITS = 56 + +BUTTON_STOP = 0x1 +BUTTON_UP = 0x2 +BUTTON_DOWN = 0x4 +BUTTON_PROG = 0x8 +BUTTON_NAMES = {BUTTON_STOP: "MY/STOP", BUTTON_UP: "UP", + BUTTON_DOWN: "DOWN", BUTTON_PROG: "PROG"} + + +def button_name(button): + return BUTTON_NAMES.get(button, "0x%X" % button) + + +def build_frame(address, button, rolling_code): + """Return the 7 obfuscated on-air bytes for one press. + + Copied from Shutter.sendCommand (operateShutters.py) — checksum over all + 14 nibbles, then the chained XOR obfuscation. + """ + frame = bytearray(7) + frame[0] = 0xA7 # "encryption key" + frame[1] = (button & 0xF) << 4 # button; low nibble becomes checksum + frame[2] = (rolling_code >> 8) & 0xFF # rolling code, big endian + frame[3] = rolling_code & 0xFF + frame[4] = (address >> 16) & 0xFF # remote address, 24 bit + frame[5] = (address >> 8) & 0xFF + frame[6] = address & 0xFF + + checksum = 0 + for octet in frame: + checksum = checksum ^ octet ^ (octet >> 4) + frame[1] |= checksum & 0x0F + + for i in range(1, 7): # obfuscation: running XOR chain + frame[i] ^= frame[i - 1] + return frame + + +def frame_to_pulses(frame, repetitions=1): + """On-air pulse list [(level, duration_us)] for one press. + + Same pulse table as Shutter.sendCommand: wake-up, then `repetitions` + frames (2 hardware-sync pairs on the first, 7 on repeats). Consecutive + same-level entries occur (e.g. software-sync low followed by a bit + starting low) — they merge on air; pulses_to_edges() models that. + """ + pulses = [(1, WAKEUP_HIGH_US), (0, WAKEUP_LOW_US)] + for rep in range(repetitions): + for _ in range(2 if rep == 0 else 7): # hardware synchronization + pulses.append((1, HW_SYNC_HALF_US)) + pulses.append((0, HW_SYNC_HALF_US)) + pulses.append((1, SW_SYNC_HIGH_US)) # software synchronization + pulses.append((0, HALF_SYMBOL_US)) + for i in range(PAYLOAD_BITS): # manchester payload + if (frame[i // 8] >> (7 - (i % 8))) & 1: + pulses.append((0, HALF_SYMBOL_US)) + pulses.append((1, HALF_SYMBOL_US)) + else: + pulses.append((1, HALF_SYMBOL_US)) + pulses.append((0, HALF_SYMBOL_US)) + pulses.append((0, INTER_FRAME_GAP_US)) # inter-frame gap + return pulses + + +def pulses_to_edges(pulses, start_us=0): + """Convert a pulse list to the (level, timestamp_us) edge events a GPIO + edge callback would deliver: consecutive same-level pulses merge into one, + an event fires at every level change. The line is assumed idle-low.""" + edges = [] + t = start_us + prev_level = 0 + for level, duration in pulses: + if level != prev_level: + edges.append((level, t)) + prev_level = level + t += duration + if prev_level != 0: + edges.append((0, t)) + return edges + + +RTSFrame = collections.namedtuple("RTSFrame", "address button rolling_code key") + + +class RTSDecoder(object): + """RTS frame decoder: a pure state machine fed (level, timestamp_us) edge + events (design doc §5.1). + + 1. Hunt for >=2 hardware-sync pairs (2560 us +/-30 %). + 2. Software sync high (4550 us) flips to payload collection. + 3. Payload durations classify as one half-symbol (640 us +/-35 %) or two; + the first out-of-tolerance duration aborts straight back to sync hunt, + re-examining the offending duration as a candidate new sync. + 4. De-obfuscate, verify checksum, emit {address, button, rollingCode}. + + Payload model: after the software-sync high the stream is 113 half-symbol + slots — index 0 is the 640 us sync tail (low), indices 2k+1 / 2k+2 are the + two manchester halves of bit k. Within a bit the halves always differ, so + bit k = NOT(level of half 2k+1) and the frame is complete once half 111 + is assigned (112 halves seen). That also means a 2-half-long run may never + start on an odd index — enforcing this catches invalid manchester early. + """ + + SYNC_TOL = 0.30 + SYM_TOL = 0.35 + HW_SYNC_MIN = int(HW_SYNC_HALF_US * (1 - SYNC_TOL)) # 1792 + # The +/-30 % windows of 2560 and 4550 overlap (3185..3328); split at the + # midpoint so every duration classifies unambiguously as one or the other. + HW_SW_SPLIT = (HW_SYNC_HALF_US + SW_SYNC_HIGH_US) // 2 # 3555 + SW_SYNC_MAX = int(SW_SYNC_HIGH_US * (1 + SYNC_TOL)) # 5915 + SYM_MIN = int(HALF_SYMBOL_US * (1 - SYM_TOL)) # 416 + SYM_SPLIT = (HALF_SYMBOL_US + 2 * HALF_SYMBOL_US) // 2 # 960 + SYM_MAX = int(2 * HALF_SYMBOL_US * (1 + SYM_TOL)) # 1728 + MIN_SYNC_HALVES = 4 # >= 2 hardware-sync pairs + + def __init__(self, on_frame=None): + self.on_frame = on_frame + self.frames_decoded = 0 + self.checksum_failures = 0 + self.payload_aborts = 0 + self.edge_count = 0 + self._last_ts = None + self._last_level = None + self._sync_halves = 0 + self._halves = None # None -> hunting; list -> collecting payload + + def reset(self): + self._sync_halves = 0 + self._halves = None + + def on_edge(self, level, ts_us): + """Feed one edge: the line changed to `level` at `ts_us` (monotonic).""" + if level not in (0, 1): # pigpio watchdog / lgpio timeout events + return + self.edge_count += 1 + if self._last_ts is None or level == self._last_level: + # First edge ever, or a missed edge left us out of phase: resync. + self._last_ts = ts_us + self._last_level = level + self.reset() + return + duration = ts_us - self._last_ts + ended_level = self._last_level # the level held since the previous edge + self._last_ts = ts_us + self._last_level = level + if self._halves is None: + self._hunt(ended_level, duration) + else: + self._collect(ended_level, duration) + + def _hunt(self, level, duration): + if self.HW_SYNC_MIN <= duration < self.HW_SW_SPLIT: + self._sync_halves += 1 + elif (level == 1 and self._sync_halves >= self.MIN_SYNC_HALVES + and self.HW_SW_SPLIT <= duration <= self.SW_SYNC_MAX): + self._halves = [] # software sync seen -> collect payload + self._sync_halves = 0 + else: + self._sync_halves = 0 + + def _collect(self, level, duration): + if duration < self.SYM_MIN or duration > self.SYM_MAX: + n = None + elif duration < self.SYM_SPLIT: + n = 1 + else: + n = 2 + if n is None or (n == 2 and len(self._halves) % 2 == 1): + self.payload_aborts += 1 + self.reset() + self._hunt(level, duration) # offending duration may be a new sync + return + self._halves.extend((level,) * n) + if len(self._halves) >= 2 * PAYLOAD_BITS: + self._finish_frame() + + def _finish_frame(self): + halves = self._halves + self.reset() + + recv = bytearray(7) + for i in range(PAYLOAD_BITS): + bit = halves[2 * i + 1] ^ 1 + recv[i // 8] |= bit << (7 - (i % 8)) + + plain = bytearray(recv) # de-obfuscation: plain[i] = recv[i] ^ recv[i-1] + for i in range(6, 0, -1): + plain[i] = recv[i] ^ recv[i - 1] + + checksum = 0 # XOR of all 14 nibbles must be 0 + for octet in plain: + checksum ^= octet ^ (octet >> 4) + if checksum & 0x0F: + self.checksum_failures += 1 + return + + frame = RTSFrame( + address=(plain[4] << 16) | (plain[5] << 8) | plain[6], + button=(plain[1] >> 4) & 0xF, + rolling_code=(plain[2] << 8) | plain[3], + key=plain[0]) + self.frames_decoded += 1 + if self.on_frame is not None: + self.on_frame(frame) + + +class PressTracker(object): + """Collapse the frame repeats of a single press into one press event. + + (address, rollingCode) uniquely identifies one press (§3) and is + remembered with a TTL; on_press fires on the first frame, on_press_end + fires with the final repeat count once the press goes quiet. + """ + + def __init__(self, on_press=None, on_press_end=None, + ttl=3.0, quiet=0.8, clock=time.monotonic): + self.on_press = on_press + self.on_press_end = on_press_end + self.presses = 0 + self._ttl = ttl + self._quiet = quiet + self._clock = clock + self._lock = threading.Lock() + self._current = None + + def on_frame(self, frame): + now = self._clock() + ended = None + with self._lock: + cur = self._current + if (cur is not None + and cur["key"] == (frame.address, frame.rolling_code) + and now - cur["last"] <= self._ttl): + cur["repeats"] += 1 + cur["last"] = now + return + ended = self._take_current() + self._current = {"key": (frame.address, frame.rolling_code), + "frame": frame, "repeats": 1, + "first": now, "last": now} + self.presses += 1 + self._emit_end(ended) + if self.on_press is not None: + self.on_press(frame) + + def poll(self): + """Call periodically; flushes a press once it has gone quiet.""" + now = self._clock() + with self._lock: + if self._current is None or now - self._current["last"] < self._quiet: + return + ended = self._take_current() + self._emit_end(ended) + + def _take_current(self): + cur, self._current = self._current, None + return cur + + def _emit_end(self, ended): + if ended is not None and self.on_press_end is not None: + self.on_press_end(ended["frame"], ended["repeats"]) + + +# ── CC1101 configuration via bit-banged SPI (design doc §5.1, Appendix A) ─── + +CC1101_SRES = 0x30 +CC1101_SRX = 0x34 +CC1101_SIDLE = 0x36 +CC1101_READ = 0x80 +CC1101_STATUS = 0xC0 # burst bit selects the status-register space +CC1101_REG_PARTNUM = 0x30 +CC1101_REG_VERSION = 0x31 +CC1101_REG_RSSI = 0x34 +CC1101_REG_MARCSTATE = 0x35 +CC1101_MARCSTATE_RX = 0x0D + +# Register map, 26 MHz crystal, 433.42 MHz ASK/OOK asynchronous serial data +# out on GDO0. FREQ is derived from the datasheet formula; the rest is a +# faithful port of SmartRC-CC1101-Driver-Lib's register set for ~100 kHz +# bandwidth (the library behind Elrindel/SomfyReceiver's confirmed-working +# example on this same physical module), with full LNA/DVGA gain instead of +# the reference's capped-gain AGCCTRL2 — capping gain reliably killed all +# receive activity on this specific hardware. Validated end-to-end on real +# hardware: loopback decodes at 100%, real remote presses decode correctly. +# +# MANCHESTER_EN and SYNC_MODE (in MDMCFG2) are set to match the reference +# but are, per the datasheet, packet-engine features tied to bit-clock +# recovery that asynchronous serial mode has none of — almost certainly +# don't-care bits here, matched only for completeness. +CC1101_RX_CONFIG = ( + (0x00, 0x2E, "IOCFG2 GDO2 high impedance (unused, not wired)"), + (0x02, 0x0D, "IOCFG0 GDO0 = asynchronous serial RX data"), + (0x06, 0x00, "PKTLEN unused in infinite-length async mode"), + (0x07, 0x04, "PKTCTRL1 no address check, no status append"), + (0x08, 0x32, "PKTCTRL0 asynchronous serial mode, no CRC, infinite length"), + (0x09, 0x00, "ADDR unused (no address check)"), + (0x0A, 0x00, "CHANNR channel 0, no channel hopping"), + (0x0B, 0x06, "FSCTRL1 IF = 26MHz*6/2^10 = 152 kHz"), + (0x0D, 0x10, "FREQ2 FREQ=0x10AB85 = round(433.42MHz * 2^16 / 26MHz)"), + (0x0E, 0xAB, "FREQ1 -> carrier 433.419995 MHz"), + (0x0F, 0x85, "FREQ0"), + (0x10, 0xC7, "MDMCFG4 RX BW 26MHz/(8*(4+0)*2^3) = 101.6 kHz; DRATE_E=7"), + (0x11, 0x93, "MDMCFG3 DRATE_M=0x93, paired with DRATE_E=7 above"), + (0x12, 0x3C, "MDMCFG2 DC-blocking filter on, ASK/OOK (MOD_FORMAT=011), " + "MANCHESTER_EN=1, SYNC_MODE=100 (likely don't-care in async " + "serial mode, see note above; matched for completeness)"), + (0x13, 0x02, "MDMCFG1 no FEC, minimal preamble (irrelevant in async mode)"), + (0x14, 0xF8, "MDMCFG0 channel spacing (irrelevant, no channel hopping)"), + (0x15, 0x47, "DEVIATN frequency deviation (FSK-only, irrelevant for OOK)"), + (0x18, 0x18, "MCSM0 auto-calibrate synthesizer on IDLE->RX"), + (0x19, 0x16, "FOCCFG frequency offset compensation"), + (0x1A, 0x1C, "BSCFG bit synchronization config"), + (0x1B, 0x03, "AGCCTRL2 full LNA/DVGA gain, 33 dB magnitude target — " + "capping DVGA gain reliably killed all receive activity on " + "this hardware"), + (0x1C, 0x00, "AGCCTRL1 no relative carrier-sense thresholds"), + (0x1D, 0x91, "AGCCTRL0 OOK decision boundary 8 dB above averaged noise " + "floor, 16-sample window"), + (0x21, 0x56, "FREND1 RX front end"), + (0x22, 0x11, "FREND0 OOK PA table index 1 (TX side unused in this POC)"), + (0x23, 0xE9, "FSCAL3 frequency synthesizer calibration"), + (0x24, 0x2A, "FSCAL2 same"), + (0x25, 0x00, "FSCAL1 same"), + (0x26, 0x1F, "FSCAL0 same"), + (0x29, 0x59, "FSTEST"), + (0x2C, 0x81, "TEST2 RX BW >= 325 kHz value (datasheet threshold, not " + "linear in bandwidth — still correct for our narrower filter)"), + (0x2D, 0x35, "TEST1 same threshold basis as TEST2"), + (0x2E, 0x09, "TEST0 VCO selection calibration disabled"), +) + + +class PigpioBitBangSpi(object): + """Bit-banged SPI on Pi 1-4 via pigpiod's built-in bb_spi_* (any GPIOs).""" + + def __init__(self, pi, sck, mosi, miso, csn, baud=50000): + self._pi = pi + self._csn = csn + pi.bb_spi_open(csn, miso, mosi, sck, baud, 0) # SPI mode 0, MSB first + + def xfer(self, data): + count, rx = self._pi.bb_spi_xfer(self._csn, data) + if count < 0: + raise RuntimeError("bb_spi_xfer failed with %d" % count) + return list(rx) + + def close(self): + try: + self._pi.bb_spi_close(self._csn) + except Exception: + pass + + +class LgpioBitBangSpi(object): + """Bit-banged SPI mode 0 with plain lgpio reads/writes (Pi 5). + + Speed is irrelevant — the CC1101 is configured once at startup — so a + software half-clock of ~10 us (~50 kHz) is plenty. + """ + + HALF_CLOCK_S = 0.00001 + + def __init__(self, handle, sck, mosi, miso, csn): + self._h = handle + self._sck, self._mosi, self._miso, self._csn = sck, mosi, miso, csn + lgpio.gpio_claim_output(handle, sck, 0) + lgpio.gpio_claim_output(handle, mosi, 0) + lgpio.gpio_claim_output(handle, csn, 1) + lgpio.gpio_claim_input(handle, miso) + + def xfer(self, data): + h = self._h + lgpio.gpio_write(h, self._csn, 0) + # The CC1101 drives SO low once its crystal is stable; wait briefly. + deadline = time.monotonic() + 0.01 + while lgpio.gpio_read(h, self._miso) and time.monotonic() < deadline: + time.sleep(0.0001) + rx = [] + for byte in data: + value = 0 + for bit in range(7, -1, -1): + lgpio.gpio_write(h, self._mosi, (byte >> bit) & 1) + time.sleep(self.HALF_CLOCK_S) + lgpio.gpio_write(h, self._sck, 1) + value = (value << 1) | lgpio.gpio_read(h, self._miso) + time.sleep(self.HALF_CLOCK_S) + lgpio.gpio_write(h, self._sck, 0) + rx.append(value) + lgpio.gpio_write(h, self._csn, 1) + return rx + + def close(self): + for gpio in (self._sck, self._mosi, self._miso, self._csn): + try: + lgpio.gpio_free(self._h, gpio) + except Exception: + pass + + +class CC1101(object): + """One-time CC1101 setup: 433.42 MHz OOK receive, demodulated data on GDO0. + + Init must prove the radio is really there and configured (§5.1): VERSION + is read first, every register write is read back, and the receiver state + is verified — any mismatch aborts startup loudly, because a mis-wired SPI + otherwise degrades silently into a deaf receiver. + """ + + def __init__(self, spi, log): + self._spi = spi + self._log = log + + def _strobe(self, cmd): + status = self._spi.xfer([cmd])[0] + if status & 0x80: # CHIP_RDYn must be low on every returned status byte + raise RuntimeError( + "CC1101 status byte 0x%02X reports chip not ready after strobe 0x%02X" + % (status, cmd)) + return status + + def _write_reg(self, addr, value): + self._spi.xfer([addr, value]) + + def _read_reg(self, addr): + return self._spi.xfer([addr | CC1101_READ, 0x00])[1] + + def _read_status_reg(self, addr): + return self._spi.xfer([addr | CC1101_STATUS, 0x00])[1] + + def configure(self): + self._spi.xfer([CC1101_SRES]) + time.sleep(0.01) + + partnum = self._read_status_reg(CC1101_REG_PARTNUM) + version = self._read_status_reg(CC1101_REG_VERSION) + if version in (0x00, 0xFF): + raise RuntimeError( + "CC1101 not responding (PARTNUM=0x%02X VERSION=0x%02X) — MISO stuck; " + "check wiring/power. Match module pins by silkscreen label " + "(MOSI may be printed SI, MISO SO)." % (partnum, version)) + self._log.info("CC1101 detected: PARTNUM=0x%02X VERSION=0x%02X " + "(genuine chips report 0x00/0x14; clones vary)", + partnum, version) + + for addr, value, note in CC1101_RX_CONFIG: + self._write_reg(addr, value) + mismatches = [] + for addr, value, note in CC1101_RX_CONFIG: + readback = self._read_reg(addr) + if readback != value: + mismatches.append("reg 0x%02X (%s): wrote 0x%02X read 0x%02X" + % (addr, note.split()[0], value, readback)) + if mismatches: + raise RuntimeError("CC1101 register read-back failed — mis-wired SPI? " + + "; ".join(mismatches)) + + self._strobe(CC1101_SIDLE) + time.sleep(0.001) + self._strobe(CC1101_SRX) + deadline = time.monotonic() + 0.5 + while time.monotonic() < deadline: + if self._read_status_reg(CC1101_REG_MARCSTATE) & 0x1F == CC1101_MARCSTATE_RX: + self._log.info("CC1101 configured: 433.42 MHz OOK, async serial data on GDO0") + return + time.sleep(0.01) + raise RuntimeError("CC1101 never entered RX (MARCSTATE=0x%02X)" + % self._read_status_reg(CC1101_REG_MARCSTATE)) + + def rssi_dbm(self): + raw = self._read_status_reg(CC1101_REG_RSSI) + return (raw - 256 if raw >= 128 else raw) / 2.0 - 74 + + def is_in_rx(self): + return self._read_status_reg(CC1101_REG_MARCSTATE) & 0x1F == CC1101_MARCSTATE_RX + + +# ── Edge sources: normalise both GPIO backends to (level, timestamp_us) ───── + +class PigpioEdgeSource(object): + """pigpio edge callbacks (Pi 1-4): pigpiod timestamps every edge daemon-side + in us ticks, so Python scheduling jitter does not affect decoding.""" + + def __init__(self, pi, gpio, on_edge, glitch_us=150): + self._pi = pi + self._gpio = gpio + self._on_edge = on_edge + self._prev_tick = None + self._ts = 0 + pi.set_mode(gpio, pigpio.INPUT) + pi.set_glitch_filter(gpio, glitch_us) # drop sub-150 us noise in the daemon + self._cb = pi.callback(gpio, pigpio.EITHER_EDGE, self._handle) + + def _handle(self, _gpio, level, tick): + if self._prev_tick is not None: + self._ts += pigpio.tickDiff(self._prev_tick, tick) # 32-bit wrap safe + self._prev_tick = tick + try: + self._on_edge(level, self._ts) + except Exception: + logging.getLogger("sniffer").exception("decoder error") + + def stop(self): + self._cb.cancel() + self._pi.set_glitch_filter(self._gpio, 0) + + +class LgpioEdgeSource(object): + """lgpio alerts (Pi 5): kernel timestamps in ns, debounce as glitch filter.""" + + def __init__(self, handle, gpio, on_edge, glitch_us=150): + self._h = handle + self._gpio = gpio + self._on_edge = on_edge + lgpio.gpio_claim_alert(handle, gpio, lgpio.BOTH_EDGES) + lgpio.gpio_set_debounce_micros(handle, gpio, glitch_us) + self._cb = lgpio.callback(handle, gpio, lgpio.BOTH_EDGES, self._handle) + + def _handle(self, _chip, _gpio, level, timestamp_ns): + try: + self._on_edge(level, timestamp_ns // 1000) # level 2 (watchdog) is ignored downstream + except Exception: + logging.getLogger("sniffer").exception("decoder error") + + def stop(self): + self._cb.cancel() + try: + lgpio.gpio_free(self._h, self._gpio) + except Exception: + pass + + +# ── Loopback test transmitter (waveform copied from Shutter.sendCommand) ──── + +def send_pulses_pigpio(pi, tx_gpio, pulses): + pi.wave_add_new() + pi.set_mode(tx_gpio, pigpio.OUTPUT) + wf = [] + for level, duration in pulses: + if level: + wf.append(pigpio.pulse(1 << tx_gpio, 0, duration)) + else: + wf.append(pigpio.pulse(0, 1 << tx_gpio, duration)) + pi.wave_add_generic(wf) + wid = pi.wave_create() + pi.wave_send_once(wid) + while pi.wave_tx_busy(): + time.sleep(0.005) + pi.wave_delete(wid) + + +def send_pulses_lgpio(handle, tx_gpio, pulses): + lgpio.gpio_claim_output(handle, tx_gpio) + wave = [lgpio.pulse(level, 1, duration) for level, duration in pulses] + lgpio.tx_wave(handle, tx_gpio, wave) + while lgpio.tx_busy(handle, tx_gpio, lgpio.TX_WAVE): + time.sleep(0.001) + lgpio.gpio_free(handle, tx_gpio) + + +# ── Sniffer application ────────────────────────────────────────────────────── + +class Sniffer(object): + STATUS_INTERVAL_S = 60 + TEST_TX_REPETITIONS = 2 + TEST_TX_BUTTONS = (BUTTON_UP, BUTTON_STOP, BUTTON_DOWN) + + def __init__(self, opts, log): + self.opts = opts + self.log = log + self.shutdown_flag = threading.Event() + self._pi = None + self._lgpio_handle = None + self._spi = None + self._cc1101 = None + self._edge_source = None + self._mqtt = None + self._tracker = PressTracker(on_press=self._on_press, + on_press_end=self._on_press_end) + self._decoder = RTSDecoder(on_frame=self._tracker.on_frame) + self._test_code = 0 + self._loopback_expected = None + self._loopback_sent = 0 + self._loopback_ok = 0 + + # -- setup --------------------------------------------------------------- + def start(self): + opts = self.opts + if IS_PI5: + if lgpio is None: + raise RuntimeError("lgpio module not available on this Pi 5") + global LGPIO_CHIP + last_error = None + for chip in (4, 0): + try: + self._lgpio_handle = lgpio.gpiochip_open(chip) + LGPIO_CHIP = chip + break + except Exception as e: + last_error = e + if self._lgpio_handle is None: + raise RuntimeError("lgpio: no usable gpiochip found: %s" % last_error) + self.log.info("Pi 5: lgpio on gpiochip%d", LGPIO_CHIP) + else: + if pigpio is None: + raise RuntimeError("pigpio module not available") + self._pi = pigpio.pi() + if not self._pi.connected: + raise RuntimeError("cannot connect to pigpiod — is it running?") + self.log.info("Pi 1-4: connected to pigpiod") + + if IS_PI5: + self._spi = LgpioBitBangSpi(self._lgpio_handle, opts.spi_sck, + opts.spi_mosi, opts.spi_miso, opts.spi_csn) + else: + self._spi = PigpioBitBangSpi(self._pi, opts.spi_sck, + opts.spi_mosi, opts.spi_miso, opts.spi_csn) + self._cc1101 = CC1101(self._spi, self.log) + self._cc1101.configure() + + if IS_PI5: + self._edge_source = LgpioEdgeSource(self._lgpio_handle, opts.rx_gpio, + self._decoder.on_edge) + else: + self._edge_source = PigpioEdgeSource(self._pi, opts.rx_gpio, + self._decoder.on_edge) + self.log.info("Listening on GPIO %d", opts.rx_gpio) + + if opts.mqtt_host: + self._start_mqtt() + if opts.test_tx_interval: + self.log.info("Loopback transmitter enabled: address 0x%06X on GPIO %d " + "every %d s", opts.test_address, opts.tx_gpio, + opts.test_tx_interval) + + def _start_mqtt(self): + if paho_mqtt is None: + self.log.warning("mqtt_host set but paho-mqtt is not installed — " + "MQTT publishing disabled") + return + try: + client = paho_mqtt.Client(paho_mqtt.CallbackAPIVersion.VERSION2) + except AttributeError: + client = paho_mqtt.Client() + if self.opts.mqtt_user: + client.username_pw_set(self.opts.mqtt_user, self.opts.mqtt_password) + client.connect_async(self.opts.mqtt_host, self.opts.mqtt_port) + client.loop_start() + self._mqtt = client + self.log.info("MQTT: publishing presses to somfy_sniffer/event on %s:%d", + self.opts.mqtt_host, self.opts.mqtt_port) + + # -- press handling ------------------------------------------------------ + def _on_press(self, frame): + if (self.opts.test_tx_interval + and frame.address == self.opts.test_address): + self._check_loopback(frame) + return + self.log.info("0x%06X %s code=%d", frame.address, + button_name(frame.button), frame.rolling_code) + + def _on_press_end(self, frame, repeats): + if (self.opts.test_tx_interval + and frame.address == self.opts.test_address): + return + self.log.info("0x%06X %s code=%d repeats=%d", frame.address, + button_name(frame.button), frame.rolling_code, repeats) + if self._mqtt is not None: + payload = json.dumps({"address": "0x%06X" % frame.address, + "button": button_name(frame.button), + "rolling_code": frame.rolling_code, + "repeats": repeats}) + self._mqtt.publish("somfy_sniffer/event", payload) + + def _check_loopback(self, frame): + expected = self._loopback_expected + if expected is not None and (frame.button, frame.rolling_code) == expected: + self._loopback_ok += 1 + self._loopback_expected = None + self.log.info("Loopback OK: %d/%d decoded (%.1f%%)", + self._loopback_ok, self._loopback_sent, + 100.0 * self._loopback_ok / self._loopback_sent) + else: + self.log.warning("Loopback MISMATCH: heard %s code=%d, expected %s", + button_name(frame.button), frame.rolling_code, expected) + + # -- test transmitter ---------------------------------------------------- + def _send_test_frame(self): + self._test_code += 1 + button = self.TEST_TX_BUTTONS[self._test_code % len(self.TEST_TX_BUTTONS)] + frame = build_frame(self.opts.test_address, button, self._test_code) + pulses = frame_to_pulses(frame, repetitions=self.TEST_TX_REPETITIONS) + self._loopback_expected = (button, self._test_code) + self._loopback_sent += 1 + self.log.info("Loopback TX #%d: 0x%06X %s code=%d", self._loopback_sent, + self.opts.test_address, button_name(button), self._test_code) + if IS_PI5: + send_pulses_lgpio(self._lgpio_handle, self.opts.tx_gpio, pulses) + else: + send_pulses_pigpio(self._pi, self.opts.tx_gpio, pulses) + + # -- main loop ----------------------------------------------------------- + def run(self): + now = time.monotonic() + next_status = now + self.STATUS_INTERVAL_S + next_tx = now + 5 if self.opts.test_tx_interval else None + while not self.shutdown_flag.is_set(): + time.sleep(0.2) + self._tracker.poll() + now = time.monotonic() + if next_tx is not None and now >= next_tx: + self._send_test_frame() + next_tx = now + self.opts.test_tx_interval + if now >= next_status: + self._log_status() + next_status = now + self.STATUS_INTERVAL_S + + def _log_status(self): + d = self._decoder + extra = "" + if self._cc1101 is not None: + try: + extra = " rssi=%.0f dBm rx=%s" % (self._cc1101.rssi_dbm(), + self._cc1101.is_in_rx()) + except Exception as e: + extra = " (CC1101 status read failed: %s)" % e + if self._loopback_sent: + extra += " loopback=%d/%d" % (self._loopback_ok, self._loopback_sent) + self.log.info("Status: edges=%d presses=%d frames=%d checksum_fail=%d " + "payload_aborts=%d%s", d.edge_count, self._tracker.presses, + d.frames_decoded, d.checksum_failures, d.payload_aborts, extra) + + def stop(self): + self.shutdown_flag.set() + if self._edge_source is not None: + self._edge_source.stop() + if self._spi is not None: + self._spi.close() + if self._mqtt is not None: + self._mqtt.loop_stop() + if self._pi is not None: + self._pi.stop() + if self._lgpio_handle is not None: + lgpio.gpiochip_close(self._lgpio_handle) + + +def parse_args(argv=None): + def hex_int(value): + return int(value, 0) + + p = argparse.ArgumentParser(description="Somfy RTS sniffer POC") + p.add_argument("--rx-gpio", type=int, default=26, + help="GPIO wired to the receiver data pin (CC1101 GDO0)") + p.add_argument("--spi-sck", type=int, default=21) + p.add_argument("--spi-mosi", type=int, default=20) + p.add_argument("--spi-miso", type=int, default=19) + p.add_argument("--spi-csn", type=int, default=16) + p.add_argument("--test-tx-interval", type=int, default=0, metavar="SECONDS", + help="transmit a loopback test frame every N seconds (0 = off)") + p.add_argument("--tx-gpio", type=int, default=4, + help="GPIO of the existing 433.42 MHz transmitter") + p.add_argument("--test-address", type=hex_int, default=0xDEC0DE, + help="dummy remote address used by the loopback transmitter") + p.add_argument("--mqtt-host", default="") + p.add_argument("--mqtt-port", type=int, default=1883) + p.add_argument("--mqtt-user", default="") + p.add_argument("--mqtt-password", default="") + p.add_argument("--verbose", action="store_true") + return p.parse_args(argv) + + +def main(argv=None): + opts = parse_args(argv) + logging.basicConfig(level=logging.DEBUG if opts.verbose else logging.INFO, + format="%(asctime)s %(levelname)-7s %(message)s", + stream=sys.stdout) + log = logging.getLogger("sniffer") + + if pigpio is None and lgpio is None: + log.error("Neither pigpio nor lgpio is available — this must run on a " + "Raspberry Pi. (The decoder unit tests run anywhere: " + "python3 -m unittest discover addons/rts_sniffer_poc)") + return 2 + + sniffer = Sniffer(opts, log) + + def _shutdown(signum, _frame): + log.info("Signal %d received, shutting down", signum) + sniffer.shutdown_flag.set() + + signal.signal(signal.SIGTERM, _shutdown) + signal.signal(signal.SIGINT, _shutdown) + + try: + sniffer.start() + log.info("RTS sniffer running — press a Somfy remote button") + sniffer.run() + except Exception: + log.exception("Fatal error") + return 1 + finally: + sniffer.stop() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/addons/rts_sniffer_poc/test_sniffer.py b/addons/rts_sniffer_poc/test_sniffer.py new file mode 100644 index 0000000..ee5b1e9 --- /dev/null +++ b/addons/rts_sniffer_poc/test_sniffer.py @@ -0,0 +1,245 @@ +# -*- coding: utf-8 -*- +"""Decoder unit tests (design doc §9) — run anywhere, no GPIO libraries needed: + + python3 -m unittest discover addons/rts_sniffer_poc + +Synthetic edge streams are generated from the same pulse tables +Shutter.sendCommand uses (via frame_to_pulses), then fed to the decoder as +(level, timestamp_us) events exactly as the GPIO edge callbacks would deliver +them. +""" + +import os +import random +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from sniffer import (BUTTON_DOWN, BUTTON_PROG, BUTTON_STOP, BUTTON_UP, + PAYLOAD_BITS, PressTracker, RTSDecoder, build_frame, + frame_to_pulses, pulses_to_edges) + +# Known-good vectors generated with the *original* Shutter.sendCommand math +# from operateShutters.py — they pin the on-air encoding independently of +# build_frame, so an encoder/decoder bug pair cannot cancel out silently. +KNOWN_FRAMES = [ + (0x279620, BUTTON_UP, 1337, bytes([0xA7, 0x8F, 0x8A, 0xB3, 0x94, 0x02, 0x22])), + (0x14A2C7, BUTTON_STOP, 42, bytes([0xA7, 0xB5, 0xB5, 0x9F, 0x8B, 0x29, 0xEE])), + (0xDEC0DE, BUTTON_DOWN, 65535, bytes([0xA7, 0xE2, 0x1D, 0xE2, 0x3C, 0xFC, 0x22])), +] + + +def decode_edges(edges): + """Feed an edge stream to a fresh decoder, return (frames, decoder).""" + frames = [] + decoder = RTSDecoder(on_frame=frames.append) + for level, ts in edges: + decoder.on_edge(level, ts) + return frames, decoder + + +def press_edges(address, button, code, repetitions=1, start_us=0): + pulses = frame_to_pulses(build_frame(address, button, code), repetitions) + return pulses_to_edges(pulses, start_us=start_us) + + +class BuildFrameTests(unittest.TestCase): + + def test_matches_sendcommand_vectors(self): + for address, button, code, expected in KNOWN_FRAMES: + self.assertEqual(bytes(build_frame(address, button, code)), expected) + + def test_deobfuscated_checksum_is_zero(self): + # Spec §3: XOR of all 14 nibbles of the de-obfuscated frame must be 0. + recv = build_frame(0x123456, BUTTON_PROG, 4242) + plain = bytearray(recv) + for i in range(6, 0, -1): + plain[i] = recv[i] ^ recv[i - 1] + checksum = 0 + for octet in plain: + checksum ^= octet ^ (octet >> 4) + self.assertEqual(checksum & 0x0F, 0) + + +class DecoderRoundTripTests(unittest.TestCase): + + def assert_decodes(self, edges, address, button, code, expected_frames=1): + frames, decoder = decode_edges(edges) + self.assertEqual(len(frames), expected_frames) + for frame in frames: + self.assertEqual(frame.address, address) + self.assertEqual(frame.button, button) + self.assertEqual(frame.rolling_code, code) + self.assertEqual(decoder.checksum_failures, 0) + return frames + + def test_single_frame_every_button(self): + for button in (BUTTON_STOP, BUTTON_UP, BUTTON_DOWN, BUTTON_PROG): + self.assert_decodes(press_edges(0x279620, button, 1337), + 0x279620, button, 1337) + + def test_repeats_all_decoded(self): + # 1 initial frame (2 hw-sync pairs) + 4 repeats (7 pairs each) + self.assert_decodes(press_edges(0x14A2C7, BUTTON_DOWN, 500, repetitions=5), + 0x14A2C7, BUTTON_DOWN, 500, expected_frames=5) + + def test_field_extremes(self): + for address, code in [(0x000001, 0), (0xFFFFFF, 0xFFFF), + (0x800000, 1), (0x14A2C7, 0x8000)]: + self.assert_decodes(press_edges(address, BUTTON_UP, code), + address, BUTTON_UP, code) + + def test_known_vector_on_air(self): + # End to end from the sendCommand-pinned bytes, bypassing build_frame. + for address, button, code, raw in KNOWN_FRAMES: + edges = pulses_to_edges(frame_to_pulses(bytearray(raw))) + self.assert_decodes(edges, address, button, code) + + def test_two_presses_back_to_back(self): + first = press_edges(0x279620, BUTTON_UP, 10) + second = press_edges(0x279620, BUTTON_STOP, 11, + start_us=first[-1][1] + 200000) + frames, _ = decode_edges(first + second) + self.assertEqual([(f.button, f.rolling_code) for f in frames], + [(BUTTON_UP, 10), (BUTTON_STOP, 11)]) + + +class DecoderToleranceTests(unittest.TestCase): + """Aged remote crystals drift and edges jitter; the decoder allows ±30 % + on syncs and ±35 % on half-symbols (design doc §5.1).""" + + def scaled_edges(self, scale): + pulses = frame_to_pulses(build_frame(0x279620, BUTTON_UP, 77)) + return pulses_to_edges([(lvl, int(dur * scale)) for lvl, dur in pulses]) + + def test_fast_remote_clock(self): + frames, _ = decode_edges(self.scaled_edges(0.80)) + self.assertEqual(len(frames), 1) + + def test_slow_remote_clock(self): + frames, _ = decode_edges(self.scaled_edges(1.25)) + self.assertEqual(len(frames), 1) + + def test_edge_jitter(self): + # ±100 us per edge keeps every duration inside tolerance; typical + # daemon/kernel timestamp jitter is tens of us (design doc §10). + for seed in range(10): + rng = random.Random(seed) + edges = [(lvl, ts + rng.randint(-100, 100)) + for lvl, ts in press_edges(0x14A2C7, BUTTON_DOWN, 900)] + frames, _ = decode_edges(edges) + self.assertEqual(len(frames), 1, "jitter seed %d failed" % seed) + + +class DecoderRobustnessTests(unittest.TestCase): + + def test_corrupted_bit_rejected_by_checksum(self): + # A flip in the last on-air byte changes exactly one de-obfuscated + # byte, which the nibble-XOR checksum catches. (A mid-frame flip + # would flip the same bit in two consecutive de-obfuscated bytes and + # cancel out of the checksum — an inherent limit of the 4-bit RTS + # checksum, not a decoder bug.) + frame = build_frame(0x279620, BUTTON_UP, 1337) + frame[6] ^= 0x10 + frames, decoder = decode_edges(pulses_to_edges(frame_to_pulses(frame))) + self.assertEqual(frames, []) + self.assertEqual(decoder.checksum_failures, 1) + + def test_truncated_frame_then_valid_frame(self): + full = press_edges(0x279620, BUTTON_UP, 1) + truncated = full[:40] # cut mid-payload, then 100 ms of silence + valid = press_edges(0x279620, BUTTON_DOWN, 2, + start_us=truncated[-1][1] + 100000) + frames, decoder = decode_edges(truncated + valid) + self.assertEqual(len(frames), 1) + self.assertEqual(frames[0].button, BUTTON_DOWN) + self.assertEqual(frames[0].rolling_code, 2) + self.assertGreaterEqual(decoder.payload_aborts, 1) + + def test_noise_produces_no_frames(self): + rng = random.Random(1234) + edges, t, level = [], 0, 0 + for _ in range(5000): + level ^= 1 + t += rng.randint(200, 3500) + edges.append((level, t)) + frames, decoder = decode_edges(edges) + self.assertEqual(frames, []) + self.assertEqual(decoder.checksum_failures, 0) + + def test_frame_decoded_after_noise(self): + rng = random.Random(99) + edges, t, level = [], 0, 0 + for _ in range(500): + level ^= 1 + t += rng.randint(200, 3500) + edges.append((level, t)) + if level == 1: # let the line settle low before the frame + t += 5000 + edges.append((0, t)) + edges += press_edges(0x14A2C7, BUTTON_STOP, 33, start_us=t + 50000) + frames, _ = decode_edges(edges) + self.assertEqual(len(frames), 1) + self.assertEqual(frames[0].address, 0x14A2C7) + + def test_single_hw_sync_pair_rejected(self): + # Fewer than 2 hardware-sync pairs must not enter payload collection. + pulses = frame_to_pulses(build_frame(0x279620, BUTTON_UP, 5)) + del pulses[2:4] # drop one of the two initial sync pairs + frames, _ = decode_edges(pulses_to_edges(pulses)) + self.assertEqual(frames, []) + + def test_duplicate_level_edge_resyncs(self): + edges = press_edges(0x279620, BUTTON_UP, 7) + decoder_frames = [] + decoder = RTSDecoder(on_frame=decoder_frames.append) + decoder.on_edge(0, 0) # spurious same-level event before the press + decoder.on_edge(0, 1000) + for level, ts in [(lvl, ts + 10000) for lvl, ts in edges]: + decoder.on_edge(level, ts) + self.assertEqual(len(decoder_frames), 1) + + +class PressTrackerTests(unittest.TestCase): + + def setUp(self): + self.now = 0.0 + self.presses = [] + self.ended = [] + self.tracker = PressTracker( + on_press=lambda f: self.presses.append(f), + on_press_end=lambda f, r: self.ended.append((f, r)), + clock=lambda: self.now) + self.decoder = RTSDecoder(on_frame=self.tracker.on_frame) + + def feed_press(self, address, button, code, repetitions, start_us=0): + for level, ts in press_edges(address, button, code, repetitions, start_us): + self.decoder.on_edge(level, ts) + + def test_repeats_collapse_into_one_press(self): + self.feed_press(0x14A2C7, BUTTON_UP, 1337, repetitions=4) + self.assertEqual(len(self.presses), 1) + self.assertEqual(self.presses[0].rolling_code, 1337) + self.now = 1.0 # quiet period elapsed + self.tracker.poll() + self.assertEqual(len(self.ended), 1) + self.assertEqual(self.ended[0][1], 4) # repeat count retained (§5.1) + + def test_new_rolling_code_is_new_press(self): + self.feed_press(0x14A2C7, BUTTON_UP, 1, repetitions=2) + self.now = 0.5 + self.feed_press(0x14A2C7, BUTTON_UP, 2, repetitions=2, start_us=10**7) + self.assertEqual(len(self.presses), 2) + self.assertEqual(len(self.ended), 1) # first press flushed by second + self.assertEqual(self.ended[0][1], 2) + + def test_ttl_expiry_splits_presses(self): + self.feed_press(0x14A2C7, BUTTON_STOP, 9, repetitions=1) + self.now = 5.0 # past the 3 s TTL: same key counts as new press + self.feed_press(0x14A2C7, BUTTON_STOP, 9, repetitions=1, start_us=5 * 10**6) + self.assertEqual(len(self.presses), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/addons/rts_sniffer_poc/translations/en.yaml b/addons/rts_sniffer_poc/translations/en.yaml new file mode 100644 index 0000000..3fbfe1b --- /dev/null +++ b/addons/rts_sniffer_poc/translations/en.yaml @@ -0,0 +1,42 @@ +--- +configuration: + rx_gpio_pin: + name: Receiver data GPIO + description: >- + GPIO wired to the receiver's demodulated data output (CC1101 GDO0). + Default is GPIO 26 (physical pin 37). + spi_sck: + name: CC1101 SPI clock GPIO + description: Bit-banged SPI SCK. Default GPIO 21 (physical pin 40). + spi_mosi: + name: CC1101 SPI MOSI GPIO + description: Pi -> radio data (module pin may be printed SI). Default GPIO 20 (pin 38). + spi_miso: + name: CC1101 SPI MISO GPIO + description: >- + Radio -> Pi data (module pin may be printed SO). Required — startup + verifies every register write by reading it back. Default GPIO 19 (pin 35). + spi_csn: + name: CC1101 SPI chip-select GPIO + description: Default GPIO 16 (physical pin 36). + test_tx_interval: + name: Loopback test interval (seconds) + description: >- + When non-zero, transmit a test frame from a dummy remote address on the + TX GPIO every N seconds and verify it decodes — validates the whole RX + chain without a physical remote. 0 disables. + tx_gpio_pin: + name: Transmitter GPIO + description: GPIO of the existing 433.42 MHz transmitter (loopback test only). Default 4. + mqtt_host: + name: MQTT host (optional) + description: When set, every press is also published to somfy_sniffer/event. + mqtt_port: + name: MQTT port + description: Default 1883. + mqtt_user: + name: MQTT username (optional) + description: Leave empty for anonymous access. + mqtt_password: + name: MQTT password (optional) + description: Only used when a username is set. diff --git a/documentation/Receiver Design.md b/documentation/Receiver Design.md index 7c0f65f..9b74fa6 100644 --- a/documentation/Receiver Design.md +++ b/documentation/Receiver Design.md @@ -95,7 +95,7 @@ trick does not exist for receivers:** commodity part. Running one unmodified (centered 500 kHz off) loses most of its sensitivity. -### 4.2 Recommended: CC1101 transceiver module (~$3) +### 4.2 CC1101 transceiver module (~$3) The CC1101 is tuned **in software**: we write its frequency registers once at startup and set it to OOK receive with *asynchronous serial output*, after @@ -104,7 +104,7 @@ receiver — demodulated 0/1 that we timestamp with GPIO edge callbacks, matchin the project's existing GPIO style. No soldering, no rare parts, exact 433.42 MHz, 3.3 V native (Pi-safe). -Wiring (SPI is only used for one-time configuration; see §5.2): +Wiring (SPI is only used for one-time configuration; see §5.1): | CC1101 pin | Signal | Default GPIO | Physical pin (Pi 4) | Note | |---|---|---|---|---| @@ -135,14 +135,7 @@ match by label, not by position (MOSI may be printed `SI`, MISO `SO`). └──────┴──────┘ ``` -### 4.3 Alternative: RXB6/RXB8 superheterodyne - -For purists who accept reduced range or manage to source a 433.42 crystal: -power at **3.3 V** (the data pin follows VCC; 5 V would damage the Pi), data -pin → `RXGPIO`. Works with the same software; only §5.2 (CC1101 init) is -skipped. Not the recommended path. - -### 4.4 Antenna +### 4.3 Antenna Today's antenna-less transmitter reaches the whole house because the *blind motors* have good factory antennas — the weak TX signal is compensated by good @@ -174,7 +167,7 @@ Internal components: the status byte returned with every transfer, read back each written register, and abort startup loudly on any mismatch (a mis-wired SPI otherwise degrades silently into a deaf receiver). Register values: see - Appendix A. Skipped when `RXType = raw` (plain receiver wired to `RXGPIO`). + Appendix A. - **Edge source** — mirrors the TX path's library split (selected by the existing `IS_PI5` flag), so the receiver runs on the exact stack the project already ships: @@ -270,9 +263,7 @@ presses with **zero MQTT/HA changes**. # (Optional) GPIO where the RF receiver's data pin is connected. # Presence of this key enables the receiver. RXGPIO = 26 -# Receiver type: cc1101 (default, configured via bit-banged SPI) or raw -RXType = cc1101 -# CC1101 bit-banged SPI pins (only used when RXType = cc1101) +# CC1101 bit-banged SPI pins RXSpiSCK = 21 RXSpiMOSI = 20 RXSpiMISO = 19 @@ -434,12 +425,12 @@ Only after the POC passes do we start the integration milestones — and | Risk | Mitigation | |---|---| | Frequency offset kills range | CC1101 tuned to exactly 433.42 MHz; POC range test before integration | -| 5 V receiver data pin damages Pi GPIO | CC1101 is 3.3 V native; RXB6 alternative documented as 3.3 V-powered only | | RF noise floods the edge callback | 150 µs kernel debounce, cheap state-machine reset, checksum, address filter; POC criterion #4 | | Receiver hears the Pi's own TX | Address self-echo filter + decode pause while `sendCommand` holds its lock | | STOP while stationary moves to stored "my" position | Already modelled by the existing intermediate-position fallback in `stop()` — physical presses inherit it, incl. the MY ping-pong (see §5.2). Requires `[ShutterIntermediatePositions]` to match the motor's stored MY | | Physical 5 s MY long-press reprograms the motor's stored MY, silently invalidating `[ShutterIntermediatePositions]` | Document in README; M3 detects it (high MY repeat count while the model says stationary), logs a warning and raises an HA notification that the configured intermediate position may have diverged | | POC add-on and Pi-Somfy add-on each start a pigpiod (DMA/`/dev/mem` contention) | Never run both at once; the POC's built-in test transmitter covers loopback without Pi-Somfy. Not an issue after integration: one process, one daemon for TX+RX | +| `operateShutters.py`'s `startGPIO()` starts pigpiod with `-l -m` — `-m` disables alerts, the mechanism `pi.callback()` uses for edge notifications. TX never needed alerts so this went unnoticed; M1's receiver adding an edge callback onto this same daemon would silently never fire (confirmed during the M0 POC: edges stayed at 0 across every CC1101 register configuration tried until `-m` was removed, at which point loopback immediately hit 100 %) | Drop `-m` from `operateShutters.py`'s pigpiod startup when M1 wires the receiver into the shared daemon | | Edge-timestamp jitter under load | Timestamps come from pigpiod (µs ticks, Pi 1–4) or the kernel (lgpio, Pi 5), not from Python; symbol tolerance ±35 % (±224 µs) vs typical jitter of tens of µs; loopback soak validates | | Positions lost on reboot (all shutters report "closed" to HA) | Persist settled positions to `[ShutterPositions]` and restore at startup (§5.6); blinds moved while the Pi is off remain a best guess until the next full up/down | From 672dcb30eb3161a0e1f4eaf2f1bb545d37a96be2 Mon Sep 17 00:00:00 2001 From: orren5 Date: Sat, 25 Jul 2026 13:11:51 +0300 Subject: [PATCH 03/25] update design --- documentation/Receiver Design.md | 253 ++++++++++++++++++++++++------- 1 file changed, 199 insertions(+), 54 deletions(-) diff --git a/documentation/Receiver Design.md b/documentation/Receiver Design.md index 9b74fa6..01de0c8 100644 --- a/documentation/Receiver Design.md +++ b/documentation/Receiver Design.md @@ -1,6 +1,8 @@ # Design: RTS Receiver — Track Physical Remote Presses -Status: **Draft / proposal** +Status: **M0 core pipeline validated on real hardware** — loopback 100 %, a +real remote decodes correctly (§7, §8); the farthest-room range test and the +24 h noise/CPU soak are still open but non-blocking. M1+ still draft/proposal Target: Pi-Somfy v3.2+ ## 1 Motivation @@ -46,7 +48,7 @@ Physical remote press **Non-goals (v1)** - Decoding encrypted Somfy io-homecontrol devices (different protocol entirely). -- Tilt/long-press handling (future work, see §10). +- Tilt/long-press handling (future work, see §11). - Replacing the TX hardware with the receiver's transceiver (future work). ## 3 Protocol background @@ -173,8 +175,12 @@ Internal components: already ships: - *Pi 1–4:* `pigpio` edge callbacks (`pi.callback(RXGPIO, EITHER_EDGE)`) on the **same pigpiod daemon the TX path already runs** — no extra footprint. - pigpiod timestamps every edge daemon-side in µs ticks, so Python - scheduling jitter does not affect decoding accuracy. + That daemon must be started **without** `-m` (disable alerts) — `-m` is + what `operateShutters.py`'s `startGPIO()` passes today, harmlessly for a + TX-only daemon, but it silently kills `pi.callback()` delivery entirely + (confirmed the hard way during the M0 POC; see §10). pigpiod timestamps + every edge daemon-side in µs ticks, so Python scheduling jitter does not + affect decoding accuracy. `pi.set_glitch_filter(RXGPIO, 150)` drops sub-150 µs noise glitches inside the daemon before they ever reach Python (the shortest real pulse is 640 µs). @@ -298,12 +304,33 @@ claim what was heard": - **v1 (config-file):** unknown presses are logged; the user copies the address into `[PhysicalRemotes]`. -- **v2 (web UI):** a "Remotes" page backed by two endpoints — - `GET /cmd/getUnheardRemotes` returns the ring buffer of recently heard - unknown addresses `{address, lastButton, count, secondsAgo}`; - `POST /cmd/assignRemote` writes the mapping to `[PhysicalRemotes]`. The flow - mirrors the existing shutter-programming UI: open page → press the physical - remote → the address appears → tick the shutter(s) it controls → save. +- **v2 (web UI), concretely, in the existing settings page** (`html/index.html` + — no separate page): the shutter/motor setup UI is already an accordion of + sections (`#collapseOne` location, `#collapseTwo` "Add/Remove Shutter", + a schedule section) inside one `#accordion`. Physical-remote pairing gets + its own sibling section, **"Physical Remotes"**, added the same way: + - A table of currently-paired remotes (address → shutter name(s)), with an + unassign action — mirrors the existing `#shutters` table + (`webserver.py`'s `addShutter`/`editShutter`/`deleteShutter` pattern). + - Below it, a "recently heard" list from `GET /cmd/getUnheardRemotes` + (`{address, lastButton, count, secondsAgo}`), each row with an "Assign" + button opening a small modal. That modal reuses the **existing + multi-select shutter picker** already in `index.html` (`