apo/DoorControl
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Control bell through socket

Browse Source
This commit is contained in:
Lynn Ochs 2022-11-14 19:16:02 +01:00
parent cb5a93c136
commit a0f164d176
3 changed files with 30 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
door_pi_control/__init__.py
View File

@ -48,15 +48,15 @@ def main():
nfc = DoorControlNfc(config, control, mqttc) nfc = DoorControlNfc(config, control, mqttc)
nfc.start() nfc.start()
logger().info("Starting socket")
socket = DoorControlSocket(config, control, nfc)
socket.start()
logger().info("Starting bell") logger().info("Starting bell")
def door_is_open(): def door_is_open():
return control.state() == control.target() == door.constants.state.OPEN return control.state() == control.target() == door.constants.state.OPEN
bell_control = bell.Control(config.bell_port, door_is_open) bell_control = bell.Control(config.bell_port, door_is_open)
bell_control.start() bell_control.start()
logger().info("Starting socket")
socket = DoorControlSocket(config, control, bell_control, nfc)
socket.start()
while True: while True:
time.sleep(60) time.sleep(60)

14
door_pi_control/bell.py
View File

@ -163,19 +163,19 @@ class Control(util.Loggable):
self._task.start() self._task.start()
def run(self): def run(self):
port = None port = None
protocol = None self._protocol = None
while True: while True:
if port is None or protocol.closed: if port is None or self._protocol.closed:
try: try:
port = serial.Serial(self._port, 9600) port = serial.Serial(self._port, 9600)
protocol = serial.threaded.ReaderThread(port, lambda: Reader(self._cb)) self._protocol = serial.threaded.ReaderThread(port, lambda: Reader(self._cb))
protocol.start() self._protocol.start()
protocol = protocol.protocol self._protocol = protocol.protocol
except Exception as e: except Exception as e:
self._logger().exception("Port opening error") self._logger().exception("Port opening error")
time.sleep(10) time.sleep(10)
continue continue
if protocol.doing_light: if self._protocol.doing_light:
protocol.send_light_msg() self._protocol.send_light_msg()
time.sleep(1) time.sleep(1)

20
door_pi_control/socket.py
View File

@ -34,10 +34,11 @@ class DoorControlSocket(util.Loggable):
self._logger().exception("Error while handling command") self._logger().exception("Error while handling command")
return True return True
def __init__(self, config, control, nfc): def __init__(self, config, control, bell, nfc):
super().__init__("socket") super().__init__("socket")
self._config = config self._config = config
self._control = control self._control = control
self._bell = bell
self._nfc = nfc self._nfc = nfc
self._fifo = self._open_control_socket() self._fifo = self._open_control_socket()
self._stop = True self._stop = True
@ -114,6 +115,23 @@ class DoorControlSocket(util.Loggable):
self._control.close() self._control.close()
else: else:
send("Missing login") send("Missing login")
elif cmd == 'bell_open':
send("Opening front door")
self._bell._protocol.open_door()
elif cmd == 'bell_light':
send("Turning the light on")
self._bell._protocol.send_light_msg()
elif cmd == 'bell_perma_light':
self._bell._protocol.doing_light ^= True
send("Turning the light on permanently: {self._bell._protocol.doing_light}")
elif cmd == 'bell_raw':
if (len(args) % 4) != 0:
send(f"Invalid number of bytes: {len(args)}")
return
data = bytes.fromhex(" ".join(args))
data = [ data[i:i+4] for i in range(0, len(data), 4) ]
send(f"Sending {repr(data)}")
self._bell._protocol.alternate_msgs(data)
elif cmd == 'rld': elif cmd == 'rld':
self._logger().debug("Reloading tokens") self._logger().debug("Reloading tokens")
send("Reloading tokens") send("Reloading tokens")