diff options
| author | Yuchen Pei <hi@ypei.me> | 2022-04-04 13:14:44 +1000 |
|---|---|---|
| committer | Yuchen Pei <hi@ypei.me> | 2022-04-04 13:45:47 +1000 |
| commit | c7e5108443febfa5995d143ed8450ab09b0cb6bd (patch) | |
| tree | db32a27d385209797d062ca94698fdb57e77d91b | |
| parent | 29584f296126c69c231ef883d286fa693ac94b98 (diff) | |
refactoring and adding web client
| -rw-r--r-- | cli_view.py | 19 | ||||
| -rw-r--r-- | html_view.py | 83 | ||||
| -rwxr-xr-x | ptv.py | 128 | ||||
| -rw-r--r-- | ptv_client.py | 73 | ||||
| -rw-r--r-- | server.py | 32 | ||||
| -rw-r--r-- | util.py | 17 |
6 files changed, 237 insertions, 115 deletions
diff --git a/cli_view.py b/cli_view.py new file mode 100644 index 0000000..9c022f9 --- /dev/null +++ b/cli_view.py @@ -0,0 +1,19 @@ +import util +import ptv_client + +def format_stop_and_route_name(stop_and_route): + stop, route = stop_and_route + return 'Stop: {}, Route: {} {} {}'.format( + stop['stop_name'], ptv_client.get_route_type(route['route_type']), + route['route_number'], route['route_name']) + +def format_stop_and_route_names(stop_and_routes): + return '\n'.join(f'[{i}] {result}' for i, result in enumerate( + map(format_stop_and_route_name, stop_and_routes))) + +def format_departures(departures, direction_names): + return '\n'.join(['estimated: {}; scheduled: {}; direction: {}'.format( + util.format_time(util.parse_time(dep['estimated_departure_utc'])), + util.format_time(util.parse_time(dep['scheduled_departure_utc'])), + direction_names[dep['direction_id']]) + for dep in departures]) diff --git a/html_view.py b/html_view.py new file mode 100644 index 0000000..2e82066 --- /dev/null +++ b/html_view.py @@ -0,0 +1,83 @@ +import util +import ptv_client + +def format_stop_and_route_name_li(stop_and_route): + stop, route = stop_and_route + return '<li>Stop: {}, Route: {} {} {}</li>'.format( + 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 """ +<tr> + <td> + <a href="/?route-type={}&stop={}&route={}">{}</a> + </td> + <td>{}</td><td>{}</td><td>{}</td> +</tr> +""".format( + route['route_type'], stop['stop_id'], route['route_id'], + 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 """ +<table> + <tr> + <th>Stop</th> + <th>Route type</th> + <th>Route number</th> + <th>Route name</th> + </tr> + {} +</table> +""".format(''.join(map(format_stop_and_route_name_tr, stop_and_routes))) + +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']]) + +def format_departure_table(departures, direction_names): + return """ +<table> + <tr> + <th>Estimated</th> + <th>Scheduled</th> + <th>Direction</th> + </tr> + {} +</table> +""".format(''.join(format_departure_tr(dep, direction_names) + for dep in departures)) + +def style(): + return """ +<style> + tr:nth-child(even) {background-color: #f2f2f2;} + tr:hover {background-color: coral;} +</style> +""" + +def html(body): + return """ +<!DOCTYPE html> +<html> +<head> + %(style)s +</head> +<body> + %(body)s +</body> +</html> +""" % {'style': style(), 'body': body} @@ -1,126 +1,24 @@ #!/usr/bin/python -# Copyright (C) 2022 Yuchen Pei. -# -# ptv.py is free software: you can redistribute it and/or modify it -# under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. - -# ptv.py is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Affero General Public License for more details. - -# You should have received a copy of the GNU Affero General Public -# License along with ptv.py. If not, see -# <https://www.gnu.org/licenses/>. - -# PTV (Public Transport Victoria) timetable cli query tool - -from hashlib import sha1 -from urllib import request, error, parse -from datetime import datetime, timezone -from zoneinfo import ZoneInfo -import hmac -import json -import pprint -import itertools -import os - -_ROUTE_TYPE = ["Train", "Tram", "Bus", "Vline", "Night Bus"] - -def get_url(request): - 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}' - 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() - 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)] - -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']]) - -def search(keyword): - return get_json(get_url(f'/v3/search/{keyword}')) - -def format_stop_and_route_name(stop_and_route): - stop, route = stop_and_route - return 'Stop: {}, Route: {} {} {}'.format( - stop['stop_name'], _ROUTE_TYPE[route['route_type']], - route['route_number'], route['route_name']) - -def format_stop_and_route_names(stop_and_routes): - return '\n'.join(f'[{i}] {result}' for i, result in enumerate( - map(format_stop_and_route_name, stop_and_routes))) - return '\n'.join(f'[{i}] {result}' for i, result in enumerate( - [format_stop_and_route_name( - stop['stop_name'], _ROUTE_TYPE[route['route_type']], - route['route_number'], route['route_name']) - for stop, route in stop_and_routes])) - -def get_departures(stop, route): - return get_json(get_url( - '/v3/departures/route_type/{}/stop/{}/route/{}'.format( - route['route_type'], stop['stop_id'], - route['route_id']))) - -def parse_time(maybe_time): - if maybe_time: - return datetime.fromisoformat(maybe_time[:-1] + '+00:00') - -def format_time(maybe_time): - if maybe_time: - return str(maybe_time.astimezone( - ZoneInfo('Australia/Melbourne')))[:-6] - -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)] - -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']} - -def format_departures(departures, direction_names): - return '\n'.join(['estimated: {}; scheduled: {}; direction: {}'.format( - format_time(parse_time(dep['estimated_departure_utc'])), - format_time(parse_time(dep['scheduled_departure_utc'])), - direction_names[dep['direction_id']]) - for dep in filter_departures(departures['departures'])]) +import ptv_client +import cli_view +import util def main(): - stop_and_routes = list(get_stop_and_routes(search(input('Query: ')))) + stop_and_routes = list( + ptv_client.get_stop_and_routes(ptv_client.search(input('Query: ')))) if not stop_and_routes: print("No results") return - print(format_stop_and_route_names(stop_and_routes)) + print(cli_view.format_stop_and_route_names(stop_and_routes)) idx = int(input('Choose a number: ')) - print(format_stop_and_route_name(stop_and_routes[idx])) - print(format_departures( - get_departures(*stop_and_routes[idx]), - get_direction_names(stop_and_routes[idx][1]['route_id']))) + print(cli_view.format_stop_and_route_name(stop_and_routes[idx])) + 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']))) if __name__ == "__main__": main() diff --git a/ptv_client.py b/ptv_client.py new file mode 100644 index 0000000..652d87b --- /dev/null +++ b/ptv_client.py @@ -0,0 +1,73 @@ +# Copyright (C) 2022 Yuchen Pei. +# +# ptv.py is free software: you can redistribute it and/or modify it +# under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. + +# ptv.py is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public +# License along with ptv.py. If not, see +# <https://www.gnu.org/licenses/>. + +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): + 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}' + 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() + 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)] + +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']]) + +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}')) + +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']} + +def get_route_type(idx): + return _ROUTE_TYPE[idx] diff --git a/server.py b/server.py new file mode 100644 index 0000000..1717ed2 --- /dev/null +++ b/server.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python +import ptv_client +import html_view +from wsgiref.simple_server import make_server +from urllib.parse import parse_qs +import util + +def application(environ, start_response): + d = parse_qs(environ['QUERY_STRING']) + if 'q' in d: + query = str(d.get('q', [''])[0]) + response_body = html_view.format_stop_and_route_name_table( + list(ptv_client.get_stop_and_routes(ptv_client.search(query)))) + else: + print(d) + departures = ptv_client.get_departures( + str(d['route-type'][0]), str(d['stop'][0]), str(d['route'][0])) + filtered_deps = util.filter_departures(departures['departures']) + direction_names = ptv_client.get_direction_names(str(d['route'][0])) + response_body = html_view.format_departure_table( + filtered_deps, direction_names) + + response_html = html_view.html(response_body) + 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() @@ -0,0 +1,17 @@ +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') + +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)] + +def format_time(maybe_time): + if maybe_time: + return str(maybe_time.astimezone( + ZoneInfo('Australia/Melbourne')))[:-6] |