73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
from datetime import datetime, timedelta
|
|
from hashlib import sha1
|
|
import hmac
|
|
import httpx
|
|
import json
|
|
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, cache, ttl=timedelta(minutes=5)):
|
|
def do():
|
|
res = httpx.get(sign(path))
|
|
res.raise_for_status()
|
|
return res.text
|
|
return json.loads(cache.get(path, ttl, do))
|
|
|
|
# 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)
|
|
|
|
def fetch_departures(stop_id, route_type_id, cache):
|
|
# TODO: Expand runs
|
|
return fetch(f'/v3/departures/route_type/{route_type_id}/stop/{stop_id}', cache)['departures']
|
|
|
|
def fetch_run(run_ref, cache):
|
|
return fetch(f'/v3/runs/{run_ref}', cache, timedelta(hours=8))['runs']
|
|
|
|
def departure_time(dep):
|
|
return datetime.fromisoformat(dep['estimated_departure_utc'] or dep['scheduled_departure_utc'])
|
|
|
|
def get_departure_data(now, cache):
|
|
bus = fetch_departures(17861, 2, cache)
|
|
train = fetch_departures(1229, 0, cache)
|
|
tram_109w = fetch_departures(2460, 1, cache)
|
|
tram_70w = fetch_departures(2161, 1, cache)
|
|
|
|
next_express_dep = None
|
|
train.sort(key=departure_time)
|
|
for dep in train:
|
|
if dep['direction_id'] == 1 and departure_time(dep) > now:
|
|
match fetch_run(dep['run_ref'], cache):
|
|
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 if departure_time(dep) > now), 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),
|
|
}
|