Honey
080cb11501
- Origin/destination now on their own lines instead of one truncated "A -> B" line, which cut off the destination entirely on the small panel. - Each leg gets a colored dot (blue = train, green = bus), using the panel's exact palette colors so they render solid, not dithered. Mode is read from EFA's product.name (confirmed live: "StadtBus"/ "RegioBus"/"SchnellBus" vs "S-Bahn"/"Regionalzug"/"Zug"). - Detail-line text switched from dithered gray to solid black -- the gray was hard to read at the small font size the panel renders at.
401 lines
18 KiB
Python
401 lines
18 KiB
Python
"""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
|
|
|
|
# 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)
|
|
|
|
# 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,
|
|
"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 colored dot per leg (blue = train, green = bus --
|
|
# pure panel colors, no dithering) followed by its line name, then
|
|
# transfer/platform/notice info. Drawn as manually-placed segments
|
|
# instead of one joined string so each leg can carry its own dot.
|
|
detail_y = top_line_y + time_font.size * 1.25
|
|
right_edge = width - pad
|
|
icon_d = detail_font.size * 0.55
|
|
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
|
|
|
|
for idx, leg in enumerate(journey["legs"]):
|
|
if cursor_x >= right_edge - icon_d:
|
|
break
|
|
dot_color = BLUE if leg["mode"] == "train" else GREEN
|
|
dot_top = detail_y + (detail_font.size - icon_d) / 2
|
|
draw.ellipse((cursor_x, dot_top, cursor_x + icon_d, dot_top + icon_d), fill=dot_color)
|
|
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"]:
|
|
draw_segment(" → ", avail=right_edge - cursor_x)
|
|
|
|
trailing_parts = []
|
|
if journey["transfers"] > 0:
|
|
trailing_parts.append(f"{journey['transfers']}x umsteigen")
|
|
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)
|