Skip to content

Instantly share code, notes, and snippets.

@jcconnell
Last active August 27, 2018 22:12
Show Gist options
  • Save jcconnell/64849ce3409fa1ee0270b108ab04e6d5 to your computer and use it in GitHub Desktop.
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
"""
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
@jcconnell
Copy link
Author

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:

sensor:
  - platform: sunrail_status
    name: Sunrail

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment