Add uk_transport component. (home-assistant#8600)

homeassistant/components/sensor/uk_transport.py

@@ -0,0 +1,275 @@
+"""Support for UK public transport data provided by transportapi.com.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/sensor.uk_transport/
+"""
+import logging
+import re
+from datetime import datetime, timedelta
+import requests
+import voluptuous as vol
+
+from homeassistant.components.sensor import PLATFORM_SCHEMA
+from homeassistant.helpers.entity import Entity
+from homeassistant.util import Throttle
+import homeassistant.helpers.config_validation as cv
+
+_LOGGER = logging.getLogger(__name__)
+
+ATTR_ATCOCODE = 'atcocode'
+ATTR_LOCALITY = 'locality'
+ATTR_STOP_NAME = 'stop_name'
+ATTR_REQUEST_TIME = 'request_time'
+ATTR_NEXT_BUSES = 'next_buses'
+ATTR_STATION_CODE = 'station_code'
+ATTR_CALLING_AT = 'calling_at'
+ATTR_NEXT_TRAINS = 'next_trains'
+
+CONF_API_APP_KEY = 'app_key'
+CONF_API_APP_ID = 'app_id'
+CONF_QUERIES = 'queries'
+CONF_MODE = 'mode'
+CONF_ORIGIN = 'origin'
+CONF_DESTINATION = 'destination'
+
+_QUERY_SCHEME = vol.Schema({
+    vol.Required(CONF_MODE):
+        vol.All(cv.ensure_list, [vol.In(list(['bus', 'train']))]),
+    vol.Required(CONF_ORIGIN): cv.string,
+    vol.Required(CONF_DESTINATION): cv.string,
+})
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Required(CONF_API_APP_ID): cv.string,
+    vol.Required(CONF_API_APP_KEY): cv.string,
+    vol.Required(CONF_QUERIES): [_QUERY_SCHEME],
+})
+
+
+def setup_platform(hass, config, add_devices, discovery_info=None):
+    """Get the uk_transport sensor."""
+    sensors = []
+    number_sensors = len(config.get(CONF_QUERIES))
+    interval = timedelta(seconds=87*number_sensors)
+
+    for query in config.get(CONF_QUERIES):
+        if 'bus' in query.get(CONF_MODE):
+            stop_atcocode = query.get(CONF_ORIGIN)
+            bus_direction = query.get(CONF_DESTINATION)
+            sensors.append(
+                UkTransportLiveBusTimeSensor(
+                    config.get(CONF_API_APP_ID),
+                    config.get(CONF_API_APP_KEY),
+                    stop_atcocode,
+                    bus_direction,
+                    interval))
+
+        elif 'train' in query.get(CONF_MODE):
+            station_code = query.get(CONF_ORIGIN)
+            calling_at = query.get(CONF_DESTINATION)
+            sensors.append(
+                UkTransportLiveTrainTimeSensor(
+                    config.get(CONF_API_APP_ID),
+                    config.get(CONF_API_APP_KEY),
+                    station_code,
+                    calling_at,
+                    interval))
+
+    add_devices(sensors, True)
+
+
+class UkTransportSensor(Entity):
+    """
+    Sensor that reads the UK transport web API.
+
+    transportapi.com provides comprehensive transport data for UK train, tube
+    and bus travel across the UK via simple JSON API. Subclasses of this
+    base class can be used to access specific types of information.
+    """
+
+    TRANSPORT_API_URL_BASE = "https://transportapi.com/v3/uk/"
+    ICON = 'mdi:train'
+
+    def __init__(self, name, api_app_id, api_app_key, url):
+        """Initialize the sensor."""
+        self._data = {}
+        self._api_app_id = api_app_id
+        self._api_app_key = api_app_key
+        self._url = self.TRANSPORT_API_URL_BASE + url
+        self._name = name
+        self._state = None
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def state(self):
+        """Return the state of the sensor."""
+        return self._state
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit this state is expressed in."""
+        return "min"
+
+    @property
+    def icon(self):
+        """Icon to use in the frontend, if any."""
+        return self.ICON
+
+    def _do_api_request(self, params):
+        """Perform an API request."""
+        request_params = dict({
+            'app_id': self._api_app_id,
+            'app_key': self._api_app_key,
+        }, **params)
+
+        response = requests.get(self._url, params=request_params)
+        if response.status_code != 200:
+            _LOGGER.warning('Invalid response from API')
+        elif 'error' in response.json():
+            if 'exceeded' in response.json()['error']:
+                self._state = 'Useage limites exceeded'
+            if 'invalid' in response.json()['error']:
+                self._state = 'Credentials invalid'
+        else:
+            self._data = response.json()
+
+
+class UkTransportLiveBusTimeSensor(UkTransportSensor):
+    """Live bus time sensor from UK transportapi.com."""
+
+    ICON = 'mdi:bus'
+
+    def __init__(self, api_app_id, api_app_key,
+                 stop_atcocode, bus_direction, interval):
+        """Construct a live bus time sensor."""
+        self._stop_atcocode = stop_atcocode
+        self._bus_direction = bus_direction
+        self._next_buses = []
+        self._destination_re = re.compile(
+            '{}'.format(bus_direction), re.IGNORECASE
+        )
+
+        sensor_name = 'Next bus to {}'.format(bus_direction)
+        stop_url = 'bus/stop/{}/live.json'.format(stop_atcocode)
+
+        UkTransportSensor.__init__(
+            self, sensor_name, api_app_id, api_app_key, stop_url
+        )
+        self.update = Throttle(interval)(self._update)
+
+    def _update(self):
+        """Get the latest live departure data for the specified stop."""
+        params = {'group': 'route', 'nextbuses': 'no'}
+
+        self._do_api_request(params)
+
+        if self._data != {}:
+            self._next_buses = []
+
+            for (route, departures) in self._data['departures'].items():
+                for departure in departures:
+                    if self._destination_re.search(departure['direction']):
+                        self._next_buses.append({
+                            'route': route,
+                            'direction': departure['direction'],
+                            'scheduled': departure['aimed_departure_time'],
+                            'estimated': departure['best_departure_estimate']
+                        })
+
+            self._state = min(map(
+                _delta_mins, [bus['scheduled'] for bus in self._next_buses]
+            ))
+
+    @property
+    def device_state_attributes(self):
+        """Return other details about the sensor state."""
+        attrs = {}
+        if self._data is not None:
+            for key in [
+                    ATTR_ATCOCODE, ATTR_LOCALITY, ATTR_STOP_NAME,
+                    ATTR_REQUEST_TIME
+            ]:
+                attrs[key] = self._data.get(key)
+            attrs[ATTR_NEXT_BUSES] = self._next_buses
+            return attrs
+
+
+class UkTransportLiveTrainTimeSensor(UkTransportSensor):
+    """Live train time sensor from UK transportapi.com."""
+
+    ICON = 'mdi:train'
+
+    def __init__(self, api_app_id, api_app_key,
+                 station_code, calling_at, interval):
+        """Construct a live bus time sensor."""
+        self._station_code = station_code
+        self._calling_at = calling_at
+        self._next_trains = []
+
+        sensor_name = 'Next train to {}'.format(calling_at)
+        query_url = 'train/station/{}/live.json'.format(station_code)
+
+        UkTransportSensor.__init__(
+            self, sensor_name, api_app_id, api_app_key, query_url
+        )
+        self.update = Throttle(interval)(self._update)
+
+    def _update(self):
+        """Get the latest live departure data for the specified stop."""
+        params = {'darwin': 'false',
+                  'calling_at': self._calling_at,
+                  'train_status': 'passenger'}
+
+        self._do_api_request(params)
+        self._next_trains = []
+
+        if self._data != {}:
+            if self._data['departures']['all'] == []:
+                self._state = 'No departures'
+            else:
+                for departure in self._data['departures']['all']:
+                    self._next_trains.append({
+                        'origin_name': departure['origin_name'],
+                        'destination_name': departure['destination_name'],
+                        'status': departure['status'],
+                        'scheduled': departure['aimed_departure_time'],
+                        'estimated': departure['expected_departure_time'],
+                        'platform': departure['platform'],
+                        'operator_name': departure['operator_name']
+                        })
+
+                self._state = min(map(
+                    _delta_mins,
+                    [train['scheduled'] for train in self._next_trains]
+                ))
+
+    @property
+    def device_state_attributes(self):
+        """Return other details about the sensor state."""
+        attrs = {}
+        if self._data is not None:
+            attrs[ATTR_STATION_CODE] = self._station_code
+            attrs[ATTR_CALLING_AT] = self._calling_at
+            if self._next_trains:
+                attrs[ATTR_NEXT_TRAINS] = self._next_trains
+            return attrs
+
+
+def _delta_mins(hhmm_time_str):
+    """Calculate time delta in minutes to a time in hh:mm format."""
+    now = datetime.now()
+    hhmm_time = datetime.strptime(hhmm_time_str, '%H:%M')
+
+    hhmm_datetime = datetime(
+        now.year, now.month, now.day,
+        hour=hhmm_time.hour, minute=hhmm_time.minute
+    )
+    if hhmm_datetime < now:
+        hhmm_datetime += timedelta(days=1)
+
+    delta_mins = (hhmm_datetime - now).seconds // 60
+    return delta_mins

tests/components/sensor/test_uk_transport.py

@@ -0,0 +1,93 @@
+"""The tests for the uk_transport platform."""
+import re
+
+import requests_mock
+import unittest
+
+from homeassistant.components.sensor.uk_transport import (
+    UkTransportSensor,
+    ATTR_ATCOCODE, ATTR_LOCALITY, ATTR_STOP_NAME, ATTR_NEXT_BUSES,
+    ATTR_STATION_CODE, ATTR_CALLING_AT, ATTR_NEXT_TRAINS,
+    CONF_API_APP_KEY, CONF_API_APP_ID)
+from homeassistant.setup import setup_component
+from tests.common import load_fixture, get_test_home_assistant
+
+BUS_ATCOCODE = '340000368SHE'
+BUS_DIRECTION = 'Wantage'
+TRAIN_STATION_CODE = 'WIM'
+TRAIN_DESTINATION_NAME = 'WAT'
+
+VALID_CONFIG = {
+    'platform': 'uk_transport',
+    CONF_API_APP_ID: 'foo',
+    CONF_API_APP_KEY: 'ebcd1234',
+    'queries': [{
+      'mode': 'bus',
+      'origin': BUS_ATCOCODE,
+      'destination': BUS_DIRECTION},
+      {
+      'mode': 'train',
+      'origin': TRAIN_STATION_CODE,
+      'destination': TRAIN_DESTINATION_NAME}]
+      }
+
+
+class TestUkTransportSensor(unittest.TestCase):
+    """Test the uk_transport platform."""
+
+    def setUp(self):
+        """Initialize values for this testcase class."""
+        self.hass = get_test_home_assistant()
+        self.config = VALID_CONFIG
+
+    def tearDown(self):
+        """Stop everything that was started."""
+        self.hass.stop()
+
+    @requests_mock.Mocker()
+    def test_bus(self, mock_req):
+        """Test for operational uk_transport sensor with proper attributes."""
+        with requests_mock.Mocker() as mock_req:
+            uri = re.compile(UkTransportSensor.TRANSPORT_API_URL_BASE + '*')
+            mock_req.get(uri, text=load_fixture('uk_transport_bus.json'))
+            self.assertTrue(
+                setup_component(self.hass, 'sensor', {'sensor': self.config}))
+
+        bus_state = self.hass.states.get('sensor.next_bus_to_wantage')
+
+        assert type(bus_state.state) == str
+        assert bus_state.name == 'Next bus to {}'.format(BUS_DIRECTION)
+        assert bus_state.attributes.get(ATTR_ATCOCODE) == BUS_ATCOCODE
+        assert bus_state.attributes.get(ATTR_LOCALITY) == 'Harwell Campus'
+        assert bus_state.attributes.get(ATTR_STOP_NAME) == 'Bus Station'
+        assert len(bus_state.attributes.get(ATTR_NEXT_BUSES)) == 2
+
+        direction_re = re.compile(BUS_DIRECTION)
+        for bus in bus_state.attributes.get(ATTR_NEXT_BUSES):
+            print(bus['direction'], direction_re.match(bus['direction']))
+            assert direction_re.search(bus['direction']) is not None
+
+    @requests_mock.Mocker()
+    def test_train(self, mock_req):
+        """Test for operational uk_transport sensor with proper attributes."""
+        with requests_mock.Mocker() as mock_req:
+            uri = re.compile(UkTransportSensor.TRANSPORT_API_URL_BASE + '*')
+            mock_req.get(uri, text=load_fixture('uk_transport_train.json'))
+            self.assertTrue(
+                setup_component(self.hass, 'sensor', {'sensor': self.config}))
+
+        train_state = self.hass.states.get('sensor.next_train_to_WAT')
+
+        assert type(train_state.state) == str
+        assert train_state.name == 'Next train to {}'.format(
+            TRAIN_DESTINATION_NAME)
+        assert train_state.attributes.get(
+            ATTR_STATION_CODE) == TRAIN_STATION_CODE
+        assert train_state.attributes.get(
+            ATTR_CALLING_AT) == TRAIN_DESTINATION_NAME
+        assert len(train_state.attributes.get(ATTR_NEXT_TRAINS)) == 25
+
+        assert train_state.attributes.get(
+            ATTR_NEXT_TRAINS)[0]['destination_name'] == 'London Waterloo'
+        assert train_state.attributes.get(
+            ATTR_NEXT_TRAINS)[0]['estimated'] == '06:13'