Add Windows installer packaging (Setup.exe + portable exe)
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>
This commit is contained in:
joyharam
@@ -0,0 +1,279 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user