From 559204bf920a59b02772f83ca46dc7b470e7f5f9 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Sat, 5 Sep 2026 03:41:15 +0000 Subject: [PATCH] Prepare the 1.0.0 release: CLI on capture(), docs, version - broadlink_cli --learn and --rflearn learn through capture() and capture_rf(), so a session no longer goes deaf when the device times out partway through the 30 s wait. New --window (seconds to listen), --keep (print every code heard) and --repeat (with --send --durations). - cli/README.md: --rfscanlearn was a typo for --rflearn (#803, #830); install line is pip install python-broadlink; examples for --frequency, --window, --keep and --repeat. - README: four device calls in the switch examples were missing await; the RF check_frequency example now unpacks the (found, frequency) tuple. - CHANGELOG: 1.0.0 heading, the CLI changes, and the pulses_to_data bytes-instead-of-bytearray note. - tests/test_capture.py timings expressed as multiples of a 30 ms unit (was 10 ms) for headroom on slower CI runners. - Version 1.0.0. --- CHANGELOG.md | 12 ++++- README.md | 16 +++---- cli/README.md | 26 +++++++++-- cli/broadlink_cli | 105 +++++++++++++++++++----------------------- pyproject.toml | 2 +- tests/test_capture.py | 57 ++++++++++++----------- 6 files changed, 120 insertions(+), 98 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cc1a750d..b1068bec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ All notable changes to this project are recorded here. The format follows Keep a Changelog; versions follow Semantic Versioning. -## Unreleased +## 1.0.0 - 2026-09-05 This is the first release of `python-broadlink`, a maintained fork of `mjg59/python-broadlink` (PyPI `broadlink`, last released as 0.19.0). The @@ -30,7 +30,15 @@ history below starts at that fork point. - Retry and timeout behaviour is unchanged: a request is repeated every second until `timeout` elapses, then `NetworkTimeoutError` is raised. - `dooya.set_percentage_and_wait` sleeps with `asyncio.sleep`. -- The CLI tools run their body under `asyncio.run`. +- The CLI tools run their body under `asyncio.run`. `broadlink_cli + --learn` and `--rflearn` use `capture()` / `capture_rf()`, so a learning + session no longer goes deaf when the device times out partway through; + `--window` sets how long to listen, `--keep` prints every code heard, and + `--send --durations --repeat N` sets the repeat count. The CLI README's + `--rfscanlearn` was a typo for `--rflearn` (mjg59/python-broadlink#803, + #830). +- `pulses_to_data` returns `bytes` (it returned a `bytearray`, against its + own annotation). - 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 diff --git a/README.md b/README.md index 6d5d4d5f..3126066d 100644 --- a/README.md +++ b/README.md @@ -11,8 +11,8 @@ A Python module and CLI for controlling Broadlink devices locally. > upstream [#839](https://github.com/mjg59/python-broadlink/issues/839) > (fix in [#841](https://github.com/mjg59/python-broadlink/pull/841)) and > adds the devices waiting in upstream's pull request queue, including the -> RM Max and RM5 Plus. Version 1.0 will be asynchronous; see `CHANGELOG.md`. -> Upstream's credit and MIT license are preserved. +> RM Max and RM5 Plus. Version 1.0 is asynchronous and adds `capture()`; +> see `CHANGELOG.md`. Upstream's credit and MIT license are preserved. ## Version 1.0 is asynchronous @@ -166,9 +166,9 @@ await device.sweep_frequency() 2. When the LED blinks, point the remote at the Broadlink device for the first time and long press the button you want to learn. 3. Check if the frequency was successfully identified: ```python3 -ok = device.check_frequency() +ok, frequency = await device.check_frequency() if ok: - print('Frequency found!') + print(f'Frequency found: {frequency} MHz') ``` 4. Enter learning mode: ```python3 @@ -251,12 +251,12 @@ await device.set_power(False) ### Checking power state ```python3 -state = device.check_power() +state = await device.check_power() ``` ### Checking energy consumption ```python3 -state = device.get_energy() +state = await device.get_energy() ``` ## Power strips @@ -269,14 +269,14 @@ await device.set_power(1, False) ### Checking power state ```python3 -state = device.check_power() +state = await device.check_power() ``` ## Light bulbs ### Fetching data ```python3 -state = device.get_state() +state = await device.get_state() ``` ### Setting state attributes diff --git a/cli/README.md b/cli/README.md index b7e48dc9..301196af 100644 --- a/cli/README.md +++ b/cli/README.md @@ -6,9 +6,9 @@ This is a command line interface for the python-broadlink API. Requirements ------------ -You need to install the module first: +You need to install the module first (Python 3.13 or newer): ``` -pip3 install broadlink +pip install python-broadlink ``` Installation @@ -67,7 +67,13 @@ broadlink_cli --device @BEDROOM.device --learn #### Learn RF code and show at console ``` -broadlink_cli --device @BEDROOM.device --rfscanlearn +broadlink_cli --device @BEDROOM.device --rflearn +``` +The device sweeps for the remote's carrier while you hold a button, then +learns the code from a short press. The sweep is unreliable on some +firmware; if you know the carrier, skip it: +``` +broadlink_cli --device @BEDROOM.device --rflearn --frequency 433.92 ``` #### Learn IR code and save to file @@ -77,7 +83,14 @@ broadlink_cli --device @BEDROOM.device --learnfile LG-TV.power #### Learn RF code and save to file ``` -broadlink_cli --device @BEDROOM.device --rfscanlearn --learnfile LG-TV.power +broadlink_cli --device @BEDROOM.device --rflearn --learnfile LG-TV.power +``` + +#### Listen for longer, or for several codes +`--window` sets how many seconds to listen (default 30); `--keep` prints +every code heard during the window instead of stopping at the first: +``` +broadlink_cli --device @BEDROOM.device --learn --window 120 --keep ``` #### Send code @@ -90,6 +103,11 @@ broadlink_cli --device @BEDROOM.device --send DATA broadlink_cli --device @BEDROOM.device --send @LG-TV.power ``` +#### Send microsecond durations, repeated +``` +broadlink_cli --device @BEDROOM.device --send --durations --repeat 2 +9000 -4500 +560 -560 +``` + #### Check temperature ``` broadlink_cli --device @BEDROOM.device --temperature diff --git a/cli/broadlink_cli b/cli/broadlink_cli index 1014986b..9512c7cb 100644 --- a/cli/broadlink_cli +++ b/cli/broadlink_cli @@ -2,13 +2,14 @@ import argparse import asyncio import base64 +import sys import time +from contextlib import aclosing from typing import List import broadlink from broadlink.const import DEFAULT_PORT -from broadlink.exceptions import ReadError, StorageError -from broadlink.remote import data_to_pulses, pulses_to_data +from broadlink.remote import CapturedSignal, data_to_pulses, pulses_to_data TIMEOUT = 30 @@ -30,6 +31,35 @@ def parse_pulses(data: List[str]) -> List[int]: return [abs(int(s)) for s in data] +def show(signal: CapturedSignal) -> None: + """Print a captured signal in every format and save it if asked.""" + raw_fmt = signal.packet.hex() + base64_fmt = base64.b64encode(signal.packet).decode('ascii') + pulse_fmt = format_pulses(signal.pulses) + + print("Packet found!") + if signal.frequency_mhz: + print("Frequency: {}MHz".format(signal.frequency_mhz)) + print("Raw:", raw_fmt) + print("Base64:", base64_fmt) + print("Pulses:", pulse_fmt) + + if args.learnfile: + print("Saving to {}".format(args.learnfile)) + with open(args.learnfile, "w") as text_file: + text_file.write(pulse_fmt if args.durations else raw_fmt) + + +async def listen(window) -> int: + """Drain a capture window, printing each signal; return how many.""" + heard = 0 + async with aclosing(window) as signals: + async for signal in signals: + heard += 1 + show(signal) + return heard + + parser = argparse.ArgumentParser(fromfile_prefix_chars='@') parser.add_argument("--device", help="device definition as 'type host mac'") parser.add_argument("--type", type=auto_int, default=0x2712, help="type of device") @@ -51,6 +81,12 @@ parser.add_argument("--learn", action="store_true", help="learn command") parser.add_argument("--rflearn", action="store_true", help="rf scan learning") parser.add_argument("--frequency", type=float, help="specify radiofrequency for learning") parser.add_argument("--learnfile", help="save learned command to a specified file") +parser.add_argument("--window", type=float, default=TIMEOUT, + help="seconds to keep listening while learning (default %(default)s)") +parser.add_argument("--keep", action="store_true", + help="keep listening for the whole window and print every code heard") +parser.add_argument("--repeat", type=int, default=0, + help="with --send --durations: extra transmissions after the first") parser.add_argument("--durations", action="store_true", help="use durations in micro seconds instead of the Broadlink format") parser.add_argument("--convert", action="store_true", help="convert input data to durations") @@ -95,40 +131,17 @@ async def main(): print("{} {}".format(key, data[key])) if args.send: data = ( - pulses_to_data(parse_pulses(args.data)) + pulses_to_data(parse_pulses(args.data), repeat=args.repeat) if args.durations else bytes.fromhex(''.join(args.data)) ) await dev.send_data(data) if args.learn or (args.learnfile and not args.rflearn): - await dev.enter_learning() print("Learning...") - start = time.time() - while time.time() - start < TIMEOUT: - await asyncio.sleep(1) - try: - data = await dev.check_data() - except (ReadError, StorageError): - continue - else: - break - else: + heard = await listen(dev.capture(window=args.window, stop_after_first=not args.keep)) + if not heard: print("No data received...") - exit(1) - - print("Packet found!") - raw_fmt = data.hex() - base64_fmt = base64.b64encode(data).decode('ascii') - pulse_fmt = format_pulses(data_to_pulses(data)) - - print("Raw:", raw_fmt) - print("Base64:", base64_fmt) - print("Pulses:", pulse_fmt) - - if args.learnfile: - print("Saving to {}".format(args.learnfile)) - with open(args.learnfile, "w") as text_file: - text_file.write(pulse_fmt if args.durations else raw_fmt) + sys.exit(1) if args.check: if await dev.check_power(): print('* ON *') @@ -187,7 +200,7 @@ async def main(): else: print("Radiofrequency not found") await dev.cancel_sweep_frequency() - exit(1) + sys.exit(1) print("Radiofrequency detected: {}MHz".format(frequency)) print("You can now let go of the button") @@ -196,34 +209,12 @@ async def main(): print("Press the button again, now a short press.") - await dev.find_rf_packet(frequency) - - start = time.time() - while time.time() - start < TIMEOUT: - await asyncio.sleep(1) - try: - data = await dev.check_data() - except (ReadError, StorageError): - continue - else: - break - else: + heard = await listen( + dev.capture_rf(window=args.window, frequency=frequency, stop_after_first=not args.keep) + ) + if not heard: print("No data received...") - exit(1) - - print("Packet found!") - raw_fmt = data.hex() - base64_fmt = base64.b64encode(data).decode('ascii') - pulse_fmt = format_pulses(data_to_pulses(data)) - - print("Raw:", raw_fmt) - print("Base64:", base64_fmt) - print("Pulses:", pulse_fmt) - - if args.learnfile: - print("Saving to {}".format(args.learnfile)) - with open(args.learnfile, "w") as text_file: - text_file.write(pulse_fmt if args.durations else raw_fmt) + sys.exit(1) if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 52abb677..76f1a2f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-broadlink" -version = "1.0.0.dev0" +version = "1.0.0" 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 14ed8c50..c4e4ce33 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -134,7 +134,12 @@ async def press_later(fake: FakeRM, packet: bytes, delay: float) -> bool: return fake.press(packet) -FAST = dict(poll_interval=0.01, rearm_interval=10.0) +# Every delay below is a multiple of this unit. The windows and presses are +# tens of milliseconds apart, which is plenty on a laptop but tight on a +# loaded CI runner, so the unit is deliberately generous. +UNIT = 0.03 + +FAST = dict(poll_interval=UNIT, rearm_interval=10.0) # ------------------------------------------------------------- IR windows @@ -146,7 +151,7 @@ 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)) + asyncio.get_running_loop().create_task(press_later(fake, IR, 3 * UNIT)) signals = [s async for s in device.capture(window=2, **FAST)] return signals @@ -166,7 +171,7 @@ async def go(): def test_capture_window_elapses_with_nothing(): device, fake = make() - signals = run(_collect(device.capture(window=0.05, **FAST))) + signals = run(_collect(device.capture(window=5 * UNIT, **FAST))) assert signals == [] assert fake.count(CMD_LEARN) == 1 assert fake.count(CMD_CHECK) >= 3 @@ -182,9 +187,9 @@ def test_capture_keeps_going_and_rearms_after_each_code(): 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)] + 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)] signals = run(go()) assert [s.packet for s in signals] == [IR, RF] @@ -202,14 +207,14 @@ async def go(): results = [] async def presses(): - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) results.append(fake.press(IR)) results.append(fake.press(RF)) # Device not armed: lost. - await asyncio.sleep(0.03) + await asyncio.sleep(3 * UNIT) 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)] + signals = [s async for s in device.capture(window=10 * UNIT, stop_after_first=False, **FAST)] return results, signals results, signals = run(go()) @@ -222,15 +227,15 @@ def test_send_during_window_rearms(): async def go(): async def send_then_press(): - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) await device.send_data(IR) fake.expire() # Whatever the send did to the session, assume the worst. - await asyncio.sleep(0.03) + await asyncio.sleep(3 * UNIT) 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)] + signals = [s async for s in device.capture(window=20 * UNIT, **FAST)] return await task, signals pressed, signals = run(go()) @@ -248,16 +253,16 @@ def test_timed_rearm_recovers_from_silent_expiry(): async def go(): async def expire_then_press(): - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) fake.expire() assert fake.press(IR) is False # Lost: the device is deaf. - await asyncio.sleep(0.05) # Past the re-arm interval. + await asyncio.sleep(5 * UNIT) # 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) + s async for s in device.capture(window=30 * UNIT, poll_interval=1 * UNIT, rearm_interval=4 * UNIT) ] return await task, signals @@ -273,7 +278,7 @@ 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: - asyncio.get_running_loop().create_task(press_later(fake, IR, 0.02)) + asyncio.get_running_loop().create_task(press_later(fake, IR, 2 * UNIT)) async for s in gen: got.append(s) if len(got) == 1: @@ -294,7 +299,7 @@ async def go(): task = asyncio.get_running_loop().create_task( _collect(device.capture(window=1, **FAST)) ) - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) with pytest.raises(e.CaptureInProgressError): await _collect(device.capture(window=1, **FAST)) assert device._capture_open is True @@ -311,7 +316,7 @@ def test_transport_timeouts_rearm_then_give_up(): fake.timeouts_to_raise = 2 async def go(): - asyncio.get_running_loop().create_task(press_later(fake, IR, 0.05)) + asyncio.get_running_loop().create_task(press_later(fake, IR, 5 * UNIT)) return [s async for s in device.capture(window=1, **FAST)] signals = run(go()) @@ -342,7 +347,7 @@ 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)) + asyncio.get_running_loop().create_task(press_later(fake, RF, 3 * UNIT)) return [s async for s in device.capture_rf(window=1, frequency=433.92, **FAST)] signals = run(go()) @@ -363,7 +368,7 @@ def test_capture_rf_yields_despite_odd_type_byte(): odd = bytes([0xB1]) + RF[1:] async def go(): - asyncio.get_running_loop().create_task(press_later(fake, odd, 0.03)) + asyncio.get_running_loop().create_task(press_later(fake, odd, 3 * UNIT)) return [s async for s in device.capture_rf(window=1, frequency=433.92, **FAST)] signals = run(go()) @@ -377,7 +382,7 @@ def test_capture_rf_below_400mhz_is_tagged_315(): device, fake = make() async def go(): - asyncio.get_running_loop().create_task(press_later(fake, RF, 0.03)) + asyncio.get_running_loop().create_task(press_later(fake, RF, 3 * UNIT)) return [s async for s in device.capture_rf(window=1, frequency=315.0, **FAST)] signals = run(go()) @@ -389,7 +394,7 @@ def test_capture_rf_sweeps_then_learns(): fake.sweep_answers = [(False, 0.0), (False, 0.0), (True, 433.92)] async def go(): - asyncio.get_running_loop().create_task(press_later(fake, RF, 0.08)) + asyncio.get_running_loop().create_task(press_later(fake, RF, 8 * UNIT)) return [s async for s in device.capture_rf(window=1, **FAST)] signals = run(go()) @@ -406,7 +411,7 @@ async def go(): def test_capture_rf_sweep_that_never_locks_is_cancelled(): device, fake = make() - signals = run(_collect(device.capture_rf(window=0.05, **FAST))) + signals = run(_collect(device.capture_rf(window=5 * UNIT, **FAST))) assert signals == [] assert fake.count(CMD_SWEEP) == 1 assert fake.count(CMD_CANCEL_SWEEP) == 1 @@ -421,12 +426,12 @@ def test_send_during_sweep_restarts_it(): async def go(): async def send(): - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) await device.send_data(IR) loop = asyncio.get_running_loop() loop.create_task(send()) - loop.create_task(press_later(fake, RF, 0.2)) + loop.create_task(press_later(fake, RF, 20 * UNIT)) return [s async for s in device.capture_rf(window=1, **FAST)] signals = run(go()) @@ -442,7 +447,7 @@ async def go(): task = asyncio.get_running_loop().create_task( _collect(device.capture(window=1, **FAST)) ) - await asyncio.sleep(0.02) + await asyncio.sleep(2 * UNIT) with pytest.raises(e.CaptureInProgressError): await _collect(device.capture_rf(window=1, frequency=433.92, **FAST)) with pytest.raises(e.CaptureInProgressError):