From db070d1f3bcd554a9578b906b60a1acbbeecbcf4 Mon Sep 17 00:00:00 2001 From: DAB-LABS <128871138+DAB-LABS@users.noreply.github.com> Date: Fri, 4 Sep 2026 13:13:27 -0700 Subject: [PATCH] Make the library asynchronous Every method that reaches a device is now a coroutine, with the same names, arguments and return values as before. Discovery, hello and setup are coroutines and xdiscover is an async generator. The packet, CRC and datetime helpers stay synchronous. There is no synchronous compatibility layer. Transport: each device keeps one UDP endpoint (asyncio DatagramProtocol) for its lifetime and serializes requests on it with an asyncio.Lock; the previous code opened a socket per call and declared a lock it never acquired. Retry and timeout behaviour is unchanged. An expired session key is re-authenticated once and the request repeated. async with / aclose() release the endpoint. Device classes are a mechanical port (async def and await); the oracle suite recorded in the previous change passes unchanged, so every method sends the same bytes and decodes the same results as 0.19.0. Transport tests use a fake endpoint and gain cases for lock serialization, endpoint reuse, stale-reply draining and re-auth. The CLI runs under asyncio.run. README and CHANGELOG describe the break. Live-checked against an RM4 Pro: discovery, hello, auth, sensors, concurrent calls, learning primitives, send, and the timeout path. --- CHANGELOG.md | 20 +++ README.md | 106 ++++++++----- broadlink/__init__.py | 74 +++++---- broadlink/alarm.py | 4 +- broadlink/climate.py | 64 ++++---- broadlink/cover.py | 88 +++++------ broadlink/device.py | 312 ++++++++++++++++++++++++++----------- broadlink/hub.py | 12 +- broadlink/light.py | 16 +- broadlink/remote.py | 60 ++++---- broadlink/sensor.py | 16 +- broadlink/switch.py | 100 ++++++------ cli/broadlink_cli | 309 +++++++++++++++++++------------------ cli/broadlink_discovery | 41 +++-- tests/test_oracle.py | 2 +- tests/test_transport.py | 334 +++++++++++++++++++++++++++------------- 16 files changed, 941 insertions(+), 617 deletions(-) mode change 100755 => 100644 broadlink/climate.py mode change 100755 => 100644 cli/broadlink_cli mode change 100755 => 100644 cli/broadlink_discovery diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ef4e048..9b44e43a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,26 @@ history below starts at that fork point. ### Changed +- **The library is asynchronous.** Every method that talks to a device is + now a coroutine: `await device.auth()`, `await device.send_data(...)`, + `await device.check_sensors()`, and so on. Discovery is + `await broadlink.discover(...)`, `broadlink.hello(...)` and `setup(...)` + are coroutines, and `xdiscover(...)` is an async generator. The packet + helpers (`pulses_to_data`, `data_to_pulses`), CRC and datetime helpers + stay synchronous. There is no synchronous compatibility layer: a call + without `await` returns a coroutine and does nothing. +- Each device keeps one UDP endpoint for its lifetime (the previous + version opened a socket per call) and serializes requests on it with an + `asyncio.Lock`. The old code declared a lock but never acquired it. + `async with device:` or `await device.aclose()` releases the endpoint; + it reopens on the next call. +- When a device reports that the session key has expired, the library + re-authenticates once and repeats the request. Callers no longer need + their own re-auth loop. +- 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`. - 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 34a82386..a8babf31 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,30 @@ A Python module and CLI for controlling Broadlink devices locally. > RM Max and RM5 Plus. Version 1.0 will be asynchronous; see `CHANGELOG.md`. > Upstream's credit and MIT license are preserved. +## Version 1.0 is asynchronous + +Every call that reaches a device is a coroutine and must be awaited. This +is the whole change from the original library's API; method names, +arguments and return values are the same. + +```python +import asyncio +import broadlink + +async def main(): + devices = await broadlink.discover(timeout=5) + device = devices[0] + await device.auth() + print(await device.check_sensors()) + +asyncio.run(main()) +``` + +Calling a device method without `await` returns a coroutine object and +sends nothing; Python prints a `RuntimeWarning: coroutine ... was never +awaited` when it is garbage collected. If you need the old synchronous +behaviour, pin the original distribution (`broadlink==0.19.0`) instead. + The following devices are supported: - **Universal remotes**: RM home, RM mini 3, RM plus, RM pro, RM pro+, RM4 mini, RM4 pro, RM4C mini, RM4S, RM4 TV mate @@ -42,11 +66,11 @@ environment, remove it first (`pip3 uninstall broadlink`); both provide the ## Basic functions -First, open Python 3 and import this module. +The examples below are written as they would appear inside an `async def` +function run with `asyncio.run(...)`, as in the snippet above. To try them +interactively, start Python with `python3 -m asyncio`, which gives you a +prompt where `await` works at the top level. -``` -python3 -``` ```python3 import broadlink ``` @@ -63,7 +87,7 @@ In order to control the device, you need to connect it to your local network. If - Manually connect to the WiFi SSID named BroadlinkProv. 2. Connect the device to your local network with the setup function. ```python3 -broadlink.setup('myssid', 'mynetworkpass', 3) +await broadlink.setup('myssid', 'mynetworkpass', 3) ``` Security mode options are (0 = none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2) @@ -72,7 +96,7 @@ Security mode options are (0 = none, 1 = WEP, 2 = WPA1, 3 = WPA2, 4 = WPA1/2) You may need to specify a broadcast address if setup is not working. ```python3 -broadlink.setup('myssid', 'mynetworkpass', 3, ip_address='192.168.0.255') +await broadlink.setup('myssid', 'mynetworkpass', 3, ip_address='192.168.0.255') ``` ### Discovery @@ -80,7 +104,7 @@ broadlink.setup('myssid', 'mynetworkpass', 3, ip_address='192.168.0.255') Use this function to discover devices: ```python3 -devices = broadlink.discover() +devices = await broadlink.discover() ``` #### Advanced options @@ -88,29 +112,29 @@ You may need to specify `local_ip_address` or `discover_ip_address` if discovery Using the IP address of your local machine: ```python3 -devices = broadlink.discover(local_ip_address='192.168.0.100') +devices = await broadlink.discover(local_ip_address='192.168.0.100') ``` Using the broadcast address of your subnet: ```python3 -devices = broadlink.discover(discover_ip_address='192.168.0.255') +devices = await broadlink.discover(discover_ip_address='192.168.0.255') ``` If the device is locked, it may not be discoverable with broadcast. In such cases, you can use the unicast version `broadlink.hello()` for direct discovery: ```python3 -device = broadlink.hello('192.168.0.16') +device = await broadlink.hello('192.168.0.16') ``` If you are a perfomance freak, use `broadlink.xdiscover()` to create devices instantly: ```python3 -for device in broadlink.xdiscover(): +async for device in broadlink.xdiscover(): print(device) # Example action. Do whatever you want here. ``` ### Authentication After discovering the device, call the `auth()` method to obtain the authentication key required for further communication: ```python3 -device.auth() +await device.auth() ``` The next steps depend on the type of device you want to control. @@ -123,12 +147,12 @@ Learning IR codes takes place in three steps. 1. Enter learning mode: ```python3 -device.enter_learning() +await device.enter_learning() ``` 2. When the LED blinks, point the remote at the Broadlink device and press the button you want to learn. 3. Get the IR packet. ```python3 -packet = device.check_data() +packet = await device.check_data() ``` ### Learning RF codes @@ -137,7 +161,7 @@ Learning RF codes takes place in six steps. 1. Sweep the frequency: ```python3 -device.sweep_frequency() +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: @@ -148,12 +172,12 @@ if ok: ``` 4. Enter learning mode: ```python3 -device.find_rf_packet() +await device.find_rf_packet() ``` 5. When the LED blinks, point the remote at the Broadlink device for the second time and short press the button you want to learn. 6. Get the RF packet: ```python3 -packet = device.check_data() +packet = await device.check_data() ``` #### Notes @@ -164,25 +188,25 @@ Universal remotes with product id 0x2712 use the same method for learning IR and You can exit the learning mode in the middle of the process by calling this method: ```python3 -device.cancel_sweep_frequency() +await device.cancel_sweep_frequency() ``` ### Sending IR/RF packets ```python3 -device.send_data(packet) +await device.send_data(packet) ``` ### Fetching sensor data ```python3 -data = device.check_sensors() +data = await device.check_sensors() ``` ## Switches ### Setting power state ```python3 -device.set_power(True) -device.set_power(False) +await device.set_power(True) +await device.set_power(False) ``` ### Checking power state @@ -199,8 +223,8 @@ state = device.get_energy() ### Setting power state ```python3 -device.set_power(1, True) # Example socket. It could be 2 or 3. -device.set_power(1, False) +await device.set_power(1, True) # Example socket. It could be 2 or 3. +await device.set_power(1, False) ``` ### Checking power state @@ -217,35 +241,35 @@ state = device.get_state() ### Setting state attributes ```python3 -devices[0].set_state(pwr=0) -devices[0].set_state(pwr=1) -devices[0].set_state(brightness=75) -devices[0].set_state(bulb_colormode=0) -devices[0].set_state(blue=255) -devices[0].set_state(red=0) -devices[0].set_state(green=128) -devices[0].set_state(bulb_colormode=1) +await devices[0].set_state(pwr=0) +await devices[0].set_state(pwr=1) +await devices[0].set_state(brightness=75) +await devices[0].set_state(bulb_colormode=0) +await devices[0].set_state(blue=255) +await devices[0].set_state(red=0) +await devices[0].set_state(green=128) +await devices[0].set_state(bulb_colormode=1) ``` ## Environment sensors ### Fetching sensor data ```python3 -data = device.check_sensors() +data = await device.check_sensors() ``` ## Hubs ### Discovering subdevices ```python3 -device.get_subdevices() +await device.get_subdevices() ``` ### Fetching data Use the DID obtained from get_subdevices() for the input parameter to query specific sub-device. ```python3 -device.get_state(did="00000000000000000000a043b0d06963") +await device.get_state(did="00000000000000000000a043b0d06963") ``` ### Setting state attributes @@ -253,13 +277,13 @@ The parameters depend on the type of subdevice that is being controlled. In this #### Turn on ```python3 -device.set_state(did="00000000000000000000a043b0d0783a", pwr=1) -device.set_state(did="00000000000000000000a043b0d0783a", pwr1=1) -device.set_state(did="00000000000000000000a043b0d0783a", pwr2=1) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr=1) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr1=1) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr2=1) ``` #### Turn off ```python3 -device.set_state(did="00000000000000000000a043b0d0783a", pwr=0) -device.set_state(did="00000000000000000000a043b0d0783a", pwr1=0) -device.set_state(did="00000000000000000000a043b0d0783a", pwr2=0) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr=0) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr1=0) +await device.set_state(did="00000000000000000000a043b0d0783a", pwr2=0) ``` diff --git a/broadlink/__init__.py b/broadlink/__init__.py index b2fa3d9a..bd77fa2a 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -1,14 +1,14 @@ #!/usr/bin/env python3 """The python-broadlink library.""" -import socket -from typing import Generator, List, Optional, Tuple, Union +from collections.abc import AsyncIterator +from typing import List, Optional, Tuple, Union from . import exceptions as e from .alarm import S1C from .climate import hvac, hysen from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_TIMEOUT from .cover import dooya, dooya2, wser -from .device import Device, ping, scan +from .device import Device, _open_endpoint, ping, scan from .hub import s3 from .light import lb1, lb2 from .remote import rm, rm4, rm4mini, rm4pro, rmmini, rmminib, rmpro @@ -238,64 +238,62 @@ def gendevice( return Device(host, mac, dev_type, name=name, is_locked=is_locked) -def hello( +async def hello( ip_address: str, port: int = DEFAULT_PORT, - timeout: int = DEFAULT_TIMEOUT, + timeout: float = DEFAULT_TIMEOUT, ) -> Device: """Direct device discovery. Useful if the device is locked. """ - try: - return next( - xdiscover( - timeout=timeout, - discover_ip_address=ip_address, - discover_ip_port=port, - ) - ) - except StopIteration as err: - raise e.NetworkTimeoutError( - -4000, - "Network timeout", - f"No response received within {timeout}s", - ) from err + async for device in xdiscover( + timeout=timeout, + discover_ip_address=ip_address, + discover_ip_port=port, + ): + return device + raise e.NetworkTimeoutError( + -4000, + "Network timeout", + f"No response received within {timeout}s", + ) -def discover( - timeout: int = DEFAULT_TIMEOUT, +async def discover( + timeout: float = DEFAULT_TIMEOUT, local_ip_address: Optional[str] = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, ) -> List[Device]: """Discover devices connected to the local network.""" - responses = scan( - timeout, local_ip_address, discover_ip_address, discover_ip_port - ) - return [gendevice(*resp) for resp in responses] + return [ + device + async for device in xdiscover( + timeout, local_ip_address, discover_ip_address, discover_ip_port + ) + ] -def xdiscover( - timeout: int = DEFAULT_TIMEOUT, +async def xdiscover( + timeout: float = DEFAULT_TIMEOUT, local_ip_address: Optional[str] = None, discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_port: int = DEFAULT_PORT, -) -> Generator[Device, None, None]: +) -> AsyncIterator[Device]: """Discover devices connected to the local network. - This function returns a generator that yields devices instantly. + Yields each device as soon as it answers. """ - responses = scan( + async for resp in scan( timeout, local_ip_address, discover_ip_address, discover_ip_port - ) - for resp in responses: + ): yield gendevice(*resp) # Setup a new Broadlink device via AP Mode. Review the README to see how to enter AP Mode. # Only tested with Broadlink RM3 Mini (Blackbean) -def setup( +async def setup( ssid: str, password: str, security_mode: int, @@ -326,8 +324,8 @@ def setup( payload[0x20] = checksum & 0xFF # Checksum 1 position payload[0x21] = checksum >> 8 # Checksum 2 position - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Internet # UDP - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - sock.sendto(payload, (ip_address, DEFAULT_PORT)) - sock.close() + transport, _ = await _open_endpoint(broadcast=True) + try: + transport.sendto(payload, (ip_address, DEFAULT_PORT)) + finally: + transport.close() diff --git a/broadlink/alarm.py b/broadlink/alarm.py index a9b5e879..2c3358de 100644 --- a/broadlink/alarm.py +++ b/broadlink/alarm.py @@ -14,11 +14,11 @@ class S1C(Device): 0x21: "Motion Sensor", } - def get_sensors_status(self) -> dict: + async def get_sensors_status(self) -> dict: """Return the state of the sensors.""" packet = bytearray(16) packet[0] = 0x06 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) count = payload[0x4] diff --git a/broadlink/climate.py b/broadlink/climate.py old mode 100755 new mode 100644 index 1a0c6006..5d75457d --- a/broadlink/climate.py +++ b/broadlink/climate.py @@ -21,14 +21,14 @@ class hysen(Device): TYPE = "HYS" - def send_request(self, request: Sequence[int]) -> bytes: + async def send_request(self, request: Sequence[int]) -> bytes: """Send a request to the device.""" packet = bytearray() packet.extend((len(request) + 2).to_bytes(2, "little")) packet.extend(request) packet.extend(CRC16.calculate(request).to_bytes(2, "little")) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) @@ -52,22 +52,22 @@ def _decode_temp(self, payload, base_index): offset = (offset_raw_value + 1) / 10 if add_offset else 0.0 return base_temp + offset - def get_temp(self) -> float: + async def get_temp(self) -> float: """Return the room temperature in degrees celsius.""" - payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08]) + payload = await self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08]) return self._decode_temp(payload, 5) - def get_external_temp(self) -> float: + async def get_external_temp(self) -> float: """Return the external temperature in degrees celsius.""" - payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08]) + payload = await self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x08]) return self._decode_temp(payload, 18) - def get_full_status(self) -> dict: + async def get_full_status(self) -> dict: """Return the state of the device. Timer schedule included. """ - payload = self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x16]) + payload = await self.send_request([0x01, 0x03, 0x00, 0x00, 0x00, 0x16]) data = {} data["remote_lock"] = payload[3] & 1 data["power"] = payload[4] & 1 @@ -127,12 +127,12 @@ def get_full_status(self) -> dict: # E.g. loop_mode = 0 ("12345,67") means Saturday and Sunday (weekend schedule) # loop_mode = 2 ("1234567") means every day, including Saturday and Sunday (weekday schedule) # The sensor command is currently experimental - def set_mode( + async def set_mode( self, auto_mode: int, loop_mode: int, sensor: int = 0 ) -> None: """Set the mode of the device.""" mode_byte = ((loop_mode + 1) << 4) + auto_mode - self.send_request([0x01, 0x06, 0x00, 0x02, mode_byte, sensor]) + await self.send_request([0x01, 0x06, 0x00, 0x02, mode_byte, sensor]) # Advanced settings # Sensor mode (SEN) sensor = 0 for internal sensor, 1 for external sensor, @@ -145,7 +145,7 @@ def set_mode( # Anti-freezing function (FrE) fre = 0 for anti-freezing function shut down, # 1 for anti-freezing function open. Factory default: 0 # Power on memory (POn) poweron = 0 for off, 1 for on. Default: 0 - def set_advanced( + async def set_advanced( self, loop_mode: int, sensor: int, @@ -158,7 +158,7 @@ def set_advanced( poweron: int, ) -> None: """Set advanced options.""" - self.send_request( + await self.send_request( [ 0x01, 0x10, @@ -182,34 +182,34 @@ def set_advanced( # For backwards compatibility only. Prefer calling set_mode directly. # Note this function invokes loop_mode=0 and sensor=0. - def switch_to_auto(self) -> None: + async def switch_to_auto(self) -> None: """Switch mode to auto.""" - self.set_mode(auto_mode=1, loop_mode=0) + await self.set_mode(auto_mode=1, loop_mode=0) - def switch_to_manual(self) -> None: + async def switch_to_manual(self) -> None: """Switch mode to manual.""" - self.set_mode(auto_mode=0, loop_mode=0) + await self.set_mode(auto_mode=0, loop_mode=0) # Set temperature for manual mode (also activates manual mode if currently in automatic) - def set_temp(self, temp: float) -> None: + async def set_temp(self, temp: float) -> None: """Set the target temperature.""" - self.send_request([0x01, 0x06, 0x00, 0x01, 0x00, int(temp * 2)]) + await self.send_request([0x01, 0x06, 0x00, 0x01, 0x00, int(temp * 2)]) # Set device on(1) or off(0), does not deactivate Wifi connectivity. # Remote lock disables control by buttons on thermostat. # heating_cooling: heating(0) cooling(1) - def set_power( + async def set_power( self, power: int = 1, remote_lock: int = 0, heating_cooling: int = 0 ) -> None: """Set the power state of the device.""" state = (heating_cooling << 7) + power - self.send_request([0x01, 0x06, 0x00, 0x00, remote_lock, state]) + await self.send_request([0x01, 0x06, 0x00, 0x00, remote_lock, state]) # set time on device # n.b. day=1 is Monday, ..., day=7 is Sunday - def set_time(self, hour: int, minute: int, second: int, day: int) -> None: + async def set_time(self, hour: int, minute: int, second: int, day: int) -> None: """Set the time.""" - self.send_request( + await self.send_request( [ 0x01, 0x10, @@ -231,7 +231,7 @@ def set_time(self, hour: int, minute: int, second: int, day: int) -> None: # {'start_hour':17, 'start_minute':30, 'temp': 22 } # Each one specifies the thermostat temp that will become effective at start_hour:start_minute # weekend is similar but only has 2 (e.g. switch on in morning and off in afternoon) - def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None: + async def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None: """Set timer schedule.""" request = [0x01, 0x10, 0x00, 0x0A, 0x00, 0x0C, 0x18] @@ -253,7 +253,7 @@ def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None: for i in range(0, 2): request.append(int(weekend[i]["temp"] * 2)) - self.send_request(request) + await self.send_request(request) class hvac(Device): @@ -343,11 +343,11 @@ def _decode(self, response: bytes) -> bytes: d_len = int.from_bytes(payload[0x08:0x0A], "little") return payload[0x0A:0x0A+d_len] - def _send(self, command: int, data: bytes = b"") -> bytes: + async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a command to the unit.""" prefix = bytes([((command << 4) | 1), 1]) packet = self._encode(prefix + data) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response)[0x02:] @@ -369,7 +369,7 @@ def _parse_state(self, data: bytes) -> dict: state["mildew"] = bool(data[0x0A] & 1 << 3) return state - def set_state( + async def set_state( self, power: bool, target_temp: float, # 16<=target_temp<=32 @@ -414,10 +414,10 @@ def set_state( data[0x0A] = display << 4 | mildew << 3 data[0x0C] = UNK2 - resp = self._send(0, data) + resp = await self._send(0, data) return self._parse_state(resp) - def get_state(self) -> dict: + async def get_state(self) -> dict: """Returns a dictionary with the unit's parameters. Returns: @@ -436,7 +436,7 @@ def get_state(self) -> dict: clean (bool): mildew (bool): """ - resp = self._send(1) + resp = await self._send(1) if len(resp) < 13: raise e.DataValidationError( @@ -447,7 +447,7 @@ def get_state(self) -> dict: return self._parse_state(resp) - def get_ac_info(self) -> dict: + async def get_ac_info(self) -> dict: """Returns dictionary with AC info. Returns: @@ -455,7 +455,7 @@ def get_ac_info(self) -> dict: power (bool): power ambient_temp (float): ambient temperature """ - resp = self._send(2) + resp = await self._send(2) if len(resp) < 22: raise e.DataValidationError( diff --git a/broadlink/cover.py b/broadlink/cover.py index 75317943..0319457a 100644 --- a/broadlink/cover.py +++ b/broadlink/cover.py @@ -1,5 +1,5 @@ """Support for covers.""" -import time +import asyncio from typing import Sequence from . import exceptions as e @@ -11,7 +11,7 @@ class dooya(Device): TYPE = "DT360E" - def _send(self, command: int, attribute: int = 0) -> int: + async def _send(self, command: int, attribute: int = 0) -> int: """Send a packet to the device.""" packet = bytearray(16) packet[0x00] = 0x09 @@ -21,42 +21,42 @@ def _send(self, command: int, attribute: int = 0) -> int: packet[0x09] = 0xFA packet[0x0A] = 0x44 - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) payload = self.decrypt(resp[0x38:]) return payload[4] - def open(self) -> int: + async def open(self) -> int: """Open the curtain.""" - return self._send(0x01) + return await self._send(0x01) - def close(self) -> int: + async def close(self) -> int: """Close the curtain.""" - return self._send(0x02) + return await self._send(0x02) - def stop(self) -> int: + async def stop(self) -> int: """Stop the curtain.""" - return self._send(0x03) + return await self._send(0x03) - def get_percentage(self) -> int: + async def get_percentage(self) -> int: """Return the position of the curtain.""" - return self._send(0x06, 0x5D) + return await self._send(0x06, 0x5D) - def set_percentage_and_wait(self, new_percentage: int) -> None: + async def set_percentage_and_wait(self, new_percentage: int) -> None: """Set the position of the curtain.""" - current = self.get_percentage() + current = await self.get_percentage() if current > new_percentage: - self.close() + await self.close() while current is not None and current > new_percentage: - time.sleep(0.2) - current = self.get_percentage() + await asyncio.sleep(0.2) + current = await self.get_percentage() elif current < new_percentage: - self.open() + await self.open() while current is not None and current < new_percentage: - time.sleep(0.2) - current = self.get_percentage() - self.stop() + await asyncio.sleep(0.2) + current = await self.get_percentage() + await self.stop() class dooya2(Device): @@ -64,7 +64,7 @@ class dooya2(Device): TYPE = "DT360E-2" - def _send(self, operation: int, data: Sequence = b""): + async def _send(self, operation: int, data: Sequence = b""): """Send a command to the device.""" packet = bytearray(12) packet[0x02] = 0xA5 @@ -89,31 +89,31 @@ def _send(self, operation: int, data: Sequence = b""): packet[0x00] = packet_len & 0xFF packet[0x01] = packet_len >> 8 - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) payload = self.decrypt(resp[0x38:]) return payload - def open(self) -> None: + async def open(self) -> None: """Open the curtain.""" - self._send(2, [0x00, 0x01, 0x00]) + await self._send(2, [0x00, 0x01, 0x00]) - def close(self) -> None: + async def close(self) -> None: """Close the curtain.""" - self._send(2, [0x00, 0x02, 0x00]) + await self._send(2, [0x00, 0x02, 0x00]) - def stop(self) -> None: + async def stop(self) -> None: """Stop the curtain.""" - self._send(2, [0x00, 0x03, 0x00]) + await self._send(2, [0x00, 0x03, 0x00]) - def get_percentage(self) -> int: + async def get_percentage(self) -> int: """Return the position of the curtain.""" - resp = self._send(1, [0x00, 0x06, 0x00]) + resp = await self._send(1, [0x00, 0x06, 0x00]) return resp[0x11] - def set_percentage(self, new_percentage: int) -> None: + async def set_percentage(self, new_percentage: int) -> None: """Set the position of the curtain.""" - self._send(2, [0x00, 0x09, new_percentage]) + await self._send(2, [0x00, 0x09, new_percentage]) class wser(Device): @@ -121,7 +121,7 @@ class wser(Device): TYPE = "WSER" - def _send(self, operation: int, data: Sequence = b""): + async def _send(self, operation: int, data: Sequence = b""): """Send a command to the device.""" packet = bytearray(12) packet[0x02] = 0xA5 @@ -146,37 +146,37 @@ def _send(self, operation: int, data: Sequence = b""): packet[0x00] = packet_len & 0xFF packet[0x01] = packet_len >> 8 - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) payload = self.decrypt(resp[0x38:]) return payload - def get_position(self) -> int: + async def get_position(self) -> int: """Return the position of the curtain.""" - resp = self._send(1, []) + resp = await self._send(1, []) position = resp[0x0E] return position - def open(self) -> int: + async def open(self) -> int: """Open the curtain.""" - resp = self._send(2, [0x4A, 0x31, 0xA0]) + resp = await self._send(2, [0x4A, 0x31, 0xA0]) position = resp[0x0E] return position - def close(self) -> int: + async def close(self) -> int: """Close the curtain.""" - resp = self._send(2, [0x61, 0x32, 0xA0]) + resp = await self._send(2, [0x61, 0x32, 0xA0]) position = resp[0x0E] return position - def stop(self) -> int: + async def stop(self) -> int: """Stop the curtain.""" - resp = self._send(2, [0x4C, 0x73, 0xA0]) + resp = await self._send(2, [0x4C, 0x73, 0xA0]) position = resp[0x0E] return position - def set_position(self, position: int) -> int: + async def set_position(self, position: int) -> int: """Set the position of the curtain.""" - resp = self._send(2, [position, 0x70, 0xA0]) + resp = await self._send(2, [position, 0x70, 0xA0]) position = resp[0x0E] return position diff --git a/broadlink/device.py b/broadlink/device.py index 22c3ebed..2dc95e0d 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -1,9 +1,19 @@ -"""Support for Broadlink devices.""" +"""Support for Broadlink devices. + +Transport layer. Every device method ends up in :meth:`Device.send_packet`, +which frames, encrypts and sends one request over UDP and waits for the one +reply. The protocol is strictly request and reply and the device never +speaks unprompted, so each device keeps a single datagram endpoint and an +``asyncio.Lock`` that serializes calls on it. +""" + +from __future__ import annotations + +import asyncio import random import socket -import threading -import time -from typing import Generator, Optional, Tuple, Union +from collections.abc import AsyncIterator +from typing import Optional, Tuple, Union from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes @@ -17,77 +27,141 @@ ) from .protocol import Datetime -HelloResponse = Tuple[int, Tuple[str, int], str, str, bool] +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} -def scan( - timeout: int = DEFAULT_TIMEOUT, - local_ip_address: Optional[str] = None, - discover_ip_address: str = DEFAULT_BCAST_ADDR, - discover_ip_port: int = DEFAULT_PORT, -) -> Generator[HelloResponse, None, None]: - """Broadcast a hello message and yield responses.""" - conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - conn.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - - if local_ip_address: - conn.bind((local_ip_address, 0)) - port = conn.getsockname()[1] - else: - local_ip_address = "0.0.0.0" - port = 0 +class _Protocol(asyncio.DatagramProtocol): + """Datagram protocol that hands every received packet to a queue.""" + + def __init__(self) -> None: + self.queue: asyncio.Queue[tuple[bytes, tuple[str, int]]] = asyncio.Queue() + self.transport: Optional[asyncio.DatagramTransport] = None + self.closed = asyncio.get_running_loop().create_future() + + def connection_made(self, transport) -> None: # type: ignore[override] + self.transport = transport + + def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: + self.queue.put_nowait((data, addr)) + + def error_received(self, exc: Exception) -> None: + # ICMP unreachable and the like. Surface it as a receive of nothing; + # the retry loop will time out and raise NetworkTimeoutError. + pass + + def connection_lost(self, exc: Optional[Exception]) -> None: + if not self.closed.done(): + self.closed.set_result(None) + + def drain(self) -> None: + """Drop anything that arrived before the current request.""" + while not self.queue.empty(): + self.queue.get_nowait() + + +async def _open_endpoint( + local_addr: Optional[tuple[str, int]] = None, + remote_addr: Optional[tuple[str, int]] = None, + broadcast: bool = False, +) -> tuple[asyncio.DatagramTransport, _Protocol]: + """Create a UDP endpoint. Tests replace this to fake the network.""" + loop = asyncio.get_running_loop() + transport, protocol = await loop.create_datagram_endpoint( + _Protocol, + local_addr=local_addr, + remote_addr=remote_addr, + family=socket.AF_INET, + allow_broadcast=broadcast, + ) + return transport, protocol # type: ignore[return-value] + + +def _hello_packet(local_ip_address: str, port: int) -> bytearray: packet = bytearray(0x30) packet[0x08:0x14] = Datetime.pack(Datetime.now()) packet[0x18:0x1C] = socket.inet_aton(local_ip_address)[::-1] packet[0x1C:0x1E] = port.to_bytes(2, "little") packet[0x26] = 6 - checksum = sum(packet, 0xBEAF) & 0xFFFF packet[0x20:0x22] = checksum.to_bytes(2, "little") + return packet - start_time = time.time() - discovered = [] - try: - while (time.time() - start_time) < timeout: - time_left = timeout - (time.time() - start_time) - conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left)) - conn.sendto(packet, (discover_ip_address, discover_ip_port)) +def _parse_hello(resp: bytes, host: tuple[str, int]) -> HelloResponse: + devtype = resp[0x34] | resp[0x35] << 8 + mac = resp[0x3A:0x40][::-1] + name = resp[0x40:].split(b"\x00")[0].decode() + is_locked = bool(resp[0x7F]) + return devtype, host, mac, name, is_locked + +async def scan( + timeout: float = DEFAULT_TIMEOUT, + local_ip_address: Optional[str] = None, + discover_ip_address: str = DEFAULT_BCAST_ADDR, + discover_ip_port: int = DEFAULT_PORT, +) -> AsyncIterator[HelloResponse]: + """Broadcast a hello message and yield responses as they arrive. + + The hello is repeated every ``DEFAULT_RETRY_INTVL`` seconds until + ``timeout`` elapses. Each device is yielded once. + """ + local_addr = (local_ip_address, 0) if local_ip_address else None + transport, protocol = await _open_endpoint(local_addr=local_addr, broadcast=True) + try: + if local_ip_address: + port = transport.get_extra_info("sockname")[1] + else: + local_ip_address = "0.0.0.0" + port = 0 + packet = _hello_packet(local_ip_address, port) + + loop = asyncio.get_running_loop() + start = loop.time() + discovered: set[tuple[tuple[str, int], bytes, int]] = set() + + while (loop.time() - start) < timeout: + transport.sendto(packet, (discover_ip_address, discover_ip_port)) + deadline = min(DEFAULT_RETRY_INTVL, timeout - (loop.time() - start)) + slot_end = loop.time() + deadline while True: + remaining = slot_end - loop.time() + if remaining <= 0: + break try: - resp, host = conn.recvfrom(1024) - except socket.timeout: + resp, host = await asyncio.wait_for(protocol.queue.get(), remaining) + except asyncio.TimeoutError: break - - devtype = resp[0x34] | resp[0x35] << 8 - mac = resp[0x3A:0x40][::-1] - - if (host, mac, devtype) in discovered: + if len(resp) < 0x80: continue - discovered.append((host, mac, devtype)) - - name = resp[0x40:].split(b"\x00")[0].decode() - is_locked = bool(resp[0x7F]) - yield devtype, host, mac, name, is_locked + entry = _parse_hello(resp, host) + key = (entry[1], entry[2], entry[0]) + if key in discovered: + continue + discovered.add(key) + yield entry finally: - conn.close() + transport.close() -def ping(ip_address: str, port: int = DEFAULT_PORT) -> None: +async def ping(ip_address: str, port: int = DEFAULT_PORT) -> None: """Send a ping packet to an address. This packet feeds the watchdog timer of firmwares >= v53. Useful to prevent reboots when the cloud cannot be reached. It must be sent every 2 minutes in such cases. """ - with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn: - conn.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + transport, _ = await _open_endpoint(broadcast=True) + try: packet = bytearray(0x30) packet[0x26] = 1 - conn.sendto(packet, (ip_address, port)) + transport.sendto(packet, (ip_address, port)) + finally: + transport.close() class Device: @@ -103,7 +177,7 @@ def __init__( host: Tuple[str, int], mac: Union[bytes, str], devtype: int, - timeout: int = DEFAULT_TIMEOUT, + timeout: float = DEFAULT_TIMEOUT, name: str = "", model: str = "", manufacturer: str = "", @@ -122,11 +196,15 @@ def __init__( self.iv = bytes.fromhex(self.__INIT_VECT) self.id = 0 self.type = self.TYPE # For backwards compatibility. - self.lock = threading.Lock() self.aes = None self.update_aes(bytes.fromhex(self.__INIT_KEY)) + self._lock: Optional[asyncio.Lock] = None + self._transport: Optional[asyncio.DatagramTransport] = None + self._protocol: Optional[_Protocol] = None + self._reauth_ok = True + def __repr__(self) -> str: """Return a formal representation of the device.""" return ( @@ -154,6 +232,14 @@ def __str__(self) -> str: ":".join(format(x, "02X") for x in self.mac), ) + async def __aenter__(self) -> "Device": + return self + + async def __aexit__(self, *exc) -> None: + await self.aclose() + + # ------------------------------------------------------------ crypto + def update_aes(self, key: bytes) -> None: """Update AES.""" self.aes = Cipher( @@ -170,7 +256,9 @@ def decrypt(self, payload: bytes) -> bytes: decryptor = self.aes.decryptor() return decryptor.update(bytes(payload)) + decryptor.finalize() - def auth(self) -> bool: + # ---------------------------------------------------------- session + + async def auth(self) -> bool: """Authenticate to the device.""" self.id = 0 self.update_aes(bytes.fromhex(self.__INIT_KEY)) @@ -181,7 +269,7 @@ def auth(self) -> bool: packet[0x2D] = 0x01 packet[0x30:0x36] = "Test 1".encode() - response = self.send_packet(0x65, packet) + response = await self.send_packet(0x65, packet, _reauth=False) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) @@ -189,7 +277,7 @@ def auth(self) -> bool: self.update_aes(payload[0x04:0x14]) return True - def hello(self, local_ip_address=None) -> bool: + async def hello(self, local_ip_address=None) -> bool: """Send a hello message to the device. Device information is checked before updating name and lock status. @@ -200,15 +288,16 @@ def hello(self, local_ip_address=None) -> bool: discover_ip_address=self.host[0], discover_ip_port=self.host[1], ) - try: - devtype, _, mac, name, is_locked = next(responses) - - except StopIteration as err: + entry = None + async for entry in responses: + break + if entry is None: raise e.NetworkTimeoutError( -4000, "Network timeout", f"No response received within {self.timeout}s", - ) from err + ) + devtype, _, mac, name, is_locked = entry if mac != self.mac: raise e.DataValidationError( @@ -230,40 +319,40 @@ def hello(self, local_ip_address=None) -> bool: self.is_locked = is_locked return True - def ping(self) -> None: + async def ping(self) -> None: """Ping the device. This packet feeds the watchdog timer of firmwares >= v53. Useful to prevent reboots when the cloud cannot be reached. It must be sent every 2 minutes in such cases. """ - ping(self.host[0], port=self.host[1]) + await ping(self.host[0], port=self.host[1]) - def get_fwversion(self) -> int: + async def get_fwversion(self) -> int: """Get firmware version.""" packet = bytearray([0x68]) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return payload[0x4] | payload[0x5] << 8 - def set_name(self, name: str) -> None: + async def set_name(self, name: str) -> None: """Set device name.""" packet = bytearray(4) packet += name.encode("utf-8") packet += bytearray(0x50 - len(packet)) packet[0x43] = self.is_locked - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) self.name = name - def set_lock(self, state: bool) -> None: + async def set_lock(self, state: bool) -> None: """Lock/unlock the device.""" packet = bytearray(4) packet += self.name.encode("utf-8") packet += bytearray(0x50 - len(packet)) packet[0x43] = bool(state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) self.is_locked = bool(state) @@ -271,8 +360,24 @@ def get_type(self) -> str: """Return device type.""" return self.type - def send_packet(self, packet_type: int, payload: bytes) -> bytes: - """Send a packet to the device.""" + # -------------------------------------------------------- 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 + + async def _endpoint(self) -> tuple[asyncio.DatagramTransport, _Protocol]: + if self._transport is None or self._transport.is_closing(): + self._transport, self._protocol = await _open_endpoint( + remote_addr=self.host + ) + return self._transport, self._protocol # type: ignore[return-value] + + def _frame(self, packet_type: int, payload: bytes) -> bytes: + """Build the wire frame for one request (advances the counter).""" self.count = ((self.count + 1) | 0x8000) & 0xFFFF packet = bytearray(0x38) packet[0x00:0x08] = bytes.fromhex("5aa5aa555aa5aa55") @@ -291,27 +396,10 @@ def send_packet(self, packet_type: int, payload: bytes) -> bytes: checksum = sum(packet, 0xBEAF) & 0xFFFF packet[0x20:0x22] = checksum.to_bytes(2, "little") + return bytes(packet) - with self.lock and socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as conn: - timeout = self.timeout - start_time = time.time() - - while True: - time_left = timeout - (time.time() - start_time) - conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left)) - conn.sendto(packet, self.host) - - try: - resp = conn.recvfrom(2048)[0] - break - except socket.timeout as err: - if (time.time() - start_time) > timeout: - raise e.NetworkTimeoutError( - -4000, - "Network timeout", - f"No response received within {timeout}s", - ) from err - + @staticmethod + def _validate(resp: bytes) -> bytes: if len(resp) < 0x30: raise e.DataValidationError( -4007, @@ -328,5 +416,55 @@ def send_packet(self, packet_type: int, payload: bytes) -> bytes: "Received data packet check error", f"Expected a checksum of {nom_checksum} and received {real_checksum}", ) + return resp + async def _exchange(self, packet: bytes) -> bytes: + """Send one frame and wait for one reply, resending on silence.""" + transport, protocol = await self._endpoint() + protocol.drain() + loop = asyncio.get_running_loop() + start = loop.time() + timeout = self.timeout + + 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) + + async def send_packet( + self, packet_type: int, payload: bytes, *, _reauth: bool = True + ) -> bytes: + """Send a packet to the device and return the raw response frame. + + If the device answers that the session key is no longer valid, the + session is re-authenticated once and the request is sent again. + """ + if self._lock is None: + self._lock = asyncio.Lock() + async with self._lock: + resp = await self._exchange(self._frame(packet_type, bytes(payload))) + + if _reauth and self._reauth_ok: + 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 return resp diff --git a/broadlink/hub.py b/broadlink/hub.py index 40dd8e2d..1d74041f 100644 --- a/broadlink/hub.py +++ b/broadlink/hub.py @@ -13,7 +13,7 @@ class s3(Device): TYPE = "S3" MAX_SUBDEVICES = 8 - def get_subdevices(self, step: int = 5) -> list: + async def get_subdevices(self, step: int = 5) -> list: """Return a list of sub devices.""" total = self.MAX_SUBDEVICES sub_devices = [] @@ -23,7 +23,7 @@ def get_subdevices(self, step: int = 5) -> list: while index < total: state = {"count": step, "index": index} packet = self._encode(14, state) - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) resp = self._decode(resp) @@ -43,18 +43,18 @@ def get_subdevices(self, step: int = 5) -> list: return sub_devices - def get_state(self, did: Optional[str] = None) -> dict: + async def get_state(self, did: Optional[str] = None) -> dict: """Return the power state of the device.""" state = {} if did is not None: state["did"] = did packet = self._encode(1, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) - def set_state( + async def set_state( self, did: Optional[str] = None, pwr1: Optional[bool] = None, @@ -73,7 +73,7 @@ def set_state( state["pwr3"] = int(bool(pwr3)) packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) diff --git a/broadlink/light.py b/broadlink/light.py index 1ae87e8f..6225887e 100644 --- a/broadlink/light.py +++ b/broadlink/light.py @@ -21,17 +21,17 @@ class ColorMode(enum.IntEnum): WHITE = 1 SCENE = 2 - def get_state(self) -> dict: + async def get_state(self) -> dict: """Return the power state of the device. Example: `{'red': 128, 'blue': 255, 'green': 128, 'pwr': 1, 'brightness': 75, 'colortemp': 2700, 'hue': 240, 'saturation': 50, 'transitionduration': 1500, 'maxworktime': 0, 'bulb_colormode': 1, 'bulb_scenes': '["@01686464,0,0,0", "#ffffff,10,0,#000000,190,0,0", "2700+100,0,0,0", "#ff0000,500,2500,#00FF00,500,2500,#0000FF,500,2500,0", "@01686464,100,2400,@01686401,100,2400,0", "@01686464,100,2400,@01686401,100,2400,@005a6464,100,2400,@005a6401,100,2400,0", "@01686464,10,0,@00000000,190,0,0", "@01686464,200,0,@005a6464,200,0,0"]', 'bulb_scene': '', 'bulb_sceneidx': 255}` """ packet = self._encode(1, {}) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) - def set_state( + async def set_state( self, pwr: Optional[bool] = None, red: Optional[int] = None, @@ -80,7 +80,7 @@ def set_state( state["bulb_sceneidx"] = int(bulb_sceneidx) packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) @@ -119,17 +119,17 @@ class ColorMode(enum.IntEnum): WHITE = 1 SCENE = 2 - def get_state(self) -> dict: + async def get_state(self) -> dict: """Return the power state of the device. Example: `{'red': 128, 'blue': 255, 'green': 128, 'pwr': 1, 'brightness': 75, 'colortemp': 2700, 'hue': 240, 'saturation': 50, 'transitionduration': 1500, 'maxworktime': 0, 'bulb_colormode': 1, 'bulb_scenes': '["@01686464,0,0,0", "#ffffff,10,0,#000000,190,0,0", "2700+100,0,0,0", "#ff0000,500,2500,#00FF00,500,2500,#0000FF,500,2500,0", "@01686464,100,2400,@01686401,100,2400,0", "@01686464,100,2400,@01686401,100,2400,@005a6464,100,2400,@005a6401,100,2400,0", "@01686464,10,0,@00000000,190,0,0", "@01686464,200,0,@005a6464,200,0,0"]', 'bulb_scene': ''}` """ packet = self._encode(1, {}) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) - def set_state( + async def set_state( self, pwr: Optional[bool] = None, red: Optional[int] = None, @@ -175,7 +175,7 @@ def set_state( state["bulb_scene"] = str(bulb_scene) packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) diff --git a/broadlink/remote.py b/broadlink/remote.py index 60c54ce2..64103882 100644 --- a/broadlink/remote.py +++ b/broadlink/remote.py @@ -52,31 +52,31 @@ class rmmini(Device): TYPE = "RMMINI" - def _send(self, command: int, data: bytes = b"") -> bytes: + async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" packet = struct.pack(" None: + async def update(self) -> None: """Update device name and lock status.""" - resp = self._send(0x1) + resp = await self._send(0x1) self.name = resp[0x48:].split(b"\x00")[0].decode() self.is_locked = bool(resp[0x87]) - def send_data(self, data: bytes) -> None: + async def send_data(self, data: bytes) -> None: """Send a code to the device.""" - self._send(0x2, data) + await self._send(0x2, data) - def enter_learning(self) -> None: + async def enter_learning(self) -> None: """Enter infrared learning mode.""" - self._send(0x3) + await self._send(0x3) - def check_data(self) -> bytes: + async def check_data(self) -> bytes: """Return the last captured code.""" - return self._send(0x4) + return await self._send(0x4) class rmpro(rmmini): @@ -84,37 +84,37 @@ class rmpro(rmmini): TYPE = "RMPRO" - def sweep_frequency(self) -> None: + async def sweep_frequency(self) -> None: """Sweep frequency.""" - self._send(0x19) + await self._send(0x19) - def check_frequency(self) -> Tuple[bool, float]: + async def check_frequency(self) -> Tuple[bool, float]: """Return True if the frequency was identified successfully.""" - resp = self._send(0x1A) + resp = await self._send(0x1A) is_found = bool(resp[0]) frequency = struct.unpack(" None: + async def find_rf_packet(self, frequency: Optional[float] = None) -> None: """Enter radiofrequency learning mode.""" payload = bytearray() if frequency: payload += struct.pack(" None: + async def cancel_sweep_frequency(self) -> None: """Cancel sweep frequency.""" - self._send(0x1E) + await self._send(0x1E) - def check_sensors(self) -> dict: + async def check_sensors(self) -> dict: """Return the state of the sensors.""" - resp = self._send(0x1) + resp = await self._send(0x1) temp = struct.unpack(" float: + async def check_temperature(self) -> float: """Return the temperature.""" - return self.check_sensors()["temperature"] + return (await self.check_sensors())["temperature"] class rmminib(rmmini): @@ -122,10 +122,10 @@ class rmminib(rmmini): TYPE = "RMMINIB" - def _send(self, command: int, data: bytes = b"") -> bytes: + async def _send(self, command: int, data: bytes = b"") -> bytes: """Send a packet to the device.""" packet = struct.pack(" dict: + async def check_sensors(self) -> dict: """Return the state of the sensors.""" - resp = self._send(0x24) + resp = await self._send(0x24) temp = struct.unpack(" float: + async def check_temperature(self) -> float: """Return the temperature.""" - return self.check_sensors()["temperature"] + return (await self.check_sensors())["temperature"] - def check_humidity(self) -> float: + async def check_humidity(self) -> float: """Return the humidity.""" - return self.check_sensors()["humidity"] + return (await self.check_sensors())["humidity"] class rm4pro(rm4mini, rmpro): diff --git a/broadlink/sensor.py b/broadlink/sensor.py index 284576fa..f0a99029 100644 --- a/broadlink/sensor.py +++ b/broadlink/sensor.py @@ -16,9 +16,9 @@ class a1(Device): ("noise", ("quiet", "normal", "noisy")), ) - def check_sensors(self) -> dict: + async def check_sensors(self) -> dict: """Return the state of the sensors.""" - data = self.check_sensors_raw() + data = await self.check_sensors_raw() for sensor, levels in self._SENSORS_AND_LEVELS: try: data[sensor] = levels[data[sensor]] @@ -26,10 +26,10 @@ def check_sensors(self) -> dict: data[sensor] = "unknown" return data - def check_sensors_raw(self) -> dict: + async def check_sensors_raw(self) -> dict: """Return the state of the sensors in raw format.""" packet = bytearray([0x1]) - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) data = self.decrypt(resp[0x38:]) @@ -47,7 +47,7 @@ class a2(Device): TYPE = "A2" - def _send(self, operation: int, data: Sequence = b""): + async def _send(self, operation: int, data: Sequence = b""): """Send a command to the device.""" packet = bytearray(12) packet[0x02] = 0xA5 @@ -72,14 +72,14 @@ def _send(self, operation: int, data: Sequence = b""): packet[0x00] = packet_len & 0xFF packet[0x01] = packet_len >> 8 - resp = self.send_packet(0x6A, packet) + resp = await self.send_packet(0x6A, packet) e.check_error(resp[0x22:0x24]) payload = self.decrypt(resp[0x38:]) return payload - def check_sensors_raw(self) -> dict: + async def check_sensors_raw(self) -> dict: """Return the state of the sensors in raw format.""" - data = self._send(1) + data = await self._send(1) return { "temperature": data[0x13] * 256 + data[0x14], diff --git a/broadlink/switch.py b/broadlink/switch.py index 8393f6b1..b41e220d 100644 --- a/broadlink/switch.py +++ b/broadlink/switch.py @@ -12,11 +12,11 @@ class sp1(Device): TYPE = "SP1" - def set_power(self, pwr: bool) -> None: + async def set_power(self, pwr: bool) -> None: """Set the power state of the device.""" packet = bytearray(4) packet[0] = bool(pwr) - response = self.send_packet(0x66, packet) + response = await self.send_packet(0x66, packet) e.check_error(response[0x22:0x24]) @@ -25,19 +25,19 @@ class sp2(Device): TYPE = "SP2" - def set_power(self, pwr: bool) -> None: + async def set_power(self, pwr: bool) -> None: """Set the power state of the device.""" packet = bytearray(16) packet[0] = 2 packet[4] = bool(pwr) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) - def check_power(self) -> bool: + async def check_power(self) -> bool: """Return the power state of the device.""" packet = bytearray(16) packet[0] = 1 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return bool(payload[0x4]) @@ -48,11 +48,11 @@ class sp2s(sp2): TYPE = "SP2S" - def get_energy(self) -> float: + async def get_energy(self) -> float: """Return the power consumption in W.""" packet = bytearray(16) packet[0] = 4 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return int.from_bytes(payload[0x4:0x7], "little") / 1000 @@ -63,36 +63,36 @@ class sp3(Device): TYPE = "SP3" - def set_power(self, pwr: bool) -> None: + async def set_power(self, pwr: bool) -> None: """Set the power state of the device.""" packet = bytearray(16) packet[0] = 2 - packet[4] = self.check_nightlight() << 1 | bool(pwr) - response = self.send_packet(0x6A, packet) + packet[4] = await self.check_nightlight() << 1 | bool(pwr) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) - def set_nightlight(self, ntlight: bool) -> None: + async def set_nightlight(self, ntlight: bool) -> None: """Set the night light state of the device.""" packet = bytearray(16) packet[0] = 2 - packet[4] = bool(ntlight) << 1 | self.check_power() - response = self.send_packet(0x6A, packet) + packet[4] = bool(ntlight) << 1 | await self.check_power() + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) - def check_power(self) -> bool: + async def check_power(self) -> bool: """Return the power state of the device.""" packet = bytearray(16) packet[0] = 1 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return bool(payload[0x4] & 1) - def check_nightlight(self) -> bool: + async def check_nightlight(self) -> bool: """Return the state of the night light.""" packet = bytearray(16) packet[0] = 1 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return bool(payload[0x4] & 2) @@ -103,10 +103,10 @@ class sp3s(sp2): TYPE = "SP3S" - def get_energy(self) -> float: + async def get_energy(self) -> float: """Return the power consumption in W.""" packet = bytearray([8, 0, 254, 1, 5, 1, 0, 0, 0, 45]) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) energy = payload[0x7:0x4:-1].hex() @@ -118,15 +118,15 @@ class sp4(Device): TYPE = "SP4" - def set_power(self, pwr: bool) -> None: + async def set_power(self, pwr: bool) -> None: """Set the power state of the device.""" - self.set_state(pwr=pwr) + await self.set_state(pwr=pwr) - def set_nightlight(self, ntlight: bool) -> None: + async def set_nightlight(self, ntlight: bool) -> None: """Set the night light state of the device.""" - self.set_state(ntlight=ntlight) + await self.set_state(ntlight=ntlight) - def set_state( + async def set_state( self, pwr: Optional[bool] = None, ntlight: Optional[bool] = None, @@ -151,23 +151,23 @@ def set_state( state["childlock"] = int(bool(childlock)) packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) return self._decode(response) - def check_power(self) -> bool: + async def check_power(self) -> bool: """Return the power state of the device.""" - state = self.get_state() + state = await self.get_state() return bool(state["pwr"]) - def check_nightlight(self) -> bool: + async def check_nightlight(self) -> bool: """Return the state of the night light.""" - state = self.get_state() + state = await self.get_state() return bool(state["ntlight"]) - def get_state(self) -> dict: + async def get_state(self) -> dict: """Get full state of device.""" packet = self._encode(1, {}) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) return self._decode(response) def _encode(self, flag: int, state: dict) -> bytes: @@ -196,9 +196,9 @@ class sp4b(sp4): TYPE = "SP4B" - def get_state(self) -> dict: + async def get_state(self) -> dict: """Get full state of device.""" - state = super().get_state() + state = await super().get_state() # Convert sensor data to float. Remove keys if sensors are not supported. sensor_attrs = ["current", "volt", "power", "totalconsum", "overload"] @@ -244,17 +244,17 @@ class bg1(Device): TYPE = "BG1" - def get_state(self) -> dict: + async def get_state(self) -> dict: """Return the power state of the device. Example: `{"pwr":1,"pwr1":1,"pwr2":0,"maxworktime":60,"maxworktime1":60,"maxworktime2":0,"idcbrightness":50}` """ packet = self._encode(1, {}) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) - def set_state( + async def set_state( self, pwr: Optional[bool] = None, pwr1: Optional[bool] = None, @@ -282,7 +282,7 @@ def set_state( state["idcbrightness"] = idcbrightness packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) @@ -321,7 +321,7 @@ class ehc31(bg1): TYPE = "EHC31" - def set_state( + async def set_state( self, pwr: Optional[bool] = None, pwr1: Optional[bool] = None, @@ -367,7 +367,7 @@ def set_state( state["childlock4"] = int(bool(childlock4)) packet = self._encode(2, state) - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) return self._decode(response) @@ -377,7 +377,7 @@ class mp1(Device): TYPE = "MP1" - def set_power_mask(self, sid_mask: int, pwr: bool) -> None: + async def set_power_mask(self, sid_mask: int, pwr: bool) -> None: """Set the power state of the device.""" packet = bytearray(16) packet[0x00] = 0x0D @@ -392,15 +392,15 @@ def set_power_mask(self, sid_mask: int, pwr: bool) -> None: packet[0x0D] = sid_mask packet[0x0E] = sid_mask if pwr else 0 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) - def set_power(self, sid: int, pwr: bool) -> None: + async def set_power(self, sid: int, pwr: bool) -> None: """Set the power state of the device.""" sid_mask = 0x01 << (sid - 1) - self.set_power_mask(sid_mask, pwr) + await self.set_power_mask(sid_mask, pwr) - def check_power_raw(self) -> int: + async def check_power_raw(self) -> int: """Return the power state of the device in raw format.""" packet = bytearray(16) packet[0x00] = 0x0A @@ -412,14 +412,14 @@ def check_power_raw(self) -> int: packet[0x07] = 0xC0 packet[0x08] = 0x01 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) return payload[0x0E] - def check_power(self) -> dict: + async def check_power(self) -> dict: """Return the power state of the device.""" - data = self.check_power_raw() + data = await self.check_power_raw() return { "s1": bool(data & 1), "s2": bool(data & 2), @@ -433,7 +433,7 @@ class mp1s(mp1): TYPE = "MP1S" - def get_state(self) -> dict: + async def get_state(self) -> dict: """Return the power state of the device. voltage in V. @@ -452,7 +452,7 @@ def get_state(self) -> dict: packet[0x08] = 0x01 packet[0x0A] = 0x04 - response = self.send_packet(0x6A, packet) + response = await self.send_packet(0x6A, packet) e.check_error(response[0x22:0x24]) payload = self.decrypt(response[0x38:]) payload_str = payload.hex()[4:-6] diff --git a/cli/broadlink_cli b/cli/broadlink_cli old mode 100755 new mode 100644 index 7913e332..1014986b --- a/cli/broadlink_cli +++ b/cli/broadlink_cli @@ -1,5 +1,6 @@ #!/usr/bin/env python3 import argparse +import asyncio import base64 import time from typing import List @@ -57,165 +58,173 @@ parser.add_argument("--joinwifi", nargs=2, help="Args are SSID PASSPHRASE to con parser.add_argument("data", nargs='*', help="Data to send or convert") args = parser.parse_args() -if args.device: - values = args.device.split() - devtype = int(values[0], 0) - host = values[1] - mac = bytearray.fromhex(values[2]) -elif args.mac: - devtype = args.type - host = args.host - mac = bytearray.fromhex(args.mac) - -if args.host or args.device: - dev = broadlink.gendevice(devtype, (host, DEFAULT_PORT), mac) - dev.auth() - -if args.joinwifi: - broadlink.setup(args.joinwifi[0], args.joinwifi[1], 4) - -if args.convert: - data = bytearray.fromhex(''.join(args.data)) - pulses = data_to_pulses(data) - print(format_pulses(pulses)) -if args.temperature: - print(dev.check_temperature()) -if args.humidity: - print(dev.check_humidity()) -if args.energy: - print(dev.get_energy()) -if args.sensors: - data = dev.check_sensors() - for key in data: - print("{} {}".format(key, data[key])) -if args.send: - data = ( - pulses_to_data(parse_pulses(args.data)) - if args.durations - else bytes.fromhex(''.join(args.data)) - ) - dev.send_data(data) -if args.learn or (args.learnfile and not args.rflearn): - dev.enter_learning() - print("Learning...") - start = time.time() - while time.time() - start < TIMEOUT: - time.sleep(1) - try: - data = dev.check_data() - except (ReadError, StorageError): - continue - else: - break - else: - 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) -if args.check: - if dev.check_power(): - print('* ON *') - else: - print('* OFF *') -if args.checknl: - if dev.check_nightlight(): - print('* ON *') - else: - print('* OFF *') -if args.turnon: - dev.set_power(True) - if dev.check_power(): - print('== Turned * ON * ==') - else: - print('!! Still OFF !!') -if args.turnoff: - dev.set_power(False) - if dev.check_power(): - print('!! Still ON !!') - else: - print('== Turned * OFF * ==') -if args.turnnlon: - dev.set_nightlight(True) - if dev.check_nightlight(): - print('== Turned * ON * ==') - else: - print('!! Still OFF !!') -if args.turnnloff: - dev.set_nightlight(False) - if dev.check_nightlight(): - print('!! Still ON !!') - else: - print('== Turned * OFF * ==') -if args.switch: - if dev.check_power(): - dev.set_power(False) - print('* Switch to OFF *') - else: - dev.set_power(True) - print('* Switch to ON *') -if args.rflearn: - if args.frequency: - frequency = args.frequency - print("Press the button you want to learn, a short press...") - else: - dev.sweep_frequency() - print("Detecting radiofrequency, press and hold the button to learn...") +async def main(): + dev = None + + if args.device: + values = args.device.split() + devtype = int(values[0], 0) + host = values[1] + mac = bytearray.fromhex(values[2]) + elif args.mac: + devtype = args.type + host = args.host + mac = bytearray.fromhex(args.mac) + + if args.host or args.device: + dev = broadlink.gendevice(devtype, (host, DEFAULT_PORT), mac) + await dev.auth() + + if args.joinwifi: + await broadlink.setup(args.joinwifi[0], args.joinwifi[1], 4) + + if args.convert: + data = bytearray.fromhex(''.join(args.data)) + pulses = data_to_pulses(data) + print(format_pulses(pulses)) + if args.temperature: + print(await dev.check_temperature()) + if args.humidity: + print(await dev.check_humidity()) + if args.energy: + print(await dev.get_energy()) + if args.sensors: + data = await dev.check_sensors() + for key in data: + print("{} {}".format(key, data[key])) + if args.send: + data = ( + pulses_to_data(parse_pulses(args.data)) + 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: - time.sleep(1) - locked, frequency = dev.check_frequency() - if locked: + await asyncio.sleep(1) + try: + data = await dev.check_data() + except (ReadError, StorageError): + continue + else: break else: - print("Radiofrequency not found") - dev.cancel_sweep_frequency() + print("No data received...") exit(1) - print("Radiofrequency detected: {}MHz".format(frequency)) - print("You can now let go of the button") + 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) + if args.check: + if await dev.check_power(): + print('* ON *') + else: + print('* OFF *') + if args.checknl: + if await dev.check_nightlight(): + print('* ON *') + else: + print('* OFF *') + if args.turnon: + await dev.set_power(True) + if await dev.check_power(): + print('== Turned * ON * ==') + else: + print('!! Still OFF !!') + if args.turnoff: + await dev.set_power(False) + if await dev.check_power(): + print('!! Still ON !!') + else: + print('== Turned * OFF * ==') + if args.turnnlon: + await dev.set_nightlight(True) + if await dev.check_nightlight(): + print('== Turned * ON * ==') + else: + print('!! Still OFF !!') + if args.turnnloff: + await dev.set_nightlight(False) + if await dev.check_nightlight(): + print('!! Still ON !!') + else: + print('== Turned * OFF * ==') + if args.switch: + if await dev.check_power(): + await dev.set_power(False) + print('* Switch to OFF *') + else: + await dev.set_power(True) + print('* Switch to ON *') + if args.rflearn: + if args.frequency: + frequency = args.frequency + print("Press the button you want to learn, a short press...") + else: + await dev.sweep_frequency() + print("Detecting radiofrequency, press and hold the button to learn...") + + start = time.time() + while time.time() - start < TIMEOUT: + await asyncio.sleep(1) + locked, frequency = await dev.check_frequency() + if locked: + break + else: + print("Radiofrequency not found") + await dev.cancel_sweep_frequency() + exit(1) - input("Press enter to continue...") + print("Radiofrequency detected: {}MHz".format(frequency)) + print("You can now let go of the button") - print("Press the button again, now a short press.") + input("Press enter to continue...") - dev.find_rf_packet(frequency) + print("Press the button again, now a short press.") - start = time.time() - while time.time() - start < TIMEOUT: - time.sleep(1) - try: - data = dev.check_data() - except (ReadError, StorageError): - continue + 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: - break - else: - 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) + 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) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/cli/broadlink_discovery b/cli/broadlink_discovery old mode 100755 new mode 100644 index 477e1bd7..779a1d21 --- a/cli/broadlink_discovery +++ b/cli/broadlink_discovery @@ -1,5 +1,6 @@ #!/usr/bin/env python3 import argparse +import asyncio import broadlink from broadlink.const import DEFAULT_BCAST_ADDR, DEFAULT_TIMEOUT @@ -11,20 +12,26 @@ parser.add_argument("--ip", default=None, help="ip address to use in the discove parser.add_argument("--dst-ip", default=DEFAULT_BCAST_ADDR, help="destination ip address to use in the discovery") args = parser.parse_args() -print("Discovering...") -devices = broadlink.discover(timeout=args.timeout, local_ip_address=args.ip, discover_ip_address=args.dst_ip) -for device in devices: - if device.auth(): - print("###########################################") - print(device.type) - print("# broadlink_cli --type {} --host {} --mac {}".format(hex(device.devtype), device.host[0], - ''.join(format(x, '02x') for x in device.mac))) - print("Device file data (to be used with --device @filename in broadlink_cli) : ") - print("{} {} {}".format(hex(device.devtype), device.host[0], ''.join(format(x, '02x') for x in device.mac))) - try: - print("temperature = {}".format(device.check_temperature())) - except (AttributeError, StorageError): - pass - print("") - else: - print("Error authenticating with device : {}".format(device.host)) + +async def main(): + print("Discovering...") + devices = await broadlink.discover(timeout=args.timeout, local_ip_address=args.ip, discover_ip_address=args.dst_ip) + for device in devices: + if await device.auth(): + print("###########################################") + print(device.type) + print("# broadlink_cli --type {} --host {} --mac {}".format(hex(device.devtype), device.host[0], + ''.join(format(x, '02x') for x in device.mac))) + print("Device file data (to be used with --device @filename in broadlink_cli) : ") + print("{} {} {}".format(hex(device.devtype), device.host[0], ''.join(format(x, '02x') for x in device.mac))) + try: + print("temperature = {}".format(await device.check_temperature())) + except (AttributeError, StorageError): + pass + print("") + else: + print("Error authenticating with device : {}".format(device.host)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_oracle.py b/tests/test_oracle.py index 7ae7d26d..f5159d62 100644 --- a/tests/test_oracle.py +++ b/tests/test_oracle.py @@ -36,7 +36,7 @@ def test_every_public_method_is_covered() -> None: # Methods on Device itself that need a live socket are covered in # test_transport.py, not here. transport_level = {"auth", "hello", "ping", "send_packet", "encrypt", "decrypt", - "update_aes"} + "update_aes", "aclose"} missing = [] for name, cls in inspect.getmembers(broadlink, inspect.isclass): if not issubclass(cls, Device): diff --git a/tests/test_transport.py b/tests/test_transport.py index 7fd8e784..37b6447d 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import socket import pytest @@ -21,60 +22,65 @@ INIT_VECT = bytes.fromhex("562e17996d093d28ddb3ba695a2e6f58") -class FakeSocket: - """A UDP socket stand-in: records sendto, replays canned recvfrom.""" +class FakeTransport: + """Datagram transport stand-in: records sendto, feeds canned replies.""" - instances: list["FakeSocket"] = [] - - def __init__(self, *args, **kwargs): - self.sent: list[tuple[bytes, tuple[str, int]]] = [] - self.inbox: list[tuple[bytes, tuple[str, int]]] = list(FakeSocket.queue) - self.timeout = None + def __init__(self, protocol, local_addr, remote_addr, broadcast, replies): + self.protocol = protocol + self.local_addr = local_addr or ("0.0.0.0", 0) + self.remote_addr = remote_addr + self.broadcast = broadcast + self.replies = list(replies) + self.sent: list[tuple[bytes, tuple[str, int] | None]] = [] self.closed = False - self.bound = None - FakeSocket.instances.append(self) - - queue: list[tuple[bytes, tuple[str, int]]] = [] - - def setsockopt(self, *args): - pass - - def settimeout(self, value): - self.timeout = value - - def bind(self, addr): - self.bound = addr - def getsockname(self): - return self.bound or ("0.0.0.0", 0) + def sendto(self, data, addr=None): + self.sent.append((bytes(data), addr or self.remote_addr)) + # Each send releases the next canned reply, if any, exactly like a + # device answering one request. + if self.replies: + self.protocol.queue.put_nowait(self.replies.pop(0)) - def sendto(self, data, addr): - self.sent.append((bytes(data), addr)) + def get_extra_info(self, name): + if name == "sockname": + return (self.local_addr[0], self.local_addr[1] or 40000) + return None - def recvfrom(self, size): - if not self.inbox: - raise socket.timeout() - return self.inbox.pop(0) + def is_closing(self): + return self.closed def close(self): self.closed = True - def __enter__(self): - return self - def __exit__(self, *exc): - self.close() +class FakeNet: + """Replacement for broadlink.device._open_endpoint.""" + + def __init__(self): + self.replies: list[tuple[bytes, tuple[str, int]]] = [] + self.endpoints: list[FakeTransport] = [] + + async def __call__(self, local_addr=None, remote_addr=None, broadcast=False): + protocol = device_module._Protocol() + transport = FakeTransport(protocol, local_addr, remote_addr, broadcast, self.replies) + self.replies = [] + protocol.connection_made(transport) + self.endpoints.append(transport) + return transport, protocol @pytest.fixture -def fake_socket(monkeypatch): - FakeSocket.instances = [] - FakeSocket.queue = [] - monkeypatch.setattr(device_module.socket, "socket", FakeSocket) - monkeypatch.setattr(broadlink.socket, "socket", FakeSocket) +def net(monkeypatch): + fake = FakeNet() + monkeypatch.setattr(device_module, "_open_endpoint", fake) + monkeypatch.setattr(broadlink, "_open_endpoint", fake) # Keep the retry loop from waiting on real time. - monkeypatch.setattr(device_module, "DEFAULT_RETRY_INTVL", 0.001) - return FakeSocket + monkeypatch.setattr(device_module, "DEFAULT_RETRY_INTVL", 0.005) + return fake + + +def run(coro): + return asyncio.run(coro) def fixed_device(cls=Device, devtype=0x2737) -> Device: @@ -86,17 +92,18 @@ def fixed_device(cls=Device, devtype=0x2737) -> Device: # ------------------------------------------------------------------ send_packet -def test_send_packet_wire_bytes(fake_socket): +def test_send_packet_wire_bytes(net): dev = fixed_device() dev.id = 0x00000001 payload = bytes([0x01]) + bytes(15) - fake_socket.queue = [(make_response(dev, bytes(16)), HOST)] + net.replies = [(make_response(dev, bytes(16)), HOST)] - resp = dev.send_packet(0x6A, payload) + resp = run(dev.send_packet(0x6A, payload)) - sock = fake_socket.instances[-1] - assert len(sock.sent) == 1 - frame, addr = sock.sent[0] + ep = net.endpoints[-1] + assert ep.remote_addr == HOST + assert len(ep.sent) == 1 + frame, addr = ep.sent[0] assert addr == HOST assert frame[0x00:0x08] == bytes.fromhex("5aa5aa555aa5aa55") assert frame[0x24:0x26] == (0x2737).to_bytes(2, "little") @@ -116,55 +123,110 @@ def test_send_packet_wire_bytes(fake_socket): assert dev.count == 0x8001 -def test_send_packet_pads_payload_to_block(fake_socket): +def test_send_packet_pads_payload_to_block(net): dev = fixed_device() - fake_socket.queue = [(make_response(dev, b""), HOST)] - dev.send_packet(0x6A, bytes(20)) - frame = fake_socket.instances[-1].sent[0][0] + net.replies = [(make_response(dev, b""), HOST)] + run(dev.send_packet(0x6A, bytes(20))) + frame = net.endpoints[-1].sent[0][0] assert len(frame) == 0x38 + 32 assert dev.decrypt(frame[0x38:]) == bytes(32) -def test_send_packet_counter_wraps_with_high_bit(fake_socket): +def test_send_packet_counter_wraps_with_high_bit(net): dev = fixed_device() dev.count = 0xFFFF - fake_socket.queue = [(make_response(dev, b""), HOST)] - dev.send_packet(0x6A, b"") + net.replies = [(make_response(dev, b""), HOST)] + run(dev.send_packet(0x6A, b"")) assert dev.count == 0x8000 -def test_send_packet_retries_then_times_out(fake_socket, monkeypatch): +def test_send_packet_reuses_one_endpoint_and_serializes(net): + dev = fixed_device() + + async def go(): + net.replies = [(make_response(dev, b""), HOST)] + await dev.send_packet(0x6A, b"") + ep = net.endpoints[-1] + ep.replies = [(make_response(dev, b""), HOST), (make_response(dev, b""), HOST)] + await asyncio.gather(dev.send_packet(0x6A, b"a"), dev.send_packet(0x6A, b"b")) + return ep + + ep = run(go()) + assert len(net.endpoints) == 1 + assert len(ep.sent) == 3 + # Counters are consecutive: the lock kept the two concurrent calls apart. + counts = [int.from_bytes(f[0x28:0x2A], "little") for f, _ in ep.sent] + assert counts == [0x8001, 0x8002, 0x8003] + + +def test_aclose_then_reopen(net): + dev = fixed_device() + + async def go(): + net.replies = [(make_response(dev, b""), HOST)] + await dev.send_packet(0x6A, b"") + await dev.aclose() + assert net.endpoints[-1].closed + net.replies = [(make_response(dev, b""), HOST)] + async with dev: + await dev.send_packet(0x6A, b"") + + run(go()) + assert len(net.endpoints) == 2 + assert net.endpoints[-1].closed # the context manager closed it + + +def test_send_packet_retries_then_times_out(net): dev = fixed_device() - dev.timeout = 0.01 - fake_socket.queue = [] # never answers + dev.timeout = 0.02 + net.replies = [] # never answers with pytest.raises(e.NetworkTimeoutError) as err: - dev.send_packet(0x6A, b"") + run(dev.send_packet(0x6A, b"")) assert err.value.errno == -4000 - assert len(fake_socket.instances[-1].sent) >= 1 + assert len(net.endpoints[-1].sent) >= 2 # resent at least once -def test_send_packet_rejects_short_response(fake_socket): +def test_send_packet_rejects_short_response(net): dev = fixed_device() - fake_socket.queue = [(bytes(0x10), HOST)] + net.replies = [(bytes(0x10), HOST)] with pytest.raises(e.DataValidationError) as err: - dev.send_packet(0x6A, b"") + run(dev.send_packet(0x6A, b"")) assert err.value.errno == -4007 -def test_send_packet_rejects_bad_checksum(fake_socket): +def test_send_packet_rejects_bad_checksum(net): dev = fixed_device() frame = bytearray(make_response(dev, b"")) frame[0x20] ^= 0xFF - fake_socket.queue = [(bytes(frame), HOST)] + net.replies = [(bytes(frame), HOST)] with pytest.raises(e.DataValidationError) as err: - dev.send_packet(0x6A, b"") + run(dev.send_packet(0x6A, b"")) assert err.value.errno == -4008 +def test_stale_reply_is_drained_before_a_request(net): + dev = fixed_device() + + async def go(): + net.replies = [(make_response(dev, b""), HOST)] + await dev.send_packet(0x6A, b"") + ep = net.endpoints[-1] + # A late packet shows up between requests; it must not be taken as + # the answer to the next one. + stale = bytearray(make_response(dev, b"")) + stale[0x20] ^= 0xFF # corrupt so it would fail validation if used + ep.protocol.queue.put_nowait((bytes(stale), HOST)) + ep.replies = [(make_response(dev, bytes([7]) + bytes(15)), HOST)] + resp = await dev.send_packet(0x6A, b"") + return dev.decrypt(resp[0x38:])[0] + + assert run(go()) == 7 + + # ------------------------------------------------------------------------- auth -def test_auth_uses_initial_key_and_installs_session_key(fake_socket): +def test_auth_uses_initial_key_and_installs_session_key(net): dev = fixed_device() dev.id = 99 # stale session; auth must reset it before sending dev.update_aes(bytes(range(16))) # stale key @@ -175,11 +237,11 @@ def test_auth_uses_initial_key_and_installs_session_key(fake_socket): # INITIAL key, which is what the device expects auth to be decrypted with. fresh = fixed_device() reply = make_response(fresh, session_id.to_bytes(4, "little") + session_key) - fake_socket.queue = [(reply, HOST)] + net.replies = [(reply, HOST)] - assert dev.auth() is True + assert run(dev.auth()) is True - frame = fake_socket.instances[-1].sent[0][0] + frame = net.endpoints[-1].sent[0][0] assert frame[0x26:0x28] == (0x65).to_bytes(2, "little") assert frame[0x30:0x34] == bytes(4) # id reset to 0 for the handshake plaintext = fresh.decrypt(frame[0x38:]) @@ -196,11 +258,57 @@ def test_auth_uses_initial_key_and_installs_session_key(fake_socket): assert dev.encrypt(bytes(16)) == probe.encrypt(bytes(16)) -def test_auth_surfaces_device_error(fake_socket): +def test_auth_surfaces_device_error(net): dev = fixed_device() - fake_socket.queue = [(make_response(dev, bytes(20), error=0xFFF9), HOST)] + net.replies = [(make_response(dev, bytes(20), error=0xFFF9), HOST)] with pytest.raises(e.AuthorizationError): - dev.auth() + run(dev.auth()) + + +def test_expired_session_is_reauthenticated_once(net): + dev = fixed_device() + dev.id = 5 + session_key = bytes.fromhex("00112233445566778899aabbccddeeff") + fresh = fixed_device() + auth_reply = make_response(fresh, (0x42).to_bytes(4, "little") + session_key) + + async def go(): + # First request: device says the control key expired (-7). + net.replies = [(make_response(dev, b"", error=0xFFF9), HOST)] + await dev._endpoint() + ep = net.endpoints[-1] + # After the expired reply the library must auth (reply 2, under the + # initial key) and resend (reply 3, under the new session key). + renewed = fixed_device() + renewed.update_aes(session_key) + ep.replies = [ + (auth_reply, HOST), + (make_response(renewed, bytes([9]) + bytes(15)), HOST), + ] + ep.replies.insert(0, (make_response(dev, b"", error=0xFFF9), HOST)) + resp = await dev.send_packet(0x6A, bytes(16)) + return ep, resp + + ep, resp = run(go()) + types = [int.from_bytes(f[0x26:0x28], "little") for f, _ in ep.sent] + assert types == [0x6A, 0x65, 0x6A] + assert dev.id == 0x42 + assert resp[0x22:0x24] == b"\x00\x00" + assert dev.decrypt(resp[0x38:])[0] == 9 + + +def test_reauth_is_not_attempted_twice(net): + dev = fixed_device() + + async def go(): + await dev._endpoint() + ep = net.endpoints[-1] + expired = (make_response(dev, b"", error=0xFFF9), HOST) + ep.replies = [expired, expired] # request fails, auth fails + return await dev.send_packet(0x6A, b"") + + with pytest.raises(e.AuthorizationError): + run(go()) # ------------------------------------------------------------------- discovery @@ -215,37 +323,53 @@ def hello_response(devtype: int, mac: bytes, name: str, locked: bool) -> bytes: return bytes(frame) -def test_scan_builds_hello_packet_and_parses_replies(fake_socket): - fake_socket.queue = [ +async def collect(aiter): + return [x async for x in aiter] + + +def test_scan_builds_hello_packet_and_parses_replies(net): + other = bytes.fromhex("34ea34000001") + net.replies = [ (hello_response(0x6026, MAC, "Bedroom RM", False), ("192.0.2.10", 80)), (hello_response(0x6026, MAC, "Bedroom RM", False), ("192.0.2.10", 80)), # dup - (hello_response(0x2711, bytes.fromhex("34ea34000001"), "Plug", True), - ("192.0.2.11", 80)), + (hello_response(0x2711, other, "Plug", True), ("192.0.2.11", 80)), ] - found = list(device_module.scan(timeout=0.01, local_ip_address="192.0.2.2")) + async def go(): + found = [] + async for entry in device_module.scan(timeout=0.02, local_ip_address="192.0.2.2"): + found.append(entry) + ep = net.endpoints[-1] + # replies are released one per send; pull the rest through + while ep.replies: + ep.protocol.queue.put_nowait(ep.replies.pop(0)) + return found + + found = run(go()) assert found == [ (0x6026, ("192.0.2.10", 80), MAC, "Bedroom RM", False), - (0x2711, ("192.0.2.11", 80), bytes.fromhex("34ea34000001"), "Plug", True), + (0x2711, ("192.0.2.11", 80), other, "Plug", True), ] - sock = fake_socket.instances[-1] - assert sock.bound == ("192.0.2.2", 0) - packet, addr = sock.sent[0] + ep = net.endpoints[-1] + assert ep.local_addr == ("192.0.2.2", 0) + assert ep.broadcast is True + packet, addr = ep.sent[0] assert addr == ("255.255.255.255", 80) assert len(packet) == 0x30 assert packet[0x26] == 6 assert packet[0x18:0x1C] == socket.inet_aton("192.0.2.2")[::-1] + assert packet[0x1C:0x1E] == (40000).to_bytes(2, "little") # bound port body = bytearray(packet) body[0x20:0x22] = b"\x00\x00" assert packet[0x20:0x22] == (sum(body, 0xBEAF) & 0xFFFF).to_bytes(2, "little") - assert sock.closed + assert ep.closed -def test_discover_and_hello_build_devices(fake_socket): - fake_socket.queue = [ +def test_discover_and_hello_build_devices(net): + net.replies = [ (hello_response(0x6026, MAC, "Bedroom RM", False), ("192.0.2.10", 80)), ] - devices = broadlink.discover(timeout=0.01) + devices = run(broadlink.discover(timeout=0.02)) assert len(devices) == 1 dev = devices[0] assert isinstance(dev, broadlink.rm4pro) @@ -255,45 +379,47 @@ def test_discover_and_hello_build_devices(fake_socket): assert dev.model == "RM4 pro" assert dev.manufacturer == "Broadlink" - fake_socket.queue = [ + net.replies = [ (hello_response(0x6026, MAC, "Bedroom RM", True), ("192.0.2.10", 80)), ] - dev = broadlink.hello("192.0.2.10", timeout=0.01) + dev = run(broadlink.hello("192.0.2.10", timeout=0.02)) assert dev.is_locked is True - assert fake_socket.instances[-1].sent[0][1] == ("192.0.2.10", 80) + assert net.endpoints[-1].sent[0][1] == ("192.0.2.10", 80) + assert net.endpoints[-1].closed -def test_hello_times_out(fake_socket): - fake_socket.queue = [] +def test_hello_times_out(net): + net.replies = [] with pytest.raises(e.NetworkTimeoutError): - broadlink.hello("192.0.2.10", timeout=0.01) + run(broadlink.hello("192.0.2.10", timeout=0.02)) -def test_device_hello_validates_identity(fake_socket): +def test_device_hello_validates_identity(net): dev = fixed_device(broadlink.rm4pro, 0x6026) - fake_socket.queue = [(hello_response(0x6026, MAC, "Renamed", True), HOST)] - assert dev.hello() is True + net.replies = [(hello_response(0x6026, MAC, "Renamed", True), HOST)] + assert run(dev.hello()) is True assert dev.name == "Renamed" assert dev.is_locked is True - fake_socket.queue = [ + net.replies = [ (hello_response(0x6026, bytes.fromhex("000000000001"), "Other", False), HOST) ] with pytest.raises(e.DataValidationError): - dev.hello() + run(dev.hello()) - fake_socket.queue = [(hello_response(0x2711, MAC, "Other", False), HOST)] + net.replies = [(hello_response(0x2711, MAC, "Other", False), HOST)] with pytest.raises(e.DataValidationError): - dev.hello() + run(dev.hello()) -def test_ping_packet(fake_socket): +def test_ping_packet(net): dev = fixed_device() - dev.ping() - packet, addr = fake_socket.instances[-1].sent[0] + run(dev.ping()) + packet, addr = net.endpoints[-1].sent[0] assert addr == HOST assert len(packet) == 0x30 assert packet[0x26] == 1 + assert net.endpoints[-1].closed # ------------------------------------------------------------------ gendevice @@ -340,10 +466,11 @@ def test_product_table_has_no_duplicate_ids(): # ----------------------------------------------------------------------- setup -def test_setup_packet(fake_socket): - broadlink.setup("MyWifi", "hunter2", 3, ip_address="192.0.2.255") - packet, addr = fake_socket.instances[-1].sent[0] +def test_setup_packet(net): + run(broadlink.setup("MyWifi", "hunter2", 3, ip_address="192.0.2.255")) + packet, addr = net.endpoints[-1].sent[0] assert addr == ("192.0.2.255", 80) + assert net.endpoints[-1].broadcast is True assert len(packet) == 0x88 assert packet[0x26] == 0x14 assert packet[68:74] == b"MyWifi" @@ -354,6 +481,7 @@ def test_setup_packet(fake_socket): body = bytearray(packet) body[0x20:0x22] = b"\x00\x00" assert packet[0x20:0x22] == (sum(body, 0xBEAF) & 0xFFFF).to_bytes(2, "little") + assert net.endpoints[-1].closed # ------------------------------------------------------------------ exceptions