The Things Network sensor for Home Assistant

thingsnetwork.py

"""
Sensor platform for Home Assistant that fetches data from the Things network.

Home Assistant: https://home-assistant.io
The Things Network: http://thethingsnetwork.org/


To use:
Copy this file to <config>/custom_components/sensor/thingsnetwork.py
Add to <config>/configuration.yaml:

sensor:
  platform: thingsnetwork
  node: 0000EF04
  variables:
    temp:
      name: Cool Temperature
      unit: "\u00b0C"
    voltage:
      unit: V
"""
import logging
import requests
from datetime import timedelta

from homeassistant.util import Throttle
from homeassistant.helpers.entity import Entity

_LOGGER = logging.getLogger(__name__)

# Return cached results if last scan was less then this time ago
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)

THING_URL = 'http://thethingsnetwork.org/api/v0/nodes/{}/?limit=1'

CONF_NODE = 'node'
CONF_VARIABLES = 'variables'
CONF_NAME = 'name'
CONF_UNIT = 'unit'


def setup_platform(hass, config, add_devices, discovery_info=None):
    """ Get the things network sensor. """

    node = config.get(CONF_NODE)
    variables = config.get(CONF_VARIABLES)

    if None in (node, variables) or not isinstance(variables, dict):
        _LOGGER.error('Did not find required string %s and dict %s', CONF_NODE,
                      CONF_VARIABLES)
        return

    thing = ThingData(node)
    thing.update()

    if thing.update_error:
        _LOGGER.error('Error retrieving data for Node from Things Network')
        return

    add_devices(
        ThingSensor(info.get(CONF_NAME, '{} {}'.format(node, variable)),
                    thing, variable, info.get(CONF_UNIT))
        for variable, info in variables.items())


# pylint: disable=too-many-instance-attributes, too-many-arguments
class ThingSensor(Entity):
    """ Implements Thing sensor for picked variables. """

    def __init__(self, name, thing, variable, unit):
        self._name = name
        self._thing = thing
        self._variable = variable
        self._unit = unit

    @property
    def name(self):
        """ The name of the sensor. """
        return self._name

    @property
    def unit_of_measurement(self):
        """ Unit the value is expressed in. """
        return self._unit

    @property
    def state(self):
        """ Returns the state of the device. """
        if self._thing.update_error:
            return 'error'
        state = self._thing.data.get(self._variable, 'n/a')

        return round(state, 2) if isinstance(state, float) else state

    def update(self):
        """ Gets the latest data from Thing API. """
        self._thing.update()


# pylint: disable=too-few-public-methods
class ThingData(object):
    """ Implements Things Network rest API v0. """

    def __init__(self, node_euid):
        self.node_euid = node_euid
        self.url = THING_URL.format(node_euid)
        self.update_error = False
        self.data = {}

    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        """ Gets the latest data from Things Network. """
        try:
            req = requests.get(self.url)
        except requests.exceptions.RequestException:
            _LOGGER.error('Error fetching thing info for %s', self.node_euid)
            self.update_error = True
            return

        try:
            data = req.json()
        except ValueError:
            _LOGGER.error('Unable to parse %s data as JSON: %s',
                          self.node_euid, req.text)
            self.update_error = True
            return

        if len(data) == 0:
            _LOGGER.error('No recorded reports found for %s', self.node_euid)
            self.update_error = True
            return

        latest = data[0]

        if 'data_json' not in latest:
            _LOGGER.error('Latest report has no data_json key.')
            self.update_error = True
            return

        self.data = latest['data_json']
        self.update_error = False