1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
#!/bin/python3
from bluepy.btle import Peripheral, ADDR_TYPE_PUBLIC, BTLEDisconnectError
import threading
import time
import sys
BEKEN_CHARACTERISTIC_NULL = 0x0001
BEKEN_CHARACTERISTIC_LAMP = 0x002a
def makemsg(r, g, b, l=0):
return bytes([
int(g > 0), g,
0x00, 0x00,
int(b > 0), b,
int(r > 0), r,
int(l > 0), l,
])
class BekenConnection:
def __init__(self, mac):
self.mac = mac
self.dev = None
self.messages = []
def keep_alive(self):
while True:
self.send(BEKEN_CHARACTERISTIC_NULL, bytes(10))
time.sleep(10)
def send(self, characteristic, message):
self.messages.append((characteristic, message))
def verify_connection(self):
while self.dev == None or self.dev.getState() == 'disc':
try:
self.dev = Peripheral(self.mac, ADDR_TYPE_PUBLIC)
except BTLEDisconnectError as e:
continue
def message_handler(self):
self.verify_connection()
while True:
if len(self.messages) < 1: continue
message = self.messages.pop(0)
self.verify_connection()
self.dev.writeCharacteristic(message[0], bytearray(message[1]))
def start_threads(self):
threading.Thread(target=self.message_handler).start()
threading.Thread(target=self.keep_alive).start()
if __name__ == "__main__":
mac = sys.argv[1]
con = BekenConnection(mac)
con.start_threads()
def user_input():
for line in sys.stdin:
r, g, b, l = [ int(x, 16) for x in [ line.strip()[i:i+2] for i in range(0, 8, 2) ] ]
con.send(BEKEN_CHARACTERISTIC_LAMP, makemsg(r, g, b, l))
threading.Thread(target=user_input).start()
|