backend.py

"""Backend interface separating real-system access from logic.

Everything that touches the Pi's kernel, sysfs, or subprocesses lives behind
``UsbipBackend``. The pure logic in ``usbip.py`` only ever calls these five
primitives, so it runs identically against ``RealBackend`` (production) and
``FakeBackend`` (tests + ``--fake`` demo mode). The only thing that ever needs
verifying on real hardware is that ``RealBackend`` reads the box correctly.
"""

from __future__ import annotations

import glob
import os
import re
import subprocess
from dataclasses import dataclass, field
from typing import Protocol

# Absolute paths, never resolved via PATH: the flush runs under passwordless
# root sudo, so a mutable PATH must never get to choose which binary runs.
USBIP = "/usr/sbin/usbip"
SS = "/usr/bin/ss"
JOURNALCTL = "/usr/bin/journalctl"

USBIP_PORT = 3240
SYSFS_USB = "/sys/bus/usb/devices"


@dataclass(frozen=True)
class RawDevice:
    """Raw per-device facts read straight from the box, no interpretation.

    Field names mirror the DRA driver's device schema (bus/host/product/
    productName/vendor/vendorName) so the page and the cluster speak the same
    vocabulary when an operator reads them side by side.
    """

    busid: str
    status: int  # 1 = available, 2 = in use (exported to a client)
    vendor: str = ""  # idVendor, e.g. "0403"
    product: str = ""  # idProduct, e.g. "6001"
    vendorName: str = ""  # manufacturer string descriptor
    productName: str = ""  # product string descriptor


@dataclass(frozen=True)
class FlushResult:
    busid: str
    ok: bool
    message: str
    steps: list[str] = field(default_factory=list)


class UsbipBackend(Protocol):
    """The five system-touching primitives. Logic depends only on this."""

    def host(self) -> str: ...
    def read_devices(self) -> list[RawDevice]: ...
    def connection_count(self) -> int: ...
    def read_logs(self, lines: int) -> list[str]: ...
    def flush(self, busid: str) -> FlushResult: ...


# --------------------------------------------------------------------------- #
# Real implementation: the thin adapter onto the actual Pi.
# --------------------------------------------------------------------------- #


class RealBackend:
    """Reads sysfs and shells out to usbip/ss/journalctl on the local Pi."""

    def __init__(self, hostname: str | None = None) -> None:
        self._host = hostname or os.uname().nodename

    def host(self) -> str:
        return self._host

    def read_devices(self) -> list[RawDevice]:
        devices: list[RawDevice] = []
        # usbip_status only exists for devices bound to the usbip-host driver,
        # so globbing it is exactly the set of exportable devices.
        for status_path in sorted(glob.glob(f"{SYSFS_USB}/*/usbip_status")):
            dev_dir = os.path.dirname(status_path)
            busid = os.path.basename(dev_dir)
            status = _read_int(status_path)
            if status is None:
                continue
            devices.append(
                RawDevice(
                    busid=busid,
                    status=status,
                    vendor=_read_str(f"{dev_dir}/idVendor"),
                    product=_read_str(f"{dev_dir}/idProduct"),
                    vendorName=_read_str(f"{dev_dir}/manufacturer"),
                    productName=_read_str(f"{dev_dir}/product"),
                )
            )
        return devices

    def connection_count(self) -> int:
        """Count established client connections to the usbip daemon.

        This is the count only: the usbip-host driver does not expose which
        busid each peer holds, so we deliberately cannot map peer -> device.
        The count feeds the surplus test in the logic layer.
        """
        try:
            out = subprocess.run(
                [SS, "-tnH", "state", "established", f"( sport = :{USBIP_PORT} )"],
                capture_output=True,
                text=True,
                timeout=5,
            ).stdout
        except (subprocess.SubprocessError, OSError):
            return 0
        return sum(1 for line in out.splitlines() if line.strip())

    def read_logs(self, lines: int) -> list[str]:
        try:
            out = subprocess.run(
                [JOURNALCTL, "-u", "usbipd", "-n", str(lines), "--no-pager", "-o", "cat"],
                capture_output=True,
                text=True,
                timeout=5,
            ).stdout
        except (subprocess.SubprocessError, OSError):
            return []
        return [ln for ln in out.splitlines() if ln.strip()]

    def flush(self, busid: str) -> FlushResult:
        """unbind then bind. Success = both commands exit cleanly.

        We do NOT assert on the final usbip_status: a healthy flush legitimately
        ends at status 2 when a waiting client reattaches instantly. If bind
        fails (device vanished mid-flush) the device is left unexported, which
        is worse than stale, so we retry bind once then report loudly.
        """
        steps: list[str] = []

        rc, msg = self._run_usbip("unbind", busid)
        steps.append(f"unbind: {msg}")
        if rc != 0:
            return FlushResult(busid, False, f"unbind failed: {msg}", steps)

        rc, msg = self._run_usbip("bind", busid)
        steps.append(f"bind: {msg}")
        if rc != 0:
            rc, msg2 = self._run_usbip("bind", busid)  # one retry
            steps.append(f"bind (retry): {msg2}")
            if rc != 0:
                return FlushResult(
                    busid,
                    False,
                    "Device may be physically gone — left unbound. "
                    "Check the device is still attached to the Pi.",
                    steps,
                )

        return FlushResult(busid, True, "Flushed; device returned to the pool.", steps)

    def _run_usbip(self, action: str, busid: str) -> tuple[int, str]:
        try:
            proc = subprocess.run(
                ["sudo", USBIP, action, "-b", busid],
                capture_output=True,
                text=True,
                timeout=15,
            )
        except subprocess.TimeoutExpired:
            return 1, "timed out"
        except OSError as exc:
            return 1, str(exc)
        out = (proc.stderr or proc.stdout or "").strip()
        return proc.returncode, out or ("ok" if proc.returncode == 0 else "failed")


