doorbelldingdongringringdoorbell/main.py

import os
import sys
import asyncio
import aiohttp
import RPi.GPIO as GPIO
import time
from aiohttp import ClientSession, CookieJar

import settings

from pyunifiprotect.unifi_protect_server import UpvServer

import logging
logger = logging.getLogger(__name__)

RELAY_ON = False
RELAY_OFF = True

def ring_bell(mac):
    try:
        doorbell = settings.DOORBELLS[mac]
        GPIO.output(doorbell['gpio'], RELAY_ON)
        time.sleep(0.5)
        GPIO.output(doorbell['gpio'], RELAY_OFF)
    except KeyError:
        logger.error('Doorbell %s not found!', mac)

def subscriber(updated):
    logger.debug('Subscription: updated=%s', updated)

    for _, data in updated.items():
        if data['event_type'] == 'ring' and data['event_ring_on']:
            logger.info('%s: %s is ringing!', data['mac'], data['name'])
            ring_bell(data['mac'])

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.check_unifi_os()
    await unifiprotect.update()

    unsub = unifiprotect.subscribe_websocket(subscriber)

    for i in range(15000):
        await asyncio.sleep(1)

    await session.close()
    unsub()

def init_relays():
    GPIO.setmode(GPIO.BCM)
    GPIO.setwarnings(False)

    for _, doorbell in settings.DOORBELLS.items():
        GPIO.setup(doorbell['gpio'], GPIO.OUT)
        GPIO.output(doorbell['gpio'], RELAY_OFF)

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(ws_listener())
    loop.close()