Macbook
1315 lines
44 KiB
Python
1315 lines
44 KiB
Python
"""Low-level USB access to j5create JUC500 cable (0711:7500 IF5)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import array
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import usb.core
|
|
import usb.util
|
|
|
|
VID = 0x0711
|
|
PID = 0x7500
|
|
PID_JUC500 = 0x7500
|
|
|
|
CABLE_MODEL_JUC500 = "JUC500"
|
|
DEFAULT_CABLE_MODEL = CABLE_MODEL_JUC500
|
|
|
|
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
|
|
|
|
MAX_PACKET = 512
|
|
# Bumped when USB connect / cable-switch logic changes (peer must restart to load).
|
|
XFER_BUILD = "2026-07-21-juc500-only"
|
|
# libusb splits into wMaxPacketSize; larger URBs cut Python→kernel round-trips.
|
|
BULK_XFER_SIZE = 64 * 1024
|
|
_IS_WIN = sys.platform == "win32"
|
|
# libusb is not reliably re-entrant across threads — serialize enum vs concurrent I/O callers.
|
|
_USB_ENUM_LOCK = threading.RLock()
|
|
_DEVICES_CACHE: list[Any] = []
|
|
_DEVICES_CACHE_AT = 0.0
|
|
_DEVICES_CACHE_TTL = 0.75
|
|
|
|
_PRODUCT_HINTS = (
|
|
"smart data",
|
|
"data link",
|
|
"wormhole",
|
|
"juc500",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CableLayout:
|
|
"""JUC500 IF5 bulk endpoint map."""
|
|
|
|
if_num: int
|
|
ep_data_out: int
|
|
ep_data_in: int
|
|
ep_ctrl_out: int
|
|
ep_ctrl_in: int
|
|
max_packet: int = 512
|
|
|
|
|
|
def default_if5_layout() -> CableLayout:
|
|
return CableLayout(
|
|
IF_VENDOR, EP_DATA_OUT, EP_DATA_IN, EP_CTRL_OUT, EP_CTRL_IN, MAX_PACKET
|
|
)
|
|
|
|
|
|
def _bulk_endpoints(intf) -> tuple[list[int], list[int], int]:
|
|
outs: list[int] = []
|
|
ins: list[int] = []
|
|
max_pkt = 512
|
|
for ep in intf:
|
|
if (ep.bmAttributes & 3) != 2:
|
|
continue
|
|
max_pkt = max(max_pkt, int(ep.wMaxPacketSize))
|
|
addr = int(ep.bEndpointAddress)
|
|
if addr & 0x80:
|
|
ins.append(addr)
|
|
else:
|
|
outs.append(addr)
|
|
return outs, ins, max_pkt
|
|
|
|
|
|
def _get_device_configuration(dev: usb.core.Device):
|
|
"""Read config descriptors; WinUSB often rejects get/set_configuration."""
|
|
try:
|
|
return dev.get_active_configuration()
|
|
except (usb.core.USBError, NotImplementedError):
|
|
try:
|
|
return dev[0]
|
|
except Exception:
|
|
pass
|
|
if _IS_WIN:
|
|
raise
|
|
try:
|
|
dev.set_configuration(1)
|
|
return dev.get_active_configuration()
|
|
except (usb.core.USBError, NotImplementedError) as exc:
|
|
raise RuntimeError("USB configuration 읽기 실패") from exc
|
|
|
|
|
|
def _is_juc500(vid: int, pid: int) -> bool:
|
|
return int(vid) == VID and int(pid) == PID_JUC500
|
|
|
|
|
|
def supports_file_transfer(info: DeviceInfo) -> bool:
|
|
"""True for JUC500 Smart Data Link IF5."""
|
|
return (
|
|
_is_juc500(int(getattr(info, "vid", VID) or VID), int(info.pid))
|
|
and int(getattr(info, "if_num", IF_VENDOR)) == IF_VENDOR
|
|
)
|
|
|
|
|
|
def device_supports_file_transfer(dev: usb.core.Device) -> bool:
|
|
return _is_juc500(int(dev.idVendor), int(dev.idProduct))
|
|
|
|
|
|
def detect_cable_layout(dev: usb.core.Device) -> CableLayout:
|
|
"""Map JUC500 IF5 bulk endpoints (0x08/0x89 + 0x0A/0x8B)."""
|
|
vid = int(dev.idVendor)
|
|
pid = int(dev.idProduct)
|
|
if not _is_juc500(vid, pid):
|
|
raise RuntimeError(
|
|
"JUC500(0711:7500) 장치가 아닙니다. "
|
|
"Smart Data Link 케이블인지 확인하세요."
|
|
)
|
|
|
|
try:
|
|
cfg = _get_device_configuration(dev)
|
|
except Exception:
|
|
if _IS_WIN:
|
|
return default_if5_layout()
|
|
raise RuntimeError(
|
|
"JUC500 IF5(0x08/0x89)를 찾지 못했습니다 — "
|
|
"케이블을 다시 연결하거나 다른 USB 포트를 사용하세요."
|
|
)
|
|
|
|
for intf in cfg:
|
|
ifnum = int(intf.bInterfaceNumber)
|
|
if ifnum != IF_VENDOR:
|
|
continue
|
|
outs, ins, max_pkt = _bulk_endpoints(intf)
|
|
if 0x08 in outs and 0x89 in ins:
|
|
ctrl_out = 0x0A if 0x0A in outs else outs[min(1, len(outs) - 1)]
|
|
ctrl_in = 0x8B if 0x8B in ins else ins[min(1, len(ins) - 1)]
|
|
return CableLayout(
|
|
IF_VENDOR, 0x08, 0x89, ctrl_out, ctrl_in, max(max_pkt, 512)
|
|
)
|
|
|
|
if _IS_WIN:
|
|
return default_if5_layout()
|
|
|
|
raise RuntimeError(
|
|
"JUC500 IF5(0x08/0x89)를 찾지 못했습니다 — "
|
|
"케이블을 다시 연결하거나 다른 USB 포트를 사용하세요."
|
|
)
|
|
|
|
|
|
def find_juc500_device(
|
|
devices: list[usb.core.Device],
|
|
*,
|
|
bus: Optional[int] = None,
|
|
address: Optional[int] = None,
|
|
serial: str = "",
|
|
) -> Optional[usb.core.Device]:
|
|
"""Pick JUC500 (0711:7500) by bus/address, serial, or first match."""
|
|
serial = (serial or "").strip()
|
|
candidates = [d for d in devices if _is_juc500(int(d.idVendor), int(d.idProduct))]
|
|
|
|
if bus is not None and address is not None:
|
|
for d in candidates:
|
|
if int(d.bus) == int(bus) and int(d.address) == int(address):
|
|
return d
|
|
|
|
if serial:
|
|
for d in candidates:
|
|
if _safe_str(d, getattr(d, "iSerialNumber", 0), "").strip() == serial:
|
|
return d
|
|
|
|
if bus is not None:
|
|
for d in candidates:
|
|
if int(d.bus) == int(bus):
|
|
return d
|
|
|
|
return candidates[0] if candidates else None
|
|
|
|
|
|
def data_link_devices(devices: list[DeviceInfo]) -> list[DeviceInfo]:
|
|
"""USB nodes usable for JUC500 IF5 file transfer."""
|
|
return [d for d in devices if supports_file_transfer(d)]
|
|
|
|
|
|
def available_cable_models(devices: list[DeviceInfo] | None = None) -> tuple[str, ...]:
|
|
"""Cable types to show in the UI — JUC500 only when present."""
|
|
if devices is None:
|
|
try:
|
|
devices = list_devices()
|
|
except Exception:
|
|
return ()
|
|
if data_link_devices(devices):
|
|
return (CABLE_MODEL_JUC500,)
|
|
return ()
|
|
|
|
|
|
def _juc500_winusb_hint() -> str:
|
|
return (
|
|
"JUC500 Mac↔Win: 공식 Wormhole 종료 → Zadig(관리자) → List All Devices → "
|
|
"0711:7500 (Smart Data Link) → Interface 5 (MI_05) → WinUSB → Replace Driver → "
|
|
"케이블 재연결 후 start.bat. Windows에서도 이 juc500 앱을 실행하세요."
|
|
)
|
|
|
|
|
|
def _package_roots() -> list[Path]:
|
|
here = Path(__file__).resolve().parent
|
|
roots = [here.parent, Path.cwd()]
|
|
if getattr(sys, "frozen", False):
|
|
roots.insert(0, Path(sys.executable).resolve().parent)
|
|
mei = getattr(sys, "_MEIPASS", None)
|
|
if mei:
|
|
roots.insert(0, Path(mei))
|
|
for base in list(roots):
|
|
p = base
|
|
for _ in range(4):
|
|
roots.append(p)
|
|
if (p / "vendor").is_dir():
|
|
break
|
|
p = p.parent
|
|
seen: set[str] = set()
|
|
out: list[Path] = []
|
|
for r in roots:
|
|
key = str(r)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
def _libusb_candidates() -> list[str]:
|
|
names = []
|
|
if _IS_WIN:
|
|
names = ["libusb-1.0.dll"]
|
|
elif sys.platform == "darwin":
|
|
names = ["libusb-1.0.dylib"]
|
|
else:
|
|
names = ["libusb-1.0.so", "libusb-1.0.so.0"]
|
|
|
|
candidates: list[str] = []
|
|
if getattr(sys, "frozen", False):
|
|
mei = getattr(sys, "_MEIPASS", None)
|
|
exe_dir = Path(sys.executable).resolve().parent
|
|
if mei:
|
|
for n in names:
|
|
candidates.append(str(Path(mei) / n))
|
|
for n in names:
|
|
candidates.append(str(exe_dir / n))
|
|
candidates.append(str(exe_dir / "_internal" / n))
|
|
|
|
for root in _package_roots():
|
|
for n in names:
|
|
candidates.append(str(root / "vendor" / n))
|
|
candidates.append(str(root / n))
|
|
|
|
if _IS_WIN:
|
|
py_dir = Path(sys.executable).resolve().parent
|
|
for n in names:
|
|
candidates.append(str(py_dir / n))
|
|
candidates.append(str(py_dir / "DLLs" / n))
|
|
pf = os.environ.get("PROGRAMFILES", r"C:\Program Files")
|
|
candidates.append(os.path.join(pf, "libusb", "libusb-1.0.dll"))
|
|
candidates.append("libusb-1.0.dll")
|
|
elif 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",
|
|
]
|
|
else:
|
|
candidates += ["libusb-1.0.so", "libusb-1.0.so.0"]
|
|
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for c in candidates:
|
|
if c and c not in seen:
|
|
seen.add(c)
|
|
out.append(c)
|
|
return out
|
|
|
|
|
|
_BACKEND = None
|
|
_BACKEND_PATH: Optional[str] = None
|
|
|
|
|
|
def _libusb_backend():
|
|
"""Locate / cache libusb-1.0 backend."""
|
|
global _BACKEND, _BACKEND_PATH
|
|
if _BACKEND is not None:
|
|
return _BACKEND
|
|
|
|
import usb.backend.libusb1
|
|
|
|
last_err: Optional[BaseException] = None
|
|
for path in _libusb_candidates():
|
|
absolute = os.path.sep in path or (len(path) > 1 and path[1] == ":")
|
|
if absolute and not os.path.exists(path):
|
|
continue
|
|
try:
|
|
be = usb.backend.libusb1.get_backend(find_library=lambda _n, p=path: p)
|
|
if be is not None:
|
|
list(usb.core.find(find_all=True, backend=be))
|
|
_BACKEND = be
|
|
_BACKEND_PATH = path
|
|
return be
|
|
except Exception as exc: # noqa: BLE001
|
|
last_err = exc
|
|
continue
|
|
|
|
try:
|
|
be = usb.backend.libusb1.get_backend()
|
|
if be is not None:
|
|
list(usb.core.find(find_all=True, backend=be))
|
|
_BACKEND = be
|
|
_BACKEND_PATH = "(system)"
|
|
return be
|
|
except Exception as exc: # noqa: BLE001
|
|
last_err = exc
|
|
|
|
hint = (
|
|
"Windows: 프로젝트의 vendor\\libusb-1.0.dll 을 "
|
|
".venv\\Scripts\\ 또는 python.exe 옆로 복사하세요."
|
|
if _IS_WIN
|
|
else "macOS: brew install libusb"
|
|
)
|
|
detail = f" ({last_err})" if last_err else ""
|
|
raise RuntimeError(f"libusb-1.0 를 찾을 수 없습니다{detail}. {hint}")
|
|
|
|
|
|
def libusb_path() -> Optional[str]:
|
|
try:
|
|
_libusb_backend()
|
|
except Exception:
|
|
return None
|
|
return _BACKEND_PATH
|
|
|
|
|
|
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_access_denied(exc: BaseException) -> bool:
|
|
text = _usb_err_text(exc)
|
|
errno = getattr(exc, "errno", None)
|
|
return (
|
|
errno in (13, 5)
|
|
or "access denied" in text
|
|
or "permission" in text
|
|
or "busy" in text
|
|
or "resource busy" in text
|
|
)
|
|
|
|
|
|
def _is_not_found(exc: BaseException) -> bool:
|
|
text = _usb_err_text(exc)
|
|
errno = getattr(exc, "errno", None)
|
|
return (
|
|
errno in (2, 19)
|
|
or "entity not found" in text
|
|
or "not found" in text
|
|
or "no device" in text
|
|
)
|
|
|
|
|
|
def _is_not_supported(exc: BaseException) -> bool:
|
|
text = _usb_err_text(exc)
|
|
errno = getattr(exc, "errno", None)
|
|
return (
|
|
errno in (95, 40)
|
|
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:
|
|
return base
|
|
if _is_access_denied(exc):
|
|
return (
|
|
f"{base}\n\n"
|
|
"Interface 5가 이미 다른 프로세스에 점유되어 있습니다.\n"
|
|
"가장 흔한 원인:\n"
|
|
" • JUC500 창/프로세스가 이미 실행 중 (start.bat 를 여러 번 실행)\n"
|
|
" • 공식 Wormhole / GoBridge 가 Interface 5 를 사용 중\n\n"
|
|
"해결:\n"
|
|
"1) 작업 관리자에서 python.exe / juc500_app 중복 종료 후 하나만 실행\n"
|
|
"2) GUI의 'Wormhole 종료' 또는 `python -m juc500_xfer kill-wormhole`\n"
|
|
"3) 그래도 안 되면 Zadig로 MI_05 → WinUSB 확인 후 케이블 재연결"
|
|
)
|
|
if _is_not_supported(exc):
|
|
return (
|
|
f"{base}\n\n"
|
|
"WinUSB가 해당 libusb API를 지원하지 않습니다. "
|
|
"Interface 5 (MI_05)에 WinUSB가 설치되어 있는지 Zadig로 확인하세요.\n"
|
|
f"{_juc500_winusb_hint()}"
|
|
)
|
|
return (
|
|
f"{base}\n"
|
|
"Zadig에서 Smart Data Link → Interface 5 (MI_05) 에 WinUSB를 설치하세요."
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DeviceInfo:
|
|
bus: int
|
|
address: int
|
|
serial: str
|
|
product: str
|
|
manufacturer: str
|
|
vid: int = VID
|
|
pid: int = PID
|
|
model: str = CABLE_MODEL_JUC500
|
|
if_num: int = IF_VENDOR
|
|
|
|
|
|
def _safe_str(dev: usb.core.Device, index: int, default: str = "") -> str:
|
|
if not index:
|
|
return default
|
|
try:
|
|
return (usb.util.get_string(dev, index) or default).strip()
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _looks_like_juc500(dev: usb.core.Device, product: str) -> bool:
|
|
if not _is_juc500(int(dev.idVendor), int(dev.idProduct)):
|
|
return False
|
|
p = (product or "").lower()
|
|
return any(h in p for h in _PRODUCT_HINTS) or int(dev.idProduct) == PID_JUC500
|
|
|
|
|
|
def iter_raw_devices():
|
|
backend = _libusb_backend()
|
|
return usb.core.find(find_all=True, backend=backend)
|
|
|
|
|
|
def find_juc500_devices() -> list[usb.core.Device]:
|
|
"""Enumerate JUC500 (0711:7500) candidates."""
|
|
with _USB_ENUM_LOCK:
|
|
found: list[usb.core.Device] = []
|
|
for dev in iter_raw_devices():
|
|
try:
|
|
vid = int(dev.idVendor)
|
|
pid = int(dev.idProduct)
|
|
except Exception:
|
|
continue
|
|
if _is_juc500(vid, pid):
|
|
found.append(dev)
|
|
continue
|
|
if vid != VID:
|
|
continue
|
|
product = _safe_str(dev, getattr(dev, "iProduct", 0), "")
|
|
if _looks_like_juc500(dev, product):
|
|
found.append(dev)
|
|
found.sort(key=lambda d: (int(d.bus), int(d.address)))
|
|
return found
|
|
|
|
|
|
def order_devices(
|
|
devices: list[DeviceInfo],
|
|
*,
|
|
serial: str | None = None,
|
|
model: str | None = None,
|
|
) -> list[DeviceInfo]:
|
|
"""Pick connection order when multiple JUC500 cables are plugged in."""
|
|
del model # JUC500-only; kwarg kept for API compatibility
|
|
if not devices:
|
|
return []
|
|
|
|
def rank(d: DeviceInfo) -> tuple[int, int, int]:
|
|
r_serial = 0 if serial and d.serial == serial.strip() else 1
|
|
return (r_serial, d.bus, d.address)
|
|
|
|
return sorted(devices, key=rank)
|
|
|
|
|
|
def list_devices(*, force: bool = False) -> list[DeviceInfo]:
|
|
global _DEVICES_CACHE, _DEVICES_CACHE_AT
|
|
now = time.monotonic()
|
|
if not force and _DEVICES_CACHE and (now - _DEVICES_CACHE_AT) < _DEVICES_CACHE_TTL:
|
|
return list(_DEVICES_CACHE)
|
|
got = _USB_ENUM_LOCK.acquire(timeout=1.5)
|
|
if not got:
|
|
return list(_DEVICES_CACHE)
|
|
try:
|
|
now = time.monotonic()
|
|
if not force and _DEVICES_CACHE and (now - _DEVICES_CACHE_AT) < _DEVICES_CACHE_TTL:
|
|
return list(_DEVICES_CACHE)
|
|
devices = _list_devices_unlocked()
|
|
_DEVICES_CACHE = devices
|
|
_DEVICES_CACHE_AT = time.monotonic()
|
|
return list(devices)
|
|
finally:
|
|
_USB_ENUM_LOCK.release()
|
|
|
|
|
|
def _list_devices_unlocked() -> list[DeviceInfo]:
|
|
out: list[DeviceInfo] = []
|
|
for dev in find_juc500_devices():
|
|
product = (
|
|
_safe_str(dev, getattr(dev, "iProduct", 0), "Smart Data Link")
|
|
or "Smart Data Link"
|
|
)
|
|
vid = int(dev.idVendor)
|
|
pid = int(dev.idProduct)
|
|
if_num = IF_VENDOR
|
|
if device_supports_file_transfer(dev):
|
|
try:
|
|
if_num = int(detect_cable_layout(dev).if_num)
|
|
except Exception:
|
|
pass
|
|
out.append(
|
|
DeviceInfo(
|
|
bus=int(dev.bus),
|
|
address=int(dev.address),
|
|
serial=_safe_str(dev, getattr(dev, "iSerialNumber", 0)),
|
|
product=product,
|
|
manufacturer=_safe_str(dev, getattr(dev, "iManufacturer", 0)),
|
|
vid=vid,
|
|
pid=pid,
|
|
model=CABLE_MODEL_JUC500,
|
|
if_num=if_num,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def _macos_ioreg_vid0711() -> list[dict[str, str]]:
|
|
"""Fallback: IORegistry may show JUC500 when libusb enumeration is empty."""
|
|
if _IS_WIN or sys.platform != "darwin":
|
|
return []
|
|
try:
|
|
proc = subprocess.run(
|
|
["ioreg", "-p", "IOUSB", "-l", "-w", "0"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
text = proc.stdout or ""
|
|
if not text:
|
|
return []
|
|
rows: list[dict[str, str]] = []
|
|
block: list[str] = []
|
|
for line in text.splitlines():
|
|
if line.startswith(" +-o ") or line.startswith(" | +-o "):
|
|
if block:
|
|
row = _parse_ioreg_usb_block(block)
|
|
if row:
|
|
rows.append(row)
|
|
block = [line]
|
|
elif block:
|
|
block.append(line)
|
|
if block:
|
|
row = _parse_ioreg_usb_block(block)
|
|
if row:
|
|
rows.append(row)
|
|
return rows
|
|
except Exception as exc: # noqa: BLE001
|
|
return [{"error": str(exc)}]
|
|
|
|
|
|
def _parse_ioreg_usb_block(lines: list[str]) -> dict[str, str] | None:
|
|
import re
|
|
|
|
body = "\n".join(lines)
|
|
vid_m = re.search(r'"idVendor"\s*=\s*(0x[0-9a-fA-F]+|\d+)', body)
|
|
if not vid_m:
|
|
return None
|
|
raw_vid = vid_m.group(1)
|
|
vid = int(raw_vid, 16) if raw_vid.lower().startswith("0x") else int(raw_vid)
|
|
if vid != VID:
|
|
return None
|
|
pid_m = re.search(r'"idProduct"\s*=\s*(0x[0-9a-fA-F]+|\d+)', body)
|
|
pid = 0
|
|
if pid_m:
|
|
raw_pid = pid_m.group(1)
|
|
pid = int(raw_pid, 16) if raw_pid.lower().startswith("0x") else int(raw_pid)
|
|
if pid != PID_JUC500:
|
|
return None
|
|
name_m = re.search(r'"USB Product Name"\s*=\s*"([^"]*)"', body)
|
|
name = name_m.group(1) if name_m else ""
|
|
loc_m = re.search(r'"locationID"\s*=\s*(0x[0-9a-fA-F]+|\d+)', body)
|
|
loc = loc_m.group(1) if loc_m else ""
|
|
return {
|
|
"vid": f"{vid:04x}",
|
|
"pid": f"{pid:04x}",
|
|
"name": name,
|
|
"location": str(loc),
|
|
}
|
|
|
|
|
|
def _windows_pnp_vid0711() -> list[dict[str, str]]:
|
|
"""Fallback: ask Windows PnP if VID_0711 is present (even without WinUSB)."""
|
|
if not _IS_WIN:
|
|
return []
|
|
ps = (
|
|
"Get-PnpDevice -PresentOnly | "
|
|
"Where-Object { $_.InstanceId -match 'VID_0711' } | "
|
|
"Select-Object Status, Class, FriendlyName, InstanceId | "
|
|
"ConvertTo-Json -Compress"
|
|
)
|
|
try:
|
|
proc = subprocess.run(
|
|
["powershell", "-NoProfile", "-Command", ps],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
raw = (proc.stdout or "").strip()
|
|
if not raw:
|
|
return []
|
|
import json
|
|
|
|
data = json.loads(raw)
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
return [
|
|
{
|
|
"status": str(d.get("Status") or ""),
|
|
"class": str(d.get("Class") or ""),
|
|
"name": str(d.get("FriendlyName") or ""),
|
|
"id": str(d.get("InstanceId") or ""),
|
|
}
|
|
for d in data
|
|
]
|
|
except Exception as exc: # noqa: BLE001
|
|
return [{"error": str(exc)}]
|
|
|
|
|
|
@dataclass
|
|
class DoctorReport:
|
|
ok: bool
|
|
libusb_path: Optional[str] = None
|
|
libusb_error: Optional[str] = None
|
|
usb_device_count: int = 0
|
|
juc500: list[dict[str, Any]] = field(default_factory=list)
|
|
sample_devices: list[str] = field(default_factory=list)
|
|
pnp_vid0711: list[dict[str, str]] = field(default_factory=list)
|
|
ioreg_vid0711: list[dict[str, str]] = field(default_factory=list)
|
|
hints: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def doctor() -> DoctorReport:
|
|
report = DoctorReport(ok=False)
|
|
try:
|
|
be = _libusb_backend()
|
|
report.libusb_path = _BACKEND_PATH
|
|
all_devs = list(usb.core.find(find_all=True, backend=be))
|
|
report.usb_device_count = len(all_devs)
|
|
for d in all_devs[:25]:
|
|
prod = _safe_str(d, getattr(d, "iProduct", 0), "")
|
|
report.sample_devices.append(
|
|
f"{d.idVendor:04x}:{d.idProduct:04x} bus={d.bus} addr={d.address} {prod!r}"
|
|
)
|
|
for info in list_devices():
|
|
report.juc500.append(asdict(info))
|
|
except Exception as exc: # noqa: BLE001
|
|
report.libusb_error = str(exc)
|
|
report.hints.append(str(exc))
|
|
|
|
report.pnp_vid0711 = _windows_pnp_vid0711()
|
|
report.ioreg_vid0711 = _macos_ioreg_vid0711()
|
|
|
|
if report.juc500:
|
|
report.ok = True
|
|
if _IS_WIN:
|
|
report.hints.append(_juc500_winusb_hint())
|
|
report.hints.append(
|
|
"JUC500 케이블이 libusb로 보입니다. GUI에서 「연결」 또는 자동 연결을 기다리세요. "
|
|
"(양쪽 PC 모두 IF5 Smart Data Link)"
|
|
)
|
|
else:
|
|
if report.libusb_error:
|
|
report.hints.append(
|
|
"libusb DLL 문제일 수 있습니다. vendor\\libusb-1.0.dll 을 "
|
|
".venv\\Scripts\\ 에 복사한 뒤 다시 실행하세요."
|
|
)
|
|
elif report.usb_device_count == 0:
|
|
report.hints.append(
|
|
"libusb가 USB 장치를 하나도 보지 못합니다. "
|
|
"다른 USB 장치도 안 보이면 DLL/권한 문제입니다."
|
|
)
|
|
elif report.pnp_vid0711 and not any("error" in x for x in report.pnp_vid0711):
|
|
report.hints.append(
|
|
"Windows는 VID_0711 케이블을 인식하지만 libusb가 열지 못합니다. "
|
|
"Zadig로 WinUSB 설치: Smart Data Link → Interface 5(MI_05). "
|
|
"공식 Wormhole 드라이버가 있으면 제거하세요."
|
|
)
|
|
elif report.ioreg_vid0711 and not any("error" in x for x in report.ioreg_vid0711):
|
|
names = ", ".join(
|
|
f"{x.get('vid', '?')}:{x.get('pid', '?')} {x.get('name', '')}".strip()
|
|
for x in report.ioreg_vid0711[:3]
|
|
)
|
|
report.hints.append(
|
|
f"macOS IORegistry에는 JUC500 케이블이 보입니다 ({names}) — "
|
|
"libusb만 못 찾는 상태입니다. USB 허브/포트를 바꾸고 케이블을 다시 연결한 뒤 "
|
|
"GUI를 재시작하세요."
|
|
)
|
|
else:
|
|
report.hints.append(
|
|
"이 PC에서 JUC500(0711:7500)이 보이지 않습니다. "
|
|
"케이블을 이 PC USB 포트에 직접 연결했는지, "
|
|
"반대쪽은 상대 PC에 꽂혀 있는지 확인하세요. "
|
|
"(양쪽 끝이 각각 다른 PC에 연결되어야 합니다.)"
|
|
)
|
|
if _IS_WIN:
|
|
report.hints.append(
|
|
"장치 관리자에서 'Smart Data Link' / '알 수 없는 장치' / "
|
|
"WORMHOLE CD 롬이 있는지 확인하세요."
|
|
)
|
|
report.hints.append(_juc500_winusb_hint())
|
|
return report
|
|
|
|
|
|
def _backend_claim_interface(
|
|
dev: usb.core.Device,
|
|
*,
|
|
ifn: int,
|
|
claimed: list[int],
|
|
required_ifn: int | None = None,
|
|
handle: int | None = None,
|
|
) -> None:
|
|
"""Claim IF for bulk I/O. WinUSB per-interface (Zadig MI_05) may return NOT_SUPPORTED."""
|
|
ctx = dev._ctx
|
|
|
|
def _do(h: int) -> None:
|
|
if ifn in ctx._claimed_intf:
|
|
if ifn not in claimed:
|
|
claimed.append(ifn)
|
|
return
|
|
try:
|
|
ctx.backend.claim_interface(h, ifn)
|
|
ctx._claimed_intf.add(ifn)
|
|
except Exception as exc:
|
|
if _IS_WIN and (
|
|
_is_not_supported(exc) or isinstance(exc, NotImplementedError)
|
|
):
|
|
ctx._claimed_intf.add(ifn)
|
|
elif _is_access_denied(exc) and (
|
|
required_ifn is None or ifn == required_ifn
|
|
):
|
|
raise
|
|
else:
|
|
raise RuntimeError(_claim_failure_message(ifn, exc)) from exc
|
|
if ifn not in claimed:
|
|
claimed.append(ifn)
|
|
|
|
if handle is not None:
|
|
_do(handle)
|
|
return
|
|
with ctx.lock:
|
|
_do(ctx.managed_open())
|
|
|
|
|
|
class Juc500Link:
|
|
"""Claim Vendor bulk IF5 and provide framed bulk I/O for JUC500."""
|
|
|
|
def __init__(
|
|
self,
|
|
serial: Optional[str] = None,
|
|
*,
|
|
bus: Optional[int] = None,
|
|
address: Optional[int] = None,
|
|
model: Optional[str] = None,
|
|
timeout_ms: int = 2000,
|
|
):
|
|
del model # JUC500-only; kwarg kept for API compatibility
|
|
self.timeout_ms = timeout_ms
|
|
self._dev: Optional[usb.core.Device] = None
|
|
self._claimed: list[int] = []
|
|
self._serial = serial
|
|
self._pick_bus = bus
|
|
self._pick_address = address
|
|
self._layout = default_if5_layout()
|
|
self._rx_stash = bytearray()
|
|
self._rx_stash_lock = threading.Lock()
|
|
self.open()
|
|
|
|
def open(self) -> None:
|
|
matches = find_juc500_devices()
|
|
if not matches:
|
|
rep = doctor()
|
|
hints = "\n".join(f"- {h}" for h in rep.hints)
|
|
raise RuntimeError(
|
|
"JUC500(0711:7500 Smart Data Link) 장치를 찾지 못했습니다.\n" + hints
|
|
)
|
|
|
|
chosen_dev = find_juc500_device(
|
|
matches,
|
|
bus=self._pick_bus,
|
|
address=self._pick_address,
|
|
serial=(self._serial or "").strip(),
|
|
)
|
|
if chosen_dev is None:
|
|
raise RuntimeError(
|
|
"JUC500(0711:7500) 파일 전송용 USB 장치를 찾지 못했습니다.\n"
|
|
+ (_juc500_winusb_hint() if _IS_WIN else "케이블을 다시 연결하세요.")
|
|
)
|
|
|
|
if self._pick_bus is not None and self._pick_address is not None:
|
|
if (
|
|
int(chosen_dev.bus) != int(self._pick_bus)
|
|
or int(chosen_dev.address) != int(self._pick_address)
|
|
):
|
|
self._pick_bus = int(chosen_dev.bus)
|
|
self._pick_address = int(chosen_dev.address)
|
|
|
|
candidates: list[usb.core.Device] = [chosen_dev]
|
|
if self._serial:
|
|
serial_matches = [
|
|
d
|
|
for d in matches
|
|
if _safe_str(d, getattr(d, "iSerialNumber", 0)) == self._serial.strip()
|
|
and device_supports_file_transfer(d)
|
|
]
|
|
if serial_matches:
|
|
candidates = serial_matches
|
|
elif (
|
|
_safe_str(chosen_dev, getattr(chosen_dev, "iSerialNumber", 0))
|
|
!= self._serial.strip()
|
|
):
|
|
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 ""
|
|
if last_err and (
|
|
_is_not_supported(last_err)
|
|
or "not supported" in str(last_err).lower()
|
|
or "unimplemented" in str(last_err).lower()
|
|
):
|
|
if _IS_WIN:
|
|
detail += (
|
|
"\n\nZadig → List All Devices → 0711:7500 → "
|
|
"Interface 5 (MI_05) → WinUSB → Replace Driver"
|
|
)
|
|
raise RuntimeError(
|
|
f"JUC500 USB Interface {self._layout.if_num} 를 열 수 없습니다."
|
|
+ detail
|
|
) from last_err
|
|
|
|
if not _IS_WIN and self._layout.if_num == IF_VENDOR:
|
|
self._claim(IF_CDC_COMM, required=False)
|
|
self._claim(IF_CDC_DATA, required=False)
|
|
|
|
def _try_open_device(self, chosen: usb.core.Device) -> None:
|
|
self._dev = chosen
|
|
self._claimed = []
|
|
self._layout = detect_cable_layout(chosen)
|
|
if _IS_WIN:
|
|
self._configure_windows()
|
|
else:
|
|
self._configure_posix()
|
|
self._claim(self._layout.if_num, required=True)
|
|
self._ensure_backend_ready()
|
|
self._probe_link()
|
|
|
|
def wake_bridge(self) -> None:
|
|
"""Best-effort ctrl/data poke so the peer host starts draining IN."""
|
|
try:
|
|
self.write_ctrl(b"\x00ALIVE", timeout_ms=120)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
self.read_ctrl(MAX_PACKET, timeout_ms=60)
|
|
except Exception:
|
|
pass
|
|
|
|
def _probe_link(self) -> None:
|
|
"""Soft-wake IF5 — bulk OUT often stalls until peer claims IN / WinUSB is bound."""
|
|
self.wake_bridge()
|
|
|
|
def _reopen_device(self) -> None:
|
|
"""Re-enumerate after hotplug / Entity not found on bulk I/O."""
|
|
if self._dev is None:
|
|
raise RuntimeError("USB 장치 없음")
|
|
bus = int(self._dev.bus)
|
|
addr = int(self._dev.address)
|
|
serial = _safe_str(self._dev, getattr(self._dev, "iSerialNumber", 0))
|
|
self._force_close()
|
|
matches = find_juc500_devices()
|
|
if not matches:
|
|
raise RuntimeError("USB 장치를 다시 찾지 못했습니다. 케이블을 확인하세요.")
|
|
chosen = find_juc500_device(
|
|
matches,
|
|
bus=bus,
|
|
address=addr,
|
|
serial=serial,
|
|
)
|
|
if chosen is None:
|
|
seed = find_juc500_device(matches, serial=serial)
|
|
if seed is None:
|
|
seed = next((x for x in matches if int(x.bus) == bus), matches[0])
|
|
for d in [seed] + [x for x in matches if x is not seed]:
|
|
try:
|
|
self._try_open_device(d)
|
|
return
|
|
except Exception:
|
|
self._force_close()
|
|
raise RuntimeError(
|
|
"USB 장치 주소가 바뀌었습니다. GUI를 재시작하거나 케이블을 다시 연결하세요."
|
|
)
|
|
self._try_open_device(chosen)
|
|
|
|
def _force_close(self) -> None:
|
|
try:
|
|
self.close()
|
|
except Exception:
|
|
pass
|
|
self._dev = None
|
|
self._claimed = []
|
|
|
|
def _configure_windows(self) -> None:
|
|
assert self._dev is not None
|
|
try:
|
|
self._dev._ctx.managed_open()
|
|
except Exception:
|
|
pass
|
|
self._ensure_cfg_cache(force=True)
|
|
|
|
def _ensure_cfg_cache(self, *, force: bool = False) -> None:
|
|
if self._dev is None:
|
|
return
|
|
try:
|
|
ctx = self._dev._ctx
|
|
if force or getattr(ctx, "_active_cfg_index", None) is None:
|
|
ctx._active_cfg_index = 0
|
|
except Exception:
|
|
pass
|
|
|
|
def _ensure_backend_ready(self) -> None:
|
|
assert self._dev is not None
|
|
ifn = self._layout.if_num
|
|
try:
|
|
try:
|
|
self._dev._ctx.managed_open()
|
|
except Exception as open_exc:
|
|
if not (
|
|
_IS_WIN
|
|
and (
|
|
_is_not_supported(open_exc)
|
|
or isinstance(open_exc, NotImplementedError)
|
|
)
|
|
):
|
|
raise
|
|
_backend_claim_interface(
|
|
self._dev,
|
|
ifn=ifn,
|
|
claimed=self._claimed,
|
|
required_ifn=ifn,
|
|
)
|
|
except Exception as exc:
|
|
if _IS_WIN and (
|
|
_is_not_supported(exc) or isinstance(exc, NotImplementedError)
|
|
):
|
|
if ifn not in self._claimed:
|
|
self._claimed.append(ifn)
|
|
self._ensure_cfg_cache(force=True)
|
|
return
|
|
if _is_access_denied(exc) and ifn == self._layout.if_num:
|
|
try:
|
|
from .wormhole_kill import kill_wormhole_processes
|
|
|
|
kill_wormhole_processes(force=True)
|
|
time.sleep(0.6)
|
|
_backend_claim_interface(
|
|
self._dev,
|
|
ifn=ifn,
|
|
claimed=self._claimed,
|
|
required_ifn=ifn,
|
|
)
|
|
self._ensure_cfg_cache(force=True)
|
|
return
|
|
except Exception:
|
|
pass
|
|
if isinstance(exc, RuntimeError):
|
|
raise
|
|
raise RuntimeError(_claim_failure_message(ifn, exc)) from exc
|
|
self._ensure_cfg_cache(force=True)
|
|
|
|
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, NotImplementedError) as exc:
|
|
if not required and (
|
|
_is_not_supported(exc) or isinstance(exc, NotImplementedError)
|
|
):
|
|
return
|
|
if _is_access_denied(exc) and ifnum == self._layout.if_num:
|
|
try:
|
|
from .wormhole_kill import kill_wormhole_processes
|
|
|
|
kill_wormhole_processes(force=True)
|
|
time.sleep(0.6)
|
|
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
|
|
usb.util.claim_interface(self._dev, ifnum)
|
|
self._claimed.append(ifnum)
|
|
return
|
|
except Exception:
|
|
pass
|
|
if required:
|
|
raise RuntimeError(_claim_failure_message(ifnum, exc)) from exc
|
|
return
|
|
self._claimed.append(ifnum)
|
|
|
|
@property
|
|
def serial(self) -> str:
|
|
assert self._dev is not None
|
|
return _safe_str(self._dev, getattr(self._dev, "iSerialNumber", 0))
|
|
|
|
@property
|
|
def product(self) -> str:
|
|
assert self._dev is not None
|
|
return _safe_str(self._dev, getattr(self._dev, "iProduct", 0), "Smart Data Link")
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
return CABLE_MODEL_JUC500
|
|
|
|
@property
|
|
def vid_pid(self) -> tuple[int, int]:
|
|
assert self._dev is not None
|
|
return int(self._dev.idVendor), int(self._dev.idProduct)
|
|
|
|
@property
|
|
def layout(self) -> CableLayout:
|
|
return self._layout
|
|
|
|
def write_data(
|
|
self,
|
|
payload: bytes,
|
|
timeout_ms: Optional[int] = None,
|
|
*,
|
|
broadcast: bool = False,
|
|
) -> int:
|
|
del broadcast # JUC500-only; kwarg kept for API compatibility
|
|
if self._dev is None:
|
|
return 0
|
|
return self._bulk_write(self._layout.ep_data_out, payload, timeout_ms)
|
|
|
|
def read_data(self, size: int = MAX_PACKET, timeout_ms: Optional[int] = None) -> bytes:
|
|
if self._dev is None:
|
|
return b""
|
|
if size == MAX_PACKET:
|
|
size = self._layout.max_packet
|
|
with self._rx_stash_lock:
|
|
if self._rx_stash:
|
|
n = min(size, len(self._rx_stash))
|
|
out = bytes(self._rx_stash[:n])
|
|
del self._rx_stash[:n]
|
|
return out
|
|
return self._bulk_read(self._layout.ep_data_in, size, timeout_ms)
|
|
|
|
def _stash_rx(self, data: bytes) -> None:
|
|
if not data:
|
|
return
|
|
with self._rx_stash_lock:
|
|
self._rx_stash.extend(data)
|
|
if len(self._rx_stash) > 8 * 1024 * 1024:
|
|
del self._rx_stash[:-4 * 1024 * 1024]
|
|
|
|
def write_ctrl(self, payload: bytes, timeout_ms: Optional[int] = None) -> int:
|
|
return self._bulk_write(self._layout.ep_ctrl_out, payload, timeout_ms)
|
|
|
|
def read_ctrl(self, size: int = MAX_PACKET, timeout_ms: Optional[int] = None) -> bytes:
|
|
if size == MAX_PACKET:
|
|
size = self._layout.max_packet
|
|
return self._bulk_read(self._layout.ep_ctrl_in, size, timeout_ms)
|
|
|
|
def _bulk_read(self, ep: int, size: int, timeout_ms: Optional[int]) -> bytes:
|
|
if self._dev is None:
|
|
return b""
|
|
to = self.timeout_ms if timeout_ms is None else timeout_ms
|
|
try:
|
|
return self._backend_bulk_read(ep, size, to)
|
|
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:
|
|
return self._backend_bulk_read(ep, bigger, to)
|
|
except usb.core.USBTimeoutError:
|
|
return b""
|
|
except usb.core.USBError as exc2:
|
|
if _is_overflow(exc2):
|
|
size = bigger
|
|
continue
|
|
if _is_not_supported(exc2):
|
|
raise RuntimeError(
|
|
f"USB 읽기 실패(EP 0x{ep:02x}): {exc2}\n"
|
|
"Windows: Interface 5(MI_05)에 WinUSB가 잡혀 있는지 "
|
|
"Zadig로 확인하세요."
|
|
) from exc2
|
|
return b""
|
|
return b""
|
|
if _is_not_supported(exc):
|
|
raise RuntimeError(
|
|
f"USB 읽기 실패(EP 0x{ep:02x}): {exc}\n"
|
|
"Windows: Interface 5(MI_05)에 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
|
|
ifn = self._layout.if_num
|
|
ctx = self._dev._ctx
|
|
with ctx.lock:
|
|
handle = ctx.managed_open()
|
|
_backend_claim_interface(
|
|
self._dev,
|
|
ifn=ifn,
|
|
claimed=self._claimed,
|
|
required_ifn=ifn,
|
|
handle=handle,
|
|
)
|
|
backend = ctx.backend
|
|
buff = array.array("B", data)
|
|
return int(backend.bulk_write(handle, ep, ifn, buff, timeout_ms))
|
|
|
|
def _backend_bulk_read(self, ep: int, size: int, timeout_ms: int) -> bytes:
|
|
assert self._dev is not None
|
|
ifn = self._layout.if_num
|
|
ctx = self._dev._ctx
|
|
with ctx.lock:
|
|
handle = ctx.managed_open()
|
|
_backend_claim_interface(
|
|
self._dev,
|
|
ifn=ifn,
|
|
claimed=self._claimed,
|
|
required_ifn=ifn,
|
|
handle=handle,
|
|
)
|
|
backend = ctx.backend
|
|
buff = usb.util.create_buffer(size)
|
|
n = int(backend.bulk_read(handle, ep, ifn, buff, timeout_ms))
|
|
if n <= 0:
|
|
return b""
|
|
return bytes(buff[:n])
|
|
|
|
def clear_data_halt(self) -> None:
|
|
if self._dev is None:
|
|
return
|
|
try:
|
|
ctx = self._dev._ctx
|
|
with ctx.lock:
|
|
handle = ctx.managed_open()
|
|
backend = ctx.backend
|
|
for ep in (
|
|
self._layout.ep_data_out,
|
|
self._layout.ep_data_in,
|
|
self._layout.ep_ctrl_out,
|
|
self._layout.ep_ctrl_in,
|
|
):
|
|
try:
|
|
backend.clear_halt(handle, ep)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
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
|
|
mp = self._layout.max_packet
|
|
if len(payload) > mp:
|
|
to = max(to, min(15000, 2000 + (len(payload) // mp) * 4))
|
|
sent = 0
|
|
view = memoryview(payload)
|
|
while sent < len(payload):
|
|
chunk = bytes(view[sent : sent + BULK_XFER_SIZE])
|
|
try:
|
|
n = self._backend_bulk_write(ep, chunk, to)
|
|
except usb.core.USBTimeoutError:
|
|
self.clear_data_halt()
|
|
if ep == self._layout.ep_data_out:
|
|
for _ in range(16):
|
|
try:
|
|
got = self._backend_bulk_read(
|
|
self._layout.ep_data_in, BULK_XFER_SIZE, 40
|
|
)
|
|
except Exception:
|
|
break
|
|
if not got:
|
|
break
|
|
self._stash_rx(got)
|
|
try:
|
|
n = self._backend_bulk_write(ep, chunk, max(to, 3000))
|
|
except usb.core.USBTimeoutError:
|
|
n = 0
|
|
off = 0
|
|
while off < len(chunk):
|
|
piece = chunk[off : off + mp]
|
|
n += self._backend_bulk_write(ep, piece, max(to, 1500))
|
|
off += len(piece)
|
|
except usb.core.USBError as exc:
|
|
if _is_not_found(exc):
|
|
try:
|
|
self._reopen_device()
|
|
n = self._backend_bulk_write(ep, chunk, to)
|
|
except Exception as exc2:
|
|
raise RuntimeError(
|
|
f"USB 쓰기 실패(EP 0x{ep:02x}): Entity not found — "
|
|
"케이블·IF5 WinUSB·상대 GUI 실행 상태를 확인하세요."
|
|
) from exc2
|
|
elif _is_not_supported(exc):
|
|
raise RuntimeError(
|
|
f"USB 쓰기 실패(EP 0x{ep:02x}): {exc}\n"
|
|
"Windows: Interface 5(MI_05)에 WinUSB가 잡혀 있는지 Zadig로 확인하세요.\n"
|
|
"최신 dist-windows 빌드인지, 공식 Wormhole가 종료됐는지도 확인하세요."
|
|
) from exc
|
|
raise
|
|
sent += int(n)
|
|
if n == 0:
|
|
break
|
|
return sent
|
|
|
|
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(self._layout.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()
|