Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Fix type hints #794

Merged
merged 1 commit into from
Apr 17, 2024
Merged

Fix type hints #794

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

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion broadlink/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,7 +269,7 @@ def discover(

def xdiscover(
timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None,
local_ip_address: str | None = None,
discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[Device, None, None]:
Expand Down
5 changes: 3 additions & 2 deletions broadlink/cover.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Support for covers."""
import time
from typing import Sequence

from . import exceptions as e
from .device import Device
Expand Down Expand Up @@ -63,7 +64,7 @@ class dooya2(Device):

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."""
packet = bytearray(12)
packet[0x02] = 0xA5
Expand Down Expand Up @@ -120,7 +121,7 @@ class wser(Device):

TYPE = "WSER"

def _send(self, operation: int, data: bytes = b""):
def _send(self, operation: int, data: Sequence = b""):
"""Send a command to the device."""
packet = bytearray(12)
packet[0x02] = 0xA5
Expand Down
2 changes: 1 addition & 1 deletion broadlink/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@

def scan(
timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None,
local_ip_address: str | None = None,
discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[HelloResponse, None, None]:
Expand Down
16 changes: 9 additions & 7 deletions broadlink/hub.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ def get_subdevices(self, step: int = 5) -> list:

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."""
state = {}
if did is not None:
Expand All @@ -55,10 +55,10 @@ def get_state(self, did: str = None) -> dict:

def set_state(
self,
did: str = None,
pwr1: bool = None,
pwr2: bool = None,
pwr3: bool = None,
did: str | None = None,
pwr1: bool | None = None,
pwr2: bool | None = None,
pwr3: bool | None = None,
) -> dict:
"""Set the power state of the device."""
state = {}
Expand All @@ -81,7 +81,9 @@ def _encode(self, flag: int, state: dict) -> bytes:
# flag: 1 for reading, 2 for writing.
packet = bytearray(12)
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)
checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x04:0x06] = checksum.to_bytes(2, "little")
Expand All @@ -91,5 +93,5 @@ def _decode(self, response: bytes) -> dict:
"""Decode a JSON packet."""
payload = self.decrypt(response[0x38:])
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
62 changes: 32 additions & 30 deletions broadlink/light.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,20 +32,20 @@ def get_state(self) -> dict:

def set_state(
self,
pwr: bool = None,
red: int = None,
blue: int = None,
green: int = None,
brightness: int = None,
colortemp: int = None,
hue: int = None,
saturation: int = None,
transitionduration: int = None,
maxworktime: int = None,
bulb_colormode: int = None,
bulb_scenes: str = None,
bulb_scene: str = None,
bulb_sceneidx: int = None,
pwr: bool | None = None,
red: int | None = None,
blue: int | None = None,
green: int | None = None,
brightness: int | None = None,
colortemp: int | None = None,
hue: int | None = None,
saturation: int | None = None,
transitionduration: int | None = None,
maxworktime: int | None = None,
bulb_colormode: int | None = None,
bulb_scenes: str | None = None,
bulb_scene: str | None = None,
bulb_sceneidx: int | None = None,
) -> dict:
"""Set the power state of the device."""
state = {}
Expand Down Expand Up @@ -101,7 +101,7 @@ def _decode(self, response: bytes) -> dict:
"""Decode a JSON packet."""
payload = self.decrypt(response[0x38:])
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


Expand Down Expand Up @@ -130,19 +130,19 @@ def get_state(self) -> dict:

def set_state(
self,
pwr: bool = None,
red: int = None,
blue: int = None,
green: int = None,
brightness: int = None,
colortemp: int = None,
hue: int = None,
saturation: int = None,
transitionduration: int = None,
maxworktime: int = None,
bulb_colormode: int = None,
bulb_scenes: str = None,
bulb_scene: str = None,
pwr: bool | None = None,
red: int | None = None,
blue: int | None = None,
green: int | None = None,
brightness: int | None = None,
colortemp: int | None = None,
hue: int | None = None,
saturation: int | None = None,
transitionduration: int | None = None,
maxworktime: int | None = None,
bulb_colormode: int | None = None,
bulb_scenes: str | None = None,
bulb_scene: str | None = None,
) -> dict:
"""Set the power state of the device."""
state = {}
Expand Down Expand Up @@ -183,7 +183,9 @@ def _encode(self, flag: int, state: dict) -> bytes:
# flag: 1 for reading, 2 for writing.
packet = bytearray(12)
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)
checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x04:0x06] = checksum.to_bytes(2, "little")
Expand All @@ -193,5 +195,5 @@ def _decode(self, response: bytes) -> dict:
"""Decode a JSON packet."""
payload = self.decrypt(response[0x38:])
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
12 changes: 6 additions & 6 deletions broadlink/remote.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
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."""
result = bytearray(4)
result[0x00] = 0x26
Expand All @@ -25,7 +25,7 @@ def pulses_to_data(pulses: t.List[int], tick: int = 32.84) -> None:
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."""
result = []
index = 4
Expand All @@ -38,8 +38,8 @@ def data_to_pulses(data: bytes, tick: int = 32.84) -> t.List[int]:
if chunk == 0:
try:
chunk = 256 * data[index] + data[index + 1]
except IndexError:
raise ValueError("Malformed data.")
except IndexError as err:
raise ValueError("Malformed data.") from err
index += 2

result.append(int(chunk * tick))
Expand Down Expand Up @@ -95,7 +95,7 @@ def check_frequency(self) -> t.Tuple[bool, float]:
frequency = struct.unpack("<I", resp[1:5])[0] / 1000.0
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."""
payload = bytearray()
if frequency:
Expand Down Expand Up @@ -129,7 +129,7 @@ def _send(self, command: int, data: bytes = b"") -> bytes:
e.check_error(resp[0x22:0x24])
payload = self.decrypt(resp[0x38:])
p_len = struct.unpack("<H", payload[:0x2])[0]
return payload[0x6 : p_len + 2]
return payload[0x6:p_len+2]


class rm4mini(rmminib):
Expand Down
4 changes: 3 additions & 1 deletion broadlink/sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
"""Support for sensors."""
from typing import Sequence

from . import exceptions as e
from .device import Device

Expand Down Expand Up @@ -45,7 +47,7 @@ class a2(Device):

TYPE = "A2"

def _send(self, operation: int, data: bytes = b""):
def _send(self, operation: int, data: Sequence = b""):
"""Send a command to the device."""
packet = bytearray(12)
packet[0x02] = 0xA5
Expand Down
71 changes: 40 additions & 31 deletions broadlink/switch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,12 +127,12 @@ def set_nightlight(self, ntlight: bool) -> None:

def set_state(
self,
pwr: bool = None,
ntlight: bool = None,
indicator: bool = None,
ntlbrightness: int = None,
maxworktime: int = None,
childlock: bool = None,
pwr: bool | None = None,
ntlight: bool | None = None,
indicator: bool | None = None,
ntlbrightness: int | None = None,
maxworktime: int | None = None,
childlock: bool | None = None,
) -> dict:
"""Set state of device."""
state = {}
Expand Down Expand Up @@ -186,7 +186,7 @@ def _decode(self, response: bytes) -> dict:
e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:])
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


Expand Down Expand Up @@ -234,7 +234,7 @@ def _decode(self, response: bytes) -> dict:
e.check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:])
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


