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

Add support for multiple devices and update documentation

Browse Source 
Add support for discovering more than one device on the network, and update
the documentation to describe that.
This commit is contained in:
Matthew Garrett 2016-10-30 14:16:40 -07:00
parent 69afd4ce52
commit 989009e21f
2 changed files with 109 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
README.md
View File

@ -1,36 +1,45 @@
Python control for Broadlink RM2 IR controllers Python control for Broadlink RM2 IR controllers
=============================================== ===============================================
A simple Python API for controlling IR controllers from [Broadlink](http://www.ibroadlink.com/rm/). At present, only RM Pro (referred to as RM2 in the codebase) devices are supported and only one device per network will be used. There is currently no support for the cloud API. A simple Python API for controlling IR controllers from [Broadlink](http://www.ibroadlink.com/rm/). At present, only RM Pro (referred to as RM2 in the codebase) and A1 sensor platform devices are supported. There is currently no support for the cloud API.
Example use Example use
----------- -----------
Discover an available device on the local network: Discover available devices on the local network:
``` ```
import broadlink import broadlink
device = broadlink.rm2() devices = broadlink.discover(timeout=5)
device.discover()
``` ```
Obtain the authentication key required for further communication: Obtain the authentication key required for further communication:
``` ```
device.auth() devices[0].auth()
``` ```
Enter learning mode: Enter learning mode:
``` ```
device.enter_learning() devices[0].enter_learning()
``` ```
Obtain an IR packet while in learning mode: Obtain an IR packet while in learning mode:
``` ```
ir_packet = device.check_data() ir_packet = devices[0].check_data()
``` ```
(This will return None if the device does not have a packet to return) (This will return None if the device does not have a packet to return)
Send an IR packet: Send an IR packet:
``` ```
device.send_data(ir_packet) devices[0].send_data(ir_packet)
```
Obtain temperature data from an RM2:
```
devices[0].check_temperature()
```
Obtain sensor data from an A1:
```
data = devices[0].check_sensors()
``` ```

154
broadlink/__init__.py
View File

@ -1,74 +1,95 @@
#!/usr/bin/python #!/usr/bin/python
from datetime import datetime from datetime import datetime
from socket import *
from Crypto.Cipher import AES from Crypto.Cipher import AES
import time import time
import random import random
import socket
class rm2: def discover(timeout=None):
def __init__(self): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.count = random.randrange(0xffff) s.connect(('8.8.8.8', 0)) # 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]
mac = responsepacket[0x3a:0x40]
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.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.iv = bytearray([0x56, 0x2e, 0x17, 0x99, 0x6d, 0x09, 0x3d, 0x28, 0xdd, 0xb3, 0xba, 0x69, 0x5a, 0x2e, 0x6f, 0x58])
s = socket(AF_INET, SOCK_DGRAM)
s.connect(('8.8.8.8', 0)) # connecting to a UDP address doesn't send packets
local_ip_address = s.getsockname()[0]
self.address = local_ip_address.split('.')
self.id = bytearray([0, 0, 0, 0]) self.id = bytearray([0, 0, 0, 0])
self.cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
def discover(self): self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.cs = socket(AF_INET, SOCK_DGRAM) self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
self.cs.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) self.cs.bind(('',0))
self.cs.setsockopt(SOL_SOCKET, SO_BROADCAST, 1)
self.cs.bind(('',0))
self.port = self.cs.getsockname()[1]
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(self.address[0])
packet[0x19] = int(self.address[1])
packet[0x1a] = int(self.address[2])
packet[0x1b] = int(self.address[3])
packet[0x1c] = self.port & 0xff
packet[0x1d] = self.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
self.cs.sendto(packet, ('255.255.255.255', 80))
response = self.cs.recvfrom(1024)
responsepacket = bytearray(response[0])
self.host = response[1]
self.mac = responsepacket[0x3a:0x40]
def auth(self): def auth(self):
payload = bytearray(0x50) payload = bytearray(0x50)
@ -102,7 +123,7 @@ class rm2:
enc_payload = response[0x38:] enc_payload = response[0x38:]
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv)) aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
payload = aes.decrypt(str(response[0x38:])) payload = aes.decrypt(str(enc_payload))
self.id = payload[0x00:0x04] self.id = payload[0x00:0x04]
self.key = payload[0x04:0x14] self.key = payload[0x04:0x14]
@ -232,3 +253,12 @@ class rm2:
aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv)) aes = AES.new(str(self.key), AES.MODE_CBC, str(self.iv))
payload = aes.decrypt(str(response[0x38:])) payload = aes.decrypt(str(response[0x38:]))
return payload[0x04:] return payload[0x04:]
class rm2(device):
def __init__ (self):
device.__init__(self, None, None)
def discover(self):
dev = discover()
self.host = dev.host
self.mac = dev.mac