evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Migrate to voluptuous (#3182)

Browse source 
🐬
This commit is contained in:
Fabian Affolter 2016-09-04 18:27:19 +02:00 • committed by Teagan Glenn
parent e9813b219e
commit cd67368bb7
1 changed files with 89 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

177
homeassistant/components/sensor/gtfs.py
View file

@ -9,29 +9,47 @@ import logging
import datetime import datetime
import threading import threading
from homeassistant.helpers.entity import Entity import voluptuous as vol
_LOGGER = logging.getLogger(__name__) from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_NAME
from homeassistant.helpers.entity import Entity
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ["https://github.com/robbiet480/pygtfs/archive/" REQUIREMENTS = ["https://github.com/robbiet480/pygtfs/archive/"
"00546724e4bbcb3053110d844ca44e2246267dd8.zip#" "00546724e4bbcb3053110d844ca44e2246267dd8.zip#"
"pygtfs==0.1.3"] "pygtfs==0.1.3"]
ICON = "mdi:train" _LOGGER = logging.getLogger(__name__)
CONF_DATA = 'data'
CONF_DESTINATION = 'destination'
CONF_ORIGIN = 'origin'
DEFAULT_NAME = 'GTFS Sensor'
DEFAULT_PATH = 'gtfs'
ICON = 'mdi:train'
TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_ORIGIN): cv.string,
vol.Required(CONF_DESTINATION): cv.string,
vol.Required(CONF_DATA): cv.isfile,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
})
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
# pylint: disable=too-many-locals # pylint: disable=too-many-locals
def get_next_departure(sched, start_station_id, end_station_id): def get_next_departure(sched, start_station_id, end_station_id):
"""Get the next departure for the given sched.""" """Get the next departure for the given sched."""
origin_station = sched.stops_by_id(start_station_id)[0] origin_station = sched.stops_by_id(start_station_id)[0]
destination_station = sched.stops_by_id(end_station_id)[0] destination_station = sched.stops_by_id(end_station_id)[0]
now = datetime.datetime.now() now = datetime.datetime.now()
day_name = now.strftime("%A").lower() day_name = now.strftime('%A').lower()
now_str = now.strftime("%H:%M:%S") now_str = now.strftime('%H:%M:%S')
from sqlalchemy.sql import text from sqlalchemy.sql import text
@ -78,9 +96,9 @@ def get_next_departure(sched, start_station_id, end_station_id):
for row in result: for row in result:
item = row item = row
today = datetime.datetime.today().strftime("%Y-%m-%d") today = datetime.datetime.today().strftime('%Y-%m-%d')
departure_time_string = "{} {}".format(today, item[2]) departure_time_string = '{} {}'.format(today, item[2])
arrival_time_string = "{} {}".format(today, item[3]) arrival_time_string = '{} {}'.format(today, item[3])
departure_time = datetime.datetime.strptime(departure_time_string, departure_time = datetime.datetime.strptime(departure_time_string,
TIME_FORMAT) TIME_FORMAT)
arrival_time = datetime.datetime.strptime(arrival_time_string, arrival_time = datetime.datetime.strptime(arrival_time_string,
@ -91,72 +109,61 @@ def get_next_departure(sched, start_station_id, end_station_id):
route = sched.routes_by_id(item[1])[0] route = sched.routes_by_id(item[1])[0]
origin_stoptime_arrival_time = "{} {}".format(today, item[4]) origin_stoptime_arrival_time = '{} {}'.format(today, item[4])
origin_stoptime_departure_time = '{} {}'.format(today, item[5])
origin_stoptime_departure_time = "{} {}".format(today, item[5]) dest_stoptime_arrival_time = '{} {}'.format(today, item[11])
dest_stoptime_depart_time = '{} {}'.format(today, item[12])
dest_stoptime_arrival_time = "{} {}".format(today, item[11])
dest_stoptime_depart_time = "{} {}".format(today, item[12])
origin_stop_time_dict = { origin_stop_time_dict = {
"Arrival Time": origin_stoptime_arrival_time, 'Arrival Time': origin_stoptime_arrival_time,
"Departure Time": origin_stoptime_departure_time, 'Departure Time': origin_stoptime_departure_time,
"Drop Off Type": item[6], "Pickup Type": item[7], 'Drop Off Type': item[6], 'Pickup Type': item[7],
"Shape Dist Traveled": item[8], "Headsign": item[9], 'Shape Dist Traveled': item[8], 'Headsign': item[9],
"Sequence": item[10] 'Sequence': item[10]
} }
destination_stop_time_dict = { destination_stop_time_dict = {
"Arrival Time": dest_stoptime_arrival_time, 'Arrival Time': dest_stoptime_arrival_time,
"Departure Time": dest_stoptime_depart_time, 'Departure Time': dest_stoptime_depart_time,
"Drop Off Type": item[13], "Pickup Type": item[14], 'Drop Off Type': item[13], 'Pickup Type': item[14],
"Shape Dist Traveled": item[15], "Headsign": item[16], 'Shape Dist Traveled': item[15], 'Headsign': item[16],
"Sequence": item[17] 'Sequence': item[17]
} }
return { return {
"trip_id": item[0], 'trip_id': item[0],
"trip": sched.trips_by_id(item[0])[0], 'trip': sched.trips_by_id(item[0])[0],
"route": route, 'route': route,
"agency": sched.agencies_by_id(route.agency_id)[0], 'agency': sched.agencies_by_id(route.agency_id)[0],
"origin_station": origin_station, 'origin_station': origin_station,
"departure_time": departure_time, 'departure_time': departure_time,
"destination_station": destination_station, 'destination_station': destination_station,
"arrival_time": arrival_time, 'arrival_time': arrival_time,
"seconds_until_departure": seconds_until, 'seconds_until_departure': seconds_until,
"minutes_until_departure": minutes_until, 'minutes_until_departure': minutes_until,
"origin_stop_time": origin_stop_time_dict, 'origin_stop_time': origin_stop_time_dict,
"destination_stop_time": destination_stop_time_dict 'destination_stop_time': destination_stop_time_dict
} }
def setup_platform(hass, config, add_devices, discovery_info=None): def setup_platform(hass, config, add_devices, discovery_info=None):
"""Get the GTFS sensor.""" """Get the GTFS sensor."""
if config.get("origin") is None: gtfs_dir = hass.config.path(DEFAULT_PATH)
_LOGGER.error("Origin must be set in the GTFS configuration!") data = config.get(CONF_DATA)
return False origin = config.get(CONF_ORIGIN)
destination = config.get(CONF_DESTINATION)
if config.get("destination") is None: name = config.get(CONF_NAME)
_LOGGER.error("Destination must be set in the GTFS configuration!")
return False
if config.get("data") is None:
_LOGGER.error("Data must be set in the GTFS configuration!")
return False
gtfs_dir = hass.config.path("gtfs")
if not os.path.exists(gtfs_dir): if not os.path.exists(gtfs_dir):
os.makedirs(gtfs_dir) os.makedirs(gtfs_dir)
if not os.path.exists(os.path.join(gtfs_dir, config["data"])): if not os.path.exists(os.path.join(gtfs_dir, data)):
_LOGGER.error("The given GTFS data file/folder was not found!") _LOGGER.error("The given GTFS data file/folder was not found!")
return False return False
import pygtfs import pygtfs
split_file_name = os.path.splitext(config["data"]) split_file_name = os.path.splitext(data)
sqlite_file = "{}.sqlite".format(split_file_name[0]) sqlite_file = "{}.sqlite".format(split_file_name[0])
joined_path = os.path.join(gtfs_dir, sqlite_file) joined_path = os.path.join(gtfs_dir, sqlite_file)
@ -164,27 +171,22 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
# pylint: disable=no-member # pylint: disable=no-member
if len(gtfs.feeds) < 1: if len(gtfs.feeds) < 1:
pygtfs.append_feed(gtfs, os.path.join(gtfs_dir, pygtfs.append_feed(gtfs, os.path.join(gtfs_dir, data))
config["data"]))
add_devices([GTFSDepartureSensor(gtfs, name, origin, destination)])
dev = []
dev.append(GTFSDepartureSensor(gtfs, config["origin"],
config["destination"]))
add_devices(dev)
# pylint: disable=too-many-instance-attributes,too-few-public-methods # pylint: disable=too-many-instance-attributes,too-few-public-methods
class GTFSDepartureSensor(Entity): class GTFSDepartureSensor(Entity):
"""Implementation of an GTFS departures sensor.""" """Implementation of an GTFS departures sensor."""
def __init__(self, pygtfs, origin, destination): def __init__(self, pygtfs, name, origin, destination):
"""Initialize the sensor.""" """Initialize the sensor."""
self._pygtfs = pygtfs self._pygtfs = pygtfs
self.origin = origin self.origin = origin
self.destination = destination self.destination = destination
self._name = "GTFS Sensor" self._name = name
self._unit_of_measurement = "min" self._unit_of_measurement = 'min'
self._state = 0 self._state = 0
self._attributes = {} self._attributes = {}
self.lock = threading.Lock() self.lock = threading.Lock()
@ -220,23 +222,22 @@ class GTFSDepartureSensor(Entity):
with self.lock: with self.lock:
self._departure = get_next_departure(self._pygtfs, self.origin, self._departure = get_next_departure(self._pygtfs, self.origin,
self.destination) self.destination)
self._state = self._departure["minutes_until_departure"] self._state = self._departure['minutes_until_departure']
origin_station = self._departure["origin_station"] origin_station = self._departure['origin_station']
destination_station = self._departure["destination_station"] destination_station = self._departure['destination_station']
origin_stop_time = self._departure["origin_stop_time"] origin_stop_time = self._departure['origin_stop_time']
destination_stop_time = self._departure["destination_stop_time"] destination_stop_time = self._departure['destination_stop_time']
agency = self._departure["agency"] agency = self._departure['agency']
route = self._departure["route"] route = self._departure['route']
trip = self._departure["trip"] trip = self._departure['trip']
name = "{} {} to {} next departure" name = '{} {} to {} next departure'
self._name = name.format(agency.agency_name, self._name = name.format(agency.agency_name,
origin_station.stop_id, origin_station.stop_id,
destination_station.stop_id) destination_station.stop_id)
# Build attributes # Build attributes
self._attributes = {} self._attributes = {}
def dict_for_table(resource): def dict_for_table(resource):
@ -247,22 +248,22 @@ class GTFSDepartureSensor(Entity):
def append_keys(resource, prefix=None): def append_keys(resource, prefix=None):
"""Properly format key val pairs to append to attributes.""" """Properly format key val pairs to append to attributes."""
for key, val in resource.items(): for key, val in resource.items():
if val == "" or val is None or key == "feed_id": if val == "" or val is None or key == 'feed_id':
continue continue
pretty_key = key.replace("_", " ") pretty_key = key.replace('_', ' ')
pretty_key = pretty_key.title() pretty_key = pretty_key.title()
pretty_key = pretty_key.replace("Id", "ID") pretty_key = pretty_key.replace('Id', 'ID')
pretty_key = pretty_key.replace("Url", "URL") pretty_key = pretty_key.replace('Url', 'URL')
if prefix is not None and \ if prefix is not None and \
pretty_key.startswith(prefix) is False: pretty_key.startswith(prefix) is False:
pretty_key = "{} {}".format(prefix, pretty_key) pretty_key = '{} {}'.format(prefix, pretty_key)
self._attributes[pretty_key] = val self._attributes[pretty_key] = val
append_keys(dict_for_table(agency), "Agency") append_keys(dict_for_table(agency), 'Agency')
append_keys(dict_for_table(route), "Route") append_keys(dict_for_table(route), 'Route')
append_keys(dict_for_table(trip), "Trip") append_keys(dict_for_table(trip), 'Trip')
append_keys(dict_for_table(origin_station), "Origin Station") append_keys(dict_for_table(origin_station), 'Origin Station')
append_keys(dict_for_table(destination_station), append_keys(dict_for_table(destination_station),
"Destination Station") 'Destination Station')
append_keys(origin_stop_time, "Origin Stop") append_keys(origin_stop_time, 'Origin Stop')
append_keys(destination_stop_time, "Destination Stop") append_keys(destination_stop_time, 'Destination Stop')