Expand All @@ -255,13 +255,13 @@ def get_state(self) -> dict:

def set_state(
self,
pwr: bool = None,
pwr1: bool = None,
pwr2: bool = None,
maxworktime: int = None,
maxworktime1: int = None,
maxworktime2: int = None,
idcbrightness: int = None,
pwr: bool | None = None,
pwr1: bool | None = None,
pwr2: bool | None = None,
maxworktime: int | None = None,
maxworktime1: int | None = None,
maxworktime2: int | None = None,
idcbrightness: int | None = None,
) -> dict:
"""Set the power state of the device."""
state = {}
Expand Down Expand Up @@ -291,7 +291,16 @@ def _encode(self, flag: int, state: dict) -> bytes:
data = json.dumps(state).encode()
length = 12 + len(data)
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)
checksum = sum(packet[0x2:], 0xBEAF) & 0xFFFF
Expand All @@ -302,7 +311,7 @@ def _decode(self, response: bytes) -> dict:
"""Decode a message."""
payload = self.decrypt(response[0x38:])
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


Expand All @@ -313,19 +322,19 @@ class ehc31(bg1):

def set_state(
self,
pwr: bool = None,
pwr1: bool = None,
pwr2: bool = None,
pwr3: bool = None,
maxworktime1: int = None,
maxworktime2: int = None,
maxworktime3: int = None,
idcbrightness: int = None,
childlock: bool = None,
childlock1: bool = None,
childlock2: bool = None,
childlock3: bool = None,
childlock4: bool = None,
pwr: bool | None = None,
pwr1: bool | None = None,
pwr2: bool | None = None,
pwr3: bool | None = None,
maxworktime1: int | None = None,
maxworktime2: int | None = None,
maxworktime3: int | None = None,
idcbrightness: int | None = None,
childlock: bool | None = None,
childlock1: bool | None = None,
childlock2: bool | None = None,
childlock3: bool | None = None,
childlock4: bool | None = None,
) -> dict:
"""Set the power state of the device."""
state = {}
Expand Down Expand Up @@ -449,7 +458,7 @@ def get_state(self) -> dict:

def get_value(start, end, factors):
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)
)
return value
Expand Down
Loading