mirror of
https://github.com/derf/travelynx
synced 2024-09-20 14:41:56 +00:00
111 lines
3.5 KiB
Python
Executable file
111 lines
3.5 KiB
Python
Executable file
#!/usr/bin/python3
|
|
|
|
# This script queries the Travelynx API if you are checked into a train. If
|
|
# yes, bahn.expert is additionally queried for the next stop, and a JSON object
|
|
# like this is written to stdout:
|
|
# {"full_text": "RE26824, next: D\u00fcren at <span fgcolor=\"#ff0000\">15:38+5</span>, dest: Aachen Hbf at <span fgcolor=\"#ff0000\">16:07+5</span>", "markup": "pango"},
|
|
# The script then exits.
|
|
#
|
|
# Configuration:
|
|
# - Place your API key from https://travelynx.de/account at
|
|
# ~/.config/travelynx.conf .
|
|
# - Then integrate into whatever generates your i3bar input.
|
|
# - Make sure you use i3bar with a pango font, so that the color tags are
|
|
# picked up.
|
|
|
|
|
|
from datetime import datetime
|
|
import dateutil
|
|
import dateutil.parser
|
|
import json
|
|
from pathlib import Path
|
|
import requests
|
|
import sys
|
|
import xdg # not pyxdg!
|
|
|
|
|
|
def format_stop(stop_name, predicted_arrival_timestamp, delay):
|
|
color = "#ffffff"
|
|
if delay > 0:
|
|
if delay <= 2:
|
|
color = "#ffff00"
|
|
else:
|
|
color = "#ff0000"
|
|
delayStr = "({:+.0f})".format(delay)
|
|
else:
|
|
delayStr = ""
|
|
if isinstance(predicted_arrival_timestamp, int):
|
|
predicted_arrival_time = datetime.fromtimestamp(predicted_arrival_timestamp)
|
|
else:
|
|
# We assume it's datetime already.
|
|
predicted_arrival_time = predicted_arrival_timestamp
|
|
return f'{stop_name} at <span fgcolor="{color}">{predicted_arrival_time:%H:%M}{delayStr}</span>'
|
|
|
|
|
|
api_key_path = Path(xdg.xdg_config_home(), "travelynx.conf")
|
|
if api_key_path.exists():
|
|
with api_key_path.open("r") as f:
|
|
api_key = f.read().strip()
|
|
else:
|
|
print(
|
|
f"Could not find Travelynx API key at {api_key_path}.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
api_base = f"https://travelynx.de/api/v1/status/{api_key}"
|
|
try:
|
|
res = requests.get(api_base)
|
|
except requests.exceptions.ConnectionError:
|
|
print(
|
|
json.dumps({"full_text": "Could not connect to travelynx", "color": "#ff0000"})
|
|
+ ","
|
|
)
|
|
sys.exit()
|
|
|
|
j = res.json()
|
|
# print(json.dumps(j, sort_keys=True, indent=4), file=sys.stderr)
|
|
|
|
if not j["checkedIn"]:
|
|
sys.exit()
|
|
|
|
out_fields = []
|
|
|
|
train = "{}{}".format(j["train"]["type"], j["train"]["no"])
|
|
out_fields.append(train)
|
|
destination_name = j["toStation"]["name"]
|
|
scheduled_arrival_timestamp = j["toStation"]["scheduledTime"]
|
|
predicted_arrival_timestamp = j["toStation"]["realTime"]
|
|
delay = (predicted_arrival_timestamp - scheduled_arrival_timestamp) / 60
|
|
|
|
try:
|
|
details_res = requests.get(f"https://bahn.expert/api/hafas/v2/details/{train}")
|
|
details = details_res.json()
|
|
# print(json.dumps(details, sort_keys=True, indent=4), file=sys.stderr)
|
|
next_stop_name = details["currentStop"]["station"]["title"]
|
|
if next_stop_name == destination_name:
|
|
out_fields.append("next")
|
|
else:
|
|
next_predicted_arrival_time = dateutil.parser.isoparse(
|
|
details["currentStop"]["arrival"]["time"]
|
|
)
|
|
next_predicted_arrival_time = next_predicted_arrival_time.astimezone(
|
|
dateutil.tz.tzlocal()
|
|
)
|
|
next_delay = details["currentStop"]["arrival"]["delay"]
|
|
out_fields.append(
|
|
"next: "
|
|
+ format_stop(next_stop_name, next_predicted_arrival_time, next_delay)
|
|
)
|
|
except requests.exceptions.ConnectionError:
|
|
pass
|
|
|
|
out_fields.append(
|
|
"dest: " + format_stop(destination_name, predicted_arrival_timestamp, delay)
|
|
)
|
|
|
|
s = ", ".join(out_fields)
|
|
|
|
out_obj = {"full_text": s, "markup": "pango"}
|
|
print(json.dumps(out_obj) + ",")
|