juc500/juc500_xfer/usb_link.py

"""Low-level USB access to the j5create JUC500 Smart Data Link cable.

Device: VID 0x0711 / PID 0x7500 (product "Smart Data Link"), composite:
  IF5 Vendor bulk is required for our protocol on Windows (WinUSB).
"""

from __future__ import annotations

import array
import os
import subprocess
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Optional

import usb.core
import usb.util

VID = 0x0711
PID = 0x7500
# Some firmware / sibling OEM IDs under the same vendor
ALT_PIDS = {0x7500, 0x5500, 0x7000, 0x0700}

IF_CDC_COMM = 0
IF_CDC_DATA = 1
IF_VENDOR = 5

EP_DATA_OUT = 0x08
EP_DATA_IN = 0x89
EP_CTRL_OUT = 0x0A
EP_CTRL_IN = 0x8B

MAX_PACKET = 512
_IS_WIN = sys.platform == "win32"

_PRODUCT_HINTS = ("smart data", "data link", "wormhole", "juc500", "oti")


def _package_roots() -> list[Path]:
    here = Path(__file__).resolve().parent
    roots = [here.parent, Path.cwd()]
    if getattr(sys, "frozen", False):
        roots.insert(0, Path(sys.executable).resolve().parent)
        mei = getattr(sys, "_MEIPASS", None)
        if mei:
            roots.insert(0, Path(mei))
    # walk up a few levels looking for vendor/
    for base in list(roots):
        p = base
        for _ in range(4):
            roots.append(p)
            if (p / "vendor").is_dir():
                break
            p = p.parent
    # unique preserve order
    seen: set[str] = set()
    out: list[Path] = []
    for r in roots:
        key = str(r)
        if key not in seen:
            seen.add(key)
            out.append(r)
    return out


def _libusb_candidates() -> list[str]:
    names = []
    if _IS_WIN:
        names = ["libusb-1.0.dll"]
    elif sys.platform == "darwin":
        names = ["libusb-1.0.dylib"]
    else:
        names = ["libusb-1.0.so", "libusb-1.0.so.0"]

    candidates: list[str] = []
    if getattr(sys, "frozen", False):
        mei = getattr(sys, "_MEIPASS", None)
        exe_dir = Path(sys.executable).resolve().parent
        if mei:
            for n in names:
                candidates.append(str(Path(mei) / n))
        for n in names:
            candidates.append(str(exe_dir / n))
            candidates.append(str(exe_dir / "_internal" / n))

    for root in _package_roots():
        for n in names:
            candidates.append(str(root / "vendor" / n))
            candidates.append(str(root / n))

    if _IS_WIN:
        # venv Scripts / beside python
        py_dir = Path(sys.executable).resolve().parent
        for n in names:
            candidates.append(str(py_dir / n))
            candidates.append(str(py_dir / "DLLs" / n))
        pf = os.environ.get("PROGRAMFILES", r"C:\Program Files")
        candidates.append(os.path.join(pf, "libusb", "libusb-1.0.dll"))
        candidates.append("libusb-1.0.dll")  # PATH search
    elif sys.platform == "darwin":
        candidates += [
            os.path.expanduser("~/homebrew/lib/libusb-1.0.dylib"),
            "/opt/homebrew/lib/libusb-1.0.dylib",
            "/usr/local/lib/libusb-1.0.dylib",
        ]
    else:
        candidates += ["libusb-1.0.so", "libusb-1.0.so.0"]

    # dedupe
    seen: set[str] = set()
    out: list[str] = []
    for c in candidates:
        if c and c not in seen:
            seen.add(c)
            out.append(c)
    return out


_BACKEND = None
_BACKEND_PATH: Optional[str] = None


