#!/usr/bin/env python3
"""Pi HS handshake sense for listen (Rev E J19).

Serial data still comes from XR UART (ttyXRUSB). J3/J4 on PI disconnects XR
GPIO1 from THVD DE/HS, so uartd / HandshakeMonitor SLV-IN is blind.

Standing map (08-25 .140, CURRENT.md):
  BCM10 OIS_OUT, BCM11 M-S_OUT — idle driven low (never high-Z)
  BCM8  OIS_IN,  BCM9  M-S_IN  — active-low vs bus driven (IN low = HS active)

Listen uses this monitor instead of XR GPIO1 when POG422_HS_PATH=pi.
Listen does not drive the bus. SuperBall slave DE uses GpioOutputLine
(BCM11 M-S_OUT, idle low). Does not bump uartd.
"""
from __future__ import annotations

import errno
import os
import shutil
import struct
import subprocess
from pathlib import Path
from typing import Any, Optional

# Rev E J19 HS OUT (park idle-low even when protocol is UART).
HS_OUT_IDLE_LOW_BCMS = (10, 11)

# Rev E J19 HS IN (not the OUT pair).
HS_IN_BCM = {
    "machine": 9,  # M-S_IN
    "ois": 8,  # OIS_IN
}

_HS_PATH_ALIASES = {
    "uart": "uart",
    "uart_hs": "uart",
    "hat": "uart",
    "xr": "uart",
    "pi": "pi",
    "pi_hs": "pi",
    "gpio": "pi",
    "j19": "pi",
}

GPIO_MAX_NAME_SIZE = 32
GPIO_V2_LINES_MAX = 64
SIZE_CHIPINFO = 68
SIZE_LINE_CONFIG = 272
SIZE_LINE_REQUEST = 592
SIZE_LINE_VALUES = 16

FLAG_INPUT = 1 << 2
FLAG_OUTPUT = 1 << 3
FLAG_EDGE_RISING = 1 << 4
FLAG_EDGE_FALLING = 1 << 5
FLAG_BIAS_PULL_UP = 1 << 8
ATTR_ID_OUTPUT_VALUES = 2
SIZE_LINE_EVENT = 48
EVENT_RISING_EDGE = 1
EVENT_FALLING_EDGE = 2
GPIO_EVENT_BUFFER = 1024

_IOC_NRBITS, _IOC_TYPEBITS, _IOC_SIZEBITS = 8, 8, 14
_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS
_IOC_WRITE, _IOC_READ = 1, 2

PINCTRL_LABELS = ("pinctrl-rp1", "pinctrl-bcm2711", "pinctrl-bcm2835", "pinctrl-bcm2712")

try:
    import fcntl
except ImportError:  # pragma: no cover - Windows unit tests
    fcntl = None  # type: ignore[assignment]


def _ioc(direction: int, typ: int, nr: int, size: int) -> int:
    value = (
        (direction << _IOC_DIRSHIFT)
        | (size << _IOC_SIZESHIFT)
        | (typ << _IOC_TYPESHIFT)
        | (nr << _IOC_NRSHIFT)
    )
    if value >= 1 << 31:
        value -= 1 << 32
    return value


GPIO_GET_CHIPINFO_IOCTL = _ioc(_IOC_READ, 0xB4, 0x01, SIZE_CHIPINFO)
GPIO_V2_GET_LINE_IOCTL = _ioc(_IOC_READ | _IOC_WRITE, 0xB4, 0x07, SIZE_LINE_REQUEST)
GPIO_V2_LINE_GET_VALUES_IOCTL = _ioc(_IOC_READ | _IOC_WRITE, 0xB4, 0x0E, SIZE_LINE_VALUES)
GPIO_V2_LINE_SET_VALUES_IOCTL = _ioc(_IOC_READ | _IOC_WRITE, 0xB4, 0x0F, SIZE_LINE_VALUES)


def normalize_hs_path(raw: str | None) -> str:
    key = (raw or "").strip().lower()
    return _HS_PATH_ALIASES.get(key, "uart" if not key else key)


def hs_in_bcm(channel_key: str) -> int:
    key = (channel_key or "machine").strip().lower()
    if key not in HS_IN_BCM:
        raise ValueError(f"no Pi HS IN BCM for channel {channel_key!r}")
    return HS_IN_BCM[key]


