Fix missing _is_not_supported and harden WinUSB config cache

Restores the helper accidentally removed when adding access-denied
handling, retries I/O after seeding pyusb config index on Windows, and
handles USB overflow reads without crashing the HELLO loop.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
joyharam
2026-07-15 11:20:59 +09:00
parent d9c0b8078c
commit 7627c17653
4 changed files with 132 additions and 5 deletions
+85 -4
View File
@@ -194,6 +194,24 @@ def _is_access_denied(exc: BaseException) -> bool:
)
def _is_not_supported(exc: BaseException) -> bool:
"""WinUSB often returns LIBUSB_ERROR_NOT_SUPPORTED for get_configuration etc."""
text = _usb_err_text(exc)
errno = getattr(exc, "errno", None)
return (
errno in (95, 40) # ENOTSUP / Windows ERROR_NOT_SUPPORTED-ish
or "not supported" in text
or "unimplemented" in text
or "enosys" in text
)
def _is_overflow(exc: BaseException) -> bool:
text = _usb_err_text(exc)
errno = getattr(exc, "errno", None)
return errno == 84 or "overflow" in text
def _claim_failure_message(ifnum: int, exc: BaseException) -> str:
base = f"USB interface {ifnum} 점유 실패: {exc}"
if not _IS_WIN:
@@ -211,6 +229,12 @@ def _claim_failure_message(ifnum: int, exc: BaseException) -> str:
"6) 케이블 뽑았다가 다시 꽂고 JUC500 재실행 → 연결\n"
"장치 관리자에서 Interface 5가 'WinUSB' / 'libusb-winusb' 인지도 확인하세요."
)
if _is_not_supported(exc):
return (
f"{base}\n\n"
"WinUSB가 해당 libusb API를 지원하지 않습니다. Interface 5(MI_05)에 "
"WinUSB가 맞는지 Zadig로 확인하고, 최신 JUC500 빌드를 사용하세요."
)
return (
f"{base}\n"
f"Zadig에서 Smart Data Link → Interface {ifnum} (MI_0{ifnum}) 에 WinUSB를 설치하세요."
@@ -431,14 +455,28 @@ class Juc500Link:
self._claim(IF_CDC_COMM, required=False)
def _configure_windows(self) -> None:
"""Avoid libusb_get_configuration — unimplemented on WinUSB.
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.
"""
assert self._dev is not None
try:
self._dev.set_configuration(1)
except usb.core.USBError:
except (usb.core.USBError, NotImplementedError):
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."""
if self._dev is None:
return
try:
ctx = self._dev._ctx
if getattr(ctx, "_active_cfg_index", None) is None:
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
@@ -506,6 +544,8 @@ class Juc500Link:
def _bulk_write(self, ep: int, payload: bytes, timeout_ms: Optional[int]) -> int:
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)
@@ -514,12 +554,22 @@ class Juc500Link:
try:
n = self._dev.write(ep, chunk.tobytes(), timeout=to)
except usb.core.USBError as exc:
if _is_not_supported(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):
raise RuntimeError(
f"USB 쓰기 실패(EP 0x{ep:02x}): {exc}\n"
"Windows: Interface 5에 WinUSB가 잡혀 있는지 Zadig로 확인하세요."
) from exc
raise
else:
raise
sent += n
if n == 0:
break
@@ -527,6 +577,8 @@ class Juc500Link:
def _bulk_read(self, ep: int, size: int, timeout_ms: Optional[int]) -> bytes:
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)
@@ -534,7 +586,36 @@ class Juc500Link:
except usb.core.USBTimeoutError:
return b""
except usb.core.USBError as exc:
if _is_overflow(exc):
for bigger in (max(size, 512) * 2, 4096, 16384, 65536):
if bigger <= size:
continue
try:
data = self._dev.read(ep, bigger, timeout=to)
return bytes(data)
except usb.core.USBTimeoutError:
return b""
except usb.core.USBError as exc2:
if _is_overflow(exc2):
size = bigger
continue
if _is_not_supported(exc2):
break
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로 확인하세요."