core/homeassistant/components/alarm_control_panel/egardia.py

"""
Interfaces with Egardia/Woonveilig alarm control panel.

For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/alarm_control_panel.egardia/
"""
import logging

import requests
import voluptuous as vol

import homeassistant.components.alarm_control_panel as alarm
import homeassistant.exceptions as exc
import homeassistant.helpers.config_validation as cv
from homeassistant.components.alarm_control_panel import PLATFORM_SCHEMA
from homeassistant.const import (
    CONF_PORT, CONF_HOST, CONF_PASSWORD, CONF_USERNAME, STATE_UNKNOWN,
    CONF_NAME, STATE_ALARM_DISARMED, STATE_ALARM_ARMED_HOME,
    STATE_ALARM_ARMED_AWAY, STATE_ALARM_TRIGGERED)

REQUIREMENTS = ['pythonegardia==1.0.22']

_LOGGER = logging.getLogger(__name__)

CONF_REPORT_SERVER_CODES = 'report_server_codes'
CONF_REPORT_SERVER_ENABLED = 'report_server_enabled'
CONF_REPORT_SERVER_PORT = 'report_server_port'
CONF_REPORT_SERVER_CODES_IGNORE = 'ignore'

DEFAULT_NAME = 'Egardia'
DEFAULT_PORT = 80
DEFAULT_REPORT_SERVER_ENABLED = False
DEFAULT_REPORT_SERVER_PORT = 52010
DOMAIN = 'egardia'

NOTIFICATION_ID = 'egardia_notification'
NOTIFICATION_TITLE = 'Egardia'

STATES = {
    'ARM': STATE_ALARM_ARMED_AWAY,
    'DAY HOME': STATE_ALARM_ARMED_HOME,
    'DISARM': STATE_ALARM_DISARMED,
    'HOME': STATE_ALARM_ARMED_HOME,
    'TRIGGERED': STATE_ALARM_TRIGGERED,
    'UNKNOWN': STATE_UNKNOWN,
}

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_HOST): cv.string,
    vol.Required(CONF_PASSWORD): cv.string,
    vol.Required(CONF_USERNAME): cv.string,
    vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
    vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
    vol.Optional(CONF_REPORT_SERVER_CODES): vol.All(cv.ensure_list),
    vol.Optional(CONF_REPORT_SERVER_ENABLED,
                 default=DEFAULT_REPORT_SERVER_ENABLED): cv.boolean,
    vol.Optional(CONF_REPORT_SERVER_PORT, default=DEFAULT_REPORT_SERVER_PORT):
        cv.port,
})


def setup_platform(hass, config, add_devices, discovery_info=None):
    """Set up the Egardia platform."""
    from pythonegardia import egardiadevice

    name = config.get(CONF_NAME)
    username = config.get(CONF_USERNAME)
    password = config.get(CONF_PASSWORD)
    host = config.get(CONF_HOST)
    port = config.get(CONF_PORT)
    rs_enabled = config.get(CONF_REPORT_SERVER_ENABLED)
    rs_port = config.get(CONF_REPORT_SERVER_PORT)
    rs_codes = config.get(CONF_REPORT_SERVER_CODES)

    try:
        egardiasystem = egardiadevice.EgardiaDevice(
            host, port, username, password, '')
    except requests.exceptions.RequestException:
        raise exc.PlatformNotReady()
    except egardiadevice.UnauthorizedError:
        _LOGGER.error("Unable to authorize. Wrong password or username")
        return False

    add_devices([EgardiaAlarm(
        name, egardiasystem, hass, rs_enabled, rs_port, rs_codes)], True)


class EgardiaAlarm(alarm.AlarmControlPanel):
    """Representation of a Egardia alarm."""

    def __init__(self, name, egardiasystem, hass, rs_enabled=False,
                 rs_port=None, rs_codes=None):
        """Initialize object."""
        self._name = name
        self._egardiasystem = egardiasystem
        self._status = STATE_UNKNOWN
        self._rs_enabled = rs_enabled
        self._rs_port = rs_port
        self._hass = hass

        if rs_codes is not None:
            self._rs_codes = rs_codes[0]
        else:
            self._rs_codes = rs_codes

        if self._rs_enabled:
            self.listen_to_system_status()

    @property
    def name(self):
        """Return the name of the device."""
        return self._name

    @property
    def state(self):
        """Return the state of the device."""
        return self._status

    def handle_system_status_event(self, event):
        """Handle egardia_system_status_event."""
        if event.data.get('status') is not None:
            statuscode = event.data.get('status')
            status = self.lookupstatusfromcode(statuscode)
            self.parsestatus(status)

    def listen_to_system_status(self):
        """Subscribe to egardia_system_status event."""
        self._hass.bus.listen(
            'egardia_system_status', self.handle_system_status_event)

    def lookupstatusfromcode(self, statuscode):
        """Look at the rs_codes and returns the status from the code."""
        status = 'UNKNOWN'
        if self._rs_codes is not None:
            statuscode = str(statuscode).strip()
            for i in self._rs_codes:
                val = str(self._rs_codes[i]).strip()
                if ',' in val:
                    splitted = val.split(',')
                    for code in splitted:
                        code = str(code).strip()
                        if statuscode == code:
                            status = i.upper()
                            break
                elif statuscode == val:
                    status = i.upper()
                    break
        return status

    def parsestatus(self, status):
        """Parse the status."""
        _LOGGER.debug("Parsing status %s", status)
        # Ignore the statuscode if it is IGNORE
        if status.lower().strip() != CONF_REPORT_SERVER_CODES_IGNORE:
            _LOGGER.debug("Not ignoring status")
            newstatus = ([v for k, v in STATES.items()
                          if status.upper() == k][0])
            self._status = newstatus
        else:
            _LOGGER.error("Ignoring status")

    def update(self):
        """Update the alarm status."""
        if not self._rs_enabled:
            status = self._egardiasystem.getstate()
            self.parsestatus(status)

    def alarm_disarm(self, code=None):
        """Send disarm command."""
        try:
            self._egardiasystem.alarm_disarm()
        except requests.exceptions.RequestException as err:
            _LOGGER.error("Egardia device exception occurred when "
                          "sending disarm command: %s", err)

    def alarm_arm_home(self, code=None):
        """Send arm home command."""
        try:
            self._egardiasystem.alarm_arm_home()
        except requests.exceptions.RequestException as err:
            _LOGGER.error("Egardia device exception occurred when "
                          "sending arm home command: %s", err)

    def alarm_arm_away(self, code=None):
        """Send arm away command."""
        try:
            self._egardiasystem.alarm_arm_away()
        except requests.exceptions.RequestException as err:
            _LOGGER.error("Egardia device exception occurred when "
                          "sending arm away command: %s", err)