Honey
06b90ffe6c
- connection_type setting (all/regional/long_distance) restricts EFA trip requests via inclMOT_N params, so ICE/IC/EC can be filtered out for commutes that only care about regional/local transit. - Surface leg.infos[].infoLinks titles (real disruption/outage notices from EFA) on each journey row, marked with a red warning icon.
356 lines
15 KiB
Python
356 lines
15 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)
|
|
|
|
# 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
|
|
legs.append({
|
|
"line": transportation.get("name") or "?",
|
|
"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 "--:--",
|
|
"lines": [leg["line"] 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 "?"
|
|
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']}")
|
|
if journey["notice"]:
|
|
detail_parts.append(journey["notice"])
|
|
|
|
detail_y = top_line_y + time_font.size * 1.25
|
|
detail_x = pad
|
|
if journey["notice"]:
|
|
draw.text((detail_x, detail_y), "⚠", font=detail_font, fill=RED)
|
|
detail_x += text_width(draw, "⚠", detail_font) + scale * 0.015
|
|
|
|
detail_line = truncate_to_width(draw, " · ".join(detail_parts), detail_font, width - pad - detail_x)
|
|
draw.text((detail_x, detail_y), 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)
|