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

Add GH Actions (#470)

Browse Source 
Add GitHub actions and run black.
This commit is contained in:
Kendell R 2020-11-05 11:00:42 -08:00 committed by Felipe Martins Diel
parent b3542a8865
commit dfd3372c36
11 changed files with 408 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
.github/workflows/flake8.yaml vendored Normal file
View File

@ -0,0 +1,32 @@
name: Python flake8
on:
push:
branches: [ main, master, dev, development ]
pull_request:
branches: [ main, master, dev, development ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.6, 3.7, 3.8, 3.9]
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install flake8 wemake-python-styleguide
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. ignore magic numbers and use double quotes and ignore numbers with zeroes before them.
# and ignore lowercase hex numbers and ignore isort incorrect imports
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=90 --ignore=WPS432,WPS339,WPS341,I --inline-quotes double --statistics

117
broadlink/__init__.py
View File

@ -20,79 +20,79 @@ SUPPORTED_TYPES = {
0x2716: (sp2, "NEO PRO", "Ankuoo"), 0x2716: (sp2, "NEO PRO", "Ankuoo"),
0x2717: (sp2, "NEO", "Ankuoo"), 0x2717: (sp2, "NEO", "Ankuoo"),
0x2719: (sp2, "SP2-compatible", "Honeywell"), 0x2719: (sp2, "SP2-compatible", "Honeywell"),
0x271a: (sp2, "SP2-compatible", "Honeywell"), 0x271A: (sp2, "SP2-compatible", "Honeywell"),
0x2720: (sp2, "SP mini", "Broadlink"), 0x2720: (sp2, "SP mini", "Broadlink"),
0x2728: (sp2, "SP2-compatible", "URANT"), 0x2728: (sp2, "SP2-compatible", "URANT"),
0x2733: (sp2, "SP3", "Broadlink"), 0x2733: (sp2, "SP3", "Broadlink"),
0x2736: (sp2, "SP mini+", "Broadlink"), 0x2736: (sp2, "SP mini+", "Broadlink"),
0x273e: (sp2, "SP mini", "Broadlink"), 0x273E: (sp2, "SP mini", "Broadlink"),
0x7530: (sp2, "SP2", "Broadlink (OEM)"), 0x7530: (sp2, "SP2", "Broadlink (OEM)"),
0x7539: (sp2, "SP2-IL", "Broadlink (OEM)"), 0x7539: (sp2, "SP2-IL", "Broadlink (OEM)"),
0x753e: (sp2, "SP mini 3", "Broadlink"), 0x753E: (sp2, "SP mini 3", "Broadlink"),
0x7540: (sp2, "MP2", "Broadlink"), 0x7540: (sp2, "MP2", "Broadlink"),
0X7544: (sp2, "SP2-CL", "Broadlink"), 0x7544: (sp2, "SP2-CL", "Broadlink"),
0x7546: (sp2, "SP2-UK/BR/IN", "Broadlink (OEM)"), 0x7546: (sp2, "SP2-UK/BR/IN", "Broadlink (OEM)"),
0x7547: (sp2, "SC1", "Broadlink"), 0x7547: (sp2, "SC1", "Broadlink"),
0x7918: (sp2, "SP2", "Broadlink (OEM)"), 0x7918: (sp2, "SP2", "Broadlink (OEM)"),
0x7919: (sp2, "SP2-compatible", "Honeywell"), 0x7919: (sp2, "SP2-compatible", "Honeywell"),
0x791a: (sp2, "SP2-compatible", "Honeywell"), 0x791A: (sp2, "SP2-compatible", "Honeywell"),
0x7d00: (sp2, "SP3-EU", "Broadlink (OEM)"), 0x7D00: (sp2, "SP3-EU", "Broadlink (OEM)"),
0x7d0d: (sp2, "SP mini 3", "Broadlink (OEM)"), 0x7D0D: (sp2, "SP mini 3", "Broadlink (OEM)"),
0x9479: (sp2, "SP3S-US", "Broadlink"), 0x9479: (sp2, "SP3S-US", "Broadlink"),
0x947a: (sp2, "SP3S-EU", "Broadlink"), 0x947A: (sp2, "SP3S-EU", "Broadlink"),
0x756c: (sp4, "SP4M", "Broadlink"), 0x756C: (sp4, "SP4M", "Broadlink"),
0x7579: (sp4, "SP4L-EU", "Broadlink"), 0x7579: (sp4, "SP4L-EU", "Broadlink"),
0x7583: (sp4, "SP mini 3", "Broadlink"), 0x7583: (sp4, "SP mini 3", "Broadlink"),
0x7d11: (sp4, "SP mini 3", "Broadlink"), 0x7D11: (sp4, "SP mini 3", "Broadlink"),
0x648b: (sp4b, "SP4M-US", "Broadlink"), 0x648B: (sp4b, "SP4M-US", "Broadlink"),
0x2712: (rm, "RM pro/pro+", "Broadlink"), 0x2712: (rm, "RM pro/pro+", "Broadlink"),
0x272a: (rm, "RM pro", "Broadlink"), 0x272A: (rm, "RM pro", "Broadlink"),
0x2737: (rm, "RM mini 3", "Broadlink"), 0x2737: (rm, "RM mini 3", "Broadlink"),
0x273d: (rm, "RM pro", "Broadlink"), 0x273D: (rm, "RM pro", "Broadlink"),
0x277c: (rm, "RM home", "Broadlink"), 0x277C: (rm, "RM home", "Broadlink"),
0x2783: (rm, "RM home", "Broadlink"), 0x2783: (rm, "RM home", "Broadlink"),
0x2787: (rm, "RM pro", "Broadlink"), 0x2787: (rm, "RM pro", "Broadlink"),
0x278b: (rm, "RM plus", "Broadlink"), 0x278B: (rm, "RM plus", "Broadlink"),
0x278f: (rm, "RM mini", "Broadlink"), 0x278F: (rm, "RM mini", "Broadlink"),
0x2797: (rm, "RM pro+", "Broadlink"), 0x2797: (rm, "RM pro+", "Broadlink"),
0x279d: (rm, "RM pro+", "Broadlink"), 0x279D: (rm, "RM pro+", "Broadlink"),
0x27a1: (rm, "RM plus", "Broadlink"), 0x27A1: (rm, "RM plus", "Broadlink"),
0x27a6: (rm, "RM plus", "Broadlink"), 0x27A6: (rm, "RM plus", "Broadlink"),
0x27a9: (rm, "RM pro+", "Broadlink"), 0x27A9: (rm, "RM pro+", "Broadlink"),
0x27c2: (rm, "RM mini 3", "Broadlink"), 0x27C2: (rm, "RM mini 3", "Broadlink"),
0x27c3: (rm, "RM pro+", "Broadlink"), 0x27C3: (rm, "RM pro+", "Broadlink"),
0x27c7: (rm, "RM mini 3", "Broadlink"), 0x27C7: (rm, "RM mini 3", "Broadlink"),
0x27cc: (rm, "RM mini 3", "Broadlink"), 0x27CC: (rm, "RM mini 3", "Broadlink"),
0x27cd: (rm, "RM mini 3", "Broadlink"), 0x27CD: (rm, "RM mini 3", "Broadlink"),
0x27d0: (rm, "RM mini 3", "Broadlink"), 0x27D0: (rm, "RM mini 3", "Broadlink"),
0x27d1: (rm, "RM mini 3", "Broadlink"), 0x27D1: (rm, "RM mini 3", "Broadlink"),
0x27de: (rm, "RM mini 3", "Broadlink"), 0x27DE: (rm, "RM mini 3", "Broadlink"),
0x51da: (rm4, "RM4 mini", "Broadlink"), 0x51DA: (rm4, "RM4 mini", "Broadlink"),
0x5f36: (rm4, "RM mini 3", "Broadlink"), 0x5F36: (rm4, "RM mini 3", "Broadlink"),
0x6026: (rm4, "RM4 pro", "Broadlink"), 0x6026: (rm4, "RM4 pro", "Broadlink"),
0x6070: (rm4, "RM4C mini", "Broadlink"), 0x6070: (rm4, "RM4C mini", "Broadlink"),
0x610e: (rm4, "RM4 mini", "Broadlink"), 0x610E: (rm4, "RM4 mini", "Broadlink"),
0x610f: (rm4, "RM4C mini", "Broadlink"), 0x610F: (rm4, "RM4C mini", "Broadlink"),
0x61a2: (rm4, "RM4 pro", "Broadlink"), 0x61A2: (rm4, "RM4 pro", "Broadlink"),
0x62bc: (rm4, "RM4 mini", "Broadlink"), 0x62BC: (rm4, "RM4 mini", "Broadlink"),
0x62be: (rm4, "RM4C mini", "Broadlink"), 0x62BE: (rm4, "RM4C mini", "Broadlink"),
0x648d: (rm4, "RM4 mini", "Broadlink"), 0x648D: (rm4, "RM4 mini", "Broadlink"),
0x649b: (rm4, "RM4 pro", "Broadlink"), 0x649B: (rm4, "RM4 pro", "Broadlink"),
0x653a: (rm4, "RM4 mini", "Broadlink"), 0x653A: (rm4, "RM4 mini", "Broadlink"),
0x2714: (a1, "e-Sensor", "Broadlink"), 0x2714: (a1, "e-Sensor", "Broadlink"),
0x4eb5: (mp1, "MP1-1K4S", "Broadlink"), 0x4EB5: (mp1, "MP1-1K4S", "Broadlink"),
0x4ef7: (mp1, "MP1-1K4S", "Broadlink (OEM)"), 0x4EF7: (mp1, "MP1-1K4S", "Broadlink (OEM)"),
0x4f1b: (mp1, "MP1-1K3S2U", "Broadlink (OEM)"), 0x4F1B: (mp1, "MP1-1K3S2U", "Broadlink (OEM)"),
0x4f65: (mp1, "MP1-1K3S2U", "Broadlink"), 0x4F65: (mp1, "MP1-1K3S2U", "Broadlink"),
0x5043: (lb1, "SB800TD", "Broadlink (OEM)"), 0x5043: (lb1, "SB800TD", "Broadlink (OEM)"),
0x504e: (lb1, "LB1", "Broadlink"), 0x504E: (lb1, "LB1", "Broadlink"),
0x60c7: (lb1, "LB1", "Broadlink"), 0x60C7: (lb1, "LB1", "Broadlink"),
0x60c8: (lb1, "LB1", "Broadlink"), 0x60C8: (lb1, "LB1", "Broadlink"),
0x6112: (lb1, "LB1", "Broadlink"), 0x6112: (lb1, "LB1", "Broadlink"),
0x2722: (S1C, "S2KIT", "Broadlink"), 0x2722: (S1C, "S2KIT", "Broadlink"),
0x4ead: (hysen, "HY02B05H", "Hysen"), 0x4EAD: (hysen, "HY02B05H", "Hysen"),
0x4e4d: (dooya, "DT360E-45/20", "Dooya"), 0x4E4D: (dooya, "DT360E-45/20", "Dooya"),
0x51e3: (bg1, "BG800/BG900", "BG Electrical"), 0x51E3: (bg1, "BG800/BG900", "BG Electrical"),
} }
@ -140,29 +140,25 @@ def hello(
def discover( def discover(
timeout: int = 10, timeout: int = 10,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = '255.255.255.255', discover_ip_address: str = "255.255.255.255",
discover_ip_port: int = 80, discover_ip_port: int = 80,
) -> List[device]: ) -> List[device]:
"""Discover devices connected to the local network.""" """Discover devices connected to the local network."""
responses = scan( responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port)
timeout, local_ip_address, discover_ip_address, discover_ip_port
)
return [gendevice(*resp) for resp in responses] return [gendevice(*resp) for resp in responses]
def xdiscover( def xdiscover(
timeout: int = 10, timeout: int = 10,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = '255.255.255.255', discover_ip_address: str = "255.255.255.255",
discover_ip_port: int = 80, discover_ip_port: int = 80,
) -> Generator[device, None, None]: ) -> Generator[device, None, None]:
"""Discover devices connected to the local network. """Discover devices connected to the local network.
This function returns a generator that yields devices instantly. This function returns a generator that yields devices instantly.
""" """
responses = scan( responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port)
timeout, local_ip_address, discover_ip_address, discover_ip_port
)
for resp in responses: for resp in responses:
yield gendevice(*resp) yield gendevice(*resp)
@ -191,13 +187,12 @@ def setup(ssid: str, password: str, security_mode: int) -> None:
payload[0x85] = pass_length # Character length of password payload[0x85] = pass_length # Character length of password
payload[0x86] = security_mode # Type of encryption payload[0x86] = security_mode # Type of encryption
checksum = sum(payload, 0xbeaf) & 0xffff checksum = sum(payload, 0xBEAF) & 0xFFFF
payload[0x20] = checksum & 0xff # Checksum 1 position payload[0x20] = checksum & 0xFF # Checksum 1 position
payload[0x21] = checksum >> 8 # Checksum 2 position payload[0x21] = checksum >> 8 # Checksum 2 position
sock = socket.socket(socket.AF_INET, # Internet sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Internet # UDP
socket.SOCK_DGRAM) # UDP
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.sendto(payload, ('255.255.255.255', 80)) sock.sendto(payload, ("255.255.255.255", 80))
sock.close() sock.close()

