"""
E-ink device client (Track B) — the poll-and-sleep bridge that pulls the current image from
`GET /display/{id}/current.{ext}` and blits it to a Pimoroni Inky Impression 13.3" Spectra 6 panel.

Runs HOST-SIDE (not in the container) via `deploy/appliance/bin/sd-eink` — GPIO/SPI aren't reachable
from the non-root app container (ADR-037), and this client must also run standalone in "satellite" mode
with no local Pieria container at all. So this module is deliberately standalone: only stdlib +
Pillow + httpx + (lazily) `inky`. No `database`/app imports — see `.ai/spec_eink_spectra6.md` §3.2/3.4.

Design note (no hardware here): mirrors `frame_push.py`'s shape — an `EinkClient` ABC + a real
`InkyClient` + a `FakeInkyClient` + a testable single-tick (`push_once` / `run_tick`) + a loop
(`run_loop`). Same "test the platform without hardware" posture as the Frame TV integration
(memory `no-frame-tv-for-testing`), applied to e-ink.
"""

from __future__ import annotations

import hashlib
import io
import logging
import os
import re
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Optional, Tuple

import httpx
from PIL import Image

logger = logging.getLogger("sd-eink")

DEFAULT_HEARTBEAT_FILE = "/run/sd-eink.state"

# fetch_fn(url) -> (content_bytes, headers_dict). Injected so push_once never calls httpx directly.
#: First retry delay after a failed tick, in seconds; doubles per consecutive failure up to
#: EINK_MIN_INTERVAL. Small enough that a device which outran its own server recovers in seconds.
ERROR_BACKOFF_BASE = 15

FetchFn = Callable[[str], Tuple[bytes, dict]]
# schedule_fn() -> schedule-state JSON dict (has a "quiet" bool).
ScheduleFn = Callable[[], dict]


# --------------------------------------------------------------------------- config

def split_conf_value(rest: str):
    """Split the text after `KEY=` into (value, trailing), matching what bash `.`-sourcing yields for
    the forms pieria.conf actually uses (ADR-137 conf comments). Only space/tab separate words, as in bash:

      KEY=value   # note   -> value  (an unquoted `#` starts a comment only when space/tab precedes it)
      KEY="a # b"  # note  -> a # b  (quotes protect a `#`)
      KEY='a # b'          -> a # b
      KEY="a"b             -> ab     (text glued to a closing quote joins the word, so KEY="90"#x -> 90#x)
      KEY=   # note        -> ""

    `trailing` is the raw suffix after the value word (it starts with space/tab, or is empty), kept so a
    writer can preserve a line's comment; writers keep it ONLY if it matches `^[ \t]+#`. An unterminated
    quote is treated as an unquoted value. Copied verbatim into eink_client.py, sd-conf and sd_setup.py
    (standalone scripts); a test pins that the source text is identical."""
    body = rest.lstrip(" \t")
    if len(body) != len(rest) and body.startswith("#"):
        return "", rest
    if body[:1] in ("'", '"'):
        end = body.find(body[0], 1)
        if end != -1:
            glued = re.match(r"[^ \t]*", body[end + 1:]).group(0)
            return body[1:end] + glued, body[end + 1 + len(glued):]
    m = re.search(r"[ \t]+#", body)
    cut = m.start() if m else len(body)
    return body[:cut].strip(" \t"), body[cut:]


def parse_conf_text(text: str) -> dict:
    """KEY -> value for every assignment line in a pieria.conf, bash-sourcing semantics (last wins)."""
    out = {}
    for raw in (text or "").split("\n"):   # bash breaks lines on \n ONLY (not \v \f \x85 ...)
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, rest = line.partition("=")
        key = key.strip()
        if key:
            out[key] = split_conf_value(rest)[0]
    return out


@dataclass
class EinkConfig:
    server_url: str
    display_id: str
    min_interval: int = 900
    saturation: float = 0.5
    orientation: str = ""   # "" = landscape (1600x1200) | "portrait" = rotate 90 (1200x1600)
    w: int = 1600
    h: int = 1200
    palette: str = "spectra6"
    fit: str = "cover"

    @classmethod
    def from_env(cls) -> EinkConfig:
        return cls(
            server_url=os.environ.get("SERVER_URL", "http://localhost:8000").rstrip("/"),
            display_id=os.environ.get("DISPLAY_ID", "default"),
            min_interval=_int_env("EINK_MIN_INTERVAL", 900),
            orientation=os.environ.get("EINK_ORIENTATION", "").strip(),
        )

    @property
    def render_size(self) -> tuple[int, int]:
        """The w,h to ASK the server for — swapped when the panel hangs portrait.

        The panel's buffer is always landscape (w x h native); `orientation=portrait` means it is
        physically rotated 90 degrees, so what the viewer sees is a h x w portrait canvas. The art must
        therefore be COMPOSED at h x w and rotated back to the native buffer at paint time — not
        composed landscape and rotated, which is what happened before this: the server framed for a
        1600x1200 landscape window and the client handed the 1200x1600 result to a 1600x1200 panel,
        so a portrait frame got both the wrong composition and a buffer the panel couldn't take.
        """
        return (self.h, self.w) if self.orientation == "portrait" else (self.w, self.h)

    @property
    def pull_url(self) -> str:
        w, h = self.render_size
        return (
            f"{self.server_url}/display/{self.display_id}/current.png"
            f"?w={w}&h={h}&palette={self.palette}&fit={self.fit}"
        )

    @property
    def schedule_url(self) -> str:
        return f"{self.server_url}/api/displays/{self.display_id}/schedule-state"


