Merge 273457f into 2b52f27

homeassistant/components/sensor/entur_public_transport.py

@@ -0,0 +1,193 @@
+"""
+Real-time information about public transport departures in Norway.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/sensor.entur_public_transport/
+"""
+from datetime import datetime, timedelta
+import logging
+
+import voluptuous as vol
+
+from homeassistant.components.sensor import PLATFORM_SCHEMA
+from homeassistant.const import (
+    ATTR_ATTRIBUTION, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME,
+    CONF_SHOW_ON_MAP)
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity import Entity
+from homeassistant.util import Throttle
+import homeassistant.util.dt as dt_util
+
+REQUIREMENTS = ['enturclient==0.1.0']
+
+_LOGGER = logging.getLogger(__name__)
+
+ATTR_NEXT_UP_IN = 'next_due_in'
+
+API_CLIENT_NAME = 'homeassistant-homeassistant'
+
+CONF_ATTRIBUTION = "Data provided by entur.org under NLOD."
+CONF_STOP_IDS = 'stop_ids'
+CONF_EXPAND_PLATFORMS = 'expand_platforms'
+
+DEFAULT_NAME = 'Entur'
+DEFAULT_ICON_KEY = 'bus'
+
+ICONS = {
+    'air': 'mdi:airplane',
+    'bus': 'mdi:bus',
+    'rail': 'mdi:train',
+    'water': 'mdi:ferry',
+}
+
+SCAN_INTERVAL = timedelta(minutes=1)
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Required(CONF_STOP_IDS): vol.All(cv.ensure_list, [cv.string]),
+    vol.Optional(CONF_EXPAND_PLATFORMS, default=True): cv.boolean,
+    vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+    vol.Optional(CONF_SHOW_ON_MAP, default=False): cv.boolean,
+})
+
+
+def due_in_minutes(timestamp: str) -> str:
+    """Get the time in minutes from a timestamp.
+
+    The timestamp should be in the format
+    year-month-yearThour:minute:second+timezone
+    """
+    if timestamp is None:
+        return None
+    diff = datetime.strptime(
+        timestamp, "%Y-%m-%dT%H:%M:%S%z") - dt_util.now()
+
+    return str(int(diff.total_seconds() / 60))
+
+
+def setup_platform(hass, config, add_entities, discovery_info=None):
+    """Set up the Entur public transport sensor."""
+    from enturclient import EnturPublicTransportData
+    from enturclient.consts import CONF_NAME as API_NAME
+
+    expand = config.get(CONF_EXPAND_PLATFORMS)
+    name = config.get(CONF_NAME)
+    show_on_map = config.get(CONF_SHOW_ON_MAP)
+    stop_ids = config.get(CONF_STOP_IDS)
+
+    stops = [s for s in stop_ids if "StopPlace" in s]
+    quays = [s for s in stop_ids if "Quay" in s]
+
+    data = EnturPublicTransportData(API_CLIENT_NAME, stops, quays, expand)
+    data.update()
+
+    proxy = EnturProxy(data)
+
+    entities = []
+    for item in data.all_stop_places_quays():
+        try:
+            given_name = "{} {}".format(
+                name, data.get_stop_info(item)[API_NAME])
+        except KeyError:
+            given_name = "{} {}".format(name, item)
+
+        entities.append(
+            EnturPublicTransportSensor(proxy, given_name, item, show_on_map))
+
+    add_entities(entities, True)
+
+
+class EnturProxy:
+    """Proxy for the Entur client.
+
+    Ensure throttle to not hit rate limiting on the API.
+    """
+
+    def __init__(self, api):
+        """Initialize the proxy."""
+        self._api = api
+
+    @Throttle(SCAN_INTERVAL)
+    def update(self) -> None:
+        """Update data in client."""
+        self._api.update()
+
+    def get_stop_info(self, stop_id: str) -> dict:
+        """Get info about specific stop place."""
+        return self._api.get_stop_info(stop_id)
+
+
+class EnturPublicTransportSensor(Entity):
+    """Implementation of a Entur public transport sensor."""
+
+    def __init__(
+            self, api: EnturProxy, name: str, stop: str, show_on_map: bool):
+        """Initialize the sensor."""
+        from enturclient.consts import ATTR_STOP_ID
+
+        self.api = api
+        self._stop = stop
+        self._show_on_map = show_on_map
+        self._name = name
+        self._data = None
+        self._state = None
+        self._icon = ICONS[DEFAULT_ICON_KEY]
+        self._attributes = {
+            ATTR_ATTRIBUTION: CONF_ATTRIBUTION,
+            ATTR_STOP_ID: self._stop,
+        }
+
+    @property
+    def name(self) -> str:
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def state(self) -> str:
+        """Return the state of the sensor."""
+        return self._state
+
+    @property
+    def device_state_attributes(self) -> dict:
+        """Return the state attributes."""
+        return self._attributes
+
+    @property
+    def unit_of_measurement(self) -> str:
+        """Return the unit this state is expressed in."""
+        return 'min'
+
+    @property
+    def icon(self) -> str:
+        """Icon to use in the frontend."""
+        return self._icon
+
+    def update(self) -> None:
+        """Get the latest data and update the states."""
+        from enturclient.consts import (
+            ATTR, ATTR_EXPECTED_AT, ATTR_NEXT_UP_AT, CONF_LOCATION,
+            CONF_LATITUDE as LAT, CONF_LONGITUDE as LONG, CONF_TRANSPORT_MODE)
+
+        self.api.update()
+
+        self._data = self.api.get_stop_info(self._stop)
+        if self._data is not None:
+            attrs = self._data[ATTR]
+            self._attributes.update(attrs)
+
+            if ATTR_NEXT_UP_AT in attrs:
+                self._attributes[ATTR_NEXT_UP_IN] = \
+                    due_in_minutes(attrs[ATTR_NEXT_UP_AT])
+
+            if CONF_LOCATION in self._data and self._show_on_map:
+                self._attributes[CONF_LATITUDE] = \
+                    self._data[CONF_LOCATION][LAT]
+                self._attributes[CONF_LONGITUDE] = \
+                    self._data[CONF_LOCATION][LONG]
+
+            if ATTR_EXPECTED_AT in attrs:
+                self._state = due_in_minutes(attrs[ATTR_EXPECTED_AT])
+            else:
+                self._state = None
+
+            self._icon = ICONS.get(
+                self._data[CONF_TRANSPORT_MODE], ICONS[DEFAULT_ICON_KEY])