28
broadlink/alarm.py
View File

@ -7,21 +7,21 @@ class S1C(device):
"""Controls a Broadlink S1C.""" """Controls a Broadlink S1C."""
_SENSORS_TYPES = { _SENSORS_TYPES = {
0x31: 'Door Sensor', # 49 as hex 0x31: "Door Sensor", # 49 as hex
0x91: 'Key Fob', # 145 as hex, as serial on fob corpse 0x91: "Key Fob", # 145 as hex, as serial on fob corpse
0x21: 'Motion Sensor' # 33 as hex 0x21: "Motion Sensor", # 33 as hex
} }
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args, **kwargs) -> None:
"""Initialize the controller.""" """Initialize the controller."""
device.__init__(self, *args, **kwargs) device.__init__(self, *args, **kwargs)
self.type = 'S1C' self.type = "S1C"
def get_sensors_status(self) -> dict: def get_sensors_status(self) -> dict:
"""Return the state of the sensors.""" """Return the state of the sensors."""
packet = bytearray(16) packet = bytearray(16)
packet[0] = 0x06 # 0x06 - get sensors info, 0x07 - probably add sensors packet[0] = 0x06 # 0x06 - get sensors info, 0x07 - probably add sensors
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
if not payload: if not payload:
@ -29,20 +29,20 @@ class S1C(device):
count = payload[0x4] count = payload[0x4]
sensor_data = payload[0x6:] sensor_data = payload[0x6:]
sensors = [ sensors = [
bytearray(sensor_data[i * 83:(i + 1) * 83]) bytearray(sensor_data[i * 83 : (i + 1) * 83])
for i in range(len(sensor_data) // 83) for i in range(len(sensor_data) // 83)
] ]
return { return {
'count': count, "count": count,
'sensors': [ "sensors": [
{ {
'status': sensor[0], "status": sensor[0],
'name': sensor[4:26].decode().strip('\x00'), "name": sensor[4:26].decode().strip("\x00"),
'type': self._SENSORS_TYPES.get(sensor[3], 'Unknown'), "type": self._SENSORS_TYPES.get(sensor[3], "Unknown"),
'order': sensor[1], "order": sensor[1],
'serial': sensor[26:30].hex(), "serial": sensor[26:30].hex(),
} }
for sensor in sensors for sensor in sensors
if any(sensor[26:30]) if any(sensor[26:30])
] ],
} }

120
broadlink/climate.py
View File

@ -33,19 +33,22 @@ class hysen(device):
request_payload.append((crc >> 8) & 0xFF) request_payload.append((crc >> 8) & 0xFF)
# send to device # send to device
response = self.send_packet(0x6a, request_payload) response = self.send_packet(0x6A, request_payload)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
response_payload = self.decrypt(response[0x38:]) response_payload = self.decrypt(response[0x38:])
# experimental check on CRC in response (first 2 bytes are len, and trailing bytes are crc) # experimental check on CRC in response (first 2 bytes are len, and trailing bytes are crc)
response_payload_len = response_payload[0] response_payload_len = response_payload[0]
if response_payload_len + 2 > len(response_payload): if response_payload_len + 2 > len(response_payload):
raise ValueError('hysen_response_error', 'first byte of response is not length') raise ValueError(
"hysen_response_error", "first byte of response is not length"
)
crc = calculate_crc16(response_payload[2:response_payload_len]) crc = calculate_crc16(response_payload[2:response_payload_len])
if (response_payload[response_payload_len] == crc & 0xFF) and ( if (response_payload[response_payload_len] == crc & 0xFF) and (
response_payload[response_payload_len + 1] == (crc >> 8) & 0xFF): response_payload[response_payload_len + 1] == (crc >> 8) & 0xFF
):
return response_payload[2:response_payload_len] return response_payload[2:response_payload_len]
raise ValueError('hysen_response_error', 'CRC check on response failed') raise ValueError("hysen_response_error", "CRC check on response failed")
def get_temp(self) -> int: def get_temp(self) -> int:
"""Return the room temperature in degrees celsius.""" """Return the room temperature in degrees celsius."""
@ -64,43 +67,53 @@ class hysen(device):
""" """
payload = self.send_request(bytearray([0x01, 0x03, 0x00, 0x00, 0x00, 0x16])) payload = self.send_request(bytearray([0x01, 0x03, 0x00, 0x00, 0x00, 0x16]))
data = {} data = {}
data['remote_lock'] = payload[3] & 1 data["remote_lock"] = payload[3] & 1
data['power'] = payload[4] & 1 data["power"] = payload[4] & 1
data['active'] = (payload[4] >> 4) & 1 data["active"] = (payload[4] >> 4) & 1
data['temp_manual'] = (payload[4] >> 6) & 1 data["temp_manual"] = (payload[4] >> 6) & 1
data['room_temp'] = (payload[5] & 255) / 2.0 data["room_temp"] = (payload[5] & 255) / 2.0
data['thermostat_temp'] = (payload[6] & 255) / 2.0 data["thermostat_temp"] = (payload[6] & 255) / 2.0
data['auto_mode'] = payload[7] & 15 data["auto_mode"] = payload[7] & 15
data['loop_mode'] = (payload[7] >> 4) & 15 data["loop_mode"] = (payload[7] >> 4) & 15
data['sensor'] = payload[8] data["sensor"] = payload[8]
data['osv'] = payload[9] data["osv"] = payload[9]
data['dif'] = payload[10] data["dif"] = payload[10]
data['svh'] = payload[11] data["svh"] = payload[11]
data['svl'] = payload[12] data["svl"] = payload[12]
data['room_temp_adj'] = ((payload[13] << 8) + payload[14]) / 2.0 data["room_temp_adj"] = ((payload[13] << 8) + payload[14]) / 2.0
if data['room_temp_adj'] > 32767: if data["room_temp_adj"] > 32767:
data['room_temp_adj'] = 32767 - data['room_temp_adj'] data["room_temp_adj"] = 32767 - data["room_temp_adj"]
data['fre'] = payload[15] data["fre"] = payload[15]
data['poweron'] = payload[16] data["poweron"] = payload[16]
data['unknown'] = payload[17] data["unknown"] = payload[17]
data['external_temp'] = (payload[18] & 255) / 2.0 data["external_temp"] = (payload[18] & 255) / 2.0
data['hour'] = payload[19] data["hour"] = payload[19]
data['min'] = payload[20] data["min"] = payload[20]
data['sec'] = payload[21] data["sec"] = payload[21]
data['dayofweek'] = payload[22] data["dayofweek"] = payload[22]
weekday = [] weekday = []
for i in range(0, 6): for i in range(0, 6):
weekday.append( weekday.append(
{'start_hour': payload[2 * i + 23], 'start_minute': payload[2 * i + 24], 'temp': payload[i + 39] / 2.0}) {
"start_hour": payload[2 * i + 23],
"start_minute": payload[2 * i + 24],
"temp": payload[i + 39] / 2.0,
}
)
data['weekday'] = weekday data["weekday"] = weekday
weekend = [] weekend = []
for i in range(6, 8): for i in range(6, 8):
weekend.append( weekend.append(
{'start_hour': payload[2 * i + 23], 'start_minute': payload[2 * i + 24], 'temp': payload[i + 39] / 2.0}) {
"start_hour": payload[2 * i + 23],
"start_minute": payload[2 * i + 24],
"temp": payload[i + 39] / 2.0,
}
)
data['weekend'] = weekend data["weekend"] = weekend
return data return data
# Change controller mode # Change controller mode
@ -140,8 +153,27 @@ class hysen(device):
poweron: int, poweron: int,
) -> None: ) -> None:
"""Set advanced options.""" """Set advanced options."""
input_payload = bytearray([0x01, 0x10, 0x00, 0x02, 0x00, 0x05, 0x0a, loop_mode, sensor, osv, dif, svh, svl, input_payload = bytearray(
(int(adj * 2) >> 8 & 0xff), (int(adj * 2) & 0xff), fre, poweron]) [
0x01,
0x10,
0x00,
0x02,
0x00,
0x05,
0x0A,
loop_mode,
sensor,
osv,
dif,
svh,
svl,
(int(adj * 2) >> 8 & 0xFF),
(int(adj * 2) & 0xFF),
fre,
poweron,
]
)
self.send_request(input_payload) self.send_request(input_payload)
# For backwards compatibility only. Prefer calling set_mode directly. # For backwards compatibility only. Prefer calling set_mode directly.
@ -169,7 +201,11 @@ class hysen(device):
# n.b. day=1 is Monday, ..., day=7 is Sunday # n.b. day=1 is Monday, ..., day=7 is Sunday
def set_time(self, hour: int, minute: int, second: int, day: int) -> None: def set_time(self, hour: int, minute: int, second: int, day: int) -> None:
"""Set the time.""" """Set the time."""
self.send_request(bytearray([0x01, 0x10, 0x00, 0x08, 0x00, 0x02, 0x04, hour, minute, second, day])) self.send_request(
bytearray(
[0x01, 0x10, 0x00, 0x08, 0x00, 0x02, 0x04, hour, minute, second, day]
)
)
# Set timer schedule # Set timer schedule
# Format is the same as you get from get_full_status. # Format is the same as you get from get_full_status.
@ -180,25 +216,25 @@ class hysen(device):
def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None: def set_schedule(self, weekday: List[dict], weekend: List[dict]) -> None:
"""Set timer schedule.""" """Set timer schedule."""
# Begin with some magic values ... # Begin with some magic values ...
input_payload = bytearray([0x01, 0x10, 0x00, 0x0a, 0x00, 0x0c, 0x18]) input_payload = bytearray([0x01, 0x10, 0x00, 0x0A, 0x00, 0x0C, 0x18])
# Now simply append times/temps # Now simply append times/temps
# weekday times # weekday times
for i in range(0, 6): for i in range(0, 6):
input_payload.append(weekday[i]['start_hour']) input_payload.append(weekday[i]["start_hour"])
input_payload.append(weekday[i]['start_minute']) input_payload.append(weekday[i]["start_minute"])
# weekend times # weekend times
for i in range(0, 2): for i in range(0, 2):
input_payload.append(weekend[i]['start_hour']) input_payload.append(weekend[i]["start_hour"])
input_payload.append(weekend[i]['start_minute']) input_payload.append(weekend[i]["start_minute"])
# weekday temperatures # weekday temperatures
for i in range(0, 6): for i in range(0, 6):
input_payload.append(int(weekday[i]['temp'] * 2)) input_payload.append(int(weekday[i]["temp"] * 2))
# weekend temperatures # weekend temperatures
for i in range(0, 2): for i in range(0, 2):
input_payload.append(int(weekend[i]['temp'] * 2)) input_payload.append(int(weekend[i]["temp"] * 2))
self.send_request(input_payload) self.send_request(input_payload)

