LAN7800 EEPROM Programmer

"""Enumerate attached LAN7800 adapters (USB VID 0x0424 / PID 0x7800).

Pure /sys walking for the device inventory (free, no privilege). For each USB
match we resolve the bound netdev (if any), its driver and current MAC, and
best-effort probe whether the EEPROM carries a valid 0xA5 signature via a
NON-interactive `ethtool -e` (sudo -n) so it can never hang waiting for a
password — if privilege is missing the eeprom state is reported "unknown".
"""
from __future__ import annotations

import glob
import os
import shutil
import subprocess

from .eeprom import LAN7800_VID, LAN7800_PID, SIG_VALUE

USB_DEVICES = "/sys/bus/usb/devices"


def _read(path: str) -> str:
    try:
        with open(path) as fh:
            return fh.read().strip()
    except OSError:
        return ""


def _hex_id(path: str) -> int:
    v = _read(path)
    try:
        return int(v, 16)
    except ValueError:
        return -1


def _find_iface(usb_dev_dir: str) -> str | None:
    # netdev lives under an interface child: <dev>:<cfg>.<intf>/net/<iface>
    for net in glob.glob(os.path.join(usb_dev_dir, "*:*", "net", "*")):
        return os.path.basename(net)
    return None


def _iface_driver(iface: str) -> str:
    link = f"/sys/class/net/{iface}/device/driver"
    if os.path.islink(link):
        return os.path.basename(os.path.realpath(link))
    return ""


def _probe_eeprom_signed(iface: str) -> str:
    """Return 'signed' | 'blank' | 'unknown'. Never blocks (sudo -n)."""
    if shutil.which("ethtool") is None:
        return "unknown"
    for cmd in (["sudo", "-n", "ethtool", "-e", iface, "raw", "on",
                 "offset", "0", "length", "1"],
                ["ethtool", "-e", iface, "raw", "on",
                 "offset", "0", "length", "1"]):
        try:
            p = subprocess.run(cmd, capture_output=True, timeout=5)
        except (subprocess.TimeoutExpired, OSError):
            continue
        if p.returncode == 0 and p.stdout:
            return "signed" if p.stdout[0] == SIG_VALUE else "blank"
        err = p.stderr.decode(errors="replace")
        if "ENODATA" in err or "No data" in err:
            return "blank"
        # password / permission / no-device -> try next form, else unknown
    return "unknown"


def diagnose_netdev(usb_path: str) -> dict:
    """Explain why a bound LAN7800 may have no netdev (best effort, no privilege)."""
    dev_dir = os.path.join(USB_DEVICES, usb_path)
    intf_drivers = {}
    for intf in glob.glob(os.path.join(dev_dir, f"{usb_path}:*")):
        drv = os.path.realpath(os.path.join(intf, "driver")) if \
            os.path.islink(os.path.join(intf, "driver")) else ""
        intf_drivers[os.path.basename(intf)] = os.path.basename(drv) if drv else None
    module_loaded = os.path.isdir("/sys/module/lan78xx")
    has_netdev = bool(_find_iface(dev_dir))
    notes = []
    if module_loaded and any(d == "lan78xx" for d in intf_drivers.values()) \
            and not has_netdev:
        notes.append("interface is bound to lan78xx but no netdev was created; "
                     "the device is likely re-enumerating/resetting (a changing "
                     "USB devnum across probes points to link/power flapping), or "
                     "PHY init failed")
    if not module_loaded:
        notes.append("lan78xx module is not loaded")
    notes.append("full cause needs `dmesg | grep lan78xx` (requires privilege)")
    return {"interface_drivers": intf_drivers, "module_loaded": module_loaded,
            "has_netdev": has_netdev, "notes": notes}


def scan_devices(probe_eeprom: bool = True) -> list[dict]:
    found = []
    for idv in glob.glob(os.path.join(USB_DEVICES, "*", "idVendor")):
        dev_dir = os.path.dirname(idv)
        if _hex_id(idv) != LAN7800_VID:
            continue
        if _hex_id(os.path.join(dev_dir, "idProduct")) != LAN7800_PID:
            continue
        name = os.path.basename(dev_dir)          # e.g. "2-1"
        iface = _find_iface(dev_dir)
        mac = _read(f"/sys/class/net/{iface}/address") if iface else ""
        driver = _iface_driver(iface) if iface else ""
        entry = {
            "usb_path": name,
            "busnum": _read(os.path.join(dev_dir, "busnum")),
            "devnum": _read(os.path.join(dev_dir, "devnum")),
            "port_path": _read(os.path.join(dev_dir, "devpath")),
            "product": _read(os.path.join(dev_dir, "product")),
            "usb_serial": _read(os.path.join(dev_dir, "serial")),
            "iface": iface,
            "driver": driver,
            "mac": mac.upper() if mac else None,
            "bound": bool(iface),
            "eeprom": "unbound",
        }
        if iface and probe_eeprom:
            entry["eeprom"] = _probe_eeprom_signed(iface)
        found.append(entry)
    found.sort(key=lambda e: e["usb_path"])
    return found