Send push notification when doorbell rings (using Tinkerforge Industrial Digital In 4 Bricklet)

doorbell.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-  

# Send push notification when doorbell rings  
# using Tinkerforge Industrial Digital In 4 Bricklet
#
# License: CC BY-SA 3.0  
# Based on http://www.tinkerforge.com/de/doc/Kits/HardwareHacking/DoorbellNotifier_Python.html
#
# The following changes were made:
#
# * Added argparse
# * Added logging
# * Added push support via Pushover (https://pushover.net/)
#

import socket, sys, signal, time, datetime, math, argparse, logging, httplib, urllib
logger = logging.getLogger('doorbell_notifier')

from tinkerforge.ip_connection import IPConnection
from tinkerforge.ip_connection import Error
from tinkerforge.bricklet_industrial_digital_in_4 import IndustrialDigitalIn4

class DoorbellNotifier:

    ipcon = None

    push_token = None
    push_user = None

    def __init__(self, args):

        self.push_token = args.push_token
        self.push_user = args.push_user

        self.ipcon = IPConnection()
        while True:
            try:
                self.ipcon.connect(args.host, args.port)
                break
            except Error as e:
                logger.error('Connection Error: ' + str(e.description))
                time.sleep(1)
            except socket.error as e:
                logger.error('Socket error: ' + str(e))
                time.sleep(1)

        self.ipcon.register_callback(IPConnection.CALLBACK_ENUMERATE,
                                     self.cb_enumerate)
        self.ipcon.register_callback(IPConnection.CALLBACK_CONNECTED,
                                     self.cb_connected)

        while True:
            try:
                self.ipcon.enumerate()
                break
            except Error as e:
                logger.error('Enumerate Error: ' + str(e.description))
                time.sleep(1)


    def cb_interrupt(self, interrupt_mask, value_mask):
        if value_mask > 0:
            logger.info('Door bell rang!')
            self.push()


    def cb_enumerate(self, uid, connected_uid, position, hardware_version,
                     firmware_version, device_identifier, enumeration_type):
        if enumeration_type == IPConnection.ENUMERATION_TYPE_CONNECTED or \
           enumeration_type == IPConnection.ENUMERATION_TYPE_AVAILABLE:
            if device_identifier == IndustrialDigitalIn4.DEVICE_IDENTIFIER:
                try:
                    self.idi4 = IndustrialDigitalIn4(uid, self.ipcon)
                    self.idi4.set_debounce_period(10000)
                    self.idi4.set_interrupt(15)
                    self.idi4.register_callback(IndustrialDigitalIn4.CALLBACK_INTERRUPT,
                                                self.cb_interrupt)
                    logger.info('Industrial Digital In 4 initialized')
                except Error as e:
                    logger.error('Industrial Digital In 4 init failed: ' + str(e.description))
                    self.idi4 = None


    def cb_connected(self, connected_reason):
        if connected_reason == IPConnection.CONNECT_REASON_AUTO_RECONNECT:
            logger.info('Auto Reconnect')

            while True:
                try:
                    self.ipcon.enumerate()
                    break
                except Error as e:
                    logger.error('Enumerate Error: ' + str(e.description))
                    time.sleep(1)

    def push(self):
        pushcon = httplib.HTTPSConnection("api.pushover.net:443")
        pushcon.request("POST", "/1/messages.json",
          urllib.urlencode({
            "token": self.push_token,
            "user": self.push_user,
            "message": "Ring ring!",
          }), { "Content-type": "application/x-www-form-urlencoded" })
        pushcon.getresponse()

    def cleanup(self):
        if self.ipcon != None:
            self.ipcon.disconnect()
        logger.info('Doorbell Notifier: End')


def signal_handler(doorbell_notifier):
    doorbell_notifier.cleanup()
    sys.exit(0)

def setup_logging():

    logger.setLevel(logging.INFO)

    fh = logging.FileHandler('/var/log/doorbell.log')
    fh.setLevel(logging.INFO)

    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)

    formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')

    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    logger.addHandler(fh)
    logger.addHandler(ch)


if __name__ == "__main__":

    parser = argparse.ArgumentParser(description='Doorbell notifier')

    parser.add_argument('--host', help="brickd host", required=True)
    parser.add_argument('--port', type=int, help="brickd port", required=True)
    parser.add_argument('--uid', help="Industrial Digital In 4 bricklet UID", required=True)
    parser.add_argument('--push-user', help="Pushover user", required=True)
    parser.add_argument('--push-token', help="Pushover token", required=True)

    args = parser.parse_args()
    setup_logging()

    logger.info('Doorbell Notifier: Start')
    doorbell_notifier = DoorbellNotifier(args)

    signal.signal(signal.SIGINT, lambda signal, frame: signal_handler(doorbell_notifier))
    while True: time.sleep(1)