8
broadlink/cover.py
View File

@ -17,12 +17,12 @@ class dooya(device):
"""Send a packet to the device.""" """Send a packet to the device."""
packet = bytearray(16) packet = bytearray(16)
packet[0] = 0x09 packet[0] = 0x09
packet[2] = 0xbb packet[2] = 0xBB
packet[3] = magic1 packet[3] = magic1
packet[4] = magic2 packet[4] = magic2
packet[9] = 0xfa packet[9] = 0xFA
packet[10] = 0x44 packet[10] = 0x44
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return payload[4] return payload[4]
@ -41,7 +41,7 @@ class dooya(device):
def get_percentage(self) -> int: def get_percentage(self) -> int:
"""Return the position of the curtain.""" """Return the position of the curtain."""
return self._send(0x06, 0x5d) return self._send(0x06, 0x5D)
def set_percentage_and_wait(self, new_percentage: int) -> None: def set_percentage_and_wait(self, new_percentage: int) -> None:
"""Set the position of the curtain.""" """Set the position of the curtain."""

130
broadlink/device.py
View File

@ -17,7 +17,7 @@ HelloResponse = Tuple[int, Tuple[str, int], str, str, bool]
def scan( def scan(
timeout: int = 10, timeout: int = 10,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = '255.255.255.255', discover_ip_address: str = "255.255.255.255",
discover_ip_port: int = 80, discover_ip_port: int = 80,
) -> Generator[HelloResponse, None, None]: ) -> Generator[HelloResponse, None, None]:
"""Broadcast a hello message and yield responses.""" """Broadcast a hello message and yield responses."""
@ -36,38 +36,38 @@ def scan(
timezone = int(time.timezone / -3600) timezone = int(time.timezone / -3600)
if timezone < 0: if timezone < 0:
packet[0x08] = 0xff + timezone - 1 packet[0x08] = 0xFF + timezone - 1
packet[0x09] = 0xff packet[0x09] = 0xFF
packet[0x0a] = 0xff packet[0x0A] = 0xFF
packet[0x0b] = 0xff packet[0x0B] = 0xFF
else: else:
packet[0x08] = timezone packet[0x08] = timezone
packet[0x09] = 0 packet[0x09] = 0
packet[0x0a] = 0 packet[0x0A] = 0
packet[0x0b] = 0 packet[0x0B] = 0
year = datetime.now().year year = datetime.now().year
packet[0x0c] = year & 0xff packet[0x0C] = year & 0xFF
packet[0x0d] = year >> 8 packet[0x0D] = year >> 8
packet[0x0e] = datetime.now().minute packet[0x0E] = datetime.now().minute
packet[0x0f] = datetime.now().hour packet[0x0F] = datetime.now().hour
subyear = str(year)[2:] subyear = str(year)[2:]
packet[0x10] = int(subyear) packet[0x10] = int(subyear)
packet[0x11] = datetime.now().isoweekday() packet[0x11] = datetime.now().isoweekday()
packet[0x12] = datetime.now().day packet[0x12] = datetime.now().day
packet[0x13] = datetime.now().month packet[0x13] = datetime.now().month
address = local_ip_address.split('.') address = local_ip_address.split(".")
packet[0x18] = int(address[3]) packet[0x18] = int(address[3])
packet[0x19] = int(address[2]) packet[0x19] = int(address[2])
packet[0x1a] = int(address[1]) packet[0x1A] = int(address[1])
packet[0x1b] = int(address[0]) packet[0x1B] = int(address[0])
packet[0x1c] = port & 0xff packet[0x1C] = port & 0xFF
packet[0x1d] = port >> 8 packet[0x1D] = port >> 8
packet[0x26] = 6 packet[0x26] = 6
checksum = sum(packet, 0xbeaf) & 0xffff checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x20] = checksum & 0xff packet[0x20] = checksum & 0xFF
packet[0x21] = checksum >> 8 packet[0x21] = checksum >> 8
starttime = time.time() starttime = time.time()
@ -85,12 +85,12 @@ def scan(
break break
devtype = response[0x34] | response[0x35] << 8 devtype = response[0x34] | response[0x35] << 8
mac = bytes(reversed(response[0x3a:0x40])) mac = bytes(reversed(response[0x3A:0x40]))
if (host, mac, devtype) in discovered: if (host, mac, devtype) in discovered:
continue continue
discovered.append((host, mac, devtype)) discovered.append((host, mac, devtype))
name = response[0x40:].split(b'\x00')[0].decode('utf-8') name = response[0x40:].split(b"\x00")[0].decode("utf-8")
is_locked = bool(response[-1]) is_locked = bool(response[-1])
yield devtype, host, mac, name, is_locked yield devtype, host, mac, name, is_locked
finally: finally:
@ -114,20 +114,20 @@ class device:
"""Initialize the controller.""" """Initialize the controller."""
self.host = host self.host = host
self.mac = bytes.fromhex(mac) if isinstance(mac, str) else mac self.mac = bytes.fromhex(mac) if isinstance(mac, str) else mac
self.devtype = devtype if devtype is not None else 0x272a self.devtype = devtype if devtype is not None else 0x272A
self.timeout = timeout self.timeout = timeout
self.name = name self.name = name
self.model = model self.model = model
self.manufacturer = manufacturer self.manufacturer = manufacturer
self.is_locked = is_locked self.is_locked = is_locked
self.count = random.randrange(0xffff) self.count = random.randrange(0xFFFF)
self.iv = bytes.fromhex('562e17996d093d28ddb3ba695a2e6f58') self.iv = bytes.fromhex("562e17996d093d28ddb3ba695a2e6f58")
self.id = bytes(4) self.id = bytes(4)
self.type = "Unknown" self.type = "Unknown"
self.lock = threading.Lock() self.lock = threading.Lock()
self.aes = None self.aes = None
key = bytes.fromhex('097628343fe99e23765c1513accf8b02') key = bytes.fromhex("097628343fe99e23765c1513accf8b02")
self.update_aes(key) self.update_aes(key)
def __repr__(self): def __repr__(self):
@ -138,7 +138,7 @@ class device:
hex(self.devtype), hex(self.devtype),
self.host[0], self.host[0],
self.host[1], self.host[1],
':'.join(format(x, '02x') for x in self.mac), ":".join(format(x, "02x") for x in self.mac),
self.name, self.name,
"Locked" if self.is_locked else "Unlocked", "Locked" if self.is_locked else "Unlocked",
) )
@ -175,24 +175,24 @@ class device:
payload[0x07] = 0x31 payload[0x07] = 0x31
payload[0x08] = 0x31 payload[0x08] = 0x31
payload[0x09] = 0x31 payload[0x09] = 0x31
payload[0x0a] = 0x31 payload[0x0A] = 0x31
payload[0x0b] = 0x31 payload[0x0B] = 0x31
payload[0x0c] = 0x31 payload[0x0C] = 0x31
payload[0x0d] = 0x31 payload[0x0D] = 0x31
payload[0x0e] = 0x31 payload[0x0E] = 0x31
payload[0x0f] = 0x31 payload[0x0F] = 0x31
payload[0x10] = 0x31 payload[0x10] = 0x31
payload[0x11] = 0x31 payload[0x11] = 0x31
payload[0x12] = 0x31 payload[0x12] = 0x31
payload[0x1e] = 0x01 payload[0x1E] = 0x01
payload[0x2d] = 0x01 payload[0x2D] = 0x01
payload[0x30] = ord('T') payload[0x30] = ord("T")
payload[0x31] = ord('e') payload[0x31] = ord("e")
payload[0x32] = ord('s') payload[0x32] = ord("s")
payload[0x33] = ord('t') payload[0x33] = ord("t")
payload[0x34] = ord(' ') payload[0x34] = ord(" ")
payload[0x35] = ord(' ') payload[0x35] = ord(" ")
payload[0x36] = ord('1') payload[0x36] = ord("1")
response = self.send_packet(0x65, payload) response = self.send_packet(0x65, payload)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
@ -233,7 +233,7 @@ class device:
def get_fwversion(self) -> int: def get_fwversion(self) -> int:
"""Get firmware version.""" """Get firmware version."""
packet = bytearray([0x68]) packet = bytearray([0x68])
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return payload[0x4] | payload[0x5] << 8 return payload[0x4] | payload[0x5] << 8
@ -241,20 +241,20 @@ class device:
def set_name(self, name: str) -> None: def set_name(self, name: str) -> None:
"""Set device name.""" """Set device name."""
packet = bytearray(4) packet = bytearray(4)
packet += name.encode('utf-8') packet += name.encode("utf-8")
packet += bytearray(0x50 - len(packet)) packet += bytearray(0x50 - len(packet))
packet[0x43] = bool(self.is_locked) packet[0x43] = bool(self.is_locked)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
self.name = name self.name = name
def set_lock(self, state: bool) -> None: def set_lock(self, state: bool) -> None:
"""Lock/unlock the device.""" """Lock/unlock the device."""
packet = bytearray(4) packet = bytearray(4)
packet += self.name.encode('utf-8') packet += self.name.encode("utf-8")
packet += bytearray(0x50 - len(packet)) packet += bytearray(0x50 - len(packet))
packet[0x43] = bool(state) packet[0x43] = bool(state)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
self.is_locked = bool(state) self.is_locked = bool(state)
@ -264,27 +264,27 @@ class device:
def send_packet(self, command: int, payload: bytes) -> bytes: def send_packet(self, command: int, payload: bytes) -> bytes:
"""Send a packet to the device.""" """Send a packet to the device."""
self.count = (self.count + 1) & 0xffff self.count = (self.count + 1) & 0xFFFF
packet = bytearray(0x38) packet = bytearray(0x38)
packet[0x00] = 0x5a packet[0x00] = 0x5A
packet[0x01] = 0xa5 packet[0x01] = 0xA5
packet[0x02] = 0xaa packet[0x02] = 0xAA
packet[0x03] = 0x55 packet[0x03] = 0x55
packet[0x04] = 0x5a packet[0x04] = 0x5A
packet[0x05] = 0xa5 packet[0x05] = 0xA5
packet[0x06] = 0xaa packet[0x06] = 0xAA
packet[0x07] = 0x55 packet[0x07] = 0x55
packet[0x24] = self.devtype & 0xff packet[0x24] = self.devtype & 0xFF
packet[0x25] = self.devtype >> 8 packet[0x25] = self.devtype >> 8
packet[0x26] = command packet[0x26] = command
packet[0x28] = self.count & 0xff packet[0x28] = self.count & 0xFF
packet[0x29] = self.count >> 8 packet[0x29] = self.count >> 8
packet[0x2a] = self.mac[5] packet[0x2A] = self.mac[5]
packet[0x2b] = self.mac[4] packet[0x2B] = self.mac[4]
packet[0x2c] = self.mac[3] packet[0x2C] = self.mac[3]
packet[0x2d] = self.mac[2] packet[0x2D] = self.mac[2]
packet[0x2e] = self.mac[1] packet[0x2E] = self.mac[1]
packet[0x2f] = self.mac[0] packet[0x2F] = self.mac[0]
packet[0x30] = self.id[3] packet[0x30] = self.id[3]
packet[0x31] = self.id[2] packet[0x31] = self.id[2]
packet[0x32] = self.id[1] packet[0x32] = self.id[1]
@ -296,16 +296,16 @@ class device:
payload = bytearray(payload) payload = bytearray(payload)
payload += bytearray(padding) payload += bytearray(padding)
checksum = sum(payload, 0xbeaf) & 0xffff checksum = sum(payload, 0xBEAF) & 0xFFFF
packet[0x34] = checksum & 0xff packet[0x34] = checksum & 0xFF
packet[0x35] = checksum >> 8 packet[0x35] = checksum >> 8
payload = self.encrypt(payload) payload = self.encrypt(payload)
for i in range(len(payload)): for i in range(len(payload)):
packet.append(payload[i]) packet.append(payload[i])
checksum = sum(packet, 0xbeaf) & 0xffff checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x20] = checksum & 0xff packet[0x20] = checksum & 0xFF
packet[0x21] = checksum >> 8 packet[0x21] = checksum >> 8
start_time = time.time() start_time = time.time()
@ -329,7 +329,7 @@ class device:
raise exception(-4007) # Length error. raise exception(-4007) # Length error.
checksum = resp[0x20] | (resp[0x21] << 8) checksum = resp[0x20] | (resp[0x21] << 8)
if sum(resp, 0xbeaf) - sum(resp[0x20:0x22]) & 0xffff != checksum: if sum(resp, 0xBEAF) - sum(resp[0x20:0x22]) & 0xFFFF != checksum:
raise exception(-4008) # Checksum error. raise exception(-4008) # Checksum error.
return resp return resp