def line_is_active(level_hi: bool, *, active_low: bool = True) -> bool:
    """IN low = bus driven when active_low (same as XR HandshakeMonitor)."""
    return (not level_hi) if active_low else level_hi


def _read_env_file_key(path: Path, key: str) -> str:
    if not path.is_file():
        return ""
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return ""
    want = key + "="
    found = ""
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#") or not stripped.startswith(want):
            continue
        found = stripped.split("=", 1)[1].strip().strip('"').strip("'")
    return found


def resolve_hs_path() -> str:
    raw = (os.environ.get("POG422_HS_PATH") or os.environ.get("hs_mode") or "").strip()
    if not raw:
        etc = Path(os.environ.get("ETHERGOLD_ETC", "/etc/ethergold"))
        raw = _read_env_file_key(etc / "ethergold.env", "POG422_HS_PATH")
    path = normalize_hs_path(raw)
    return path if path in ("uart", "pi") else "uart"


def park_hs_out_idle_low() -> dict[str, Any]:
    """Drive BCM10 OIS_OUT / BCM11 M-S_OUT output-low.

    Always, independent of POG422_HS_PATH. UART protocol still uses XR GPIO1
    for HS sense/drive; this only prevents TXS A-side pull-ups from holding DE
    if J3/J4 are on PI while config still says UART.
    Best-effort: missing pinctrl (Windows / non-Pi) is skipped, not fatal.
    """
    pinctrl = shutil.which("pinctrl")
    if not pinctrl:
        return {"ok": True, "skipped": "no-pinctrl"}
    lines: list[str] = []
    for bcm in HS_OUT_IDLE_LOW_BCMS:
        try:
            completed = subprocess.run(
                [pinctrl, "set", str(bcm), "op", "dl"],
                check=False,
                capture_output=True,
                text=True,
                timeout=8,
            )
        except OSError as exc:
            return {"ok": False, "error": str(exc), "lines": lines}
        if completed.returncode != 0:
            err = (completed.stderr or completed.stdout or "").strip()
            return {"ok": False, "error": f"BCM{bcm}: {err}", "lines": lines}
        lines.append(f"BCM{bcm}=op dl")
    return {"ok": True, "pins": lines}


def _read_chip_info(path: str) -> Optional[tuple[str, str, int]]:
    if fcntl is None:
        return None
    fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC)
    try:
        buf = bytearray(SIZE_CHIPINFO)
        try:
            fcntl.ioctl(fd, GPIO_GET_CHIPINFO_IOCTL, buf, True)
        except OSError:
            return None
    finally:
        os.close(fd)
    name = buf[0:GPIO_MAX_NAME_SIZE].split(b"\x00", 1)[0].decode("ascii", "replace")
    label = buf[GPIO_MAX_NAME_SIZE : 2 * GPIO_MAX_NAME_SIZE]
    label = label.split(b"\x00", 1)[0].decode("ascii", "replace")
    (lines,) = struct.unpack_from("<I", buf, 2 * GPIO_MAX_NAME_SIZE)
    return name, label, int(lines)


def _find_header_chip(explicit: Optional[str] = None) -> tuple[str, int]:
    if explicit:
        info = _read_chip_info(explicit)
        if info is None:
            raise OSError(f"cannot read chip info from {explicit}")
        return explicit, info[2]
    chips: list[tuple[str, str, int]] = []
    dev = Path("/dev")
    if dev.is_dir():
        for entry in sorted(dev.glob("gpiochip*")):
            info = _read_chip_info(str(entry))
            if info is not None:
                chips.append((str(entry), info[1], info[2]))
    if not chips:
        raise OSError("no /dev/gpiochip* found")
    for want in PINCTRL_LABELS:
        for path, label, lines in chips:
            if label == want:
                return path, lines
    for path, _label, lines in chips:
        if lines >= 28:
            return path, lines
    listing = ", ".join(f"{p}:{lab}({n})" for p, lab, n in chips)
    raise OSError(f"no header GPIO controller among: {listing}")


def decode_gpio_v2_line_events(raw: bytes, *, active_low: bool = True) -> list[dict]:
    """Parse gpio_v2_line_event records. Rising pin = inactive when active_low."""
    out: list[dict] = []
    n = len(raw) - (len(raw) % SIZE_LINE_EVENT)
    for off in range(0, n, SIZE_LINE_EVENT):
        timestamp_ns, evid = struct.unpack_from("<QI", raw, off)
        pin_hi = evid == EVENT_RISING_EDGE
        active = line_is_active(pin_hi, active_low=active_low)
        out.append(
            {
                "kind": "active" if active else "inactive",
                "timestamp_ns": int(timestamp_ns),
                "gpio_event_id": int(evid),
            }
        )
    return out


