<s> """<STR_LIT>""" <EOL> import os <EOL> import logging <EOL> import datetime <EOL> from homeassistant . helpers . entity import Entity <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> REQUIREMENTS = [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] <EOL> ICON = "<STR_LIT>" <EOL> TIME_FORMAT = "<STR_LIT>" <EOL> def get_next_departure ( sched , start_station_id , end_station_id ) : <EOL> """<STR_LIT>""" <EOL> origin_station = sched . stops_by_id ( start_station_id ) [ <NUM_LIT:0> ] <EOL> destination_station = sched . stops_by_id ( end_station_id ) [ <NUM_LIT:0> ] <EOL> now = datetime . datetime . now ( ) <EOL> day_name = now . strftime ( "<STR_LIT>" ) . lower ( ) <EOL> now_str = now . strftime ( "<STR_LIT>" ) <EOL> from sqlalchemy . sql import text <EOL> sql_query = text ( """<STR_LIT>""" . format ( day_name = day_name ) ) <EOL> result = sched . engine . execute ( sql_query , now_str = now_str , <EOL> origin_station_id = origin_station . id , <EOL> end_station_id = destination_station . id ) <EOL> item = { } <EOL> for row in result : <EOL> item = row <EOL> today = datetime . datetime . today ( ) . strftime ( "<STR_LIT>" ) <EOL> departure_time_string = "<STR_LIT>" . format ( today , item [ <NUM_LIT:2> ] ) <EOL> arrival_time_string = "<STR_LIT>" . format ( today , item [ <NUM_LIT:3> ] ) <EOL> departure_time = datetime . datetime . strptime ( departure_time_string , <EOL> TIME_FORMAT ) <EOL> arrival_time = datetime . datetime . strptime ( arrival_time_string , <EOL> TIME_FORMAT ) <EOL> seconds_until = ( departure_time - datetime . datetime . now ( ) ) . total_seconds ( ) <EOL> minutes_until = int ( seconds_until / <NUM_LIT> ) <EOL> route = sched . routes_by_id ( item [ <NUM_LIT:1> ] ) [ <NUM_LIT:0> ] <EOL> origin_stoptime_arrival_time = "<STR_LIT>" . format ( today , item [ <NUM_LIT:4> ] ) <EOL> origin_stoptime_departure_time = "<STR_LIT>" . format ( today , item [ <NUM_LIT:5> ] ) <EOL> dest_stoptime_arrival_time = "<STR_LIT>" . format ( today , item [ <NUM_LIT:11> ] ) <EOL> dest_stoptime_depart_time = "<STR_LIT>" . format ( today , item [ <NUM_LIT:12> ] ) <EOL> origin_stop_time_dict = { <EOL> "<STR_LIT>" : origin_stoptime_arrival_time , <EOL> "<STR_LIT>" : origin_stoptime_departure_time , <EOL> "<STR_LIT>" : item [ <NUM_LIT:6> ] , "<STR_LIT>" : item [ <NUM_LIT:7> ] , <EOL> "<STR_LIT>" : item [ <NUM_LIT:8> ] , "<STR_LIT>" : item [ <NUM_LIT:9> ] , <EOL> "<STR_LIT>" : item [ <NUM_LIT:10> ] <EOL> } <EOL> destination_stop_time_dict = { <EOL> "<STR_LIT>" : dest_stoptime_arrival_time , <EOL> "<STR_LIT>" : dest_stoptime_depart_time , <EOL> "<STR_LIT>" : item [ <NUM_LIT> ] , "<STR_LIT>" : item [ <NUM_LIT> ] , <EOL> "<STR_LIT>" : item [ <NUM_LIT:15> ] , "<STR_LIT>" : item [ <NUM_LIT:16> ] , <EOL> "<STR_LIT>" : item [ <NUM_LIT> ] <EOL> } <EOL> return { <EOL> "<STR_LIT>" : item [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : sched . trips_by_id ( item [ <NUM_LIT:0> ] ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : route , <EOL> "<STR_LIT>" : sched . agencies_by_id ( route . agency_id ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : origin_station , <EOL> "<STR_LIT>" : departure_time , <EOL> "<STR_LIT>" : destination_station , <EOL> "<STR_LIT>" : arrival_time , <EOL> "<STR_LIT>" : seconds_until , <EOL> "<STR_LIT>" : minutes_until , <EOL> "<STR_LIT>" : origin_stop_time_dict , <EOL> "<STR_LIT>" : destination_stop_time_dict <EOL> } <EOL> def setup_platform ( hass , config , add_devices , discovery_info = None ) : <EOL> """<STR_LIT>""" <EOL> if config . get ( "<STR_LIT>" ) is None : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return False <EOL> if config . get ( "<STR_LIT>" ) is None : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return False <EOL> if config . get ( "<STR_LIT:data>" ) is None : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return False <EOL> gtfs_dir = hass . config . path ( "<STR_LIT>" ) <EOL> if not os . path . exists ( gtfs_dir ) : <EOL> os . makedirs ( gtfs_dir ) <EOL> if not os . path . exists ( os . path . join ( gtfs_dir , config [ "<STR_LIT:data>" ] ) ) : <EOL> _LOGGER . error ( "<STR_LIT>" ) <EOL> return False <EOL> dev = [ ] <EOL> dev . append ( GTFSDepartureSensor ( config [ "<STR_LIT:data>" ] , gtfs_dir , <EOL> config [ "<STR_LIT>" ] , config [ "<STR_LIT>" ] ) ) <EOL> add_devices ( dev ) <EOL> class GTFSDepartureSensor ( Entity ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data_source , gtfs_folder , origin , destination ) : <EOL> """<STR_LIT>""" <EOL> self . _data_source = data_source <EOL> self . _gtfs_folder = gtfs_folder <EOL> self . origin = origin <EOL> self . destination = destination <EOL> self . _name = "<STR_LIT>" <EOL> self . _unit_of_measurement = "<STR_LIT>" <EOL> self . _state = <NUM_LIT:0> <EOL> self . _attributes = { } <EOL> self . update ( ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _name <EOL> @ property <EOL> def state ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _state <EOL> @ property <EOL> def unit_of_measurement ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _unit_of_measurement <EOL> @ property <EOL> def device_state_attributes ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _attributes <EOL> @ property <EOL> def icon ( self ) : <EOL> """<STR_LIT>""" <EOL> return ICON <EOL> def update ( self ) : <EOL> """<STR_LIT>""" <EOL> import pygtfs <EOL> split_file_name = os . path . splitext ( self . _data_source ) <EOL> sqlite_file = "<STR_LIT>" . format ( split_file_name [ <NUM_LIT:0> ] ) <EOL> gtfs = pygtfs . Schedule ( os . path . join ( self . _gtfs_folder , sqlite_file ) ) <EOL> if len ( gtfs . feeds ) < <NUM_LIT:1> : <EOL> pygtfs . append_feed ( gtfs , os . path . join ( self . _gtfs_folder , <EOL> self . _data_source ) ) <EOL> self . _departure = get_next_departure ( gtfs , self . origin , <EOL> self . destination ) <EOL> self . _state = self . _departure [ "<STR_LIT>" ] <EOL> origin_station = self . _departure [ "<STR_LIT>" ] <EOL> destination_station = self . _departure [ "<STR_LIT>" ] <EOL> origin_stop_time = self . _departure [ "<STR_LIT>" ] <EOL> destination_stop_time = self . _departure [ "<STR_LIT>" ] <EOL> agency = self . _departure [ "<STR_LIT>" ] <EOL> route = self . _departure [ "<STR_LIT>" ] <EOL> trip = self . _departure [ "<STR_LIT>" ] <EOL> name = "<STR_LIT>" <EOL> self . _name = name . format ( agency . agency_name , <EOL> origin_station . stop_id , <EOL> destination_station . stop_id ) <EOL> self . _attributes = { } <EOL> def dict_for_table ( resource ) : <EOL> """<STR_LIT>""" <EOL> return dict ( ( col , getattr ( resource , col ) ) <EOL> for col in resource . __table__ . columns . keys ( ) ) <EOL> def append_keys ( resource , prefix = None ) : <EOL> """<STR_LIT>""" <EOL> for key , val in resource . items ( ) : <EOL> if val == "<STR_LIT>" or val is None or key == "<STR_LIT>" : <EOL> continue <EOL> pretty_key = key . replace ( "<STR_LIT:_>" , "<STR_LIT:U+0020>" ) <EOL> pretty_key = pretty_key . title ( ) <EOL> pretty_key = pretty_key . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> pretty_key = pretty_key . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if prefix is not None and pretty_key . startswith ( prefix ) is False : <EOL> pretty_key = "<STR_LIT>" . format ( prefix , pretty_key ) <EOL> self . _attributes [ pretty_key ] = val <EOL> append_keys ( dict_for_table ( agency ) , "<STR_LIT>" ) <EOL> append_keys ( dict_for_table ( route ) , "<STR_LIT>" ) <EOL> append_keys ( dict_for_table ( trip ) , "<STR_LIT>" ) <EOL> append_keys ( dict_for_table ( origin_station ) , "<STR_LIT>" ) <EOL> append_keys ( dict_for_table ( destination_station ) , "<STR_LIT>" ) <EOL> append_keys ( origin_stop_time , "<STR_LIT>" ) <EOL> append_keys ( destination_stop_time , "<STR_LIT>" ) </s>
