Thu, 06 Dec 2018 19:35:38 +0200
bus live information
#!/usr/bin/env python3 from flask import Flask, render_template, abort, send_from_directory, redirect from datetime import datetime, date, time, timedelta from os import path, listdir, environ from configparser import ConfigParser import locale app = Flask(__name__) from misc import * from busroute import reduce_schedule import buses regions = {} def reset_locale(): ''' Resets the locale to system so that the system language settings do not mess with the formatting. ''' locale.setlocale(locale.LC_ALL, locale.getdefaultlocale()) def activate_locale(language = None): ''' Activates either the locale for the provided language, or the default locale. Returns such an object that resets the locale upon function exit. ''' language = language or language_for_page() class result: def __enter__(self): if language: locale.setlocale(locale.LC_ALL, tr('locale', 'other', language = language)) def __exit__(self, *args): reset_locale() return result() def simplify_name(region_name, replace = False): region = regions.get(region_name) if region: if replace and 'replacement' in region: return simplify_name(region['replacement']) return tr(region.get('name', region_name), 'region_short_name', default = region.get('short_name', region_name)) else: return tr(region_name, 'region_short_name') reset_locale() # Load translations class Translator: def __init__(self, languages = None): self.languages = languages or dict() def load_language(self, file_path): language_name = path.splitext(path.basename(file_path))[0] ini = ConfigParser() ini.read(path.join(file_path)) self.languages[language_name] = ini def __call__(self, name, *sections, language = None, default = None): language = language or language_for_page() for section in sections: try: print('Trying:', repr(language), repr(section), repr(name)) return self.languages[language][section][name] except KeyError: try: return profile['tr:' + language + ':' + section][name] except KeyError: pass else: return default or (name[:1].upper() + name[1:]) def load_region(self, region): for key, value in region.items(): if ':' in key: name_type, language = key.split(':', 1) if name_type.endswith('name') and language != '': section = 'region_' + name_type if section not in self.languages[language]: self.languages[language][section] = {} print(repr(language), repr(section), repr(region['name']), '=', repr(value)) self.languages[language][section][region['name']] = value def load_regions(self, regions): for region in regions.values(): self.load_region(region) def __repr__(self): return 'Translator(languages = ' + repr(self.languages) + ')' tr = Translator() for file in listdir('tr'): tr.load_language(path.join('tr', file)) def language_for_page(): ''' Returns the code of which language to use for the page. ''' from flask import request if request.args.get('untranslated') is not None: return None else: for language_name in tr.languages: if request.args.get(language_name) is not None: return language_name else: return request.accept_languages.best_match(tr.languages) def sign_elements(schedule_entry, format = 'medium'): ''' For an entry in a bus stop schedule, find out where the connection is leading to. Returns a list of places, possibly empty. ''' from math import ceil trip_length = schedule_entry['trip'].length - schedule_entry['stop'].traveled_distance regions = schedule_entry['trip'].concise_schedule(schedule_entry['stop']) return [ name for name in reduce_schedule( regions, trip_length = trip_length, format = format ) ] def sign(schedule_entry, format = 'medium'): ''' For an entry in a bus stop schedule, find out where the connection is leading to. Returns a string. ''' sign = sign_elements(schedule_entry, format = format) if sign: return ' - '.join(tr(simplify_name(place), 'region_name', 'region_short_name') for place in sign) else: return schedule_entry['trip'].schedule[-1].stop.name def long_form_sign(schedule_entry, format = 'long'): from math import ceil trip_length = schedule_entry['trip'].length - schedule_entry['stop'].traveled_distance sign = reduce_schedule(schedule_entry['trip'].concise_schedule(schedule_entry['stop']), trip_length = trip_length, format = format) if sign: return { 'destination': tr(sign[-1], 'region_short_name'), 'via': [tr(place, 'region_short_name') for place in sign[:-1]], } else: return { 'destination': schedule_entry['trip'].schedule[-1].stop.name, 'via': [], } def imminent(schedule_entry): return (schedule_entry['time'] - now()) <= timedelta(minutes = 3) def first_halt_in_trip_in_place(trip, place): for halt in trip.schedule: if halt.stop.region == place: return halt else: return None place_abbreviations = ConfigParser() place_abbreviations.read('abbreviations.ini') def place_abbreviation(place): try: return place_abbreviations['abbreviations'][place] except KeyError: return place def trip_description(trip): entries = [trip.from_place] old_places = None starting_halt = None while len(entries) < 3: remaining_length = trip.length if starting_halt: remaining_length -= starting_halt.traveled_distance places = reduce_schedule(trip.concise_schedule(starting_stop = starting_halt), trip_length = remaining_length, format = 'short') new_places = set(places) - set(entries) if not new_places or places == old_places: break for place in places: if place in new_places: starting_halt = first_halt_in_trip_in_place(trip, place) entries += [place] break old_places = places if trip.to_place not in entries: entries += [trip.to_place] return entries def trip_abbreviation(trip): return trip.route.reference + ':' + '-'.join(map(place_abbreviation, trip_description(trip))) def split_route_ref(route_ref): try: return list(parse_route_ref(route_ref)) except ValueError: return ['', route_ref, ''] @app.route('/stop/<reference>') def bus_stop_schedule(reference): from buses import bus_stops schedule = [] try: bus_stop = bus_stops[reference] except KeyError: abort(404) for schedule_entry in bus_stop.schedule(max_amount = 100, max_past = 4, arrivals = True): route_ref = schedule_entry['trip'].route.reference schedule.append({ 'time': time_representation(schedule_entry['time']), 'timestamp': int(schedule_entry['time'].timestamp()), 'route': route_ref, 'route-splice': split_route_ref(route_ref), 'sign': sign(schedule_entry), 'trip': schedule_entry['stop'].trip.name, 'night': is_night_time(schedule_entry['time']), 'imminent': imminent(schedule_entry), 'id': str(schedule_entry['stop'].uuid), 'gone': schedule_entry['gone'], }) return render_template( 'stop.html', schedule = schedule, ref = bus_stop.code, name = bus_stop.code + ' ' + tr(bus_stop.name, 'bus-stops'), link_to_map = bus_stop.location.link_to_map, region = hasattr(bus_stop, 'region_name') and bus_stop.region or None, location = bus_stop.location, cluster = bus_stop.cluster.url_name if len(bus_stop.cluster.stops) > 1 else None, tr = tr, typename = bus_stop.typename, service = min(bus_stop.services), ) def week_schedule(bus_stop, start_date = today(), **kwargs): for i in range(7): try: yield from bus_stop.schedule_for_day(start_date + timedelta(i), **kwargs) except ValueError: # went outside bounds return def route_key(route): match = re.search(r'^([a-zA-Z]*)(\d+)(.*)$', route) if match: groups = match.groups() return (groups[0], int(groups[1]), groups[2]) else: return (route,) def parse_route_ref(route_ref): from re import search match = search(r'^([^0-9]*)([0-9]+)(.*)$', route_ref) try: return match.group(1), int(match.group(2)), match.group(3) except AttributeError: raise ValueError(route_ref) def condense_route_list(route_list): def prepare_range_pool(range_pool): if len(range_pool) < 3: yield from map(str, range_pool) else: yield str(min(range_pool)) + '-' + str(max(range_pool)) range_pool = [] for route in route_list: try: route_int = int(route) except ValueError: yield from prepare_range_pool(range_pool) range_pool = [] yield route else: if not range_pool or route_int - 1 in range_pool: range_pool.append(route_int) else: yield from prepare_range_pool(range_pool) range_pool = [route_int] if range_pool: yield from prepare_range_pool(range_pool) def is_weekend_night(time): from datetime import timedelta adjusted_time = time - timedelta(hours = 4, minutes = 30) return adjusted_time.weekday() in [4, 5] and is_night_time(time) encircled = '⒜⒝⒞⒟⒠⒡⒢⒣⒤⒥⒦⒧⒨⒩⒪⒫⒬⒭⒮⒯⒰⒱⒲⒳⒴⒵' #encircled = 'ⓐⓑⓒⓓⓔⓕⓖⓗⓘⓙⓚⓛⓜⓝⓞⓟⓠⓡⓢⓣⓤⓥⓦⓧⓨⓩ' def encircle(char): from string import ascii_uppercase try: return encircled[ascii_uppercase.index(char.upper())] except IndexError: return char def variant_abbreviations(variants): '''Makes a mapping of route destination variants to letter abbreviations''' suggestion = [variant[-1][0].upper() for variant in variants] if len(set(suggestion)) != len(suggestion): from string import ascii_uppercase suggestion = ascii_uppercase return dict(zip(variants, suggestion)) def schedule_entry_hash(schedule_entry): return schedule_entry['trip'].name, schedule_entry['time'] def describe(bus_stop, week_schedule): schedule = [] from collections import defaultdict, Counter from busroute import simplify_name destinations_per_route = defaultdict(Counter) counts_per_variant = {} def route_key(route_ref): from math import log try: return () + parse_route_ref(route_ref) except ValueError: return () def filter_names(names): if len(names) == 1 and names[0] == (bus_stop.region and simplify_name(bus_stop.region)): return type(names)() else: return names data = [] names = [] from collections import defaultdict night_routes = defaultdict(lambda: True) num_leaves = 0 trip_mapping = {} for schedule_entry in week_schedule: from busroute import greatly_simplify_name sign_tuple = tuple(sign_elements(schedule_entry, format = 'short')) sign_tuple = tuple(greatly_simplify_name(k) for k in sign_tuple) route = schedule_entry['trip'].route.reference night_routes[route] &= is_night_time(schedule_entry['time']) destinations_per_route[route][sign_tuple] += 1 trip_mapping[schedule_entry_hash(schedule_entry)] = (route, sign_tuple) num_leaves += 1 night_routes = {key for key, value in night_routes.items() if value} routes_per_destination = defaultdict(set) all_variants = [] variant_names = {} # Collect all variants for route, tally in destinations_per_route.items(): variants = variant_abbreviations(tally.keys()) winner, count = tally.most_common()[0] for destination, count in tally.items(): route_name = route if destination != winner: route_name += encircle(variants[destination]) if route in night_routes: night_routes.add(route_name) variant_names[route, destination] = route_name all_variants.append({ 'name': route_name, 'destination': filter_names(destination), 'count': count }) counts_per_variant[route_name] = count all_variants.sort(key = lambda k: k['count']) def route_limit(route): return (route in night_routes) and 6 or 20 rare_variants = {variant['name'] for variant in all_variants if variant['count'] < route_limit(variant['name'])} route_variant_count = len(all_variants) for variant in all_variants: routes_per_destination[variant['destination']].add(variant['name']) for key in routes_per_destination: routes_per_destination[key] = sorted(routes_per_destination[key], key = route_key) def route_len(route): length = 0 for char in route: if char.isdigit(): length += 1 else: break return length or len(route) from math import inf def route_key(route): from math import log return ( counts_per_variant.get(route, 0) < route_limit(route), route in night_routes, route_len(route), str(route) ) def routes_key(routes): return min(route_key(route) for route in routes) # Convert routes per destination to item pairs so that we can split it into rare and non-rare route_destination_pairs = list() for regions, routes in routes_per_destination.items(): common_routes = set(route for route in routes if counts_per_variant[route] >= route_limit(route)) rare_routes = set(routes) - common_routes if common_routes: route_destination_pairs.append((regions, common_routes)) if rare_routes: route_destination_pairs.append((regions, rare_routes)) result = [] rare_variant_groups = set() for regions, routes in sorted(route_destination_pairs, key = lambda pair: routes_key(pair[1])): routes_tuple = tuple(condense_route_list(sorted(routes, key = route_key))) result.append(( routes_tuple, ' - '.join(tr(region, 'region_short_name') for region in regions) )) if all(variant in rare_variants for variant in routes): rare_variant_groups.add(routes_tuple) return { 'night-routes': night_routes, 'all-night-routes': lambda entry, description: all(route in description['night-routes'] for route in entry[0]), 'simple': route_variant_count <= 1, 'description': result, 'variant-map': {k:variant_names[v] for k, v in trip_mapping.items()}, 'rare-variants': rare_variants, 'rare-variant-groups': rare_variant_groups, } @app.route('/stop_description/<reference>') def bus_stop_description(reference): from buses import bus_stops from pprint import pformat try: bus_stop = bus_stops[reference] except KeyError: abort(404) return '<pre>' + pformat(describe(bus_stop, week_schedule(bus_stop, arrivals = True))) + '</pre>' @app.route('/api/describe_destination/<trip_reference>/<stop_reference>/<int:index>') def describe_destination(trip_reference, stop_reference, index): from buses import bus_stops, all_trips from busroute import simplify_name from flask import jsonify try: trip = all_trips[trip_reference] bus_stop = bus_stops[stop] schedule_entry = [schedule_entry for schedule_entry in trip.schedule if schedule_entry.stop.reference == stop_reference][index] except KeyError: abort(404) except ValueError: abort(404) return jsonify(long_form_sign({'trip': trip, 'stop': schedule_entry})) @app.route('/api/trip_abbreviation/<block_id>/<int:arrival_time_offset>') def api_trip_abbreviation(block_id, arrival_time_offset): from buses import trips_by_vehicle_info from flask import jsonify from datetime import timedelta try: trip = trips_by_vehicle_info[block_id, timedelta(seconds = arrival_time_offset)] except KeyError: abort(404) return jsonify({ 'abbreviation': trip_abbreviation(trip), }) def service_start_time(): from datetime import date, datetime, timedelta result = datetime.now().replace(hour = 0, minute = 0, second = 0, microsecond = 0) if datetime.now().hour < 4: result -= timedelta(1) return result @app.route('/find_halt/<stop_reference>/<blockref>/<int:originalaimeddeparturetime>') def find_halt(stop_reference, blockref, originalaimeddeparturetime): from datetime import datetime from flask import jsonify info = (blockref, datetime.fromtimestamp(originalaimeddeparturetime) - service_start_time()) trip = buses.trips_by_vehicle_info[info] try: return jsonify({ 'id': [str(halt.uuid) for halt in buses.trips_by_vehicle_info[info].schedule if halt.stop.reference == stop_reference][0], }) except: abort(404) def current_bus_day(): from datetime import date, datetime, timedelta day = date.today() if datetime.now().hour < 5: day -= timedelta(1) return day @app.route('/stop_display/<reference>') def stop_display(reference): from buses import bus_stops schedule = [] try: bus_stop = bus_stops[reference] except KeyError: abort(404) for i, schedule_entry in enumerate(bus_stop.schedule_for_day(current_bus_day(), arrivals = False)): schedule.append({ 'time_data': schedule_entry['time'], 'time': time_representation(schedule_entry['time']), 'timestamp': int(schedule_entry['time'].timestamp()), 'route': schedule_entry['trip'].route.reference, 'sign': long_form_sign(schedule_entry, format = 'medium'), 'trip': schedule_entry['stop'].trip.name, 'night': is_night_time(schedule_entry['time']), 'imminent': imminent(schedule_entry), 'index': i, }) from pprint import pprint pprint(schedule) def destination_key(schedule_entry): sign = schedule_entry['sign'] return (sign['destination'],) + tuple(sign['via']) def filter_schedule(schedule, *, key): used = set() for schedule_entry in schedule: key_value = key(schedule_entry) if key_value not in used: used.add(key_value) yield schedule_entry schedule = list(filter_schedule(schedule, key = destination_key))[:6] if schedule: num_imminent_leaves = max(1, len([schedule_entry for schedule_entry in schedule if schedule_entry['time_data'] - schedule[0]['time_data'] < timedelta(minutes = 3)])) else: num_imminent_leaves = 1 return render_template( 'stop_display.html', schedule = schedule, ref = bus_stop.code, name = tr(bus_stop.name, 'bus-stops'), link_to_map = bus_stop.location.link_to_map, region = bus_stop.region, location = bus_stop.location, cluster = bus_stop.cluster.url_name if len(bus_stop.cluster.stops) > 1 else None, num_imminent_leaves = num_imminent_leaves, tr = tr, ) def time_representation(time, relative = True): time_difference = time - now() #if relative and timedelta(minutes = -1) < time_difference < timedelta(minutes = 1): # return tr('right-now', 'misc-text') #elif relative and time_difference > timedelta(0) and time_difference < timedelta(minutes = 10): # return '%dm' % round(time_difference.seconds / 60) if time.date() == today(): return '%d:%02d' % (time.hour, time.minute) elif time_difference < timedelta(7): with activate_locale(): return time.strftime('%-a %H:%M').replace(' ', '\xa0') else: with activate_locale(): return time.strftime('%-d.%-m. %H:%M').replace(' ', '\xa0') def make_cluster(cluster): schedule = list({ 'time': time_representation(schedule_entry['time']), 'route': schedule_entry['trip'].route.reference, 'route-splice': split_route_ref(schedule_entry['trip'].route.reference), 'sign': sign(schedule_entry), 'trip': schedule_entry['stop'].trip.name, 'night': is_night_time(schedule_entry['time']), 'stop': schedule_entry['stop'].stop, 'stop_id': schedule_entry['stop'].stop.reference, 'stop_code': schedule_entry['stop'].stop.code, 'stop_name': tr(schedule_entry['stop'].stop.name, 'bus-stops', 'places'), 'imminent': imminent(schedule_entry), 'typename': schedule_entry['stop'].stop.typename, } for schedule_entry in cluster.schedule(max_amount = 100)) stops_in_cluster = sorted( ({ 'id': stop.reference, 'code': stop.code, 'name': tr(stop.name, 'bus-stops'), } for stop in cluster.stops), key = lambda stop: (len(stop['id']), stop['id']) ) return render_template( 'cluster.html', schedule = schedule, name = tr(cluster.name, 'places', 'place-clusters', 'bus-stops'), link_to_map = cluster.center.link_to_map, location = cluster.center, stops_in_cluster = stops_in_cluster, amount_of_stops_in_cluster = len(stops_in_cluster), tr = tr, service = min(set.union(*[bus_stop.services for bus_stop in cluster.stops])), ) @app.route('/stop_cluster/<cluster_name>') def cluster_schedule(cluster_name): from buses import bus_stops, clusters_by_name try: cluster = clusters_by_name[cluster_name] except KeyError: abort(404) else: return make_cluster(cluster) @app.route('/cluster') def custom_cluster(): from flask import request from buses import bus_stops, CustomBusStopCluster if 'stops' in request.args: cluster = CustomBusStopCluster( name = request.args.get('name', 'Aikataulu'), stops = {bus_stops[stop_code] for stop_code in str.split(request.args['stops'], ';')}, ) return make_cluster(cluster) else: return render_template('custom_cluster.html') def day_class(weekday): if weekday < 5: return 'working-day' elif weekday == 5: return 'saturday' else: assert weekday == 6 return 'sunday' class Week: def __init__(self, year, weeknumber): self.year, self.weeknumber = int(year), int(weeknumber) def normalize(self): while self.weeknumber > 52: self.weeknumber -= 52 self.year += 1 while self.weeknumber < 1: self.weeknumber += 52 self.year -= 1 def __repr__(self): return str.format( 'Week({year}, {weeknumber})', year = self.year, weeknumber = self.weeknumber ) def __add__(self, numweeks): newweek = Week(year = self.year, weeknumber = self.weeknumber + numweeks) newweek.normalize() return newweek def __sub__(self, numweeks): newweek = Week(year = self.year, weeknumber = self.weeknumber - numweeks) newweek.normalize() return newweek def monday(self): from datetime import datetime datestring = str.format( '{year}-W{weeknumber}-1', year = self.year, weeknumber = self.weeknumber, ) return datetime.strptime(datestring, "%Y-W%W-%w").date() @staticmethod def fromstring(string): try: year, weeknumber = string.split('W', 1) return Week(year = int(year), weeknumber = int(weeknumber)) except ValueError: raise ValueError('bad week string: ' + repr(string)) def __str__(self): return str(self.year) + 'W' + str(self.weeknumber) @staticmethod def now(): from datetime import date cal = date.today().isocalendar() return Week(year = cal[0], weeknumber = cal[1]) @app.route('/stop_week/<stop_reference>') def stop_week(stop_reference): from buses import bus_stops from flask import request from datetime import date if 'week' in request.args: try: week = Week.fromstring(request.args['week']) except ValueError: abort(400) else: week = Week.now() if 'routes' in request.args: filtered_routes = set(request.args['routes'].split(';')) route_filter = lambda route: route in filtered_routes else: route_filter = lambda route: True if 'dest' in request.args: dests = {bus_stops.get(dest, None) for dest in request.args['dest'].split(';')} dests.discard(None) dest_filter = lambda trip: any(trip.contains_stop(dest) for dest in dests) else: dest_filter = lambda trip: True schedule = [] try: bus_stop = bus_stops[stop_reference] except KeyError: abort(404) week_model = {} bus_stop_schedule = list(week_schedule(bus_stop, start_date = week.monday(), arrivals = True, allow_gone = True)) description = describe(bus_stop, bus_stop_schedule) for schedule_entry in bus_stop_schedule: route_ref = schedule_entry['trip'].route.reference if route_filter(route_ref) and dest_filter(schedule_entry['trip']): time = schedule_entry['time'] date = schedule_entry['date'] if date not in week_model: week_model[date] = dict() day_model = week_model[date] if time.hour not in day_model: day_model[time.hour] = list() hour_model = day_model[time.hour] list.append(hour_model, { 'route': description['variant-map'][schedule_entry_hash(schedule_entry)], 'route-splice': split_route_ref(route_ref), 'trip': schedule_entry['stop'].trip.name, 'night': is_night_time(schedule_entry['time']), 'minute': time.minute, }) for day_offset in range(7): from datetime import date, datetime, timedelta day = week.monday() + timedelta(day_offset) try: day_model = week_model[day] except KeyError: week_model[day] = {} else: def hour_key(x): return (x - 5) % 24 # Fill in missing hours hours = set(dict.keys(day_model)) sorted_hours = sorted(hours, key = hour_key) start_hour = sorted_hours[0] end_hour = sorted_hours[-1] + 1 for hour in range(start_hour, end_hour): hour_start = datetime(day.year, day.month, day.day, hour, 0) if hour not in day_model and hour_start >= datetime.now(): day_model[hour] = [] # Sort the hours, so that 5am is first and 4am is last. from collections import OrderedDict week_model[day] = OrderedDict( sorted( day_model.items(), key = lambda pair: hour_key(pair[0]), ) ) week_model = [ { 'day': day, 'schedule': schedule, 'day-class': day_class(day.weekday()) } for day, schedule in week_model.items() ] week_model = sorted(week_model, key = lambda day: day['day']) return render_template( 'stop_week.html', ref = bus_stop.code, name = tr(bus_stop.name, 'bus-stops'), tr = tr, week = week_model, description = description, typename = bus_stop.typename, curweek = week, next_week = week + 1, previous_week = week - 1, ) @app.route('/trip/<trip_reference>') def trip(trip_reference): from flask import request from buses import all_trips, shapes from busroute import simplify_name,greatly_simplify_name try: trip = all_trips[trip_reference] except KeyError: abort(404) schedule = [] region = '' for halt in trip.schedule: stop_time = datetime.combine(today(), time()) + halt.arrival_time formatted_time = time_representation(stop_time) if profile['regions']['use-regions']: if halt.stop.region != region and not (region and not halt.stop.region): if len(schedule) and not schedule[-1]['name']: schedule[-1]['name'] = tr(halt.stop.region or '', 'region_name') else: schedule.append({ 'name': tr(halt.stop.region or '', 'region_name'), 'time': formatted_time, 'stops': [], 'index': len(schedule), }) region = halt.stop.region else: schedule.append({ 'name': tr(halt.stop.name or '', 'bus-stops'), 'time': formatted_time, 'stops': [], 'index': len(schedule), }) schedule[-1]['stops'].append({ 'time': formatted_time, 'id': halt.stop.reference, 'code': halt.stop.code, 'ref': halt.stop.code or halt.stop.reference, 'name': tr(halt.stop.name, 'bus-stops'), 'typename': halt.stop.typename, 'arrival_time': time_representation(datetime.combine(today(), time()) + halt.arrival_time), 'departure_time': time_representation(datetime.combine(today(), time()) + halt.departure_time), 'location': str(halt.stop.location), }) sign = trip.concise_schedule() try: sign = [sign[0], sign[-1]] except IndexError: sign = [trip.schedule[0].stop.name, trip.schedule[-1].stop.name] return render_template('trip.html', schedule = schedule, trip_reference = trip_reference, route = trip.route.reference, #description = ' - '.join(tr(place, 'region_name') for place in sign), description = greatly_simplify_name(sign[-1]), night = is_night_time(datetime.combine(today(), time()) + trip.schedule[-1].arrival_time), tr = tr, length = trip.length / 1000, service = trip.route.service, shape = ','.join(str.format('[{}, {}]', point[0], point[1]) for point in shapes[trip.shape]), stops = list( dict( ref = halt.stop.reference, name = halt.stop.name, position = str.format('[{}]', str(halt.stop.location)), ) for halt in trip.schedule ), ) @app.route('/route/<names>') def route_page(names): from buses import routes from collections import defaultdict from busroute import greatly_simplify_name routes = {routes[name.upper()] for name in names.split(';')} schedules = defaultdict(list) for route in routes: for trip in route.trips: if trip.is_served_at(today()): schedules[trip.schedule[0].stop].append({ 'name': trip.reference, 'from': trip.from_place, 'to': trip.to_place, 'description': ' - '.join( greatly_simplify_name(place) for place in trip_description(trip) ), 'gone': datetime.combine(today(), time()) + trip.schedule[-1].arrival_time < now(), 'time': time_representation(datetime.combine(today(), time()) + trip.schedule[0].departure_time), 'route': route.reference, }) return render_template('route.html', name = route.reference + ' ' + route.description, tr = tr, schedules = schedules, ) @app.route('/interesting') def interesting(): from buses import all_trips, services_for_day from busroute import simplify_name from collections import Counter, defaultdict from datetime import datetime trip_counts = Counter() interesting_trips = defaultdict(list) for trip_reference, trip in all_trips.items(): if len(trip.service.dates) <= 3: for date in trip.service.dates: moment = trip.schedule[0].departure_datetime(date) if moment > datetime.now(): interesting_trips[date].append((trip, moment)) for date in interesting_trips: interesting_trips[date].sort(key = lambda tuple: tuple[1]) interesting_trips = [(date, interesting_trips[date]) for date in sorted(interesting_trips.keys())] data = [ { 'date': date, 'leaves': [{ 'route': trip.route.reference, 'departure': moment, 'trip': trip.reference, 'stop': trip.schedule[0].stop, 'description': ' - '.join( simplify_name(place, replace = True) for place in trip_description(trip) ), } for trip, moment in leaves] } for date, leaves in interesting_trips ] return render_template( 'interesting.html', data = data, tr = tr, ) @app.route('/static/<path:path>') def static_file(path): return send_from_directory(path.join('static', path)) from argparse import ArgumentParser parser = ArgumentParser() parser.add_argument('gtfs_zip_path') parser.add_argument('profile_path') if __name__ == '__main__': parser.add_argument('-p', '--port', type = int, default = 5000) parser.add_argument('-d', '--debug', action = 'store_true') args = parser.parse_args() profile.read(args.profile_path) if profile['regions']['use-regions']: from regions import parse_regions regions = parse_regions(profile['regions']['osm-path']) tr.load_regions(regions) import busroute busroute.regions = regions buses.load_buses(args.gtfs_zip_path) if __name__ == '__main__': app.run(debug = args.debug, port = args.port)