Added new Washington State DOT sensor. (#5496)

* Added new Washington State DOT sensor.

* Minor changes from review for WSDOT.

* Update wsdot.py

homeassistant/components/sensor/wsdot.py

@@ -0,0 +1,143 @@
+"""
+Support for Washington State Department of Transportation (WSDOT) data.
+
+Data provided by WSDOT is documented at http://wsdot.com/traffic/api/
+"""
+import logging
+import re
+from datetime import datetime, timezone, timedelta
+
+import requests
+import voluptuous as vol
+
+from homeassistant.components.sensor import PLATFORM_SCHEMA
+from homeassistant.const import (
+    CONF_API_KEY, CONF_NAME, ATTR_ATTRIBUTION, CONF_ID
+    )
+from homeassistant.helpers.entity import Entity
+import homeassistant.helpers.config_validation as cv
+
+
+_LOGGER = logging.getLogger(__name__)
+
+CONF_TRAVEL_TIMES = 'travel_time'
+
+# API codes for travel time details
+ATTR_ACCESS_CODE = 'AccessCode'
+ATTR_TRAVEL_TIME_ID = 'TravelTimeID'
+ATTR_CURRENT_TIME = 'CurrentTime'
+ATTR_AVG_TIME = 'AverageTime'
+ATTR_NAME = 'Name'
+ATTR_TIME_UPDATED = 'TimeUpdated'
+ATTR_DESCRIPTION = 'Description'
+ATTRIBUTION = "Data provided by WSDOT"
+
+SCAN_INTERVAL = timedelta(minutes=3)
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Required(CONF_API_KEY): cv.string,
+    vol.Optional(CONF_TRAVEL_TIMES): [{
+        vol.Required(CONF_ID): cv.string,
+        vol.Optional(CONF_NAME): cv.string}]
+})
+
+
+def setup_platform(hass, config, add_devices, discovery_info=None):
+    """Get the WSDOT sensor."""
+    sensors = []
+    for travel_time in config.get(CONF_TRAVEL_TIMES):
+        name = (travel_time.get(CONF_NAME) or
+                travel_time.get(CONF_ID))
+        sensors.append(
+            WashingtonStateTravelTimeSensor(
+                name,
+                config.get(CONF_API_KEY),
+                travel_time.get(CONF_ID)))
+    add_devices(sensors, True)
+
+
+class WashingtonStateTransportSensor(Entity):
+    """
+    Sensor that reads the WSDOT web API.
+
+    WSDOT provides ferry schedules, toll rates, weather conditions,
+    mountain pass conditions, and more. Subclasses of this
+    can read them and make them available.
+    """
+
+    ICON = 'mdi:car'
+
+    def __init__(self, name, access_code):
+        """Initialize the sensor."""
+        self._data = {}
+        self._access_code = access_code
+        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 icon(self):
+        """Icon to use in the frontend, if any."""
+        return self.ICON
+
+
+class WashingtonStateTravelTimeSensor(WashingtonStateTransportSensor):
+    """Travel time sensor from WSDOT."""
+
+    RESOURCE = ('http://www.wsdot.wa.gov/Traffic/api/TravelTimes/'
+                'TravelTimesREST.svc/GetTravelTimeAsJson')
+    ICON = 'mdi:car'
+
+    def __init__(self, name, access_code, travel_time_id):
+        """Construct a travel time sensor."""
+        self._travel_time_id = travel_time_id
+        WashingtonStateTransportSensor.__init__(self, name, access_code)
+
+    def update(self):
+        """Get the latest data from WSDOT."""
+        params = {ATTR_ACCESS_CODE: self._access_code,
+                  ATTR_TRAVEL_TIME_ID: self._travel_time_id}
+
+        response = requests.get(self.RESOURCE, params, timeout=10)
+        if response.status_code != 200:
+            _LOGGER.warning('Invalid response from WSDOT API.')
+        else:
+            self._data = response.json()
+        self._state = self._data.get(ATTR_CURRENT_TIME)
+
+    @property
+    def device_state_attributes(self):
+        """Return other details about the sensor state."""
+        if self._data is not None:
+            attrs = {ATTR_ATTRIBUTION: ATTRIBUTION}
+            for key in [ATTR_AVG_TIME, ATTR_NAME, ATTR_DESCRIPTION,
+                        ATTR_TRAVEL_TIME_ID]:
+                attrs[key] = self._data.get(key)
+            attrs[ATTR_TIME_UPDATED] = _parse_wsdot_timestamp(
+                self._data.get(ATTR_TIME_UPDATED))
+            return attrs
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit this state is expressed in."""
+        return "min"
+
+
+def _parse_wsdot_timestamp(timestamp):
+    """Convert WSDOT timestamp to datetime."""
+    if not timestamp:
+        return None
+    # ex: Date(1485040200000-0800)
+    milliseconds, tzone = re.search(
+        r'Date\((\d+)([+-]\d\d)\d\d\)', timestamp).groups()
+    return datetime.fromtimestamp(int(milliseconds) / 1000,
+                                  tz=timezone(timedelta(hours=int(tzone))))

tests/components/sensor/test_wsdot.py

@@ -0,0 +1,64 @@
+"""The tests for the WSDOT platform."""
+import re
+import unittest
+from datetime import timedelta, datetime, timezone
+
+import requests_mock
+
+from homeassistant.components.sensor import wsdot
+from homeassistant.components.sensor.wsdot import (
+    WashingtonStateTravelTimeSensor, ATTR_DESCRIPTION,
+    ATTR_TIME_UPDATED, CONF_API_KEY, CONF_NAME,
+    CONF_ID, CONF_TRAVEL_TIMES, SCAN_INTERVAL)
+from homeassistant.bootstrap import setup_component
+from tests.common import load_fixture, get_test_home_assistant
+
+
+class TestWSDOT(unittest.TestCase):
+    """Test the WSDOT platform."""
+
+    def add_entities(self, new_entities, update_before_add=False):
+        """Mock add entities."""
+        if update_before_add:
+            for entity in new_entities:
+                entity.update()
+
+        for entity in new_entities:
+            self.entities.append(entity)
+
+    def setUp(self):
+        """Initialize values for this testcase class."""
+        self.hass = get_test_home_assistant()
+        self.config = {
+            CONF_API_KEY: 'foo',
+            SCAN_INTERVAL: timedelta(seconds=120),
+            CONF_TRAVEL_TIMES: [{
+                CONF_ID: 96,
+                CONF_NAME: 'I90 EB'}],
+        }
+        self.entities = []
+
+    def tearDown(self):  # pylint: disable=invalid-name
+        """Stop everything that was started."""
+        self.hass.stop()
+
+    def test_setup_with_config(self):
+        """Test the platform setup with configuration."""
+        self.assertTrue(
+            setup_component(self.hass, 'sensor', {'wsdot': self.config}))
+
+    @requests_mock.Mocker()
+    def test_setup(self, mock_req):
+        """Test for operational WSDOT sensor with proper attributes."""
+        uri = re.compile(WashingtonStateTravelTimeSensor.RESOURCE + '*')
+        mock_req.get(uri, text=load_fixture('wsdot.json'))
+        wsdot.setup_platform(self.hass, self.config, self.add_entities)
+        self.assertEqual(len(self.entities), 1)
+        sensor = self.entities[0]
+        self.assertEqual(sensor.name, 'I90 EB')
+        self.assertEqual(sensor.state, 11)
+        self.assertEqual(sensor.device_state_attributes[ATTR_DESCRIPTION],
+                         'Downtown Seattle to Downtown Bellevue via I-90')
+        self.assertEqual(sensor.device_state_attributes[ATTR_TIME_UPDATED],
+                         datetime(2017, 1, 21, 15, 10,
+                                  tzinfo=timezone(timedelta(hours=-8))))

tests/fixtures/wsdot.json

@@ -0,0 +1,20 @@
+{"Description": "Downtown Seattle to Downtown Bellevue via I-90",
+ "TimeUpdated": "/Date(1485040200000-0800)/",
+ "Distance": 10.6,
+ "EndPoint": {"Direction": "N",
+  "Description": "I-405 @ NE 8th St in Bellevue",
+  "Longitude": -122.18797,
+  "MilePost": 13.6,
+  "Latitude": 47.61361,
+  "RoadName": "I-405"},
+ "StartPoint": {"Direction": "S",
+  "Description": "I-5 @ University St in Seattle",
+  "Longitude": -122.331759,
+  "MilePost": 165.83,
+  "Latitude": 47.609294,
+  "RoadName": "I-5"},
+ "CurrentTime": 11,
+ "TravelTimeID": 96,
+ "Name": "Seattle-Bellevue via I-90 (EB AM)",
+ "AverageTime": 11}
+