def _libusb_backend():
    """Locate / cache libusb-1.0 backend."""
    global _BACKEND, _BACKEND_PATH
    if _BACKEND is not None:
        return _BACKEND

    import usb.backend.libusb1

    last_err: Optional[BaseException] = None
    for path in _libusb_candidates():
        absolute = os.path.sep in path or (len(path) > 1 and path[1] == ":")
        if absolute and not os.path.exists(path):
            continue
        try:
            be = usb.backend.libusb1.get_backend(find_library=lambda _n, p=path: p)
            if be is not None:
                # smoke: enumerate (may be empty)
                list(usb.core.find(find_all=True, backend=be))
                _BACKEND = be
                _BACKEND_PATH = path
                return be
        except Exception as exc:  # noqa: BLE001
            last_err = exc
            continue

    try:
        be = usb.backend.libusb1.get_backend()
        if be is not None:
            list(usb.core.find(find_all=True, backend=be))
            _BACKEND = be
            _BACKEND_PATH = "(system)"
            return be
    except Exception as exc:  # noqa: BLE001
        last_err = exc

    hint = (
        "Windows: 프로젝트의 vendor\\libusb-1.0.dll 을 "
        ".venv\\Scripts\\ 또는 python.exe 옆로 복사하세요."
        if _IS_WIN
        else "macOS: brew install libusb"
    )
    detail = f" ({last_err})" if last_err else ""
    raise RuntimeError(f"libusb-1.0 를 찾을 수 없습니다{detail}. {hint}")


def libusb_path() -> Optional[str]:
    try:
        _libusb_backend()
    except Exception:
        return None
    return _BACKEND_PATH


def _usb_err_text(exc: BaseException) -> str:
    parts = [str(exc)]
    strerror = getattr(exc, "strerror", None)
    if strerror and strerror not in parts[0]:
        parts.append(str(strerror))
    return " ".join(parts).lower()


def _is_access_denied(exc: BaseException) -> bool:
    text = _usb_err_text(exc)
    errno = getattr(exc, "errno", None)
    return (
        errno in (13, 5)
        or "access denied" in text
        or "permission" in text
        or "busy" in text
        or "resource busy" in text
    )


def _is_not_supported(exc: BaseException) -> bool:
    """WinUSB often returns LIBUSB_ERROR_NOT_SUPPORTED for get_configuration etc."""
    text = _usb_err_text(exc)
    errno = getattr(exc, "errno", None)
    return (
        errno in (95, 40)  # ENOTSUP / Windows ERROR_NOT_SUPPORTED-ish
        or "not supported" in text
        or "unimplemented" in text
        or "enosys" in text
    )


def _is_overflow(exc: BaseException) -> bool:
    text = _usb_err_text(exc)
    errno = getattr(exc, "errno", None)
    return errno == 84 or "overflow" in text


def _claim_failure_message(ifnum: int, exc: BaseException) -> str:
    base = f"USB interface {ifnum} 점유 실패: {exc}"
    if not _IS_WIN:
        return base
    if _is_access_denied(exc):
        return (
            f"{base}\n\n"
            "Windows가 Interface 5를 다른 드라이버(공식 Wormhole 등)로 붙잡고 있습니다.\n"
            "아래를 순서대로 하세요:\n"
            "1) GUI의 'Wormhole 종료' 버튼 또는 `python -m juc500_xfer kill-wormhole`\n"
            "2) Zadig(https://zadig.akeo.ie/) 실행 (관리자)\n"
            "3) Options → List All Devices\n"
            "4) 목록에서 'Smart Data Link (Interface 5)' 또는 InstanceId에 MI_05 가 있는 항목 선택\n"
            "5) Driver 를 WinUSB 로 바꾼 뒤 Replace Driver\n"
            "6) 케이블 뽑았다가 다시 꽂고 JUC500 재실행 → 연결\n"
            "장치 관리자에서 Interface 5가 'WinUSB' / 'libusb-winusb' 인지도 확인하세요."
        )
    if _is_not_supported(exc):
        return (
            f"{base}\n\n"
            "WinUSB가 해당 libusb API를 지원하지 않습니다. Interface 5(MI_05)에 "
            "WinUSB가 맞는지 Zadig로 확인하고, 최신 JUC500 빌드를 사용하세요."
        )
    return (
        f"{base}\n"
        f"Zadig에서 Smart Data Link → Interface {ifnum} (MI_0{ifnum}) 에 WinUSB를 설치하세요."
    )


