homeassistant-config/custom_components/fhem/__init__.py

"""FHEM integration"""

import logging
import voluptuous as vol
import asyncio
from collections import defaultdict
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP

_LOGGER = logging.getLogger(__name__)
CONF_CUL_DEVICE_NAME = 'cul_device_name'
CONF_FHEM_SENSOR_TYPE = 'fhem_sensor_type'
CONF_FHEM_IDS = 'fhem_ids'
DOMAIN = 'fhem'
DATA_FHEM = "data_fhem"
CONFIG_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({
        vol.Required(CONF_HOST): cv.string,
        vol.Optional(CONF_PORT, default=7072): cv.port,
        vol.Required(CONF_CUL_DEVICE_NAME): cv.string,
    })
}, extra=vol.ALLOW_EXTRA)


async def async_setup(hass, config):
    connection = FhemConnection(hass, config)
    hass.data[DATA_FHEM] = connection
    await connection.start()
    return True


class FhemConnection:

    def __init__(self, hass, config):
        self.hass = hass
        self._host = config[DOMAIN][CONF_HOST]
        self._port = config[DOMAIN][CONF_PORT]
        self._cul_device_name = config[DOMAIN][CONF_CUL_DEVICE_NAME]

        self.connected = False
        self.reconnect_time_start = 1
        self.reconnect_time_max = 60
        self.reconnect_time = self.reconnect_time_start
        self._devices = defaultdict(list)
        self._run = False
        self._writer = None
        self._connection_last_state = 'UNKNOWN'

    def register_device(self, id, d):
        self._devices[id].append(d)
        if self._writer:
            self._writer.writelines([
                "displayattr .*\n".encode(),
            ])

    async def _update_all_devices(self):
        for device_list in self._devices.values():
            for device in device_list:
                await device.async_update_ha_state()

    async def start(self):
        self._run = True
        self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.stop)
        self.hass.loop.create_task(self._connection())

    async def stop(self):
        self._run = False
        self.connected = False

    async def _connection(self):
        try:
            reader, writer = await asyncio.open_connection(self._host, self._port)
            _LOGGER.info("Connected to FHEM {}:{}".format(self._host, self._port))
            self._connection_last_state = 'CONNECTED'

            self._writer = writer
            self.connected = True
            await self._update_all_devices()

            self.reconnect_time = self.reconnect_time_start
            writer.writelines([
                "displayattr .*\n".encode(),
                "inform on\n".encode(),
            ])
            while self._run:
                line = await reader.readline()
                if not line:
                    _LOGGER.warning("FHEM disconnected: {}".format(line))
                    raise OSError("Disconnect")
                line = line.decode()
                _LOGGER.debug("FHEM received line: {}".format(line))
                await self._process_line(line)
        except OSError:
            if self._connection_last_state != 'FAILED':
                self.hass.components.persistent_notification.async_create("FHEM connection failed",
                                                                          title="No FHEM connection")
                _LOGGER.error("Connection to FHEM failed {}:{}".format(self._host, self._port))
            self._connection_last_state = 'FAILED'

            self.connected = False
            await self._update_all_devices()
            await asyncio.sleep(self.reconnect_time)
            self.reconnect_time = min(2 * self.reconnect_time, self.reconnect_time_max)
            self.hass.loop.create_task(self._connection())

    async def _process_line(self, line):
        if line.startswith(self._cul_device_name + " "):  # Status update message
            _, device_name, command = line.split(" ", 2)
            for device in self._devices[device_name]:
                _LOGGER.debug("FHEM line received (device): " + device_name + ": " + line)
                await device.line_received(command.strip())
        else:  # potential response to displayattr
            split_line = line.split(" ", 1)
            if len(split_line) == 2:
                device_name, command = split_line
                for device in self._devices[device_name]:
                    await device.line_received(command.strip())

    def write_line(self, line):
        if self._writer:
            line += '\n'
            self._writer.write(line.encode())

    def fhem_set(self, id, *arguments):
        """
        Send command to FHEM using this device
            :param arguments: string or list of strings containing command parameters
        """
        arguments = " ".join([str(a) for a in arguments])
        self._writer.write("set {} {}\n".format(id, arguments).encode())


def device_error_reporting(hass, received_line, component_type, component_name):
    if received_line.startswith('overheat'):
        overheat = received_line.split(':')[1]
        overheat = overheat.strip().lower()
        assert overheat == 'on' or overheat == 'off'
        if overheat == 'on':
            text = "FHEM: {} overheated: <br><b>{}</b>".format(component_type, component_name)
            hass.components.persistent_notification.async_create(text, title="{} overheat".format(component_type))
    elif received_line.startswith('overload'):
        overload = received_line.split(':')[1]
        overload = overload.strip().lower()
        assert overload == 'on' or overload == 'off'
        if overload == 'on':
            text = "FHEM: {} overloaded: <br><b>{}</b>".format(component_type, component_name)
            hass.components.persistent_notification.async_create(text, title="{} overloaded".format(component_type))