travelynx/contrib/i3bar-snippet.py

#!/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) + ",")