"""Next connections for a fixed daily commute (origin -> destination), via regional EFA (Elektronische Fahrplanauskunft) instances -- transit-planner backends run by regional transit authorities, not Deutsche Bahn itself. This deliberately avoids bahn.de's own internal trip-planner API. DB has been actively blocking third-party access to its own backends recently (see the Departures screen's docstring and the outage that motivated it) -- EFA instances aren't DB's infrastructure, so they aren't part of that clampdown. Several are tried in sequence (not parallel, to keep this synchronous like the rest of the app); first one to return usable journeys wins. Ported and simplified from github.com/EiS94/db_info (a Home Assistant integration), which queries these same EFA instances plus bahn.de's API in parallel and scores the results -- dropped the bahn.de source and the parallel/scoring machinery for a simpler, more DB-independent first version. """ 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 GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search" EFA_SOURCES = [ { "name": "vrr", "url": "https://www.vrr.de/vrr-efa/XML_TRIP_REQUEST2", # vrr.de hangs until timeout for trips outside its home region # instead of returning a fast error, so it's only queried when both # ends of the trip roughly fall within North Rhine-Westphalia. "bounds": {"lat_min": 50.3, "lat_max": 52.6, "lon_min": 5.7, "lon_max": 9.5}, }, {"name": "bahnland-bayern", "url": "https://bahnland-bayern.de/efa/XML_TRIP_REQUEST2", "bounds": None}, {"name": "efa", "url": "https://www.efa.de/hit-efa/XML_TRIP_REQUEST2", "bounds": None}, {"name": "efa-bw", "url": "http://www.efa-bw.de/nvbw/XML_TRIP_REQUEST2", "bounds": None}, ] # Some EFA instances reject the default requests/python User-Agent as bot # traffic and answer with a 403. REQUEST_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Accept": "application/json", } RED = (200, 0, 0) BLACK = (0, 0, 0) MUTED = (110, 110, 110) def _in_bounds(lat, lon, bounds) -> bool: if bounds is None: return True return bounds["lat_min"] <= lat <= bounds["lat_max"] and bounds["lon_min"] <= lon <= bounds["lon_max"] def _parse_time(iso_str): if not iso_str: return None try: return datetime.fromisoformat(iso_str) except ValueError: return None class JourneyScreen(Screen): id = "journey" display_name = "Journey" poll_interval = 120 # fallback if update_minutes isn't in settings yet def default_settings(self) -> dict: return { "origin_query": "", "origin_label": "", "origin_lat": None, "origin_lon": None, "destination_query": "", "destination_label": "", "destination_lat": None, "destination_lon": None, "results": 3, "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 geocode(self, query: str): """Resolve a place name to coordinates -- same free Open-Meteo geocoding WeatherScreen uses, no key needed.""" response = requests.get( GEOCODE_URL, params={"name": query, "count": 1, "language": "de"}, timeout=HTTP_TIMEOUT ) response.raise_for_status() results = response.json().get("results") or [] if not results: return None top = results[0] label_parts = [top.get("name"), top.get("admin1"), top.get("country")] return { "latitude": top["latitude"], "longitude": top["longitude"], "label": ", ".join(p for p in label_parts if p), } def check(self, settings: dict): o_lat, o_lon = settings.get("origin_lat"), settings.get("origin_lon") d_lat, d_lon = settings.get("destination_lat"), settings.get("destination_lon") if o_lat is None or o_lon is None or d_lat is None or d_lon is None: return None try: results = max(1, min(5, int(settings.get("results") or 3))) except (TypeError, ValueError): results = 3 params = self._build_params(o_lat, o_lon, d_lat, d_lon, results) for source in EFA_SOURCES: if not (_in_bounds(o_lat, o_lon, source["bounds"]) and _in_bounds(d_lat, d_lon, source["bounds"])): continue try: response = requests.get( source["url"], params=params, headers=REQUEST_HEADERS, timeout=HTTP_TIMEOUT ) response.raise_for_status() data = response.json() except (requests.exceptions.RequestException, ValueError) as exc: logger.warning("Journey source '%s' failed, trying next: %s", source["name"], exc) continue journeys = self._parse_journeys(data, results) if journeys: logger.debug("Journey data from '%s'", source["name"]) return journeys return None def _build_params(self, o_lat, o_lon, d_lat, d_lon, results) -> dict: now = datetime.now().astimezone() return { "outputFormat": "rapidJSON", "language": "de", "type_origin": "coord", "name_origin": f"{o_lon}:{o_lat}:WGS84[dd.ddddd]", "type_destination": "coord", "name_destination": f"{d_lon}:{d_lat}:WGS84[dd.ddddd]", "itdDateDayMonthYear": now.strftime("%d.%m.%Y"), "itdTime": now.strftime("%H:%M"), "itdTripDateTimeDepArr": "dep", "coordOutputFormat": "WGS84[dd.ddddd]", "useRealtime": "1", "calcOneDirection": "1", "calcNumberOfTrips": str(results), "ptOptionsActive": "1", "lineRestriction": "400", "serverInfo": "0", } def _parse_journeys(self, data: dict, results: int): journeys = [] for raw_journey in (data.get("journeys") or [])[:results]: legs = [] for leg in raw_journey.get("legs", []): transportation = leg.get("transportation") or {} product = transportation.get("product") or {} is_walk = ( product.get("class") == 100 or product.get("name") == "footpath" or not transportation.get("name") ) if is_walk: continue legs.append({ "line": transportation.get("name") or "?", "origin": leg.get("origin") or {}, "destination": leg.get("destination") or {}, }) if not legs: continue first, last = legs[0], legs[-1] dep_planned = _parse_time(first["origin"].get("departureTimePlanned")) dep_estimated = _parse_time(first["origin"].get("departureTimeEstimated")) arr_planned = _parse_time(last["destination"].get("arrivalTimePlanned")) arr_estimated = _parse_time(last["destination"].get("arrivalTimeEstimated")) if dep_planned is None: continue dep_actual = dep_estimated or dep_planned delay_min = int((dep_estimated - dep_planned).total_seconds() // 60) if dep_estimated else 0 arr_actual = arr_estimated or arr_planned # EFA returns UTC timestamps (Z suffix) -- astimezone() with no # args converts to the system's local timezone before display. # Without this, times showed 2h behind (CEST is UTC+2), making # "next" departures look like they'd already left. journeys.append({ "departure": dep_actual.astimezone().strftime("%H:%M"), "delay_min": max(0, delay_min), "arrival": arr_actual.astimezone().strftime("%H:%M") if arr_actual else "--:--", "lines": [leg["line"] for leg in legs], "transfers": len(legs) - 1, "platform": (first["origin"].get("properties") or {}).get("platformName") or "", }) return journeys or None # --------------------------------------------------------------- render 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) have_route = settings.get("origin_lat") is not None and settings.get("destination_lat") is not None if not have_route: self._render_message(draw, width, height, scale, "Keine Strecke konfiguriert") return image if not state: self._render_message(draw, width, height, scale, "Keine Verbindungen verfügbar") return image pad = width * 0.04 origin = settings.get("origin_label") or settings.get("origin_query") or "?" destination = settings.get("destination_label") or settings.get("destination_query") or "?" header_font = load_font(size=int(scale * 0.05), bold=True) header = truncate_to_width(draw, f"{origin} → {destination}", header_font, width - 2 * pad) draw.text((pad, height * 0.045), header, 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) time_font = load_font(size=int(scale * 0.07), bold=True) arrow_font = load_font(size=int(scale * 0.05)) detail_font = load_font(size=int(scale * 0.038)) for i, journey in enumerate(state): row_top_y = row_top + i * row_h if i > 0: draw.line((pad, row_top_y, width - pad, row_top_y), fill=(228, 228, 228), width=1) top_line_y = row_top_y + row_h * 0.14 dep_color = RED if journey["delay_min"] > 0 else BLACK dep_text = journey["departure"] if journey["delay_min"] > 0: dep_text += f" +{journey['delay_min']}" draw.text((pad, top_line_y), dep_text, font=time_font, fill=dep_color) dep_w = text_width(draw, dep_text, time_font) arrow_x = pad + dep_w + scale * 0.02 draw.text((arrow_x, top_line_y + (time_font.size - arrow_font.size) / 2), "→", font=arrow_font, fill=MUTED) arrow_w = text_width(draw, "→", arrow_font) draw.text((arrow_x + arrow_w + scale * 0.02, top_line_y), journey["arrival"], font=time_font, fill=BLACK) detail_parts = [" → ".join(journey["lines"])] if journey["transfers"] > 0: detail_parts.append(f"{journey['transfers']}x umsteigen") if journey["platform"]: detail_parts.append(f"Gl. {journey['platform']}") detail_line = truncate_to_width(draw, " · ".join(detail_parts), detail_font, width - 2 * pad) draw.text((pad, top_line_y + time_font.size * 1.25), detail_line, font=detail_font, fill=MUTED) 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=MUTED)