def _request_input_line(
    chip_path: str,
    chip_lines: int,
    line: int,
    consumer: str,
    *,
    edges: bool = True,
    event_buffer_size: int = GPIO_EVENT_BUFFER,
) -> int:
    if fcntl is None:
        raise OSError("fcntl/GPIO not available")
    if line >= chip_lines:
        raise OSError(f"GPIO line {line} out of range for {chip_path} ({chip_lines} lines)")
    fd = os.open(chip_path, os.O_RDONLY | os.O_CLOEXEC)
    try:
        buf = bytearray(SIZE_LINE_REQUEST)
        struct.pack_into("<I", buf, 0, line)
        name = consumer.encode("ascii", "replace")[: GPIO_MAX_NAME_SIZE - 1]
        off = GPIO_V2_LINES_MAX * 4
        buf[off : off + len(name)] = name
        cfg = off + GPIO_MAX_NAME_SIZE
        flags = FLAG_INPUT | FLAG_BIAS_PULL_UP
        if edges:
            flags |= FLAG_EDGE_RISING | FLAG_EDGE_FALLING
        struct.pack_into("<Q", buf, cfg, flags)
        struct.pack_into("<I", buf, cfg + 8, 0)
        tail = cfg + SIZE_LINE_CONFIG
        struct.pack_into("<I", buf, tail, 1)
        struct.pack_into("<I", buf, tail + 4, int(event_buffer_size if edges else 0))
        try:
            fcntl.ioctl(fd, GPIO_V2_GET_LINE_IOCTL, buf, True)
        except OSError as exc:
            if exc.errno == errno.EBUSY:
                raise OSError(
                    f"GPIO{line} already claimed on {chip_path} "
                    f"(another listen/collector? gpioinfo {chip_path} {line})"
                ) from exc
            if exc.errno in (errno.EACCES, errno.EPERM):
                raise OSError(
                    f"permission denied on {chip_path} — user must be in group gpio"
                ) from exc
            raise OSError(f"GPIO_V2_GET_LINE failed on line {line}: {exc}") from exc
        (req_fd,) = struct.unpack_from("<i", buf, SIZE_LINE_REQUEST - 4)
        if req_fd < 0:
            raise OSError(f"kernel returned invalid GPIO request fd {req_fd}")
        return req_fd
    finally:
        os.close(fd)


def _read_line_hi(req_fd: int) -> Optional[bool]:
    if fcntl is None:
        return None
    buf = bytearray(SIZE_LINE_VALUES)
    struct.pack_into("<QQ", buf, 0, 0, 1)
    try:
        fcntl.ioctl(req_fd, GPIO_V2_LINE_GET_VALUES_IOCTL, buf, True)
    except OSError:
        return None
    (bits, _mask) = struct.unpack_from("<QQ", buf, 0)
    return bool(bits & 1)


def _request_output_line(chip_path: str, chip_lines: int, line: int, consumer: str) -> int:
    """Claim one line as output, default low. Same gpio_v2_line_request layout as IN."""
    if fcntl is None:
        raise OSError("fcntl/GPIO not available")
    if line >= chip_lines:
        raise OSError(f"GPIO line {line} out of range for {chip_path} ({chip_lines} lines)")
    fd = os.open(chip_path, os.O_RDONLY | os.O_CLOEXEC)
    try:
        buf = bytearray(SIZE_LINE_REQUEST)
        struct.pack_into("<I", buf, 0, line)
        name = consumer.encode("ascii", "replace")[: GPIO_MAX_NAME_SIZE - 1]
        off = GPIO_V2_LINES_MAX * 4
        buf[off : off + len(name)] = name
        cfg = off + GPIO_MAX_NAME_SIZE
        struct.pack_into("<Q", buf, cfg, FLAG_OUTPUT)
        struct.pack_into("<I", buf, cfg + 8, 1)
        attr = cfg + 32
        struct.pack_into("<I", buf, attr, ATTR_ID_OUTPUT_VALUES)
        struct.pack_into("<Q", buf, attr + 8, 0)
        struct.pack_into("<Q", buf, attr + 16, 1)
        tail = cfg + SIZE_LINE_CONFIG
        struct.pack_into("<I", buf, tail, 1)
        struct.pack_into("<I", buf, tail + 4, 1)
        try:
            fcntl.ioctl(fd, GPIO_V2_GET_LINE_IOCTL, buf, True)
        except OSError as exc:
            if exc.errno == errno.EBUSY:
                raise OSError(
                    f"GPIO{line} already claimed on {chip_path} (pinctrl get {line})"
                ) from exc
            if exc.errno in (errno.EACCES, errno.EPERM):
                raise OSError(
                    f"permission denied on {chip_path} — user must be in group gpio"
                ) from exc
            raise OSError(f"GPIO_V2_GET_LINE OUTPUT failed on line {line}: {exc}") from exc
        (req_fd,) = struct.unpack_from("<i", buf, SIZE_LINE_REQUEST - 4)
        if req_fd < 0:
            raise OSError(f"kernel returned invalid GPIO request fd {req_fd}")
        return req_fd
    finally:
        os.close(fd)


