tanner/doorbelldingdongringringdoorbell
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Integrate GPIO relays

Browse Source
This commit is contained in:
Tanner Collin 2021-10-15 15:07:28 -06:00
parent 9e49c728ed
commit d9499faaba
1 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
main.py
View File

@ -2,6 +2,8 @@ import os
import sys import sys
import asyncio import asyncio
import aiohttp import aiohttp
import RPi.GPIO as GPIO
import time
from aiohttp import ClientSession, CookieJar from aiohttp import ClientSession, CookieJar
import settings import settings
@ -11,12 +13,25 @@ from pyunifiprotect.unifi_protect_server import UpvServer
import logging import logging
logger = logging.getLogger(__name__) 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): def subscriber(updated):
logger.debug('Subscription: updated=%s', updated) logger.debug('Subscription: updated=%s', updated)
for _, data in updated.items(): for _, data in updated.items():
if data['event_type'] == 'ring' and data['event_ring_on']: if data['event_type'] == 'ring' and data['event_ring_on']:
logger.info('%s is ringing!', data['name']) logger.info('%s: %s is ringing!', data['mac'], data['name'])
ring_bell(data['mac'])
async def ws_listener(): async def ws_listener():
session = ClientSession(cookie_jar=CookieJar(unsafe=True)) session = ClientSession(cookie_jar=CookieJar(unsafe=True))
@ -40,6 +55,14 @@ async def ws_listener():
await session.close() await session.close()
unsub() 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__': if __name__ == '__main__':
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()