68 lines
2.4 KiB
Python
68 lines
2.4 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))
|
|
try:
|
|
res.raise_for_status()
|
|
except:
|
|
try:
|
|
print(res.json())
|
|
except:
|
|
pass
|
|
raise
|
|
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')] , [2415, 2460]),
|
|
Route('766' , 15800, 2, [(13, 'N'), (207, 'S')] , [17861] ),
|
|
Route('612' , 13024, 2, [(13, 'N'), (158, 'S')] , [17861] ),
|
|
]
|
|
|
|
def fetch_departures(stop_id, route_type_id):
|
|
deps = fetch(f'/v3/departures/route_type/{route_type_id}/stop/{stop_id}')['departures']
|
|
for dep in deps:
|
|
for key in ['estimated_departure_utc', 'scheduled_departure_utc']:
|
|
if dep[key]:
|
|
dep[key] = datetime.fromisoformat(dep[key])
|
|
deps.sort(key=lambda dep: dep['estimated_departure_utc'] or dep['scheduled_departure_utc'])
|
|
return deps
|
|
|
|
def get_departure_data():
|
|
by_route_type = defaultdict(list)
|
|
for route_type, stop in set((r.route_type_id, s) for r in local_routes for s in r.stops):
|
|
by_route_type[route_type] += fetch_departures(stop, route_type)
|
|
return by_route_type
|