joyharam
ec99c5eac6
Skip get_configuration/kernel-driver APIs on Windows, seed pyusb config cache, and require only Vendor interface 5 for transfers. Co-authored-by: Cursor <cursoragent@cursor.com>
354 lines
12 KiB
Python
354 lines
12 KiB
Python
"""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()
|