52
broadlink/light.py
View File

@ -11,14 +11,14 @@ class lb1(device):
state_dict = [] state_dict = []
effect_map_dict = { effect_map_dict = {
'lovely color': 0, "lovely color": 0,
'flashlight': 1, "flashlight": 1,
'lightning': 2, "lightning": 2,
'color fading': 3, "color fading": 3,
'color breathing': 4, "color breathing": 4,
'multicolor breathing': 5, "multicolor breathing": 5,
'color jumping': 6, "color jumping": 6,
'multicolor jumping': 7, "multicolor jumping": 7,
} }
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args, **kwargs) -> None:
@ -26,36 +26,38 @@ class lb1(device):
device.__init__(self, *args, **kwargs) device.__init__(self, *args, **kwargs)
self.type = "SmartBulb" self.type = "SmartBulb"
def send_command(self, command: str, type: str = 'set') -> None: def send_command(self, command: str, type: str = "set") -> None:
"""Send a command to the device.""" """Send a command to the device."""
packet = bytearray(16+(int(len(command)/16) + 1)*16) packet = bytearray(16 + (int(len(command) / 16) + 1) * 16)
packet[0x00] = 0x0c + len(command) & 0xff packet[0x00] = 0x0C + len(command) & 0xFF
packet[0x02] = 0xa5 packet[0x02] = 0xA5
packet[0x03] = 0xa5 packet[0x03] = 0xA5
packet[0x04] = 0x5a packet[0x04] = 0x5A
packet[0x05] = 0x5a packet[0x05] = 0x5A
packet[0x08] = 0x02 if type == "set" else 0x01 # 0x01 => query, # 0x02 => set packet[0x08] = 0x02 if type == "set" else 0x01 # 0x01 => query, # 0x02 => set
packet[0x09] = 0x0b packet[0x09] = 0x0B
packet[0x0a] = len(command) packet[0x0A] = len(command)
packet[0x0e:] = map(ord, command) packet[0x0E:] = map(ord, command)
checksum = sum(packet, 0xbeaf) & 0xffff checksum = sum(packet, 0xBEAF) & 0xFFFF
packet[0x06] = checksum & 0xff # Checksum 1 position packet[0x06] = checksum & 0xFF # Checksum 1 position
packet[0x07] = checksum >> 8 # Checksum 2 position packet[0x07] = checksum >> 8 # Checksum 2 position
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x36:0x38]) check_error(response[0x36:0x38])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
responseLength = int(payload[0x0a]) | (int(payload[0x0b]) << 8) responseLength = int(payload[0x0A]) | (int(payload[0x0B]) << 8)
if responseLength > 0: if responseLength > 0:
self.state_dict = json.loads(payload[0x0e:0x0e+responseLength]) self.state_dict = json.loads(payload[0x0E : 0x0E + responseLength])
def set_json(self, jsonstr: str) -> str: def set_json(self, jsonstr: str) -> str:
"""Send a command to the device and return state.""" """Send a command to the device and return state."""
reconvert = json.loads(jsonstr) reconvert = json.loads(jsonstr)
if 'bulb_sceneidx' in reconvert.keys(): if "bulb_sceneidx" in reconvert.keys():
reconvert['bulb_sceneidx'] = self.effect_map_dict.get(reconvert['bulb_sceneidx'], 255) reconvert["bulb_sceneidx"] = self.effect_map_dict.get(
reconvert["bulb_sceneidx"], 255
)
self.send_command(json.dumps(reconvert)) self.send_command(json.dumps(reconvert))
return json.dumps(self.state_dict) return json.dumps(self.state_dict)

