AKP
d20ccc03f4
Add `formattedSchedules.json` Add `requirements.txt` Add `scheduleExport.py` Add `stations.json` Add `__main__.py`
97 lines
No EOL
2.8 KiB
Python
97 lines
No EOL
2.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import sys
|
|
import datetime
|
|
import requests
|
|
|
|
|
|
def process_file(fname: str):
|
|
tiploc_to_crs = {}
|
|
res = []
|
|
|
|
with open(fname) as f:
|
|
while True:
|
|
line = f.readline()
|
|
if line == "":
|
|
# EOF just spits out empty lines
|
|
break
|
|
line = json.loads(line)
|
|
|
|
if "TiplocV1" in line:
|
|
line = line["TiplocV1"]
|
|
if line["crs_code"] is not None:
|
|
tiploc_to_crs[line["tiploc_code"]] = line["crs_code"]
|
|
continue
|
|
|
|
if "JsonScheduleV1" not in line:
|
|
continue
|
|
|
|
line = line["JsonScheduleV1"]
|
|
|
|
if line.get("train_status", "") != "P":
|
|
continue
|
|
|
|
start_date = datetime.datetime.strptime(line["schedule_start_date"], "%Y-%m-%d")
|
|
end_date = datetime.datetime.strptime(line["schedule_end_date"], "%Y-%m-%d")
|
|
now = datetime.datetime.now()
|
|
|
|
if start_date - now > datetime.timedelta(0) or end_date - now < datetime.timedelta(0):
|
|
continue
|
|
|
|
output = {}
|
|
|
|
output["days"] = line["schedule_days_runs"]
|
|
output["toc"] = line["atoc_code"]
|
|
output["signallingID"] = line["schedule_segment"]["signalling_id"]
|
|
|
|
stops = []
|
|
|
|
for schloc in line["schedule_segment"]["schedule_location"]:
|
|
tiploc = schloc["tiploc_code"]
|
|
|
|
if tiploc not in tiploc_to_crs:
|
|
print(f"WARNING: no CRS for TIPLOC {tiploc}", file=sys.stderr)
|
|
continue
|
|
|
|
pubdep = schloc.get("public_departure")
|
|
pubarr = schloc.get("public_arrival")
|
|
|
|
if pubdep is None and pubarr is None:
|
|
continue
|
|
|
|
stops.append({
|
|
"time": pubdep if pubdep is not None else pubarr,
|
|
"crs": tiploc_to_crs[tiploc],
|
|
})
|
|
|
|
output["stops"] = stops
|
|
|
|
if len(stops) != 0:
|
|
res.append(output)
|
|
|
|
with open("formattedSchedules.json", "w") as f:
|
|
json.dump(res, f, indent=1)
|
|
|
|
with open("stations.json", "w") as f:
|
|
json.dump(fetch_crs_info(tiploc_to_crs.values()), f, indent=1)
|
|
|
|
|
|
def fetch_crs_info(all_crs):
|
|
res = {}
|
|
|
|
req_body = "[out:json];" + ("".join(map(lambda x: f'node["ref:crs"="{x}"]; out;', all_crs)))
|
|
r = requests.post("http://overpass-api.de/api/interpreter", data={"data": req_body})
|
|
r.raise_for_status()
|
|
j = r.json()
|
|
for x in j["elements"]:
|
|
res[x["tags"]["ref:crs"]] = {"lat": x["lat"], "lon": x["lon"], "name": x["tags"]["name"]}
|
|
|
|
return res
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
FNAME = sys.argv[1]
|
|
|
|
process_file(FNAME) |