@dataclass
class DeviceInfo:
    bus: int
    address: int
    serial: str
    product: str
    manufacturer: str
    vid: int = VID
    pid: int = PID


def _safe_str(dev: usb.core.Device, index: int, default: str = "") -> str:
    if not index:
        return default
    try:
        return (usb.util.get_string(dev, index) or default).strip()
    except Exception:
        return default


def _looks_like_juc500(dev: usb.core.Device, product: str) -> bool:
    if dev.idVendor != VID:
        return False
    if int(dev.idProduct) in ALT_PIDS or int(dev.idProduct) == PID:
        return True
    p = (product or "").lower()
    return any(h in p for h in _PRODUCT_HINTS)


def iter_raw_devices():
    backend = _libusb_backend()
    return usb.core.find(find_all=True, backend=backend)


def find_juc500_devices() -> list[usb.core.Device]:
    found: list[usb.core.Device] = []
    for dev in iter_raw_devices():
        product = _safe_str(dev, getattr(dev, "iProduct", 0), "")
        if _looks_like_juc500(dev, product):
            found.append(dev)
    # Prefer exact 0711:7500 first
    found.sort(key=lambda d: (0 if (d.idVendor == VID and d.idProduct == PID) else 1))
    return found


def list_devices() -> list[DeviceInfo]:
    out: list[DeviceInfo] = []
    for dev in find_juc500_devices():
        out.append(
            DeviceInfo(
                bus=int(dev.bus),
                address=int(dev.address),
                serial=_safe_str(dev, getattr(dev, "iSerialNumber", 0)),
                product=_safe_str(dev, getattr(dev, "iProduct", 0), "Smart Data Link")
                or "Smart Data Link",
                manufacturer=_safe_str(dev, getattr(dev, "iManufacturer", 0)),
                vid=int(dev.idVendor),
                pid=int(dev.idProduct),
            )
        )
    return out


def _windows_pnp_vid0711() -> list[dict[str, str]]:
    """Fallback: ask Windows PnP if VID_0711 is present (even without WinUSB)."""
    if not _IS_WIN:
        return []
    ps = (
        "Get-PnpDevice -PresentOnly | "
        "Where-Object { $_.InstanceId -match 'VID_0711' } | "
        "Select-Object Status, Class, FriendlyName, InstanceId | "
        "ConvertTo-Json -Compress"
    )
    try:
        proc = subprocess.run(
            ["powershell", "-NoProfile", "-Command", ps],
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
        )
        raw = (proc.stdout or "").strip()
        if not raw:
            return []
        import json

        data = json.loads(raw)
        if isinstance(data, dict):
            data = [data]
        return [
            {
                "status": str(d.get("Status") or ""),
                "class": str(d.get("Class") or ""),
                "name": str(d.get("FriendlyName") or ""),
                "id": str(d.get("InstanceId") or ""),
            }
            for d in data
        ]
    except Exception as exc:  # noqa: BLE001
        return [{"error": str(exc)}]


@dataclass
class DoctorReport:
    ok: bool
    libusb_path: Optional[str] = None
    libusb_error: Optional[str] = None
    usb_device_count: int = 0
    juc500: list[dict[str, Any]] = field(default_factory=list)
    sample_devices: list[str] = field(default_factory=list)
    pnp_vid0711: list[dict[str, str]] = field(default_factory=list)
    hints: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