36
broadlink/remote.py
View File

@ -17,45 +17,45 @@ class rm(device):
"""Return the last captured code.""" """Return the last captured code."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x04) packet.append(0x04)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return payload[len(self._request_header) + 4:] return payload[len(self._request_header) + 4 :]
def send_data(self, data: bytes) -> None: def send_data(self, data: bytes) -> None:
"""Send a code to the device.""" """Send a code to the device."""
packet = bytearray(self._code_sending_header) packet = bytearray(self._code_sending_header)
packet += bytearray([0x02, 0x00, 0x00, 0x00]) packet += bytearray([0x02, 0x00, 0x00, 0x00])
packet += data packet += data
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def enter_learning(self) -> None: def enter_learning(self) -> None:
"""Enter infrared learning mode.""" """Enter infrared learning mode."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x03) packet.append(0x03)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def sweep_frequency(self) -> None: def sweep_frequency(self) -> None:
"""Sweep frequency.""" """Sweep frequency."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x19) packet.append(0x19)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def cancel_sweep_frequency(self) -> None: def cancel_sweep_frequency(self) -> None:
"""Cancel sweep frequency.""" """Cancel sweep frequency."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x1e) packet.append(0x1E)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def check_frequency(self) -> bool: def check_frequency(self) -> bool:
"""Return True if the frequency was identified successfully.""" """Return True if the frequency was identified successfully."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x1a) packet.append(0x1A)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
if payload[len(self._request_header) + 4] == 1: if payload[len(self._request_header) + 4] == 1:
@ -65,8 +65,8 @@ class rm(device):
def find_rf_packet(self) -> bool: def find_rf_packet(self) -> bool:
"""Enter radiofrequency learning mode.""" """Enter radiofrequency learning mode."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(0x1b) packet.append(0x1B)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
if payload[len(self._request_header) + 4] == 1: if payload[len(self._request_header) + 4] == 1:
@ -77,10 +77,10 @@ class rm(device):
"""Return the state of the sensors in raw format.""" """Return the state of the sensors in raw format."""
packet = bytearray(self._request_header) packet = bytearray(self._request_header)
packet.append(command) packet.append(command)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return bytearray(payload[len(self._request_header) + 4:]) return bytearray(payload[len(self._request_header) + 4 :])
def check_temperature(self) -> int: def check_temperature(self) -> int:
"""Return the temperature.""" """Return the temperature."""
@ -90,7 +90,7 @@ class rm(device):
def check_sensors(self) -> dict: def check_sensors(self) -> dict:
"""Return the state of the sensors.""" """Return the state of the sensors."""
data = self._check_sensors(0x1) data = self._check_sensors(0x1)
return {'temperature': data[0x0] + data[0x1] / 10.0} return {"temperature": data[0x0] + data[0x1] / 10.0}
class rm4(rm): class rm4(rm):
@ -100,8 +100,8 @@ class rm4(rm):
"""Initialize the controller.""" """Initialize the controller."""
device.__init__(self, *args, **kwargs) device.__init__(self, *args, **kwargs)
self.type = "RM4" self.type = "RM4"
self._request_header = b'\x04\x00' self._request_header = b"\x04\x00"
self._code_sending_header = b'\xda\x00' self._code_sending_header = b"\xda\x00"
def check_temperature(self) -> int: def check_temperature(self) -> int:
"""Return the temperature.""" """Return the temperature."""
@ -117,6 +117,6 @@ class rm4(rm):
"""Return the state of the sensors.""" """Return the state of the sensors."""
data = self._check_sensors(0x24) data = self._check_sensors(0x24)
return { return {
'temperature': data[0x0] + data[0x1] / 100.0, "temperature": data[0x0] + data[0x1] / 100.0,
'humidity': data[0x2] + data[0x3] / 100.0 "humidity": data[0x2] + data[0x3] / 100.0,
} }

