Paperwall/screens/navidrome.py

"""Navidrome "now playing" screen -- ported from the InkyPi plugin.

Same Subsonic API logic as before (salted-MD5 auth, filter on state=="playing"
rather than the unreliable minutesAgo, live positionMs for elapsed time), now
rendering directly with Pillow instead of HTML/CSS + headless Chromium.
"""

import hashlib
import logging
import random
import time
from io import BytesIO

import requests
from PIL import Image, ImageDraw, ImageEnhance

from screens.base import Screen
from screens.render_utils import load_font, text_width, truncate_to_width, wrap_to_lines

logger = logging.getLogger(__name__)

SUBSONIC_API_VERSION = "1.16.1"
CLIENT_NAME = "Paperwall"
HTTP_TIMEOUT = 10

REMEMBERED_FIELDS = (
    "id", "title", "artist", "album", "year", "genre",
    "coverArt", "duration", "starred", "track", "username", "playerName",
)


class NavidromeScreen(Screen):
    id = "navidrome"
    display_name = "Navidrome Now Playing"
    poll_interval = 20

    def default_settings(self) -> dict:
        return {
            "url": "",
            "username": "",
            "password": "",
            "filter_user": "",
            "show_progress": False,
            "show_genre": True,
            "show_source": True,
            "boost_cover": True,
            "grayscale_cover": False,
            "_last_track": None,
            "_last_track_at": None,
        }

    # ---------------------------------------------------------------- check

    def check(self, settings: dict):
        url = (settings.get("url") or "").strip().rstrip("/")
        username = (settings.get("username") or "").strip()
        password = settings.get("password") or ""
        if not url or not username or not password:
            return None

        only_user = (settings.get("filter_user") or "").strip()
        try:
            entry = self._fetch_now_playing(url, username, password, only_user)
        except requests.exceptions.RequestException as exc:
            logger.error("Navidrome request failed: %s", exc)
            return None

        if entry is not None:
            settings["_last_track"] = {k: entry.get(k) for k in REMEMBERED_FIELDS}
            settings["_last_track_at"] = time.time()

        return entry

    def _auth_params(self, username: str, password: str) -> dict:
        salt = "%08x" % random.getrandbits(32)
        token = hashlib.md5((password + salt).encode("utf-8")).hexdigest()
        return {"u": username, "t": token, "s": salt, "v": SUBSONIC_API_VERSION, "c": CLIENT_NAME, "f": "json"}

    def _call(self, url: str, username: str, password: str, method: str, params: dict | None = None) -> dict:
        query = self._auth_params(username, password)
        query.update(params or {})
        response = requests.get(f"{url}/rest/{method}", params=query, timeout=HTTP_TIMEOUT)
        response.raise_for_status()
        payload = response.json().get("subsonic-response", {})
        if payload.get("status") != "ok":
            message = payload.get("error", {}).get("message", "unknown error")
            raise RuntimeError(f"Navidrome API error on {method}: {message}")
        return payload

    def _fetch_now_playing(self, url: str, username: str, password: str, only_user: str):
        """Return the currently-playing entry, or None.

        Navidrome's `minutesAgo` counts from when the session *started*, not
        since the last heartbeat, so a track over 5 minutes long would look
        "stale" under a naive staleness filter while still genuinely playing.
        Filtering on `state == "playing"` instead is correct -- Navidrome
        expires stopped sessions server-side already.
        """
        payload = self._call(url, username, password, "getNowPlaying")
        entries = payload.get("nowPlaying", {}).get("entry", []) or []

        if only_user:
            entries = [e for e in entries if e.get("username") == only_user]

        entries = [e for e in entries if e.get("state") == "playing"]
        if not entries:
            return None

        return min(entries, key=lambda e: e.get("minutesAgo") or 0)

    # --------------------------------------------------------------- render

    def render(self, state, settings: dict, resolution: tuple[int, int]) -> Image.Image:
        idle = state is None
        entry = state if not idle else settings.get("_last_track")

        width, height = resolution
        image = Image.new("RGB", (width, height), "white")
        draw = ImageDraw.Draw(image)

        if entry is None:
            self._render_empty(draw, width, height)
            return image

        track = self._build_track(entry, settings, idle)
        cover = None
        if entry.get("coverArt"):
            cover = self._fetch_cover(settings, entry["coverArt"])

        landscape = width > height
        if landscape:
            self._render_landscape(image, draw, track, cover, settings, idle, width, height)
        else:
            self._render_portrait(image, draw, track, cover, settings, idle, width, height)

        return image

    def _build_track(self, entry: dict, settings: dict, idle: bool) -> dict:
        duration = int(entry.get("duration") or 0)
        if idle:
            elapsed = 0
            since = time.time() - float(settings.get("_last_track_at") or time.time())
            minutes_ago = (int(since // 60) // 5) * 5
        else:
            elapsed = int((entry.get("positionMs") or 0) / 1000)
            elapsed = min(elapsed, duration) if duration else elapsed
            minutes_ago = entry.get("minutesAgo") or 0

        return {
            "title": entry.get("title") or "Unknown title",
            "artist": entry.get("artist") or "Unknown artist",
            "album": entry.get("album") or "",
            "year": entry.get("year"),
            "genre": entry.get("genre"),
            "starred": bool(entry.get("starred")),
            "player": entry.get("playerName") or "",
            "listener": entry.get("username") or "",
            "minutes_ago": minutes_ago,
            "duration": duration,
            "duration_text": _format_time(duration),
            "elapsed": elapsed,
            "elapsed_text": _format_time(elapsed),
            "progress_pct": (elapsed / duration) if duration else 0,
        }

    def _fetch_cover(self, settings: dict, cover_art_id: str):
        url = (settings.get("url") or "").strip().rstrip("/")
        username = (settings.get("username") or "").strip()
        password = settings.get("password") or ""
        query = self._auth_params(username, password)
        query.update({"id": cover_art_id, "size": 600})
        try:
            response = requests.get(f"{url}/rest/getCoverArt", params=query, timeout=HTTP_TIMEOUT)
            response.raise_for_status()
            image = Image.open(BytesIO(response.content)).convert("RGB")
        except (requests.exceptions.RequestException, OSError) as exc:
            logger.warning("Could not load cover art %s: %s", cover_art_id, exc)
            return None

        if settings.get("boost_cover", True):
            image = ImageEnhance.Contrast(image).enhance(1.35)
            image = ImageEnhance.Brightness(image).enhance(1.05)

        if settings.get("grayscale_cover", False):
            image = image.convert("L").convert("RGB")

        return image

    # ------------------------------------------------------------- layouts

    def _render_empty(self, draw: ImageDraw.ImageDraw, width: int, height: int):
        font = load_font(size=int(min(width, height) * 0.06))
        msg = "Nichts läuft gerade"
        mw = text_width(draw, msg, font)
        draw.text(((width - mw) / 2, height / 2), msg, font=font, fill=(140, 140, 140))

    def _render_landscape(self, image, draw, track, cover, settings, idle, width, height):
        pad = int(width * 0.025)
        # Cap at a fraction of width too, not just height -- a naive full-height
        # square cover can eat almost the whole canvas on a wide/short panel and
        # leave the text column too narrow to wrap sensibly.
        cover_size = min(height - 2 * pad, int(width * 0.5))
        cover_x, cover_y = pad, pad

        self._paste_cover(image, cover, cover_x, cover_y, cover_size)

        meta_x = cover_x + cover_size + int(width * 0.03)
        meta_w = width - meta_x - pad
        self._draw_meta(draw, track, settings, idle, meta_x, pad, meta_w, height - 2 * pad, min(width, height))

    def _render_portrait(self, image, draw, track, cover, settings, idle, width, height):
        pad = int(width * 0.06)
        cover_size = width - 2 * pad
        cover_x, cover_y = pad, pad

        self._paste_cover(image, cover, cover_x, cover_y, cover_size)

        meta_y = cover_y + cover_size + int(height * 0.03)
        meta_h = height - meta_y - pad
        self._draw_meta(
            draw, track, settings, idle, pad, meta_y, width - 2 * pad, meta_h, min(width, height), center=True
        )

    def _paste_cover(self, image, cover, x, y, size):
        if cover is not None:
            fitted = _cover_fit(cover, size)
            image.paste(fitted, (x, y))
        else:
            draw = ImageDraw.Draw(image)
            draw.rectangle((x, y, x + size, y + size), outline=(60, 60, 60), width=2)
            font = load_font(size=int(size * 0.3))
            note = "♪"
            nw = text_width(draw, note, font)
            draw.text((x + (size - nw) / 2, y + size * 0.35), note, font=font, fill=(150, 150, 150))

    def _draw_meta(self, draw, track, settings, idle, x, y, w, h, scale, center=False):
        status_font = load_font(size=int(scale * 0.038), bold=True)
        title_font = load_font(size=int(scale * 0.075), bold=True)
        artist_font = load_font(size=int(scale * 0.05))
        album_font = load_font(size=int(scale * 0.036))
        source_font = load_font(size=int(scale * 0.03))

        def centered_x(text, font):
            tw = text_width(draw, text, font)
            return x + (w - tw) / 2 if center else x

        cursor = y

        status = "ZULETZT GESPIELT" if idle else "● NOW PLAYING"
        draw.text((centered_x(status, status_font), cursor), status, font=status_font, fill=(100, 100, 100))
        cursor += status_font.size * 1.8

        title = track["title"] + (" ★" if track["starred"] else "")
        for line in wrap_to_lines(draw, title, title_font, w, max_lines=2):
            draw.text((centered_x(line, title_font), cursor), line, font=title_font, fill="black")
            cursor += title_font.size * 1.15

        cursor += title_font.size * 0.2
        artist_line = truncate_to_width(draw, track["artist"], artist_font, w)
        draw.text((centered_x(artist_line, artist_font), cursor), artist_line, font=artist_font, fill=(30, 30, 30))
        cursor += artist_font.size * 1.4

        if track["album"]:
            album_line = track["album"]
            if track["year"]:
                album_line += f" · {track['year']}"
            album_line = truncate_to_width(draw, album_line, album_font, w)
            draw.text((centered_x(album_line, album_font), cursor), album_line, font=album_font, fill=(110, 110, 110))
            cursor += album_font.size * 1.4

        if settings.get("show_genre", True) and track["genre"]:
            genre_line = truncate_to_width(draw, track["genre"].upper(), album_font, w)
            draw.text((centered_x(genre_line, album_font), cursor), genre_line, font=album_font, fill=(150, 150, 150))
            cursor += album_font.size * 1.4

        if settings.get("show_progress", False) and not idle and track["duration"]:
            bar_y = cursor + status_font.size * 0.3
            bar_h = max(3, int(scale * 0.012))
            draw.rectangle((x, bar_y, x + w, bar_y + bar_h), outline=(60, 60, 60))
            fill_w = int(w * min(1.0, track["progress_pct"]))
            if fill_w > 0:
                draw.rectangle((x, bar_y, x + fill_w, bar_y + bar_h), fill=(60, 60, 60))
            cursor = bar_y + bar_h + status_font.size * 0.4
            times = f"{track['elapsed_text']}  /  {track['duration_text']}"
            draw.text((centered_x(times, source_font), cursor), times, font=source_font, fill=(110, 110, 110))
            cursor += source_font.size * 1.6
        elif track["duration"]:
            duration_badge = track["duration_text"]
            draw.text((centered_x(duration_badge, source_font), cursor), duration_badge, font=source_font, fill=(110, 110, 110))
            cursor += source_font.size * 1.6

        if settings.get("show_source", True):
            if idle:
                source_line = f"vor {track['minutes_ago']} min"
            else:
                parts = [p for p in (track["player"], track["listener"]) if p]
                source_line = " · ".join(parts)
            if source_line:
                source_line = truncate_to_width(draw, source_line, source_font, w)
                # Prefer pinning to the bottom of the available area, but never
                # overlap whatever the cursor already drew above it -- long
                # titles/metadata can push past where a fixed bottom position
                # would otherwise land.
                bottom_y = max(cursor + status_font.size * 0.4, y + h - source_font.size * 1.3)
                draw.text((centered_x(source_line, source_font), bottom_y), source_line, font=source_font, fill=(140, 140, 140))


def _cover_fit(image: Image.Image, size: int) -> Image.Image:
    """Center-crop to a square, then resize -- mirrors the CSS object-fit: cover behavior."""
    w, h = image.size
    side = min(w, h)
    left = (w - side) // 2
    top = (h - side) // 2
    cropped = image.crop((left, top, left + side, top + side))
    return cropped.resize((size, size), Image.LANCZOS)


def _format_time(seconds) -> str:
    seconds = int(seconds or 0)
    return f"{seconds // 60}:{seconds % 60:02d}"