Last active
August 27, 2018 22:12
-
-
Save jcconnell/64849ce3409fa1ee0270b108ab04e6d5 to your computer and use it in GitHub Desktop.
This is a custom component for Home Assistant that provides Sunrail train status. Instructions in first comment
This file contains hidden or 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
""" | |
Custom Sensor for Sunrail train times. | |
""" | |
from collections import defaultdict | |
from datetime import timedelta, datetime, time | |
import logging | |
import voluptuous as vol | |
from homeassistant.components.sensor import PLATFORM_SCHEMA | |
from homeassistant.const import (ATTR_ATTRIBUTION, CONF_NAME) | |
from homeassistant.helpers.entity import Entity | |
from homeassistant.util import slugify | |
from homeassistant.util.distance import convert | |
from homeassistant.util.dt import now | |
import homeassistant.helpers.config_validation as cv | |
REQUIREMENTS = ['sunrail==1.0.2'] | |
_LOGGER = logging.getLogger(__name__) | |
CONF_ATTRIBUTION = "Provided by Sunrail.com" | |
CONF_DIRECTION = 'direction' | |
CONF_INCLUDE_STATIONS = 'include_stations' | |
CONF_INCLUDE_TRAINS = 'include_trains' | |
CONF_EXCLUDE_STATIONS = 'exclude_stations' | |
CONF_EXCLUDE_TRAINS = 'exclude_trains' | |
DOMAIN = 'Sunrail' | |
EVENT_INCIDENT = '{}_incident'.format(DOMAIN) | |
SCAN_INTERVAL = timedelta(minutes=5) | |
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ | |
vol.Required(CONF_NAME): cv.string, | |
vol.Optional(CONF_DIRECTION): cv.string, | |
vol.Optional(CONF_INCLUDE_STATIONS): vol.All(cv.ensure_list, [cv.string]), | |
vol.Optional(CONF_INCLUDE_TRAINS): vol.All(cv.ensure_list, [cv.string]), | |
vol.Optional(CONF_EXCLUDE_STATIONS): vol.All(cv.ensure_list, [cv.string]), | |
vol.Optional(CONF_EXCLUDE_TRAINS): vol.All(cv.ensure_list, [cv.string]), | |
}) | |
def setup_platform(hass, config, add_devices, discovery_info=None): | |
"""Set up the Sunrail platform.""" | |
name = config.get(CONF_NAME) | |
direction = config.get(CONF_DIRECTION) | |
include_stations = config.get(CONF_INCLUDE_STATIONS) | |
include_trains = config.get(CONF_INCLUDE_TRAINS) | |
exclude_stations = config.get(CONF_EXCLUDE_STATIONS) | |
exclude_trains = config.get(CONF_EXCLUDE_TRAINS) | |
sunrail_sensor = SunrailSensor(hass, name, direction, | |
include_stations, | |
include_trains, | |
exclude_stations, | |
exclude_trains) | |
sunrail_sensor.update() | |
if sunrail_sensor.state is None: | |
_LOGGER.error("Unable to setup Sunrail sensor.") | |
return | |
add_devices([sunrail_sensor], True) | |
class SunrailSensor(Entity): | |
"""Representation of a Sunrail Sensor.""" | |
def __init__(self, hass, name, direction, | |
include_stations, include_trains, | |
exclude_stations, exclude_trains): | |
"""Initialize the Sunrail sensor.""" | |
import sunrail | |
self._hass = hass | |
self._name = name | |
self._direction = direction | |
self._include_stations = include_stations | |
self._include_trains = include_trains | |
self._exclude_stations = exclude_stations | |
self._exclude_trains = exclude_trains | |
self._sunrail = sunrail.SunRail( | |
include_stations=self._include_stations, | |
exclude_stations=self._exclude_stations, | |
include_trains=self._include_trains, | |
exclude_trains=self._exclude_trains, | |
direction=self._direction) | |
self._attributes = None | |
self._state = None | |
self._previous_alerts = set() | |
@property | |
def name(self): | |
"""Return the name of the sensor.""" | |
if datetime.now().time() < time(10,00): | |
return "South {}".format(self._name) | |
return "North {}".format(self._name) | |
@property | |
def state(self): | |
"""Return the state of the sensor.""" | |
return self._state | |
@property | |
def device_state_attributes(self): | |
"""Return the state attributes.""" | |
return self._attributes | |
def _alert_event(self, alert): | |
"""Fire if an event occurs.""" | |
data = { | |
'id': alert.get('id'), | |
'message': alert.get('message'), | |
'timestamp': alert.get('currentdate'), | |
} | |
self._hass.bus.fire(EVENT_INCIDENT, data) | |
def update(self): | |
"""Update device state.""" | |
self._sunrail.update() | |
status = self._sunrail.get_next() | |
alerts = self._sunrail.get_alerts() | |
self._state = "ON-TIME" if alerts is None else "DELAYED" | |
attr = { ATTR_ATTRIBUTION: CONF_ATTRIBUTION } | |
if alerts is not None: | |
for alert in alerts: | |
if alert.get('id') not in self._previous_alerts: | |
self._alert_event(alert) | |
self._previous_alerts.add(alert.get('id')) | |
attr.update({'alert': alert.get('message')}) | |
if datetime.now().time() < time(10,00): | |
for station in status.get('S'): | |
attr.update({station['station']:station['arrival_time']}) | |
else: | |
for station in status.get('N'): | |
attr.update({station['station']:station['arrival_time']}) | |
self._attributes = attr |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Last updated: 8/27/18
Info
Add this file to your Home Assistant configuration custom component directory. eg:
~/.homeassistant/custom_components/sensor/sunrail_status.py
Update your Home Assistant
configuration.yaml
file. Here's an example:More configuration options are available:
include_stations
,exclude_stations
,include_trains
,exclude_trains
,direction
Notes
I typically ride South in the morning and return North after 10AM. If your schedule is different, you may want to adjust the time on line
132
.Improvements
I plan to make this location aware using the device_tracker platform