20
broadlink/sensor.py
View File

@ -7,9 +7,9 @@ class a1(device):
"""Controls a Broadlink A1.""" """Controls a Broadlink A1."""
_SENSORS_AND_LEVELS = ( _SENSORS_AND_LEVELS = (
('light', ('dark', 'dim', 'normal', 'bright')), ("light", ("dark", "dim", "normal", "bright")),
('air_quality', ('excellent', 'good', 'normal', 'bad')), ("air_quality", ("excellent", "good", "normal", "bad")),
('noise', ('quiet', 'normal', 'noisy')), ("noise", ("quiet", "normal", "noisy")),
) )
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args, **kwargs) -> None:
@ -24,20 +24,20 @@ class a1(device):
try: try:
data[sensor] = levels[data[sensor]] data[sensor] = levels[data[sensor]]
except IndexError: except IndexError:
data[sensor] = 'unknown' data[sensor] = "unknown"
return data return data
def check_sensors_raw(self) -> dict: def check_sensors_raw(self) -> dict:
"""Return the state of the sensors in raw format.""" """Return the state of the sensors in raw format."""
packet = bytearray([0x1]) packet = bytearray([0x1])
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
data = bytearray(payload[0x4:]) data = bytearray(payload[0x4:])
return { return {
'temperature': data[0x0] + data[0x1] / 10.0, "temperature": data[0x0] + data[0x1] / 10.0,
'humidity': data[0x2] + data[0x3] / 10.0, "humidity": data[0x2] + data[0x3] / 10.0,
'light': data[0x4], "light": data[0x4],
'air_quality': data[0x6], "air_quality": data[0x6],
'noise': data[0x8], "noise": data[0x8],
} }