def doctor() -> DoctorReport:
    report = DoctorReport(ok=False)
    try:
        be = _libusb_backend()
        report.libusb_path = _BACKEND_PATH
        all_devs = list(usb.core.find(find_all=True, backend=be))
        report.usb_device_count = len(all_devs)
        for d in all_devs[:25]:
            prod = _safe_str(d, getattr(d, "iProduct", 0), "")
            report.sample_devices.append(
                f"{d.idVendor:04x}:{d.idProduct:04x} bus={d.bus} addr={d.address} {prod!r}"
            )
        for info in list_devices():
            report.juc500.append(asdict(info))
    except Exception as exc:  # noqa: BLE001
        report.libusb_error = str(exc)
        report.hints.append(str(exc))

    report.pnp_vid0711 = _windows_pnp_vid0711()

    if report.juc500:
        report.ok = True
        report.hints.append("JUC500 장치가 libusb로 보입니다. 「연결」을 진행하세요.")
    else:
        if report.libusb_error:
            report.hints.append(
                "libusb DLL 문제일 수 있습니다. vendor\\libusb-1.0.dll 을 "
                ".venv\\Scripts\\ 에 복사한 뒤 다시 실행하세요."
            )
        elif report.usb_device_count == 0:
            report.hints.append(
                "libusb가 USB 장치를 하나도 보지 못합니다. "
                "다른 USB 장치도 안 보이면 DLL/권한 문제입니다."
            )
        elif report.pnp_vid0711 and not any("error" in x for x in report.pnp_vid0711):
            report.hints.append(
                "Windows는 VID_0711 케이블을 인식하지만 libusb가 JUC500을 못 엽니다. "
                "Zadig로 Smart Data Link → Interface 5(MI_05)에 WinUSB를 설치하세요. "
                "공식 Wormhole 드라이버가 있으면 제거하세요."
            )
        else:
            report.hints.append(
                "이 PC에서 JUC500(0711:7500)이 보이지 않습니다. "
                "케이블을 이 PC USB 포트에 직접 연결했는지, "
                "반대쪽은 상대 PC에 꽂혀 있는지 확인하세요. "
                "(양쪽 끝이 각각 다른 PC에 연결되어야 합니다.)"
            )
            if _IS_WIN:
                report.hints.append(
                    "장치 관리자에서 'Smart Data Link' / '알 수 없는 장치' / "
                    "WORMHOLE CD 롬이 있는지 확인하세요."
                )
    return report


