"""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 Official Wormhole software is Intel/x86_64-only and uses the deprecated IOUSBDevice stack, which does not work reliably on Apple Silicon. This module opens interfaces 1 and 5 via libusb for a Silicon-native pathway. """ 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 # Interface numbers IF_CDC_COMM = 0 IF_CDC_DATA = 1 IF_VENDOR = 5 # Vendor interface endpoints (file payload) EP_DATA_OUT = 0x08 EP_DATA_IN = 0x89 # Vendor interface endpoints (keepalive / control mux) EP_CTRL_OUT = 0x0A EP_CTRL_IN = 0x8B # CDC data endpoints (alternate / reserved) EP_CDC_OUT = 0x02 EP_CDC_IN = 0x83 MAX_PACKET = 512 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 sys.platform == "win32": 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 path and (path.endswith((".dll", ".dylib", ".so", ".so.0")) or os.path.exists(path)): if path.endswith((".dll", ".dylib", ".so", ".so.0")) and not os.path.exists(path): # bare DLL name: allow loader PATH search via find_library if os.path.sep in path or (len(path) > 1 and path[1] == ":"): 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 @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 CDC+Vendor 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) 를 찾지 못했습니다. " "케이블이 연결되어 있는지 확인하세요." ) 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 # Avoid set_configuration() if the OS already activated config 1 # (mass-storage CD is mounted). Re-setting can disrupt the volume. try: cfg = self._dev.get_active_configuration() except usb.core.USBError: self._dev.set_configuration() cfg = self._dev.get_active_configuration() if cfg.bConfigurationValue != 1: self._dev.set_configuration(1) for ifnum in (IF_CDC_DATA, IF_VENDOR): self._claim(ifnum) # CDC Comm is optional (interrupt notifications) try: self._claim(IF_CDC_COMM) except usb.core.USBError: pass def _claim(self, ifnum: int) -> None: assert self._dev is not None 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) @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 # Split into max-packet chunks for bridge chips that dislike large URBs sent = 0 view = memoryview(payload) while sent < len(payload): chunk = view[sent : sent + MAX_PACKET] n = self._dev.write(ep, chunk.tobytes(), timeout=to) 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"" 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): chunk = reader(MAX_PACKET, timeout_ms=50) 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()