Add Departures screen (train departure board)
New screens/departures.py using the free, unofficial db-vendo-client REST
API (v6.db.transport.rest) -- no key needed, same "no API key" philosophy
as Weather's Open-Meteo integration. Community-run infra though, not
official DB, and it was actually down (503, then timing out entirely) for
the whole time I was building this -- couldn't live-verify against real
data the way Open-Meteo was earlier, so I built the parsing against the
project's documented example response and stress-tested render() with
mocked requests.get() covering on-time/delayed/cancelled/missing-platform/
very-long-direction cases instead. check() fails soft (logs a warning,
returns None) on any request error, matching the "service might be down"
reality of this API -- the screen shows a plain "not available" message
rather than erroring out.
Settings: station name (resolved to a station ID via /locations, same
pattern as Weather's city->coordinates geocoding), how many departures to
show (3/5/8), and its own configurable update interval via the
get_poll_interval() mechanism built for Weather (defaults to 2 min --
departure boards go stale much faster than weather does).
Caught two real layout bugs while stress-testing render() before shipping:
line names ("ICE 1234") and delayed/cancelled time text ("Fällt aus") were
both wider than the columns I'd budgeted for them, with the second one
overlapping the direction text when right-aligned past its column's own
left edge. Fixed by measuring actual worst-case widths at the chosen font
size instead of guessing, and truncating every field to its column as a
defensive backstop regardless.
This commit is contained in:
1 parent
889c062212
commit
8c9b3e1bfa
6 files changed
+287
-2
No files matched your search
@@ -1,4 +1,5 @@
|
||||
from screens.clock import ClockScreen
|
||||
from screens.departures import DeparturesScreen
|
||||
from screens.navidrome import NavidromeScreen
|
||||
from screens.picture import PictureScreen
|
||||
from screens.weather import WeatherScreen
|
||||
@@ -6,6 +7,7 @@ from screens.weather import WeatherScreen
|
||||
SCREEN_CLASSES = {
|
||||
ClockScreen.id: ClockScreen,
|
||||
WeatherScreen.id: WeatherScreen,
|
||||
DeparturesScreen.id: DeparturesScreen,
|
||||
PictureScreen.id: PictureScreen,
|
||||
NavidromeScreen.id: NavidromeScreen,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
"""Train departures via the free, unofficial db-vendo-client REST API
|
||||
(v6.db.transport.rest) -- no API key needed, wraps Deutsche Bahn's own
|
||||
backend the same way a lot of hobby projects/Home Assistant integrations do.
|
||||
|
||||
It's community-run on donated infrastructure, not an official DB service --
|
||||
expect occasional downtime. check() degrades to None on any failure (via
|
||||
the same try/except pattern as the Navidrome screen), which renders a
|
||||
clean "not available" state rather than crashing the scheduler tick.
|
||||
"""
|
||||
|
||||
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
|
||||
API_BASE = "https://v6.db.transport.rest"
|
||||
RESULTS_CHOICES = (3, 5, 8)
|
||||
|
||||
BLACK = (0, 0, 0)
|
||||
RED = (200, 0, 0)
|
||||
MUTED = (110, 110, 110)
|
||||
FAINT_LINE = (228, 228, 228)
|
||||
|
||||
|
||||
class DeparturesScreen(Screen):
|
||||
id = "departures"
|
||||
display_name = "Departures"
|
||||
poll_interval = 120 # fallback if update_minutes isn't in settings yet
|
||||
|
||||
def default_settings(self) -> dict:
|
||||
return {
|
||||
"station": "",
|
||||
"station_id": None,
|
||||
"station_label": "",
|
||||
"results": 5,
|
||||
"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 find_station(self, query: str):
|
||||
"""Resolve a station name to {id, name} via /locations. Prefers an
|
||||
actual station/stop over addresses/POIs. Returns None if nothing
|
||||
matched (or the request itself fails -- let that raise, same as
|
||||
WeatherScreen.geocode(), so the settings route can show an error)."""
|
||||
response = requests.get(
|
||||
f"{API_BASE}/locations",
|
||||
params={"query": query, "results": 5, "poi": "false", "addresses": "false"},
|
||||
timeout=HTTP_TIMEOUT,
|
||||
)
|
||||
response.raise_for_status()
|
||||
for entry in response.json() or []:
|
||||
if entry.get("type") in ("station", "stop") and entry.get("id"):
|
||||
return {"id": entry["id"], "name": entry.get("name") or query}
|
||||
return None
|
||||
|
||||
def check(self, settings: dict):
|
||||
station_id = settings.get("station_id")
|
||||
if not station_id:
|
||||
return None
|
||||
|
||||
results = settings.get("results", 5)
|
||||
if results not in RESULTS_CHOICES:
|
||||
results = 5
|
||||
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{API_BASE}/stops/{station_id}/departures",
|
||||
params={"duration": 90, "results": results},
|
||||
timeout=HTTP_TIMEOUT,
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
except (requests.exceptions.RequestException, ValueError) as exc:
|
||||
logger.warning("Departures request failed (service may be down): %s", exc)
|
||||
return None
|
||||
|
||||
# Documented example response is a bare array; be defensive in case
|
||||
# a future version wraps it in {"departures": [...]} instead.
|
||||
raw = payload.get("departures", payload) if isinstance(payload, dict) else payload
|
||||
if not raw:
|
||||
return None
|
||||
|
||||
departures = []
|
||||
for entry in raw[:results]:
|
||||
line = (entry.get("line") or {}).get("name") or "?"
|
||||
planned = entry.get("plannedWhen") or entry.get("when")
|
||||
delay_sec = entry.get("delay")
|
||||
delay_min = int(delay_sec // 60) if isinstance(delay_sec, (int, float)) and delay_sec > 0 else 0
|
||||
departures.append({
|
||||
"line": line,
|
||||
"direction": entry.get("direction") or "",
|
||||
"time": _format_clock(planned),
|
||||
"delay_min": delay_min,
|
||||
"cancelled": bool(entry.get("cancelled")),
|
||||
"platform": entry.get("platform") or entry.get("plannedPlatform") or "",
|
||||
})
|
||||
|
||||
return departures or None
|
||||
|
||||
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)
|
||||
|
||||
if not settings.get("station_id"):
|
||||
self._render_message(draw, width, height, scale, "Keine Station konfiguriert")
|
||||
return image
|
||||
if not state:
|
||||
self._render_message(draw, width, height, scale, "Keine Abfahrten verfügbar")
|
||||
return image
|
||||
|
||||
pad = width * 0.04
|
||||
header_font = load_font(size=int(scale * 0.055), bold=True)
|
||||
station_label = settings.get("station_label") or settings.get("station") or ""
|
||||
draw.text((pad, height * 0.045), station_label, 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)
|
||||
|
||||
line_font = load_font(size=int(scale * 0.05), bold=True)
|
||||
dir_font = load_font(size=int(scale * 0.042))
|
||||
time_font = load_font(size=int(scale * 0.05), bold=True)
|
||||
platform_font = load_font(size=int(scale * 0.032))
|
||||
|
||||
# Sized against actually-measured worst cases at this font size:
|
||||
# "ICE 1234" ~98px, "21:09 +12" ~115px, "Fällt aus" ~96px. Every
|
||||
# field is still truncated to its column as a safety net for
|
||||
# anything longer, but these widths mean that won't normally fire.
|
||||
line_w = scale * 0.275
|
||||
time_w = scale * 0.325
|
||||
platform_w = scale * 0.1
|
||||
dir_x = pad + line_w + scale * 0.015
|
||||
dir_w = max(20, (width - pad * 2) - line_w - time_w - platform_w - scale * 0.015)
|
||||
time_x = dir_x + dir_w
|
||||
platform_x = time_x + time_w
|
||||
|
||||
for i, dep in enumerate(state):
|
||||
row_top_y = row_top + i * row_h
|
||||
row_cy = row_top_y + row_h / 2
|
||||
if i > 0:
|
||||
draw.line((pad, row_top_y, width - pad, row_top_y), fill=FAINT_LINE, width=1)
|
||||
|
||||
line_text = truncate_to_width(draw, dep["line"], line_font, line_w - 6)
|
||||
self._draw_centered_v(draw, line_text, line_font, pad, row_cy, BLACK)
|
||||
|
||||
direction = truncate_to_width(draw, dep["direction"], dir_font, dir_w - 10)
|
||||
self._draw_centered_v(draw, direction, dir_font, dir_x, row_cy, (60, 60, 60))
|
||||
|
||||
if dep["cancelled"]:
|
||||
time_text, time_color = "Fällt aus", RED
|
||||
elif dep["delay_min"] > 0:
|
||||
time_text, time_color = f"{dep['time']} +{dep['delay_min']}", RED
|
||||
else:
|
||||
time_text, time_color = dep["time"], BLACK
|
||||
time_text = truncate_to_width(draw, time_text, time_font, time_w - 12)
|
||||
tw = text_width(draw, time_text, time_font)
|
||||
self._draw_centered_v(draw, time_text, time_font, time_x + time_w - tw - 8, row_cy, time_color)
|
||||
|
||||
if dep["platform"]:
|
||||
platform_text = truncate_to_width(draw, dep["platform"], platform_font, platform_w - 4)
|
||||
pw = text_width(draw, platform_text, platform_font)
|
||||
self._draw_centered_v(
|
||||
draw, platform_text, platform_font,
|
||||
platform_x + (platform_w - pw) / 2, row_cy, 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)
|
||||
|
||||
def _draw_centered_v(self, draw, text, font, x, cy, fill):
|
||||
left, top, right, bottom = draw.textbbox((0, 0), text, font=font)
|
||||
draw.text((x, cy - (bottom - top) / 2 - top), text, font=font, fill=fill)
|
||||
|
||||
|
||||
def _format_clock(iso_str) -> str:
|
||||
if not iso_str:
|
||||
return "--:--"
|
||||
try:
|
||||
return datetime.fromisoformat(iso_str).strftime("%H:%M")
|
||||
except (TypeError, ValueError):
|
||||
return "--:--"
|
||||
Reference in new issue
Block a user