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."""