doorbelldingdongringringdoorbell/main.py

import os
import sys
import asyncio
import aiohttp
from aiohttp import ClientSession, CookieJar

import settings

from pyunifiprotect.unifi_protect_server import UpvServer

import logging
logger = logging.getLogger(__name__)

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 is ringing!', data['name'])

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()

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