joyharam
8774a4843e
PyInstaller onedir/onefile specs, Inno Setup script, and GitHub Actions workflow to produce JUC500-Setup-win64.exe and portable builds. Co-authored-by: Cursor <cursoragent@cursor.com>
280 lines
9.1 KiB
Python
280 lines
9.1 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
|
||
|
||
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()
|