From 0bc04d2b40b101f44b5b7d4f8e2c0d0cb40d8827 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sat, 5 Sep 2026 21:52:32 +0000 Subject: [PATCH 1/2] Fix the issues found in the second review of 1.0.1 A second, adversarial review of 1.0.1 and a re-test of the first review's findings turned up three real defects in the transport and capture code, none in the wire format. This fixes them and takes the smaller items along. Tested on 3.13 and 3.14, 257 tests, oracle fixtures unchanged. Technical details: - Reply matching remembers every recently used counter, not only timed-out ones, so the second answer to a resent request is dropped instead of being taken as the next request's reply. - auth() resets the session and installs the new key under the request lock, so a queued request is never framed with id 0. - A new capture window gives asyncio's finalizer a turn to close a dropped generator, then refuses if the old window is still alive, instead of taking it from a paused consumer. A refused attempt no longer displaces the live window. - An undecodable returned packet is logged and skipped; the window re-arms. - EndpointClosedError (-4013), a subclass of ConnectionClosedError, for aclose() during a request. - hello() closes the scan generator; TimeoutError spelling; unused protocol future removed; debug logging on device and remote. - README: Closing section, Timing section with the bench numbers, Python support note, return-value differences from 0.19.0. - Version 1.0.2. --- CHANGELOG.md | 52 +++++++++++++++++ README.md | 61 ++++++++++++++++++-- broadlink/__init__.py | 16 ++++-- broadlink/device.py | 120 ++++++++++++++++++++++++---------------- broadlink/exceptions.py | 11 ++++ broadlink/remote.py | 110 +++++++++++++++++++++++------------- pyproject.toml | 2 +- tests/test_capture.py | 110 +++++++++++++++++++++++++++++++----- tests/test_transport.py | 93 ++++++++++++++++++++++++++++++- 9 files changed, 462 insertions(+), 113 deletions(-) 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..d7f24064 100644 --- a/README.md +++ b/README.md @@ -16,9 +16,12 @@ 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 @@ -52,8 +55,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. ``` @@ -144,6 +171,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 @@ -217,10 +265,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..07d293b3 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 """The python-broadlink library.""" +import contextlib from collections.abc import AsyncIterator from typing import List, Optional, Tuple, Union @@ -258,12 +259,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", diff --git a/broadlink/device.py b/broadlink/device.py index 4ac3988c..fb1ed473 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -11,6 +11,8 @@ import asyncio import collections +import contextlib +import logging import random import socket from collections.abc import AsyncIterator @@ -28,6 +30,8 @@ ) from .protocol import Datetime +_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 @@ -35,10 +39,12 @@ # -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.""" @@ -50,7 +56,6 @@ 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() def connection_made(self, transport) -> None: # type: ignore[override] self.transport = transport @@ -64,8 +69,7 @@ def error_received(self, exc: Exception) -> None: pass def connection_lost(self, exc: Optional[Exception]) -> None: - if not self.closed.done(): - self.closed.set_result(None) + pass def drain(self) -> None: """Drop anything that arrived before the current request.""" @@ -144,7 +148,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 @@ -214,7 +218,7 @@ def __init__( 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._recent: collections.deque[int] = collections.deque(maxlen=_RECENT_MAX) self._reauth_lock: Optional[asyncio.Lock] = None self._auth_generation = 0 @@ -272,23 +276,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:]) - - 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 +308,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: + break if entry is None: raise e.NetworkTimeoutError( -4000, @@ -388,6 +402,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] @@ -401,6 +416,7 @@ async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: 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 +466,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 +477,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 +493,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 +535,14 @@ 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..82c91513 100644 --- a/broadlink/exceptions.py +++ b/broadlink/exceptions.py @@ -71,6 +71,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 +152,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/remote.py b/broadlink/remote.py index 4527ccc6..18caf9ec 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -2,6 +2,7 @@ import asyncio import enum +import logging import struct import time import weakref @@ -11,6 +12,8 @@ from . import exceptions as e from .device import Device +_LOGGER = logging.getLogger(__name__) + TICK = 8192 / 269 """Duration of one Broadlink timing unit in microseconds (about 30.45 us). @@ -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( @@ -329,15 +342,17 @@ async def _capture_loop( kind: SignalKind, frequency_mhz: Optional[float], *, - prev: Optional[weakref.ReferenceType] = None, - claim: bool = True, + claim: Optional[list] = 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): @@ -436,12 +469,13 @@ 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, + claim=holder, ) - self._window = weakref.ref(gen) + holder.append(weakref.ref(gen)) return gen async def _capture_rf_loop( @@ -452,11 +486,11 @@ async def _capture_rf_loop( 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 @@ -476,7 +510,7 @@ 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, + claim=None, ) try: async for signal in inner: diff --git a/pyproject.toml b/pyproject.toml index c5224d2e..c4b644e4 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" diff --git a/tests/test_capture.py b/tests/test_capture.py index a6381fbf..b73d0161 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -312,22 +312,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 +334,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 +446,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)] diff --git a/tests/test_transport.py b/tests/test_transport.py index f1b59098..f1b1122d 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -262,6 +262,95 @@ 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 +373,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 From 679c3866e9ab7c9dd48ccfe797ce177a51c627d7 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sat, 5 Sep 2026 21:54:14 +0000 Subject: [PATCH 2/2] Format with ruff and widen the lint rules No behaviour change: 257 tests pass before and after, the oracle fixtures are byte-identical, and repr/str output is unchanged. This is the one-time formatting pass the pyproject comment promised once the async port landed. Technical details: - ruff format over the tree; CI now runs ruff format --check. - Lint set widened from E9/F/I to E, W, F, I, UP, B, ASYNC, RUF. Safe autofixes applied (typing modernised to the 3.13 spellings, f-strings for the percent formatting in repr/str and the exceptions). - Ignored with a reason in pyproject: ASYNC109 (protocol timeout, not a cancellation scope), RUF012 and E721 (inherited class tables and the exception __eq__), RUF006 in tests (helper tasks fired on purpose), and E501 in the three upstream modules with long example payloads. - zip() calls carry an explicit strict=; hello()'s first-reply loop carries a noqa with its reason. --- .github/workflows/ci.yml | 2 + README.md | 14 +- broadlink/__init__.py | 13 +- broadlink/alarm.py | 1 + broadlink/climate.py | 37 +--- broadlink/const.py | 1 + broadlink/cover.py | 3 +- broadlink/device.py | 69 +++---- broadlink/exceptions.py | 7 +- broadlink/helpers.py | 7 +- broadlink/hub.py | 18 +- broadlink/light.py | 64 +++---- broadlink/protocol.py | 1 + broadlink/remote.py | 52 +++--- broadlink/sensor.py | 3 +- broadlink/switch.py | 64 +++---- pyproject.toml | 19 +- tests/oracle/cases.py | 395 ++++++++++++++++++++++++++++++++------- tests/oracle/harness.py | 7 +- tests/test_capture.py | 41 +++- tests/test_oracle.py | 12 +- tests/test_remote.py | 3 +- tests/test_transport.py | 12 +- 23 files changed, 564 insertions(+), 281 deletions(-) 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/README.md b/README.md index d7f24064..e6231ff1 100644 --- a/README.md +++ b/README.md @@ -27,12 +27,14 @@ returning `bytes`, the unused `Device.lock` attribute removed, and 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()) ``` @@ -121,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) @@ -130,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 @@ -146,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: @@ -223,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 diff --git a/broadlink/__init__.py b/broadlink/__init__.py index 07d293b3..632afdf3 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -1,8 +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 @@ -224,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: @@ -277,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 @@ -292,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 fb1ed473..2315f307 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -16,7 +16,6 @@ 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 @@ -32,7 +31,7 @@ _LOGGER = logging.getLogger(__name__) -HelloResponse = Tuple[int, Tuple[str, int], bytes, str, bool] +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; @@ -55,7 +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.transport: asyncio.DatagramTransport | None = None def connection_made(self, transport) -> None: # type: ignore[override] self.transport = transport @@ -68,7 +67,7 @@ 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: + def connection_lost(self, exc: Exception | None) -> None: pass def drain(self) -> None: @@ -78,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.""" @@ -115,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]: @@ -188,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 = "", @@ -214,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._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: Optional[asyncio.Lock] = None + 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: @@ -286,7 +275,7 @@ async def auth(self) -> bool: packet[0x04:0x14] = [0x31] * 16 packet[0x1E] = 0x01 packet[0x2D] = 0x01 - packet[0x30:0x36] = "Test 1".encode() + packet[0x30:0x36] = b"Test 1" if self._lock is None: self._lock = asyncio.Lock() @@ -317,7 +306,7 @@ async def hello(self, local_ip_address=None) -> bool: discover_ip_port=self.host[1], ) ) as responses: - async for entry in responses: + async for entry in responses: # noqa: B007 - first reply only break if entry is None: raise e.NetworkTimeoutError( @@ -412,9 +401,7 @@ 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] @@ -537,9 +524,7 @@ async def send_packet(self, packet_type: int, payload: bytes) -> bytes: 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 - ) + _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() diff --git a/broadlink/exceptions.py b/broadlink/exceptions.py index 82c91513..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).""" 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, @@ -106,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 @@ -139,7 +139,7 @@ class ParsedPacket: kind: SignalKind repeat: int - pulses: List[int] + pulses: list[int] type_byte: int @@ -170,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. @@ -220,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: @@ -340,9 +340,9 @@ async def _capture_loop( poll_interval: float, rearm_interval: float, kind: SignalKind, - frequency_mhz: Optional[float], + frequency_mhz: float | None, *, - claim: Optional[list] = None, + 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 @@ -426,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: @@ -448,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, @@ -472,7 +472,11 @@ def capture_rf( self._check_window() holder: list = [] gen = self._capture_rf_loop( - window, frequency, stop_after_first, poll_interval, rearm_interval, + window, + frequency, + stop_after_first, + poll_interval, + rearm_interval, claim=holder, ) holder.append(weakref.ref(gen)) @@ -481,7 +485,7 @@ def capture_rf( async def _capture_rf_loop( self, window: float, - frequency: Optional[float], + frequency: float | None, stop_after_first: bool, poll_interval: float, rearm_interval: float, @@ -509,7 +513,13 @@ 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, + arm, + window, + stop_after_first, + poll_interval, + rearm_interval, + kind, + frequency, claim=None, ) try: @@ -518,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() @@ -566,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 c4b644e4..cde5452c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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 b73d0161..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( @@ -588,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 f1b1122d..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) @@ -335,7 +337,9 @@ def sendto(data, addr=None): 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")) + 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()) @@ -515,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())