kamp/python-broadlink
1
0
Fork 0
mirror of https://github.com/mjg59/python-broadlink.git synced 2024-11-21 22:51:41 +01:00
Code

Fix type hints (#794)

Browse Source
This commit is contained in:
Felipe Martins Diel 2024-04-17 02:49:13 -03:00 committed by GitHub
parent 1e11558613
commit c4979562c8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 95 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
broadlink/__init__.py
View File

@ -269,7 +269,7 @@ def discover(
def xdiscover( def xdiscover(
timeout: int = DEFAULT_TIMEOUT, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str | None = None,
discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = DEFAULT_PORT, discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[Device, None, None]: ) -> t.Generator[Device, None, None]:

5
broadlink/cover.py
View File

@ -1,5 +1,6 @@
"""Support for covers.""" """Support for covers."""
import time import time
from typing import Sequence
from . import exceptions as e from . import exceptions as e
from .device import Device from .device import Device
@ -63,7 +64,7 @@ class dooya2(Device):
TYPE = "DT360E-2" TYPE = "DT360E-2"
def _send(self, operation: int, data: bytes = b""): def _send(self, operation: int, data: Sequence = b""):
"""Send a command to the device.""" """Send a command to the device."""
packet = bytearray(12) packet = bytearray(12)
packet[0x02] = 0xA5 packet[0x02] = 0xA5
@ -120,7 +121,7 @@ class wser(Device):
TYPE = "WSER" TYPE = "WSER"
def _send(self, operation: int, data: bytes = b""): def _send(self, operation: int, data: Sequence = b""):
"""Send a command to the device.""" """Send a command to the device."""
packet = bytearray(12) packet = bytearray(12)
packet[0x02] = 0xA5 packet[0x02] = 0xA5

2
broadlink/device.py
View File

@ -22,7 +22,7 @@ HelloResponse = t.Tuple[int, t.Tuple[str, int], str, str, bool]
def scan( def scan(
timeout: int = DEFAULT_TIMEOUT, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str | None = None,
discover_ip_address: str = DEFAULT_BCAST_ADDR, discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = DEFAULT_PORT, discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[HelloResponse, None, None]: ) -> t.Generator[HelloResponse, None, None]:

16
broadlink/hub.py
View File

@ -42,7 +42,7 @@ class s3(Device):
return sub_devices return sub_devices
def get_state(self, did: str = None) -> dict: def get_state(self, did: str | None = None) -> dict:
"""Return the power state of the device.""" """Return the power state of the device."""
state = {} state = {}
if did is not None: if did is not None:
@ -55,10 +55,10 @@ class s3(Device):
def set_state( def set_state(
self, self,
did: str = None, did: str | None = None,
pwr1: bool = None, pwr1: bool | None = None,
pwr2: bool = None, pwr2: bool | None = None,
pwr3: bool = None, pwr3: bool | None = None,
) -> dict: ) -> dict:
"""Set the power state of the device.""" """Set the power state of the device."""
state = {} state = {}
@ -81,7 +81,9 @@ class s3(Device):
# flag: 1 for reading, 2 for writing. # flag: 1 for reading, 2 for writing.
packet = bytearray(12) packet = bytearray(12)
data = json.dumps(state, separators=(",", ":")).encode() data = json.dumps(state, separators=(",", ":")).encode()
struct.pack_into("<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)) struct.pack_into(
"<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)
)
packet.extend(data) packet.extend(data)
checksum = sum(packet, 0xBEAF) & 0xFFFF checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x04:0x06] = checksum.to_bytes(2, "little") packet[0x04:0x06] = checksum.to_bytes(2, "little")
@ -91,5 +93,5 @@ class s3(Device):
"""Decode a JSON packet.""" """Decode a JSON packet."""
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0x08)[0] js_len = struct.unpack_from("<I", payload, 0x08)[0]
state = json.loads(payload[0x0C : 0x0C + js_len]) state = json.loads(payload[0x0C:0x0C+js_len])
return state return state

62
broadlink/light.py
View File

@ -32,20 +32,20 @@ class lb1(Device):
def set_state( def set_state(
self, self,
pwr: bool = None, pwr: bool | None = None,
red: int = None, red: int | None = None,
blue: int = None, blue: int | None = None,
green: int = None, green: int | None = None,
brightness: int = None, brightness: int | None = None,
colortemp: int = None, colortemp: int | None = None,
hue: int = None, hue: int | None = None,
saturation: int = None, saturation: int | None = None,
transitionduration: int = None, transitionduration: int | None = None,
maxworktime: int = None, maxworktime: int | None = None,
bulb_colormode: int = None, bulb_colormode: int | None = None,
bulb_scenes: str = None, bulb_scenes: str | None = None,
bulb_scene: str = None, bulb_scene: str | None = None,
bulb_sceneidx: int = None, bulb_sceneidx: int | None = None,
) -> dict: ) -> dict:
"""Set the power state of the device.""" """Set the power state of the device."""
state = {} state = {}
@ -101,7 +101,7 @@ class lb1(Device):
"""Decode a JSON packet.""" """Decode a JSON packet."""
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0xA)[0] js_len = struct.unpack_from("<I", payload, 0xA)[0]
state = json.loads(payload[0xE : 0xE + js_len]) state = json.loads(payload[0xE:0xE+js_len])
return state return state
@ -130,19 +130,19 @@ class lb2(Device):
def set_state( def set_state(
self, self,
pwr: bool = None, pwr: bool | None = None,
red: int = None, red: int | None = None,
blue: int = None, blue: int | None = None,
green: int = None, green: int | None = None,
brightness: int = None, brightness: int | None = None,
colortemp: int = None, colortemp: int | None = None,
hue: int = None, hue: int | None = None,
saturation: int = None, saturation: int | None = None,
transitionduration: int = None, transitionduration: int | None = None,
maxworktime: int = None, maxworktime: int | None = None,
bulb_colormode: int = None, bulb_colormode: int | None = None,
bulb_scenes: str = None, bulb_scenes: str | None = None,
bulb_scene: str = None, bulb_scene: str | None = None,
) -> dict: ) -> dict:
"""Set the power state of the device.""" """Set the power state of the device."""
state = {} state = {}
@ -183,7 +183,9 @@ class lb2(Device):
# flag: 1 for reading, 2 for writing. # flag: 1 for reading, 2 for writing.
packet = bytearray(12) packet = bytearray(12)
data = json.dumps(state, separators=(",", ":")).encode() data = json.dumps(state, separators=(",", ":")).encode()
struct.pack_into("<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)) struct.pack_into(
"<HHHBBI", packet, 0, 0xA5A5, 0x5A5A, 0, flag, 0x0B, len(data)
)
packet.extend(data) packet.extend(data)
checksum = sum(packet, 0xBEAF) & 0xFFFF checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x04:0x06] = checksum.to_bytes(2, "little") packet[0x04:0x06] = checksum.to_bytes(2, "little")
@ -193,5 +195,5 @@ class lb2(Device):
"""Decode a JSON packet.""" """Decode a JSON packet."""
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0x08)[0] js_len = struct.unpack_from("<I", payload, 0x08)[0]
state = json.loads(payload[0x0C : 0x0C + js_len]) state = json.loads(payload[0x0C:0x0C+js_len])
return state return state

