Paperwall/screens/departures.py

"""Departure board via the free, unofficial db-vendo-client REST API
(v6.db.transport.rest) -- no API key needed, wraps Deutsche Bahn's HAFAS
backend the same way a lot of hobby projects/Home Assistant integrations do.

Not train-only -- HAFAS covers regional/long-distance trains, S-Bahn,
U-Bahn/subway, trams, and buses in one system, so a bus stop works exactly
the same way as a train station here: search for it by name like any other
stop, nothing in check()/render() cares which mode a given departure is.

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)
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, BLACK)

            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, BLACK,
                )

        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)

    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 "--:--"