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)