mirror of
https://github.com/mjg59/python-broadlink.git
synced 2024-09-21 04:20:36 +02:00
d490c5b71e
Look at the device ID and figure out what type of device it is, and provide appropriate classes for each that only support the functionality available on that device.
338 lines
10 KiB
Python
Executable File
338 lines
10 KiB
Python
Executable File
#!/usr/bin/python
|
|
|
|
from datetime import datetime
|
|
from Crypto.Cipher import AES
|
|
import time
|
|
import random
|
|
import socket
|
|
|
|
def discover(timeout=None):
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.connect(('8.8.8.8', 53)) # connecting to a UDP address doesn't send packets
|
|
local_ip_address = s.getsockname()[0]
|
|
address = local_ip_address.split('.')
|
|
cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
cs.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
cs.bind(('',0))
|
|
port = cs.getsockname()[1]
|
|
starttime = time.time()
|
|
|
|
devices = []
|
|
|
|
timezone = time.timezone/-3600
|
|
packet = bytearray(0x30)
|
|
|
|
year = datetime.now().year
|
|
|
|
if timezone < 0:
|
|
packet[0x08] = 0xff + timezone - 1
|
|
packet[0x09] = 0xff
|
|
packet[0x0a] = 0xff
|
|
packet[0x0b] = 0xff
|
|
else:
|
|
packet[0x08] = timezone
|
|
packet[0x09] = 0
|
|
packet[0x0a] = 0
|
|
packet[0x0b] = 0
|
|
packet[0x0c] = year & 0xff
|
|
packet[0x0d] = year >> 8
|
|
packet[0x0e] = datetime.now().minute
|
|
packet[0x0f] = datetime.now().hour
|
|
subyear = str(year)[2:]
|
|
packet[0x10] = int(subyear)
|
|
packet[0x11] = datetime.now().isoweekday()
|
|
packet[0x12] = datetime.now().day
|
|
packet[0x13] = datetime.now().month
|
|
packet[0x18] = int(address[0])
|
|
packet[0x19] = int(address[1])
|
|
packet[0x1a] = int(address[2])
|
|
packet[0x1b] = int(address[3])
|
|
packet[0x1c] = port & 0xff
|
|
packet[0x1d] = port >> 8
|
|
packet[0x26] = 6
|
|
checksum = 0xbeaf
|
|
|
|
for i in range(len(packet)):
|
|
checksum += packet[i]
|
|
checksum = checksum & 0xffff
|
|
packet[0x20] = checksum & 0xff
|
|
packet[0x21] = checksum >> 8
|
|
|
|
cs.sendto(packet, ('255.255.255.255', 80))
|
|
if timeout is None:
|
|
response = cs.recvfrom(1024)
|
|
responsepacket = bytearray(response[0])
|
|
host = response[1]
|
|
mac = responsepacket[0x3a:0x40]
|
|
return device(host=host, mac=mac)
|
|
else:
|
|
while (time.time() - starttime) < timeout:
|
|
cs.settimeout(timeout - (time.time() - starttime))
|
|
try:
|
|
response = cs.recvfrom(1024)
|
|
except socket.timeout:
|
|
return devices
|
|
responsepacket = bytearray(response[0])
|
|
host = response[1]
|
|
devtype = responsepacket[0x34] | responsepacket[0x35] << 8
|
|
mac = responsepacket[0x3a:0x40]
|
|
if devtype == 0: # SP1
|
|
devices.append(sp1(host=host, mac=mac))
|
|
if devtype == 0x2711: # SP2
|
|
devices.append(sp2(host=host, mac=mac))
|
|
if devtype == 0x2719 or devtype == 0x7919 or devtype == 0x271a or devtype == 0x791a: # Honeywell SP2
|
|
devices.append(sp2(host=host, mac=mac))
|
|
if devtype == 0x2720: # SPMini
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype == 0x753e: # SP3
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype == 0x2728: # SPMini2
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype == 0x2733 or devtype == 0x273e: # OEM branded SPMini
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype >= 0x7530 and devtype <= 0x7918: # OEM branded SPMini2
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype == 0x2736: # SPMiniPlus
|
|
devices.append(sp2(host=host, mac=mac))
|
|
elif devtype == 0x2712: # RM2
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x2737: # RM Mini
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x273d: # RM Pro Phicomm
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x2783: # RM2 Home Plus
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x277c: # RM2 Home Plus GDT
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x272a: # RM2 Pro Plus
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x2787: # RM2 Pro Plus2
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x278b: # RM2 Pro Plus BL
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x278f: # RM Mini Shate
|
|
devices.append(rm(host=host, mac=mac))
|
|
elif devtype == 0x2714: # A1
|
|
devices.append(a1(host=host, mac=mac))
|
|
else:
|
|
devices.append(device(host=host, mac=mac))
|
|
|
|
class device:
|
|
def __init__(self, host, mac):
|
|
self.host = host
|
|
self.mac = mac
|
|
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.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.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_BROADCAST, 1)
|
|
self.cs.bind(('',0))
|
|
|
|
def auth(self):
|
|
payload = bytearray(0x50)
|
|
payload[0x04] = 0x31
|
|
payload[0x05] = 0x31
|
|
payload[0x06] = 0x31
|
|
payload[0x07] = 0x31
|
|
payload[0x08] = 0x31
|
|
payload[0x09] = 0x31
|
|
payload[0x0a] = 0x31
|
|
payload[0x0b] = 0x31
|
|
payload[0x0c] = 0x31
|
|
payload[0x0d] = 0x31
|
|
payload[0x0e] = 0x31
|
|
payload[0x0f] = 0x31
|
|
payload[0x10] = 0x31
|
|
payload[0x11] = 0x31
|
|
payload[0x12] = 0x31
|
|
payload[0x1e] = 0x01
|
|
payload[0x2d] = 0x01
|
|
payload[0x30] = 'T'
|
|
payload[0x31] = 'e'
|
|
payload[0x32] = 's'
|
|
payload[0x33] = 't'
|
|
payload[0x34] = ' '
|
|
payload[0x35] = ' '
|
|
payload[0x36] = '1'
|
|
|
|
response = self.send_packet(0x65, payload)
|
|
|
|
enc_payload = response[0x38:]
|
|
|
|
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
|
|
payload = aes.decrypt(str(enc_payload))
|
|
|
|
self.id = payload[0x00:0x04]
|
|
self.key = payload[0x04:0x14]
|
|
|
|
def send_packet(self, command, payload):
|
|
self.count = (self.count + 1) & 0xffff
|
|
packet = bytearray(0x38)
|
|
packet[0x00] = 0x5a
|
|
packet[0x01] = 0xa5
|
|
packet[0x02] = 0xaa
|
|
packet[0x03] = 0x55
|
|
packet[0x04] = 0x5a
|
|
packet[0x05] = 0xa5
|
|
packet[0x06] = 0xaa
|
|
packet[0x07] = 0x55
|
|
packet[0x24] = 0x2a
|
|
packet[0x25] = 0x27
|
|
packet[0x26] = command
|
|
packet[0x28] = self.count & 0xff
|
|
packet[0x29] = self.count >> 8
|
|
packet[0x2a] = self.mac[0]
|
|
packet[0x2b] = self.mac[1]
|
|
packet[0x2c] = self.mac[2]
|
|
packet[0x2d] = self.mac[3]
|
|
packet[0x2e] = self.mac[4]
|
|
packet[0x2f] = self.mac[5]
|
|
packet[0x30] = self.id[0]
|
|
packet[0x31] = self.id[1]
|
|
packet[0x32] = self.id[2]
|
|
packet[0x33] = self.id[3]
|
|
|
|
checksum = 0xbeaf
|
|
for i in range(len(payload)):
|
|
checksum += payload[i]
|
|
checksum = checksum & 0xffff
|
|
|
|
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
|
|
payload = aes.encrypt(str(payload))
|
|
|
|
packet[0x34] = checksum & 0xff
|
|
packet[0x35] = checksum >> 8
|
|
|
|
for i in range(len(payload)):
|
|
packet.append(payload[i])
|
|
|
|
checksum = 0xbeaf
|
|
for i in range(len(packet)):
|
|
checksum += packet[i]
|
|
checksum = checksum & 0xffff
|
|
packet[0x20] = checksum & 0xff
|
|
packet[0x21] = checksum >> 8
|
|
|
|
self.cs.sendto(packet, self.host)
|
|
response = self.cs.recvfrom(1024)
|
|
return response[0]
|
|
|
|
|
|
class sp1(device):
|
|
def __init__ (self, host, mac):
|
|
device.__init__(self, host, mac)
|
|
|
|
def set_power(self, state):
|
|
packet = bytearray(4)
|
|
packet[0] = state
|
|
self.send_packet(0x66, packet)
|
|
|
|
|
|
class sp2(device):
|
|
def __init__ (self, host, mac):
|
|
device.__init__(self, host, mac)
|
|
|
|
def set_power(self, state):
|
|
packet = bytearray(8)
|
|
packet[0] = 2
|
|
packet[4] = state
|
|
self.send_packet(0x6a, packet)
|
|
|
|
|
|
class a1(device):
|
|
def __init__ (self, host, mac):
|
|
device.__init__(self, host, mac)
|
|
|
|
def check_sensors(self):
|
|
packet = bytearray(16)
|
|
packet[0] = 1
|
|
response = self.send_packet(0x6a, packet)
|
|
err = ord(response[0x22]) | (ord(response[0x23]) << 8)
|
|
if err == 0:
|
|
data = {}
|
|
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
|
|
payload = aes.decrypt(str(response[0x38:]))
|
|
data['temperature'] = (ord(payload[0x4]) * 10 + ord(payload[0x5])) / 10.0
|
|
data['humidity'] = (ord(payload[0x6]) * 10 + ord(payload[0x7])) / 10.0
|
|
light = ord(payload[0x8])
|
|
if light == 0:
|
|
data['light'] = 'dark'
|
|
elif light == 1:
|
|
data['light'] = 'dim'
|
|
elif light == 2:
|
|
data['light'] = 'normal'
|
|
elif light == 3:
|
|
data['light'] = 'bright'
|
|
else:
|
|
data['light'] = 'unknown'
|
|
air_quality = ord(payload[0x0a])
|
|
if air_quality == 0:
|
|
data['air_quality'] = 'excellent'
|
|
elif air_quality == 1:
|
|
data['air_quality'] = 'good'
|
|
elif air_quality == 2:
|
|
data['air_quality'] = 'normal'
|
|
elif air_quality == 3:
|
|
data['air_quality'] = 'bad'
|
|
else:
|
|
data['air_quality'] = 'unknown'
|
|
noise = ord(payload[0xc])
|
|
if noise == 0:
|
|
data['noise'] = 'quiet'
|
|
elif noise == 1:
|
|
data['noise'] = 'normal'
|
|
elif noise == 2:
|
|
data['noise'] = 'noisy'
|
|
else:
|
|
data['noise'] = 'unknown'
|
|
return data
|
|
|
|
|
|
class rm(device):
|
|
def __init__ (self, host, mac):
|
|
device.__init__(self, host, mac)
|
|
|
|
def check_data(self):
|
|
packet = bytearray(16)
|
|
packet[0] = 4
|
|
response = self.send_packet(0x6a, packet)
|
|
err = ord(response[0x22]) | (ord(response[0x23]) << 8)
|
|
if err == 0:
|
|
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
|
|
payload = aes.decrypt(str(response[0x38:]))
|
|
return payload[0x04:]
|
|
|
|
def send_data(self, data):
|
|
packet = bytearray([0x02, 0x00, 0x00, 0x00])
|
|
packet += data
|
|
self.send_packet(0x6a, packet)
|
|
|
|
def enter_learning(self):
|
|
packet = bytearray(16)
|
|
packet[0] = 3
|
|
self.send_packet(0x6a, packet)
|
|
|
|
def check_temperature(self):
|
|
packet = bytearray(16)
|
|
packet[0] = 1
|
|
response = self.send_packet(0x6a, packet)
|
|
err = ord(response[0x22]) | (ord(response[0x23]) << 8)
|
|
if err == 0:
|
|
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
|
|
payload = aes.decrypt(str(response[0x38:]))
|
|
temp = (ord(payload[0x4]) * 10 + ord(payload[0x5])) / 10.0
|
|
return temp
|
|
|
|
# For legay compatibility - don't use this
|
|
class rm2(rm):
|
|
def __init__ (self):
|
|
device.__init__(self, None, None)
|
|
|
|
def discover(self):
|
|
dev = discover()
|
|
self.host = dev.host
|
|
self.mac = dev.mac
|