Abstract pulse_relay, improve watchdog

This commit is contained in:
Tanner Collin 2021-10-16 01:33:57 +01:00
parent f8e5e9cba2
commit 59fcd694b6

41
main.py
View File

@ -11,6 +11,7 @@ import asyncio
import aiohttp
import RPi.GPIO as GPIO
import time
from signal import *
from aiohttp import ClientSession, CookieJar
import settings
@ -20,12 +21,24 @@ from pyunifiprotect.unifi_protect_server import UpvServer
RELAY_ON = False
RELAY_OFF = True
allow_watchdog = False
def set_relay(pin, state):
GPIO.output(pin, state)
logging.info('Set relay on pin %s to %s', pin, 'ON' if state == RELAY_ON else 'OFF')
def pulse_relay(pin):
set_relay(pin, RELAY_ON)
time.sleep(0.25)
set_relay(pin, RELAY_OFF)
def ring_bell(mac):
global allow_watchdog
allow_watchdog = True
try:
doorbell = settings.DOORBELLS[mac]
GPIO.output(doorbell['gpio'], RELAY_ON)
time.sleep(0.5)
GPIO.output(doorbell['gpio'], RELAY_OFF)
pulse_relay(doorbell['gpio'])
except KeyError:
logging.error('Doorbell %s not found!', mac)
@ -37,6 +50,10 @@ def subscriber(updated):
logging.info('%s: %s is ringing!', data['mac'], data['name'])
ring_bell(data['mac'])
def feed_watchdog():
with open('/dev/watchdog', 'w') as wdt:
wdt.write('1')
async def ws_listener():
session = ClientSession(cookie_jar=CookieJar(unsafe=True))
@ -53,7 +70,10 @@ async def ws_listener():
unsub = unifiprotect.subscribe_websocket(subscriber)
for i in range(15000):
while True:
if not DEBUG and allow_watchdog:
logging.debug('Feeding watchdog...')
feed_watchdog()
await asyncio.sleep(1)
await session.close()
@ -62,7 +82,7 @@ async def ws_listener():
def disable_relays_on_exit(*args):
logging.info('Exiting, disabling relays...')
for _, doorbell in settings.DOORBELLS.items():
GPIO.output(doorbell['gpio'], RELAY_OFF)
set_relay(doorbell['gpio'], RELAY_OFF)
logging.info('Goodbye.')
os._exit(0)
@ -72,24 +92,17 @@ def init():
for _, doorbell in settings.DOORBELLS.items():
GPIO.setup(doorbell['gpio'], GPIO.OUT)
GPIO.output(doorbell['gpio'], RELAY_OFF)
pulse_relay(doorbell['gpio'])
time.sleep(1)
logging.info('GPIO initialized')
for sig in (SIGABRT, SIGILL, SIGINT, SIGSEGV, SIGTERM):
signal(sig, disable_relays_on_exit)
logging.info('Signals initialized')
async def feed_watchdog():
while True:
with open('/dev/watchdog', 'w') as wdt:
wdt.write('1')
await asyncio.sleep(1)
if __name__ == '__main__':
init()
loop = asyncio.get_event_loop()
if not DEBUG:
loop.create_task(feed_watchdog())
loop.run_until_complete(ws_listener())
loop.close()