89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime
|
|
from hashlib import sha1
|
|
import hmac
|
|
import httpx
|
|
import os
|
|
|
|
dev_id = os.environ['PTV_USER_ID']
|
|
api_key = os.environ['PTV_API_KEY']
|
|
|
|
def sign(request):
|
|
request += '&' if ('?' in request) else '?'
|
|
request += 'devid='
|
|
request += dev_id
|
|
hashed = hmac.new(api_key.encode(), request.encode(), sha1)
|
|
return f'https://timetableapi.ptv.vic.gov.au{request}&signature={hashed.hexdigest().upper()}'
|
|
|
|
def fetch(path):
|
|
res = httpx.get(sign(path))
|
|
res.raise_for_status()
|
|
return res.json()
|
|
|
|
# Lilydale route 9 (type 0) (dir 1 in, 8 out) (stop 1229)
|
|
# Belgrave route 2 (type 0) (dir 1 in, 2 out) (stop 1229)
|
|
# 109 route 722 (type 1) (dir 2 E, 3 W) (stop 2415 E, 2460 W)
|
|
# 70 route 940 (type 1) (dir 28 E, 29 W) (stop 2161 E, 2162 W)
|
|
# 766 route 15800 (type 2) (dir 13 N, 207 S) (stop 17861)
|
|
# 612 route 13024 (type 2) (dir 13 N, 158 S) (stop 17861)
|
|
|
|
class Route:
|
|
def __init__(self, name, route_id, route_type_id, directions, stops):
|
|
self.name = name
|
|
self.route_id = route_id
|
|
self.route_type_id = route_type_id
|
|
self.directions = directions
|
|
self.stops = stops
|
|
|
|
local_routes = [
|
|
Route('Lilydale', 9 , 0, [(1, 'in'), (8, 'out')] , [1229] ),
|
|
Route('Belgrave', 2 , 0, [(1, 'in'), (2, 'out')] , [1229] ),
|
|
Route('109' , 722 , 1, [(2, 'E'), (3, 'W')] , [2415, 2460]),
|
|
Route('70' , 940 , 1, [(28, 'E'), (29, 'W')] , [2162, 2161]),
|
|
Route('766' , 15800, 2, [(13, 'N'), (188, 'S')] , [17861] ),
|
|
Route('612' , 13024, 2, [(13, 'N'), (139, 'S')] , [17861] ),
|
|
]
|
|
# exp 951624
|
|
# non-exp 951826
|
|
|
|
def fetch_departures(stop_id, route_type_id):
|
|
return fetch(f'/v3/departures/route_type/{route_type_id}/stop/{stop_id}')['departures']
|
|
|
|
def fetch_run(run_ref):
|
|
return fetch(f'/v3/runs/{run_ref}')['runs']
|
|
|
|
def departure_time(dep):
|
|
return datetime.fromisoformat(dep['estimated_departure_utc'] or dep['scheduled_departure_utc'])
|
|
|
|
def get_departure_data():
|
|
bus = fetch_departures(17861, 2)
|
|
train = fetch_departures(1229, 0)
|
|
tram_109w = fetch_departures(2460, 1)
|
|
tram_70w = fetch_departures(2161, 1)
|
|
|
|
next_express_dep = None
|
|
train.sort(key=departure_time)
|
|
for dep in train:
|
|
if dep['direction_id'] == 1:
|
|
match fetch_run(dep['run_ref']):
|
|
case [run]:
|
|
if run['express_stop_count'] > 1: # Ignore East Richmond
|
|
next_express_dep = dep
|
|
break
|
|
|
|
def earliest(deps):
|
|
return min((departure_time(dep) for dep in deps), default=None)
|
|
|
|
return {
|
|
'612 S': earliest(dep for dep in bus if dep['route_id'] == 13024 and dep['direction_id'] == 139),
|
|
'612 N': earliest(dep for dep in bus if dep['route_id'] == 13024 and dep['direction_id'] == 13),
|
|
'766 N': earliest(dep for dep in bus if dep['route_id'] == 15800 and dep['direction_id'] == 13),
|
|
|
|
'Union W': earliest(dep for dep in train if dep['direction_id'] == 1 and dep is not next_express_dep),
|
|
'Union W Express': departure_time(next_express_dep) if next_express_dep else None,
|
|
'Union E': earliest(dep for dep in train if dep['direction_id'] in {2, 8}),
|
|
|
|
'109 W': earliest(tram_109w),
|
|
'70 W': earliest(tram_70w),
|
|
}
|