#!/usr/bin/python from datetime import datetime from Crypto.Cipher import AES import time import random import socket def gendevice(devtype, host, mac): if devtype == 0: # SP1 return sp1(host=host, mac=mac) if devtype == 0x2711: # SP2 return sp2(host=host, mac=mac) if devtype == 0x2719 or devtype == 0x7919 or devtype == 0x271a or devtype == 0x791a: # Honeywell SP2 return sp2(host=host, mac=mac) if devtype == 0x2720: # SPMini return sp2(host=host, mac=mac) elif devtype == 0x753e: # SP3 return sp2(host=host, mac=mac) elif devtype == 0x2728: # SPMini2 return sp2(host=host, mac=mac) elif devtype == 0x2733 or devtype == 0x273e: # OEM branded SPMini return sp2(host=host, mac=mac) elif devtype >= 0x7530 and devtype <= 0x7918: # OEM branded SPMini2 return sp2(host=host, mac=mac) elif devtype == 0x2736: # SPMiniPlus return sp2(host=host, mac=mac) elif devtype == 0x2712: # RM2 return rm(host=host, mac=mac) elif devtype == 0x2737: # RM Mini return rm(host=host, mac=mac) elif devtype == 0x273d: # RM Pro Phicomm return rm(host=host, mac=mac) elif devtype == 0x2783: # RM2 Home Plus return rm(host=host, mac=mac) elif devtype == 0x277c: # RM2 Home Plus GDT return rm(host=host, mac=mac) elif devtype == 0x272a: # RM2 Pro Plus return rm(host=host, mac=mac) elif devtype == 0x2787: # RM2 Pro Plus2 return rm(host=host, mac=mac) elif devtype == 0x278b: # RM2 Pro Plus BL return rm(host=host, mac=mac) elif devtype == 0x278f: # RM Mini Shate return rm(host=host, mac=mac) elif devtype == 0x2714: # A1 return a1(host=host, mac=mac) else: return device(host=host, mac=mac) 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 = int(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] devtype = responsepacket[0x34] | responsepacket[0x35] << 8 return gendevice(devtype, host, 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] dev = gendevice(devtype, host, mac) devices.append(dev) class device: def __init__(self, host, mac, timeout=10): self.host = host self.mac = mac self.timeout = timeout 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] = ord('T') payload[0x31] = ord('e') payload[0x32] = ord('s') payload[0x33] = ord('t') payload[0x34] = ord(' ') payload[0x35] = ord(' ') payload[0x36] = ord('1') response = self.send_packet(0x65, payload) enc_payload = response[0x38:] aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.decrypt(bytes(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(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.encrypt(bytes(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 starttime = time.time() while True: try: self.cs.sendto(packet, self.host) self.cs.settimeout(1) response = self.cs.recvfrom(1024) break except socket.timeout: if (time.time() - starttime) < self.timeout: pass raise return bytes(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): """Sets the power state of the smart plug.""" packet = bytearray(16) packet[0] = 2 packet[4] = 1 if state else 0 self.send_packet(0x6a, packet) def check_power(self): """Returns the power state of the smart plug.""" packet = bytearray(16) packet[0] = 1 response = self.send_packet(0x6a, packet) err = response[0x22] | (response[0x23] << 8) if err == 0: aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.decrypt(bytes(response[0x38:])) return bool(payload[0x4]) 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 = response[0x22] | (response[0x23] << 8) if err == 0: data = {} aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.decrypt(bytes(response[0x38:])) data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0 data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0 light = 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 = 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 = 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 = response[0x22] | (response[0x23] << 8) if err == 0: aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.decrypt(bytes(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 = response[0x22] | (response[0x23] << 8) if err == 0: aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv)) payload = aes.decrypt(bytes(response[0x38:])) temp = (payload[0x4] * 10 + 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