Honey
d5757a00df
Open-Meteo's geocoder is built for city-level weather lookups -- it knows
"Münster" but has no idea what "Münster Hauptbahnhof" means, so any
specific station name failed to resolve. Reused it from Weather without
checking it actually fit this use case.
The db_info reference project sidesteps this entirely by requiring the
user to pick existing Home Assistant person/zone entities, which already
carry precise coordinates from HA's own location system -- no equivalent
here, so this needed real geocoding.
Switched geocode() to query the same EFA instances' XML_STOPFINDER_REQUEST
endpoint instead -- the same system that answers the trip request, so
location resolution is guaranteed consistent with what trip planning
understands. Verified live: correctly resolves specific stations ("Münster
Hauptbahnhof" -> "Münster (Westf), Hbf", even the "Muenster" ue-spelling
the user actually typed), plain city names (falls back to the main
station), and district/neighborhood names (e.g. "Hiltrup" -> the actual
Hiltrup stop, not just the vague suburb entry) -- all via matchQuality
scoring and the API's own isBest flag. Re-verified the full pipeline
end-to-end through the real Flask route afterward.
312 lines
13 KiB
Python
312 lines
13 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)
|
|
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/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
|
|
|
|
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["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) -> 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)
|