Skip to content

Fix the issues found in the review of 1.0.0 #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DAB-LABS merged 1 commit into from
Sep 5, 2026
Merged

Fix the issues found in the review of 1.0.0 #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 55 additions & 5 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,51 @@
All notable changes to this project are recorded here. The format follows
Keep a Changelog; versions follow Semantic Versioning.

## 1.0.1 - 2026-09-05

Fixes from an independent review of 1.0.0, most of them in the transport.
None changes the wire format or the public API.

### Fixed

- A reply to a request that had already timed out could be delivered as the
reply to the next request on the same device, because the persistent
endpoint (new in 1.0.0) is not thrown away between calls the way the old
per-call socket was. Replies are now matched to their request by the
packet counter the device echoes at offset 0x28; a reply carrying the
counter of a request that already timed out is discarded, and a reply
whose counter matches nothing the device sent is still accepted, so
firmware that does not echo the counter is unaffected. Confirmed on an
RM4 Pro, which echoes it.
- `capture()` treated only `StorageError` (-5) as "nothing captured yet".
Some firmware answers `ReadError` (-10); both are now treated as "nothing
yet", matching what the original CLI and Home Assistant do while polling.
The CLI's `--learn` and `--rflearn` inherit the fix.
- Abandoning a capture generator without closing it (for example `break`
out of `async for` to take one code) no longer blocks the next
`capture()` on the same device: opening a new window closes an abandoned
one. Opening a window while another is actively being iterated still
raises `CaptureInProgressError`. A new read-only `Device.capture_active`
property reports whether a window is open.
- Re-authentication is now shared between concurrent callers: when several
requests hit an expired session key at once, the library authenticates
once and every caller retries, instead of one caller re-authenticating
and the others surfacing the raw error. The logged-out code (-2) now
triggers re-authentication as well, matching Home Assistant's own retry.
- Changing `device.host` after the endpoint is open now reopens it against
the new address instead of continuing to talk to the old one.
- `aclose()` while a request is in flight fails that request at once with
`ConnectionClosedError` instead of waiting out the timeout.

### Documentation

- The README explains that `broadlink` and `python-broadlink` install the
same package name and cannot coexist, and how to recover if both were
installed.
- The changelog no longer describes the carried-over device commits as
"intact" (they were squash-merged with `Co-authored-by` credit) and no
longer overstates what the oracle records.

## 1.0.0 - 2026-09-05