def _set_line_hi(req_fd: int, high: bool) -> None:
    if fcntl is None:
        raise OSError("fcntl/GPIO not available")
    buf = bytearray(SIZE_LINE_VALUES)
    struct.pack_into("<QQ", buf, 0, 1 if high else 0, 1)
    fcntl.ioctl(req_fd, GPIO_V2_LINE_SET_VALUES_IOCTL, buf, True)


class GpioOutputLine:
    """One BCM header pin as output (SuperBall slave DE). Idle low. ioctl, not pinctrl."""

    def __init__(
        self,
        bcm: int,
        *,
        consumer: str = "pog422-de",
        chip: Optional[str] = None,
    ) -> None:
        self.bcm = int(bcm)
        chip_path, chip_lines = _find_header_chip(chip)
        self._req_fd = _request_output_line(
            chip_path, chip_lines, self.bcm, consumer[: GPIO_MAX_NAME_SIZE - 1]
        )
        self._high = False

    def set(self, high: bool) -> None:
        _set_line_hi(self._req_fd, bool(high))
        self._high = bool(high)

    def close(self) -> None:
        fd = getattr(self, "_req_fd", -1)
        if fd is not None and fd >= 0:
            try:
                _set_line_hi(fd, False)
            except OSError:
                pass
            try:
                os.close(fd)
            except OSError:
                pass
            self._req_fd = -1
        self._high = False

    def __enter__(self) -> "GpioOutputLine":
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()


class GpioHandshakeMonitor:
    """BCM8/9 HS IN. Poll `is_active()` plus kernel-stamped `drain_edges()`."""

    def __init__(
        self,
        channel_key: str,
        *,
        active_low: bool = True,
        chip: Optional[str] = None,
    ) -> None:
        self.channel_key = (channel_key or "machine").strip().lower()
        self.active_low = active_low
        self.bcm = hs_in_bcm(self.channel_key)
        chip_path, chip_lines = _find_header_chip(chip)
        self._req_fd = _request_input_line(
            chip_path, chip_lines, self.bcm, f"pog422-hs-{self.channel_key}"
        )
        try:
            os.set_blocking(self._req_fd, False)
        except OSError:
            pass
        self.gpio_name = f"BCM{self.bcm}"

    def is_active(self) -> bool:
        level_hi = _read_line_hi(self._req_fd)
        if level_hi is None:
            return False
        return line_is_active(level_hi, active_low=self.active_low)

    def is_active_cached(self) -> bool:
        return self.is_active()

    def drain_edges(self) -> list[dict]:
        """Kernel-stamped rise/fall since last drain. Empty if none pending."""
        fd = getattr(self, "_req_fd", -1)
        if fd is None or fd < 0:
            return []
        chunks = bytearray()
        while True:
            try:
                raw = os.read(fd, SIZE_LINE_EVENT * 32)
            except BlockingIOError:
                break
            except OSError:
                break
            if not raw:
                break
            chunks.extend(raw)
            if len(raw) < SIZE_LINE_EVENT * 32:
                break
        if not chunks:
            return []
        return decode_gpio_v2_line_events(bytes(chunks), active_low=self.active_low)

    def close(self) -> None:
        fd = getattr(self, "_req_fd", -1)
        if fd is not None and fd >= 0:
            try:
                os.close(fd)
            except OSError:
                pass
            self._req_fd = -1

    def __enter__(self) -> "GpioHandshakeMonitor":
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()