class Juc500Link:
    """Claim Vendor (+ optional CDC) interfaces and provide framed bulk I/O."""

    def __init__(self, serial: Optional[str] = None, timeout_ms: int = 2000):
        self.timeout_ms = timeout_ms
        self._dev: Optional[usb.core.Device] = None
        self._claimed: list[int] = []
        self._serial = serial
        self.open()

    def open(self) -> None:
        matches = find_juc500_devices()
        if not matches:
            rep = doctor()
            hints = "\n".join(f"- {h}" for h in rep.hints)
            raise RuntimeError(
                "JUC500(Smart Data Link / 0711:7500) 장치를 찾지 못했습니다.\n" + hints
            )

        candidates = matches
        if self._serial:
            candidates = [
                d
                for d in matches
                if _safe_str(d, getattr(d, "iSerialNumber", 0)) == self._serial.strip()
            ]
            if not candidates:
                raise RuntimeError(f"시리얼 '{self._serial}' 에 해당하는 JUC500 이 없습니다.")

        self._dev = None
        last_err: Exception | None = None
        for chosen in candidates:
            try:
                self._try_open_device(chosen)
                break
            except Exception as exc:  # noqa: BLE001
                last_err = exc
                self._force_close()
        else:
            detail = f"\n마지막 오류: {last_err}" if last_err else ""
            raise RuntimeError(
                "JUC500 USB Interface 5 를 열 수 없습니다." + detail
            ) from last_err

        # Optional CDC — ignore failures (WinUSB often only on IF5)
        self._claim(IF_CDC_DATA, required=False)
        self._claim(IF_CDC_COMM, required=False)

    def _try_open_device(self, chosen: usb.core.Device) -> None:
        self._dev = chosen
        self._claimed = []
        if _IS_WIN:
            self._configure_windows()
        else:
            self._configure_posix()
        self._claim(IF_VENDOR, required=True)
        # Force backend path readiness (no Device.write get_configuration)
        self._ensure_backend_ready()

    def _force_close(self) -> None:
        try:
            self.close()
        except Exception:
            pass
        self._dev = None
        self._claimed = []

    def _configure_windows(self) -> None:
        """WinUSB: never call get_configuration / prefer skipping set_configuration.

        pyusb Device.read/write resolve endpoints via get_active_configuration(),
        which calls libusb_get_configuration() — NOT_SUPPORTED on WinUSB.
        We seed caches and perform all bulk I/O through the backend directly.
        """
        assert self._dev is not None
        # set_configuration is frequently NOT_SUPPORTED when the composite is
        # already configured by Windows — skip rather than confuse WinUSB.
        try:
            self._dev._ctx.managed_open()
        except Exception:
            pass
        self._ensure_cfg_cache(force=True)

    def _ensure_cfg_cache(self, *, force: bool = False) -> None:
        """Seed pyusb config index so leftover Device.* paths skip get_configuration."""
        if self._dev is None:
            return
        try:
            ctx = self._dev._ctx
            if force or getattr(ctx, "_active_cfg_index", None) is None:
                ctx._active_cfg_index = 0
        except Exception:
            pass

    def _ensure_backend_ready(self) -> None:
        assert self._dev is not None
        ctx = self._dev._ctx
        with ctx.lock:
            handle = ctx.managed_open()
            if IF_VENDOR not in ctx._claimed_intf:
                ctx.backend.claim_interface(handle, IF_VENDOR)
                ctx._claimed_intf.add(IF_VENDOR)
            if IF_VENDOR not in self._claimed:
                self._claimed.append(IF_VENDOR)
        self._ensure_cfg_cache(force=True)

    def _configure_posix(self) -> None:
        assert self._dev is not None
        try:
            cfg = self._dev.get_active_configuration()
        except usb.core.USBError:
            self._dev.set_configuration(1)
            cfg = self._dev.get_active_configuration()
        if cfg.bConfigurationValue != 1:
            self._dev.set_configuration(1)

    def _claim(self, ifnum: int, *, required: bool) -> None:
        assert self._dev is not None
        if not _IS_WIN:
            try:
                if self._dev.is_kernel_driver_active(ifnum):
                    self._dev.detach_kernel_driver(ifnum)
            except (NotImplementedError, usb.core.USBError):
                pass
        try:
            usb.util.claim_interface(self._dev, ifnum)
        except usb.core.USBError as exc:
            # Official Wormhole often holds IF5 — kill and retry once.
            if _is_access_denied(exc) and ifnum == IF_VENDOR:
                try:
                    from .wormhole_kill import kill_wormhole_processes

                    kill_wormhole_processes(force=True)
                    time.sleep(0.6)
                    if not _IS_WIN:
                        try:
                            if self._dev.is_kernel_driver_active(ifnum):
                                self._dev.detach_kernel_driver(ifnum)
                        except (NotImplementedError, usb.core.USBError):
                            pass
                    usb.util.claim_interface(self._dev, ifnum)
                    self._claimed.append(ifnum)
                    return
                except Exception:
                    pass
            if required:
                raise RuntimeError(_claim_failure_message(ifnum, exc)) from exc
            return
        self._claimed.append(ifnum)

    @property
    def serial(self) -> str:
        assert self._dev is not None
        return _safe_str(self._dev, getattr(self._dev, "iSerialNumber", 0))

    def write_data(self, payload: bytes, timeout_ms: Optional[int] = None) -> int:
        return self._bulk_write(EP_DATA_OUT, payload, timeout_ms)

    def read_data(self, size: int = MAX_PACKET, timeout_ms: Optional[int] = None) -> bytes:
        return self._bulk_read(EP_DATA_IN, size, timeout_ms)

    def write_ctrl(self, payload: bytes, timeout_ms: Optional[int] = None) -> int:
        return self._bulk_write(EP_CTRL_OUT, payload, timeout_ms)

    def read_ctrl(self, size: int = MAX_PACKET, timeout_ms: Optional[int] = None) -> bytes:
        return self._bulk_read(EP_CTRL_IN, size, timeout_ms)

    def _bulk_write(self, ep: int, payload: bytes, timeout_ms: Optional[int]) -> int:
        """Bulk OUT via libusb backend (skips pyusb get_configuration on WinUSB)."""
        assert self._dev is not None
        to = self.timeout_ms if timeout_ms is None else timeout_ms
        sent = 0
        view = memoryview(payload)
        while sent < len(payload):
            chunk = bytes(view[sent : sent + MAX_PACKET])
            try:
                n = self._backend_bulk_write(ep, chunk, to)
            except usb.core.USBTimeoutError:
                raise
            except usb.core.USBError as exc:
                if _is_not_supported(exc):
                    raise RuntimeError(
                        f"USB 쓰기 실패(EP 0x{ep:02x}): {exc}\n"
                        "Windows: Interface 5(MI_05)에 WinUSB가 잡혀 있는지 Zadig로 확인하세요.\n"
                        "최신 dist-windows 빌드인지, 공식 Wormhole가 종료됐는지도 확인하세요."
                    ) from exc
                raise
            sent += int(n)
            if n == 0:
                break
        return sent

    def _bulk_read(self, ep: int, size: int, timeout_ms: Optional[int]) -> bytes:
        """Bulk IN via libusb backend (skips pyusb get_configuration on WinUSB)."""
        assert self._dev is not None
        to = self.timeout_ms if timeout_ms is None else timeout_ms
        try:
            return self._backend_bulk_read(ep, size, to)
        except usb.core.USBTimeoutError:
            return b""
        except usb.core.USBError as exc:
            if _is_overflow(exc):
                for bigger in (max(size, 512) * 2, 4096, 16384, 65536):
                    if bigger <= size:
                        continue
                    try:
                        return self._backend_bulk_read(ep, bigger, to)
                    except usb.core.USBTimeoutError:
                        return b""
                    except usb.core.USBError as exc2:
                        if _is_overflow(exc2):
                            size = bigger
                            continue
                        if _is_not_supported(exc2):
                            raise RuntimeError(
                                f"USB 읽기 실패(EP 0x{ep:02x}): {exc2}\n"
                                "Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
                            ) from exc2
                        return b""
                return b""
            if _is_not_supported(exc):
                raise RuntimeError(
                    f"USB 읽기 실패(EP 0x{ep:02x}): {exc}\n"
                    "Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
                ) from exc
            return b""

    def _backend_bulk_write(self, ep: int, data: bytes, timeout_ms: int) -> int:
        assert self._dev is not None
        ctx = self._dev._ctx
        with ctx.lock:
            handle = ctx.managed_open()
            if IF_VENDOR not in ctx._claimed_intf:
                ctx.backend.claim_interface(handle, IF_VENDOR)
                ctx._claimed_intf.add(IF_VENDOR)
            buff = array.array("B", data)
            return int(ctx.backend.bulk_write(handle, ep, IF_VENDOR, buff, timeout_ms))

    def _backend_bulk_read(self, ep: int, size: int, timeout_ms: int) -> bytes:
        assert self._dev is not None
        ctx = self._dev._ctx
        with ctx.lock:
            handle = ctx.managed_open()
            if IF_VENDOR not in ctx._claimed_intf:
                ctx.backend.claim_interface(handle, IF_VENDOR)
                ctx._claimed_intf.add(IF_VENDOR)
            buff = usb.util.create_buffer(size)
            n = int(ctx.backend.bulk_read(handle, ep, IF_VENDOR, buff, timeout_ms))
            if n <= 0:
                return b""
            return bytes(buff[:n])

    def drain(self, duration_s: float = 0.2) -> None:
        deadline = time.monotonic() + duration_s
        while time.monotonic() < deadline:
            got = False
            for reader in (self.read_data, self.read_ctrl):
                try:
                    chunk = reader(MAX_PACKET, timeout_ms=50)
                except RuntimeError:
                    break
                if chunk:
                    got = True
            if not got:
                break

    def close(self) -> None:
        if self._dev is None:
            return
        for ifnum in list(self._claimed):
            try:
                usb.util.release_interface(self._dev, ifnum)
            except Exception:
                pass
        self._claimed.clear()
        try:
            usb.util.dispose_resources(self._dev)
        except Exception:
            pass
        self._dev = None

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

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