From e55bc74b2a82c54a549aef225138df9714953a95 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sat, 5 Sep 2026 03:18:30 +0000 Subject: [PATCH] Add capture() and capture_rf() with packet helpers A learning session is arm, poll, wait, re-arm, one code at a time, with a device that leaves learning mode without saying so and a send that ends the session. capture() and capture_rf() own that loop and hand back clean signals, so a consumer subscribes and reads instead of reimplementing the dance (as the remote platform, the receiver PR and others each did). - capture(window, stop_after_first, poll_interval, rearm_interval): async generator yielding CapturedSignal. Re-arms on a timer (default 15 s, under the 25-40 s the device was measured to hold a session) and after any send_data (a send ends the session; both from the bench). One window per device; a second raises CaptureInProgressError. - capture_rf(window, frequency, ...) on the Pro classes: takes the carrier directly, or sweeps for it when not given. The sweep is unreliable on some firmware, so the known-frequency path is primary. - CapturedSignal: device packet, pulses at the corrected tick, kind, repeat, and the RF carrier the packet does not itself record. - pulses_to_data gains kind and repeat; parse_packet is the inverse; SignalKind names the bands. A returned RF packet does not always carry the canonical type byte (an RM4 Pro sends 0xB1 for 433 MHz), so kind is read by band and a capture is tagged from what it armed, never dropped on the byte. - One shared front-end lock and a transmit generation counter already live on the device; capture reads the counter so a concurrent send re-arms the window. Tests drive the loops against a scripted device that models the bench findings; the transport oracle fixtures are unchanged. --- CHANGELOG.md | 16 ++ README.md | 40 +++ broadlink/exceptions.py | 9 + broadlink/remote.py | 341 ++++++++++++++++++++++++- tests/test_capture.py | 548 ++++++++++++++++++++++++++++++++++++++++ tests/test_oracle.py | 3 + 6 files changed, 952 insertions(+), 5 deletions(-) create mode 100644 tests/test_capture.py diff --git a/CHANGELOG.md b/CHANGELOG.md index fb74a4b4..cc1a750d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,22 @@ history below starts at that fork point. ### Added +- `capture()` and `capture_rf()`, async generators that own the arm, poll, + timeout and re-arm loop of a learning session and yield each signal as a + `CapturedSignal` (device packet, decoded pulses at the correct tick, + kind, repeat count, and for RF the carrier frequency). They re-arm on a + timer, because the device leaves learning mode silently, and after any + `send_data`, because a transmission ends the session; both intervals and + the poll cadence were set from a bench on an RM4 Pro. Only one window can + be open per device. `capture_rf()` (Pro models only) takes the carrier + frequency directly and falls back to the on-device sweep when it is not + given. +- Packet helpers: `pulses_to_data` takes `kind` and `repeat`, `parse_packet` + is its inverse, and `SignalKind` names the IR, 433 MHz and 315 MHz bands. + A device's returned RF packet does not always use the canonical type byte + (an RM4 Pro answers a 433 MHz capture with 0xB1, not 0xB2), so the kind is + 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); RM5 plus 0x5224 with a new `rm5plus` class (#831, Anil Daoud); RM mini 3 diff --git a/README.md b/README.md index f3fe226f..6d5d4d5f 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,46 @@ You can exit the learning mode in the middle of the process by calling this meth await device.cancel_sweep_frequency() ``` +### Capturing signals + +`capture()` wraps the arm, poll, timeout and re-arm dance above into one +async generator that yields each signal it hears as a `CapturedSignal`: + +```python3 +from contextlib import aclosing + +async with aclosing(device.capture(window=30)) as signals: + async for signal in signals: + print(signal.kind, len(signal.pulses), "pulses") + await other_device.send_data(signal.packet) +``` + +By default the window closes after the first signal. Pass +`stop_after_first=False` to keep it open for the whole `window` (in seconds; +`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. + +`CapturedSignal` carries the device's own `packet` bytes (ready for +`send_data`), the decoded `pulses` in microseconds at the correct 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. + +RF works the same way on the Pro models, with the carrier as the one extra +input: + +```python3 +async with aclosing(device.capture_rf(window=30, frequency=433.92)) as signals: + async for signal in signals: + ... +``` + +Pass `frequency` whenever you know it. Without it the device first sweeps +for the carrier while you hold a button down, then learns the code from a +fresh press; the sweep is unreliable on some firmware and can report a +carrier it never really locked, so the known-frequency path is preferred. + ### Sending IR/RF packets ```python3 await device.send_data(packet) diff --git a/broadlink/exceptions.py b/broadlink/exceptions.py index 2343ad6e..8f2ecc6c 100644 --- a/broadlink/exceptions.py +++ b/broadlink/exceptions.py @@ -95,6 +95,15 @@ class StorageError(BroadlinkException): """Storage error.""" +class CaptureInProgressError(BroadlinkException): + """A capture window is already open on this device. + + A universal remote has one receiver, so only one ``capture`` or + ``capture_rf`` window can be open at a time. Close the running one + before opening another. + """ + + class WriteError(BroadlinkException): """Write error.""" diff --git a/broadlink/remote.py b/broadlink/remote.py index 2aa3c464..9f905618 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -1,7 +1,11 @@ """Support for universal remotes.""" +import asyncio +import enum import struct -from typing import List, Optional, Tuple +import time +from dataclasses import dataclass, field +from typing import AsyncIterator, Awaitable, Callable, List, Optional, Tuple from . import exceptions as e from .device import Device @@ -16,11 +20,73 @@ device were unaffected because both directions shared the constant. """ +DEFAULT_POLL_INTERVAL = 0.5 +"""Seconds between ``check_data`` polls while a capture window is open.""" -def pulses_to_data(pulses: List[int], tick: float = TICK) -> bytes: - """Convert a microsecond duration sequence into a Broadlink IR packet.""" +DEFAULT_REARM_INTERVAL = 15.0 +"""Seconds after which an open capture window re-enters learning mode. + +The RM4 Pro leaves learning mode silently between 25 s and 40 s after +``enter_learning`` (bench, 2026-09-04), and any ``send_data`` also ends the +session, while ``check_data`` keeps answering with the same "nothing yet" +error, so an open window has to re-arm on a timer and after every send. +""" + + +class SignalKind(enum.IntEnum): + """The kind of signal a packet carries. + + The values are the canonical type bytes the library writes when it + builds a packet (protocol.md offset 0x00). Packets a device returns + from a learn session do not always use exactly these bytes -- an RM4 + Pro returns 0xB1 for a 433 MHz capture, not 0xB2 -- so read a returned + packet's kind with ``classify`` rather than by equality. + """ + + IR = 0x26 + RF_433 = 0xB2 + RF_315 = 0xD7 + + @property + def is_rf(self) -> bool: + return self is not SignalKind.IR + + @classmethod + def classify(cls, type_byte: int) -> "SignalKind": + """Map a packet's raw first byte to a kind, tolerantly. + + The RF learn path returns bytes in the 0xB_ (433 MHz) and 0xD_ + (315 MHz) ranges whose low bits are not documented and vary by + firmware, so classify by range rather than by exact value. Raises + ``ValueError`` for a byte in no known range. + """ + if type_byte == cls.IR: + return cls.IR + if type_byte & 0xF0 == 0xB0: + return cls.RF_433 + if type_byte & 0xF0 == 0xD0: + return cls.RF_315 + raise ValueError(f"Unknown packet type 0x{type_byte:02x}") + + +def pulses_to_data( + pulses: List[int], + tick: float = TICK, + *, + kind: SignalKind = SignalKind.IR, + repeat: int = 0, +) -> bytes: + """Convert a microsecond duration sequence into a Broadlink packet. + + ``kind`` selects the type byte (IR, RF 433 MHz or RF 315 MHz) and + ``repeat`` is the number of extra transmissions the device performs + after the first, 0 to 255 (protocol.md offset 0x01). + """ + if not 0 <= repeat <= 0xFF: + raise ValueError("repeat must be between 0 and 255") result = bytearray(4) - result[0x00] = 0x26 + result[0x00] = SignalKind(kind) + result[0x01] = repeat for pulse in pulses: div, mod = divmod(round(pulse / tick), 256) @@ -33,7 +99,7 @@ def pulses_to_data(pulses: List[int], tick: float = TICK) -> bytes: result[0x02] = data_len & 0xFF result[0x03] = data_len >> 8 - return result + return bytes(result) def data_to_pulses(data: bytes, tick: float = TICK) -> List[int]: @@ -58,11 +124,98 @@ def data_to_pulses(data: bytes, tick: float = TICK) -> List[int]: return result +@dataclass(frozen=True) +class ParsedPacket: + """The parts of a Broadlink packet: kind, repeat count and timings. + + ``type_byte`` is the packet's raw first byte; ``kind`` is that byte + classified into a band (see ``SignalKind.classify``), which for a + device-returned RF packet is not always the canonical value. + """ + + kind: SignalKind + repeat: int + pulses: List[int] + type_byte: int + + +def parse_packet(data: bytes, tick: float = TICK) -> ParsedPacket: + """Split a Broadlink packet into its kind, repeat count and timings. + + Raises ``ValueError`` if the packet is shorter than its header or the + type byte is in no known band (IR, 433 MHz or 315 MHz). + """ + if len(data) < 4: + raise ValueError("Malformed data.") + kind = SignalKind.classify(data[0x00]) + return ParsedPacket(kind, data[0x01], data_to_pulses(data, tick), data[0x00]) + + +@dataclass(frozen=True) +class CapturedSignal: + """One signal captured by a universal remote. + + ``packet`` is the device's own bytes, ready for ``send_data`` and for + storage; ``pulses`` is the same signal as microsecond durations at the + corrected tick. ``kind`` is the band the signal was captured on; + ``type_byte`` is the packet's raw first byte, which for RF is not always + the canonical value for the band. ``frequency_mhz`` is set for RF + captures only and holds the carrier the device swept to or was given, + which the packet itself does not record. + """ + + packet: bytes + kind: SignalKind + pulses: List[int] = field(repr=False) + repeat: int = 0 + frequency_mhz: Optional[float] = None + type_byte: Optional[int] = None + captured_at: float = field(default_factory=time.time, repr=False) + + @classmethod + def from_packet( + cls, + packet: bytes, + frequency_mhz: Optional[float] = None, + *, + kind: Optional[SignalKind] = None, + ) -> "CapturedSignal": + """Build a signal from a device-returned packet. + + ``kind`` overrides the band read from the packet's type byte. A + capture window knows what it armed, so it passes the kind it armed + for and a signal is never dropped over an unexpected type byte; the + raw byte is still kept in ``type_byte``. The timings are read from + the packet regardless of the type byte. + """ + if len(packet) < 4: + raise ValueError("Malformed data.") + type_byte = packet[0x00] + if kind is None: + kind = SignalKind.classify(type_byte) + return cls( + bytes(packet), + kind, + data_to_pulses(packet), + packet[0x01], + frequency_mhz, + type_byte, + ) + + class rmmini(Device): """Controls a Broadlink RM mini 3.""" TYPE = "RMMINI" + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + # Bumped by every transmission. An open capture window compares it + # 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 + self._capture_open = False + async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" packet = struct.pack(" None: async def send_data(self, data: bytes) -> None: """Send a code to the device.""" + self._tx_generation += 1 await self._send(0x2, data) async def enter_learning(self) -> None: @@ -89,6 +243,104 @@ async def check_data(self) -> bytes: """Return the last captured code.""" return await self._send(0x4) + def capture( + self, + window: float = 30.0, + *, + stop_after_first: bool = True, + poll_interval: float = DEFAULT_POLL_INTERVAL, + rearm_interval: float = DEFAULT_REARM_INTERVAL, + ) -> AsyncIterator[CapturedSignal]: + """Open an infrared capture window and yield what the device hears. + + The device is put into learning mode and polled every + ``poll_interval`` seconds. Each code it reports is yielded as a + ``CapturedSignal``. With ``stop_after_first`` the window closes + after the first code; otherwise the device is re-armed after each + code (it holds one code per learning session) and the window stays + open until ``window`` seconds have passed. ``window=0`` keeps it + open until the generator is closed. + + The device leaves learning mode on its own after a while without + saying so, so the window re-arms it every ``rearm_interval`` seconds + and after every ``send_data`` on the same device. Closing the + 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; + a second raises ``CaptureInProgressError``. + """ + return self._capture_loop( + self.enter_learning, + window, + stop_after_first, + poll_interval, + rearm_interval, + SignalKind.IR, + None, + ) + + async def _capture_loop( + self, + arm: Callable[[], Awaitable[None]], + window: float, + stop_after_first: bool, + poll_interval: float, + rearm_interval: float, + kind: SignalKind, + frequency_mhz: Optional[float], + ) -> AsyncIterator[CapturedSignal]: + 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 self._capture_open: + raise e.CaptureInProgressError("A capture window is already open") + + self._capture_open = True + try: + loop = asyncio.get_running_loop() + deadline = loop.time() + window if window else None + timeouts = 0 + + await arm() + armed_at = loop.time() + generation = self._tx_generation + + while True: + now = loop.time() + if deadline is not None and now >= deadline: + return + delay = poll_interval + if deadline is not None: + delay = min(delay, deadline - now) + await asyncio.sleep(delay) + + try: + data = await self.check_data() + except e.StorageError: + data = b"" # The device's answer for "nothing yet". + except e.NetworkTimeoutError: + timeouts += 1 + if timeouts >= 3: + raise + generation = -1 # Re-arm; the device's state is unknown. + continue + 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. + + now = loop.time() + if generation != self._tx_generation or now - armed_at >= rearm_interval: + await arm() + armed_at = loop.time() + generation = self._tx_generation + finally: + self._capture_open = False + class rmpro(rmmini): """Controls a Broadlink RM pro.""" @@ -117,6 +369,85 @@ async def cancel_sweep_frequency(self) -> None: """Cancel sweep frequency.""" await self._send(0x1E) + async def capture_rf( + self, + window: float = 30.0, + *, + frequency: Optional[float] = None, + stop_after_first: bool = True, + poll_interval: float = DEFAULT_POLL_INTERVAL, + rearm_interval: float = DEFAULT_REARM_INTERVAL, + ) -> AsyncIterator[CapturedSignal]: + """Open a radio frequency capture window and yield what the device hears. + + With ``frequency`` (in MHz, for example 433.92) the device goes + straight into RF learning mode on that carrier. Without it the + device first sweeps for the carrier while the user HOLDS a button on + the remote, and only then learns the code from a fresh press; the + sweep is unreliable on some firmware and can report a carrier it + never really locked, so pass the frequency whenever it is known. + + The window, polling, re-arm and stop-after-first semantics are those + of ``capture``; the sweep counts against the same ``window``. A + ``send_data`` during the sweep restarts it. Each ``CapturedSignal`` + carries the carrier in ``frequency_mhz``, which the packet itself + does not record. + """ + if self._capture_open: + raise e.CaptureInProgressError("A capture window is already open") + if window < 0 or poll_interval <= 0: + raise ValueError("window must be 0 or positive, poll_interval positive") + + loop = asyncio.get_running_loop() + deadline = loop.time() + window if window else None + + if frequency is None: + self._capture_open = True + try: + frequency = await self._sweep(deadline, poll_interval) + finally: + self._capture_open = False + if frequency is None: + return + if deadline is not None: + window = max(deadline - loop.time(), 0.0) + if window == 0: + return + + async def arm() -> None: + await self.find_rf_packet(frequency) + + kind = SignalKind.RF_315 if frequency < 400 else SignalKind.RF_433 + async for signal in self._capture_loop( + arm, window, stop_after_first, poll_interval, rearm_interval, kind, frequency + ): + yield signal + + async def _sweep( + self, deadline: Optional[float], poll_interval: float + ) -> Optional[float]: + """Sweep for the remote's carrier; return it in MHz, or None if the + window ran out first.""" + loop = asyncio.get_running_loop() + await self.sweep_frequency() + generation = self._tx_generation + while True: + now = loop.time() + if deadline is not None and now >= deadline: + await self.cancel_sweep_frequency() + return None + delay = poll_interval + if deadline is not None: + delay = min(delay, deadline - now) + await asyncio.sleep(delay) + if generation != self._tx_generation: + await self.sweep_frequency() + generation = self._tx_generation + continue + found, frequency = await self.check_frequency() + if found: + return frequency + async def check_sensors(self) -> dict: """Return the state of the sensors.""" resp = await self._send(0x1) diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 00000000..14ed8c50 --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,548 @@ +"""Capture windows and packet helpers, against a scripted universal remote. + +The fake replaces ``send_packet`` on a real device instance, decodes the +command the class framed, and behaves like an RM4 Pro as measured on the +bench: it holds one code per learning session, answers ``check_data`` with +``StorageError`` -5 until a code lands, and drops presses while not armed. +""" + +from __future__ import annotations + +import asyncio +import struct +from contextlib import aclosing + +import pytest + +import broadlink +from broadlink import exceptions as e +from broadlink.remote import ( + CapturedSignal, + SignalKind, + data_to_pulses, + parse_packet, + pulses_to_data, +) +from tests.oracle.harness import HOST, MAC, make_response + +IR = pulses_to_data([9000, 4500, 560, 560, 560, 1690]) +RF = pulses_to_data([300, 900, 300, 900], kind=SignalKind.RF_433) + +CMD_SEND = 0x02 +CMD_LEARN = 0x03 +CMD_CHECK = 0x04 +CMD_SWEEP = 0x19 +CMD_CHECK_FREQ = 0x1A +CMD_FIND_RF = 0x1B +CMD_CANCEL_SWEEP = 0x1E + + +class FakeRM: + """A scripted RM: one receiver, one code per arm, silent expiry.""" + + def __init__(self, device: broadlink.Device, framing: str) -> None: + self.device = device + self.framing = framing # "rmmini" ( bool: + """A remote is pressed at the device. Captured only while armed.""" + if not self.armed: + return False + self.pending = packet + self.armed = False # One code per learning session. + return True + + def expire(self) -> None: + """The device leaves learning mode without telling anyone.""" + self.armed = False + + # -- fake transport + async def send_packet(self, packet_type: int, payload: bytes) -> bytes: + assert packet_type == 0x6A + if self.framing == "rmmini": + command = struct.unpack(" tuple[bytes, int | str]: + if command == CMD_LEARN: + self.armed = True + self.rf_frequency = None + return b"", 0 + if command == CMD_FIND_RF: + self.armed = True + self.rf_frequency = struct.unpack(" 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]: + 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" + return device, FakeRM(device, framing) + + +def run(coro): + return asyncio.run(coro) + + +async def press_later(fake: FakeRM, packet: bytes, delay: float) -> bool: + await asyncio.sleep(delay) + return fake.press(packet) + + +FAST = dict(poll_interval=0.01, rearm_interval=10.0) + + +# ------------------------------------------------------------- IR windows + + +@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) + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, IR, 0.03)) + signals = [s async for s in device.capture(window=2, **FAST)] + return signals + + signals = run(go()) + assert len(signals) == 1 + sig = signals[0] + assert isinstance(sig, CapturedSignal) + assert sig.packet == IR + assert sig.kind is SignalKind.IR + assert sig.pulses == data_to_pulses(IR) + assert sig.frequency_mhz is None + assert fake.commands[0][0] == CMD_LEARN + assert fake.count(CMD_LEARN) == 1 + assert fake.count(CMD_CHECK) >= 2 + assert device._capture_open is False + + +def test_capture_window_elapses_with_nothing(): + device, fake = make() + signals = run(_collect(device.capture(window=0.05, **FAST))) + assert signals == [] + assert fake.count(CMD_LEARN) == 1 + assert fake.count(CMD_CHECK) >= 3 + assert device._capture_open is False + + +async def _collect(gen): + return [s async for s in gen] + + +def test_capture_keeps_going_and_rearms_after_each_code(): + device, fake = make() + + async def go(): + loop = asyncio.get_running_loop() + loop.create_task(press_later(fake, IR, 0.02)) + loop.create_task(press_later(fake, RF, 0.06)) + return [s async for s in device.capture(window=0.12, stop_after_first=False, **FAST)] + + signals = run(go()) + assert [s.packet for s in signals] == [IR, RF] + # Armed once at the start and once after each code. + assert fake.count(CMD_LEARN) == 3 + + +def test_press_between_code_and_rearm_is_lost_but_next_is_not(): + """The device holds one code per session; a second press before the + window re-arms is gone, as measured on the bench.""" + device, fake = make() + + async def go(): + loop = asyncio.get_running_loop() + results = [] + + async def presses(): + await asyncio.sleep(0.02) + results.append(fake.press(IR)) + results.append(fake.press(RF)) # Device not armed: lost. + await asyncio.sleep(0.03) + results.append(fake.press(RF)) # Re-armed by then. + + loop.create_task(presses()) + signals = [s async for s in device.capture(window=0.1, stop_after_first=False, **FAST)] + return results, signals + + results, signals = run(go()) + assert results == [True, False, True] + assert [s.packet for s in signals] == [IR, RF] + + +def test_send_during_window_rearms(): + device, fake = make() + + async def go(): + async def send_then_press(): + await asyncio.sleep(0.02) + await device.send_data(IR) + fake.expire() # Whatever the send did to the session, assume the worst. + await asyncio.sleep(0.03) + return fake.press(RF) + + loop = asyncio.get_running_loop() + task = loop.create_task(send_then_press()) + signals = [s async for s in device.capture(window=0.2, **FAST)] + return await task, signals + + pressed, signals = run(go()) + assert pressed is True + assert [s.packet for s in signals] == [RF] + assert fake.count(CMD_SEND) == 1 + assert fake.count(CMD_LEARN) == 2 + learn_positions = [i for i, (c, _) in enumerate(fake.commands) if c == CMD_LEARN] + send_position = next(i for i, (c, _) in enumerate(fake.commands) if c == CMD_SEND) + assert learn_positions[0] < send_position < learn_positions[1] + + +def test_timed_rearm_recovers_from_silent_expiry(): + device, fake = make() + + async def go(): + async def expire_then_press(): + await asyncio.sleep(0.02) + fake.expire() + assert fake.press(IR) is False # Lost: the device is deaf. + await asyncio.sleep(0.05) # Past the re-arm interval. + return fake.press(IR) + + loop = asyncio.get_running_loop() + task = loop.create_task(expire_then_press()) + signals = [ + s async for s in device.capture(window=0.3, poll_interval=0.01, rearm_interval=0.04) + ] + return await task, signals + + pressed, signals = run(go()) + assert pressed is True + assert len(signals) == 1 + assert fake.count(CMD_LEARN) >= 2 + + +def test_open_ended_window_runs_until_closed(): + device, fake = make() + + async def go(): + got = [] + async with aclosing(device.capture(window=0, stop_after_first=False, **FAST)) as gen: + asyncio.get_running_loop().create_task(press_later(fake, IR, 0.02)) + async for s in gen: + got.append(s) + if len(got) == 1: + break + return got + + got = run(go()) + assert len(got) == 1 + assert device._capture_open is False + # Closing sends nothing further to the device. + assert fake.commands[-1][0] in (CMD_CHECK, CMD_LEARN) + + +def test_second_window_is_refused(): + device, fake = make() + + async def go(): + task = asyncio.get_running_loop().create_task( + _collect(device.capture(window=1, **FAST)) + ) + await asyncio.sleep(0.02) + with pytest.raises(e.CaptureInProgressError): + await _collect(device.capture(window=1, **FAST)) + assert device._capture_open is True + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + run(go()) + assert device._capture_open is False + + +def test_transport_timeouts_rearm_then_give_up(): + device, fake = make() + fake.timeouts_to_raise = 2 + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, IR, 0.05)) + return [s async for s in device.capture(window=1, **FAST)] + + signals = run(go()) + assert len(signals) == 1 + assert fake.count(CMD_LEARN) >= 2 # Re-armed after the timeouts. + + device, fake = make() + fake.timeouts_to_raise = 3 + with pytest.raises(e.NetworkTimeoutError): + run(_collect(device.capture(window=1, **FAST))) + assert device._capture_open is False + + +def test_capture_rejects_bad_arguments(): + device, _ = make() + with pytest.raises(ValueError): + run(_collect(device.capture(window=-1))) + with pytest.raises(ValueError): + run(_collect(device.capture(poll_interval=0))) + with pytest.raises(ValueError): + run(_collect(device.capture(rearm_interval=0))) + + +# ------------------------------------------------------------- RF windows + + +def test_capture_rf_with_known_frequency_skips_the_sweep(): + device, fake = make() + + async def go(): + asyncio.get_running_loop().create_task(press_later(fake, RF, 0.03)) + return [s async for s in device.capture_rf(window=1, frequency=433.92, **FAST)] + + signals = run(go()) + assert len(signals) == 1 + sig = signals[0] + assert sig.kind is SignalKind.RF_433 + assert sig.frequency_mhz == 433.92 + assert sig.packet == RF + assert fake.count(CMD_SWEEP) == 0 + assert fake.commands[0] == (CMD_FIND_RF, struct.pack(" kinds.index(CMD_CHECK_FREQ) + assert fake.commands[find][1] == struct.pack(" 0 diff --git a/tests/test_oracle.py b/tests/test_oracle.py index f5159d62..6cdd4d2b 100644 --- a/tests/test_oracle.py +++ b/tests/test_oracle.py @@ -37,6 +37,9 @@ def test_every_public_method_is_covered() -> None: # test_transport.py, not here. 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"} missing = [] for name, cls in inspect.getmembers(broadlink, inspect.isclass): if not issubclass(cls, Device):