def _int_env(key: str, fallback: int) -> int:
    try:
        return int(str(os.environ.get(key, fallback)).strip())
    except (TypeError, ValueError):
        return fallback


def _float_env(key: str, fallback: float) -> float:
    try:
        return float(str(os.environ.get(key, fallback)).strip())
    except (TypeError, ValueError):
        return fallback


# --------------------------------------------------------------------------- panel abstraction

class EinkClient(ABC):
    """Minimal interface over an e-ink panel, so the pusher is hardware-agnostic + testable."""

    resolution: Optional[Tuple[int, int]] = None

    @abstractmethod
    def show(self, image: Image.Image, saturation: float) -> None:
        """Blit an image to the panel (blocking; a full refresh is ~20-35s)."""


class InkyClient(EinkClient):
    """Thin wrapper over the Pimoroni `inky` library. THE ONLY HARDWARE-DEPENDENT CODE.
    `inky_factory` is injectable purely so tests can verify the mapping without the real library or a
    panel; the default factory does the `inky` import INSIDE itself so importing this module on a
    laptop with no `inky` installed never fails."""

    def __init__(self, inky_factory=None):
        self._inky_factory = inky_factory or self._default_factory
        self._inky = None

    @staticmethod
    def _default_factory():
        from inky.auto import auto  # lazy: optional dep, only needed on real hardware
        return auto()

    def _panel(self):
        if self._inky is None:
            self._inky = self._inky_factory()
        return self._inky

    @property
    def resolution(self):
        return self._panel().resolution

    def show(self, image: Image.Image, saturation: float) -> None:
        panel = self._panel()
        panel.set_image(image, saturation=saturation)
        panel.show()


class FakeInkyClient(EinkClient):
    """In-memory EinkClient for tests: records calls, optionally raises to exercise error paths."""

    def __init__(self, fail_on: Optional[str] = None, resolution: Tuple[int, int] = (1600, 1200)):
        self.calls: list = []
        self.shown: list = []
        self.resolution = resolution
        self._fail_on = fail_on

    def show(self, image: Image.Image, saturation: float) -> None:
        if self._fail_on == "show":
            raise RuntimeError("simulated failure in show")
        self.calls.append(("show", image.size, saturation))
        self.shown.append((image.size, saturation))


# --------------------------------------------------------------------------- HTTP (real fetch/schedule)

def make_http_fetch_fn(timeout: float = 15.0) -> FetchFn:
    """Real FetchFn: GET the pull URL, return (body bytes, headers dict)."""

    def _fetch(url: str) -> Tuple[bytes, dict]:
        r = httpx.get(url, timeout=timeout)
        r.raise_for_status()
        return r.content, dict(r.headers)

    return _fetch


def make_http_schedule_fn(cfg: EinkConfig, timeout: float = 8.0) -> ScheduleFn:
    """Real ScheduleFn: GET the display's schedule-state JSON."""

    def _schedule() -> dict:
        r = httpx.get(cfg.schedule_url, timeout=timeout)
        r.raise_for_status()
        return r.json()

    return _schedule


# --------------------------------------------------------------------------- orchestration

