199 lines
7.4 KiB
Python
199 lines
7.4 KiB
Python
from . import mqtt, util
|
|
from .util import timestamp
|
|
import time
|
|
import datetime
|
|
import serial
|
|
import serial.threaded
|
|
import threading
|
|
|
|
msgs = {
|
|
"push_1": bytes.fromhex("B0 A0 6F 08"),
|
|
"push_2": bytes.fromhex("B5 A0 6F 18"),
|
|
"release": bytes.fromhex("B2 A0 6F 08"),
|
|
"voice_closed": bytes.fromhex("B5 02 60 08"),
|
|
"ring_hs": bytes.fromhex("91 02 60 08"),
|
|
"ring_meyer": bytes.fromhex("91 01 60 08"),
|
|
"open_push": bytes.fromhex("96 a0 6f a4"),
|
|
"open_release": bytes.fromhex("96 a0 6f a0"),
|
|
"light_push": bytes.fromhex("bd a0 6f a4"),
|
|
"light_release":bytes.fromhex("bd a0 6f a0"),
|
|
}
|
|
|
|
class Reader(util.Loggable):
|
|
def __init__(self, should_open_immediately_cb = None, mqttc = None):
|
|
super().__init__("bell")
|
|
self.tmp = bytes()
|
|
self.data = []
|
|
self.lock = threading.RLock()
|
|
|
|
self.cv = threading.Condition(self.lock)
|
|
|
|
self.times = []
|
|
self.start = None
|
|
self.last_symbol_timestamp = None
|
|
self.light_push_time = None
|
|
|
|
self.doing_light = False
|
|
self.closed = False
|
|
self.open_now_cb = should_open_immediately_cb
|
|
|
|
self._events = mqtt.Value(mqttc, "door/bell/event")
|
|
self._last_ring = datetime.datetime.now() - datetime.timedelta(days=1)
|
|
|
|
def connection_lost(self, exc):
|
|
if exc:
|
|
print(exc)
|
|
traceback.print_exc()
|
|
self._logger().info("Closed")
|
|
self.closed = True
|
|
|
|
def connection_made(self, transport):
|
|
# super(Reader, self).connection_made(transport)
|
|
self.port = transport
|
|
self._logger().info("Opened")
|
|
|
|
def chksum(self, x):
|
|
return x[0] ^ x[1] ^ x[2] ^ x[3]
|
|
|
|
def write(self, data):
|
|
with self.lock:
|
|
self.port.write(bytes([0xA8]) + data + bytes([self.chksum(data), 0xA3]))
|
|
|
|
def data_received(self, data):
|
|
self.tmp += data
|
|
data= self.tmp
|
|
|
|
while len(data) >= 7:
|
|
while len(data) >= 7 and (data[0] != 0xA8 or data[-1] != 0xA3 or self.chksum(data[1:-2]) != data[-2]):
|
|
if 0xA8 in data[1:]:
|
|
next = data[1:].index(0xA8)
|
|
data = data[1+next:]
|
|
else:
|
|
data = bytes()
|
|
|
|
if len(data) >= 7:
|
|
msg = data[1:5]
|
|
with self.cv:
|
|
if len(self.data) and msg == self.data[-1]:
|
|
pass
|
|
#log("Repeats")
|
|
else:
|
|
for k,v in msgs.items():
|
|
if v == msg:
|
|
self._logger().debug("Known message: %s: %s" % (k, v.hex()))
|
|
break
|
|
else:
|
|
self._logger().debug("Unknown message: %s" % (msg.hex(),))
|
|
|
|
self.data.append(data[1:5])
|
|
self.cv.notify()
|
|
|
|
if msgs['light_push'] == msg and not self.light_push_time:
|
|
self.light_push_time = datetime.datetime.now()
|
|
elif msgs['light_release'] == msg and self.light_push_time != None:
|
|
dt = (datetime.datetime.now() - self.light_push_time).total_seconds()
|
|
self.light_push_time = None
|
|
self._logger().debug(f"Light time: {dt}")
|
|
if dt >= 3:
|
|
self.doing_light = not self.doing_light
|
|
elif msgs['ring_hs'] == msg:
|
|
if not self.start:
|
|
self.start = datetime.datetime.now()
|
|
if self.last_symbol_timestamp and (self.start - self.last_symbol_timestamp).total_seconds() > 5:
|
|
self.times = []
|
|
elif msgs['release'] == msg and self.start:
|
|
open_now = self.open_now_cb and self.open_now_cb()
|
|
if open_now:
|
|
self._logger().info("Opening after a single button press")
|
|
self.open_door()
|
|
else:
|
|
t = datetime.datetime.now()
|
|
self.last_symbol_timestamp = t
|
|
dt = (t - self.start).total_seconds()
|
|
self.start = None
|
|
self.times = self.times[-3:] + [dt]
|
|
avg = sum(self.times) / len(self.times)
|
|
morse = [x>avg for x in self.times]
|
|
if len(self.times) == 4:
|
|
self._logger().info(", ".join([{True: "dah", False: "dit"}[x > avg] for x in self.times]))
|
|
|
|
if (t - self._last_ring).total_seconds() > 30:
|
|
self._events("ring", force=True)
|
|
self._last_ring = t
|
|
|
|
if morse == [False, False, True, False]:
|
|
self._events("morse", force=True)
|
|
self.open_door()
|
|
data = data[7:]
|
|
|
|
self.tmp = data
|
|
self._events("")
|
|
|
|
def alternate_msgs(self, names, times = 3, *, dt = 0.04):
|
|
for _ in range(times):
|
|
for m in names:
|
|
if type(m) == str:
|
|
self.write(msgs[m])
|
|
else:
|
|
self.write(m)
|
|
time.sleep(dt)
|
|
|
|
def send_light_msg(self):
|
|
self.alternate_msgs(['light_push'])
|
|
time.sleep(0.3)
|
|
self.alternate_msgs(['light_release'])
|
|
|
|
def send_open_msg(self):
|
|
self.alternate_msgs(['open_push'])
|
|
time.sleep(0.3)
|
|
self.alternate_msgs(['open_release'])
|
|
|
|
def open_door(self):
|
|
self._logger().info("Opening")
|
|
self._events("open", force=True)
|
|
self._events("")
|
|
self.alternate_msgs(['light_push', 'open_push'], dt = 0.04)
|
|
time.sleep(0.3)
|
|
self.alternate_msgs(['light_release', 'open_release'], dt = 0.04);
|
|
|
|
def get(self, block = False):
|
|
with self.cv:
|
|
if len(self.data) == 0:
|
|
if block:
|
|
self.cv.wait_for(lambda: len(self.data) > 0)
|
|
else:
|
|
return None
|
|
rv = self.data[0]
|
|
self.data = self.data[1:]
|
|
return rv
|
|
|
|
class Control(util.Loggable):
|
|
def __init__(self, port, mqttc = None, open_immediately_cb = None):
|
|
super().__init__("bell")
|
|
self._port = port
|
|
self._task = None
|
|
self._cb = open_immediately_cb
|
|
self._mqttc = mqttc
|
|
def start(self):
|
|
if not self._task:
|
|
self._task = threading.Thread(target = self.run, daemon=True)
|
|
self._task.start()
|
|
def run(self):
|
|
port = None
|
|
self._protocol = None
|
|
while True:
|
|
if port is None or self._protocol.closed:
|
|
try:
|
|
port = serial.Serial(self._port, 9600)
|
|
self._protocol = serial.threaded.ReaderThread(port, lambda: Reader(self._cb, self._mqttc))
|
|
self._protocol.start()
|
|
self._protocol = self._protocol.protocol
|
|
except Exception as e:
|
|
self._logger().exception("Port opening error")
|
|
time.sleep(10)
|
|
continue
|
|
|
|
if self._protocol.doing_light:
|
|
self._protocol.send_light_msg()
|
|
time.sleep(1)
|