diff options
| author | Yuchen Pei <hi@ypei.me> | 2022-04-05 12:47:11 +1000 |
|---|---|---|
| committer | Yuchen Pei <hi@ypei.me> | 2022-04-07 21:00:13 +1000 |
| commit | 60415a2d640364ff7348d4580995494479203f59 (patch) | |
| tree | 08cf94fdcbfd726f8e3eaeeba8395b64adbf105a | |
| parent | 64caef983381f8f1938b4baa748705c149ad7534 (diff) | |
linting changes
| -rw-r--r-- | html_view.py | 41 | ||||
| -rwxr-xr-x | ptv.py | 10 | ||||
| -rw-r--r-- | ptv_client.py | 48 | ||||
| -rw-r--r-- | server.py | 28 | ||||
| -rw-r--r-- | util.py | 19 |
5 files changed, 93 insertions, 53 deletions
diff --git a/html_view.py b/html_view.py index f4dc7b0..98306cb 100644 --- a/html_view.py +++ b/html_view.py @@ -1,6 +1,7 @@ +from urllib import parse import util import ptv_client -from urllib import parse + def format_stop_and_route_name_li(stop_and_route): stop, route = stop_and_route @@ -8,10 +9,12 @@ def format_stop_and_route_name_li(stop_and_route): stop['stop_name'], ptv_client.get_route_type(route['route_type']), route['route_number'], route['route_name']) + def format_stop_and_route_name_ol(stop_and_routes): return '<ol>{}</ol>'.format(''.join( map(format_stop_and_route_name_li, stop_and_routes))) + def format_stop_and_route_name_tr(stop_and_route): stop, route = stop_and_route return """ @@ -23,11 +26,11 @@ def format_stop_and_route_name_tr(stop_and_route): </td> <td>{}</td><td>{}</td><td>{}</td> </tr> -""".format( - route['route_type'], stop['stop_id'], route['route_id'], - parse.quote(stop['stop_name']), parse.quote(route['route_number']), - stop['stop_name'], ptv_client.get_route_type(route['route_type']), - route['route_number'], route['route_name']) +""".format(route['route_type'], stop['stop_id'], route['route_id'], + parse.quote(stop['stop_name']), parse.quote(route['route_number']), + stop['stop_name'], ptv_client.get_route_type(route['route_type']), + route['route_number'], route['route_name']) + def format_stop_and_route_name_table(stop_and_routes): return """ @@ -42,19 +45,24 @@ def format_stop_and_route_name_table(stop_and_routes): </table> """.format(''.join(map(format_stop_and_route_name_tr, stop_and_routes))) + def format_departure_title(stop_name, route_type, route_number): return 'Departures of {} {} at {}'.format( ptv_client.get_route_type(route_type), route_number, stop_name) + def format_departure_tr(departure, direction_names): return """ <tr> <td>{}</td><td>{}</td><td>{}</td> </tr> """.format( - util.format_time(util.parse_time(departure['estimated_departure_utc'])), - util.format_time(util.parse_time(departure['scheduled_departure_utc'])), - direction_names[departure['direction_id']]) + util.format_time(util.parse_time( + departure['estimated_departure_utc'])), + util.format_time(util.parse_time( + departure['scheduled_departure_utc'])), + direction_names[departure['direction_id']]) + def format_departure_table(departures, direction_names): return """ @@ -66,8 +74,9 @@ def format_departure_table(departures, direction_names): </tr> {} </table> -""".format(''.join(format_departure_tr(dep, direction_names) - for dep in departures)) +""".format(''.join( + format_departure_tr(dep, direction_names) for dep in departures)) + def style(): return """ @@ -77,7 +86,8 @@ def style(): </style> """ -def html(title, body): + +def html(title: str, body): return """ <!DOCTYPE html> <html> @@ -90,7 +100,12 @@ def html(title, body): %(body)s </body> </html> -""" % {'title': title, 'style': style(), 'body': body} +""" % { + 'title': title, + 'style': style(), + 'body': body + } + def landing_page(): title = 'PTV timetable search tool' @@ -4,6 +4,7 @@ import ptv_client import cli_view import util + def main(): stop_and_routes = list( ptv_client.get_stop_and_routes(ptv_client.search(input('Query: ')))) @@ -16,9 +17,12 @@ def main(): departures = ptv_client.get_departures_from_stop_and_route( *stop_and_routes[idx]) filtered_deps = util.filter_departures(departures['departures']) - print(cli_view.format_departures( - filtered_deps, - ptv_client.get_direction_names(stop_and_routes[idx][1]['route_id']))) + print( + cli_view.format_departures( + filtered_deps, + ptv_client.get_direction_names( + stop_and_routes[idx][1]['route_id']))) + if __name__ == "__main__": main() diff --git a/ptv_client.py b/ptv_client.py index 652d87b..ec89134 100644 --- a/ptv_client.py +++ b/ptv_client.py @@ -18,56 +18,70 @@ from hashlib import sha1 from urllib import request, error, parse import hmac import json -import pprint import itertools import os _ROUTE_TYPE = ["Train", "Tram", "Bus", "Vline", "Night Bus"] -def get_url(request): + +def get_url(req): key = str.encode(os.environ.get('PTVKEY')) dev_id = os.environ.get('PTVID') - request = parse.quote(request) + ('&' if ('?' in request) else '?') - raw = request + f'devid={dev_id}' + req = parse.quote(req) + ('&' if ('?' in req) else '?') + raw = req + f'devid={dev_id}' hashed = hmac.new(key, raw.encode('utf-8'), sha1) signature = hashed.hexdigest().upper() return f'http://timetableapi.ptv.vic.gov.au{raw}&signature={signature}' + def get_json(url): try: - document = request.urlopen(url).read().decode() + with request.urlopen(url) as opened: + document = opened.read().decode() + return json.loads(document) except error.HTTPError: print("http error!") - return - return json.loads(document) + def get_stops(search_result): - return [{'routes': stop['routes'], - 'stop_name': stop['stop_name'], - 'stop_id': stop['stop_id']} for stop in search_result(stops)] + return [{ + 'routes': stop['routes'], + 'stop_name': stop['stop_name'], + 'stop_id': stop['stop_id'] + } for stop in search_result['stops']] + def get_stop_and_routes(search_result): - return itertools.chain.from_iterable( - [[(stop, route) for route in stop['routes']] - for stop in search_result['stops']]) + return itertools.chain.from_iterable([[(stop, route) + for route in stop['routes']] + for stop in search_result['stops']]) + def search(keyword): return get_json(get_url(f'/v3/search/{keyword}')) + def get_departures(route_type, stop_id, route_id): - return get_json(get_url( - f'/v3/departures/route_type/{route_type}/stop/{stop_id}/route/{route_id}')) + return get_json( + get_url(f'/v3/departures/route_type/{route_type}/stop/{stop_id}' + f'/route/{route_id}')) + def get_departures_from_stop_and_route(stop, route): return get_departures(route['route_type'], stop['stop_id'], route['route_id']) + def get_directions(route_id): return get_json(get_url(f'/v3/directions/route/{route_id}')) + def get_direction_names(route_id): - return {dir['direction_id']: dir['direction_name'] for dir in - get_directions(route_id)['directions']} + return { + dir['direction_id']: dir['direction_name'] + for dir in get_directions(route_id)['directions'] + } + def get_route_type(idx): return _ROUTE_TYPE[idx] @@ -1,13 +1,14 @@ -#!/usr/bin/env python -import ptv_client -import html_view from wsgiref.simple_server import make_server from urllib.parse import parse_qs +import ptv_client +import html_view import util + def is_dep_query(params): - return all(key in params for key in - ('route-type', 'stop', 'route', 'stop-name', 'route-number')) + return all(key in params + for key in ('route-type', 'stop', 'route', 'stop-name')) + def application(environ, start_response): params = parse_qs(environ['QUERY_STRING']) @@ -18,25 +19,26 @@ def application(environ, start_response): title = f'Search results for {query}' response_html = html_view.html(title, response_body) elif is_dep_query(params): - departures = ptv_client.get_departures( - str(params['route-type'][0]), str(params['stop'][0]), str(params['route'][0])) + departures = ptv_client.get_departures(str(params['route-type'][0]), + str(params['stop'][0]), + str(params['route'][0])) filtered_deps = util.filter_departures(departures['departures']) - direction_names = ptv_client.get_direction_names(str(params['route'][0])) + direction_names = ptv_client.get_direction_names( + str(params['route'][0])) response_body = html_view.format_departure_table( filtered_deps, direction_names) title = html_view.format_departure_title( str(params['stop-name'][0]), int(params['route-type'][0]), - str(params['route-number'][0])) + str(params.get('route-number', [''])[0])) response_html = html_view.html(title, response_body) else: response_html = html_view.landing_page() - response_headers = [ - ('Content-Type', 'text/html'), - ('Content-Length', str(len(response_html))) - ] + response_headers = [('Content-Type', 'text/html'), + ('Content-Length', str(len(response_html)))] start_response('200 OK', response_headers) return [response_html.encode()] + httpd = make_server('localhost', 8052, application) httpd.serve_forever() @@ -1,17 +1,22 @@ from datetime import datetime, timezone from zoneinfo import ZoneInfo + def parse_time(maybe_time): if maybe_time: return datetime.fromisoformat(maybe_time[:-1] + '+00:00') - + return None + + def filter_departures(departures): - return [dep for dep in departures - if dep['scheduled_departure_utc'] and - parse_time(dep['scheduled_departure_utc']) > - datetime.now().astimezone(timezone.utc)] + return [ + dep for dep in departures if dep['scheduled_departure_utc'] + and parse_time(dep['scheduled_departure_utc']) > + datetime.now().astimezone(timezone.utc) + ] + def format_time(maybe_time): if maybe_time: - return str(maybe_time.astimezone( - ZoneInfo('Australia/Melbourne')))[:-6] + return str(maybe_time.astimezone(ZoneInfo('Australia/Melbourne')))[:-6] + return None |