"""Departure board via the free, unofficial db-vendo-client REST API (v6.db.transport.rest) -- no API key needed, wraps Deutsche Bahn's HAFAS backend the same way a lot of hobby projects/Home Assistant integrations do. Not train-only -- HAFAS covers regional/long-distance trains, S-Bahn, U-Bahn/subway, trams, and buses in one system, so a bus stop works exactly the same way as a train station here: search for it by name like any other stop, nothing in check()/render() cares which mode a given departure is. It's community-run on donated infrastructure, not an official DB service -- expect occasional downtime. check() degrades to None on any failure (via the same try/except pattern as the Navidrome screen), which renders a clean "not available" state rather than crashing the scheduler tick. """ import logging from datetime import datetime import requests from PIL import Image, ImageDraw from screens.base import Screen from screens.render_utils import load_font, text_width, truncate_to_width logger = logging.getLogger(__name__) HTTP_TIMEOUT = 10 API_BASE = "https://v6.db.transport.rest" RESULTS_CHOICES = (3, 5, 8) BLACK = (0, 0, 0) RED = (200, 0, 0) FAINT_LINE = (228, 228, 228) class DeparturesScreen(Screen): id = "departures" display_name = "Departures" poll_interval = 120 # fallback if update_minutes isn't in settings yet def default_settings(self) -> dict: return { "station": "", "station_id": None, "station_label": "", "results": 5, "update_minutes": 2, } def get_poll_interval(self, settings: dict) -> int: raw = settings.get("update_minutes") try: minutes = int(raw) if raw is not None else 2 except (TypeError, ValueError): minutes = 2 return max(1, minutes) * 60 def find_station(self, query: str): """Resolve a station name to {id, name} via /locations. Prefers an actual station/stop over addresses/POIs. Returns None if nothing matched (or the request itself fails -- let that raise, same as WeatherScreen.geocode(), so the settings route can show an error).""" response = requests.get( f"{API_BASE}/locations", params={"query": query, "results": 5, "poi": "false", "addresses": "false"}, timeout=HTTP_TIMEOUT, ) response.raise_for_status() for entry in response.json() or []: if entry.get("type") in ("station", "stop") and entry.get("id"): return {"id": entry["id"], "name": entry.get("name") or query} return None def check(self, settings: dict): station_id = settings.get("station_id") if not station_id: return None results = settings.get("results", 5) if results not in RESULTS_CHOICES: results = 5 try: response = requests.get( f"{API_BASE}/stops/{station_id}/departures", params={"duration": 90, "results": results}, timeout=HTTP_TIMEOUT, ) response.raise_for_status() payload = response.json() except (requests.exceptions.RequestException, ValueError) as exc: logger.warning("Departures request failed (service may be down): %s", exc) return None # Documented example response is a bare array; be defensive in case # a future version wraps it in {"departures": [...]} instead. raw = payload.get("departures", payload) if isinstance(payload, dict) else payload if not raw: return None departures = [] for entry in raw[:results]: line = (entry.get("line") or {}).get("name") or "?" planned = entry.get("plannedWhen") or entry.get("when") delay_sec = entry.get("delay") delay_min = int(delay_sec // 60) if isinstance(delay_sec, (int, float)) and delay_sec > 0 else 0 departures.append({ "line": line, "direction": entry.get("direction") or "", "time": _format_clock(planned), "delay_min": delay_min, "cancelled": bool(entry.get("cancelled")), "platform": entry.get("platform") or entry.get("plannedPlatform") or "", }) return departures or None def render(self, state, settings: dict, resolution: tuple[int, int]) -> Image.Image: width, height = resolution image = Image.new("RGB", (width, height), "white") draw = ImageDraw.Draw(image) scale = min(width, height) if not settings.get("station_id"): self._render_message(draw, width, height, scale, "Keine Station konfiguriert") return image if not state: self._render_message(draw, width, height, scale, "Keine Abfahrten verfügbar") return image pad = width * 0.04 header_font = load_font(size=int(scale * 0.055), bold=True) station_label = settings.get("station_label") or settings.get("station") or "" draw.text((pad, height * 0.045), station_label, font=header_font, fill=BLACK) header_bottom = height * 0.045 + header_font.size * 1.35 draw.rectangle((pad, header_bottom, width - pad, header_bottom + 2), fill=BLACK) row_top = header_bottom + height * 0.03 row_h = (height - row_top - height * 0.03) / len(state) line_font = load_font(size=int(scale * 0.05), bold=True) dir_font = load_font(size=int(scale * 0.042)) time_font = load_font(size=int(scale * 0.05), bold=True) platform_font = load_font(size=int(scale * 0.032)) # Sized against actually-measured worst cases at this font size: # "ICE 1234" ~98px, "21:09 +12" ~115px, "Fällt aus" ~96px. Every # field is still truncated to its column as a safety net for # anything longer, but these widths mean that won't normally fire. line_w = scale * 0.275 time_w = scale * 0.325 platform_w = scale * 0.1 dir_x = pad + line_w + scale * 0.015 dir_w = max(20, (width - pad * 2) - line_w - time_w - platform_w - scale * 0.015) time_x = dir_x + dir_w platform_x = time_x + time_w for i, dep in enumerate(state): row_top_y = row_top + i * row_h row_cy = row_top_y + row_h / 2 if i > 0: draw.line((pad, row_top_y, width - pad, row_top_y), fill=FAINT_LINE, width=1) line_text = truncate_to_width(draw, dep["line"], line_font, line_w - 6) self._draw_centered_v(draw, line_text, line_font, pad, row_cy, BLACK) direction = truncate_to_width(draw, dep["direction"], dir_font, dir_w - 10) self._draw_centered_v(draw, direction, dir_font, dir_x, row_cy, BLACK) if dep["cancelled"]: time_text, time_color = "Fällt aus", RED elif dep["delay_min"] > 0: time_text, time_color = f"{dep['time']} +{dep['delay_min']}", RED else: time_text, time_color = dep["time"], BLACK time_text = truncate_to_width(draw, time_text, time_font, time_w - 12) tw = text_width(draw, time_text, time_font) self._draw_centered_v(draw, time_text, time_font, time_x + time_w - tw - 8, row_cy, time_color) if dep["platform"]: platform_text = truncate_to_width(draw, dep["platform"], platform_font, platform_w - 4) pw = text_width(draw, platform_text, platform_font) self._draw_centered_v( draw, platform_text, platform_font, platform_x + (platform_w - pw) / 2, row_cy, BLACK, ) return image def _render_message(self, draw, width, height, scale, msg): font = load_font(size=int(scale * 0.06)) mw = text_width(draw, msg, font) draw.text(((width - mw) / 2, height / 2), msg, font=font, fill=BLACK) def _draw_centered_v(self, draw, text, font, x, cy, fill): left, top, right, bottom = draw.textbbox((0, 0), text, font=font) draw.text((x, cy - (bottom - top) / 2 - top), text, font=font, fill=fill) def _format_clock(iso_str) -> str: if not iso_str: return "--:--" try: return datetime.fromisoformat(iso_str).strftime("%H:%M") except (TypeError, ValueError): return "--:--"