12
broadlink/remote.py
View File

@ -6,7 +6,7 @@ from . import exceptions as e
from .device import Device from .device import Device
def pulses_to_data(pulses: t.List[int], tick: int = 32.84) -> None: def pulses_to_data(pulses: t.List[int], tick: float = 32.84) -> bytes:
"""Convert a microsecond duration sequence into a Broadlink IR packet.""" """Convert a microsecond duration sequence into a Broadlink IR packet."""
result = bytearray(4) result = bytearray(4)
result[0x00] = 0x26 result[0x00] = 0x26
@ -25,7 +25,7 @@ def pulses_to_data(pulses: t.List[int], tick: int = 32.84) -> None:
return result return result
def data_to_pulses(data: bytes, tick: int = 32.84) -> t.List[int]: def data_to_pulses(data: bytes, tick: float = 32.84) -> t.List[int]:
"""Parse a Broadlink packet into a microsecond duration sequence.""" """Parse a Broadlink packet into a microsecond duration sequence."""
result = [] result = []
index = 4 index = 4
@ -38,8 +38,8 @@ def data_to_pulses(data: bytes, tick: int = 32.84) -> t.List[int]:
if chunk == 0: if chunk == 0:
try: try:
chunk = 256 * data[index] + data[index + 1] chunk = 256 * data[index] + data[index + 1]
except IndexError: except IndexError as err:
raise ValueError("Malformed data.") raise ValueError("Malformed data.") from err
index += 2 index += 2
result.append(int(chunk * tick)) result.append(int(chunk * tick))
@ -95,7 +95,7 @@ class rmpro(rmmini):
frequency = struct.unpack("<I", resp[1:5])[0] / 1000.0 frequency = struct.unpack("<I", resp[1:5])[0] / 1000.0
return is_found, frequency return is_found, frequency
def find_rf_packet(self, frequency: float = None) -> None: def find_rf_packet(self, frequency: float | None = None) -> None:
"""Enter radiofrequency learning mode.""" """Enter radiofrequency learning mode."""
payload = bytearray() payload = bytearray()
if frequency: if frequency:
@ -129,7 +129,7 @@ class rmminib(rmmini):
e.check_error(resp[0x22:0x24]) e.check_error(resp[0x22:0x24])
payload = self.decrypt(resp[0x38:]) payload = self.decrypt(resp[0x38:])
p_len = struct.unpack("<H", payload[:0x2])[0] p_len = struct.unpack("<H", payload[:0x2])[0]
return payload[0x6 : p_len + 2] return payload[0x6:p_len+2]
class rm4mini(rmminib): class rm4mini(rmminib):

