maurice/snix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

code clean up (#243)

Browse source
This commit is contained in:
Daniel Høyer Iversen 2019-05-19 17:54:14 +02:00 • committed by GitHub
parent c9a1c106a7
commit a75f98720e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 774 additions and 770 deletions
Download patch file Download diff file Expand all files Collapse all files

132
broadlink/__init__.py
View file

@ -1,6 +1,7 @@
#!/usr/bin/python #!/usr/bin/python
from datetime import datetime from datetime import datetime
try: try:
from Crypto.Cipher import AES from Crypto.Cipher import AES
except ImportError as e: except ImportError as e:
@ -9,7 +10,6 @@ except ImportError as e:
import time import time
import random import random
import socket import socket
import sys
import threading import threading
import codecs import codecs
@ -60,6 +60,7 @@ def gendevice(devtype, host, mac):
return device(host=host, mac=mac, devtype=devtype) return device(host=host, mac=mac, devtype=devtype)
return deviceClass(host=host, mac=mac, devtype=devtype) return deviceClass(host=host, mac=mac, devtype=devtype)
def discover(timeout=None, local_ip_address=None): def discover(timeout=None, local_ip_address=None):
if local_ip_address is None: if local_ip_address is None:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@ -122,9 +123,8 @@ def discover(timeout=None, local_ip_address=None):
mac = responsepacket[0x3a:0x40] mac = responsepacket[0x3a:0x40]
devtype = responsepacket[0x34] | responsepacket[0x35] << 8 devtype = responsepacket[0x34] | responsepacket[0x35] << 8
return gendevice(devtype, host, mac) return gendevice(devtype, host, mac)
else:
while (time.time() - starttime) < timeout: while (time.time() - starttime) < timeout:
cs.settimeout(timeout - (time.time() - starttime)) cs.settimeout(timeout - (time.time() - starttime))
try: try:
@ -140,7 +140,6 @@ def discover(timeout=None, local_ip_address=None):
return devices return devices
class device: class device:
def __init__(self, host, mac, devtype, timeout=10): def __init__(self, host, mac, devtype, timeout=10):
self.host = host self.host = host
@ -148,8 +147,10 @@ class device:
self.devtype = devtype self.devtype = devtype
self.timeout = timeout self.timeout = timeout
self.count = random.randrange(0xffff) self.count = random.randrange(0xffff)
self.key = bytearray([0x09, 0x76, 0x28, 0x34, 0x3f, 0xe9, 0x9e, 0x23, 0x76, 0x5c, 0x15, 0x13, 0xac, 0xcf, 0x8b, 0x02]) self.key = bytearray(
self.iv = bytearray([0x56, 0x2e, 0x17, 0x99, 0x6d, 0x09, 0x3d, 0x28, 0xdd, 0xb3, 0xba, 0x69, 0x5a, 0x2e, 0x6f, 0x58]) [0x09, 0x76, 0x28, 0x34, 0x3f, 0xe9, 0x9e, 0x23, 0x76, 0x5c, 0x15, 0x13, 0xac, 0xcf, 0x8b, 0x02])
self.iv = bytearray(
[0x56, 0x2e, 0x17, 0x99, 0x6d, 0x09, 0x3d, 0x28, 0xdd, 0xb3, 0xba, 0x69, 0x5a, 0x2e, 0x6f, 0x58])
self.id = bytearray([0, 0, 0, 0]) self.id = bytearray([0, 0, 0, 0])
self.cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@ -255,7 +256,7 @@ class device:
packet[0x33] = self.id[3] packet[0x33] = self.id[3]
# pad the payload for AES encryption # pad the payload for AES encryption
if len(payload)>0: if payload:
numpad = (len(payload) // 16 + 1) * 16 numpad = (len(payload) // 16 + 1) * 16
payload = payload.ljust(numpad, b"\x00") payload = payload.ljust(numpad, b"\x00")
@ -314,9 +315,7 @@ class mp1(device):
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) self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8)
def set_power(self, sid, state): def set_power(self, sid, state):
"""Sets the power state of the smart power strip.""" """Sets the power state of the smart power strip."""
@ -337,9 +336,10 @@ class mp1(device):
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
state = payload[0x0e] state = payload[0x0e]
else: else:
state = ord(payload[0x0e]) state = ord(payload[0x0e])
@ -398,19 +398,12 @@ class sp2(device):
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
if payload[0x4] == 1 or payload[0x4] == 3 or payload[0x4] == 0xFD: return bool(payload[0x4] == 1 or payload[0x4] == 3 or payload[0x4] == 0xFD)
state = True return bool(ord(payload[0x4]) == 1 or ord(payload[0x4]) == 3 or ord(payload[0x4]) == 0xFD)
else:
state = False
else:
if ord(payload[0x4]) == 1 or ord(payload[0x4]) == 3 or ord(payload[0x4]) == 0xFD:
state = True
else:
state = False
return state
def check_nightlight(self): def check_nightlight(self):
"""Returns the power state of the smart plug.""" """Returns the power state of the smart plug."""
@ -418,30 +411,25 @@ class sp2(device):
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
if payload[0x4] == 2 or payload[0x4] == 3 or payload[0x4] == 0xFF: return bool(payload[0x4] == 2 or payload[0x4] == 3 or payload[0x4] == 0xFF)
state = True return bool(ord(payload[0x4]) == 2 or ord(payload[0x4]) == 3 or ord(payload[0x4]) == 0xFF)
else:
state = False
else:
if ord(payload[0x4]) == 2 or ord(payload[0x4]) == 3 or ord(payload[0x4]) == 0xFF:
state = True
else:
state = False
return state
def get_energy(self): def get_energy(self):
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)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x07]) == int: if isinstance(payload[0x7], int):
energy = int(hex(payload[0x07] * 256 + payload[0x06])[2:]) + int(hex(payload[0x05])[2:]) / 100.0 energy = int(hex(payload[0x07] * 256 + payload[0x06])[2:]) + int(hex(payload[0x05])[2:]) / 100.0
else: else:
energy = int(hex(ord(payload[0x07]) * 256 + ord(payload[0x06]))[2:]) + int(hex(ord(payload[0x05]))[2:])/100.0 energy = int(hex(ord(payload[0x07]) * 256 + ord(payload[0x06]))[2:]) + int(
hex(ord(payload[0x05]))[2:]) / 100.0
return energy return energy
@ -455,10 +443,11 @@ class a1(device):
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
data = {} data = {}
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0 data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0
data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0 data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0
light = payload[0x8] light = payload[0x8]
@ -505,10 +494,11 @@ class a1(device):
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
data = {} data = {}
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0 data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0
data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0 data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0
data['light'] = payload[0x8] data['light'] = payload[0x8]
@ -533,7 +523,8 @@ class rm(device):
packet[0] = 4 packet[0] = 4
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
return payload[0x04:] return payload[0x04:]
@ -562,7 +553,8 @@ class rm(device):
packet[0] = 0x1a packet[0] = 0x1a
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return False
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if payload[0x04] == 1: if payload[0x04] == 1:
return True return True
@ -573,7 +565,8 @@ class rm(device):
packet[0] = 0x1b packet[0] = 0x1b
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return False
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if payload[0x04] == 1: if payload[0x04] == 1:
return True return True
@ -584,9 +577,10 @@ class rm(device):
packet[0] = 1 packet[0] = 1
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return False
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
if type(payload[0x4]) == int: if isinstance(payload[0x4], int):
temp = (payload[0x4] * 10 + payload[0x5]) / 10.0 temp = (payload[0x4] * 10 + payload[0x5]) / 10.0
else: else:
temp = (ord(payload[0x4]) * 10 + ord(payload[0x5])) / 10.0 temp = (ord(payload[0x4]) * 10 + ord(payload[0x5])) / 10.0
@ -642,12 +636,11 @@ class hysen(device):
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 = CRC16(modbus_flag=True).calculate(bytes(response_payload[2:response_payload_len])) crc = CRC16(modbus_flag=True).calculate(bytes(response_payload[2:response_payload_len]))
if (response_payload[response_payload_len] == crc & 0xFF) and (response_payload[response_payload_len+1] == (crc >> 8) & 0xFF): if (response_payload[response_payload_len] == crc & 0xFF) and (
response_payload[response_payload_len + 1] == (crc >> 8) & 0xFF):
return response_payload[2:response_payload_len] return response_payload[2:response_payload_len]
else:
raise ValueError('hysen_response_error', 'CRC check on response failed') raise ValueError('hysen_response_error', 'CRC check on response failed')
# Get current room temperature in degrees celsius # Get current room temperature in degrees celsius
def get_temp(self): def get_temp(self):
payload = self.send_request(bytearray([0x01, 0x03, 0x00, 0x00, 0x00, 0x08])) payload = self.send_request(bytearray([0x01, 0x03, 0x00, 0x00, 0x00, 0x08]))
@ -689,19 +682,22 @@ class hysen(device):
weekday = [] weekday = []
for i in range(0, 6): for i in range(0, 6):
weekday.append({'start_hour':payload[2*i + 23], 'start_minute':payload[2*i + 24],'temp':payload[i + 39]/2.0}) weekday.append(
{'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({'start_hour':payload[2*i + 23], 'start_minute':payload[2*i + 24],'temp':payload[i + 39]/2.0}) weekend.append(
{'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
# auto_mode = 1 for auto (scheduled/timed) mode, 0 for manual mode. # auto_mode = 1 for auto (scheduled/timed) mode, 0 for manual mode.
# Manual mode will activate last used temperature. In typical usage call set_temp to activate manual control and set temp. # Manual mode will activate last used temperature.
# In typical usage call set_temp to activate manual control and set temp.
# loop_mode refers to index in [ "12345,67", "123456,7", "1234567" ] # loop_mode refers to index in [ "12345,67", "123456,7", "1234567" ]
# E.g. loop_mode = 0 ("12345,67") means Saturday and Sunday follow the "weekend" schedule # E.g. loop_mode = 0 ("12345,67") means Saturday and Sunday follow the "weekend" schedule
# loop_mode = 2 ("1234567") means every day (including Saturday and Sunday) follows the "weekday" schedule # loop_mode = 2 ("1234567") means every day (including Saturday and Sunday) follows the "weekday" schedule
@ -712,19 +708,23 @@ class hysen(device):
self.send_request(bytearray([0x01, 0x06, 0x00, 0x02, mode_byte, sensor])) self.send_request(bytearray([0x01, 0x06, 0x00, 0x02, mode_byte, sensor]))
# Advanced settings # Advanced settings
# Sensor mode (SEN) sensor = 0 for internal sensor, 1 for external sensor, 2 for internal control temperature, external limit temperature. Factory default: 0. # Sensor mode (SEN) sensor = 0 for internal sensor, 1 for external sensor,
# 2 for internal control temperature, external limit temperature. Factory default: 0.
# Set temperature range for external sensor (OSV) osv = 5..99. Factory default: 42C # Set temperature range for external sensor (OSV) osv = 5..99. Factory default: 42C
# Deadzone for floor temprature (dIF) dif = 1..9. Factory default: 2C # Deadzone for floor temprature (dIF) dif = 1..9. Factory default: 2C
# Upper temperature limit for internal sensor (SVH) svh = 5..99. Factory default: 35C # Upper temperature limit for internal sensor (SVH) svh = 5..99. Factory default: 35C
# Lower temperature limit for internal sensor (SVL) svl = 5..99. Factory default: 5C # Lower temperature limit for internal sensor (SVL) svl = 5..99. Factory default: 5C
# Actual temperature calibration (AdJ) adj = -0.5. Prescision 0.1C # Actual temperature calibration (AdJ) adj = -0.5. Prescision 0.1C
# Anti-freezing function (FrE) fre = 0 for anti-freezing function shut down, 1 for anti-freezing function open. Factory default: 0 # 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 power on memory off, 1 for power on memory on. Factory default: 0 # Power on memory (POn) poweron = 0 for power on memory off, 1 for power on memory on. Factory default: 0
def set_advanced(self, loop_mode, sensor, osv, dif, svh, svl, adj, fre, poweron): def set_advanced(self, loop_mode, sensor, osv, dif, svh, svl, adj, fre, poweron):
input_payload = bytearray([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]) input_payload = bytearray([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. Note this function invokes loop_mode=0 and sensor=0. # For backwards compatibility only. Prefer calling set_mode directly.
# Note this function invokes loop_mode=0 and sensor=0.
def switch_to_auto(self): def switch_to_auto(self):
self.set_mode(auto_mode=1, loop_mode=0) self.set_mode(auto_mode=1, loop_mode=0)
@ -735,7 +735,8 @@ class hysen(device):
def set_temp(self, temp): def set_temp(self, temp):
self.send_request(bytearray([0x01, 0x06, 0x00, 0x01, 0x00, int(temp * 2)])) self.send_request(bytearray([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. # Set device on(1) or off(0), does not deactivate Wifi connectivity.
# Remote lock disables control by buttons on thermostat.
def set_power(self, power=1, remote_lock=0): def set_power(self, power=1, remote_lock=0):
self.send_request(bytearray([0x01, 0x06, 0x00, 0x00, remote_lock, power])) self.send_request(bytearray([0x01, 0x06, 0x00, 0x00, remote_lock, power]))
@ -787,6 +788,7 @@ class S1C(device):
""" """
Its VERY VERY VERY DIRTY IMPLEMENTATION of S1C Its VERY VERY VERY DIRTY IMPLEMENTATION of S1C
""" """
def __init__(self, *a, **kw): def __init__(self, *a, **kw):
device.__init__(self, *a, **kw) device.__init__(self, *a, **kw)
self.type = 'S1C' self.type = 'S1C'
@ -796,13 +798,14 @@ class S1C(device):
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)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
payload = aes.decrypt(bytes(response[0x38:])) payload = aes.decrypt(bytes(response[0x38:]))
if payload: if not payload:
head = payload[:4] return None
count = payload[0x4] #need to fix for python 2.x count = payload[0x4]
sensors = payload[0x6:] sensors = payload[0x6:]
sensors_a = [bytearray(sensors[i * 83:(i + 1) * 83]) for i in range(len(sensors) // 83)] sensors_a = [bytearray(sensors[i * 83:(i + 1) * 83]) for i in range(len(sensors) // 83)]
@ -847,7 +850,8 @@ class dooya(device):
packet[10] = 0x44 packet[10] = 0x44
response = self.send_packet(0x6a, packet) response = self.send_packet(0x6a, packet)
err = response[0x22] | (response[0x23] << 8) err = response[0x22] | (response[0x23] << 8)
if err == 0: if err != 0:
return None
payload = self.decrypt(bytes(response[0x38:])) payload = self.decrypt(bytes(response[0x38:]))
return ord(payload[4]) return ord(payload[4])