From 0d539cd4700236340ac13248158d2ae6487b28e9 Mon Sep 17 00:00:00 2001 From: Marcel Weschke Date: Thu, 14 Aug 2025 16:34:05 +0200 Subject: [PATCH] updated functionality, --light (terminal mode) or Textual mode (as default) --- RMV_Digital_Board_light.py | 206 ++++++++++++++++++++++--------------- 1 file changed, 121 insertions(+), 85 deletions(-) diff --git a/RMV_Digital_Board_light.py b/RMV_Digital_Board_light.py index 271c41d..29ae31c 100644 --- a/RMV_Digital_Board_light.py +++ b/RMV_Digital_Board_light.py @@ -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()