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 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 _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) 작업 관리자에서 Wormhole / SmartDataLink / j5create 관련 프로그램 모두 종료\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' 인지도 확인하세요."
        )
    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
            )

        chosen = None
        for dev in matches:
            if self._serial:
                s = _safe_str(dev, getattr(dev, "iSerialNumber", 0))
                if s != self._serial.strip():
                    continue
            chosen = dev
            break
        if chosen is None:
            raise RuntimeError(f"시리얼 '{self._serial}' 에 해당하는 JUC500 이 없습니다.")

        self._dev = chosen

        if _IS_WIN:
            self._configure_windows()
        else:
            self._configure_posix()

        self._claim(IF_VENDOR, required=True)
        self._claim(IF_CDC_DATA, required=False)
        self._claim(IF_CDC_COMM, required=False)

    def _configure_windows(self) -> None:
        assert self._dev is not None
        try:
            self._dev.set_configuration(1)
        except usb.core.USBError:
            pass
        try:
            ctx = self._dev._ctx
            if getattr(ctx, "_active_cfg_index", None) is None:
                ctx._active_cfg_index = 0
        except Exception:
            pass

    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:
            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:
        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 = view[sent : sent + MAX_PACKET]
            try:
                n = self._dev.write(ep, chunk.tobytes(), timeout=to)
            except usb.core.USBError as exc:
                if _is_not_supported(exc):
                    raise RuntimeError(
                        f"USB 쓰기 실패(EP 0x{ep:02x}): {exc}\n"
                        "Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
                    ) from exc
                raise
            sent += n
            if n == 0:
                break
        return sent

    def _bulk_read(self, ep: int, size: int, timeout_ms: Optional[int]) -> bytes:
        assert self._dev is not None
        to = self.timeout_ms if timeout_ms is None else timeout_ms
        try:
            data = self._dev.read(ep, size, timeout=to)
            return bytes(data)
        except usb.core.USBTimeoutError:
            return b""
        except usb.core.USBError as exc:
            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 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()