"""Low-level USB access to the j5create JUC500 Smart Data Link cable. The cable enumerates as USB VID:PID 0711:7500 with six interfaces: 0 CDC Communication (control notifications) 1 CDC Data (bulk 0x02 / 0x83) – proprietary control path 2 Mass Storage (WORMHOLE virtual CD, claimed by OS) 3 HID mouse (KM switch, claimed by OS) 4 HID keyboard (KM switch, claimed by OS) 5 Vendor Specific (bulk 0x08/0x89 + 0x0A/0x8B) – data bridge On Windows, WinUSB does not implement libusb_get_configuration / kernel-driver detach. Only Interface 5 (Vendor) is required for our protocol; CDC interfaces are optional. """ from __future__ import annotations import os import sys import time from dataclasses import dataclass from typing import Optional import usb.core import usb.util VID = 0x0711 PID = 0x7500 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 EP_CDC_OUT = 0x02 EP_CDC_IN = 0x83 MAX_PACKET = 512 _IS_WIN = sys.platform == "win32" def _libusb_backend(): """Locate libusb-1.0 on macOS / Windows / Linux.""" import usb.backend.libusb1 candidates = [] if getattr(sys, "frozen", False): mei = getattr(sys, "_MEIPASS", None) if mei: candidates += [ os.path.join(mei, "libusb-1.0.dll"), os.path.join(mei, "libusb-1.0.dylib"), os.path.join(mei, "libusb-1.0.so"), ] exe_dir = os.path.dirname(os.path.abspath(sys.executable)) candidates += [ os.path.join(exe_dir, "libusb-1.0.dll"), os.path.join(exe_dir, "_internal", "libusb-1.0.dll"), ] if 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", ] elif _IS_WIN: here = os.path.dirname(os.path.abspath(__file__)) root = os.path.dirname(here) candidates += [ os.path.join(root, "vendor", "libusb-1.0.dll"), os.path.join(os.environ.get("PROGRAMFILES", ""), "libusb", "libusb-1.0.dll"), "libusb-1.0.dll", ] else: candidates += ["libusb-1.0.so", "libusb-1.0.so.0"] for path in candidates: if not path: continue absolute = os.path.sep in path or (len(path) > 1 and path[1] == ":") if absolute and not os.path.exists(path): continue be = usb.backend.libusb1.get_backend(find_library=lambda _n, p=path: p) if be is not None: return be be = usb.backend.libusb1.get_backend() if be is None: raise RuntimeError( "libusb-1.0 를 찾을 수 없습니다. " "macOS: brew install libusb / Windows: vendor/libusb-1.0.dll 배치" ) return be 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_not_supported(exc: BaseException) -> bool: text = _usb_err_text(exc) return ( "not supported" in text or "unimplemented" in text or "enosys" in text ) @dataclass class DeviceInfo: bus: int address: int serial: str product: str manufacturer: str def list_devices() -> list[DeviceInfo]: backend = _libusb_backend() out: list[DeviceInfo] = [] for dev in usb.core.find(find_all=True, idVendor=VID, idProduct=PID, backend=backend): try: serial = usb.util.get_string(dev, dev.iSerialNumber) or "" except Exception: serial = "" try: product = usb.util.get_string(dev, dev.iProduct) or "Smart Data Link" except Exception: product = "Smart Data Link" try: manufacturer = usb.util.get_string(dev, dev.iManufacturer) or "" except Exception: manufacturer = "" out.append( DeviceInfo( bus=dev.bus, address=dev.address, serial=serial.strip(), product=product.strip(), manufacturer=manufacturer.strip(), ) ) return out 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: backend = _libusb_backend() matches = list( usb.core.find(find_all=True, idVendor=VID, idProduct=PID, backend=backend) ) if not matches: raise RuntimeError( "JUC500(0711:7500 Smart Data Link) 를 찾지 못했습니다. " "케이블이 연결되어 있는지 확인하세요." + ( " Windows: Zadig로 Interface 5에 WinUSB를 설치했는지 확인하세요." if _IS_WIN else "" ) ) chosen = None for dev in matches: if self._serial: try: s = (usb.util.get_string(dev, dev.iSerialNumber) or "").strip() except Exception: s = "" 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() # Protocol uses Vendor IF5 only. CDC is best-effort. 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: """WinUSB: libusb_get_configuration() is often unimplemented. pyusb caches the active config index after set_configuration(). If set fails (already configured), we seed the cache so later write()/read() do not call get_configuration() and raise 'Operation not supported or unimplemented on this platform'. """ assert self._dev is not None try: self._dev.set_configuration(1) except usb.core.USBError: pass # Ensure pyusb ResourceManager will not call get_configuration(). try: ctx = self._dev._ctx if getattr(ctx, "_active_cfg_index", None) is None: ctx._active_cfg_index = 0 # device has a single config 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: hint = "" if _IS_WIN: hint = ( f"\nZadig에서 Smart Data Link → Interface {ifnum} 에 " "WinUSB 드라이버를 설치하세요. " "(데이터 전송은 Interface 5가 필수입니다.)" ) raise RuntimeError(f"USB interface {ifnum} 점유 실패: {exc}{hint}") from exc return self._claimed.append(ifnum) @property def serial(self) -> str: assert self._dev is not None try: return (usb.util.get_string(self._dev, self._dev.iSerialNumber) or "").strip() except Exception: return "" 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 # Transient pipe errors — treat like empty during poll 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()