requirements_all.txt

@@ -338,6 +338,9 @@ elkm1-lib==0.7.12
 # homeassistant.components.enocean
 enocean==0.40
 
+# homeassistant.components.sensor.entur_public_transport
+enturclient==0.1.0
+
 # homeassistant.components.sensor.envirophat
 # envirophat==0.0.6
 

requirements_test_all.txt

@@ -58,6 +58,9 @@ defusedxml==0.5.0
 # homeassistant.components.sensor.dsmr
 dsmr_parser==0.12
 
+# homeassistant.components.sensor.entur_public_transport
+enturclient==0.1.0
+
 # homeassistant.components.sensor.season
 ephem==3.7.6.0
 

script/gen_requirements_all.py

@@ -46,6 +46,7 @@
     'coinmarketcap',
     'defusedxml',
     'dsmr_parser',
+    'enturclient',
     'ephem',
     'evohomeclient',
     'feedparser',

tests/components/sensor/test_entur_public_transport.py

@@ -0,0 +1,66 @@
+"""The tests for the entur platform."""
+from datetime import datetime
+import unittest
+from unittest.mock import patch
+
+from enturclient.api import RESOURCE
+from enturclient.consts import ATTR_EXPECTED_AT, ATTR_ROUTE, ATTR_STOP_ID
+import requests_mock
+
+from homeassistant.components.sensor.entur_public_transport import (
+    CONF_EXPAND_PLATFORMS, CONF_STOP_IDS)
+from homeassistant.setup import setup_component
+import homeassistant.util.dt as dt_util
+
+from tests.common import get_test_home_assistant, load_fixture
+
+VALID_CONFIG = {
+    'platform': 'entur_public_transport',
+    CONF_EXPAND_PLATFORMS: False,
+    CONF_STOP_IDS: [
+        'NSR:StopPlace:548',
+        'NSR:Quay:48550',
+    ]
+}
+
+FIXTURE_FILE = 'entur_public_transport.json'
+TEST_TIMESTAMP = datetime(2018, 10, 10, 7, tzinfo=dt_util.UTC)
+
+
+class TestEnturPublicTransportSensor(unittest.TestCase):
+    """Test the entur platform."""
+
+    def setUp(self):
+        """Initialize values for this testcase class."""
+        self.hass = get_test_home_assistant()
+
+    def tearDown(self):
+        """Stop everything that was started."""
+        self.hass.stop()
+
+    @requests_mock.Mocker()
+    @patch(
+        'homeassistant.components.sensor.entur_public_transport.dt_util.now',
+        return_value=TEST_TIMESTAMP)
+    def test_setup(self, mock_req, mock_patch):
+        """Test for correct sensor setup with state and proper attributes."""
+        mock_req.post(RESOURCE,
+                      text=load_fixture(FIXTURE_FILE),
+                      status_code=200)
+        self.assertTrue(
+            setup_component(self.hass, 'sensor', {'sensor': VALID_CONFIG}))
+
+        state = self.hass.states.get('sensor.entur_bergen_stasjon')
+        assert state.state == '28'
+        assert state.attributes.get(ATTR_STOP_ID) == 'NSR:StopPlace:548'
+        assert state.attributes.get(ATTR_ROUTE) == "59 Bergen"
+        assert state.attributes.get(ATTR_EXPECTED_AT) \
+            == '2018-10-10T09:28:00+0200'
+
+        state = self.hass.states.get('sensor.entur_fiskepiren_platform_2')
+        assert state.state == '0'
+        assert state.attributes.get(ATTR_STOP_ID) == 'NSR:Quay:48550'
+        assert state.attributes.get(ATTR_ROUTE) \
+            == "5 Stavanger Airport via Forum"
+        assert state.attributes.get(ATTR_EXPECTED_AT) \
+            == '2018-10-10T09:00:00+0200'