mwio/RMV-Abfahrtstafel-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated functionality, --light (terminal mode) or Textual mode (as default)

Browse Source
This commit is contained in:
Marcel Weschke
2025-08-14 16:34:05 +02:00
parent dc4e8c6ffa
commit 0d539cd470
1 changed files with 121 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
RMV_Digital_Board_light.py
View File

@@ -39,14 +39,18 @@ 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
# 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"
@@ -101,20 +105,7 @@ line_colors = {
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.
"""
def fetch_departures(start_id, max_journeys=12):
params = {
"id": start_id,
"format": "json",
@@ -125,10 +116,6 @@ def fetch_departures(start_id, max_journeys=5, debug=False):
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", [])
@@ -138,47 +125,21 @@ def fetch_departures(start_id, max_journeys=5, debug=False):
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.
"""
def build_table_data(departures):
"""Erzeugt Datenlisten für West- und Ost-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
excluded_lines = {"81", "M36"}
seen = set()
westbound = []
eastbound = []
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
@@ -202,53 +163,128 @@ def build_table(departures):
seen.add(key)
if any(k in dest for k in west_keywords):
westbound.append((dep_time, line, dest, mins_str))
westbound.append((line, dest, mins_str))
elif any(k in dest for k in east_keywords):
eastbound.append((dep_time, line, dest, mins_str))
eastbound.append((line, dest, mins_str))
# Sortieren
westbound.sort(key=lambda x: x[0])
eastbound.sort(key=lambda x: x[0])
westbound = sorted(westbound, key=lambda x: x[2])[:MAX_PER_DIRECTION]
eastbound = sorted(eastbound, key=lambda x: x[2])[:MAX_PER_DIRECTION]
# 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))
return now, westbound, eastbound
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.
"""
# -------------------
# Terminalmodus
# -------------------
def run_terminal_mode():
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))
departures = fetch_departures(START_ID)
now, west, east = build_table_data(departures)
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(poll_interval)
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)
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(description="RMV Live-Abfahrtstafel für Buchrainplatz")
parser.add_argument("--debug", action="store_true", help="Zeige API-JSON-Ausgabe zur Fehlersuche")
parser = argparse.ArgumentParser()
parser.add_argument("--light", action="store_true", help="Terminalmodus")
args = parser.parse_args()
main(debug=args.debug)
if args.light:
run_terminal_mode()
else:
RMVApp().run()