From d72bdf58e026a4a30ddd23b280b3d53ab93fc730 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sun, 6 Sep 2026 02:31:41 +0000 Subject: [PATCH] Fix the issues found in the review of 1.0.4 A fifth review, of 1.0.4, found one more place where the connected socket could surface an error the original library never raised, and made a fair case that the capture window claim was more machinery than the job needed. This fixes the first, simplifies the second, and takes two small items along. No change to the wire format or the public API. Tested on 3.13 and 3.14, 272 tests under -X dev with ResourceWarning as an error, oracle fixtures unchanged, and live against an RM4 Pro. Technical details: - EHOSTUNREACH and Windows's ConnectionResetError are treated as silence in _exchange, as ECONNREFUSED already was: logged, the timeout decides, the socket is dropped afterwards. Home Assistant tolerates a timeout for a few polls but marks a device unavailable on the first OSError, and the original's unconnected socket never saw any of these. Measured on the bench before changing anything: neither an on-link address with no host nor an off-subnet one produced the ICMP here, so this is insurance for networks that do. - The window claim is a flag set on first iteration and cleared in the generator's finally (so asyncio's finalizer releases a dropped window), replacing the weakref and ag_frame inspection. Same behaviour, same tests; CaptureInProgressError now always comes from the first iteration. - discover() closes the xdiscover() generator it drains. - A comment explains why the RM Max sits in rmpro (upstream #838). - README: a device belongs to the event loop it first talks on. - Version 1.0.5. --- CHANGELOG.md | 36 ++++++++++ README.md | 10 ++- broadlink/__init__.py | 12 ++-- broadlink/device.py | 32 +++++++-- broadlink/remote.py | 147 +++++++++++++++++++++------------------- pyproject.toml | 2 +- tests/test_loopback.py | 2 +- tests/test_transport.py | 34 ++++++++++ 8 files changed, 190 insertions(+), 85 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bbc9c4cb..18f56eab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,42 @@ All notable changes to this project are recorded here. The format follows Keep a Changelog; versions follow Semantic Versioning. +## 1.0.5 - 2026-09-06 + +Fixes from a fifth review, of 1.0.4. No change to the wire format or the +public API. + +### Fixed + +- A connected socket can learn from ICMP that a host cannot be reached + (`EHOSTUNREACH`, typically a router answering for a device that is off), + and 1.0.4 raised that as an `OSError` at once. The original library's + unconnected socket never saw it and simply timed out, and Home + Assistant tolerates a timeout for a few polls where it marks a device + unavailable on the first `OSError`. Host unreachable, and Windows's + `ConnectionResetError` for port unreachable, are now treated as silence + like port unreachable already was: logged, the timeout decides, and the + socket is still dropped afterwards. Measured on the bench first: on the + test network neither an on-link address with no host behind it nor an + off-subnet one produced the ICMP, so this is insurance for networks + that do, not a fix for one that reproduced. +- `discover()` closes the `xdiscover()` generator it drains, like + `hello()` and `xdiscover()` itself. + +### Changed + +- The capture window claim is a flag set on the window's first iteration + and cleared when its generator finishes or is closed, including by + asyncio's finalizer, in place of the weak reference and frame + inspection used since 1.0.0. Same behaviour, pinned by the same tests: + a dropped window gets one turn to be finalized and then blocks nobody, + a held or paused window is refused to a newcomer. One visible + difference: `CaptureInProgressError` now always comes from the new + window's first iteration, never from the `capture()` call itself. +- A comment next to the RM Max entry says why it sits in `rmpro` + (upstream #838's text says rm4pro, its tested diff says rmpro). +- README: a device belongs to the event loop it first talks on. + ## 1.0.4 - 2026-09-06 Fixes from a fourth review, of 1.0.3, which drove the real socket path the diff --git a/README.md b/README.md index 329348f2..2a3884dd 100644 --- a/README.md +++ b/README.md @@ -205,6 +205,11 @@ async with device: await device.aclose() ``` +A device belongs to the event loop it first talks on: its socket and its +locks are bound to that loop, so a script that runs several +`asyncio.run(...)` calls should create the device inside each one rather +than reuse it across them. + 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`. A request that fails for a network @@ -290,9 +295,8 @@ By default the window closes after the first signal. Pass 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: opening a second one raises `CaptureInProgressError` -while the first is still held, either from the `capture()` call itself or -from the new window's first iteration, depending on what the first window -was doing at that moment. Always close a window you leave early +from the new window's first iteration 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. diff --git a/broadlink/__init__.py b/broadlink/__init__.py index 0dd588d3..59c1012e 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -116,6 +116,9 @@ 0x27A6: ("RM plus", "Broadlink"), 0x27A9: ("RM pro+", "Broadlink"), 0x27C3: ("RM pro+", "Broadlink"), + # The RM Max answers the RM pro framing; the RM4 framing (length + # prefix) gets "device is locked" from it. Tested on hardware in + # upstream #838, whose text says rm4pro but whose diff says rmpro. 0xAF8B: ("RM Max", "Broadlink"), }, rmminib: { @@ -281,12 +284,9 @@ async def discover( discover_ip_port: int = DEFAULT_PORT, ) -> list[Device]: """Discover devices connected to the local network.""" - return [ - device - async for device in xdiscover( - timeout, local_ip_address, discover_ip_address, discover_ip_port - ) - ] + devices = xdiscover(timeout, local_ip_address, discover_ip_address, discover_ip_port) + async with contextlib.aclosing(devices): + return [device async for device in devices] async def xdiscover( diff --git a/broadlink/device.py b/broadlink/device.py index c28f3006..45665345 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -12,6 +12,7 @@ import asyncio import collections import contextlib +import errno import logging import random import socket @@ -53,6 +54,22 @@ error the socket reported (address ``None``), or ``_CLOSED``.""" +def _is_silence(item: object) -> bool: + """True for the socket errors that mean "no device answered". + + A connected datagram socket learns from ICMP that nobody is listening + (port unreachable, ``ConnectionRefusedError``; ``ConnectionResetError`` + on Windows) or that the host cannot be reached (``EHOSTUNREACH``, from + a router answering for a host that is off). The original library used + an unconnected socket that never received any of these and simply + timed out, and callers such as Home Assistant treat a timeout more + leniently than an ``OSError``, so these are treated as silence. + """ + if isinstance(item, ConnectionRefusedError | ConnectionResetError): + return True + return isinstance(item, OSError) and item.errno == errno.EHOSTUNREACH + + class _Protocol(asyncio.DatagramProtocol): """Datagram protocol that hands every received packet to a queue. @@ -570,12 +587,15 @@ async def _exchange(self, packet: bytes) -> bytes: raise e.EndpointClosedError( -4013, "Endpoint closed", "The device endpoint was closed" ) - if isinstance(resp, ConnectionRefusedError): - # ICMP port unreachable: the host is up and nothing is - # listening, or the device is rebooting. The original - # library's unconnected socket never saw these, so keep - # waiting and let the timeout decide, as it did. - _LOGGER.debug("%s: port unreachable, still waiting", self.host[0]) + if _is_silence(resp): + # ICMP unreachable of one kind or another: the host is up + # with nothing listening, the device is off or rebooting, + # or a router answered for it. The original library's + # unconnected socket never saw these, so keep waiting and + # let the timeout decide, as it did. + _LOGGER.debug( + "%s: unreachable (%s), still waiting", self.host[0], resp + ) continue if isinstance(resp, Exception): # A send failure (no route, address gone) or a fatal diff --git a/broadlink/remote.py b/broadlink/remote.py index 6840833c..e90251a3 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -1,11 +1,11 @@ """Support for universal remotes.""" import asyncio +import contextlib import enum import logging import struct import time -import weakref from collections.abc import AsyncGenerator, Awaitable, Callable from dataclasses import dataclass, field from typing import Self @@ -221,51 +221,39 @@ def __init__(self, *args, **kwargs) -> None: # against the value it saw when it armed the device and re-arms # after any send, since the device has one front end for both. self._tx_generation = 0 - # Weak reference to the async generator of the current capture - # window, if any. See _claim_window. - self._window: weakref.ReferenceType | None = None + # True while a capture window holds the device's receiver. Set by + # the window on its first iteration, cleared when its generator + # finishes or is closed, including by asyncio's finalizer. + self._capturing = False @property def capture_active(self) -> bool: """True while a capture window is open on this device.""" - window = self._window() if self._window is not None else None - return window is not None and window.ag_frame is not None + return self._capturing - 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 not None and old.ag_frame is not None and old.ag_running: - raise e.CaptureInProgressError("A capture window is already open") - - async def _claim_window(self, new: weakref.ReferenceType) -> None: - """Make sure the previous window is really gone, then register ``new``. + async def _claim_window(self) -> None: + """Take the receiver for a new window, or refuse. 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. + loop iteration once nothing references it, and closing it releases + the claim. Give that a turn. If the claim is still held after that, + someone still has the window, whether inside ``__anext__`` or + paused between signals, and the new window is refused rather than + taken from under them. """ - 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" - ) - if self._window is not prev: - # Another claimant got in during the two turns above. - raise e.CaptureInProgressError("A capture window is already open") - self._window = new + if self._capturing: + await asyncio.sleep(0) + await asyncio.sleep(0) + if self._capturing: + raise e.CaptureInProgressError( + "A capture window is already open; close it with aclose() first" + ) + self._capturing = True + + def _release_window(self) -> None: + self._capturing = False async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" @@ -319,13 +307,11 @@ def capture( 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 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. + ``CaptureInProgressError`` on its first iteration. A window whose + generator was dropped without being closed is finalized by asyncio + on the next loop iteration and does not block. """ - self._check_window() - holder: list = [] - gen = self._capture_loop( + return self._capture_loop( self.enter_learning, window, stop_after_first, @@ -333,10 +319,8 @@ def capture( rearm_interval, SignalKind.IR, None, - claim=holder, + claim=True, ) - holder.append(weakref.ref(gen)) - return gen async def _capture_loop( self, @@ -348,18 +332,43 @@ async def _capture_loop( kind: SignalKind, frequency_mhz: float | None, *, - claim: list | None = None, + claim: bool, ) -> AsyncGenerator[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. + # ``claim`` is False when the caller already holds the window, 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(claim[0]) + await self._claim_window() + try: + body = self._capture_body( + arm, + window, + stop_after_first, + poll_interval, + rearm_interval, + kind, + frequency_mhz, + ) + async with contextlib.aclosing(body): + async for signal in body: + yield signal + finally: + if claim: + self._release_window() + async def _capture_body( + self, + arm: Callable[[], Awaitable[None]], + window: float, + stop_after_first: bool, + poll_interval: float, + rearm_interval: float, + kind: SignalKind, + frequency_mhz: float | None, + ) -> AsyncGenerator[CapturedSignal]: loop = asyncio.get_running_loop() deadline = loop.time() + window if window else None timeouts = 0 @@ -475,18 +484,9 @@ def capture_rf( Each ``CapturedSignal`` carries the carrier in ``frequency_mhz``, which the packet itself does not record. """ - self._check_window() - holder: list = [] - gen = self._capture_rf_loop( - window, - frequency, - stop_after_first, - poll_interval, - rearm_interval, - claim=holder, + return self._capture_rf_loop( + window, frequency, stop_after_first, poll_interval, rearm_interval ) - holder.append(weakref.ref(gen)) - return gen async def _capture_rf_loop( self, @@ -495,13 +495,26 @@ async def _capture_rf_loop( stop_after_first: bool, poll_interval: float, rearm_interval: float, - *, - claim: list, ) -> AsyncGenerator[CapturedSignal]: if window < 0 or poll_interval <= 0: raise ValueError("window must be 0 or positive, poll_interval positive") - await self._claim_window(claim[0]) + await self._claim_window() + try: + async for signal in self._capture_rf_body( + window, frequency, stop_after_first, poll_interval, rearm_interval + ): + yield signal + finally: + self._release_window() + async def _capture_rf_body( + self, + window: float, + frequency: float | None, + stop_after_first: bool, + poll_interval: float, + rearm_interval: float, + ) -> AsyncGenerator[CapturedSignal]: loop = asyncio.get_running_loop() deadline = loop.time() + window if window else None @@ -526,13 +539,11 @@ async def arm() -> None: rearm_interval, kind, frequency, - claim=None, + claim=False, ) - try: + async with contextlib.aclosing(inner): async for signal in inner: yield signal - finally: - await inner.aclose() 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 diff --git a/pyproject.toml b/pyproject.toml index ffa751f4..e07e4e22 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.4" +version = "1.0.5" description = "Python API for controlling Broadlink devices" readme = "README.md" license = "MIT" diff --git a/tests/test_loopback.py b/tests/test_loopback.py index 4921fcaf..ff294eee 100644 --- a/tests/test_loopback.py +++ b/tests/test_loopback.py @@ -107,4 +107,4 @@ async def go(): await dev.aclose() assert asyncio.run(go()) == 9 - assert "port unreachable" in caplog.text + assert "unreachable" in caplog.text diff --git a/tests/test_transport.py b/tests/test_transport.py index ff5d6370..a179c3c5 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -483,6 +483,40 @@ def failing_sendto(data, addr=None): assert run(go()) +@pytest.mark.parametrize( + "err", + [ + ConnectionRefusedError(111, "Connection refused"), + ConnectionResetError(10054, "Connection reset"), + OSError(113, "No route to host"), + ], + ids=["port-unreachable", "windows-reset", "host-unreachable"], +) +def test_icmp_unreachable_is_silence_not_an_error(net, err): + """0.19.0's unconnected socket never saw ICMP errors: a device that was + off, rebooting, or behind a router answering for it produced silence + and then NetworkTimeoutError, which Home Assistant tolerates for a few + polls where it would mark the device unavailable on an OSError. The + connected socket must keep that contract.""" + dev = fixed_device() + dev.timeout = 0.05 + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + + def icmp_sendto(data, addr=None): + ep.protocol.error_received(err) + + ep.sendto = icmp_sendto + with pytest.raises(e.NetworkTimeoutError): + await dev.send_packet(0x6A, b"") + assert ep.closed # dropped after the timeout, so the next call heals + assert dev._transport is None + + run(go()) + + def test_transport_lost_with_error_wakes_the_request(net): """If asyncio closes the transport from its side, the waiting request is told instead of waiting out its timeout."""