main.py

@@ -1,5 +1,6 @@
 import os, logging
 DEBUG = os.environ.get('DEBUG')
+NO_WATCHDOG = os.environ.get('NO_WATCHDOG')
 logging.basicConfig(
         format='[%(asctime)s] %(levelname)s %(module)s/%(funcName)s - %(message)s',
         level=logging.DEBUG if DEBUG else logging.INFO)
@@ -9,26 +10,26 @@ import json
 import os
 import sys
 import asyncio
-
+import aiohttp
-try:
 import RPi.GPIO as GPIO
-    IS_PI = True
-except ModuleNotFoundError:
-    IS_PI = False
-
 import time
 from signal import *
+from aiohttp import ClientSession, CookieJar
 
-import unifi
 import settings
 
+from pyunifiprotect.unifi_protect_server import UpvServer
+from pyunifiprotect.exceptions import NvrError
+
 RELAY_ON = False
 RELAY_OFF = True
 
+allow_watchdog = False
+
 cooldown_time = time.time()
 
 def set_relay(pin, state):
-    if IS_PI: GPIO.output(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):
@@ -36,8 +37,12 @@ def pulse_relay(pin):
     time.sleep(0.5)  # atomic
     set_relay(pin, RELAY_OFF)
 
-def ring_bell(camera):
+def ring_bell(mac):
-    global cooldown_time
+    global allow_watchdog, cooldown_time
+
+    if not allow_watchdog and not DEBUG and not NO_WATCHDOG:
+        logging.info('Enabling watchdog...')
+        allow_watchdog = True
 
     if time.time() - cooldown_time < 2:
         logging.info('Cooldown skipping.')
@@ -46,30 +51,71 @@ def ring_bell(camera):
     cooldown_time = time.time()
 
     try:
-        doorbell = settings.DOORBELLS[camera]
+        doorbell = settings.DOORBELLS[mac]
         pulse_relay(doorbell['gpio'])
     except KeyError:
-        logging.error('Doorbell %s not found!', camera)
+        logging.error('Doorbell %s not found!', mac)
 
+def subscriber(updates):
+    logging.debug('Subscription: updates=%s', updates)
 
-async def process_message(msg):
+    for _, data in updates.items():
-    if msg.get('type', '') != 'ring':
+        if data['event_type'] == 'ring' and data['event_ring_on']:
-        return
+            logging.info('%s: %s is ringing!', data['mac'], data['name'])
+            ring_bell(data['mac'])
 
-    logging.info('Ring message: %s', msg)
+def feed_watchdog():
+    with open('/dev/watchdog', 'w') as wdt:
+        wdt.write('1')
 
-    ring_bell(msg['camera'])
+async def ws_listener():
+    session = ClientSession(cookie_jar=CookieJar(unsafe=True))
+
+    unifiprotect = UpvServer(
+        session,
+        settings.UFP_ADDRESS,
+        settings.UFP_PORT,
+        settings.UFP_USERNAME,
+        settings.UFP_PASSWORD,
+    )
+
+    await unifiprotect.update()
+
+    unsub = unifiprotect.subscribe_websocket(subscriber)
+
+    logging.info('Connecting to websocket.')
+    await asyncio.sleep(2)
 
-async def main():
     while True:
         try:
-            async for msg in unifi.connect():
+            updates = await unifiprotect.update()
-                await process_message(msg)
+            logging.debug('')
-        except BaseException as e:
+            logging.debug('Updates: %s', json.dumps(updates, indent=4))
+        except NvrError:
+            logging.error('Error updating connection. Reconnecting...')
+            break
+
+        active_ws = await unifiprotect.check_ws()
+        if not active_ws:
+            logging.error('Websocket unactive. Reconnecting...')
+            break
+
+        if allow_watchdog:
+            feed_watchdog()
+
+        await asyncio.sleep(1)
+
+    await session.close()
+    unsub()
+
+async def connect():
+    while True:
+        try:
+            await ws_listener()
+        except NvrError as e:
             logging.error('Error connecting to Unifi Protect: %s. Trying again...', str(e))
             await asyncio.sleep(3)
 
-
 def disable_relays_on_exit(*args):
     logging.info('Exiting, disabling relays...')
     for _, doorbell in settings.DOORBELLS.items():
@@ -78,11 +124,11 @@ def disable_relays_on_exit(*args):
     os._exit(0)
 
 def init():
-    if IS_PI: GPIO.setmode(GPIO.BCM)
+    GPIO.setmode(GPIO.BCM)
-    if IS_PI: GPIO.setwarnings(False)
+    GPIO.setwarnings(False)
 
     for _, doorbell in settings.DOORBELLS.items():
-        if IS_PI: GPIO.setup(doorbell['gpio'], GPIO.OUT)
+        GPIO.setup(doorbell['gpio'], GPIO.OUT)
         set_relay(doorbell['gpio'], RELAY_OFF)
         #pulse_relay(doorbell['gpio'])
         time.sleep(1)
@@ -99,5 +145,5 @@ if __name__ == '__main__':
     init()
 
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(main())
+    loop.run_until_complete(connect())
     loop.close()

requirements.txt

@@ -1,9 +1 @@
-aiohttp==3.8.1
+pyunifiprotect==1.1.0
-aiosignal==1.2.0
-async-timeout==4.0.2
-attrs==21.4.0
-charset-normalizer==2.0.12
-frozenlist==1.3.0
-idna==3.3
-multidict==6.0.2
-yarl==1.7.2

unifi.py

@@ -1,41 +0,0 @@
-import asyncio
-import aiohttp
-import zlib
-import struct
-import json
-
-import settings
-
-HEADER_LENGTH = 8
-
-async def connect():
-    data = dict(
-        username=settings.UFP_USERNAME,
-        password=settings.UFP_PASSWORD,
-        rememberMe=True,
-    )
-
-    async with aiohttp.ClientSession() as session:
-        async with session.post(settings.UFP_ADDRESS + '/api/auth/login', json=data, ssl=False) as resp:
-            cookie = resp.cookies['TOKEN']
-
-        headers = {'cookie': cookie.key + '=' + cookie.value}
-        async with session.ws_connect(settings.UFP_ADDRESS + '/proxy/protect/ws/updates', headers=headers, ssl=False) as ws:
-            async for msg in ws:
-                packet_type, payload_format, deflated, unknown, payload_size = struct.unpack('!bbbbi', msg.data[0:HEADER_LENGTH])
-                action_start = HEADER_LENGTH
-                action_packet = zlib.decompress(msg.data[action_start:])
-                data_start = payload_size + 2*HEADER_LENGTH
-                data_packet = zlib.decompress(msg.data[data_start:])
-
-                yield json.loads(data_packet.decode())
-
-
-async def test():
-    async for msg in connect():
-        print(msg)
-
-if __name__ == '__main__':
-    loop = asyncio.get_event_loop()
-    loop.run_until_complete(test())
-    loop.close()