DoorControl/door_pi_control/bell.py

182 lines
6.6 KiB
Python
Raw Normal View History

2022-11-06 13:30:11 +00:00
from . import mqtt, util
from .util import timestamp
import time
2022-11-06 17:46:58 +00:00
import datetime
2022-11-06 13:30:11 +00:00
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):
2022-11-11 15:37:44 +00:00
super().__init__("bell")
2022-11-06 13:30:11 +00:00
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
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:
open_now = self.open_now_cb and self.open_now_cb()
if open_now:
2022-11-07 05:29:37 +00:00
self._logger().info("Opening after a single button press")
2022-11-06 13:30:11 +00:00
self.open_door()
elif 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:
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 morse == [False, False, True, False]:
self.open_door()
data = data[7:]
self.tmp = data
def alternate_msgs(self, names, times = 3, *, dt = 0.04):
for _ in range(times):
for m in names:
self.write(msgs[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.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, open_immediately_cb = None):
2022-11-11 15:37:44 +00:00
super().__init__("bell")
2022-11-06 13:30:11 +00:00
self._port = port
self._task = None
self._cb = open_immediately_cb
def start(self):
if not self._task:
self._task = threading.Thread(target = self.run, daemon=True)
self._task.start()
def run(self):
port = None
protocol = None
while True:
if port is None or protocol.closed:
try:
port = serial.Serial(self._port, 9600)
protocol = serial.threaded.ReaderThread(port, lambda: Reader(self._cb))
protocol.start()
protocol = protocol.protocol
except Exception as e:
self._logger().exception("Port opening error")
time.sleep(10)
continue
if protocol.doing_light:
protocol.send_light_msg()
time.sleep(1)