def push_once(
    cfg: EinkConfig,
    fetch_fn: FetchFn,
    client: EinkClient,
    *,
    last_etag: Optional[str] = None,
    interval: Optional[int] = None,
) -> dict:
    """Pull the current image and paint it if it changed. Returns a status dict; never raises for
    expected conditions (unchanged). A fetch or client.show error propagates to the caller
    (run_tick handles it) — matches frame_push's contract.

    `interval` (ADR-121) is this client's own actual pull cadence — the sleep it settled on after its
    LAST tick, i.e. max(display_time, EINK_MIN_INTERVAL). Sent as `&interval=` so the server can widen
    its KNOWN-display window to match a cadence slower than 2x display_time (a fixed EINK_MIN_INTERVAL
    floor made a 30s playlist's panel disappear from /api/remote/displays for 14 of every 15 minutes —
    the 2026-09-20 bench incident). run_tick supplies the previous tick's `sleep`; the very first tick
    of a run has none yet, so it's omitted (the server's un-widened fallback covers that one request)."""
    url = cfg.pull_url
    if interval:
        url = f"{url}&interval={int(interval)}"
    content, headers = fetch_fn(url)

    etag = headers.get("ETag") or headers.get("etag")
    if not etag:
        # Content-hash fallback so dedupe still works if the server ever omits the header.
        etag = "sha256:" + hashlib.sha256(content).hexdigest()[:16]

    if last_etag is not None and etag == last_etag:
        return {"status": "unchanged", "etag": etag}

    image = Image.open(io.BytesIO(content))
    image.load()
    if cfg.orientation == "portrait":
        # The server composed this at (h, w) portrait; rotate it back onto the panel's native
        # landscape buffer. Net effect: the panel gets its normal w x h, the viewer sees portrait art.
        image = image.rotate(90, expand=True)

    client.show(image, cfg.saturation)

    refresh_after = headers.get("X-Refresh-After") or headers.get("x-refresh-after")
    try:
        refresh_after = int(refresh_after)
    except (TypeError, ValueError):
        refresh_after = cfg.min_interval

    return {"status": "painted", "etag": etag, "refresh_after": refresh_after}


def _write_heartbeat() -> None:
    path = os.environ.get("EINK_HEARTBEAT_FILE", DEFAULT_HEARTBEAT_FILE)
    try:
        with open(path, "w") as f:
            f.write(str(time.time()))
    except OSError:
        pass  # e.g. no /run perms on a laptop/test box — non-fatal, watchdog-friendly only


def run_tick(
    cfg: EinkConfig,
    fetch_fn: FetchFn,
    client: EinkClient,
    schedule_fn: ScheduleFn,
    state: dict,
) -> dict:
    """A single scheduler tick (factored out so it's testable without the infinite loop). Never
    raises — errors are caught and reported in the returned status dict."""
    try:
        sched = schedule_fn()
    except Exception as e:
        logger.warning(f"[eink] schedule-state fetch failed, proceeding to paint: {e}")
        sched = {}

    if sched.get("quiet"):
        # E-ink holds its image at zero power — night/quiet means STOP refreshing, not blank.
        return {"status": "quiet", "sleep": cfg.min_interval}

    try:
        res = push_once(cfg, fetch_fn, client, last_etag=state.get("last_etag"),
                         interval=state.get("last_interval"))
    except Exception as e:
        # Escalating backoff, starting SHORT. An all-in-one Pi boots its e-ink client seconds before its
        # own Docker container binds the port, so the very first tick after a fresh setup reliably gets
        # connection-refused. Sleeping the full steady-state interval on that one failure left a
        # just-configured panel holding a stale image for 15 minutes — indistinguishable from a broken
        # device to the person who just set it up (found on hardware 2026-07-21).
        fails = state["error_streak"] = state.get("error_streak", 0) + 1
        backoff = min(cfg.min_interval, ERROR_BACKOFF_BASE * (2 ** (fails - 1)))
        logger.error(f"[eink] tick failed (attempt {fails}): {e} — retrying in {backoff}s")
        return {"status": "error", "reason": str(e) or e.__class__.__name__,
                "attempt": fails, "sleep": backoff}

    state["error_streak"] = 0
    res = dict(res)
    if res["status"] == "painted":
        state["last_etag"] = res["etag"]
        _write_heartbeat()
        res["sleep"] = max(res.get("refresh_after", cfg.min_interval), cfg.min_interval)
    else:  # unchanged
        res["sleep"] = cfg.min_interval
    # Remembered for the NEXT tick's `&interval=` (see push_once) — one tick behind, which converges
    # immediately and matches steady state since a playlist's display_time rarely changes tick to tick.
    state["last_interval"] = res["sleep"]
    return res


def run_loop(
    cfg: EinkConfig,
    fetch_fn: Optional[FetchFn] = None,
    client: Optional[EinkClient] = None,
    schedule_fn: Optional[ScheduleFn] = None,
) -> None:
    """Periodic pull-and-paint loop. Builds real httpx fetch/schedule fns and an InkyClient unless
    injected (tests / --dry-run inject fakes)."""
    fetch_fn = fetch_fn or make_http_fetch_fn()
    schedule_fn = schedule_fn or make_http_schedule_fn(cfg)
    client = client or InkyClient()

    logger.info(f"[eink] loop started (display={cfg.display_id}, server={cfg.server_url})")
    state: dict = {}
    while True:
        res = run_tick(cfg, fetch_fn, client, schedule_fn, state)
        if res.get("status") not in ("unchanged", "quiet"):
            logger.info(f"[eink] tick: {res}")
        time.sleep(max(1, res.get("sleep", cfg.min_interval)))