mirror of
https://github.com/home-assistant/core.git
synced 2025-05-31 03:07:06 +00:00
b096004449
Uses the (schiene)[https://pypi.python.org/pypi/schiene/0.14] API to communicate with the webserver of bahn.de and pulls iformation about a specific connection from the (bahn.de)[http://www.bahn.de/p/view/index.shtml] webpage. The departure time of the next train for the given connection is shown. In case of delay, the delay is also shown. Additional `ATTRIBUTES` are used to inform about e.g. the type of the train, price and if it is ontime. Usage: sensor: platform: deutsche_bahn from: name_of_start_station to: name_of_final_station Problems: I'm testing it for quite some time, but I have never seen the `ATTRIBUTES` in case of a delayed train. The `ATTRIBUTES` are directly passed from the `schiene` API. So this usecase has not been tested yet. deutsche_bahn ist not supporting the `schiene` api unlike in the swiss_public_transport case. It's not guaranteed that `schiene` will work forever, infact it can happen that Bahn AG will intentionally brake the API at some point. In the past Bahn AG has not allways been very supportive to the opensource community.
99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
'''
|
|
homeassistant.components.sensor.bahn
|
|
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
The deutsche_bahn sensor tells you if your next train is on time, or delayed.
|
|
|
|
'''
|
|
|
|
import logging
|
|
from datetime import timedelta, datetime
|
|
from homeassistant.util import Throttle
|
|
from homeassistant.helpers.entity import Entity
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
REQUIREMENTS = ['schiene==0.14']
|
|
|
|
ICON = 'mdi:train'
|
|
|
|
# Return cached results if last scan was less then this time ago
|
|
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=120)
|
|
|
|
|
|
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
|
|
""" Add the Bahn Sensor. """
|
|
start = config.get('from')
|
|
goal = config.get('to')
|
|
|
|
if start is None:
|
|
_LOGGER.error('Missing required variable: "from"')
|
|
return False
|
|
|
|
if goal is None:
|
|
_LOGGER.error('Missing required variable: "to"')
|
|
return False
|
|
|
|
dev = []
|
|
dev.append(DeutscheBahnSensor(start, goal))
|
|
add_devices_callback(dev)
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class DeutscheBahnSensor(Entity):
|
|
"""Implement a DeutscheBahn sensor
|
|
start: starting station
|
|
goal: target station"""
|
|
def __init__(self, start, goal):
|
|
self._name = start + ' to ' + goal
|
|
self.data = SchieneData(start, goal)
|
|
self.update()
|
|
|
|
@property
|
|
def name(self):
|
|
""" return the name."""
|
|
return self._name
|
|
|
|
@property
|
|
def icon(self):
|
|
""" Icon for the frontend"""
|
|
return ICON
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the depature time of the next train"""
|
|
return self._state
|
|
|
|
@property
|
|
def state_attributes(self):
|
|
return self.data.connections[0]
|
|
|
|
def update(self):
|
|
""" Gets the latest delay from bahn.de and updates the state"""
|
|
self.data.update()
|
|
self._state = self.data.connections[0].get('departure', 'Unknown')
|
|
delay = self.data.connections[0].get('delay',
|
|
{'delay_departure': 0,
|
|
'delay_arrival': 0})
|
|
if delay['delay_departure'] != 0:
|
|
self._state += " + {}".format(delay['delay_departure'])
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
class SchieneData(object):
|
|
""" Pulls data from the bahn.de web page"""
|
|
def __init__(self, start, goal):
|
|
import schiene
|
|
self.start = start
|
|
self.goal = goal
|
|
self.schiene = schiene.Schiene()
|
|
self.connections = [{}]
|
|
|
|
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
|
def update(self):
|
|
""" update connection data"""
|
|
self.connections = self.schiene.connections(self.start,
|
|
self.goal,
|
|
datetime.now())
|
|
for con in self.connections:
|
|
if 'details' in con:
|
|
con.pop('details') # details info is not usefull
|