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/Home Assistant/addon/pi_somfy/DOCS.md b/Home Assistant/addon/pi_somfy/DOCS.md index 4bb3dd8..f9e1fde 100644 --- a/Home Assistant/addon/pi_somfy/DOCS.md +++ b/Home Assistant/addon/pi_somfy/DOCS.md @@ -19,9 +19,24 @@ This add-on runs [Pi-Somfy](https://github.com/Nickduino/Pi-Somfy) directly on y ## Configuration -| Option | Default | Description | -|------------|---------|------------------------------------------------| -| `gpio_pin` | `4` | GPIO pin number for the 433.42 MHz transmitter | +| Option | Default | Description | +|---------------|---------|---------------------------------------------------------------------------| +| `gpio_pin` | `4` | GPIO pin number for the 433.42 MHz transmitter | +| `rx_gpio_pin` | (none) | GPIO wired to a CC1101 receiver's data output. Leave blank to disable the physical-remote receiver entirely. | +| `spi_sck` | `21` | CC1101 bit-banged SPI clock GPIO (only used when `rx_gpio_pin` is set) | +| `spi_mosi` | `20` | CC1101 bit-banged SPI MOSI GPIO (only used when `rx_gpio_pin` is set) | +| `spi_miso` | `19` | CC1101 bit-banged SPI MISO GPIO (only used when `rx_gpio_pin` is set) | +| `spi_csn` | `16` | CC1101 bit-banged SPI chip-select GPIO (only used when `rx_gpio_pin` is set) | + +### Physical remote receiver (optional) + +Setting `rx_gpio_pin` enables listening for physical Somfy RTS remote button +presses via a CC1101 receiver module, so a physical remote and the app stay +in sync. Pairing a physical remote to a shutter is not yet exposed in the +web UI — add it manually to +`[PhysicalRemotes]` in `/data/operateShutters.conf` (map the remote's +address to a comma-separated list of shutterId(s), same format as the +`[Shutters]` section). ## Web UI diff --git a/Home Assistant/addon/pi_somfy/Dockerfile b/Home Assistant/addon/pi_somfy/Dockerfile index ddb20d0..28e0c71 100644 --- a/Home Assistant/addon/pi_somfy/Dockerfile +++ b/Home Assistant/addon/pi_somfy/Dockerfile @@ -3,6 +3,7 @@ FROM ${BUILD_FROM} # Install system dependencies (Debian base) # Build 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 \ @@ -24,11 +25,12 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ && 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 \ + && 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/* diff --git a/Home Assistant/addon/pi_somfy/config.yaml b/Home Assistant/addon/pi_somfy/config.yaml index 681ec42..b08c084 100644 --- a/Home Assistant/addon/pi_somfy/config.yaml +++ b/Home Assistant/addon/pi_somfy/config.yaml @@ -20,6 +20,7 @@ privileged: - SYS_RAWIO devices: - /dev/mem + - /dev/vcio - /dev/gpiochip0 - /dev/gpiochip4 map: @@ -32,7 +33,16 @@ ports_description: watchdog: "http://[HOST]:[PORT:80]/cmd/getConfig" options: gpio_pin: 4 + spi_sck: 21 + spi_mosi: 20 + spi_miso: 19 + spi_csn: 16 schema: gpio_pin: "int(0,27)" + 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)" discovery: - pi_somfy diff --git a/Home Assistant/addon/pi_somfy/patch_pigpiod.py b/Home Assistant/addon/pi_somfy/patch_pigpiod.py new file mode 100644 index 0000000..abba040 --- /dev/null +++ b/Home Assistant/addon/pi_somfy/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/Home Assistant/addon/pi_somfy/run.sh b/Home Assistant/addon/pi_somfy/run.sh index f3354c3..6f4e03b 100644 --- a/Home Assistant/addon/pi_somfy/run.sh +++ b/Home Assistant/addon/pi_somfy/run.sh @@ -20,6 +20,30 @@ else sed -i "/^\[General\]/a TXGPIO = ${GPIO_PIN}" "${CONFIG_FILE}" fi +# RX receiver (physical Somfy remotes) is optional — only write RXGPIO/RXSpi* +# into the config file if the user set rx_gpio_pin, so operateShutters.py's +# `config.RXGPIO is not None` gate stays unset (receiver disabled) otherwise. +if bashio::config.has_value 'rx_gpio_pin'; then + 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') + bashio::log.info "RX receiver enabled: GPIO ${RX_GPIO_PIN} (CC1101)" + + for entry in "RXGPIO:${RX_GPIO_PIN}" "RXSpiSCK:${SPI_SCK}" "RXSpiMOSI:${SPI_MOSI}" "RXSpiMISO:${SPI_MISO}" "RXSpiCSN:${SPI_CSN}"; do + key="${entry%%:*}" + value="${entry#*:}" + if grep -q "^${key}" "${CONFIG_FILE}"; then + sed -i "s/^${key}.*/${key} = ${value}/" "${CONFIG_FILE}" + else + sed -i "/^\[General\]/a ${key} = ${value}" "${CONFIG_FILE}" + fi + done +else + bashio::log.info "RX receiver disabled (set rx_gpio_pin in add-on options to enable)" +fi + # Ensure log location exists and is writable sed -i "s|^LogLocation.*|LogLocation = /data/|" "${CONFIG_FILE}" @@ -50,7 +74,10 @@ if [ "${IS_PI5}" = true ]; then bashio::log.info "Pi 5 detected — using lgpio (no pigpiod needed)" else bashio::log.info "Starting pigpiod..." - pigpiod -l -m + # Deliberately not passing -m (disable alerts): -m silently prevents + # pi.callback() from ever delivering edge notifications, which the RX + # receiver needs when rx_gpio_pin is set. + pigpiod -l sleep 1 if ! pgrep -x pigpiod > /dev/null; then diff --git a/Home Assistant/custom_components/pi_somfy/cover.py b/Home Assistant/custom_components/pi_somfy/cover.py index 7505dcb..77dc1a2 100644 --- a/Home Assistant/custom_components/pi_somfy/cover.py +++ b/Home Assistant/custom_components/pi_somfy/cover.py @@ -100,10 +100,22 @@ def is_closing(self) -> bool: @callback def _handle_coordinator_update(self) -> None: - """Handle updated data from the coordinator.""" - pos = self.current_cover_position - if pos is not None and (pos <= 0 or pos >= 100): - self._moving = None + """Handle updated data from the coordinator. + + Reads Pi-Somfy's own movement-state signal (populated by any trigger + source — the app, a physical remote, or the web UI — not just + commands issued through Home Assistant) rather than only clearing + the optimistic local flag once position settles. This overwrites + the optimistic set from async_open_cover/async_close_cover/ + async_set_cover_position with the server-authoritative value as soon + as the next poll lands, which also happens to confirm HA-issued + commands almost immediately since the coordinator refreshes right + after sending one. + """ + shutter = self.coordinator.data.get(self._shutter_id) + if shutter is not None: + state = shutter.get("movementState") + self._moving = state if state in ("opening", "closing") else None self.async_write_ha_state() async def async_open_cover(self, **kwargs: Any) -> None: diff --git a/README.md b/README.md index 5a99586..c24735b 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,45 @@ OK. now this all should look like this. Note that some of the pictures are a bit ![Pi Connection](documentation/Connection.jpg)
![RF Transmitter Connection](documentation/Sender.jpg)
+### 2.1 Receiver (optional) — tracking physical remotes + +Everything above covers sending commands. Optionally, Pi-Somfy can also +*listen* for presses on your existing physical Somfy remotes: a small +receiver module decodes the same RTS radio protocol your remotes already +speak, so pressing one updates Pi-Somfy's tracked shutter position (and +Home Assistant, via MQTT or the custom component) immediately — not just +commands issued through the app itself. + +This needs a second, separate piece of 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** | + +To enable it, set `RXGPIO` (and, if wired to non-default pins, +`RXSpiSCK`/`RXSpiMOSI`/`RXSpiMISO`/`RXSpiCSN`) in `operateShutters.conf`'s +`[General]` section — leave `RXGPIO` unset/commented out to run without a +receiver, exactly as before. If you're running the Home Assistant add-on, +the same options are exposed directly in its configuration page as +`rx_gpio_pin`/`spi_sck`/`spi_mosi`/`spi_miso`/`spi_csn`. + +Once enabled, pair a physical remote to a shutter from the web UI's +"Physical Remotes" section (see §5 below): press a button on the remote, +find it listed under "Recently Heard", and assign it to one or more +shutters. + ## 3 Software If you are new to using a Raspberry Pi and Linux please refer to other sources for coming up to speed with the environment. Having a base knowledge will go a long way. This [site](https://www.raspberrypi.org/help/) is a great place to start if you are new to these topics. @@ -206,6 +245,8 @@ Click the "Add" button, select the name for your shutter (this is also the name 1. Finally, it's time to program your shutters schedule. To do so, use the "Scheduled Operations" menu.
![Screenshot](documentation/p3.png)
+1. If you've wired up the optional CC1101 receiver (§2.1), pair your physical remotes using the "Physical Remotes" menu. Press a button on the remote, find it listed under "Recently Heard", click the assign icon, and pick which shutter(s) it should control. + ## 6 Alexa Integration diff --git a/config.py b/config.py index 8973050..cc90f21 100644 --- a/config.py +++ b/config.py @@ -132,6 +132,16 @@ def __init__(self, filename = None, section = None, log = None): self.ShuttersByName = {} self.Schedule = {} self.Password = "" + self.PhysicalRemotes = {} + self.ShutterPositions = {} + # Optional (unlike TXGPIO, not guaranteed present in defaultConfig.conf) + # — must default to None so `is not None` checks don't raise + # AttributeError on a config file that omits them. + self.RXGPIO = None + self.RXSpiSCK = None + self.RXSpiMOSI = None + self.RXSpiMISO = None + self.RXSpiCSN = None try: self.config = RawConfigParser(strict=False) @@ -150,7 +160,8 @@ def __init__(self, filename = None, section = None, log = None): # -------------------- MyConfig::LoadConfig----------------------------------- def LoadConfig(self): - parameters = {'LogLocation': str, 'Latitude': float, 'Longitude': float, 'SendRepeat': int, 'UseHttps': bool, 'HTTPPort': int, 'HTTPSPort': int, 'TXGPIO': int, 'RTS_Address': str, "Password": str} + parameters = {'LogLocation': str, 'Latitude': float, 'Longitude': float, 'SendRepeat': int, 'UseHttps': bool, 'HTTPPort': int, 'HTTPSPort': int, 'TXGPIO': int, 'RTS_Address': str, "Password": str, + 'RXGPIO': int, 'RXSpiSCK': int, 'RXSpiMOSI': int, 'RXSpiMISO': int, 'RXSpiCSN': int} for key, type in parameters.items(): try: @@ -201,7 +212,23 @@ def LoadConfig(self): except Exception as e1: self.LogErrorLine("Missing config file or config file entries in Section Scheduler for key "+key+": " + str(e1)) return False - + + remotes = self.GetList(section="PhysicalRemotes") + for key, value in remotes: + try: + self.PhysicalRemotes[key] = [s.strip() for s in value.split(",")] + except Exception as e1: + self.LogErrorLine("Missing config file or config file entries in Section PhysicalRemotes for key "+key+": " + str(e1)) + return False + + positions = self.GetList(section="ShutterPositions") + for key, value in positions: + try: + self.ShutterPositions[key] = int(value) + except Exception as e1: + self.LogErrorLine("Missing config file or config file entries in Section ShutterPositions for key "+key+": " + str(e1)) + return False + return True #---------------------MyConfig::setLocation--------------------------------- @@ -314,7 +341,14 @@ def WriteValue(self, Entry, Value, section = None): last_data_line = i if not in_target_section: - raise Exception("NOT ABLE TO FIND SECTION:" + sect) + # Auto-create the missing section rather than failing: + # fresh installs get every section from defaultConfig.conf, + # but a pre-M1 config file being upgraded in place may be + # missing newer sections (e.g. [PhysicalRemotes], + # [ShutterPositions]) entirely. + lines.append("") + lines.append("[" + sect + "]") + section_header_line = len(lines) - 1 if key_line >= 0: # Replace existing key @@ -346,3 +380,70 @@ def WriteValue(self, Entry, Value, section = None): except Exception as e1: self.LogError("Error in WriteValue: " + str(e1)) return False + + #---------------------MyConfig::RemoveValue---------------------------------- + # Deletes a single key from a section — WriteValue can only replace/insert + # a key, never remove one, which a real "unassign" (e.g. [PhysicalRemotes]) + # needs, unlike [Shutters]'s soft-delete-via-flag convention which doesn't + # fit a plain "key = value" row shape. Mirrors WriteValue's exact + # scan-then-atomic-rewrite pattern. Idempotent: removing an already-absent + # key (or a section that doesn't exist at all) is not an error. + def RemoveValue(self, Entry, section = None): + + sect = section if section is not None else self.Section + + try: + with self.CriticalLock: + with open(self.FileName, 'r') as f: + lines = f.read().splitlines() + + in_target_section = False + key_line = -1 + + for i, line in enumerate(lines): + m_sect = self._RE_SECTION.match(line) + if m_sect: + if in_target_section: + break # reached next section, stop + if m_sect.group(1).strip().lower() == sect.lower(): + in_target_section = True + continue + + if in_target_section: + m_kv = self._RE_KEY_VALUE.match(line) + if m_kv and m_kv.group(2).strip() == Entry: + key_line = i + break + + if not in_target_section or key_line < 0: + return True # nothing to remove — idempotent, not an error + + del lines[key_line] + content = "\n".join(lines) + "\n" + + # Atomic write: write to temp file, then replace + tmp = self.FileName + ".tmp" + with open(tmp, 'w') as f: + f.write(content) + f.flush() + os.fsync(f.fileno()) + if sys.version_info[0] >= 3: + os.replace(tmp, self.FileName) + else: + if os.path.exists(self.FileName): + os.remove(self.FileName) + os.rename(tmp, self.FileName) + + # RawConfigParser.read() only merges in additions/updates, it + # never drops an option that disappeared from the file — the + # removed key would otherwise stay stale in the cached view. + try: + self.config.remove_option(sect, Entry) + except Exception: + pass + self.config.read(self.FileName) + return True + + except Exception as e1: + self.LogError("Error in RemoveValue: " + str(e1)) + return False diff --git a/defaultConfig.conf b/defaultConfig.conf index bb0599b..fe7c046 100644 --- a/defaultConfig.conf +++ b/defaultConfig.conf @@ -49,6 +49,21 @@ HTTPSPort = 443 # each instance is set to a different value to avoid possible conflicts RTS_Address = 0x279620 +# (Optional) GPIO wired to the RTS receiver's demodulated data output +# (CC1101 GDO0). When set, Pi-Somfy listens for physical Somfy remote +# button presses and updates the tracked shutter position accordingly +# Leave commented out to disable the receiver entirely. +# Example: GPIO 26 (physical pin 37) +# RXGPIO = 26 + +# (Optional) Bit-banged SPI pins used to configure the CC1101 receiver +# module. Only used when RXGPIO is set above. Defaults shown below; change +# only if wired differently. +#RXSpiSCK = 21 +#RXSpiMOSI = 20 +#RXSpiMISO = 19 +#RXSpiCSN = 16 + ################################################################### ################################################################### ## LIST OF ALL SHUTTERS REGISTERED @@ -131,4 +146,29 @@ EnableDiscovery = true # [Scheduler] +################################################################### +################################################################### +## PHYSICAL REMOTE PAIRING (used by the RTS receiver, when RXGPIO is set) +################################################################### +################################################################### +# +# Maps a physical Somfy remote's address (24bit identifier, same format as +# [Shutters]) to the shutterId(s) it should control. The config value is a +# comma delimited list of shutterId(s) (the addresses used as keys in +# [Shutters]) — list more than one to fan a single remote channel out to a +# group of shutters. +# +# Example: a remote at address 0x27ABCD operating two shutters: +#0x27ABCD = 0x279620, 0x14A2C7 +# +[PhysicalRemotes] + + +# Last known position (0-100 %) of each shutter, keyed by the same +# shutterId used in [Shutters]. Automatically maintained by Pi-Somfy as +# shutters move — like [ShutterRollingCodes], you normally won't edit this +# by hand. Seeds the tracked position on startup so it isn't lost across a +# restart. +# +[ShutterPositions] diff --git a/html/index.html b/html/index.html index 02027d7..ca88972 100755 --- a/html/index.html +++ b/html/index.html @@ -228,6 +228,58 @@

+
+

+ +

+
+
+
+ + + + + + + + + + +
Remote AddressShutter(s)Actions +
+ + +
+
+
+
+
+
+
+
Recently Heard (Unpaired)
+
+
+ + + + + + + + + + +
Remote AddressLast ButtonActions +
+ +
+
+
+
+
+
@@ -377,10 +435,12 @@

    -
  1. Move the shutter to the My position: . +
  2. Move the shutter to the My position: .
  3. Press to clear the My position.
    4. +
  4. Also clear it in Pi-Somfy's own tracking: +
@@ -393,5 +453,24 @@

+ \ No newline at end of file diff --git a/html/operateShutters.css b/html/operateShutters.css index f3d65b9..7d329e5 100644 --- a/html/operateShutters.css +++ b/html/operateShutters.css @@ -562,6 +562,11 @@ button[class*="press-button-"]:hover { background-color: #dee2e6; box-shadow: va } .shutterRemote:hover { box-shadow: var(--shadow-md); transform: translateY(-2px); } .shutterRemote .name { text-align: center; font-size: 14px; font-weight: 700; color: #666; margin-bottom: 4px; } +.shutterRemote .position { text-align: center; font-size: 20px; font-weight: 700; color: var(--color-dark); margin-bottom: 2px; } +.shutterRemote .state { text-align: center; font-size: 12px; font-weight: 600; color: var(--color-primary); min-height: 1.2em; margin-bottom: 2px; } +.shutterRemote .myPosition { text-align: center; font-size: 11px; font-weight: 600; color: #999; min-height: 1.2em; margin-bottom: 2px; } +.shutterRemote .myPosition.unset { color: var(--color-warning); } +.shutterRemote.moving { box-shadow: 0 0 0 2px var(--color-primary); } .shutterRemote a { display: block; margin-left: auto; margin-right: auto; } .shutterRemote a svg { width: 60px; height: auto; } diff --git a/html/operateShutters.js b/html/operateShutters.js index 650dbbd..68fba02 100755 --- a/html/operateShutters.js +++ b/html/operateShutters.js @@ -626,6 +626,9 @@ var marker; var config; var modalCallerIconElement; var configShutter; +var assignRemoteAddress; +var unheardRemotesInterval = null; +var manualOperationInterval = null; const buttonStop = 0x1; const buttonUp = 0x2; @@ -661,12 +664,19 @@ function GetStartupInfo(initMap) config = result; setupTableShutters(); setupTableSchedule(); + setupTableRemotes(); if (config.Longitude == 0) { $('#collapseOne').collapse('show'); } else if (Object.keys(config.Shutters).length == 0){ - $('.accordion-collapse.show').collapse('toggle'); + $('.accordion-collapse.show').collapse('toggle'); $('#collapseTwo').collapse('show'); } + // Manual Operation is expanded by default (no shown.collapse + // fires for content already open at load), so start polling + // here once rather than only from the collapse event. + if (manualOperationInterval === null && $('#collapseFour').hasClass('show')) { + manualOperationInterval = setInterval(refreshMovementStates, 3000); + } $(".loader").removeClass("is-active"); }); } @@ -856,6 +866,21 @@ function pressButtons(id, buttons, longPress, confirmMessage) { }, "json"); } +function setIntermediatePosition(id, position) { + var url = baseurl.concat("setIntermediatePosition"); + $.post(url, + {shutter: id, position: position}, + function(result, status){ + if ((status=="success") && (result.status == "OK")) { + config.ShutterIntermediatePositions[id] = result.intermediatePosition; + updateManualOperationState(); + BootstrapDialog.show({type: BootstrapDialog.TYPE_INFO, title: 'Information', message:'My position saved.'}); + } else { + BootstrapDialog.show({type: BootstrapDialog.TYPE_DANGER, title: 'Error', message:'Received Error from Server: '+result.message}); + } + }, "json"); +} + function deleteShutter(id) { var url = baseurl.concat("deleteShutter"); $.post( url, @@ -915,6 +940,32 @@ function deleteSchedule(id) { }, "json"); } +function assignRemote(address, shutterIds) { + var url = baseurl.concat("assignRemote"); + $.post( url, + {address: address, "shutterIds[]": shutterIds}, + function(result, status){ + if ((status=="success") && (result.status == "OK")) { + GetStartupInfo(false); + } else { + BootstrapDialog.show({type: BootstrapDialog.TYPE_DANGER, title: 'Error', message:'Received Error from Server: '+result.message, onhide: function(){GetStartupInfo(false);}}); + } + }, "json"); +} + +function unassignRemote(address) { + var url = baseurl.concat("unassignRemote"); + $.post( url, + {address: address}, + function(result, status){ + if ((status=="success") && (result.status == "OK")) { + GetStartupInfo(false); + } else { + BootstrapDialog.show({type: BootstrapDialog.TYPE_DANGER, title: 'Error', message:'Received Error from Server: '+result.message, onhide: function(){GetStartupInfo(false);}}); + } + }, "json"); +} + function setupTableShutters () { @@ -929,17 +980,52 @@ function setupTableShutters () { ''; $("#shutters").append(row); - var cell = '
' + + var cell = '
' + '
'+config.Shutters[shutter]+'
' + + '
' + + '
' + + '
' + '' + '' + '' + '
'; $("#action_manual").append(cell); }); - + $("#shuttersCount").text($("#shutters").find('tr').length-1); - + updateManualOperationState(); +} + +// Reflects config.MovementStates ('opening'/'closing'/'stopped'/null), +// config.Positions (0-100), and config.ShutterIntermediatePositions (the +// stored "My" position, 0-100 or null if unset) onto each Manual Operation +// card — kept current by refreshMovementStates(), polled while that section +// is visible, so a physical remote press (or a command from elsewhere) +// shows up here too, not just commands sent from this page. +function updateManualOperationState() { + if (!config.MovementStates || !config.Positions) { return; } + Object.keys(config.MovementStates).forEach(function(shutter) { + var state = config.MovementStates[shutter]; + var card = $('.shutterRemote[name="'+shutter+'"]'); + var label = (state == "opening") ? "Opening..." : (state == "closing") ? "Closing..." : ""; + card.find(".state").text(label); + card.find(".position").text(config.Positions[shutter] + "%"); + card.toggleClass("moving", state == "opening" || state == "closing"); + var myPosition = config.ShutterIntermediatePositions ? config.ShutterIntermediatePositions[shutter] : null; + var myPositionUnset = (myPosition === null || myPosition === undefined); + card.find(".myPosition").text(myPositionUnset ? "⚠ My position not set" : "My: " + myPosition + "%") + .toggleClass("unset", myPositionUnset); + }); +} + +function refreshMovementStates() { + $.getJSON(baseurl.concat("getConfig"), {}, function(result, status){ + if (!(status=="success")) { return; } + config.MovementStates = result.MovementStates; + config.Positions = result.Positions; + config.ShutterIntermediatePositions = result.ShutterIntermediatePositions; + updateManualOperationState(); + }); } function setupTableSchedule () { @@ -1059,6 +1145,61 @@ function setupTableSchedule () { $('.editbox.in').show(); } +function setupTableRemotes () { + $("#remotes").find("tr:gt(0)").remove(); + var addresses = Object.keys(config.PhysicalRemotes); + addresses.sort().forEach(function(address) { + var shutterNames = config.PhysicalRemotes[address].map(function(shutterId) { + return config.Shutters[shutterId] || shutterId; + }).join(", "); + var row = '' + + ''+address+'' + + ''+shutterNames+'' + + '' + $("#action_remotes").html() + '' + + ''; + $("#remotes").append(row); + }); + $("#remotesCount").text($("#remotes").find('tr').length-1); + initTooltips(); +} + +// Polled only while the "Physical Remotes" section is expanded (started/ +// stopped by the shown.collapse/hidden.collapse handlers in setupListeners). +function refreshUnheardRemotes() { + var url = baseurl.concat("getUnheardRemotes"); + $.getJSON(url, {}, function(result, status){ + if (!(status=="success") || result.status != "OK") { return; } + $("#unheardRemotes").find("tr:gt(0)").remove(); + var pairedAddresses = Object.keys(config.PhysicalRemotes); + result.remotes.forEach(function(remote) { + if (pairedAddresses.includes(remote.address)) { return; } // paired since last poll + var row = '' + + ''+remote.address+'' + + ''+remote.buttonName+'' + + '' + $("#action_unheard_remotes").html() + '' + + ''; + $("#unheardRemotes").append(row); + }); + initTooltips(); + }); +} + +function openAssignRemoteModal(address, currentShutterIds) { + assignRemoteAddress = address; + $("#assign-remote-address").text(address); + var select = $("#assignRemoteShutters"); + if (select.data('multiselect')) { + select.multiselect('destroy'); + } + select.empty(); + var shutterIds = Object.keys(config.Shutters); + shutterIds.sort(function(a, b) { return config.Shutters[a].toLowerCase() > config.Shutters[b].toLowerCase()}).forEach(function(shutterId) { + var selected = currentShutterIds.includes(shutterId) ? " selected" : ""; + select.append(''); + }); + select.multiselect({dropUp: true, maxHeight: 200, includeSelectAllOption: true, buttonWidth: '130px', nonSelectedText: 'None', allSelectedText: 'All', numberDisplayed: 1, nSelectedText: 'Selected', buttonClass: 'btn btn-secondary'}); + $('#assign-remote').modal('show'); +} function clockDelayValUpdate(obj) { if ($(obj).val() != parseInt($(obj).val())){ @@ -1366,9 +1507,33 @@ function setupListeners() { // Edit row on configure button click $(document).on("click", ".configureShutters", function(){ configShutter = $(this).parents("tr").attr('name'); + var current = config.ShutterIntermediatePositions[configShutter]; + $('#intermediatePositionInput').val((current === null || current === undefined) ? "" : current); $('#configure-shutter').modal('show'); }); + // Reassign an already-paired remote + $(document).on("click", ".editRemote", function(){ + var address = $(this).parents("tr").attr('name'); + openAssignRemoteModal(address, config.PhysicalRemotes[address] || []); + }); + + // Assign a remote heard under "Recently Heard" + $(document).on("click", ".assignRemoteBtn", function(){ + var address = $(this).parents("tr").attr('name'); + openAssignRemoteModal(address, []); + }); + + $('#assign-remote-save').on("click", function(){ + var shutterIds = $("#assignRemoteShutters").val() || []; + if (shutterIds.length == 0) { + BootstrapDialog.show({type: BootstrapDialog.TYPE_DANGER, title: 'Error', message:'Select at least one shutter.'}); + return; + } + assignRemote(assignRemoteAddress, shutterIds); + $('#assign-remote').modal('hide'); + }); + // Edit row on edit button click $(document).on("click", ".editSchedule", function(){ $(this).parents("tr").find('.editbox').toggle(); @@ -1397,6 +1562,8 @@ function setupListeners() { deleteShutter(rowId); } else if (tableId == "schedule") { deleteSchedule(rowId); + } else if (tableId == "remotes") { + unassignRemote(rowId); } } $('#confirm-delete').modal('hide'); @@ -1423,6 +1590,27 @@ function setupListeners() { } }) + $('#collapseFour').on('shown.collapse', function () { + if (manualOperationInterval === null) { + refreshMovementStates(); + manualOperationInterval = setInterval(refreshMovementStates, 3000); + } + }) + + $('#collapseFour').on('hidden.collapse', function () { + clearInterval(manualOperationInterval); + manualOperationInterval = null; + }) + + $('#collapseFive').on('shown.collapse', function () { + refreshUnheardRemotes(); + unheardRemotesInterval = setInterval(refreshUnheardRemotes, 3000); + }) + + $('#collapseFive').on('hidden.collapse', function () { + clearInterval(unheardRemotesInterval); + }) + $('#program-new-shutter-ok').on("click", function(){ // We are good, the hide.modal event will take care of refreshing the main window }); @@ -1470,7 +1658,16 @@ function setupListeners() { $(document).on("click", '.press-button-prog-long', function(){ pressButtons(configShutter, buttonProg, true); }); - + + $('#save-intermediate-position').on("click", function(){ + setIntermediatePosition(configShutter, $('#intermediatePositionInput').val()); + }); + + $('#clear-intermediate-position').on("click", function(){ + $('#intermediatePositionInput').val(""); + setIntermediatePosition(configShutter, ""); + }); + // Shutter Commands $(document).on("click", ".up", function(){ diff --git a/mqtt.py b/mqtt.py index f1b5b46..19c2bf6 100644 --- a/mqtt.py +++ b/mqtt.py @@ -193,10 +193,8 @@ def receiveMessageFromMQTT(self, client, userdata, message): if action == "command": if msg == "OPEN": - self._publish_state(shutter_id, "opening") self.shutter.rise(shutter_id) elif msg == "CLOSE": - self._publish_state(shutter_id, "closing") self.shutter.lower(shutter_id) elif msg == "STOP": self.shutter.stop(shutter_id) @@ -207,16 +205,12 @@ def receiveMessageFromMQTT(self, client, userdata, message): target = int(msg) current = self.shutter.getPosition(shutter_id) if target >= 100: - self._publish_state(shutter_id, "opening") self.shutter.rise(shutter_id) elif target <= 0: - self._publish_state(shutter_id, "closing") self.shutter.lower(shutter_id) elif target > current: - self._publish_state(shutter_id, "opening") self.shutter.risePartial(shutter_id, target) elif target < current: - self._publish_state(shutter_id, "closing") self.shutter.lowerPartial(shutter_id, target) else: @@ -296,6 +290,15 @@ def set_state(self, shutter_id, level): else: self._publish_state(shutter_id, "stopped") + def set_movement_state(self, shutter_id, state): + """Callback invoked when a shutter starts moving or stops — fired + identically for software (TX) and physical-remote-triggered + movement. Complements set_state's position-derived open/closed/ + stopped signal with the transient opening/closing state during + travel.""" + self.LogInfo("Shutter " + shutter_id + " movement state: " + state) + self._publish_state(shutter_id, state) + def run(self): self.connected_flag = False self.LogInfo("Starting MQTT thread") @@ -319,6 +322,7 @@ def run(self): self.t.on_message = self.receiveMessageFromMQTT self.t.on_disconnect = self.on_disconnect self.shutter.registerCallBack(self.set_state) + self.shutter.registerMovementCallBack(self.set_movement_state) # Initial connection with retry error = 0 diff --git a/operateShutters.py b/operateShutters.py index 663fd2e..7a192ea 100755 --- a/operateShutters.py +++ b/operateShutters.py @@ -129,6 +129,7 @@ def registerCommand(self, commandDirection): def __init__(self, log = None, config = None): super(Shutter, self).__init__() self.lock = threading.Lock() + self.transmitting = threading.Event() if log is not None: self.log = log if config is not None: @@ -140,9 +141,19 @@ def __init__(self, log = None, config = None): self.TXGPIO=4 # 433.42 MHz emitter on GPIO 4 self.frame = bytearray(7) self.callback = [] + self.movementCallback = [] + self.movementStateList = {} self.shutterStateList = {} self.shutterStateLock = threading.Lock() + # Seed persisted positions, only for shutters + # that both still exist in [Shutters] and have a settled position + # recorded in [ShutterPositions], so getShutterState's lazy-init + # fallback only ever applies to shutters that have never settled. + for shutterId, position in self.config.ShutterPositions.items(): + if shutterId in self.config.Shutters: + self.shutterStateList[shutterId] = self.ShutterState(position) + def getShutterState(self, shutterId, initialPosition = None): with self.shutterStateLock: if shutterId not in self.shutterStateList: @@ -153,10 +164,47 @@ def getPosition(self, shutterId): state = self.getShutterState(shutterId, 0) return state.position + # Live-interpolated position while a move is in progress, for UI display + # only. Mirrors _simulateStop's elapsed-time/direction math (the same + # "how far did it get" estimate a MY-press stop would compute) but never + # mutates state or settles a position — internal logic that decides + # partial-move direction etc. keeps using getPosition()'s last-settled + # value, since an in-flight estimate isn't precise enough to act on. + def getDisplayPosition(self, shutterId): + state = self.getShutterState(shutterId, 0) + movementState = self.movementStateList.get(shutterId) + if movementState not in ('opening', 'closing'): + return state.position + + secondsSinceLastCommand = time.monotonic() - state.lastCommandTime + if secondsSinceLastCommand <= 0: + return state.position + + if movementState == 'opening': + duration = self.config.Shutters[shutterId]['durationUp'] + if duration <= 0 or secondsSinceLastCommand >= duration: + return 100 # elapsed time already covers the full travel — + # report the target, not the stale pre-move + # position, even if settling hasn't fired yet + durationPercentage = secondsSinceLastCommand / duration * 100 + estimated = state.position + durationPercentage if state.position > 0 else durationPercentage + return min(100, int(round(estimated))) + else: + duration = self.config.Shutters[shutterId]['durationDown'] + if duration <= 0 or secondsSinceLastCommand >= duration: + return 0 + durationPercentage = secondsSinceLastCommand / duration * 100 + estimated = state.position - durationPercentage if state.position < 100 else 100 - durationPercentage + return max(0, int(round(estimated))) + def setPosition(self, shutterId, newPosition): state = self.getShutterState(shutterId) with self.shutterStateLock: state.position = newPosition + # Every call site of setPosition() is already a "position settled" + # moment (end of waitAndSetFinalPosition, stop()'s immediate branch, + # the partial-move completions). + self.config.WriteValue(shutterId, str(newPosition), section="ShutterPositions") for function in self.callback: function(shutterId, newPosition) @@ -170,16 +218,28 @@ def waitAndSetFinalPosition(self, shutterId, timeToWait, newPosition): # Only set new position if registerCommand has not been called in between if state.lastCommandTime == oldLastCommandTime: self.LogDebug("["+self.config.Shutters[shutterId]['name']+"] Set new final position: " + str(newPosition)) + # A full/partial move that runs out its timer uninterrupted has + # genuinely stopped — fire this before setPosition() so, as + # elsewhere, the position callback's more precise open/closed + # inference has the last word. Without this, 'opening'/'closing' + # would stay stuck forever once a move completes naturally, + # since nothing else clears it for this (non-explicit-stop) path. + self._fireMovement(shutterId, 'stopped') self.setPosition(shutterId, newPosition) else: self.LogDebug("["+self.config.Shutters[shutterId]['name']+"] Discard final position. Position is now: " + str(state.position)) def lower(self, shutterId): - state = self.getShutterState(shutterId, 100) - self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Going down") self.sendCommand(shutterId, self.buttonDown, self.config.SendRepeat) + self._simulateDown(shutterId) + + # Position-model update for a "down" command, shared by the TX path + # (lower(), above) and the physical-remote path (recordExternalCommand). + def _simulateDown(self, shutterId): + state = self.getShutterState(shutterId, 100) state.registerCommand('down') + self._fireMovement(shutterId, 'closing') # wait and set final position only if not interrupted in between timeToWait = state.position/100*self.config.Shutters[shutterId]['durationDown'] @@ -192,18 +252,25 @@ def lowerPartial(self, shutterId, percentage): self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Going down") self.sendCommand(shutterId, self.buttonDown, self.config.SendRepeat) state.registerCommand('down') + self._fireMovement(shutterId, 'closing') time.sleep((state.position-percentage)/100*self.config.Shutters[shutterId]['durationDown']) self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Stop at partial position requested") self.sendCommand(shutterId, self.buttonStop, self.config.SendRepeat) + self._fireMovement(shutterId, 'stopped') self.setPosition(shutterId, percentage) def rise(self, shutterId): - state = self.getShutterState(shutterId, 0) - self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Going up") self.sendCommand(shutterId, self.buttonUp, self.config.SendRepeat) + self._simulateUp(shutterId) + + # Position-model update for an "up" command, shared by the TX path + # (rise(), above) and the physical-remote path (recordExternalCommand). + def _simulateUp(self, shutterId): + state = self.getShutterState(shutterId, 0) state.registerCommand('up') + self._fireMovement(shutterId, 'opening') # wait and set final position only if not interrupted in between timeToWait = (100-state.position)/100*self.config.Shutters[shutterId]['durationUp'] @@ -216,51 +283,67 @@ def risePartial(self, shutterId, percentage): self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Going up") self.sendCommand(shutterId, self.buttonUp, self.config.SendRepeat) state.registerCommand('up') + self._fireMovement(shutterId, 'opening') time.sleep((percentage-state.position)/100*self.config.Shutters[shutterId]['durationUp']) self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Stop at partial position requested") self.sendCommand(shutterId, self.buttonStop, self.config.SendRepeat) + self._fireMovement(shutterId, 'stopped') self.setPosition(shutterId, percentage) def stop(self, shutterId): - state = self.getShutterState(shutterId, 50) - self.LogInfo("["+self.config.Shutters[shutterId]['name']+"] Stopping") self.sendCommand(shutterId, self.buttonStop, self.config.SendRepeat) + self._simulateStop(shutterId) + + # Position-model update for a "stop"/"my" command, shared by the TX path + # (stop(), above) and the physical-remote path (recordExternalCommand). + # Inherits the full MY-button ping-pong the motors implement from the + # elapsed-time/fallback logic below — no special-casing needed for + # physical presses. + def _simulateStop(self, shutterId): + state = self.getShutterState(shutterId, 50) self.LogDebug("["+shutterId+"] Previous position: " + str(state.position)) - secondsSinceLastCommand = int(round(time.monotonic() - state.lastCommandTime)) - self.LogDebug("["+shutterId+"] Seconds since last command: " + str(secondsSinceLastCommand)) # Compute position based on time elapsed since last command & command direction setupDurationDown = self.config.Shutters[shutterId]['durationDown'] setupDurationUp = self.config.Shutters[shutterId]['durationUp'] + # Whether the shutter is genuinely moving right now comes from + # movementStateList (set by _simulateUp/_simulateDown, and cleared + # back to 'stopped' the instant ANY move settles — full, partial, or + # a previous MY-position correction), not from comparing elapsed + # time against the shutter's *full* 0-100 duration. That old + # elapsed-time heuristic misclassified a MY press shortly after a + # completed partial/MY move (which settles in less than the full + # duration) as "still mid-travel", computing a bogus interrupted + # position instead of correctly falling through to the MY-position + # logic below — the shutter was already stationary the whole time. + movementState = self.movementStateList.get(shutterId) + self.LogDebug("["+shutterId+"] Movement state: " + str(movementState)) + fallback = False - if state.lastCommandDirection == 'up': - if secondsSinceLastCommand > 0 and secondsSinceLastCommand < setupDurationUp: - durationPercentage = int(round(secondsSinceLastCommand/setupDurationUp * 100)) + if movementState == 'opening': + secondsSinceLastCommand = time.monotonic() - state.lastCommandTime + self.LogDebug("["+shutterId+"] Seconds since last command: " + str(round(secondsSinceLastCommand, 2))) + durationPercentage = int(round(secondsSinceLastCommand/setupDurationUp * 100)) if setupDurationUp > 0 else 100 self.LogDebug("["+shutterId+"] Up duration percentage: " + str(durationPercentage) + ", State position: "+ str(state.position)) if state.position > 0: # after rise from previous position newPosition = min (100 , state.position + durationPercentage) else: # after rise from fully closed newPosition = durationPercentage - else: #fallback - self.LogWarn("["+shutterId+"] Too much time since up command.") - fallback = True - elif state.lastCommandDirection == 'down': - if secondsSinceLastCommand > 0 and secondsSinceLastCommand < setupDurationDown: - durationPercentage = int(round(secondsSinceLastCommand/setupDurationDown * 100)) + elif movementState == 'closing': + secondsSinceLastCommand = time.monotonic() - state.lastCommandTime + self.LogDebug("["+shutterId+"] Seconds since last command: " + str(round(secondsSinceLastCommand, 2))) + durationPercentage = int(round(secondsSinceLastCommand/setupDurationDown * 100)) if setupDurationDown > 0 else 100 self.LogDebug("["+shutterId+"] Down duration percentage: " + str(durationPercentage) + ", State position: "+ str(state.position)) if state.position < 100: # after lower from previous position newPosition = max (0 , state.position - durationPercentage) else: # after down from fully opened newPosition = 100 - durationPercentage - else: #fallback - self.LogWarn("["+shutterId+"] Too much time since down command.") - fallback = True - else: # consecutive stops - self.LogWarn("["+shutterId+"] Stop pressed while stationary.") + else: # already stopped (fully, partially, or from a previous MY correction) + self.LogInfo("["+shutterId+"] Stop pressed while stationary.") fallback = True if fallback == True: # Let's assume it will end on the intermediate position ! If it exists ! @@ -272,14 +355,24 @@ def stop(self, shutterId): self.LogInfo("["+shutterId+"] Motor expected to move to intermediate position "+str(intermediatePosition)) if state.position > intermediatePosition: state.registerCommand('down') + self._fireMovement(shutterId, 'closing') timeToWait = abs(state.position - intermediatePosition) / 100*self.config.Shutters[shutterId]['durationDown'] else: state.registerCommand('up') + self._fireMovement(shutterId, 'opening') timeToWait = abs(state.position - intermediatePosition) / 100*self.config.Shutters[shutterId]['durationUp'] # wait and set final intermediate position only if not interrupted in between t = threading.Thread(target = self.waitAndSetFinalPosition, args = (shutterId, timeToWait, intermediatePosition)) t.start() - return + return # genuinely moving again — not a stop; the settled state + # reports via setPosition()'s existing position callback + # when the thread completes, same as rise()/lower(). + + # Fire the movement event before setPosition(): if the computed position + # lands exactly on 0/100, setPosition()'s existing position callback + # publishes the more precise open/closed state, and should have the + # last word on the retained MQTT topic, not this 'stopped' event. + self._fireMovement(shutterId, 'stopped') # Save computed position self.setPosition(shutterId, newPosition) @@ -297,6 +390,42 @@ def program(self, shutterId): def registerCallBack(self, callbackFunction): self.callback.append(callbackFunction) + # Register for transient movement-state notifications ('opening'/ + # 'closing'/'stopped'), fired identically for the TX/software path and + # the physical-remote path — complements registerCallBack's position- + # based open/closed/stopped signal with the "movement in progress" + # state nothing else reports today. + def registerMovementCallBack(self, callbackFunction): + self.movementCallback.append(callbackFunction) + + def _fireMovement(self, shutterId, movementState): + self.movementStateList[shutterId] = movementState + for function in self.movementCallback: + function(shutterId, movementState) + + # Last movement state fired for a shutter ('opening'/'closing'/'stopped'), + # or None if it's never moved this run — lets a REST poller (getStatus) + # see the same signal MQTT gets pushed, without needing its own callback. + def getMovementState(self, shutterId): + return self.movementStateList.get(shutterId) + + # Update the position model for a button press heard from a physical RTS + # remote — dispatches to the same _simulate* methods the TX path uses, + # with no RF transmission. + def recordExternalCommand(self, shutterId, button): + name = self.config.Shutters[shutterId]['name'] + if button == self.buttonUp: + self.LogInfo("["+name+"] Physical remote: UP") + self._simulateUp(shutterId) + elif button == self.buttonDown: + self.LogInfo("["+name+"] Physical remote: DOWN") + self._simulateDown(shutterId) + elif button == self.buttonStop: + self.LogInfo("["+name+"] Physical remote: STOP/MY") + self._simulateStop(shutterId) + else: + self.LogWarn("["+name+"] Physical remote: unhandled button 0x%X" % button) + def sendCommand(self, shutterId, button, repetition): #Sending a frame # Sending more than two repetitions after the original frame means a button kept pressed and moves the blind in steps # to adjust the tilt. Sending the original frame and three repetitions is the smallest adjustment, sending the original @@ -307,6 +436,7 @@ def sendCommand(self, shutterId, button, repetition): #Sending a frame self.lock.acquire() try: self.LogDebug("sendCommand: Lock acquired") + self.transmitting.set() checksum = 0 teleco = int(shutterId, 16) @@ -410,6 +540,7 @@ def sendCommand(self, shutterId, button, repetition): #Sending a frame pi.stop() finally: + self.transmitting.clear() self.lock.release() self.LogDebug("sendCommand: Lock released") @@ -520,6 +651,7 @@ def __init__(self, args = None): self.schedule = Schedule(log = self.log, config = self.config) self.scheduler = None self.webServer = None + self.receiver = None if (args.echo == True): self.alexa = Alexa(kwargs={'log':self.log, 'shutter': self.shutter, 'config': self.config}) @@ -528,6 +660,11 @@ def __init__(self, args = None): from mqtt import MQTT self.mqtt = MQTT(kwargs={'log':self.log, 'shutter': self.shutter, 'config': self.config}) + # Enabled when RXGPIO is present in [General] — no new CLI flag. + if not WINDOWS and self.config.RXGPIO is not None: + from receiver import Receiver + self.receiver = Receiver(kwargs={'log':self.log, 'shutter': self.shutter, 'config': self.config, 'console': self.console}) + self.ProcessCommand(args); #------------------------ operateShutters::IsLoaded ----------------------------- @@ -563,41 +700,53 @@ def startGPIO(self): return False # pigpio path for Pi 1/2/3/4 + # Deliberately not passing -m (disable alerts): -m silently prevents + # pi.callback() from ever delivering edge notifications, which the + # receiver (Receiver, when config.RXGPIO is set) needs. + # No sudo: __init__ already refused to run unless os.geteuid() == 0, + # so this process is already root in every deployment (standalone, + # systemd, or a container) — sudo here would just add a dependency + # this code doesn't need. if sys.version_info[0] < 3: import commands - status, process = commands.getstatusoutput('sudo pidof pigpiod') + status, process = commands.getstatusoutput('pidof pigpiod') if status: # it wasn't running, so start it self.LogInfo ("pigpiod was not running") - commands.getstatusoutput('sudo pigpiod -l -m') # try to start it + commands.getstatusoutput('pigpiod -l') # try to start it time.sleep(0.5) # check it again - status, process = commands.getstatusoutput('sudo pidof pigpiod') + status, process = commands.getstatusoutput('pidof pigpiod') else: import subprocess - status, process = subprocess.getstatusoutput('sudo pidof pigpiod') + status, process = subprocess.getstatusoutput('pidof pigpiod') if status: # it wasn't running, so start it self.LogInfo ("pigpiod was not running") - subprocess.getstatusoutput('sudo pigpiod -l -m') # try to start it + subprocess.getstatusoutput('pigpiod -l') # try to start it time.sleep(0.5) # check it again - status, process = subprocess.getstatusoutput('sudo pidof pigpiod') + status, process = subprocess.getstatusoutput('pidof pigpiod') if not status: # if it was started successfully (or was already running)... pigpiod_process = process self.LogInfo ("pigpiod is running, process ID is {} ".format(pigpiod_process)) + self.LogConsole("pigpiod is running, process ID is {} ".format(pigpiod_process)) try: pi = pigpio.pi() # local GPIO only if not pi.connected: self.LogError("pigpio connection could not be established. Check logs to get more details.") + self.LogConsole("pigpio connection could not be established.") return False else: self.LogInfo("pigpio's pi instantiated.") + self.LogConsole("pigpio's pi instantiated.") except Exception as e: start_pigpiod_exception = str(e) self.LogError("problem instantiating pi: {}".format(start_pigpiod_exception)) + self.LogConsole("problem instantiating pi: {}".format(start_pigpiod_exception)) else: self.LogError("start pigpiod was unsuccessful.") + self.LogConsole("start pigpiod was unsuccessful (pidof pigpiod status={}).".format(status)) return False return True @@ -634,6 +783,9 @@ def ProcessCommand(self, args): if (args.mqtt == True): self.mqtt.daemon = True self.mqtt.start() + if self.receiver is not None: + self.receiver.daemon = True + self.receiver.start() self.scheduler.join() elif ((args.shutterName != "") and (args.press)): @@ -661,7 +813,10 @@ def ProcessCommand(self, args): if (args.mqtt == True): self.mqtt.daemon = True self.mqtt.start() - self.webServer = FlaskAppWrapper(name='WebServer', static_url_path=os.path.dirname(os.path.realpath(__file__))+'/html', log = self.log, shutter = self.shutter, schedule = self.schedule, config = self.config) + if self.receiver is not None: + self.receiver.daemon = True + self.receiver.start() + self.webServer = FlaskAppWrapper(name='WebServer', static_url_path=os.path.dirname(os.path.realpath(__file__))+'/html', log = self.log, shutter = self.shutter, schedule = self.schedule, config = self.config, receiver = self.receiver) self.webServer.run() else: parser.print_help() @@ -672,11 +827,16 @@ def ProcessCommand(self, args): if (args.mqtt == True): self.mqtt.daemon = True self.mqtt.start() + if self.receiver is not None: + self.receiver.daemon = True + self.receiver.start() if (args.echo == True): self.alexa.join() if (args.mqtt == True): self.mqtt.join() + if self.receiver is not None: + self.receiver.join() self.LogInfo ("Process Command Completed....") self.Close(); @@ -708,6 +868,11 @@ def Close(self, signum = None, frame = None): self.mqtt.shutdown_flag.set() self.mqtt.join() self.LogError("MQTT Listener stopped. Now exiting.") + if self.receiver is not None: + self.LogError("Stopping RTS Receiver. This can take up to 1 second...") + self.receiver.shutdown_flag.set() + self.receiver.join() + self.LogError("RTS Receiver stopped. Now exiting.") if self.webServer is not None: self.LogError("Stopping WebServer. This can take up to 1 second...") self.webServer.shutdown_server() diff --git a/receiver.py b/receiver.py new file mode 100644 index 0000000..3cccbd3 --- /dev/null +++ b/receiver.py @@ -0,0 +1,763 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +"""Somfy RTS receiver — decodes physical remote presses and updates the +Shutter position model. + +Configures a CC1101 receiver module over bit-banged SPI for 433.42 MHz OOK +reception, and decodes the RTS frame format (manchester-encoded, checksummed, +obfuscated via a running XOR chain). Runs as a Pi-Somfy service thread with a +TX-pause gate, so the receiver never tries to decode our own transmissions, +and a self-echo filter as a second line of defense. Known physical remotes +are mapped to shutters via [PhysicalRemotes] in the config file. +""" + +import collections +import os +import sys +import threading +import time + +from config import MyLog + +# GPIO libraries are only needed on real hardware. Import lazily, and detect +# the Pi model independently below, so the decoder classes stay unit-testable +# on any dev machine. (operateShutters.py's own IS_PI5/LGPIO_CHIP aren't +# imported here on purpose: that module hard-imports ephem/pigpio/lgpio at +# load time, which would make `import receiver` fail on a plain dev machine.) +try: + import pigpio +except ImportError: + pigpio = None +try: + import lgpio +except ImportError: + lgpio = None + + +# ── Pi model detection (copied from operateShutters.py — see note above) ──── +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) ───────── +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"} + +# Bit-banged SPI GPIO defaults — used when RXSpiSCK/MOSI/MISO/CSN aren't set +# in [General]. +DEFAULT_SPI_SCK = 21 +DEFAULT_SPI_MOSI = 20 +DEFAULT_SPI_MISO = 19 +DEFAULT_SPI_CSN = 16 + + +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. Used by test_receiver.py to + build synthetic edge streams; the receiver itself only decodes. + """ + 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). + """ + 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. + + 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 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 ────────────────────────────────── + +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 here)"), + (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: 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. `log` is a MyLog-style + object (LogInfo/LogError), not a stdlib logger. + """ + + 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.LogInfo("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.LogInfo("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, log, glitch_us=150): + self._pi = pi + self._gpio = gpio + self._on_edge = on_edge + self._log = log + 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 as e: + self._log.LogError("Receiver: decoder error: " + str(e)) + + 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, log, glitch_us=150): + self._h = handle + self._gpio = gpio + self._on_edge = on_edge + self._log = log + 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 as e: + self._log.LogError("Receiver: decoder error: " + str(e)) + + def stop(self): + self._cb.cancel() + try: + lgpio.gpio_free(self._h, self._gpio) + except Exception: + pass + + +class Receiver(threading.Thread, MyLog): + """Service thread: listens for physical Somfy RTS remote presses and + updates Shutter's position model. + + Constructed and started the same way as Scheduler/MQTT (kwargs={'log', + 'shutter', 'config'}), gated on config.RXGPIO being set — no new CLI flag. + All hardware bring-up happens in run(), never in __init__, so a + mis-wired/missing CC1101 disables the receiver (logs an error and + returns) instead of crashing the whole process. + """ + + def __init__(self, group=None, target=None, name=None, args=(), kwargs=None): + threading.Thread.__init__(self, group=group, target=target, name="Receiver") + self.shutdown_flag = threading.Event() + self.args = args + self.kwargs = kwargs + if kwargs["log"] is not None: + self.log = kwargs["log"] + if kwargs["shutter"] is not None: + self.shutter = kwargs["shutter"] + if kwargs["config"] is not None: + self.config = kwargs["config"] + # Optional: streams hardware bring-up status to the console/add-on + # log (self.log alone is file-only), same as operateShutters.py's + # own startGPIO() diagnostics. None-safe — LogConsole no-ops without it. + self.console = kwargs.get("console") + + self._pi = None + self._lgpio_handle = None + self._spi = None + self._cc1101 = None + self._edge_source = None + self._tx_gated = False + # Presses from remotes not found in [PhysicalRemotes] — visible for + # a future pairing UI — for now just logs them. + self._unknown_remotes = collections.deque(maxlen=32) + + self._tracker = PressTracker(on_press=self._dispatch, on_press_end=self._on_press_end) + self._decoder = RTSDecoder(on_frame=self._on_frame) + + # -- hardware bring-up ----------------------------------------------------- + def _start_hardware(self): + rx_gpio = self.config.RXGPIO + spi_sck = self.config.RXSpiSCK if self.config.RXSpiSCK is not None else DEFAULT_SPI_SCK + spi_mosi = self.config.RXSpiMOSI if self.config.RXSpiMOSI is not None else DEFAULT_SPI_MOSI + spi_miso = self.config.RXSpiMISO if self.config.RXSpiMISO is not None else DEFAULT_SPI_MISO + spi_csn = self.config.RXSpiCSN if self.config.RXSpiCSN is not None else DEFAULT_SPI_CSN + + 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.LogInfo("Receiver: Pi 5, lgpio on gpiochip%d" % LGPIO_CHIP) + self.LogConsole("Receiver: Pi 5, lgpio on gpiochip%d" % LGPIO_CHIP) + self._spi = LgpioBitBangSpi(self._lgpio_handle, spi_sck, spi_mosi, spi_miso, spi_csn) + 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.LogInfo("Receiver: connected to pigpiod") + self.LogConsole("Receiver: connected to pigpiod") + self._spi = PigpioBitBangSpi(self._pi, spi_sck, spi_mosi, spi_miso, spi_csn) + + self._cc1101 = CC1101(self._spi, self) + self._cc1101.configure() + self.LogConsole("Receiver: CC1101 configured") + + if IS_PI5: + self._edge_source = LgpioEdgeSource(self._lgpio_handle, rx_gpio, self._on_edge, self) + else: + self._edge_source = PigpioEdgeSource(self._pi, rx_gpio, self._on_edge, self) + self.LogInfo("Receiver: listening on GPIO %d" % rx_gpio) + self.LogConsole("Receiver: listening on GPIO %d" % rx_gpio) + + def _stop_hardware(self): + if self._edge_source is not None: + self._edge_source.stop() + if self._spi is not None: + self._spi.close() + if self._pi is not None: + self._pi.stop() + if self._lgpio_handle is not None: + lgpio.gpiochip_close(self._lgpio_handle) + + # -- edge / frame / press pipeline ----------------------------------------- + # Sits between the edge source and the decoder: drop edges while our own + # TX path is transmitting, and resync the decoder once TX ends — found + # during earlier testing that a receiver left running during TX otherwise decodes + # noise/reflections of our own frame. + def _on_edge(self, level, ts_us): + if self.shutter.transmitting.is_set(): + self._tx_gated = True + return + if self._tx_gated: + self._tx_gated = False + self._decoder.reset() + self._decoder.on_edge(level, ts_us) + + # Sits between the decoder and the press tracker: a second, independent + # defense against self-echo (catches cleanly-decoded frames outside the + # TX-gated window, e.g. multipath) — neither defense subsumes the other. + def _on_frame(self, frame): + if ("0x%06x" % frame.address) in self.config.Shutters: + return + self._tracker.on_frame(frame) + + def _dispatch(self, frame): + address = "0x%06x" % frame.address + shutterIds = self.config.PhysicalRemotes.get(address) + if not shutterIds: + msg = ("Receiver: unknown remote " + address + " pressed " + + button_name(frame.button) + " (code=" + str(frame.rolling_code) + ")") + self.LogInfo(msg) + self.LogConsole(msg) + self._unknown_remotes.append((address, frame.button, frame.rolling_code)) + return + self.LogConsole("Receiver: " + address + " pressed " + button_name(frame.button) + + " -> " + ", ".join(shutterIds)) + for shutterId in shutterIds: + if shutterId not in self.config.Shutters: + self.LogWarn("Receiver: [PhysicalRemotes] " + address + + " maps to unknown shutterId " + shutterId) + continue + self.shutter.recordExternalCommand(shutterId, frame.button) + + def _on_press_end(self, frame, repeats): + self.LogDebug("Receiver: 0x%06x %s repeats=%d" % + (frame.address, button_name(frame.button), repeats)) + + # -- main loop --------------------------------------------------------------- + def run(self): + try: + self._start_hardware() + except Exception as e: + self.LogError("Receiver: hardware init failed — receiver disabled: " + str(e)) + self.LogConsole("Receiver: hardware init failed — receiver disabled: " + str(e)) + self._stop_hardware() + return + try: + while not self.shutdown_flag.is_set(): + time.sleep(0.2) + self._tracker.poll() + finally: + self._stop_hardware() diff --git a/test_config.py b/test_config.py new file mode 100644 index 0000000..2a85565 --- /dev/null +++ b/test_config.py @@ -0,0 +1,94 @@ +# -*- coding: utf-8 -*- +"""config.py unit tests — run anywhere, no GPIO libraries needed: + + python3 -m unittest discover + +Covers RemoveValue (new in M2, needed for a real "unassign" of a +[PhysicalRemotes] entry) and WriteValue's auto-create-missing-section +behavior, using a real temp file so the exact on-disk INI text is verified, +not just the in-memory RawConfigParser view. +""" + +import os +import tempfile +import unittest + +from config import MyConfig + + +class ConfigTestCase(unittest.TestCase): + + def setUp(self): + fd, self.path = tempfile.mkstemp(suffix=".conf") + os.close(fd) + with open(self.path, "w") as f: + f.write("[General]\nLogLocation = /tmp/\n\n" + "[Shutters]\n0x111111 = Test,True,10\n") + self.cfg = MyConfig(filename=self.path, section="General") + + def tearDown(self): + os.remove(self.path) + + def read_file(self): + with open(self.path) as f: + return f.read() + + +class WriteValueTests(ConfigTestCase): + + def test_write_value_auto_creates_missing_section(self): + ok = self.cfg.WriteValue("0xaaaaaa", "shutter1,shutter2", section="PhysicalRemotes") + self.assertTrue(ok) + self.assertIn("[PhysicalRemotes]", self.read_file()) + self.assertEqual( + self.cfg.ReadValue("0xaaaaaa", section="PhysicalRemotes"), + "shutter1,shutter2") + + def test_write_value_replaces_existing_key(self): + self.cfg.WriteValue("0x111111", "Renamed,True,20", section="Shutters") + self.assertEqual( + self.cfg.ReadValue("0x111111", section="Shutters"), + "Renamed,True,20") + # still exactly one row for this key + rows = [k for k, v in self.cfg.GetList(section="Shutters") if k == "0x111111"] + self.assertEqual(len(rows), 1) + + +class RemoveValueTests(ConfigTestCase): + + def test_remove_existing_key_deletes_line(self): + self.cfg.WriteValue("0xaaaaaa", "shutter1", section="PhysicalRemotes") + ok = self.cfg.RemoveValue("0xaaaaaa", section="PhysicalRemotes") + self.assertTrue(ok) + self.assertFalse(self.cfg.HasOption("0xaaaaaa", section="PhysicalRemotes")) + self.assertNotIn("0xaaaaaa", self.read_file()) + + def test_remove_nonexistent_key_is_idempotent(self): + ok = self.cfg.RemoveValue("0xdeadbe", section="Shutters") + self.assertTrue(ok) + # unrelated existing key untouched + self.assertTrue(self.cfg.HasOption("0x111111", section="Shutters")) + + def test_remove_from_nonexistent_section_returns_true(self): + ok = self.cfg.RemoveValue("whatever", section="NoSuchSection") + self.assertTrue(ok) + + def test_write_then_remove_round_trip(self): + self.cfg.WriteValue("0xbbbbbb", "shutterX", section="PhysicalRemotes") + self.assertEqual( + self.cfg.ReadValue("0xbbbbbb", section="PhysicalRemotes"), "shutterX") + self.cfg.RemoveValue("0xbbbbbb", section="PhysicalRemotes") + self.assertIsNone( + self.cfg.ReadValue("0xbbbbbb", section="PhysicalRemotes", default=None)) + + def test_remove_does_not_disturb_other_keys_in_section(self): + self.cfg.WriteValue("0xaaaaaa", "shutter1", section="PhysicalRemotes") + self.cfg.WriteValue("0xbbbbbb", "shutter2", section="PhysicalRemotes") + self.cfg.RemoveValue("0xaaaaaa", section="PhysicalRemotes") + self.assertFalse(self.cfg.HasOption("0xaaaaaa", section="PhysicalRemotes")) + self.assertEqual( + self.cfg.ReadValue("0xbbbbbb", section="PhysicalRemotes"), "shutter2") + + +if __name__ == "__main__": + unittest.main() diff --git a/test_receiver.py b/test_receiver.py new file mode 100644 index 0000000..049cd80 --- /dev/null +++ b/test_receiver.py @@ -0,0 +1,617 @@ +# -*- coding: utf-8 -*- +"""Receiver unit tests — run anywhere, no GPIO libraries needed: + + python3 -m unittest discover + +Decoder tests feed synthetic (level, timestamp_us) edge streams, built from +the same pulse tables Shutter.sendCommand uses, directly to RTSDecoder/ +PressTracker. The remaining classes test receiver-specific behavior: +TX-pause gating, self-echo filtering, [PhysicalRemotes] group fan-out, and +the dispatch equivalence between a physical remote press and the app's own +buttons. +""" + +import logging +import random +import threading +import time +import unittest + +from receiver import (BUTTON_DOWN, BUTTON_PROG, BUTTON_STOP, BUTTON_UP, + PressTracker, Receiver, RTSDecoder, build_frame, + frame_to_pulses, pulses_to_edges) + +logging.getLogger("test_receiver").addHandler(logging.NullHandler()) +LOG = logging.getLogger("test_receiver") +LOG.setLevel(logging.CRITICAL) # keep test output clean; correctness is asserted, not logged + +# 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): + # 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.""" + + 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. + 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 + + 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) + + +# ── Receiver integration: TX-gate, self-echo, [PhysicalRemotes] fan-out ───── + +class FakeShutter(object): + """Minimal Shutter stand-in exposing only what Receiver touches.""" + + def __init__(self): + self.transmitting = threading.Event() + self.commands = [] # [(shutterId, button), ...] + + def recordExternalCommand(self, shutterId, button): + self.commands.append((shutterId, button)) + + +class FakeConfig(object): + + def __init__(self, shutters=None, physical_remotes=None): + self.Shutters = shutters or {} + self.PhysicalRemotes = physical_remotes or {} + self.RXGPIO = None + self.RXSpiSCK = None + self.RXSpiMOSI = None + self.RXSpiMISO = None + self.RXSpiCSN = None + + +def make_receiver(shutters=None, physical_remotes=None): + shutter = FakeShutter() + config = FakeConfig(shutters, physical_remotes) + receiver = Receiver(kwargs={'log': LOG, 'shutter': shutter, 'config': config}) + return receiver, shutter, config + + +class TxGateTests(unittest.TestCase): + + def test_edges_dropped_while_transmitting(self): + receiver, shutter, _ = make_receiver() + shutter.transmitting.set() + for level, ts in press_edges(0x111111, BUTTON_UP, 1): + receiver._on_edge(level, ts) + self.assertEqual(receiver._decoder.frames_decoded, 0) + self.assertEqual(shutter.commands, []) + + def test_decoder_resets_after_gated_partial_frame(self): + receiver, shutter, _ = make_receiver( + shutters={"0x02aaaa": {}}, + physical_remotes={"0x222222": ["0x02aaaa"]}) + + shutter.transmitting.set() + interrupted = press_edges(0x111111, BUTTON_UP, 1) + midpoint = len(interrupted) // 2 + for level, ts in interrupted[:midpoint]: + receiver._on_edge(level, ts) # dropped: gated + shutter.transmitting.clear() + for level, ts in interrupted[midpoint:]: + receiver._on_edge(level, ts) # tail arrives post-gate; must not wedge the decoder + self.assertEqual(shutter.commands, []) + + # A subsequent, unrelated, complete press must still decode cleanly — + # proving the decoder was reset rather than left mid-payload. + next_press = press_edges(0x222222, BUTTON_STOP, 2, + start_us=interrupted[-1][1] + 200000) + for level, ts in next_press: + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, [("0x02aaaa", BUTTON_STOP)]) + + +class SelfEchoTests(unittest.TestCase): + + def test_own_shutter_address_is_dropped(self): + receiver, shutter, _ = make_receiver(shutters={"0x111111": {}}) + for level, ts in press_edges(0x111111, BUTTON_UP, 1): + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, []) + + def test_other_address_is_not_dropped(self): + receiver, shutter, _ = make_receiver( + shutters={"0x111111": {}, "0x02aaaa": {}}, + physical_remotes={"0x222222": ["0x02aaaa"]}) + for level, ts in press_edges(0x222222, BUTTON_DOWN, 1): + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, [("0x02aaaa", BUTTON_DOWN)]) + + +class PhysicalRemotesFanOutTests(unittest.TestCase): + + def test_group_fans_out_to_every_shutter(self): + receiver, shutter, _ = make_receiver( + shutters={"0x02aaaa": {}, "0x02bbbb": {}}, + physical_remotes={"0x279620": ["0x02aaaa", "0x02bbbb"]}) + for level, ts in press_edges(0x279620, BUTTON_DOWN, 42): + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, + [("0x02aaaa", BUTTON_DOWN), ("0x02bbbb", BUTTON_DOWN)]) + + def test_unknown_remote_is_recorded_not_dispatched(self): + receiver, shutter, _ = make_receiver() + for level, ts in press_edges(0xABCDEF, BUTTON_UP, 1): + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, []) + self.assertEqual(len(receiver._unknown_remotes), 1) + self.assertEqual(receiver._unknown_remotes[0][0], "0xabcdef") + + def test_unmapped_shutter_id_in_group_is_skipped(self): + receiver, shutter, _ = make_receiver( + shutters={"0x02aaaa": {}}, # 0x02cccc deliberately absent + physical_remotes={"0x279620": ["0x02aaaa", "0x02cccc"]}) + for level, ts in press_edges(0x279620, BUTTON_UP, 1): + receiver._on_edge(level, ts) + self.assertEqual(shutter.commands, [("0x02aaaa", BUTTON_UP)]) + + +# ── Simulation equivalence: physical remote vs. app button ────────────────── +# Needs the real Shutter class, which means operateShutters.py's full +# requirements.txt (ephem, pigpio/lgpio, Flask, paho-mqtt) — unlike the pure +# decoder tests above, this isn't optional-dependency-light, so skip cleanly +# if those aren't installed rather than failing collection for the whole file. +try: + from operateShutters import Shutter + _HAVE_SHUTTER = True +except Exception: + _HAVE_SHUTTER = False + + +@unittest.skipUnless(_HAVE_SHUTTER, "operateShutters and its dependencies " + "(ephem, pigpio/lgpio, Flask, paho-mqtt) are required " + "to test Shutter directly") +class SimulationEquivalenceTests(unittest.TestCase): + """recordExternalCommand (physical-remote path) must dispatch to exactly + the same _simulate* methods rise()/lower()/stop() use — a + physical press updates the position model identically to the equivalent + app button, minus the RF transmission.""" + + class _FakeConfig(object): + def __init__(self, shutters): + self.TXGPIO = None + self.ShutterPositions = {} + self.Shutters = shutters + self.SendRepeat = 1 + + def WriteValue(self, entry, value, section=None): + pass + + def _make_shutter(self): + shutters = {"0x02aaaa": {"name": "test", "durationDown": 20, + "durationUp": 20, "intermediatePosition": None}} + return Shutter(log=LOG, config=self._FakeConfig(shutters)) + + def test_up_button_dispatches_to_simulate_up(self): + shutter = self._make_shutter() + calls = [] + shutter._simulateUp = lambda shutterId: calls.append(shutterId) + shutter.recordExternalCommand("0x02aaaa", Shutter.buttonUp) + self.assertEqual(calls, ["0x02aaaa"]) + + def test_down_button_dispatches_to_simulate_down(self): + shutter = self._make_shutter() + calls = [] + shutter._simulateDown = lambda shutterId: calls.append(shutterId) + shutter.recordExternalCommand("0x02aaaa", Shutter.buttonDown) + self.assertEqual(calls, ["0x02aaaa"]) + + def test_stop_button_dispatches_to_simulate_stop(self): + shutter = self._make_shutter() + calls = [] + shutter._simulateStop = lambda shutterId: calls.append(shutterId) + shutter.recordExternalCommand("0x02aaaa", Shutter.buttonStop) + self.assertEqual(calls, ["0x02aaaa"]) + + +@unittest.skipUnless(_HAVE_SHUTTER, "operateShutters and its dependencies " + "(ephem, pigpio/lgpio, Flask, paho-mqtt) are required " + "to test Shutter directly") +class MovementCallbackTests(unittest.TestCase): + """registerMovementCallBack fires 'opening'/'closing'/'stopped' for both + the TX/software path and the physical-remote path identically (M2 §5.4).""" + + class _FakeConfig(object): + def __init__(self, shutters): + self.TXGPIO = None + self.ShutterPositions = {} + self.Shutters = shutters + self.SendRepeat = 1 + + def WriteValue(self, entry, value, section=None): + pass + + def _make_shutter(self, duration=20, intermediatePosition=None): + shutters = {"0x02aaaa": {"name": "test", "durationDown": duration, + "durationUp": duration, + "intermediatePosition": intermediatePosition}} + shutter = Shutter(log=LOG, config=self._FakeConfig(shutters)) + shutter.sendCommand = lambda *a, **kw: None # never touch real hardware + self.events = [] + shutter.registerMovementCallBack( + lambda shutterId, state: self.events.append(("movement", state))) + shutter.registerCallBack( + lambda shutterId, position: self.events.append(("position", position))) + return shutter + + def movements(self): + return [e[1] for e in self.events if e[0] == "movement"] + + def test_simulate_up_fires_opening_immediately(self): + shutter = self._make_shutter() + shutter._simulateUp("0x02aaaa") + self.assertEqual(self.movements(), ["opening"]) + + def test_get_movement_state_reflects_last_fired_event(self): + shutter = self._make_shutter() + self.assertIsNone(shutter.getMovementState("0x02aaaa")) + # Start mid-travel: at position 0 (or 100), the very next _simulateUp/ + # _simulateDown call computes a near-zero timeToWait for its spawned + # settle thread (already at the target end), which can race ahead and + # fire 'stopped' before this test's own assertions run. + shutter.getShutterState("0x02aaaa", 50) + shutter._simulateUp("0x02aaaa") + self.assertEqual(shutter.getMovementState("0x02aaaa"), "opening") + shutter._simulateDown("0x02aaaa") + self.assertEqual(shutter.getMovementState("0x02aaaa"), "closing") + + def test_simulate_down_fires_closing_immediately(self): + shutter = self._make_shutter() + shutter._simulateDown("0x02aaaa") + self.assertEqual(self.movements(), ["closing"]) + + def test_simulate_stop_normal_fires_stopped_before_position_settles(self): + shutter = self._make_shutter(duration=100) + state = shutter.getShutterState("0x02aaaa", 50) + state.registerCommand('up') + shutter._fireMovement("0x02aaaa", 'opening') # mark as genuinely moving + self.events = [] # discard the setup event above; assert only on _simulateStop's own effects + state.lastCommandTime = time.monotonic() - 1.0 # ~1s into a 100s travel + shutter._simulateStop("0x02aaaa") + self.assertEqual(self.movements(), ["stopped"]) + # The movement event must precede the position-settle event, so a + # more-precise open/closed publish (from the position callback) + # always has the last word on the retained MQTT topic. + self.assertEqual([e[0] for e in self.events], ["movement", "position"]) + + def test_simulate_stop_intermediate_fallback_fires_closing_not_stopped(self): + # position (80) above the stored MY position (30): motor moves down. + shutter = self._make_shutter(duration=20, intermediatePosition=30) + shutter.getShutterState("0x02aaaa", 80) + shutter._simulateStop("0x02aaaa") + self.assertEqual(self.movements(), ["closing"]) + + def test_simulate_stop_intermediate_fallback_fires_opening_not_stopped(self): + # position (30) below the stored MY position (80): motor moves up. + shutter = self._make_shutter(duration=20, intermediatePosition=80) + shutter.getShutterState("0x02aaaa", 30) + shutter._simulateStop("0x02aaaa") + self.assertEqual(self.movements(), ["opening"]) + + def test_simulate_stop_stationary_fallback_fires_stopped(self): + shutter = self._make_shutter(duration=20, intermediatePosition=None) + shutter.getShutterState("0x02aaaa", 50) + shutter._simulateStop("0x02aaaa") + self.assertEqual(self.movements(), ["stopped"]) + + def test_stop_after_settled_partial_move_goes_to_my_position(self): + # Regression: _simulateStop used to infer "still moving" from + # elapsed-time-since-last-command vs. the shutter's FULL duration — + # which wrongly matched here, since a partial move settles in far + # less time than the full duration. A MY press shortly afterward + # must correctly see the shutter as already stopped and evaluate + # the MY-position fallback, not compute a bogus interrupted-move + # position as if it were still rising. + shutter = self._make_shutter(duration=20, intermediatePosition=45) + # Drive the same state transitions risePartial(shutterId, 60) would + # (registerCommand -> 'opening' -> 'stopped' + setPosition), without + # its real time.sleep(...) — risePartial blocks synchronously for the + # move's duration, unlike _simulateUp/_simulateDown which settle on a + # background thread, so calling it here would make this test take + # (60/100)*20s = 12 real seconds. + state = shutter.getShutterState("0x02aaaa", 0) + state.registerCommand('up') + shutter._fireMovement("0x02aaaa", 'opening') + shutter._fireMovement("0x02aaaa", 'stopped') + shutter.setPosition("0x02aaaa", 60) + self.assertEqual(shutter.getPosition("0x02aaaa"), 60) + self.events = [] + shutter._simulateStop("0x02aaaa") + # 60 -> intermediatePosition 45 is a move down (closing), not the + # bogus "still opening" interpolation the old elapsed-time + # heuristic would have produced. + self.assertEqual(self.movements(), ["closing"]) + + def test_rise_partial_fires_opening_then_stopped(self): + shutter = self._make_shutter(duration=0) + shutter.risePartial("0x02aaaa", 80) + self.assertEqual(self.movements(), ["opening", "stopped"]) + + def test_lower_partial_fires_closing_then_stopped(self): + shutter = self._make_shutter(duration=0) + shutter.lowerPartial("0x02aaaa", 20) + self.assertEqual(self.movements(), ["closing", "stopped"]) + + def test_record_external_command_up_fires_same_event_as_simulate_up(self): + shutter = self._make_shutter() + shutter.recordExternalCommand("0x02aaaa", Shutter.buttonUp) + self.assertEqual(self.movements(), ["opening"]) + + def test_full_move_fires_stopped_when_it_completes_naturally(self): + # waitAndSetFinalPosition's background thread (spawned by + # _simulateUp/_simulateDown for a full, uninterrupted move) must + # clear 'opening'/'closing' back to 'stopped' once it reaches 100/0 + # — this is the only path that settles without an explicit stop + # command, so nothing else fires this event for it. + shutter = self._make_shutter(duration=0.05) + shutter._simulateUp("0x02aaaa") + for _ in range(50): + if self.movements() == ["opening", "stopped"]: + break + time.sleep(0.02) + self.assertEqual(self.movements(), ["opening", "stopped"]) + + def test_display_position_equals_settled_position_when_stationary(self): + shutter = self._make_shutter() + shutter.getShutterState("0x02aaaa", 42) + self.assertEqual(shutter.getDisplayPosition("0x02aaaa"), 42) + + def test_display_position_interpolates_while_opening(self): + shutter = self._make_shutter(duration=100) + state = shutter.getShutterState("0x02aaaa", 0) + state.registerCommand('up') + shutter._fireMovement("0x02aaaa", 'opening') + state.lastCommandTime = time.monotonic() - 25.0 # 25% into a 100s move + self.assertEqual(shutter.getDisplayPosition("0x02aaaa"), 25) + + def test_display_position_interpolates_while_closing(self): + shutter = self._make_shutter(duration=100) + state = shutter.getShutterState("0x02aaaa", 100) + state.registerCommand('down') + shutter._fireMovement("0x02aaaa", 'closing') + state.lastCommandTime = time.monotonic() - 30.0 # 30% into a 100s move + self.assertEqual(shutter.getDisplayPosition("0x02aaaa"), 70) + + def test_display_position_reports_target_once_elapsed_exceeds_duration(self): + # Right before waitAndSetFinalPosition's background thread actually + # settles (a real race window), a poll should report the target + # (100/0), not the stale pre-move position. + shutter = self._make_shutter(duration=10) + state = shutter.getShutterState("0x02aaaa", 0) + state.registerCommand('up') + shutter._fireMovement("0x02aaaa", 'opening') + state.lastCommandTime = time.monotonic() - 15.0 # already past the 10s duration + self.assertEqual(shutter.getDisplayPosition("0x02aaaa"), 100) + + def test_display_position_interpolates_during_my_position_fallback_move(self): + # STOP/MY pressed while stationary and away from the stored MY + # position (the bug report this is fixing): the motor moves toward + # intermediatePosition, and getDisplayPosition should track it live, + # not just show the stale starting position until it settles. + shutter = self._make_shutter(duration=20, intermediatePosition=30) + shutter.getShutterState("0x02aaaa", 80) # above the MY position -> closing + shutter._simulateStop("0x02aaaa") + self.assertEqual(self.movements(), ["closing"]) + state = shutter.getShutterState("0x02aaaa") + # 5s at the configured 100/20=5%/s rate -> 25 points off 80, landing + # mid-travel toward (not yet at) the 30 target — this move's actual + # full duration (per _simulateStop's own math) is 10s (50 points at + # 5%/s), so 5s in is genuinely halfway there, not coincidentally at + # the destination. + state.lastCommandTime = time.monotonic() - 5.0 + self.assertEqual(shutter.getDisplayPosition("0x02aaaa"), 55) + + +if __name__ == "__main__": + unittest.main() diff --git a/test_webserver.py b/test_webserver.py new file mode 100644 index 0000000..dfa0469 --- /dev/null +++ b/test_webserver.py @@ -0,0 +1,303 @@ +# -*- coding: utf-8 -*- +"""webserver.py unit tests for the M2 Physical Remotes endpoints. + +Needs Flask (and receiver.py's import, which only needs config.py — no +pigpio/lgpio required just to import webserver.py), so this whole file +skips cleanly if Flask isn't installed rather than failing collection. + + python3 -m unittest discover +""" + +import json +import logging +import unittest + +logging.getLogger("test_webserver").addHandler(logging.NullHandler()) +LOG = logging.getLogger("test_webserver") +LOG.setLevel(logging.CRITICAL) + +try: + from webserver import FlaskAppWrapper + _HAVE_WEBSERVER = True +except Exception: + _HAVE_WEBSERVER = False + + +def result_of(response): + """processCommand() returns Response(json.dumps(...)) without setting + content_type=application/json, so Flask's own response.get_json() + returns None — parse the raw body instead (true for every endpoint, + not just the ones under test here).""" + return json.loads(response.data) + + +class FakeConfig(object): + def __init__(self, password=""): + self.Password = password + self.Latitude = 0 + self.Longitude = 0 + self.Shutters = { + "0x02aaaa": {"name": "Shutter A", "durationDown": 20, "durationUp": 20, "intermediatePosition": None}, + "0x02bbbb": {"name": "Shutter B", "durationDown": 20, "durationUp": 20, "intermediatePosition": None}, + } + self.PhysicalRemotes = {} + self.written = [] # [(entry, value, section), ...] + self.removed = [] # [(entry, section), ...] + + def WriteValue(self, entry, value, section=None): + self.written.append((entry, value, section)) + return True + + def RemoveValue(self, entry, section=None): + self.removed.append((entry, section)) + return True + + +class FakeShutter(object): + def __init__(self, movement_states=None, positions=None, display_positions=None): + self._movement_states = movement_states or {} + self._positions = positions or {} + # Defaults to _positions unless a test needs to prove getConfig/ + # getStatus call getDisplayPosition specifically, not getPosition. + self._display_positions = display_positions if display_positions is not None else self._positions + + def getPosition(self, shutterId): + return self._positions.get(shutterId, 50) + + def getDisplayPosition(self, shutterId): + return self._display_positions.get(shutterId, 50) + + def getMovementState(self, shutterId): + return self._movement_states.get(shutterId) + + +class FakeSchedule(object): + def getScheduleAsDict(self): + return {} + + +class FakeReceiver(object): + def __init__(self, unknown_remotes=()): + self._unknown_remotes = list(unknown_remotes) + + +def make_client(config=None, receiver=None, shutter=None): + config = config or FakeConfig() + wrapper = FlaskAppWrapper(name="test", static_url_path="", log=LOG, + shutter=shutter or FakeShutter(), schedule=FakeSchedule(), + config=config, receiver=receiver) + return wrapper.app.test_client(), config + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class GetConfigMovementStatesTests(unittest.TestCase): + """getConfig also exposes MovementStates and Positions (ungated, unlike + getStatus) so the Pi-Somfy web UI's own Manual Operation panel can poll + them without needing password plumbing.""" + + def test_includes_movement_states_per_shutter(self): + shutter = FakeShutter(movement_states={"0x02aaaa": "closing"}) + client, _ = make_client(shutter=shutter) + result = result_of(client.get("/cmd/getConfig")) + self.assertEqual(result["MovementStates"], {"0x02aaaa": "closing", "0x02bbbb": None}) + + def test_includes_positions_per_shutter(self): + shutter = FakeShutter(positions={"0x02aaaa": 75}) + client, _ = make_client(shutter=shutter) + result = result_of(client.get("/cmd/getConfig")) + self.assertEqual(result["Positions"], {"0x02aaaa": 75, "0x02bbbb": 50}) + + def test_positions_use_display_position_not_settled_position(self): + # getConfig must report the live-interpolated estimate (for a moving + # shutter), not the last-settled value used for internal decisions. + shutter = FakeShutter(positions={"0x02aaaa": 10}, display_positions={"0x02aaaa": 55}) + client, _ = make_client(shutter=shutter) + result = result_of(client.get("/cmd/getConfig")) + self.assertEqual(result["Positions"]["0x02aaaa"], 55) + + def test_includes_intermediate_positions_per_shutter(self): + config = FakeConfig() + config.Shutters["0x02aaaa"]["intermediatePosition"] = 40 + client, _ = make_client(config=config) + result = result_of(client.get("/cmd/getConfig")) + self.assertEqual(result["ShutterIntermediatePositions"], {"0x02aaaa": 40, "0x02bbbb": None}) + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class SetIntermediatePositionTests(unittest.TestCase): + """The 'My Position' wizard step's Save button writes here, so Pi-Somfy's + own tracked intermediatePosition matches whatever value was just stored + on the real motor via the long-press procedure.""" + + def test_sets_position_and_updates_cache(self): + client, config = make_client() + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0x02aaaa", "position": "40"})) + self.assertEqual(result, {"status": "OK", "intermediatePosition": 40}) + self.assertEqual(config.written, [("0x02aaaa", "40", "ShutterIntermediatePositions")]) + self.assertEqual(config.Shutters["0x02aaaa"]["intermediatePosition"], 40) + + def test_clears_position_with_blank_value(self): + client, config = make_client() + config.Shutters["0x02aaaa"]["intermediatePosition"] = 40 + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0x02aaaa", "position": ""})) + self.assertEqual(result, {"status": "OK", "intermediatePosition": None}) + self.assertEqual(config.written, [("0x02aaaa", "None", "ShutterIntermediatePositions")]) + self.assertIsNone(config.Shutters["0x02aaaa"]["intermediatePosition"]) + + def test_rejects_unknown_shutter(self): + client, config = make_client() + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0xnotreal", "position": "40"})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_rejects_out_of_range_position(self): + client, config = make_client() + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0x02aaaa", "position": "101"})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_rejects_non_numeric_position(self): + client, config = make_client() + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0x02aaaa", "position": "abc"})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_requires_password_when_configured(self): + client, config = make_client(config=FakeConfig(password="secret")) + result = result_of(client.post("/cmd/setIntermediatePosition", + data={"shutter": "0x02aaaa", "position": "40"})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class GetStatusTests(unittest.TestCase): + """getStatus exposes the same movementState signal MQTT gets pushed, so + the HA custom component's REST polling can show opening/closing for any + trigger source (physical remote, web UI, or Home Assistant itself).""" + + def test_includes_movement_state_per_shutter(self): + shutter = FakeShutter(movement_states={"0x02aaaa": "opening"}) + client, _ = make_client(shutter=shutter) + result = result_of(client.get("/cmd/getStatus")) + self.assertEqual(result["status"], "OK") + self.assertEqual(result["shutters"]["0x02aaaa"]["movementState"], "opening") + + def test_position_uses_display_position_not_settled_position(self): + shutter = FakeShutter(positions={"0x02aaaa": 10}, display_positions={"0x02aaaa": 55}) + client, _ = make_client(shutter=shutter) + result = result_of(client.get("/cmd/getStatus")) + self.assertEqual(result["shutters"]["0x02aaaa"]["position"], 55) + + def test_movement_state_is_none_when_never_moved(self): + client, _ = make_client(shutter=FakeShutter()) + result = result_of(client.get("/cmd/getStatus")) + self.assertIsNone(result["shutters"]["0x02aaaa"]["movementState"]) + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class GetUnheardRemotesTests(unittest.TestCase): + + def test_empty_when_no_receiver(self): + client, _ = make_client(receiver=None) + result = result_of(client.get("/cmd/getUnheardRemotes")) + self.assertEqual(result, {"status": "OK", "remotes": []}) + + def test_lists_unknown_remotes(self): + receiver = FakeReceiver([("0xabcdef", 2, 100)]) + client, _ = make_client(receiver=receiver) + result = result_of(client.get("/cmd/getUnheardRemotes")) + self.assertEqual(result["status"], "OK") + self.assertEqual(len(result["remotes"]), 1) + self.assertEqual(result["remotes"][0]["address"], "0xabcdef") + self.assertEqual(result["remotes"][0]["buttonName"], "UP") + + def test_dedupes_by_address_keeping_most_recent(self): + receiver = FakeReceiver([("0xabcdef", 2, 100), ("0xabcdef", 4, 101)]) + client, _ = make_client(receiver=receiver) + result = result_of(client.get("/cmd/getUnheardRemotes")) + self.assertEqual(len(result["remotes"]), 1) + self.assertEqual(result["remotes"][0]["rollingCode"], 101) + + def test_ungated_even_with_password_set(self): + client, _ = make_client(config=FakeConfig(password="secret"), receiver=FakeReceiver()) + result = result_of(client.get("/cmd/getUnheardRemotes")) + self.assertEqual(result["status"], "OK") + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class AssignRemoteTests(unittest.TestCase): + + def test_assigns_and_updates_config(self): + client, config = make_client() + result = result_of(client.post("/cmd/assignRemote", + data={"address": "0xABCDEF", "shutterIds[]": ["0x02aaaa", "0x02bbbb"]})) + self.assertEqual(result["status"], "OK") + # normalized to lowercase before writing + self.assertEqual(config.written, [("0xabcdef", "0x02aaaa,0x02bbbb", "PhysicalRemotes")]) + self.assertEqual(config.PhysicalRemotes["0xabcdef"], ["0x02aaaa", "0x02bbbb"]) + + def test_rejects_unknown_shutter(self): + client, config = make_client() + result = result_of(client.post("/cmd/assignRemote", + data={"address": "0xabcdef", "shutterIds[]": ["0xnotreal"]})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_rejects_empty_selection(self): + client, config = make_client() + result = result_of(client.post("/cmd/assignRemote", data={"address": "0xabcdef"})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_requires_password_when_configured(self): + client, config = make_client(config=FakeConfig(password="secret")) + result = result_of(client.post("/cmd/assignRemote", + data={"address": "0xabcdef", "shutterIds[]": ["0x02aaaa"]})) + self.assertEqual(result["status"], "ERROR") + self.assertEqual(config.written, []) + + def test_succeeds_with_correct_password(self): + client, config = make_client(config=FakeConfig(password="secret")) + result = result_of(client.post("/cmd/assignRemote", + data={"address": "0xabcdef", "shutterIds[]": ["0x02aaaa"]}, + headers={"Password": "secret"})) + self.assertEqual(result["status"], "OK") + + +@unittest.skipUnless(_HAVE_WEBSERVER, "Flask is required to test webserver.py") +class UnassignRemoteTests(unittest.TestCase): + + def test_unassigns_existing_mapping(self): + config = FakeConfig() + config.PhysicalRemotes["0xabcdef"] = ["0x02aaaa"] + client, _ = make_client(config=config) + result = result_of(client.post("/cmd/unassignRemote", data={"address": "0xabcdef"})) + self.assertEqual(result["status"], "OK") + self.assertNotIn("0xabcdef", config.PhysicalRemotes) + self.assertEqual(config.removed, [("0xabcdef", "PhysicalRemotes")]) + + def test_second_unassign_reports_not_assigned(self): + config = FakeConfig() + config.PhysicalRemotes["0xabcdef"] = ["0x02aaaa"] + client, _ = make_client(config=config) + client.post("/cmd/unassignRemote", data={"address": "0xabcdef"}) + result = result_of(client.post("/cmd/unassignRemote", data={"address": "0xabcdef"})) + self.assertEqual(result["status"], "ERROR") + + def test_requires_password_when_configured(self): + config = FakeConfig(password="secret") + config.PhysicalRemotes["0xabcdef"] = ["0x02aaaa"] + client, _ = make_client(config=config) + result = result_of(client.post("/cmd/unassignRemote", data={"address": "0xabcdef"})) + self.assertEqual(result["status"], "ERROR") + self.assertIn("0xabcdef", config.PhysicalRemotes) + + +if __name__ == "__main__": + unittest.main() diff --git a/webserver.py b/webserver.py index d11457e..d59a91e 100644 --- a/webserver.py +++ b/webserver.py @@ -11,6 +11,7 @@ try: from config import MyLog + from receiver import button_name except Exception as e1: print("\n\nThis program requires the modules located from the same github repository that are not present.\n") print("Error: " + str(e1)) @@ -20,14 +21,15 @@ class FlaskAppWrapper(MyLog): app = None - def __init__(self, name = __name__, static_url_path = '', log = None, shutter = None, schedule = None, config = None): + def __init__(self, name = __name__, static_url_path = '', log = None, shutter = None, schedule = None, config = None, receiver = None): if log != None: self.log = log logging.getLogger('werkzeug').setLevel(logging.ERROR) - + self.shutter = shutter self.schedule = schedule self.config = config + self.receiver = receiver self.app = Flask(import_name=name, static_url_path="", static_folder=static_url_path) self.app.after_request(self.add_header) @@ -58,7 +60,7 @@ def requestMain(self): def processCommand(self, command): self.LogDebug(request.url + " ( "+ request.method + " ): command=" + command) try: - if command in ["up", "down", "stop", "program", "press", "getConfig", "getStatus", "setPosition", "addSchedule", "editSchedule", "deleteSchedule", "addShutter", "editShutter", "deleteShutter", "setLocation" ]: + if command in ["up", "down", "stop", "program", "press", "getConfig", "getStatus", "setPosition", "addSchedule", "editSchedule", "deleteSchedule", "addShutter", "editShutter", "deleteShutter", "setLocation", "getUnheardRemotes", "assignRemote", "unassignRemote", "setIntermediatePosition" ]: self.LogInfo("processing Command \"" + command + "\" with parameters: "+str(request.values)) result = getattr(self, command)(request.values) return Response(json.dumps(result), status=200) @@ -234,13 +236,77 @@ def deleteSchedule(self, params): def getConfig(self, params): shutters = {} durations = {} + movementStates = {} + positions = {} + intermediatePositions = {} for k in self.config.Shutters: - shutters[k] = self.config.Shutters[k]['name'] - durations[k] = self.config.Shutters[k]['durationDown'] - obj = {'Latitude': self.config.Latitude, 'Longitude': self.config.Longitude, 'Shutters': shutters, 'ShutterDurations': durations, 'Schedule': self.schedule.getScheduleAsDict()} + shutters[k] = self.config.Shutters[k]['name'] + durations[k] = self.config.Shutters[k]['durationDown'] + movementStates[k] = self.shutter.getMovementState(k) + positions[k] = self.shutter.getDisplayPosition(k) + intermediatePositions[k] = self.config.Shutters[k]['intermediatePosition'] + obj = {'Latitude': self.config.Latitude, 'Longitude': self.config.Longitude, 'Shutters': shutters, 'ShutterDurations': durations, 'Schedule': self.schedule.getScheduleAsDict(), 'PhysicalRemotes': self.config.PhysicalRemotes, 'MovementStates': movementStates, 'Positions': positions, 'ShutterIntermediatePositions': intermediatePositions} self.LogDebug("getConfig called, sending: "+json.dumps(obj)) return obj + def getUnheardRemotes(self, params): + remotes = {} + if self.receiver is not None: + # deque is oldest->newest; a plain dict keyed by address keeps + # only the most recent press per address as we iterate forward. + for address, button, rolling_code in self.receiver._unknown_remotes: + remotes[address] = {'address': address, 'button': button, + 'buttonName': button_name(button), + 'rollingCode': rolling_code} + return {'status': 'OK', 'remotes': list(remotes.values())} + + def assignRemote(self, params): + if not self.validatePassword(): + return {'status': 'ERROR'} + address = params.get('address', 0, type=str).strip().lower() + shutterIds = params.to_dict(flat=False).get('shutterIds[]', []) + if not address: + return {'status': 'ERROR', 'message': 'Address is required'} + if not shutterIds: + return {'status': 'ERROR', 'message': 'Select at least one shutter'} + unknown = [s for s in shutterIds if s not in self.config.Shutters] + if unknown: + return {'status': 'ERROR', 'message': 'Unknown shutter(s): ' + ", ".join(unknown)} + self.config.WriteValue(address, ",".join(shutterIds), section="PhysicalRemotes") + self.config.PhysicalRemotes[address] = shutterIds + return {'status': 'OK'} + + def unassignRemote(self, params): + if not self.validatePassword(): + return {'status': 'ERROR'} + address = params.get('address', 0, type=str).strip().lower() + if address not in self.config.PhysicalRemotes: + return {'status': 'ERROR', 'message': 'Remote is not assigned'} + if not self.config.RemoveValue(address, section="PhysicalRemotes"): + return {'status': 'ERROR', 'message': 'Failed to update config file'} + del self.config.PhysicalRemotes[address] + return {'status': 'OK'} + + def setIntermediatePosition(self, params): + if not self.validatePassword(): + return {'status': 'ERROR'} + shutter = params.get('shutter', 0, type=str) + if shutter not in self.config.Shutters: + return {'status': 'ERROR', 'message': 'Shutter does not exist'} + raw = params.get('position', 0, type=str).strip() + if raw == "": + position = None + else: + try: + position = int(raw) + except ValueError: + return {'status': 'ERROR', 'message': 'Position must be a whole number between 0 and 100'} + if position < 0 or position > 100: + return {'status': 'ERROR', 'message': 'Position must be between 0 and 100'} + self.config.WriteValue(shutter, str(position), section="ShutterIntermediatePositions") + self.config.Shutters[shutter]['intermediatePosition'] = position + return {'status': 'OK', 'intermediatePosition': position} + def getStatus(self, params): if not self.validatePassword(): return {'status': 'ERROR'} @@ -248,9 +314,10 @@ def getStatus(self, params): for k in self.config.Shutters: shutters[k] = { 'name': self.config.Shutters[k]['name'], - 'position': self.shutter.getPosition(k), + 'position': self.shutter.getDisplayPosition(k), 'durationUp': self.config.Shutters[k]['durationUp'], - 'durationDown': self.config.Shutters[k]['durationDown'] + 'durationDown': self.config.Shutters[k]['durationDown'], + 'movementState': self.shutter.getMovementState(k) } return {'status': 'OK', 'shutters': shutters}