diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec1ee0c2..2a14f56e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,6 +24,8 @@ jobs: pip install -e ".[dev]" - name: Lint run: ruff check . + - name: Format check + run: ruff format --check . - name: Test run: pytest diff --git a/CHANGELOG.md b/CHANGELOG.md index a26eef85..4fcfd7b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,58 @@ All notable changes to this project are recorded here. The format follows Keep a Changelog; versions follow Semantic Versioning. +## 1.0.2 - 2026-09-05 + +Fixes from a second, adversarial review of 1.0.1 and a re-test of the +first review's findings. No change to the wire format or the public API. + +### Fixed + +- 1.0.1's reply matching dropped a late reply to a request that had + timed out, but not the second reply to a request that was resent after a + silent second and then answered twice. That duplicate carries the counter + of a request that succeeded, and it could still be taken as the answer + to the next request. The library now remembers every recently used + counter and drops any reply carrying one other than the current + request's. A reply whose counter the device has not used recently is + still accepted, for firmware that may not echo it. +- `auth()` reset the session id and key before taking the request lock, so + a request already queued behind the lock could be framed with device id + 0 and the initial key. The reset, the exchange and the install of the + new key now happen as one unit under the lock. +- 1.0.1 let a new capture window close one that a consumer had abandoned, + using "is the generator running right now" as the test. That cannot + tell an abandoned window from one whose consumer is awaiting something + between signals, which the README's own example does. A new window now + gives asyncio's finalizer one turn to close a genuinely dropped + generator and then refuses if the old window is still alive, rather + than taking it. A refused attempt no longer displaces the live window. +- A packet the device returned that cannot be decoded (a declared length + running into a truncated escape) no longer ends the capture window; it + is logged and the window re-arms. +- `aclose()` during a request now raises `EndpointClosedError`, a subclass + of `ConnectionClosedError` with code -4013 in the error table, so a + caller that closed the device on purpose can tell that apart from the + device's own "logged out" answer. +- `hello()` closes the discovery generator it breaks out of instead of + leaving the socket to the finalizer; `asyncio.TimeoutError` is spelled + `TimeoutError`; an unused future on the protocol object is gone. + +### Added + +- Debug logging on the `broadlink.device` and `broadlink.remote` loggers: + endpoint open and close, resends, dropped late replies, timeouts, + re-authentication, capture arm and re-arm, captured packets. +- README: a "Closing" section on the persistent socket, a "Timing" section + with the bench measurement of the tick fix (5.4 percent short before, + 0.6 percent short after, on an RM4 Pro against an independent + receiver), a note that Python 3.13 is a support decision, and the short + list of return-value differences from 0.19.0. + +### Changed + +- The code is formatted with `ruff format` and CI checks it. + ## 1.0.1 - 2026-09-05 Fixes from an independent review of 1.0.0, most of them in the transport. diff --git a/README.md b/README.md index 8edca5e6..e6231ff1 100644 --- a/README.md +++ b/README.md @@ -16,20 +16,25 @@ A Python module and CLI for controlling Broadlink devices locally. ## Version 1.0 is asynchronous -Every call that reaches a device is a coroutine and must be awaited. This -is the whole change from the original library's API; method names, -arguments and return values are the same. +Every call that reaches a device is a coroutine and must be awaited. That +is the main change from the original library's API: method names and +arguments are the same, and so are return values, with the small +exceptions listed in `CHANGELOG.md` (the IR tick constant, `pulses_to_data` +returning `bytes`, the unused `Device.lock` attribute removed, and +`timeout` parameters typed as floats). ```python import asyncio import broadlink + async def main(): devices = await broadlink.discover(timeout=5) device = devices[0] await device.auth() print(await device.check_sensors()) + asyncio.run(main()) ``` @@ -52,8 +57,32 @@ The following devices are supported: - **Thermostats**: Hysen HY02B05H - **Hubs**: S3 +## Timing + +The original library converted microseconds to the device's timing units +with the constant 32.84, which is the right ratio applied the wrong way +round, and it shortened every IR code built from microsecond timings by +about 7 percent. Codes learned from a remote and replayed through the same +device were never affected, which is why it went unnoticed for years. +Version 1.0 uses 8192/269 (about 30.45 us per unit), the value implied by +`protocol.md`, and rounds to the nearest unit instead of truncating. + +Measured on an RM4 Pro against an independent receiver, the same NEC frame +packed with the old constant arrived 5.4 percent short of its intended +length; packed with the corrected constant it arrived 0.6 percent short, +twice, thirteen hours apart, within 22 us of itself. Packets learned by +the device and replayed by name are unchanged. Anything that stores +microsecond timings produced by the old `data_to_pulses` (which reported +them about 7.8 percent long) and re-encodes them with the new +`pulses_to_data` will lengthen by that amount; store the device packet +instead, as `CapturedSignal.packet` does. + ## Installation +Python 3.13 or newer. That is a support decision rather than a technical +one: the code runs on 3.11, but the versions tested in CI are 3.13 and +3.14 and those are the ones Home Assistant ships. + Use pip3 to install the latest version of this module. ``` @@ -94,7 +123,7 @@ In order to control the device, you need to connect it to your local network. If - Manually connect to the WiFi SSID named BroadlinkProv. 2. Connect the device to your local network with the setup function. ```python3 -await broadlink.setup('myssid', 'mynetworkpass', 3) +await broadlink.setup("myssid", "mynetworkpass", 3) ``` Security mode options are (0 = none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2) @@ -103,7 +132,7 @@ Security mode options are (0 = none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2) You may need to specify a broadcast address if setup is not working. ```python3 -await broadlink.setup('myssid', 'mynetworkpass', 3, ip_address='192.168.0.255') +await broadlink.setup("myssid", "mynetworkpass", 3, ip_address="192.168.0.255") ``` ### Discovery @@ -119,17 +148,17 @@ You may need to specify `local_ip_address` or `discover_ip_address` if discovery Using the IP address of your local machine: ```python3 -devices = await broadlink.discover(local_ip_address='192.168.0.100') +devices = await broadlink.discover(local_ip_address="192.168.0.100") ``` Using the broadcast address of your subnet: ```python3 -devices = await broadlink.discover(discover_ip_address='192.168.0.255') +devices = await broadlink.discover(discover_ip_address="192.168.0.255") ``` If the device is locked, it may not be discoverable with broadcast. In such cases, you can use the unicast version `broadlink.hello()` for direct discovery: ```python3 -device = await broadlink.hello('192.168.0.16') +device = await broadlink.hello("192.168.0.16") ``` If you are a perfomance freak, use `broadlink.xdiscover()` to create devices instantly: @@ -144,6 +173,27 @@ After discovering the device, call the `auth()` method to obtain the authenticat await device.auth() ``` +### Closing + +Each device keeps one UDP socket open for its lifetime (the original +library opened a new one for every call). Close it when you are done with +the device, either with the context manager or explicitly: + +```python3 +async with device: + await device.auth() + print(await device.check_sensors()) + +# or +await device.aclose() +``` + +The socket reopens by itself on the next call, so closing is cheap and +safe to do at any time. A request that is in flight when `aclose()` runs +fails with `EndpointClosedError`. An integration that creates devices +should close them when it unloads; a device that is never closed holds +its socket until it is garbage collected. + The next steps depend on the type of device you want to control. ## Universal remotes @@ -175,7 +225,7 @@ await device.sweep_frequency() ```python3 ok, frequency = await device.check_frequency() if ok: - print(f'Frequency found: {frequency} MHz') + print(f"Frequency found: {frequency} MHz") ``` 4. Enter learning mode: ```python3 @@ -217,10 +267,13 @@ By default the window closes after the first signal. Pass `window=0` runs until the generator is closed), re-arming after each signal because the device holds only one code per learning session. A universal remote has a single receiver, so only one capture window can be open on a -device at a time. +device at a time: opening a second one raises `CaptureInProgressError` +while the first is still held. Always close a window you leave early +(`aclosing` above does it), otherwise it stays open until Python collects +the generator. `CapturedSignal` carries the device's own `packet` bytes (ready for -`send_data`), the decoded `pulses` in microseconds at the correct tick, the +`send_data`), the decoded `pulses` in microseconds at the corrected tick, the `kind` (`SignalKind.IR`, `RF_433` or `RF_315`), the `repeat` count, and for RF the `frequency_mhz` the packet itself does not record. diff --git a/broadlink/__init__.py b/broadlink/__init__.py index eab43e72..632afdf3 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -1,7 +1,9 @@ #!/usr/bin/env python3 """The python-broadlink library.""" + +import contextlib from collections.abc import AsyncIterator -from typing import List, Optional, Tuple, Union +from typing import Optional, Union from . import exceptions as e from .alarm import S1C @@ -223,8 +225,8 @@ def gendevice( dev_type: int, - host: Tuple[str, int], - mac: Union[bytes, str], + host: tuple[str, int], + mac: bytes | str, name: str = "", is_locked: bool = False, ) -> Device: @@ -258,12 +260,15 @@ async def hello( Useful if the device is locked. """ - async for device in xdiscover( - timeout=timeout, - discover_ip_address=ip_address, - discover_ip_port=port, - ): - return device + async with contextlib.aclosing( + xdiscover( + timeout=timeout, + discover_ip_address=ip_address, + discover_ip_port=port, + ) + ) as devices: + async for device in devices: + return device raise e.NetworkTimeoutError( -4000, "Network timeout", @@ -273,10 +278,10 @@ async def hello( async def discover( timeout: float = DEFAULT_TIMEOUT, - local_ip_address: Optional[str] = None, + local_ip_address: str | None = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, -) -> List[Device]: +) -> list[Device]: """Discover devices connected to the local network.""" return [ device @@ -288,7 +293,7 @@ async def discover( async def xdiscover( timeout: float = DEFAULT_TIMEOUT, - local_ip_address: Optional[str] = None, + local_ip_address: str | None = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, ) -> AsyncIterator[Device]: diff --git a/broadlink/alarm.py b/broadlink/alarm.py index 2c3358de..a7d30549 100644 --- a/broadlink/alarm.py +++ b/broadlink/alarm.py @@ -1,4 +1,5 @@ """Support for alarm kits.""" + from . import exceptions as e from .device import Device diff --git a/broadlink/climate.py b/broadlink/climate.py index 5d75457d..6841f87d 100644 --- a/broadlink/climate.py +++ b/broadlink/climate.py @@ -1,7 +1,8 @@ """Support for climate control.""" + import enum import struct -from typing import List, Sequence +from collections.abc import Sequence from . import exceptions as e from .device import Device @@ -33,7 +34,7 @@ async def send_request(self, request: Sequence[int]) -> bytes: payload = self.decrypt(response[0x38:]) p_len = int.from_bytes(payload[:0x02], "little") - nom_crc = int.from_bytes(payload[p_len:p_len+2], "little") + nom_crc = int.from_bytes(payload[p_len : p_len + 2], "little") real_crc = CRC16.calculate(payload[0x02:p_len]) if nom_crc != real_crc: @@ -83,9 +84,7 @@ async def get_full_status(self) -> dict: data["dif"] = payload[10] data["svh"] = payload[11] data["svl"] = payload[12] - data["room_temp_adj"] = ( - int.from_bytes(payload[13:15], "big", signed=True) / 10.0 - ) + data["room_temp_adj"] = int.from_bytes(payload[13:15], "big", signed=True) / 10.0 data["fre"] = payload[15] data["poweron"] = payload[16] data["unknown"] = payload[17] @@ -127,9 +126,7 @@ async def get_full_status(self) -> dict: # E.g. loop_mode = 0 ("12345,67") means Saturday and Sunday (weekend schedule) # loop_mode = 2 ("1234567") means every day, including Saturday and Sunday (weekday schedule) # The sensor command is currently experimental - async def set_mode( - self, auto_mode: int, loop_mode: int, sensor: int = 0 - ) -> None: + async def set_mode(self, auto_mode: int, loop_mode: int, sensor: int = 0) -> None: """Set the mode of the device.""" mode_byte = ((loop_mode + 1) << 4) + auto_mode await self.send_request([0x01, 0x06, 0x00, 0x02, mode_byte, sensor]) @@ -210,19 +207,7 @@ async def set_power( async def set_time(self, hour: int, minute: int, second: int, day: int) -> None: """Set the time.""" await self.send_request( - [ - 0x01, - 0x10, - 0x00, - 0x08, - 0x00, - 0x02, - 0x04, - hour, - minute, - second, - day - ] + [0x01, 0x10, 0x00, 0x08, 0x00, 0x02, 0x04, hour, minute, second, day] ) # Set timer schedule @@ -231,7 +216,7 @@ async def set_time(self, hour: int, minute: int, second: int, day: int) -> None: # {'start_hour':17, 'start_minute':30, 'temp': 22 } # Each one specifies the thermostat temp that will become effective at start_hour:start_minute # weekend is similar but only has 2 (e.g. switch on in morning and off in afternoon) - async def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None: + async def set_schedule(self, weekday: list[dict], weekend: list[dict]) -> None: """Set timer schedule.""" request = [0x01, 0x10, 0x00, 0x0A, 0x00, 0x0C, 0x18] @@ -317,9 +302,7 @@ def _encode(self, data: bytes) -> bytes: """Encode data for transport.""" packet = bytearray(10) p_len = 10 + len(data) - struct.pack_into( - " bytes: # payload[0x2:0x8] == bytes([0xbb, 0x00, 0x07, 0x00, 0x00, 0x00]) payload = self.decrypt(response[0x38:]) p_len = int.from_bytes(payload[:0x02], "little") - nom_crc = int.from_bytes(payload[p_len:p_len+2], "little") + nom_crc = int.from_bytes(payload[p_len : p_len + 2], "little") real_crc = CRC16.calculate(payload[0x02:p_len], polynomial=0x9BE4) if nom_crc != real_crc: @@ -341,7 +324,7 @@ def _decode(self, response: bytes) -> bytes: ) d_len = int.from_bytes(payload[0x08:0x0A], "little") - return payload[0x0A:0x0A+d_len] + return payload[0x0A : 0x0A + d_len] async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a command to the unit.""" diff --git a/broadlink/const.py b/broadlink/const.py index 19c37f52..0ebf6cf4 100644 --- a/broadlink/const.py +++ b/broadlink/const.py @@ -1,4 +1,5 @@ """Constants.""" + DEFAULT_BCAST_ADDR = "255.255.255.255" DEFAULT_PORT = 80 DEFAULT_RETRY_INTVL = 1 diff --git a/broadlink/cover.py b/broadlink/cover.py index 0319457a..5c38bd04 100644 --- a/broadlink/cover.py +++ b/broadlink/cover.py @@ -1,6 +1,7 @@ """Support for covers.""" + import asyncio -from typing import Sequence +from collections.abc import Sequence from . import exceptions as e from .device import Device diff --git a/broadlink/device.py b/broadlink/device.py index 4ac3988c..2315f307 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -11,10 +11,11 @@ import asyncio import collections +import contextlib +import logging import random import socket from collections.abc import AsyncIterator -from typing import Optional, Tuple, Union from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes @@ -28,17 +29,21 @@ ) from .protocol import Datetime -HelloResponse = Tuple[int, Tuple[str, int], bytes, str, bool] +_LOGGER = logging.getLogger(__name__) + +HelloResponse = tuple[int, tuple[str, int], bytes, str, bool] # Device error codes that mean the session key is no longer accepted and a # fresh auth() will fix it. -2: logged out; -7: control key expired; # -4012: control id error. _REAUTH_CODES = {-2, -7, -4012} -# How many timed-out request counters to remember, so that a reply to one -# of them arriving late is recognised and dropped instead of being taken as -# the answer to a later request. -_ABANDONED_MAX = 32 +# How many recently used request counters to remember. A reply carrying one +# of them (other than the current request's) is a late or duplicate answer +# to an earlier request and is dropped rather than taken as the answer to +# the current one. 64 covers a burst of resends comfortably and ages out +# long before the 16-bit counter wraps. +_RECENT_MAX = 64 _CLOSED = (None, None) """Sentinel put on the receive queue when the endpoint is closed.""" @@ -49,8 +54,7 @@ class _Protocol(asyncio.DatagramProtocol): def __init__(self) -> None: self.queue: asyncio.Queue[tuple[bytes, tuple[str, int]]] = asyncio.Queue() - self.transport: Optional[asyncio.DatagramTransport] = None - self.closed = asyncio.get_running_loop().create_future() + self.transport: asyncio.DatagramTransport | None = None def connection_made(self, transport) -> None: # type: ignore[override] self.transport = transport @@ -63,9 +67,8 @@ def error_received(self, exc: Exception) -> None: # the retry loop will time out and raise NetworkTimeoutError. pass - def connection_lost(self, exc: Optional[Exception]) -> None: - if not self.closed.done(): - self.closed.set_result(None) + def connection_lost(self, exc: Exception | None) -> None: + pass def drain(self) -> None: """Drop anything that arrived before the current request.""" @@ -74,8 +77,8 @@ def drain(self) -> None: async def _open_endpoint( - local_addr: Optional[tuple[str, int]] = None, - remote_addr: Optional[tuple[str, int]] = None, + local_addr: tuple[str, int] | None = None, + remote_addr: tuple[str, int] | None = None, broadcast: bool = False, ) -> tuple[asyncio.DatagramTransport, _Protocol]: """Create a UDP endpoint. Tests replace this to fake the network.""" @@ -111,7 +114,7 @@ def _parse_hello(resp: bytes, host: tuple[str, int]) -> HelloResponse: async def scan( timeout: float = DEFAULT_TIMEOUT, - local_ip_address: Optional[str] = None, + local_ip_address: str | None = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, ) -> AsyncIterator[HelloResponse]: @@ -144,7 +147,7 @@ async def scan( break try: resp, host = await asyncio.wait_for(protocol.queue.get(), remaining) - except asyncio.TimeoutError: + except TimeoutError: break if len(resp) < 0x80: continue @@ -184,8 +187,8 @@ class Device: def __init__( self, - host: Tuple[str, int], - mac: Union[bytes, str], + host: tuple[str, int], + mac: bytes | str, devtype: int, timeout: float = DEFAULT_TIMEOUT, name: str = "", @@ -210,42 +213,32 @@ def __init__( self.aes = None self.update_aes(bytes.fromhex(self.__INIT_KEY)) - self._lock: Optional[asyncio.Lock] = None - self._transport: Optional[asyncio.DatagramTransport] = None - self._protocol: Optional[_Protocol] = None - self._endpoint_addr: Optional[Tuple[str, int]] = None - self._abandoned: collections.deque[int] = collections.deque(maxlen=_ABANDONED_MAX) - self._reauth_lock: Optional[asyncio.Lock] = None + self._lock: asyncio.Lock | None = None + self._transport: asyncio.DatagramTransport | None = None + self._protocol: _Protocol | None = None + self._endpoint_addr: tuple[str, int] | None = None + self._recent: collections.deque[int] = collections.deque(maxlen=_RECENT_MAX) + self._reauth_lock: asyncio.Lock | None = None self._auth_generation = 0 def __repr__(self) -> str: """Return a formal representation of the device.""" return ( - "%s.%s(%s, mac=%r, devtype=%r, timeout=%r, name=%r, " - "model=%r, manufacturer=%r, is_locked=%r)" - ) % ( - self.__class__.__module__, - self.__class__.__qualname__, - self.host, - self.mac, - self.devtype, - self.timeout, - self.name, - self.model, - self.manufacturer, - self.is_locked, + f"{self.__class__.__module__}.{self.__class__.__qualname__}(" + f"{self.host}, mac={self.mac!r}, devtype={self.devtype!r}, " + f"timeout={self.timeout!r}, name={self.name!r}, " + f"model={self.model!r}, manufacturer={self.manufacturer!r}, " + f"is_locked={self.is_locked!r})" ) def __str__(self) -> str: """Return a readable representation of the device.""" - return "%s (%s / %s:%s / %s)" % ( - self.name or "Unknown", - " ".join(filter(None, [self.manufacturer, self.model, hex(self.devtype)])), - *self.host, - ":".join(format(x, "02X") for x in self.mac), - ) + ident = " ".join(filter(None, [self.manufacturer, self.model, hex(self.devtype)])) + mac = ":".join(format(x, "02X") for x in self.mac) + name = self.name or "Unknown" + return f"{name} ({ident} / {self.host[0]}:{self.host[1]} / {mac})" - async def __aenter__(self) -> "Device": + async def __aenter__(self) -> Device: return self async def __aexit__(self, *exc) -> None: @@ -272,23 +265,31 @@ def decrypt(self, payload: bytes) -> bytes: # ---------------------------------------------------------- session async def auth(self) -> bool: - """Authenticate to the device.""" - self.id = 0 - self.update_aes(bytes.fromhex(self.__INIT_KEY)) + """Authenticate to the device. + The session reset, the exchange and the install of the new key all + happen while holding the request lock, so a request queued behind + the lock is never framed with the initial key or device id 0. + """ packet = bytearray(0x50) packet[0x04:0x14] = [0x31] * 16 packet[0x1E] = 0x01 packet[0x2D] = 0x01 - packet[0x30:0x36] = "Test 1".encode() - - response = await self.send_packet(0x65, packet, _reauth=False) - e.check_error(response[0x22:0x24]) - payload = self.decrypt(response[0x38:]) + packet[0x30:0x36] = b"Test 1" - self.id = int.from_bytes(payload[:0x4], "little") - self.update_aes(payload[0x04:0x14]) - self._auth_generation += 1 + if self._lock is None: + self._lock = asyncio.Lock() + self._reauth_lock = asyncio.Lock() + async with self._lock: + self.id = 0 + self.update_aes(bytes.fromhex(self.__INIT_KEY)) + response = await self._exchange(self._frame(0x65, bytes(packet))) + e.check_error(response[0x22:0x24]) + payload = self.decrypt(response[0x38:]) + self.id = int.from_bytes(payload[:0x4], "little") + self.update_aes(payload[0x04:0x14]) + self._auth_generation += 1 + _LOGGER.debug("%s: authenticated, session id %d", self.host[0], self.id) return True async def hello(self, local_ip_address=None) -> bool: @@ -296,15 +297,17 @@ async def hello(self, local_ip_address=None) -> bool: Device information is checked before updating name and lock status. """ - responses = scan( - timeout=self.timeout, - local_ip_address=local_ip_address, - discover_ip_address=self.host[0], - discover_ip_port=self.host[1], - ) entry = None - async for entry in responses: - break + async with contextlib.aclosing( + scan( + timeout=self.timeout, + local_ip_address=local_ip_address, + discover_ip_address=self.host[0], + discover_ip_port=self.host[1], + ) + ) as responses: + async for entry in responses: # noqa: B007 - first reply only + break if entry is None: raise e.NetworkTimeoutError( -4000, @@ -388,6 +391,7 @@ async def aclose(self) -> None: self._endpoint_addr = None if transport is not None: transport.close() + _LOGGER.debug("%s: endpoint closed", self.host[0]) if protocol is not None: protocol.queue.put_nowait(_CLOSED) # type: ignore[arg-type] @@ -397,10 +401,9 @@ async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: # old address, so drop it. await self.aclose() if self._transport is None or self._transport.is_closing(): - self._transport, self._protocol = await _open_endpoint( - remote_addr=self.host - ) + self._transport, self._protocol = await _open_endpoint(remote_addr=self.host) self._endpoint_addr = self.host + _LOGGER.debug("%s: endpoint opened", self.host[0]) return self._transport, self._protocol # type: ignore[return-value] def _frame(self, packet_type: int, payload: bytes) -> bytes: @@ -450,8 +453,9 @@ async def _exchange(self, packet: bytes) -> bytes: Replies carry the request's packet counter (offset 0x28), so a reply is matched to the request by counter. A reply whose counter belongs - to a request that already timed out is dropped; one with a counter - this device has never sent is accepted, for firmware that may not + to any other recent request (a late answer, or the second answer to + a request that was resent) is dropped; one with a counter this + device has not used recently is accepted, for firmware that may not echo it. """ transport, protocol = await self._endpoint() @@ -460,9 +464,14 @@ async def _exchange(self, packet: bytes) -> bytes: start = loop.time() timeout = self.timeout count = int.from_bytes(packet[0x28:0x2A], "little") + self._recent.append(count) + sends = 0 while True: transport.sendto(packet) + sends += 1 + if sends > 1: + _LOGGER.debug("%s: no reply, resending (%d)", self.host[0], sends) resend_at = loop.time() + DEFAULT_RETRY_INTVL while True: now = loop.time() @@ -471,32 +480,34 @@ async def _exchange(self, packet: bytes) -> bytes: wait = min(resend_at, start + timeout) - now try: resp, _ = await asyncio.wait_for(protocol.queue.get(), max(wait, 0)) - except asyncio.TimeoutError: + except TimeoutError: if loop.time() - start >= timeout: break if loop.time() >= resend_at: break # Resend. continue if resp is None: - raise e.ConnectionClosedError( - -4013, "Connection closed", "The device endpoint was closed" + raise e.EndpointClosedError( + -4013, "Endpoint closed", "The device endpoint was closed" ) resp = self._validate(resp) reply_count = int.from_bytes(resp[0x28:0x2A], "little") - if reply_count == count or reply_count not in self._abandoned: + if reply_count == count or reply_count not in self._recent: return resp - # A late answer to a request we gave up on: keep waiting. + _LOGGER.debug( + "%s: dropped a reply for an earlier request (counter 0x%04x)", + self.host[0], + reply_count, + ) if loop.time() - start >= timeout: - self._abandoned.append(count) + _LOGGER.debug("%s: no reply within %ss", self.host[0], timeout) raise e.NetworkTimeoutError( -4000, "Network timeout", f"No response received within {timeout}s", ) from None - async def send_packet( - self, packet_type: int, payload: bytes, *, _reauth: bool = True - ) -> bytes: + async def send_packet(self, packet_type: int, payload: bytes) -> bytes: """Send a packet to the device and return the raw response frame. If the device answers that the session key is no longer valid, the @@ -511,14 +522,12 @@ async def send_packet( async with self._lock: resp = await self._exchange(self._frame(packet_type, bytes(payload))) - if _reauth: - code = int.from_bytes(resp[0x22:0x24], "little", signed=True) - if code in _REAUTH_CODES: - async with self._reauth_lock: # type: ignore[union-attr] - if self._auth_generation == generation: - await self.auth() - async with self._lock: - resp = await self._exchange( - self._frame(packet_type, bytes(payload)) - ) + code = int.from_bytes(resp[0x22:0x24], "little", signed=True) + if code in _REAUTH_CODES: + _LOGGER.debug("%s: device answered %d, re-authenticating", self.host[0], code) + async with self._reauth_lock: # type: ignore[union-attr] + if self._auth_generation == generation: + await self.auth() + async with self._lock: + resp = await self._exchange(self._frame(packet_type, bytes(payload))) return resp diff --git a/broadlink/exceptions.py b/broadlink/exceptions.py index 8f2ecc6c..f5b56d10 100644 --- a/broadlink/exceptions.py +++ b/broadlink/exceptions.py @@ -1,4 +1,5 @@ """Exceptions for Broadlink devices.""" + import collections import struct @@ -22,7 +23,7 @@ def __init__(self, *args, **kwargs): def __str__(self): """Return str(self).""" if self.errno is not None: - return "[Errno %s] %s" % (self.errno, self.strerror) + return f"[Errno {self.errno}] {self.strerror}" return self.strerror def __eq__(self, other): @@ -42,13 +43,13 @@ def __init__(self, *args, **kwargs): """Initialize the exception.""" errors = args[0][:] if args else [] counter = collections.Counter(errors) - strerror = "Multiple errors occurred: %s" % counter + strerror = f"Multiple errors occurred: {counter}" super().__init__(strerror, **kwargs) self.errors = errors def __repr__(self): """Return repr(self).""" - return "MultipleErrors(%r)" % self.errors + return f"MultipleErrors({self.errors!r})" def __str__(self): """Return str(self).""" @@ -71,6 +72,16 @@ class ConnectionClosedError(BroadlinkException): """Connection closed error.""" +class EndpointClosedError(ConnectionClosedError): + """The library's own endpoint was closed while a request was in flight. + + Raised locally by ``Device.aclose()``, not by the device. It is a + subclass of ``ConnectionClosedError`` so existing handlers still catch + it, and a distinct class so a caller that closed the device on purpose + can tell it apart from the device's "logged out" (-2) answer. + """ + + class StructureAbnormalError(BroadlinkException): """Structure abnormal error.""" @@ -142,6 +153,7 @@ class UnknownError(BroadlinkException): -4010: (DataValidationError, "Received encrypted data packet length error"), -4011: (DataValidationError, "Received encrypted data packet check error"), -4012: (AuthorizationError, "Device control ID error"), + -4013: (EndpointClosedError, "Endpoint closed"), } diff --git a/broadlink/helpers.py b/broadlink/helpers.py index e7b3d4c9..c41cb84e 100644 --- a/broadlink/helpers.py +++ b/broadlink/helpers.py @@ -1,5 +1,6 @@ """Helper functions and classes.""" -from typing import Dict, List, Sequence + +from collections.abc import Sequence class CRC16: @@ -8,10 +9,10 @@ class CRC16: CRC tables are cached for performance. """ - _cache: Dict[int, List[int]] = {} + _cache: dict[int, list[int]] = {} @classmethod - def get_table(cls, polynomial: int) -> List[int]: + def get_table(cls, polynomial: int) -> list[int]: """Return the CRC-16 table for a polynomial.""" try: crc_table = cls._cache[polynomial] diff --git a/broadlink/hub.py b/broadlink/hub.py index 1d74041f..6dd4ed40 100644 --- a/broadlink/hub.py +++ b/broadlink/hub.py @@ -1,7 +1,7 @@ """Support for hubs.""" + import json import struct -from typing import Optional from . import exceptions as e from .device import Device @@ -43,7 +43,7 @@ async def get_subdevices(self, step: int = 5) -> list: return sub_devices - async def get_state(self, did: Optional[str] = None) -> dict: + async def get_state(self, did: str | None = None) -> dict: """Return the power state of the device.""" state = {} if did is not None: @@ -56,10 +56,10 @@ async def get_state(self, did: Optional[str] = None) -> dict: async def set_state( self, - did: Optional[str] = None, - pwr1: Optional[bool] = None, - pwr2: Optional[bool] = None, - pwr3: Optional[bool] = None, + did: str | None = None, + pwr1: bool | None = None, + pwr2: bool | None = None, + pwr3: bool | None = None, ) -> dict: """Set the power state of the device.""" state = {} @@ -82,9 +82,7 @@ def _encode(self, flag: int, state: dict) -> bytes: # flag: 1 for reading, 2 for writing. packet = bytearray(12) data = json.dumps(state, separators=(",", ":")).encode() - struct.pack_into( - " dict: """Decode a JSON packet.""" payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" dict: async def set_state( self, - pwr: Optional[bool] = None, - red: Optional[int] = None, - blue: Optional[int] = None, - green: Optional[int] = None, - brightness: Optional[int] = None, - colortemp: Optional[int] = None, - hue: Optional[int] = None, - saturation: Optional[int] = None, - transitionduration: Optional[int] = None, - maxworktime: Optional[int] = None, - bulb_colormode: Optional[int] = None, - bulb_scenes: Optional[str] = None, - bulb_scene: Optional[str] = None, - bulb_sceneidx: Optional[int] = None, + pwr: bool | None = None, + red: int | None = None, + blue: int | None = None, + green: int | None = None, + brightness: int | None = None, + colortemp: int | None = None, + hue: int | None = None, + saturation: int | None = None, + transitionduration: int | None = None, + maxworktime: int | None = None, + bulb_colormode: int | None = None, + bulb_scenes: str | None = None, + bulb_scene: str | None = None, + bulb_sceneidx: int | None = None, ) -> dict: """Set the power state of the device.""" state = {} @@ -102,7 +102,7 @@ def _decode(self, response: bytes) -> dict: """Decode a JSON packet.""" payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" dict: async def set_state( self, - pwr: Optional[bool] = None, - red: Optional[int] = None, - blue: Optional[int] = None, - green: Optional[int] = None, - brightness: Optional[int] = None, - colortemp: Optional[int] = None, - hue: Optional[int] = None, - saturation: Optional[int] = None, - transitionduration: Optional[int] = None, - maxworktime: Optional[int] = None, - bulb_colormode: Optional[int] = None, - bulb_scenes: Optional[str] = None, - bulb_scene: Optional[str] = None, + pwr: bool | None = None, + red: int | None = None, + blue: int | None = None, + green: int | None = None, + brightness: int | None = None, + colortemp: int | None = None, + hue: int | None = None, + saturation: int | None = None, + transitionduration: int | None = None, + maxworktime: int | None = None, + bulb_colormode: int | None = None, + bulb_scenes: str | None = None, + bulb_scene: str | None = None, ) -> dict: """Set the power state of the device.""" state = {} @@ -184,9 +184,7 @@ def _encode(self, flag: int, state: dict) -> bytes: # flag: 1 for reading, 2 for writing. packet = bytearray(12) data = json.dumps(state, separators=(",", ":")).encode() - struct.pack_into( - " dict: """Decode a JSON packet.""" payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" "SignalKind": def pulses_to_data( - pulses: List[int], + pulses: list[int], tick: float = TICK, *, kind: SignalKind = SignalKind.IR, @@ -103,7 +106,7 @@ def pulses_to_data( return bytes(result) -def data_to_pulses(data: bytes, tick: float = TICK) -> List[int]: +def data_to_pulses(data: bytes, tick: float = TICK) -> list[int]: """Parse a Broadlink packet into a microsecond duration sequence.""" result = [] index = 4 @@ -136,7 +139,7 @@ class ParsedPacket: kind: SignalKind repeat: int - pulses: List[int] + pulses: list[int] type_byte: int @@ -167,19 +170,19 @@ class CapturedSignal: packet: bytes kind: SignalKind - pulses: List[int] = field(repr=False) + pulses: list[int] = field(repr=False) repeat: int = 0 - frequency_mhz: Optional[float] = None - type_byte: Optional[int] = None + frequency_mhz: float | None = None + type_byte: int | None = None captured_at: float = field(default_factory=time.time, repr=False) @classmethod def from_packet( cls, packet: bytes, - frequency_mhz: Optional[float] = None, + frequency_mhz: float | None = None, *, - kind: Optional[SignalKind] = None, + kind: SignalKind | None = None, ) -> "CapturedSignal": """Build a signal from a device-returned packet. @@ -217,7 +220,7 @@ def __init__(self, *args, **kwargs) -> None: self._tx_generation = 0 # Weak reference to the async generator of the current capture # window, if any. See _claim_window. - self._window: Optional[weakref.ReferenceType] = None + self._window: weakref.ReferenceType | None = None @property def capture_active(self) -> bool: @@ -225,30 +228,38 @@ def capture_active(self) -> bool: window = self._window() if self._window is not None else None return window is not None and window.ag_frame is not None - def _check_window(self) -> Optional[weakref.ReferenceType]: - """Refuse a new window while another is being iterated; return the - reference to a previous window that the new one should close.""" + def _check_window(self) -> None: + """Fail fast at call time if another window is being iterated now.""" old = self._window() if self._window is not None else None - if old is None or old.ag_frame is None: - return None - if old.ag_running: + if old is not None and old.ag_frame is not None and old.ag_running: raise e.CaptureInProgressError("A capture window is already open") - return self._window - async def _claim_window(self, prev: Optional[weakref.ReferenceType]) -> None: - """Close a previous window whose consumer walked away from it. + async def _claim_window(self, new: weakref.ReferenceType) -> None: + """Make sure the previous window is really gone, then register ``new``. - A window whose consumer is still iterating it is live and a new one - is refused at call time (``_check_window``). One the consumer broke - out of without closing the generator is not live; it is closed here - so that it cannot block the device forever. + A consumer that walked away from a window without closing it (for + example ``break`` out of ``async for`` with no ``aclosing``) leaves + the generator to asyncio's finalizer, which closes it on the next + loop iteration once nothing references it. Give that a turn. If the + window is still alive after that, someone still holds it, whether + they are inside ``__anext__`` or paused between signals, and the new + window is refused rather than taken from under them. """ - old = prev() if prev is not None else None - if old is None or old.ag_frame is None: - return - if old.ag_running: - raise e.CaptureInProgressError("A capture window is already open") - await old.aclose() + prev = self._window + if prev is not None: + old = prev() + if old is not None and old.ag_frame is not None: + if old.ag_running: + raise e.CaptureInProgressError("A capture window is already open") + del old # Hold no reference while the finalizer gets its turn. + await asyncio.sleep(0) + await asyncio.sleep(0) + old = prev() + if old is not None and old.ag_frame is not None: + raise e.CaptureInProgressError( + "A capture window is already open; close it with aclose() first" + ) + self._window = new async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" @@ -301,11 +312,13 @@ def capture( generator sends nothing further; the device times out by itself. Use ``contextlib.aclosing`` (or iterate to the end) so the window is released promptly. Only one capture window can be open per device: - opening one while another is being iterated raises - ``CaptureInProgressError``, and opening one after breaking out of - another without closing it closes the old one. + opening one while another is still held raises + ``CaptureInProgressError``. A window whose generator was dropped + without being closed is finalized by asyncio on the next loop + iteration and does not block. """ - prev = self._check_window() + self._check_window() + holder: list = [] gen = self._capture_loop( self.enter_learning, window, @@ -314,9 +327,9 @@ def capture( rearm_interval, SignalKind.IR, None, - prev=prev, + claim=holder, ) - self._window = weakref.ref(gen) + holder.append(weakref.ref(gen)) return gen async def _capture_loop( @@ -327,17 +340,19 @@ async def _capture_loop( poll_interval: float, rearm_interval: float, kind: SignalKind, - frequency_mhz: Optional[float], + frequency_mhz: float | None, *, - prev: Optional[weakref.ReferenceType] = None, - claim: bool = True, + claim: list | None = None, ) -> AsyncIterator[CapturedSignal]: + # ``claim`` carries a weak reference to this generator (filled in by + # the caller after creating it); None means the caller owns the + # window claim, as capture_rf does for its inner loop. if window < 0: raise ValueError("window must be 0 (open-ended) or positive") if poll_interval <= 0 or rearm_interval <= 0: raise ValueError("poll_interval and rearm_interval must be positive") if claim: - await self._claim_window(prev) + await self._claim_window(claim[0]) loop = asyncio.get_running_loop() deadline = loop.time() + window if window else None @@ -346,6 +361,7 @@ async def _capture_loop( await arm() armed_at = loop.time() generation = self._tx_generation + _LOGGER.debug("%s: capture window armed (%s)", self.host[0], kind.name) while True: now = loop.time() @@ -372,16 +388,33 @@ async def _capture_loop( timeouts = 0 if data: - yield CapturedSignal.from_packet(data, frequency_mhz, kind=kind) - if stop_after_first: - return - generation = -1 # One code per session: re-arm. + try: + signal = CapturedSignal.from_packet(data, frequency_mhz, kind=kind) + except ValueError as err: + # A packet the device returned but we cannot decode. Log + # it, re-arm and keep the window open. + _LOGGER.warning( + "%s: ignoring an undecodable capture (%s): %s", + self.host[0], + err, + data.hex(), + ) + generation = -1 + else: + _LOGGER.debug( + "%s: captured %d bytes (%s)", self.host[0], len(data), kind.name + ) + yield signal + if stop_after_first: + return + generation = -1 # One code per session: re-arm. now = loop.time() if generation != self._tx_generation or now - armed_at >= rearm_interval: await arm() armed_at = loop.time() generation = self._tx_generation + _LOGGER.debug("%s: capture window re-armed", self.host[0]) class rmpro(rmmini): @@ -393,14 +426,14 @@ async def sweep_frequency(self) -> None: """Sweep frequency.""" await self._send(0x19) - async def check_frequency(self) -> Tuple[bool, float]: + async def check_frequency(self) -> tuple[bool, float]: """Return True if the frequency was identified successfully.""" resp = await self._send(0x1A) is_found = bool(resp[0]) frequency = struct.unpack(" None: + async def find_rf_packet(self, frequency: float | None = None) -> None: """Enter radiofrequency learning mode.""" payload = bytearray() if frequency: @@ -415,7 +448,7 @@ def capture_rf( self, window: float = 30.0, *, - frequency: Optional[float] = None, + frequency: float | None = None, stop_after_first: bool = True, poll_interval: float = DEFAULT_POLL_INTERVAL, rearm_interval: float = DEFAULT_REARM_INTERVAL, @@ -436,27 +469,32 @@ def capture_rf( Each ``CapturedSignal`` carries the carrier in ``frequency_mhz``, which the packet itself does not record. """ - prev = self._check_window() + self._check_window() + holder: list = [] gen = self._capture_rf_loop( - window, frequency, stop_after_first, poll_interval, rearm_interval, - prev=prev, + window, + frequency, + stop_after_first, + poll_interval, + rearm_interval, + claim=holder, ) - self._window = weakref.ref(gen) + holder.append(weakref.ref(gen)) return gen async def _capture_rf_loop( self, window: float, - frequency: Optional[float], + frequency: float | None, stop_after_first: bool, poll_interval: float, rearm_interval: float, *, - prev: Optional[weakref.ReferenceType], + claim: list, ) -> AsyncIterator[CapturedSignal]: if window < 0 or poll_interval <= 0: raise ValueError("window must be 0 or positive, poll_interval positive") - await self._claim_window(prev) + await self._claim_window(claim[0]) loop = asyncio.get_running_loop() deadline = loop.time() + window if window else None @@ -475,8 +513,14 @@ async def arm() -> None: kind = SignalKind.RF_315 if frequency < 400 else SignalKind.RF_433 inner = self._capture_loop( - arm, window, stop_after_first, poll_interval, rearm_interval, kind, frequency, - claim=False, + arm, + window, + stop_after_first, + poll_interval, + rearm_interval, + kind, + frequency, + claim=None, ) try: async for signal in inner: @@ -484,9 +528,7 @@ async def arm() -> None: finally: await inner.aclose() - async def _sweep( - self, deadline: Optional[float], poll_interval: float - ) -> Optional[float]: + async def _sweep(self, deadline: float | None, poll_interval: float) -> float | None: """Sweep for the remote's carrier; return it in MHz, or None if the window ran out first.""" loop = asyncio.get_running_loop() @@ -532,7 +574,7 @@ async def _send(self, command: int, data: bytes = b"") -> bytes: e.check_error(resp[0x22:0x24]) payload = self.decrypt(resp[0x38:]) p_len = struct.unpack(" None: async def set_state( self, - pwr: Optional[bool] = None, - ntlight: Optional[bool] = None, - indicator: Optional[bool] = None, - ntlbrightness: Optional[int] = None, - maxworktime: Optional[int] = None, - childlock: Optional[bool] = None, + pwr: bool | None = None, + ntlight: bool | None = None, + indicator: bool | None = None, + ntlbrightness: int | None = None, + maxworktime: int | None = None, + childlock: bool | None = None, ) -> dict: """Set state of device.""" state = {} @@ -187,7 +187,7 @@ def _decode(self, response: bytes) -> dict: e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" dict: e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" dict: async def set_state( self, - pwr: Optional[bool] = None, - pwr1: Optional[bool] = None, - pwr2: Optional[bool] = None, - maxworktime: Optional[int] = None, - maxworktime1: Optional[int] = None, - maxworktime2: Optional[int] = None, - idcbrightness: Optional[int] = None, + pwr: bool | None = None, + pwr1: bool | None = None, + pwr2: bool | None = None, + maxworktime: int | None = None, + maxworktime1: int | None = None, + maxworktime2: int | None = None, + idcbrightness: int | None = None, ) -> dict: """Set the power state of the device.""" state = {} @@ -312,7 +312,7 @@ def _decode(self, response: bytes) -> dict: """Decode a message.""" payload = self.decrypt(response[0x38:]) js_len = struct.unpack_from(" dict: """Set the power state of the device.""" state = {} @@ -459,8 +459,8 @@ async def get_state(self) -> dict: def get_value(start, end, factors): value = sum( - int(payload_str[i-2:i]) * factor - for i, factor in zip(range(start, end, -2), factors) + int(payload_str[i - 2 : i]) * factor + for i, factor in zip(range(start, end, -2), factors, strict=False) ) return value diff --git a/pyproject.toml b/pyproject.toml index c5224d2e..cde5452c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.1" +version = "1.0.2" description = "Python API for controlling Broadlink devices" readme = "README.md" license = "MIT" @@ -57,10 +57,23 @@ line-length = 90 target-version = "py313" [tool.ruff.lint] -# Start from the upstream flake8 gate (syntax errors and undefined names) -# plus pyflakes and import hygiene. Style rules widen once the async port lands. -select = ["E9", "F", "I"] +select = ["E", "W", "F", "I", "UP", "B", "ASYNC", "RUF"] +ignore = [ + # The device timeout parameters are the protocol timeout, not a + # cancellation scope. + "ASYNC109", + # Inherited from upstream and left alone: mutable class-level tables on + # device classes, and the type comparison in BroadlinkException.__eq__. + "RUF012", + "E721", +] [tool.ruff.lint.per-file-ignores] # The package __init__ re-exports the public API. "broadlink/__init__.py" = ["F401"] +# Tests fire helper tasks without keeping a reference on purpose. +"tests/*" = ["RUF006"] +# Long example payloads and comments inherited from upstream. +"broadlink/climate.py" = ["E501"] +"broadlink/light.py" = ["E501"] +"broadlink/switch.py" = ["E501"] diff --git a/tests/oracle/cases.py b/tests/oracle/cases.py index df782acb..df265748 100644 --- a/tests/oracle/cases.py +++ b/tests/oracle/cases.py @@ -44,7 +44,9 @@ def rmminib_payload(body: bytes) -> str: def hysen_payload(body: bytes) -> str: """hysen.send_request: [len][body][crc16(body)]; returns body.""" p_len = len(body) + 2 - return hexb(struct.pack(" str: @@ -120,7 +122,9 @@ def sensor(status, order, stype, name, serial): def hysen_status_body() -> bytes: body = bytearray(48) body[3] = 0x01 # remote_lock - body[4] = 0b1101_0001 # heating_cooling=1, temp_manual=1, active=1, offset add=0, power=1 + body[4] = ( + 0b1101_0001 # heating_cooling=1, temp_manual=1, active=1, offset add=0, power=1 + ) body[5] = 43 # room temp 21.5 body[6] = 44 # thermostat temp 22.0 body[7] = 0x21 # loop_mode 2, auto_mode 1 @@ -145,13 +149,13 @@ def hysen_status_body() -> bytes: def hvac_state_data() -> bytes: data = bytearray(2 + 13) s = memoryview(data)[2:] - s[0x00] = (int(24) - 8 << 3) | 2 # target 24, swing_v POS2 + s[0x00] = (24 - 8 << 3) | 2 # target 24, swing_v POS2 s[0x01] = (7 << 5) | 0b100 # swing_h OFF s[0x03] = 2 << 5 # speed MID s[0x04] = 1 << 6 # preset TURBO (bits 6-7; bit 7 doubles as the half degree) s[0x05] = (1 << 5) | (1 << 2) # mode COOL, sleep s[0x08] = (1 << 5) | (1 << 2) | 0b11 # power, clean, health - s[0x0A] = (1 << 4) # display + s[0x0A] = 1 << 4 # display return bytes(data) @@ -212,30 +216,63 @@ def all_cases() -> list[dict]: # Device base ------------------------------------------------------- add(case("Device", 0x0000, "get_fwversion", responses=[fw_payload(0x1234)])) - add(case("Device", 0x0000, "set_name", "Living room", responses=[EMPTY], attrs=["name"])) + add( + case( + "Device", 0x0000, "set_name", "Living room", responses=[EMPTY], attrs=["name"] + ) + ) add(case("Device", 0x0000, "set_lock", True, responses=[EMPTY], attrs=["is_locked"])) - add(case("Device", 0x0000, "set_lock", False, responses=[EMPTY], attrs=["is_locked"], - setup={"name": "Kitchen"})) + add( + case( + "Device", + 0x0000, + "set_lock", + False, + responses=[EMPTY], + attrs=["is_locked"], + setup={"name": "Kitchen"}, + ) + ) add(case("Device", 0x0000, "get_type")) # RM family --------------------------------------------------------- - for cls, devtype, payload in (("rmmini", 0x2737, rmmini_payload), - ("rmpro", 0x272A, rmmini_payload), - ("rmminib", 0x5F36, rmminib_payload), - ("rm4mini", 0x51DA, rmminib_payload), - ("rm4pro", 0x6026, rmminib_payload), - ("rm", 0x2712, rmmini_payload), - ("rm4", 0x62BE, rmminib_payload)): + for cls, devtype, payload in ( + ("rmmini", 0x2737, rmmini_payload), + ("rmpro", 0x272A, rmmini_payload), + ("rmminib", 0x5F36, rmminib_payload), + ("rm4mini", 0x51DA, rmminib_payload), + ("rm4pro", 0x6026, rmminib_payload), + ("rm", 0x2712, rmmini_payload), + ("rm4", 0x62BE, rmminib_payload), + ): add(case(cls, devtype, "send_data", b(IR_CODE), responses=[payload(b"")])) add(case(cls, devtype, "enter_learning", responses=[payload(b"")])) add(case(cls, devtype, "check_data", responses=[payload(IR_CODE)])) upd = rmminib_update_payload if payload is rmminib_payload else rm_update_payload - add(case(cls, devtype, "update", responses=[upd("Bedroom RM", True)], - attrs=["name", "is_locked"])) + add( + case( + cls, + devtype, + "update", + responses=[upd("Bedroom RM", True)], + attrs=["name", "is_locked"], + ) + ) for cls, devtype in (("rmpro", 0x272A), ("rm", 0x2712)): - add(case(cls, devtype, "check_sensors", responses=[rmmini_payload(bytes([23, 4]))])) - add(case(cls, devtype, "check_temperature", responses=[rmmini_payload(bytes([23, 4]))])) + add( + case( + cls, devtype, "check_sensors", responses=[rmmini_payload(bytes([23, 4]))] + ) + ) + add( + case( + cls, + devtype, + "check_temperature", + responses=[rmmini_payload(bytes([23, 4]))], + ) + ) for cls, devtype in (("rm4mini", 0x51DA), ("rm4pro", 0x6026), ("rm4", 0x62BE)): body = bytes([24, 35, 51, 20]) @@ -243,15 +280,23 @@ def all_cases() -> list[dict]: add(case(cls, devtype, "check_temperature", responses=[rmminib_payload(body)])) add(case(cls, devtype, "check_humidity", responses=[rmminib_payload(body)])) - for cls, devtype, payload in (("rmpro", 0x272A, rmmini_payload), - ("rm4pro", 0x6026, rmminib_payload), - ("rm", 0x2712, rmmini_payload), - ("rm4", 0x62BE, rmminib_payload)): + for cls, devtype, payload in ( + ("rmpro", 0x272A, rmmini_payload), + ("rm4pro", 0x6026, rmminib_payload), + ("rm", 0x2712, rmmini_payload), + ("rm4", 0x62BE, rmminib_payload), + ): add(case(cls, devtype, "sweep_frequency", responses=[payload(b"")])) found = bytes([1]) + struct.pack(" list[dict]: add(case(cls, devtype, "set_power", True, responses=[EMPTY])) add(case(cls, devtype, "check_power", responses=[on])) add(case(cls, devtype, "check_power", responses=[off])) - add(case("sp2s", 0x2728, "get_energy", - responses=["00000000" + (1234).to_bytes(3, "little").hex() + "00" * 9])) - add(case("sp3s", 0x947A, "get_energy", - responses=["0000000000" + "341200" + "00" * 8])) # bytes 5..7 = 34 12 00 + add( + case( + "sp2s", + 0x2728, + "get_energy", + responses=["00000000" + (1234).to_bytes(3, "little").hex() + "00" * 9], + ) + ) + add( + case("sp3s", 0x947A, "get_energy", responses=["0000000000" + "341200" + "00" * 8]) + ) # bytes 5..7 = 34 12 00 nl_on = "00000000" + "03" + "00" * 11 # power bit0, nightlight bit1 add(case("sp3", 0x753E, "set_power", True, responses=[nl_on, EMPTY])) add(case("sp3", 0x753E, "set_nightlight", True, responses=[on, EMPTY])) add(case("sp3", 0x753E, "check_power", responses=[nl_on])) add(case("sp3", 0x753E, "check_nightlight", responses=[nl_on])) - sp4_state = {"pwr": 1, "ntlight": 0, "indicator": 1, "ntlbrightness": 50, - "maxworktime": 0, "childlock": 0} + sp4_state = { + "pwr": 1, + "ntlight": 0, + "indicator": 1, + "ntlbrightness": 50, + "maxworktime": 0, + "childlock": 0, + } add(case("sp4", 0x7579, "get_state", responses=[json12_payload(sp4_state)])) add(case("sp4", 0x7579, "set_power", True, responses=[json12_payload(sp4_state)])) - add(case("sp4", 0x7579, "set_nightlight", False, responses=[json12_payload(sp4_state)])) - add(case("sp4", 0x7579, "set_state", pwr=True, ntlbrightness=25, childlock=True, - responses=[json12_payload(sp4_state)])) + add( + case( + "sp4", 0x7579, "set_nightlight", False, responses=[json12_payload(sp4_state)] + ) + ) + add( + case( + "sp4", + 0x7579, + "set_state", + pwr=True, + ntlbrightness=25, + childlock=True, + responses=[json12_payload(sp4_state)], + ) + ) add(case("sp4", 0x7579, "check_power", responses=[json12_payload(sp4_state)])) add(case("sp4", 0x7579, "check_nightlight", responses=[json12_payload(sp4_state)])) - sp4b_state = dict(sp4_state, current=120, volt=230500, power=27600, - totalconsum=-1, overload=0) + sp4b_state = dict( + sp4_state, current=120, volt=230500, power=27600, totalconsum=-1, overload=0 + ) add(case("sp4b", 0x5115, "get_state", responses=[json14_payload(sp4b_state)])) - add(case("sp4b", 0x5115, "set_state", pwr=False, responses=[json14_payload(sp4b_state)])) + add( + case( + "sp4b", 0x5115, "set_state", pwr=False, responses=[json14_payload(sp4b_state)] + ) + ) add(case("sp4b", 0x5115, "check_power", responses=[json14_payload(sp4b_state)])) - bg_state = {"pwr": 1, "pwr1": 1, "pwr2": 0, "maxworktime": 60, "maxworktime1": 60, - "maxworktime2": 0, "idcbrightness": 50} + bg_state = { + "pwr": 1, + "pwr1": 1, + "pwr2": 0, + "maxworktime": 60, + "maxworktime1": 60, + "maxworktime2": 0, + "idcbrightness": 50, + } add(case("bg1", 0x51E3, "get_state", responses=[json14_payload(bg_state)])) - add(case("bg1", 0x51E3, "set_state", pwr1=True, maxworktime2=15, - responses=[json14_payload(bg_state)])) + add( + case( + "bg1", + 0x51E3, + "set_state", + pwr1=True, + maxworktime2=15, + responses=[json14_payload(bg_state)], + ) + ) add(case("ehc31", 0x6480, "get_state", responses=[json14_payload(bg_state)])) - add(case("ehc31", 0x6480, "set_state", pwr3=True, childlock=True, childlock4=False, - responses=[json14_payload(bg_state)])) + add( + case( + "ehc31", + 0x6480, + "set_state", + pwr3=True, + childlock=True, + childlock4=False, + responses=[json14_payload(bg_state)], + ) + ) add(case("mp1", 0x4EB5, "set_power_mask", 0b0101, True, responses=[EMPTY])) add(case("mp1", 0x4EB5, "set_power", 1, True, responses=[EMPTY])) @@ -310,42 +410,105 @@ def all_cases() -> list[dict]: # Sensors ----------------------------------------------------------- add(case("a1", 0x2714, "check_sensors", responses=[a1_payload()])) - add(case("a1", 0x2714, "check_sensors", responses=[a1_payload(light=9, air=9, noise=9)])) + add( + case( + "a1", 0x2714, "check_sensors", responses=[a1_payload(light=9, air=9, noise=9)] + ) + ) add(case("a1", 0x2714, "check_sensors_raw", responses=[a1_payload()])) add(case("a2", 0x4F60, "check_sensors_raw", responses=[a2_payload()])) # Lights ------------------------------------------------------------ - lb_state = {"red": 128, "blue": 255, "green": 128, "pwr": 1, "brightness": 75, - "colortemp": 2700, "hue": 240, "saturation": 50, - "transitionduration": 1500, "maxworktime": 0, "bulb_colormode": 1, - "bulb_scenes": "[]", "bulb_scene": "", "bulb_sceneidx": 255} + lb_state = { + "red": 128, + "blue": 255, + "green": 128, + "pwr": 1, + "brightness": 75, + "colortemp": 2700, + "hue": 240, + "saturation": 50, + "transitionduration": 1500, + "maxworktime": 0, + "bulb_colormode": 1, + "bulb_scenes": "[]", + "bulb_scene": "", + "bulb_sceneidx": 255, + } add(case("lb1", 0x60C7, "get_state", responses=[json14_payload(lb_state)])) - add(case("lb1", 0x60C7, "set_state", pwr=True, brightness=50, bulb_colormode=1, - bulb_scene="", responses=[json14_payload(lb_state)])) + add( + case( + "lb1", + 0x60C7, + "set_state", + pwr=True, + brightness=50, + bulb_colormode=1, + bulb_scene="", + responses=[json14_payload(lb_state)], + ) + ) add(case("lb2", 0xA4F4, "get_state", responses=[json12_payload(lb_state)])) - add(case("lb2", 0xA4F4, "set_state", pwr=False, red=1, green=2, blue=3, - transitionduration=200, responses=[json12_payload(lb_state)])) + add( + case( + "lb2", + 0xA4F4, + "set_state", + pwr=False, + red=1, + green=2, + blue=3, + transitionduration=200, + responses=[json12_payload(lb_state)], + ) + ) # Climate ----------------------------------------------------------- status = hysen_payload(hysen_status_body()) ack = hysen_payload(bytes([0x01, 0x06, 0x00, 0x02, 0x21, 0x00])) - add(case("hysen", 0x4EAD, "send_request", [0x01, 0x03, 0x00, 0x00, 0x00, 0x08], - responses=[status])) + add( + case( + "hysen", + 0x4EAD, + "send_request", + [0x01, 0x03, 0x00, 0x00, 0x00, 0x08], + responses=[status], + ) + ) add(case("hysen", 0x4EAD, "get_temp", responses=[status])) add(case("hysen", 0x4EAD, "get_external_temp", responses=[status])) add(case("hysen", 0x4EAD, "get_full_status", responses=[status])) add(case("hysen", 0x4EAD, "set_mode", 1, 2, responses=[ack])) add(case("hysen", 0x4EAD, "set_mode", 0, 0, 1, responses=[ack])) - add(case("hysen", 0x4EAD, "set_advanced", 0, 0, 42, 2, 35, 5, -0.5, 0, 1, - responses=[ack])) + add( + case( + "hysen", + 0x4EAD, + "set_advanced", + 0, + 0, + 42, + 2, + 35, + 5, + -0.5, + 0, + 1, + responses=[ack], + ) + ) add(case("hysen", 0x4EAD, "switch_to_auto", responses=[ack])) add(case("hysen", 0x4EAD, "switch_to_manual", responses=[ack])) add(case("hysen", 0x4EAD, "set_temp", 21.5, responses=[ack])) add(case("hysen", 0x4EAD, "set_power", 1, 0, 1, responses=[ack])) add(case("hysen", 0x4EAD, "set_time", 14, 30, 5, 3, responses=[ack])) - sched_wd = [{"start_hour": 6 + i, "start_minute": 15, "temp": 20 + i} for i in range(6)] - sched_we = [{"start_hour": 8, "start_minute": 0, "temp": 21}, - {"start_hour": 22, "start_minute": 30, "temp": 17.5}] + sched_wd = [ + {"start_hour": 6 + i, "start_minute": 15, "temp": 20 + i} for i in range(6) + ] + sched_we = [ + {"start_hour": 8, "start_minute": 0, "temp": 21}, + {"start_hour": 22, "start_minute": 30, "temp": 17.5}, + ] add(case("hysen", 0x4EAD, "set_schedule", sched_wd, sched_we, responses=[ack])) # A corrupted CRC must be rejected. bad = bytearray.fromhex(status) @@ -355,12 +518,69 @@ def all_cases() -> list[dict]: add(case("hvac", 0x4E2A, "get_state", responses=[hvac_payload(hvac_state_data())])) add(case("hvac", 0x4E2A, "get_ac_info", responses=[hvac_payload(hvac_info_data())])) add(case("hvac", 0x4E2A, "get_state", responses=[hvac_payload(b"\x00\x00\x01")])) - add(case("hvac", 0x4E2A, "set_state", True, 22.5, 1, 2, 0, 7, 0, False, False, True, - False, False, False, responses=[hvac_payload(hvac_state_data())])) - add(case("hvac", 0x4E2A, "set_state", True, 24, 4, 3, 2, 0, 0, False, False, True, - False, False, False, responses=[hvac_payload(hvac_state_data())])) - add(case("hvac", 0x4E2A, "set_state", True, 24, 2, 1, 1, 0, 0, False, False, True, - False, False, False, responses=[hvac_payload(hvac_state_data())])) + add( + case( + "hvac", + 0x4E2A, + "set_state", + True, + 22.5, + 1, + 2, + 0, + 7, + 0, + False, + False, + True, + False, + False, + False, + responses=[hvac_payload(hvac_state_data())], + ) + ) + add( + case( + "hvac", + 0x4E2A, + "set_state", + True, + 24, + 4, + 3, + 2, + 0, + 0, + False, + False, + True, + False, + False, + False, + responses=[hvac_payload(hvac_state_data())], + ) + ) + add( + case( + "hvac", + 0x4E2A, + "set_state", + True, + 24, + 2, + 1, + 1, + 0, + 0, + False, + False, + True, + False, + False, + False, + responses=[hvac_payload(hvac_state_data())], + ) + ) # Covers ------------------------------------------------------------ pos = "00000000" + "32" + "00" * 11 # payload[4] = 50 @@ -379,12 +599,29 @@ def all_cases() -> list[dict]: # Hub and alarm ----------------------------------------------------- subs1 = {"total": 3, "list": [{"did": "a1", "pwr1": 1}, {"did": "a2", "pwr1": 0}]} subs2 = {"total": 3, "list": [{"did": "a2", "pwr1": 0}, {"did": "a3", "pwr1": 1}]} - add(case("s3", 0xA59C, "get_subdevices", 2, - responses=[json12_payload(subs1), json12_payload(subs2)])) + add( + case( + "s3", + 0xA59C, + "get_subdevices", + 2, + responses=[json12_payload(subs1), json12_payload(subs2)], + ) + ) add(case("s3", 0xA59C, "get_state", responses=[json12_payload({"pwr1": 1})])) add(case("s3", 0xA59C, "get_state", "a1", responses=[json12_payload({"pwr1": 1})])) - add(case("s3", 0xA59C, "set_state", "a1", True, None, False, - responses=[json12_payload({"pwr1": 1, "pwr3": 0})])) + add( + case( + "s3", + 0xA59C, + "set_state", + "a1", + True, + None, + False, + responses=[json12_payload({"pwr1": 1, "pwr3": 0})], + ) + ) add(case("S1C", 0x2722, "get_sensors_status", responses=[s1c_payload()])) # Error path shared by every class: a non-zero device error code. @@ -395,8 +632,22 @@ def all_cases() -> list[dict]: def error_cases() -> list[dict]: """Cases whose canned response carries a device error code.""" return [ - {"cls": "rmmini", "devtype": 0x2737, "method": "enter_learning", - "args": [], "kwargs": {}, "responses": [EMPTY], "error_code": 0xFFFB}, - {"cls": "sp2", "devtype": 0x2711, "method": "check_power", - "args": [], "kwargs": {}, "responses": [EMPTY], "error_code": 0xFFF9}, + { + "cls": "rmmini", + "devtype": 0x2737, + "method": "enter_learning", + "args": [], + "kwargs": {}, + "responses": [EMPTY], + "error_code": 0xFFFB, + }, + { + "cls": "sp2", + "devtype": 0x2711, + "method": "check_power", + "args": [], + "kwargs": {}, + "responses": [EMPTY], + "error_code": 0xFFF9, + }, ] diff --git a/tests/oracle/harness.py b/tests/oracle/harness.py index 9fb8d2d0..8c5c5254 100644 --- a/tests/oracle/harness.py +++ b/tests/oracle/harness.py @@ -72,8 +72,7 @@ def __call__(self, packet_type: int, payload: bytes) -> bytes: self.sent.append((packet_type, bytes(payload))) if not self.responses: raise AssertionError( - f"method sent more packets than canned responses " - f"({len(self.sent)} sent)" + f"method sent more packets than canned responses ({len(self.sent)} sent)" ) return make_response(self.device, self.responses.pop(0), self.error) @@ -116,7 +115,7 @@ def run_case(case: dict) -> dict: responses = [bytes.fromhex(r) for r in case.get("responses", [])] recorder = Recorder(device, responses, case.get("error_code", 0)) - target = getattr(device, "send_packet") + target = device.send_packet if inspect.iscoroutinefunction(target): device.send_packet = recorder.async_call # type: ignore[method-assign] else: @@ -132,7 +131,7 @@ def run_case(case: dict) -> dict: if inspect.isawaitable(result): result = asyncio.run(_await(result)) outcome["result"] = normalize(result) - except Exception as err: # noqa: BLE001 - the error type IS the oracle + except Exception as err: outcome["error"] = f"{type(err).__name__}: {err}" outcome["sent"] = [[ptype, payload.hex()] for ptype, payload in recorder.sent] diff --git a/tests/test_capture.py b/tests/test_capture.py index a6381fbf..82389f69 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -111,7 +111,9 @@ def handle(self, command: int, data: bytes) -> tuple[bytes, int | str]: self.sweeping = False return b"", 0 if command == CMD_CHECK_FREQ: - found, freq = self.sweep_answers.pop(0) if self.sweep_answers else (False, 0.0) + found, freq = ( + self.sweep_answers.pop(0) if self.sweep_answers else (False, 0.0) + ) return bytes([found]) + struct.pack(" int: return sum(1 for c, _ in self.commands if c == command) -def make(cls_name: str = "rm4pro", devtype: int = 0x649B) -> tuple[broadlink.Device, FakeRM]: +def make( + cls_name: str = "rm4pro", devtype: int = 0x649B +) -> tuple[broadlink.Device, FakeRM]: cls = getattr(broadlink, cls_name) device = cls(HOST, MAC, devtype, name="Bench", model="Test", manufacturer="Test") framing = "rmmini" if cls_name in {"rmmini", "rmpro", "rm"} else "rmminib" @@ -146,8 +150,10 @@ async def press_later(fake: FakeRM, packet: bytes, delay: float) -> bool: # ------------------------------------------------------------- IR windows -@pytest.mark.parametrize("cls_name,devtype", [("rm4pro", 0x649B), ("rmpro", 0x272A), - ("rm4mini", 0x51DA), ("rm5plus", 0x5224)]) +@pytest.mark.parametrize( + "cls_name,devtype", + [("rm4pro", 0x649B), ("rmpro", 0x272A), ("rm4mini", 0x51DA), ("rm5plus", 0x5224)], +) def test_capture_yields_first_signal_and_closes(cls_name, devtype): device, fake = make(cls_name, devtype) @@ -190,7 +196,12 @@ async def go(): loop = asyncio.get_running_loop() loop.create_task(press_later(fake, IR, 2 * UNIT)) loop.create_task(press_later(fake, RF, 6 * UNIT)) - return [s async for s in device.capture(window=12 * UNIT, stop_after_first=False, **FAST)] + return [ + s + async for s in device.capture( + window=12 * UNIT, stop_after_first=False, **FAST + ) + ] signals = run(go()) assert [s.packet for s in signals] == [IR, RF] @@ -215,7 +226,12 @@ async def presses(): results.append(fake.press(RF)) # Re-armed by then. loop.create_task(presses()) - signals = [s async for s in device.capture(window=10 * UNIT, stop_after_first=False, **FAST)] + signals = [ + s + async for s in device.capture( + window=10 * UNIT, stop_after_first=False, **FAST + ) + ] return results, signals results, signals = run(go()) @@ -263,7 +279,10 @@ async def expire_then_press(): loop = asyncio.get_running_loop() task = loop.create_task(expire_then_press()) signals = [ - s async for s in device.capture(window=30 * UNIT, poll_interval=1 * UNIT, rearm_interval=4 * UNIT) + s + async for s in device.capture( + window=30 * UNIT, poll_interval=1 * UNIT, rearm_interval=4 * UNIT + ) ] return await task, signals @@ -278,7 +297,9 @@ def test_open_ended_window_runs_until_closed(): async def go(): got = [] - async with aclosing(device.capture(window=0, stop_after_first=False, **FAST)) as gen: + async with aclosing( + device.capture(window=0, stop_after_first=False, **FAST) + ) as gen: asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) async for s in gen: got.append(s) @@ -294,7 +315,7 @@ async def go(): def test_second_window_is_refused(): - device, fake = make() + device, _fake = make() async def go(): task = asyncio.get_running_loop().create_task( @@ -312,22 +333,20 @@ async def go(): assert device.capture_active is False -def test_abandoned_window_is_closed_by_the_next_one(): - """A consumer that breaks out of the loop without closing the generator - must not block the device; the next window closes the old one.""" +def test_dropped_window_does_not_block_the_next_one(): + """Breaking out of ``async for`` without closing the generator leaves + it to asyncio's finalizer; the next capture() gives that a turn and + proceeds.""" device, fake = make() async def go(): asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) - first = device.capture(window=1, stop_after_first=False, **FAST) - async for s in first: + got = None + async for s in device.capture(window=1, stop_after_first=False, **FAST): got = s - break # Walk away without aclose(). - assert device.capture_active is True # The old generator is suspended. - assert not first.ag_running + break # No reference kept; the generator is collectable. asyncio.get_running_loop().create_task(press_later(fake, RF, 2 * UNIT)) second = [s async for s in device.capture(window=1, **FAST)] - assert first.ag_frame is None # Closed by the second window. return got, second got, second = run(go()) @@ -336,17 +355,100 @@ async def go(): assert device.capture_active is False -def test_unstarted_window_does_not_block(): +def test_held_window_is_not_taken_by_the_next_one(): + """A window the consumer still holds is refused to a newcomer, even if + the consumer is not inside the generator at that instant, until the + consumer closes it.""" device, fake = make() async def go(): - _unused = device.capture(window=1, **FAST) # never iterated + asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) + first = device.capture(window=1, stop_after_first=False, **FAST) + async for _ in first: + break # Still referenced by ``first``. + with pytest.raises(e.CaptureInProgressError): + await _collect(device.capture(window=1, **FAST)) + assert device.capture_active is True + # A refused attempt must not have displaced the live window. + with pytest.raises(e.CaptureInProgressError): + await _collect(device.capture_rf(window=1, frequency=433.92, **FAST)) + assert device.capture_active is True + await first.aclose() + assert device.capture_active is False + asyncio.get_running_loop().create_task(press_later(fake, RF, 2 * UNIT)) + return [s async for s in device.capture(window=1, **FAST)] + + second = run(go()) + assert [s.packet for s in second] == [RF] + + +def test_paused_consumer_keeps_its_window(): + """The README's own loop awaits between signals. An intruder calling + capture() during that pause must be refused, and the consumer must + keep receiving.""" + device, fake = make() + + async def go(): + loop = asyncio.get_running_loop() + got = [] + intruder = {"error": None, "signals": None} + + async def consumer(): + async with aclosing( + device.capture(window=30 * UNIT, stop_after_first=False, **FAST) + ) as window: + async for s in window: + got.append(s) + await asyncio.sleep(4 * UNIT) # Paused, not running. + + async def intrude(): + await asyncio.sleep(3 * UNIT) # During the consumer's pause. + try: + intruder["signals"] = await _collect(device.capture(window=1, **FAST)) + except e.CaptureInProgressError as err: + intruder["error"] = err + + loop.create_task(press_later(fake, IR, 2 * UNIT)) + loop.create_task(press_later(fake, RF, 12 * UNIT)) + task = loop.create_task(consumer()) + await intrude() + await task + return got, intruder + + got, intruder = run(go()) + assert isinstance(intruder["error"], e.CaptureInProgressError) + assert intruder["signals"] is None + assert [s.packet for s in got] == [IR, RF] + + +def test_unreferenced_unstarted_window_does_not_block(): + device, fake = make() + + async def go(): + device.capture(window=1, **FAST) # created and dropped, never iterated asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) return [s async for s in device.capture(window=1, **FAST)] assert len(run(go())) == 1 +def test_undecodable_packet_does_not_end_the_window(): + """A returned packet whose declared length runs into a truncated escape + is logged and skipped; the window re-arms and the next signal lands.""" + device, fake = make() + bad = bytes([0x26, 0x00, 0x03, 0x00, 0x10, 0x00]) # escape with no bytes after + + async def go(): + loop = asyncio.get_running_loop() + loop.create_task(press_later(fake, bad, 2 * UNIT)) + loop.create_task(press_later(fake, IR, 6 * UNIT)) + return [s async for s in device.capture(window=1, **FAST)] + + signals = run(go()) + assert [s.packet for s in signals] == [IR] + assert fake.count(CMD_LEARN) >= 2 # Re-armed after the bad one. + + def test_older_firmware_read_error_means_nothing_yet(): device, fake = make() fake.nothing_yet_code = -10 # ReadError @@ -365,9 +467,10 @@ def test_capture_rf_abandoned_then_ir_window(): async def go(): asyncio.get_running_loop().create_task(press_later(fake, RF, 2 * UNIT)) - rf = device.capture_rf(window=1, frequency=433.92, stop_after_first=False, **FAST) - async for _ in rf: - break + async for _ in device.capture_rf( + window=1, frequency=433.92, stop_after_first=False, **FAST + ): + break # Dropped, not held. asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) return [s async for s in device.capture(window=1, **FAST)] @@ -506,7 +609,7 @@ async def send(): def test_capture_rf_refused_while_ir_window_open(): - device, fake = make() + device, _fake = make() async def go(): task = asyncio.get_running_loop().create_task( diff --git a/tests/test_oracle.py b/tests/test_oracle.py index 6cdd4d2b..c0a1accb 100644 --- a/tests/test_oracle.py +++ b/tests/test_oracle.py @@ -35,8 +35,16 @@ def test_every_public_method_is_covered() -> None: covered = {(e["case"]["cls"], e["case"]["method"]) for e in ENTRIES} # Methods on Device itself that need a live socket are covered in # test_transport.py, not here. - transport_level = {"auth", "hello", "ping", "send_packet", "encrypt", "decrypt", - "update_aes", "aclose"} + transport_level = { + "auth", + "hello", + "ping", + "send_packet", + "encrypt", + "decrypt", + "update_aes", + "aclose", + } # Capture windows drive several requests over time; they are covered # with a scripted device in test_capture.py. transport_level |= {"capture", "capture_rf"} diff --git a/tests/test_remote.py b/tests/test_remote.py index c0f7f06a..72cd02ef 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -1,4 +1,5 @@ """Tests for the tick constant used by pulses_to_data / data_to_pulses (GH #839).""" + import unittest from broadlink.remote import TICK, data_to_pulses, pulses_to_data @@ -34,7 +35,7 @@ def test_round_trip(self): pulses = [9000, 4500, 560, 1690, 560, 560] packet = pulses_to_data(pulses) decoded = data_to_pulses(packet) - for original, result in zip(pulses, decoded): + for original, result in zip(pulses, decoded, strict=True): self.assertAlmostEqual(result, original, delta=TICK) def test_true_microsecond_nec_leader_is_now_correct(self): diff --git a/tests/test_transport.py b/tests/test_transport.py index f1b59098..414bb044 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -62,7 +62,9 @@ def __init__(self): async def __call__(self, local_addr=None, remote_addr=None, broadcast=False): protocol = device_module._Protocol() - transport = FakeTransport(protocol, local_addr, remote_addr, broadcast, self.replies) + transport = FakeTransport( + protocol, local_addr, remote_addr, broadcast, self.replies + ) self.replies = [] protocol.connection_made(transport) self.endpoints.append(transport) @@ -262,6 +264,97 @@ async def answer_later(): assert run(go()) == 2 +def test_second_answer_to_a_resent_request_is_not_taken_as_next_reply(net): + """The retry path: a request goes unanswered for a resend interval, is + resent with the same counter, and the device answers both copies. The + second answer must not be taken as the reply to the next request.""" + dev = fixed_device() + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + first = stamped(dev, bytes([1]) + bytes(15), 0x8001) + second = stamped(dev, bytes([2]) + bytes(15), 0x8002) + sends = {"n": 0} + + def sendto(data, addr=None): + ep.sent.append((bytes(data), addr or ep.remote_addr)) + sends["n"] += 1 + if sends["n"] == 2: # The resend of request 1 gets answered... + ep.protocol.queue.put_nowait((first, HOST)) + + ep.sendto = sendto + resp1 = await dev.send_packet(0x6A, b"") # count 0x8001, answered on resend + assert dev.decrypt(resp1[0x38:])[0] == 1 + assert sends["n"] == 2 + + # ...and the original copy's answer shows up while request 2 waits, + # followed by request 2's own answer. + async def late_then_real(): + await asyncio.sleep(0.003) + ep.protocol.queue.put_nowait((first, HOST)) # duplicate, counter 0x8001 + await asyncio.sleep(0.003) + ep.protocol.queue.put_nowait((second, HOST)) + + asyncio.get_running_loop().create_task(late_then_real()) + resp2 = await dev.send_packet(0x6A, b"") # count 0x8002 + return dev.decrypt(resp2[0x38:])[0] + + assert run(go()) == 2 + + +def test_auth_never_lets_a_queued_request_out_with_id_zero(net): + """A request queued behind the lock while auth() runs must be framed + after the new session is installed, never with the initial key and + device id 0.""" + dev = fixed_device() + dev.id = 5 + session_key = bytes.fromhex("00112233445566778899aabbccddeeff") + fresh = fixed_device() + auth_reply = make_response(fresh, (0x42).to_bytes(4, "little") + session_key) + renewed = fixed_device() + renewed.update_aes(session_key) + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + authed = {"done": False} + + loop = asyncio.get_running_loop() + + def sendto(data, addr=None): + ep.sent.append((bytes(data), addr or ep.remote_addr)) + ptype = int.from_bytes(data[0x26:0x28], "little") + if ptype == 0x65: + authed["done"] = True + reply = auth_reply + elif not authed["done"]: + reply = make_response(dev, b"", error=0xFFF9) # -7 expired + else: + reply = make_response(renewed, bytes([7]) + bytes(15)) + # Answer a little later, like a real device, so the caller + # suspends and the second caller queues behind the lock. + loop.call_later(0.002, ep.protocol.queue.put_nowait, (reply, ep.remote_addr)) + + ep.sendto = sendto + a, b = await asyncio.gather( + dev.send_packet(0x6A, b"a"), dev.send_packet(0x6A, b"b") + ) + return ep, a, b + + ep, a, b = run(go()) + ids = [ + (int.from_bytes(f[0x26:0x28], "little"), int.from_bytes(f[0x30:0x34], "little")) + for f, _ in ep.sent + ] + for ptype, dev_id in ids: + if ptype != 0x65: + assert dev_id in (5, 0x42), ids + assert [p for p, _ in ids].count(0x65) == 1 + assert dev.decrypt(a[0x38:])[0] == 7 + assert dev.decrypt(b[0x38:])[0] == 7 + + def test_reply_with_unknown_counter_is_accepted(net): """Firmware that does not echo the counter must keep working.""" dev = fixed_device() @@ -284,8 +377,10 @@ async def go(): await asyncio.sleep(0.01) t0 = asyncio.get_running_loop().time() await dev.aclose() - with pytest.raises(e.ConnectionClosedError): + with pytest.raises(e.EndpointClosedError) as err: await task + assert isinstance(err.value, e.ConnectionClosedError) + assert err.value.errno == -4013 return asyncio.get_running_loop().time() - t0 assert run(go()) < 1.0 @@ -424,7 +519,9 @@ def sendto(data, addr=None): ep.protocol.queue.put_nowait((reply, ep.remote_addr)) ep.sendto = sendto - a, b = await asyncio.gather(dev.send_packet(0x6A, b"a"), dev.send_packet(0x6A, b"b")) + a, b = await asyncio.gather( + dev.send_packet(0x6A, b"a"), dev.send_packet(0x6A, b"b") + ) return ep, {dev.decrypt(a[0x38:])[0], dev.decrypt(b[0x38:])[0]} ep, values = run(go())