aidev/juc500
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Bypass pyusb Device I/O on WinUSB with direct libusb bulk transfers

Browse Source 
Device.write/read always call get_configuration which WinUSB does not
implement. Route all framed I/O through the backend with IF5 claimed.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
joyharam
2026-07-15 11:29:01 +09:00
parent 7627c17653
commit ff60cee1cc
3 changed files with 114 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
juc500_xfer/protocol.py
+6 -1
View File
@@ -193,7 +193,12 @@ class Session:
try:
self.send(MsgType.HELLO, self.peer_name.encode("utf-8"))
except Exception as exc:
self.log(f"HELLO 재전송 실패: {exc}")
# Peer not reading yet (FIFO full) → USB timeout; keep retrying quietly.
msg = str(exc)
if "timed out" in msg.lower() or "timeout" in msg.lower():
self.log("상대 수신 대기 중… (HELLO)")
else:
self.log(f"HELLO 재전송 실패: {exc}")
self._connected.wait(timeout=min(2.0, remaining))
return self._peer_hello or "peer"
juc500_xfer/usb_link.py
+104 -60
View File
@@ -6,6 +6,7 @@ Device: VID 0x0711 / PID 0x7500 (product "Smart Data Link"), composite:
from __future__ import annotations
import array
import os
import subprocess
import sys
@@ -432,55 +433,93 @@ class Juc500Link:
"JUC500(Smart Data Link / 0711:7500) 장치를 찾지 못했습니다.\n" + hints
)
chosen = None
for dev in matches:
if self._serial:
s = _safe_str(dev, getattr(dev, "iSerialNumber", 0))
if s != self._serial.strip():
continue
chosen = dev
break
if chosen is None:
raise RuntimeError(f"시리얼 '{self._serial}' 에 해당하는 JUC500 이 없습니다.")
candidates = matches
if self._serial:
candidates = [
d
for d in matches
if _safe_str(d, getattr(d, "iSerialNumber", 0)) == self._serial.strip()
]
if not candidates:
raise RuntimeError(f"시리얼 '{self._serial}' 에 해당하는 JUC500 이 없습니다.")
self._dev = None
last_err: Exception | None = None
for chosen in candidates:
try:
self._try_open_device(chosen)
break
except Exception as exc: # noqa: BLE001
last_err = exc
self._force_close()
else:
detail = f"\n마지막 오류: {last_err}" if last_err else ""
raise RuntimeError(
"JUC500 USB Interface 5 를 열 수 없습니다." + detail
) from last_err
# Optional CDC — ignore failures (WinUSB often only on IF5)
self._claim(IF_CDC_DATA, required=False)
self._claim(IF_CDC_COMM, required=False)
def _try_open_device(self, chosen: usb.core.Device) -> None:
self._dev = chosen
self._claimed = []
if _IS_WIN:
self._configure_windows()
else:
self._configure_posix()
self._claim(IF_VENDOR, required=True)
self._claim(IF_CDC_DATA, required=False)
self._claim(IF_CDC_COMM, required=False)
# Force backend path readiness (no Device.write get_configuration)
self._ensure_backend_ready()
def _force_close(self) -> None:
try:
self.close()
except Exception:
pass
self._dev = None
self._claimed = []
def _configure_windows(self) -> None:
"""Avoid libusb_get_configuration — unimplemented on WinUSB.
"""WinUSB: never call get_configuration / prefer skipping set_configuration.
Every pyusb read/write resolves endpoints via get_active_configuration(),
which calls backend.get_configuration() when the cache is empty and then
raises 'Operation not supported or unimplemented on this platform'.
Always seed _active_cfg_index after open.
pyusb Device.read/write resolve endpoints via get_active_configuration(),
which calls libusb_get_configuration() — NOT_SUPPORTED on WinUSB.
We seed caches and perform all bulk I/O through the backend directly.
"""
assert self._dev is not None
# set_configuration is frequently NOT_SUPPORTED when the composite is
# already configured by Windows — skip rather than confuse WinUSB.
try:
self._dev.set_configuration(1)
except (usb.core.USBError, NotImplementedError):
self._dev._ctx.managed_open()
except Exception:
pass
self._ensure_cfg_cache(force=True)
def _ensure_cfg_cache(self, *, force: bool = False) -> None:
"""Seed pyusb config index so I/O never calls get_configuration on WinUSB."""
"""Seed pyusb config index so leftover Device.* paths skip get_configuration."""
if self._dev is None:
return
try:
ctx = self._dev._ctx
if force or getattr(ctx, "_active_cfg_index", None) is None:
# Composite JUC500 exposes a single configuration at index 0.
ctx._active_cfg_index = 0
except Exception:
pass
def _ensure_backend_ready(self) -> None:
assert self._dev is not None
ctx = self._dev._ctx
with ctx.lock:
handle = ctx.managed_open()
if IF_VENDOR not in ctx._claimed_intf:
ctx.backend.claim_interface(handle, IF_VENDOR)
ctx._claimed_intf.add(IF_VENDOR)
if IF_VENDOR not in self._claimed:
self._claimed.append(IF_VENDOR)
self._ensure_cfg_cache(force=True)
def _configure_posix(self) -> None:
assert self._dev is not None
try:
@@ -543,46 +582,36 @@ class Juc500Link:
return self._bulk_read(EP_CTRL_IN, size, timeout_ms)
def _bulk_write(self, ep: int, payload: bytes, timeout_ms: Optional[int]) -> int:
"""Bulk OUT via libusb backend (skips pyusb get_configuration on WinUSB)."""
assert self._dev is not None
if _IS_WIN:
self._ensure_cfg_cache()
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]
chunk = bytes(view[sent : sent + MAX_PACKET])
try:
n = self._dev.write(ep, chunk.tobytes(), timeout=to)
n = self._backend_bulk_write(ep, chunk, to)
except usb.core.USBTimeoutError:
raise
except usb.core.USBError as exc:
if _IS_WIN and _is_not_supported(exc):
self._ensure_cfg_cache(force=True)
try:
n = self._dev.write(ep, chunk.tobytes(), timeout=to)
except usb.core.USBError as exc2:
raise RuntimeError(
f"USB 쓰기 실패(EP 0x{ep:02x}): {exc2}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
) from exc2
elif _is_not_supported(exc):
if _is_not_supported(exc):
raise RuntimeError(
f"USB 쓰기 실패(EP 0x{ep:02x}): {exc}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
"Windows: Interface 5(MI_05)에 WinUSB가 잡혀 있는지 Zadig로 확인하세요.\n"
"최신 dist-windows 빌드인지, 공식 Wormhole가 종료됐는지도 확인하세요."
) from exc
else:
raise
sent += n
raise
sent += int(n)
if n == 0:
break
return sent
def _bulk_read(self, ep: int, size: int, timeout_ms: Optional[int]) -> bytes:
"""Bulk IN via libusb backend (skips pyusb get_configuration on WinUSB)."""
assert self._dev is not None
if _IS_WIN:
self._ensure_cfg_cache()
to = self.timeout_ms if timeout_ms is None else timeout_ms
try:
data = self._dev.read(ep, size, timeout=to)
return bytes(data)
return self._backend_bulk_read(ep, size, to)
except usb.core.USBTimeoutError:
return b""
except usb.core.USBError as exc:
@@ -591,8 +620,7 @@ class Juc500Link:
if bigger <= size:
continue
try:
data = self._dev.read(ep, bigger, timeout=to)
return bytes(data)
return self._backend_bulk_read(ep, bigger, to)
except usb.core.USBTimeoutError:
return b""
except usb.core.USBError as exc2:
@@ -600,28 +628,44 @@ class Juc500Link:
size = bigger
continue
if _is_not_supported(exc2):
break
raise RuntimeError(
f"USB 읽기 실패(EP 0x{ep:02x}): {exc2}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
) from exc2
return b""
return b""
if _is_not_supported(exc):
if _IS_WIN:
self._ensure_cfg_cache(force=True)
try:
data = self._dev.read(ep, size, timeout=to)
return bytes(data)
except usb.core.USBTimeoutError:
return b""
except usb.core.USBError as exc2:
raise RuntimeError(
f"USB 읽기 실패(EP 0x{ep:02x}): {exc2}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
) from exc2
raise RuntimeError(
f"USB 읽기 실패(EP 0x{ep:02x}): {exc}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
) from exc
return b""
def _backend_bulk_write(self, ep: int, data: bytes, timeout_ms: int) -> int:
assert self._dev is not None
ctx = self._dev._ctx
with ctx.lock:
handle = ctx.managed_open()
if IF_VENDOR not in ctx._claimed_intf:
ctx.backend.claim_interface(handle, IF_VENDOR)
ctx._claimed_intf.add(IF_VENDOR)
buff = array.array("B", data)
return int(ctx.backend.bulk_write(handle, ep, IF_VENDOR, buff, timeout_ms))
def _backend_bulk_read(self, ep: int, size: int, timeout_ms: int) -> bytes:
assert self._dev is not None
ctx = self._dev._ctx
with ctx.lock:
handle = ctx.managed_open()
if IF_VENDOR not in ctx._claimed_intf:
ctx.backend.claim_interface(handle, IF_VENDOR)
ctx._claimed_intf.add(IF_VENDOR)
buff = usb.util.create_buffer(size)
n = int(ctx.backend.bulk_read(handle, ep, IF_VENDOR, buff, timeout_ms))
if n <= 0:
return b""
return bytes(buff[:n])
def drain(self, duration_s: float = 0.2) -> None:
deadline = time.monotonic() + duration_s
while time.monotonic() < deadline:
packaging/windows/README-WIN.txt
+4 -2
View File
@@ -12,7 +12,9 @@ JUC500 — Windows 11 사용 안내
5) Mac 쪽에서도 같은 프로그램을 실행한 뒤 연결합니다.
오류 "Operation not supported or unimplemented on this platform"
→ 구버전 버그이거나 Interface 5 WinUSB 미설치. 최신 Setup/portable 로 업데이트하고
Zadig에서 Interface 5 가 WinUSB 인지 다시 확인하세요.
→ pyusb가 WinUSB에서 get_configuration 을 호출할 때 발생합니다.
2026-07-15 이후 빌드(backend 직접 전송)로 교체하세요.
그래도 같으면 Zadig에서 Interface 5(MI_05) = WinUSB 인지 확인하고
케이블을 뽑았다 다시 꽂은 뒤 JUC500 을 재실행하세요.
포터블: JUC500-portable.exe 는 설치 없이 실행할 수 있습니다.