112
broadlink/switch.py
View File

@ -17,19 +17,19 @@ class mp1(device):
def set_power_mask(self, sid_mask: int, state: bool) -> None: def set_power_mask(self, sid_mask: int, state: bool) -> None:
"""Set the power state of the device.""" """Set the power state of the device."""
packet = bytearray(16) packet = bytearray(16)
packet[0x00] = 0x0d packet[0x00] = 0x0D
packet[0x02] = 0xa5 packet[0x02] = 0xA5
packet[0x03] = 0xa5 packet[0x03] = 0xA5
packet[0x04] = 0x5a packet[0x04] = 0x5A
packet[0x05] = 0x5a packet[0x05] = 0x5A
packet[0x06] = 0xb2 + ((sid_mask << 1) if state else sid_mask) packet[0x06] = 0xB2 + ((sid_mask << 1) if state else sid_mask)
packet[0x07] = 0xc0 packet[0x07] = 0xC0
packet[0x08] = 0x02 packet[0x08] = 0x02
packet[0x0a] = 0x03 packet[0x0A] = 0x03
packet[0x0d] = sid_mask packet[0x0D] = sid_mask
packet[0x0e] = sid_mask if state else 0 packet[0x0E] = sid_mask if state else 0
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def set_power(self, sid: int, state: bool) -> None: def set_power(self, sid: int, state: bool) -> None:
@ -40,30 +40,30 @@ class mp1(device):
def check_power_raw(self) -> bool: def check_power_raw(self) -> bool:
"""Return the power state of the device in raw format.""" """Return the power state of the device in raw format."""
packet = bytearray(16) packet = bytearray(16)
packet[0x00] = 0x0a packet[0x00] = 0x0A
packet[0x02] = 0xa5 packet[0x02] = 0xA5
packet[0x03] = 0xa5 packet[0x03] = 0xA5
packet[0x04] = 0x5a packet[0x04] = 0x5A
packet[0x05] = 0x5a packet[0x05] = 0x5A
packet[0x06] = 0xae packet[0x06] = 0xAE
packet[0x07] = 0xc0 packet[0x07] = 0xC0
packet[0x08] = 0x01 packet[0x08] = 0x01
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return payload[0x0e] return payload[0x0E]
def check_power(self) -> dict: def check_power(self) -> dict:
"""Return the power state of the device.""" """Return the power state of the device."""
state = self.check_power_raw() state = self.check_power_raw()
if state is None: if state is None:
return {'s1': None, 's2': None, 's3': None, 's4': None} return {"s1": None, "s2": None, "s3": None, "s4": None}
data = {} data = {}
data['s1'] = bool(state & 0x01) data["s1"] = bool(state & 0x01)
data['s2'] = bool(state & 0x02) data["s2"] = bool(state & 0x02)
data['s3'] = bool(state & 0x04) data["s3"] = bool(state & 0x04)
data['s4'] = bool(state & 0x08) data["s4"] = bool(state & 0x08)
return data return data
@ -80,8 +80,8 @@ class bg1(device):
Example: `{"pwr":1,"pwr1":1,"pwr2":0,"maxworktime":60,"maxworktime1":60,"maxworktime2":0,"idcbrightness":50}` Example: `{"pwr":1,"pwr1":1,"pwr2":0,"maxworktime":60,"maxworktime1":60,"maxworktime2":0,"idcbrightness":50}`
""" """
packet = self._encode(1, b'{}') packet = self._encode(1, b"{}")
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
return self._decode(response) return self._decode(response)
@ -98,22 +98,22 @@ class bg1(device):
"""Set the power state of the device.""" """Set the power state of the device."""
data = {} data = {}
if pwr is not None: if pwr is not None:
data['pwr'] = int(bool(pwr)) data["pwr"] = int(bool(pwr))
if pwr1 is not None: if pwr1 is not None:
data['pwr1'] = int(bool(pwr1)) data["pwr1"] = int(bool(pwr1))
if pwr2 is not None: if pwr2 is not None:
data['pwr2'] = int(bool(pwr2)) data["pwr2"] = int(bool(pwr2))
if maxworktime is not None: if maxworktime is not None:
data['maxworktime'] = maxworktime data["maxworktime"] = maxworktime
if maxworktime1 is not None: if maxworktime1 is not None:
data['maxworktime1'] = maxworktime1 data["maxworktime1"] = maxworktime1
if maxworktime2 is not None: if maxworktime2 is not None:
data['maxworktime2'] = maxworktime2 data["maxworktime2"] = maxworktime2
if idcbrightness is not None: if idcbrightness is not None:
data['idcbrightness'] = idcbrightness data["idcbrightness"] = idcbrightness
js = json.dumps(data).encode('utf8') js = json.dumps(data).encode("utf8")
packet = self._encode(2, js) packet = self._encode(2, js)
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
return self._decode(response) return self._decode(response)
@ -129,20 +129,22 @@ class bg1(device):
# 0x0e- json data # 0x0e- json data
packet = bytearray(14) packet = bytearray(14)
length = 4 + 2 + 2 + 4 + len(js) length = 4 + 2 + 2 + 4 + len(js)
struct.pack_into('<HHHHBBI', packet, 0, length, 0xa5a5, 0x5a5a, 0x0000, flag, 0x0b, len(js)) struct.pack_into(
"<HHHHBBI", packet, 0, length, 0xA5A5, 0x5A5A, 0x0000, flag, 0x0B, len(js)
)
for i in range(len(js)): for i in range(len(js)):
packet.append(js[i]) packet.append(js[i])
checksum = sum(packet[0x08:], 0xc0ad) & 0xffff checksum = sum(packet[0x08:], 0xC0AD) & 0xFFFF
packet[0x06] = checksum & 0xff packet[0x06] = checksum & 0xFF
packet[0x07] = checksum >> 8 packet[0x07] = checksum >> 8
return packet return packet
def _decode(self, response: bytes) -> dict: def _decode(self, response: bytes) -> dict:
"""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
@ -178,7 +180,7 @@ class sp2(device):
packet[4] = 3 if state else 2 packet[4] = 3 if state else 2
else: else:
packet[4] = 1 if state else 0 packet[4] = 1 if state else 0
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def set_nightlight(self, state: bool) -> None: def set_nightlight(self, state: bool) -> None:
@ -189,14 +191,14 @@ class sp2(device):
packet[4] = 3 if state else 1 packet[4] = 3 if state else 1
else: else:
packet[4] = 2 if state else 0 packet[4] = 2 if state else 0
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
def check_power(self) -> bool: def check_power(self) -> bool:
"""Return the power state of the device.""" """Return the power state of the device."""
packet = bytearray(16) packet = bytearray(16)
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return bool(payload[0x4] == 1 or payload[0x4] == 3 or payload[0x4] == 0xFD) return bool(payload[0x4] == 1 or payload[0x4] == 3 or payload[0x4] == 0xFD)
@ -205,7 +207,7 @@ class sp2(device):
"""Return the state of the night light.""" """Return the state of the night light."""
packet = bytearray(16) packet = bytearray(16)
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return bool(payload[0x4] == 2 or payload[0x4] == 3 or payload[0x4] == 0xFF) return bool(payload[0x4] == 2 or payload[0x4] == 3 or payload[0x4] == 0xFF)
@ -213,10 +215,13 @@ class sp2(device):
def get_energy(self) -> int: def get_energy(self) -> int:
"""Return the energy state of the device.""" """Return the energy state of the device."""
packet = bytearray([8, 0, 254, 1, 5, 1, 0, 0, 0, 45]) packet = bytearray([8, 0, 254, 1, 5, 1, 0, 0, 0, 45])
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6A, packet)
check_error(response[0x22:0x24]) check_error(response[0x22:0x24])
payload = self.decrypt(response[0x38:]) payload = self.decrypt(response[0x38:])
return int(hex(payload[0x07] * 256 + payload[0x06])[2:]) + int(hex(payload[0x05])[2:]) / 100.0 return (
int(hex(payload[0x07] * 256 + payload[0x06])[2:])
+ int(hex(payload[0x05])[2:]) / 100.0
)
class sp4(device): class sp4(device):
@ -314,7 +319,18 @@ class sp4b(sp4):
payload = json.dumps(state, separators=(",", ":")).encode() payload = json.dumps(state, separators=(",", ":")).encode()
packet = bytearray(14) packet = bytearray(14)
length = 4 + 2 + 2 + 4 + len(payload) length = 4 + 2 + 2 + 4 + len(payload)
struct.pack_into('<HHHHBBI', packet, 0, length, 0xA5A5, 0x5A5A, 0x0000, flag, 0x0B, len(payload)) struct.pack_into(
"<HHHHBBI",
packet,
0,
length,
0xA5A5,
0x5A5A,
0x0000,
flag,
0x0B,
len(payload),
)
packet.extend(payload) packet.extend(payload)
checksum = sum(packet[0x8:], 0xC0AD) & 0xFFFF checksum = sum(packet[0x8:], 0xC0AD) & 0xFFFF
packet[0x06] = checksum & 0xFF packet[0x06] = checksum & 0xFF

22
setup.py
View File

@ -8,21 +8,21 @@ from setuptools import setup, find_packages
version = '0.16.0' version = '0.16.0'
setup( setup(
name='broadlink', name="broadlink",
version=version, version=version,
author='Matthew Garrett', author="Matthew Garrett",
author_email='mjg59@srcf.ucam.org', author_email="mjg59@srcf.ucam.org",
url='http://github.com/mjg59/python-broadlink', url="http://github.com/mjg59/python-broadlink",
packages=find_packages(), packages=find_packages(),
scripts=[], scripts=[],
install_requires=['cryptography>=2.1.1'], install_requires=["cryptography>=2.1.1"],
description='Python API for controlling Broadlink IR controllers', description="Python API for controlling Broadlink IR controllers",
classifiers=[ classifiers=[
'Development Status :: 4 - Beta', "Development Status :: 4 - Beta",
'Intended Audience :: Developers', "Intended Audience :: Developers",
'License :: OSI Approved :: MIT License', "License :: OSI Approved :: MIT License",
'Operating System :: OS Independent', "Operating System :: OS Independent",
'Programming Language :: Python', "Programming Language :: Python",
], ],
include_package_data=True, include_package_data=True,
zip_safe=False, zip_safe=False,