Honey
684d706afc
Not needed any more -- the Journey screen already covers the same regional-transit need via EFA backends, and Departures depended on the free db-vendo-client API that DB has been actively blocking third parties from anyway. Removed screens/departures.py, its settings template, its route, its registration in screens/__init__.py, and its nav link. Reworded the docstring/hint references in journey.py and settings_journey.html that pointed at the now-gone screen. Existing config.json entries referencing "departures" (rotation list, pinned screen) are harmless leftovers -- both scheduler lookups already filter to screens that still exist, so nothing needs a migration step. Verified: app boots clean, /settings/departures 404s, every other settings page still loads fine, no leftover references outside of journey.py's own unrelated "departure time" wording.
423 lines
19 KiB
Python
423 lines
19 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 --
|
|
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)
|