forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add london_underground (home-assistant#8272)
* Add tube_state Add tube_state sensor * Final cleanup * Make corrections Correct PLATFORM_SCHEMA * Fix space * Make test pass * Correct format of test Test still failing, don’t understand why * correct description * Make test pass Preferred method below returns None state = self.hass.states.get('sensor.london_overground') * Format for hound * indent * Make requested changes to test, not working Test fails with: AssertionError: assert 0 > 0 where 0 = len([]) Surely I need tube_state.setup_platform ? * Fixed test Config was wrong * Change component name to london_tube * Update name to london_underground Make consistent * cleanup
- Loading branch information
1 parent
05ced33
commit 865865c
Showing
3 changed files
with
638 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,135 @@ | ||
""" | ||
Sensor for checking the status of London Underground tube lines. | ||
For more details about this component, please refer to the documentation at | ||
https://home-assistant.io/components/sensor.tube-state | ||
""" | ||
import logging | ||
from datetime import timedelta | ||
|
||
import voluptuous as vol | ||
import requests | ||
|
||
import homeassistant.helpers.config_validation as cv | ||
from homeassistant.components.sensor import PLATFORM_SCHEMA | ||
from homeassistant.helpers.entity import Entity | ||
from homeassistant.util import Throttle | ||
|
||
_LOGGER = logging.getLogger(__name__) | ||
|
||
ATTRIBUTION = "Powered by TfL Open Data" | ||
CONF_LINE = 'line' | ||
SCAN_INTERVAL = timedelta(seconds=30) | ||
TUBE_LINES = [ | ||
'Bakerloo', | ||
'Central', | ||
'Circle', | ||
'District', | ||
'DLR', | ||
'Hammersmith & City', | ||
'Jubilee', | ||
'London Overground', | ||
'Metropolitan', | ||
'Northern', | ||
'Piccadilly', | ||
'TfL Rail', | ||
'Victoria', | ||
'Waterloo & City'] | ||
URL = 'https://api.tfl.gov.uk/line/mode/tube,overground,dlr,tflrail/status' | ||
|
||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ | ||
vol.Required(CONF_LINE): | ||
vol.All(cv.ensure_list, [vol.In(list(TUBE_LINES))]), | ||
}) | ||
|
||
|
||
def setup_platform(hass, config, add_devices, discovery_info=None): | ||
"""Set up the Tube sensor.""" | ||
data = TubeData() | ||
data.update() | ||
sensors = [] | ||
for line in config.get(CONF_LINE): | ||
sensors.append(LondonTubeSensor(line, data)) | ||
|
||
add_devices(sensors, True) | ||
|
||
|
||
class LondonTubeSensor(Entity): | ||
"""Sensor that reads the status of a line from TubeData.""" | ||
|
||
ICON = 'mdi:subway' | ||
|
||
def __init__(self, name, data): | ||
"""Initialize the sensor.""" | ||
self._name = name | ||
self._data = data | ||
self._state = None | ||
self._description = 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 | ||
|
||
@property | ||
def device_state_attributes(self): | ||
"""Return other details about the sensor state.""" | ||
attrs = {} | ||
attrs['Description'] = self._description | ||
return attrs | ||
|
||
def update(self): | ||
"""Update the sensor.""" | ||
self._data.update() | ||
self._state = self._data.data[self.name]['State'] | ||
self._description = self._data.data[self.name]['Description'] | ||
|
||
|
||
class TubeData(object): | ||
"""Get the latest tube data from TFL.""" | ||
|
||
def __init__(self): | ||
"""Initialize the TubeData object.""" | ||
self.data = None | ||
|
||
# Update only once in scan interval. | ||
@Throttle(SCAN_INTERVAL) | ||
def update(self): | ||
"""Get the latest data from TFL.""" | ||
response = requests.get(URL) | ||
if response.status_code != 200: | ||
_LOGGER.warning("Invalid response from API") | ||
else: | ||
self.data = parse_api_response(response.json()) | ||
|
||
|
||
def parse_api_response(response): | ||
"""Take in the TFL API json response.""" | ||
lines = [line['name'] for line in response] | ||
data_dict = dict.fromkeys(lines) | ||
|
||
for line in response: | ||
statuses = [status['statusSeverityDescription'] | ||
for status in line['lineStatuses']] | ||
state = ' + '.join(sorted(set(statuses))) | ||
|
||
if state == 'Good Service': | ||
reason = 'Nothing to report' | ||
else: | ||
reason = ' *** '.join( | ||
[status['reason'] for status in line['lineStatuses']]) | ||
|
||
attr = {'State': state, 'Description': reason} | ||
data_dict[line['name']] = attr | ||
|
||
return data_dict |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,38 @@ | ||
"""The tests for the tube_state platform.""" | ||
import unittest | ||
import requests_mock | ||
|
||
from homeassistant.components.sensor.london_underground import CONF_LINE, URL | ||
from homeassistant.setup import setup_component | ||
from tests.common import load_fixture, get_test_home_assistant | ||
|
||
VALID_CONFIG = { | ||
'platform': 'london_underground', | ||
CONF_LINE: [ | ||
'London Overground', | ||
] | ||
} | ||
|
||
|
||
class TestLondonTubeSensor(unittest.TestCase): | ||
"""Test the tube_state 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_setup(self, mock_req): | ||
"""Test for operational tube_state sensor with proper attributes.""" | ||
mock_req.get(URL, text=load_fixture('london_underground.json')) | ||
self.assertTrue( | ||
setup_component(self.hass, 'sensor', {'sensor': self.config})) | ||
|
||
state = self.hass.states.get('sensor.london_overground') | ||
assert state.state == 'Minor Delays' | ||
assert state.attributes.get('Description') == 'something' |
Oops, something went wrong.