This is the first release of `python-broadlink`, a maintained fork of
Expand Down Expand Up @@ -39,6 +84,8 @@ history below starts at that fork point.
#830).
- `pulses_to_data` returns `bytes` (it returned a `bytearray`, against its
own annotation).
- The device's request lock is now a private `_lock` that is actually
acquired; the unused public `Device.lock` attribute is gone.
- Packaging moved to `pyproject.toml`; `setup.py` and the stale
`requirements.txt` pin are gone. The distribution name is now
`python-broadlink`; the import name stays `broadlink`. Python 3.13 or
Expand Down Expand Up @@ -80,7 +127,8 @@ history below starts at that fork point.
read by band and a capture is tagged from what it armed rather than the
byte.
- Devices, carried over from pull requests against the original repository
with their authors' commits intact: RM Max 0xAF8B (#838, Alexey Masolov);
with their authors credited (the changes were squash-merged with
`Co-authored-by` trailers naming each author): RM Max 0xAF8B (#838, Alexey Masolov);
RM5 plus 0x5224 with a new `rm5plus` class (#831, Anil Daoud); RM mini 3
OEM 0xA544 (#823, Bartłomiej Nogaś); RM mini 3 CMCC 0x27C8 (#802,
shuxin); LB26 R1 0xA517 (#812, techitapart); SP mini 3-AL 0x7D15 (#805,
Expand All @@ -92,7 +140,9 @@ history below starts at that fork point.
issue if either does not behave.
- `cryptography` 43 or newer is required, the first release with wheels for
Python 3.13 (supersedes mjg59/python-broadlink#749).
- A test suite. The `tests/oracle` package records the exact request bytes
every public method of every device class sends, and the results it
decodes from canned responses, so that later changes to the transport
can be checked byte for byte against the original behavior.
- A test suite. The `tests/oracle` package records, for every public method
of every device class, the request each one hands to the transport (its
packet type and plaintext payload) and the result it decodes from a canned
response, so that a later reimplementation can be checked against the
original method by method; the framing, encryption and checksum layer is
covered separately by `tests/test_transport.py`.
13 changes: 10 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,9 +60,16 @@ Use pip3 to install the latest version of this module.
pip3 install python-broadlink
```

If the original `broadlink` distribution is also installed in the same
environment, remove it first (`pip3 uninstall broadlink`); both provide the
`broadlink` package.
Both this distribution and the original `broadlink` install a package named
`broadlink`, so only one can be present in an environment at a time. Pip
does not warn about this: installing one on top of the other appears to
succeed, and whichever was installed last is the one that `import broadlink`
finds. If both were installed, uninstall both (`pip3 uninstall broadlink
python-broadlink`) and reinstall this one, since `pip3 uninstall broadlink`
alone removes the shared files and leaves `python-broadlink` registered but
unimportable. This matters most where another package pins `broadlink`:
installing it into the same environment silently replaces this async
library with the original synchronous one.

## Basic functions

Expand Down
118 changes: 86 additions & 32 deletions broadlink/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from __future__ import annotations

import asyncio
import collections
import random
import socket
from collections.abc import AsyncIterator
Expand All @@ -30,8 +31,17 @@
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. -7: control key expired; -4012: control id error.
_REAUTH_CODES = {-7, -4012}
# fresh auth() will fix it. -2: logged out; -7: control key expired;
# -4012: control id error.
_REAUTH_CODES = {-2, -7, -4012}

# How many timed-out request counters to remember, so that a reply to one
# of them arriving late is recognised and dropped instead of being taken as
# the answer to a later request.
_ABANDONED_MAX = 32

_CLOSED = (None, None)
"""Sentinel put on the receive queue when the endpoint is closed."""


class _Protocol(asyncio.DatagramProtocol):
Expand Down Expand Up @@ -203,7 +213,10 @@ def __init__(
self._lock: Optional[asyncio.Lock] = None
self._transport: Optional[asyncio.DatagramTransport] = None
self._protocol: Optional[_Protocol] = None
self._reauth_ok = True
self._endpoint_addr: Optional[Tuple[str, int]] = None
self._abandoned: collections.deque[int] = collections.deque(maxlen=_ABANDONED_MAX)
self._reauth_lock: Optional[asyncio.Lock] = None
self._auth_generation = 0

def __repr__(self) -> str:
"""Return a formal representation of the device."""
Expand Down Expand Up @@ -275,6 +288,7 @@ async def auth(self) -> bool:

self.id = int.from_bytes(payload[:0x4], "little")
self.update_aes(payload[0x04:0x14])
self._auth_generation += 1
return True

async def hello(self, local_ip_address=None) -> bool:
Expand Down Expand Up @@ -363,17 +377,30 @@ def get_type(self) -> str:
# -------------------------------------------------------- transport

async def aclose(self) -> None:
"""Close the device's endpoint. It is reopened on the next call."""
if self._transport is not None:
self._transport.close()
self._transport = None
self._protocol = None
"""Close the device's endpoint. It is reopened on the next call.

A request in flight fails at once with ``ConnectionClosedError``
rather than waiting out its timeout.
"""
transport, protocol = self._transport, self._protocol
self._transport = None
self._protocol = None
self._endpoint_addr = None
if transport is not None:
transport.close()
if protocol is not None:
protocol.queue.put_nowait(_CLOSED) # type: ignore[arg-type]

async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]:
if self._transport is not None and self._endpoint_addr != self.host:
# The caller changed host; the connected socket points at the
# 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._endpoint_addr = self.host
return self._transport, self._protocol # type: ignore[return-value]

def _frame(self, packet_type: int, payload: bytes) -> bytes:
Expand Down Expand Up @@ -419,28 +446,53 @@ def _validate(resp: bytes) -> bytes:
return resp

async def _exchange(self, packet: bytes) -> bytes:
"""Send one frame and wait for one reply, resending on silence."""
"""Send one frame and wait for its reply, resending on silence.

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
echo it.
"""
transport, protocol = await self._endpoint()
protocol.drain()
loop = asyncio.get_running_loop()
start = loop.time()
timeout = self.timeout
count = int.from_bytes(packet[0x28:0x2A], "little")

while True:
transport.sendto(packet)
time_left = timeout - (loop.time() - start)
wait = min(DEFAULT_RETRY_INTVL, time_left)
try:
resp, _ = await asyncio.wait_for(protocol.queue.get(), max(wait, 0))
except asyncio.TimeoutError:
if (loop.time() - start) >= timeout:
raise e.NetworkTimeoutError(
-4000,
"Network timeout",
f"No response received within {timeout}s",
) from None
continue
return self._validate(resp)
resend_at = loop.time() + DEFAULT_RETRY_INTVL
while True:
now = loop.time()
if now - start >= timeout:
break
wait = min(resend_at, start + timeout) - now
try:
resp, _ = await asyncio.wait_for(protocol.queue.get(), max(wait, 0))
except asyncio.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"
)
resp = self._validate(resp)
reply_count = int.from_bytes(resp[0x28:0x2A], "little")
if reply_count == count or reply_count not in self._abandoned:
return resp
# A late answer to a request we gave up on: keep waiting.
if loop.time() - start >= timeout:
self._abandoned.append(count)
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
Expand All @@ -449,22 +501,24 @@ async def send_packet(

If the device answers that the session key is no longer valid, the
session is re-authenticated once and the request is sent again.
Concurrent callers that hit the same expired key share one
re-authentication and each retry once.
"""
if self._lock is None:
self._lock = asyncio.Lock()
self._reauth_lock = asyncio.Lock()
generation = self._auth_generation
async with self._lock:
resp = await self._exchange(self._frame(packet_type, bytes(payload)))

if _reauth and self._reauth_ok:
if _reauth:
code = int.from_bytes(resp[0x22:0x24], "little", signed=True)
if code in _REAUTH_CODES:
self._reauth_ok = False
try:
await self.auth()
async with self._lock:
resp = await self._exchange(
self._frame(packet_type, bytes(payload))
)
finally:
self._reauth_ok = True
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
Loading
Loading