#!/usr/bin/env python # -*- coding: utf-8 -*- # Autor: Marcel Weschke - 2025 # Linienfahrpläne-Quelle: # # Linie 15: https://www.rmv.de/c/fileadmin/import/timetable/traffiQ_2022_AHF_Linie_15_ab_2021_12_12.pdf # Linie 16: https://www.rmv.de/c/fileadmin/import/timetable/traffiQ_2022_Buch_Linie_16_ab_2021_12_12.pdf # # ============================================================================= # RMV Live-Abfahrtstafel für Buchrainplatz # # Beispiel-Programmausgabe im Terminal: # # ➜ Desktop python RMV_DigiAnzeige_Buchreinplatz.py # # Buchrainplatz – Live-Abfahrtstafel – 11:57:10 # # ← Richtung Frankfurt # (Westen) # ┏━━━━━━━┳━━━━━━┳━━━━┓ # ┃ Linie ┃ Ziel ┃ In ┃ # ┡━━━━━━━╇━━━━━━╇━━━━┩ # └───────┴──────┴────┘ # # → Richtung Oberrad / # Offenbach (Osten) # ┏━━━━━━━┳━━━━━━┳━━━━┓ # ┃ Linie ┃ Ziel ┃ In ┃ # ┡━━━━━━━╇━━━━━━╇━━━━┩ # └───────┴──────┴────┘ # # ============================================================================= # %% Load libraries import argparse import requests import time import json import os import sys from datetime import datetime from rich.console import Console, Group from rich.table import Table from rich.live import Live from rich.align import Align # %% Main # Option 1: Manuell API Key laden: #API_KEY = "" #"DEIN_API_KEY_HIER" # Option 2: Automatisch API Key aus einer Textdatei laden: try: with open("apikey.txt", "r", encoding="utf-8") as f: API_KEY = f.read().strip() except FileNotFoundError: raise FileNotFoundError("Fehler: Datei 'apikey.txt' nicht gefunden. Bitte API-Key in dieser Datei hinterlegen.") except Exception as e: raise RuntimeError(f"Fehler beim Laden des API-Key: {e}") START_ID = "3001605" # Buchrainplatz #START_ID = "3000906" # Lokalbahnhof/Textorstraße MAX_PER_DIRECTION = 12 URL = "https://www.rmv.de/hapi/departureBoard" # Linienfarben # Colorcode-Quelle: https://commons.wikimedia.org/wiki/Template:Frankfurt_transit_icons line_colors = { "S1": "#0096da", # S-Bahn-lines "S2": "#ee161f", "S3": "#00aa96", "S4": "#fac800", "S5": "#965b33", "S6": "#f27718", "S7": "#1b533d", "S8": "#8dc63c", "S9": "#91208f", "11": "#fbbd02", # Tram-lines "12": "#c83710", "14": "#f14517", "16": "#f4730e", "15": "#fbbd02", "17": "#f14517", "18": "#f89d06", "19": "#ef3200", "20": "#f79700", "21": "#f89d06", "U1": "#a50010", # U-Bahn-lines "U2": "#00a850", "U3": "#333f8d", "U4": "#e12d82", "U5": "#015c1b", "U6": "#007ec6", "U7": "#df9100", "U8": "#c066a3", "U9": "#ffd700" } console = Console() def fetch_departures(start_id, max_journeys=5, debug=False): """ Ruft Abfahrtsdaten für eine gegebene RMV-Haltestellen-ID von der RMV-HAFAS-API ab. Args: starrt_id (str): Die Haltestellen-ID (z. B. '3001605' für Buchrainplatz). max_journeys (int, optional): Maximale Anzahl zurückzugebender Fahrten. Default: 5. debug (bool, optional): Wenn True, wird die vollständige API-JSON-Ausgabe angezeigt. Returns: list[dict]: Liste der Abfahrten mit Feldern wie Linie, Ziel, Datum und Uhrzeit. """ params = { "id": start_id, "format": "json", "accessId": API_KEY, "limit": max_journeys } resp = requests.get(URL, params=params, timeout=10) resp.raise_for_status() data = resp.json() if debug: console.print("[yellow]RAW API Response:[/yellow]") console.print_json(json.dumps(data, indent=2)) departures = [] if "DepartureBoard" in data and isinstance(data["DepartureBoard"], dict): dep = data["DepartureBoard"].get("Departure", []) departures = dep if isinstance(dep, list) else [dep] elif "Departure" in data: departures = data["Departure"] if isinstance(data["Departure"], list) else [data["Departure"]] return departures def build_table(departures): """ Baut zwei getrennte Tabellen (West- und Ost-Richtung) mit den nächsten Abfahrten. Die Richtung wird anhand von Schlüsselwörtern im Ziel bestimmt. Duplikate werden gefiltert, und die Abfahrten werden nach Zeit sortiert. Zusätzlich werden Linien mit definierten Farben dargestellt und die Anzahl der angezeigten Einträge pro Richtung begrenzt. Args: departures (list[dict]): Liste von Abfahrtsdatensätzen aus der RMV-API. Returns: rich.console.Group: Zwei ausgerichtete Rich-Tabellen, eine für jede Richtung. """ now = datetime.now().strftime("%H:%M:%S") table_west = Table(title=f"← Richtung Frankfurt (Westen) – {now}") table_west.add_column("Linie", justify="center", style="cyan", no_wrap=True) table_west.add_column("Ziel", justify="left", style="magenta") table_west.add_column("In", justify="right", style="green") table_east = Table(title="→ Richtung Oberrad / Offenbach (Osten)") table_east.add_column("Linie", justify="center", style="cyan", no_wrap=True) table_east.add_column("Ziel", justify="left", style="magenta") table_east.add_column("In", justify="right", style="green") west_keywords = ["Frankfurt", "Ginnheim", "Lokalbahnhof", "Haardtwaldplatz"] east_keywords = ["Oberrad", "Offenbach", "Stadtgrenze", "Louisa", "Hugo-Junkers", "Balduinstraße"] excluded_lines = {"81", "M36"} # zu excludierende Buslinien seen = set() westbound = [] eastbound = [] for dep in departures: line = dep.get("Product", [{}])[0].get("line", "?") if isinstance(dep.get("Product"), list) else dep.get("Product", {}).get("line", "?") if line in excluded_lines: continue dest = dep.get("direction", "").strip() if not dest: continue date = dep.get("rtDate") or dep.get("date") time_ = dep.get("rtTime") or dep.get("time") if not date or not time_: continue try: dep_time = datetime.fromisoformat(f"{date}T{time_}") except ValueError: continue mins = int((dep_time - datetime.now()).total_seconds() / 60) mins_str = f"{mins} Min" if mins >= 0 else "Jetzt" key = (line, dest, dep_time) if key in seen: continue seen.add(key) if any(k in dest for k in west_keywords): westbound.append((dep_time, line, dest, mins_str)) elif any(k in dest for k in east_keywords): eastbound.append((dep_time, line, dest, mins_str)) # Sortieren westbound.sort(key=lambda x: x[0]) eastbound.sort(key=lambda x: x[0]) # Begrenzen auf MAX_PER_DIRECTION westbound = westbound[:MAX_PER_DIRECTION] eastbound = eastbound[:MAX_PER_DIRECTION] # Tabellen füllen (Linien einfärben) for _, line, dest, mins_str in westbound: line_style = f"[{line_colors[line]}]{line}[/]" if line in line_colors else line table_west.add_row(line_style, dest, mins_str) for _, line, dest, mins_str in eastbound: line_style = f"[{line_colors[line]}]{line}[/]" if line in line_colors else line table_east.add_row(line_style, dest, mins_str) return Group(Align.center(table_west), Align.center(table_east)) def main(poll_interval=30, debug=False): """ Hauptschleife: Holt regelmäßig Abfahrtsdaten und aktualisiert die Anzeige. Args: poll_interval (int, optional): Abfrageintervall in Sekunden. Default: 30. debug (bool, optional): Wenn True, wird die API-JSON-Ausgabe angezeigt. """ with Live(console=console, refresh_per_second=1) as live: while True: try: departures = fetch_departures(START_ID, debug=debug) table = build_table(departures) live.update(Align.center(table)) except Exception as e: console.print(f"[red]Fehler bei API-Aufruf:[/red] {e}") time.sleep(poll_interval) if __name__ == "__main__": parser = argparse.ArgumentParser(description="RMV Live-Abfahrtstafel für Buchrainplatz") parser.add_argument("--debug", action="store_true", help="Zeige API-JSON-Ausgabe zur Fehlersuche") args = parser.parse_args() main(debug=args.debug)