4
broadlink/sensor.py
View File

@ -1,4 +1,6 @@
"""Support for sensors.""" """Support for sensors."""
from typing import Sequence
from . import exceptions as e from . import exceptions as e
from .device import Device from .device import Device
@ -45,7 +47,7 @@ class a2(Device):
TYPE = "A2" TYPE = "A2"
def _send(self, operation: int, data: bytes = b""): def _send(self, operation: int, data: Sequence = b""):
"""Send a command to the device.""" """Send a command to the device."""
packet = bytearray(12) packet = bytearray(12)
packet[0x02] = 0xA5 packet[0x02] = 0xA5

71
broadlink/switch.py
View File

@ -127,12 +127,12 @@ class sp4(Device):
def set_state( def set_state(
self, self,
pwr: bool = None, pwr: bool | None = None,
ntlight: bool = None, ntlight: bool | None = None,
indicator: bool = None, indicator: bool | None = None,
ntlbrightness: int = None, ntlbrightness: int | None = None,
maxworktime: int = None, maxworktime: int | None = None,
childlock: bool = None, childlock: bool | None = None,
) -> dict: ) -> dict:
"""Set state of device.""" """Set state of device."""
state = {} state = {}
@ -186,7 +186,7 @@ class sp4(Device):
e.check_error(response[0x22:0x24]) e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0x08)[0] js_len = struct.unpack_from("<I", payload, 0x08)[0]
state = json.loads(payload[0x0C : 0x0C + js_len]) state = json.loads(payload[0x0C:0x0C+js_len])
return state return state
@ -234,7 +234,7 @@ class sp4b(sp4):
e.check_error(response[0x22:0x24]) e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0xA)[0] js_len = struct.unpack_from("<I", payload, 0xA)[0]
state = json.loads(payload[0x0E : 0x0E + js_len]) state = json.loads(payload[0x0E:0x0E+js_len])
return state return state
@ -255,13 +255,13 @@ class bg1(Device):
def set_state( def set_state(
self, self,
pwr: bool = None, pwr: bool | None = None,
pwr1: bool = None, pwr1: bool | None = None,
pwr2: bool = None, pwr2: bool | None = None,
maxworktime: int = None, maxworktime: int | None = None,
maxworktime1: int = None, maxworktime1: int | None = None,
maxworktime2: int = None, maxworktime2: int | None = None,
idcbrightness: int = None, idcbrightness: int | None = None,
) -> dict: ) -> dict:
"""Set the power state of the device.""" """Set the power state of the device."""
state = {} state = {}
@ -291,7 +291,16 @@ class bg1(Device):
data = json.dumps(state).encode() data = json.dumps(state).encode()
length = 12 + len(data) length = 12 + len(data)
struct.pack_into( struct.pack_into(
"<HHHHBBI", packet, 0, length, 0xA5A5, 0x5A5A, 0x0000, flag, 0x0B, len(data) "<HHHHBBI",
packet,
0,
length,
0xA5A5,
0x5A5A,
0x0000,
flag,
0x0B,
len(data),
) )
packet.extend(data) packet.extend(data)
checksum = sum(packet[0x2:], 0xBEAF) & 0xFFFF checksum = sum(packet[0x2:], 0xBEAF) & 0xFFFF
@ -302,7 +311,7 @@ class bg1(Device):
"""Decode a message.""" """Decode a message."""
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
js_len = struct.unpack_from("<I", payload, 0x0A)[0] js_len = struct.unpack_from("<I", payload, 0x0A)[0]
state = json.loads(payload[0x0E : 0x0E + js_len]) state = json.loads(payload[0x0E:0x0E+js_len])
return state return state
@ -313,19 +322,19 @@ class ehc31(bg1):
def set_state( def set_state(
self, self,
pwr: bool = None, pwr: bool | None = None,
pwr1: bool = None, pwr1: bool | None = None,
pwr2: bool = None, pwr2: bool | None = None,
pwr3: bool = None, pwr3: bool | None = None,
maxworktime1: int = None, maxworktime1: int | None = None,
maxworktime2: int = None, maxworktime2: int | None = None,
maxworktime3: int = None, maxworktime3: int | None = None,
idcbrightness: int = None, idcbrightness: int | None = None,
childlock: bool = None, childlock: bool | None = None,
childlock1: bool = None, childlock1: bool | None = None,
childlock2: bool = None, childlock2: bool | None = None,
childlock3: bool = None, childlock3: bool | None = None,
childlock4: bool = None, childlock4: bool | None = None,
) -> dict: ) -> dict:
"""Set the power state of the device.""" """Set the power state of the device."""
state = {} state = {}
@ -449,7 +458,7 @@ class mp1s(mp1):
def get_value(start, end, factors): def get_value(start, end, factors):
value = sum( value = sum(
int(payload_str[i - 2 : i]) * factor int(payload_str[i-2:i]) * factor
for i, factor in zip(range(start, end, -2), factors) for i, factor in zip(range(start, end, -2), factors)
) )
return value return value