def _read_str(path: str) -> str:
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            return fh.read().strip()
    except OSError:
        return ""


def _read_int(path: str) -> int | None:
    raw = _read_str(path)
    try:
        return int(raw)
    except ValueError:
        return None


# --------------------------------------------------------------------------- #
# Fake implementation: pure in-memory state for tests and --fake demo mode.
# --------------------------------------------------------------------------- #


class FakeBackend:
    """In-memory backend. Reproduces states real hardware can't stage on demand
    (surplus, deep hub trees, specific log sequences) and mutates on flush so
    the UI can be exercised end to end without a Pi."""

    def __init__(
        self,
        hostname: str = "10.71.20.99 (fake)",
        devices: list[RawDevice] | None = None,
        connections: int = 0,
        logs: list[str] | None = None,
    ) -> None:
        self._host = hostname
        self._devices = list(devices) if devices is not None else _default_fake_devices()
        self._connections = connections
        self._logs = list(logs) if logs is not None else _default_fake_logs()

    def host(self) -> str:
        return self._host

    def read_devices(self) -> list[RawDevice]:
        return list(self._devices)

    def connection_count(self) -> int:
        return self._connections

    def read_logs(self, lines: int) -> list[str]:
        return self._logs[-lines:]

    def flush(self, busid: str) -> FlushResult:
        idx = next((i for i, d in enumerate(self._devices) if d.busid == busid), None)
        if idx is None:
            return FlushResult(busid, False, "No such device on this host.", [])
        # Model the real surprise: a waiting client reattaches immediately, so a
        # flushed device snaps back to status 2 rather than resting at 1.
        dev = self._devices[idx]
        self._devices[idx] = RawDevice(
            busid=dev.busid,
            status=2 if self._connections > 0 else 1,
            vendor=dev.vendor,
            product=dev.product,
            vendorName=dev.vendorName,
            productName=dev.productName,
        )
        self._logs.append(f"usbipd: info: bind device on busid {busid}: complete")
        return FlushResult(
            busid, True, "Flushed; device returned to the pool.", ["unbind: ok", "bind: ok"]
        )


def _default_fake_devices() -> list[RawDevice]:
    """The real ambiguous case from the beamline, plus a nested extender hub."""
    return [
        RawDevice("3-1.1", 2, "0403", "6001", "STMicroelectronics", "Standa 8SMC5"),
        RawDevice("3-1.4", 1, "0403", "6001", "STMicroelectronics", "Standa 8SMC5"),
        RawDevice("3-1.2.1", 1, "1a86", "7523", "Acme", "Extender-hub motor A"),
        RawDevice("3-1.2.4", 2, "1a86", "7523", "Acme", "Extender-hub motor B"),
    ]


def _default_fake_logs() -> list[str]:
    return [
        "usbipd: info: connection from 172.23.71.223:53750",
        "usbipd: info: received request: 0x8003(6)",
        "usbipd: info: found requested device: 3-1.1",
        "usbipd: info: request 0x8003(6): failed",
        "usbipd: info: connection from 172.23.71.223:53751",
        "usbipd: info: received request: 0x8003(6)",
        "usbipd: info: found requested device: 3-1.1",
        "usbipd: info: request 0x8003(6): failed",
    ]