308 lines
11 KiB
Python
308 lines
11 KiB
Python
#!/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 re
|
||
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
|
||
|
||
# Neu für Webversion
|
||
from textual.app import App, ComposeResult
|
||
from textual.containers import Container
|
||
from textual.widgets import Header, Footer, Input, DataTable, Static
|
||
from textual.color import Color
|
||
|
||
# %% 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=12):
|
||
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()
|
||
|
||
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 _parse_mins(cell) -> int:
|
||
"""Wandelt '15 Min', '0 Min' oder 'Jetzt' in eine Zahl um (für Sortierung)."""
|
||
if isinstance(cell, int):
|
||
return cell
|
||
s = str(cell).strip().lower()
|
||
if s == "jetzt":
|
||
return 0
|
||
m = re.search(r"-?\d+", s)
|
||
return int(m.group()) if m else 999 # Unbekanntes/fehlendes ans Ende
|
||
|
||
def build_table_data(departures):
|
||
"""Erzeugt Datenlisten für West- und Ost-Richtung."""
|
||
now = datetime.now().strftime("%H:%M:%S")
|
||
|
||
west_keywords = ["Frankfurt", "Ginnheim", "Lokalbahnhof", "Haardtwaldplatz"]
|
||
east_keywords = ["Oberrad", "Offenbach", "Stadtgrenze", "Louisa", "Hugo-Junkers", "Balduinstraße"]
|
||
excluded_lines = {"81", "M36"}
|
||
|
||
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((line, dest, mins_str))
|
||
elif any(k in dest for k in east_keywords):
|
||
eastbound.append((line, dest, mins_str))
|
||
|
||
westbound = sorted(westbound, key=lambda x: x[2])[:MAX_PER_DIRECTION]
|
||
eastbound = sorted(eastbound, key=lambda x: x[2])[:MAX_PER_DIRECTION]
|
||
|
||
return now, westbound, eastbound
|
||
|
||
|
||
# -------------------
|
||
# Terminalmodus
|
||
# -------------------
|
||
def run_terminal_mode():
|
||
with Live(console=console, refresh_per_second=1) as live:
|
||
while True:
|
||
try:
|
||
departures = fetch_departures(START_ID)
|
||
now, west, east = build_table_data(departures)
|
||
|
||
# KORREKTE NUMERISCHE SORTIERUNG
|
||
west.sort(key=lambda row: _parse_mins(row[2]))
|
||
east.sort(key=lambda row: _parse_mins(row[2]))
|
||
|
||
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")
|
||
for line, dest, mins in west:
|
||
style = f"[{line_colors[line]}]{line}[/]" if line in line_colors else line
|
||
table_west.add_row(style, dest, mins)
|
||
|
||
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")
|
||
for line, dest, mins in east:
|
||
style = f"[{line_colors[line]}]{line}[/]" if line in line_colors else line
|
||
table_east.add_row(style, dest, mins)
|
||
|
||
live.update(Align.center(Group(Align.center(table_west), Align.center(table_east))))
|
||
except Exception as e:
|
||
console.print(f"[red]Fehler bei API-Aufruf:[/red] {e}")
|
||
time.sleep(30)
|
||
|
||
|
||
# -------------------
|
||
# Webmodus mit Textual
|
||
# -------------------
|
||
class RMVApp(App):
|
||
CSS_PATH = None
|
||
CSS = """
|
||
Screen {
|
||
align: center middle;
|
||
}
|
||
#input_id {
|
||
width: 40;
|
||
margin: 1;
|
||
}
|
||
"""
|
||
|
||
BINDINGS = [("q", "quit", "Beenden")]
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.current_id = START_ID
|
||
self.table_west = DataTable()
|
||
self.table_east = DataTable()
|
||
|
||
def compose(self) -> ComposeResult:
|
||
yield Header()
|
||
yield Input(
|
||
value=self.current_id,
|
||
placeholder="START_ID eingeben und Enter drücken",
|
||
id="input_id"
|
||
)
|
||
yield Container(
|
||
Static("← Richtung Frankfurt (Westen)"),
|
||
self.table_west,
|
||
Static(""), # Leerzeile als Abstand
|
||
Static("→ Richtung Oberrad / Offenbach (Osten)"),
|
||
self.table_east
|
||
)
|
||
yield Footer()
|
||
|
||
async def on_mount(self) -> None:
|
||
# Spaltenüberschriften einmalig setzen
|
||
self.table_west.add_columns("Linie", "Ziel", "In")
|
||
self.table_east.add_columns("Linie", "Ziel", "In")
|
||
|
||
await self.refresh_tables()
|
||
self.set_interval(30, self.refresh_tables)
|
||
|
||
async def on_input_submitted(self, event: Input.Submitted) -> None:
|
||
"""START_ID ändern, wenn Enter gedrückt wird"""
|
||
self.current_id = event.value.strip()
|
||
await self.refresh_tables()
|
||
|
||
async def refresh_tables(self) -> None:
|
||
try:
|
||
departures = fetch_departures(self.current_id)
|
||
_, west, east = build_table_data(departures)
|
||
|
||
# KORREKTE NUMERISCHE SORTIERUNG
|
||
west.sort(key=lambda row: _parse_mins(row[2]))
|
||
east.sort(key=lambda row: _parse_mins(row[2]))
|
||
|
||
self.update_table(self.table_west, west)
|
||
self.update_table(self.table_east, east)
|
||
except Exception as e:
|
||
self.console.print(f"[red]Fehler:[/red] {e}")
|
||
|
||
def update_table(self, table: DataTable, rows):
|
||
"""Befüllt eine DataTable mit neuen Daten und Farbcodes"""
|
||
table.clear() # Nur Zeilen löschen, Spalten bleiben
|
||
for line, dest, mins in rows:
|
||
color_hex = line_colors.get(line)
|
||
if color_hex:
|
||
table.add_row(f"[{color_hex}]{line}[/]", dest, mins)
|
||
else:
|
||
table.add_row(line, dest, mins)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--light", action="store_true", help="Terminalmodus")
|
||
args = parser.parse_args()
|
||
|
||
if args.light:
|
||
run_terminal_mode()
|
||
else:
|
||
RMVApp().run()
|