diff --git a/homeassistant/components/sensor/wsdot.py b/homeassistant/components/sensor/wsdot.py new file mode 100644 index 00000000000..a3230a88eb3 --- /dev/null +++ b/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)))) diff --git a/tests/components/sensor/test_wsdot.py b/tests/components/sensor/test_wsdot.py new file mode 100644 index 00000000000..4a2dc345f10 --- /dev/null +++ b/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)))) diff --git a/tests/fixtures/wsdot.json b/tests/fixtures/wsdot.json new file mode 100644 index 00000000000..de5dc80579f --- /dev/null +++ b/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} +