"""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 -- 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 ICON_BUS, ICON_TRAIN, load_font, load_icon_font, text_width, truncate_to_width logger = logging.getLogger(__name__) HTTP_TIMEOUT = 10 # Each EFA instance exposes both endpoints under the same base path -- # XML_TRIP_REQUEST2 for routing, XML_STOPFINDER_REQUEST for resolving a # place name to a stop + coordinates. Using the stopfinder from the same # system that will answer the trip request (rather than a generic # city-level geocoder) is what makes searching for e.g. "Münster # Hauptbahnhof" -- a specific station, not just a city -- work at all. EFA_SOURCES = [ { "name": "vrr", "base": "https://www.vrr.de/vrr-efa/", # 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", "base": "https://bahnland-bayern.de/efa/", "bounds": None}, {"name": "efa", "base": "https://www.efa.de/hit-efa/", "bounds": None}, {"name": "efa-bw", "base": "http://www.efa-bw.de/nvbw/", "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) # Pure panel colors (from the epd4in0e's 6-color quantization palette in # display/waveshare_epd/epd4in0e.py) -- used for the train/bus mode dots so # they render as solid color on the real hardware instead of dithering like # an off-palette RGB value would. BLUE = (0, 0, 255) GREEN = (0, 255, 0) # EFA's inclMOT_N flags select which transport modes a trip request is # allowed to use -- N is an index into EFA's internal mode table, not the # "class" value seen in responses. Long-distance rail (IC/EC/ICE) sits at # 14-16; everything else usable for a regional/local commute (regional # trains, S-Bahn, buses, trams, etc.) is the rest. Ported from db_info's # _build_mot_params()/_build_produktgattungen(). _MOT_LONG_DISTANCE = {f"inclMOT_{i}": "true" for i in (14, 15, 16)} _MOT_REGIONAL = {f"inclMOT_{i}": "true" for i in (1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 17, 20)} def _mot_params(connection_type: str) -> dict: if connection_type == "regional": return {"includedMeans": "checkbox", **_MOT_REGIONAL} if connection_type == "long_distance": return {"includedMeans": "checkbox", **_MOT_LONG_DISTANCE} return {} 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, "connection_type": "all", } 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/station name to coordinates via the same EFA instances used for trip requests (XML_STOPFINDER_REQUEST) -- not a generic city-level geocoder, which has no idea what "Hauptbahnhof" means. Tries each source in order (no bounds pre-filter here, since we don't have coordinates yet to check bounds against); returns the first usable match.""" params = { "outputFormat": "rapidJSON", "language": "de", "type_sf": "any", "name_sf": query, "anyMaxSizeHitList": 5, "coordOutputFormat": "WGS84[dd.ddddd]", } for source in EFA_SOURCES: try: response = requests.get( source["base"] + "XML_STOPFINDER_REQUEST", params=params, headers=REQUEST_HEADERS, timeout=HTTP_TIMEOUT, ) response.raise_for_status() data = response.json() except (requests.exceptions.RequestException, ValueError) as exc: logger.warning("Stopfinder source '%s' failed, trying next: %s", source["name"], exc) continue match = self._best_location(data) if match: return match return None def _best_location(self, data: dict): locations = data.get("locations") or [] if not locations: return None best = next((loc for loc in locations if loc.get("isBest")), locations[0]) coord = best.get("coord") if not coord or len(coord) < 2: return None return {"latitude": coord[0], "longitude": coord[1], "label": best.get("name") or ""} 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 connection_type = settings.get("connection_type") or "all" params = self._build_params(o_lat, o_lon, d_lat, d_lon, results, connection_type) 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["base"] + "XML_TRIP_REQUEST2", 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, connection_type: str = "all") -> dict: now = datetime.now().astimezone() params = { "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", } params.update(_mot_params(connection_type)) return params 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 # EFA's product.name is a reliable human-readable mode string # ("StadtBus", "RegioBus", "SchnellBus", "Niederflurbus" for # buses; "S-Bahn", "Regionalzug", "Zug" for trains) -- more # robust than product.class, whose numeric values aren't a # clean bus/train split (bus classes 5-7 sit right next to # S-Bahn's class 1). Confirmed live against VRR. is_bus = "bus" in (product.get("name") or "").lower() legs.append({ "line": transportation.get("name") or "?", "mode": "bus" if is_bus else "train", "origin": leg.get("origin") or {}, "destination": leg.get("destination") or {}, "infos": leg.get("infos") 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 # Service notices (disruptions, elevator outages, etc.) come as # leg.infos[].infoLinks[].title -- collect titles across all legs # of this journey, deduped, since the same notice often repeats # on every leg it affects. notice_titles = [] for leg in legs: for info in leg["infos"]: for link in info.get("infoLinks") or []: title = link.get("title") or link.get("subtitle") if title and title not in notice_titles: notice_titles.append(title) # Layover between consecutive legs: next leg's (estimated, # falling back to planned) departure minus this leg's (likewise) # arrival. None when either side is missing a timestamp, so the # render step can fall back to a plain "umsteigen" label. transfer_minutes = [] for cur, nxt in zip(legs, legs[1:]): arr = _parse_time(cur["destination"].get("arrivalTimeEstimated")) or _parse_time(cur["destination"].get("arrivalTimePlanned")) dep = _parse_time(nxt["origin"].get("departureTimeEstimated")) or _parse_time(nxt["origin"].get("departureTimePlanned")) transfer_minutes.append(max(0, int((dep - arr).total_seconds() // 60)) if arr and dep else None) # 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 "--:--", "legs": [{"name": leg["line"], "mode": leg["mode"]} for leg in legs], "transfers": len(legs) - 1, "transfer_minutes": transfer_minutes, "platform": (first["origin"].get("properties") or {}).get("platformName") or "", "notice": "; ".join(notice_titles), }) 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 "?" # Origin/destination stacked on their own lines rather than one # "A -> B" line -- on the small panel a single combined line # truncated too early to show the destination at all. header_font = load_font(size=int(scale * 0.045), bold=True) origin_line = truncate_to_width(draw, origin, header_font, width - 2 * pad) dest_line = truncate_to_width(draw, f"→ {destination}", header_font, width - 2 * pad) line_gap = header_font.size * 1.22 header_top = height * 0.035 draw.text((pad, header_top), origin_line, font=header_font, fill=BLACK) draw.text((pad, header_top + line_gap), dest_line, font=header_font, fill=BLACK) header_bottom = header_top + line_gap * 2 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=BLACK) 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 line: a real train/bus pictogram per leg (Font Awesome # Solid, vendored in fonts/fa-solid-900.ttf -- confirmed live to # stay recognizable even at this render size, unlike a hand-drawn # shape or emoji, which either look unclear that small or render # as tofu since DejaVu Sans has no color-emoji glyphs). Blue for # train, green for bus -- pure panel colors, no dithering. # Followed by the line name, then transfer/platform/notice info. # Drawn as manually-placed segments instead of one joined string # so each leg can carry its own icon. detail_y = top_line_y + time_font.size * 1.25 right_edge = width - pad icon_d = detail_font.size * 1.05 icon_font = load_icon_font(size=int(icon_d)) cursor_x = pad def draw_segment(text, color=BLACK, avail=None): nonlocal cursor_x if avail is not None: text = truncate_to_width(draw, text, detail_font, avail) draw.text((cursor_x, detail_y), text, font=detail_font, fill=color) cursor_x += text_width(draw, text, detail_font) return text def draw_mode_icon(x, mode): color = BLUE if mode == "train" else GREEN glyph = ICON_TRAIN if mode == "train" else ICON_BUS l, t, r, b = draw.textbbox((0, 0), glyph, font=icon_font) gw, gh = r - l, b - t top = detail_y + (detail_font.size - icon_d) / 2 draw.text((x - l, top + (icon_d - gh) / 2 - t), glyph, font=icon_font, fill=color) for idx, leg in enumerate(journey["legs"]): if cursor_x >= right_edge - icon_d: break draw_mode_icon(cursor_x, leg["mode"]) cursor_x += icon_d + scale * 0.01 shown = draw_segment(leg["name"], avail=right_edge - cursor_x) if idx < len(journey["legs"]) - 1 and cursor_x < right_edge and shown == leg["name"]: gap = journey["transfer_minutes"][idx] sep = f" →{gap}min→ " if gap is not None else " → umsteigen → " draw_segment(sep, avail=right_edge - cursor_x) trailing_parts = [] if journey["platform"]: trailing_parts.append(f"Gl. {journey['platform']}") if trailing_parts and cursor_x < right_edge: draw_segment(" · " + " · ".join(trailing_parts), avail=right_edge - cursor_x) if journey["notice"] and cursor_x < right_edge - icon_d: cursor_x += scale * 0.01 draw_segment("⚠ ", color=RED) draw_segment(journey["notice"], avail=right_edge - cursor_x) 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)