Macbook
2046 lines
72 KiB
Python
2046 lines
72 KiB
Python
"""Low-level USB access to j5create JUC500/JUC700 and OEM SmartKMLink cables.
|
||
|
||
Supported models share the J5FX IF5 bulk protocol (0x08/0x89 + 0x0A/0x8B):
|
||
JUC500 — VID 0x0711 PID 0x7500 — IF5
|
||
JUC700 — VID 0x0711 PID 0x7000–0x700F
|
||
• Windows: 700x composite IF5 (MI_05) and/or 7500 sibling
|
||
• Apple Silicon / Mac↔Win: Duet Bus RC IF0 bulk when IF5 absent
|
||
SmartKMLink — VID 0x0EA0 PID 0x7301 — same IF5 layout as JUC500 (OEM)
|
||
"""
|
||
|
||
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 # default / JUC500
|
||
PID_JUC500 = 0x7500
|
||
PID_JUC700_MIN = 0x7000
|
||
PID_JUC700_MAX = 0x700F
|
||
|
||
# OEM KM Link (same IF5 endpoints as JUC500 Smart Data Link)
|
||
VID_SMARTKMLINK = 0x0EA0
|
||
PID_SMARTKMLINK = 0x7301
|
||
|
||
# JUC500 + JUC700 (MCT USB3.0 Duet Bus) + sibling OEM IDs under the same vendor
|
||
ALT_PIDS = (
|
||
{PID_JUC500, 0x5500, 0x0700}
|
||
| {pid for pid in range(PID_JUC700_MIN, PID_JUC700_MAX + 1)}
|
||
)
|
||
|
||
CABLE_MODEL_JUC500 = "JUC500"
|
||
CABLE_MODEL_JUC700 = "JUC700"
|
||
CABLE_MODEL_SMARTKMLINK = "SmartKMLink"
|
||
CABLE_MODEL_GENERIC = "Smart Data Link"
|
||
# Default when multiple cables are plugged in (override via JUC500_CABLE_MODEL env)
|
||
DEFAULT_CABLE_MODEL = CABLE_MODEL_JUC500
|
||
|
||
KNOWN_CABLE_MODELS = (
|
||
CABLE_MODEL_JUC500,
|
||
CABLE_MODEL_JUC700,
|
||
CABLE_MODEL_SMARTKMLINK,
|
||
)
|
||
|
||
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-20-mac-if0-bcast"
|
||
# 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",
|
||
"juc700",
|
||
"duet",
|
||
"dss",
|
||
"oti",
|
||
"smartkmlink",
|
||
"km link",
|
||
"kmlink",
|
||
)
|
||
|
||
|
||
def _is_smartkmlink(vid: int, pid: int) -> bool:
|
||
return int(vid) == VID_SMARTKMLINK and int(pid) == PID_SMARTKMLINK
|
||
|
||
|
||
def cable_model_from(vid: int, pid: int, product: str = "") -> str:
|
||
"""Map USB IDs to a user-facing cable model (auto-detected, no manual pick)."""
|
||
vid = int(vid)
|
||
pid = int(pid)
|
||
if _is_smartkmlink(vid, pid):
|
||
return CABLE_MODEL_SMARTKMLINK
|
||
p = (product or "").lower()
|
||
if "smartkmlink" in p or "km link" in p or "kmlink" in p:
|
||
return CABLE_MODEL_SMARTKMLINK
|
||
if vid != VID:
|
||
return CABLE_MODEL_GENERIC
|
||
if pid == PID_JUC500 or pid == 0x0700:
|
||
return CABLE_MODEL_JUC500
|
||
if PID_JUC700_MIN <= pid <= PID_JUC700_MAX:
|
||
return CABLE_MODEL_JUC700
|
||
if any(h in p for h in ("juc700", "dss", "duet bus")):
|
||
return CABLE_MODEL_JUC700
|
||
if any(h in p for h in ("juc500", "wormhole", "smart data", "data link")):
|
||
return CABLE_MODEL_JUC500
|
||
if pid in ALT_PIDS:
|
||
return CABLE_MODEL_GENERIC
|
||
return CABLE_MODEL_GENERIC
|
||
|
||
|
||
def _has_juc700_marker(peers: list[DeviceInfo]) -> bool:
|
||
return any(PID_JUC700_MIN <= int(d.pid) <= PID_JUC700_MAX for d in peers)
|
||
|
||
|
||
def resolve_cable_model(
|
||
vid: int,
|
||
pid: int,
|
||
product: str = "",
|
||
*,
|
||
serial: str = "",
|
||
bus: int = -1,
|
||
peers: list[DeviceInfo] | None = None,
|
||
) -> str:
|
||
"""Resolve dongle family — JUC700's file-transfer IF5 node is still pid 7500."""
|
||
base = cable_model_from(vid, pid, product)
|
||
if base == CABLE_MODEL_JUC700 or peers is None:
|
||
return base
|
||
markers = [
|
||
d
|
||
for d in peers
|
||
if PID_JUC700_MIN <= int(d.pid) <= PID_JUC700_MAX
|
||
]
|
||
if not markers:
|
||
return base
|
||
serial = (serial or "").strip()
|
||
for m in markers:
|
||
ms = (m.serial or "").strip()
|
||
if serial and ms and serial == ms:
|
||
return CABLE_MODEL_JUC700
|
||
if bus >= 0 and int(m.bus) == int(bus):
|
||
return CABLE_MODEL_JUC700
|
||
return base
|
||
|
||
|
||
def supported_cable_models() -> tuple[str, ...]:
|
||
return (CABLE_MODEL_JUC500, CABLE_MODEL_JUC700)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CableLayout:
|
||
"""Per-model USB interface + 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 _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 default_if5_layout() -> CableLayout:
|
||
"""Smart Data Link file-transfer path (JUC500 + JUC700 both use IF5 bulk)."""
|
||
return CableLayout(
|
||
IF_VENDOR, EP_DATA_OUT, EP_DATA_IN, EP_CTRL_OUT, EP_CTRL_IN, MAX_PACKET
|
||
)
|
||
|
||
|
||
def default_juc700_if0_layout(max_packet: int = 1024) -> CableLayout:
|
||
"""JUC700 Duet Bus IF0 — prefer 0x01/0x81 (Mac↔Win probe)."""
|
||
return CableLayout(0, 0x01, 0x81, 0x03, 0x83, max(max_packet, 512))
|
||
|
||
|
||
# Preferred (data_out, data_in, ctrl_out, ctrl_in) for JUC700 IF0.
|
||
# 0x01/0x81 pair measured slightly more responsive toward Win peer drain.
|
||
_JUC700_IF0_EP_PRESETS: tuple[tuple[int, int, int, int], ...] = (
|
||
(0x01, 0x81, 0x03, 0x83),
|
||
(0x0D, 0x8D, 0x0E, 0x8E),
|
||
(0x03, 0x83, 0x01, 0x81),
|
||
(0x0E, 0x8E, 0x0D, 0x8D),
|
||
)
|
||
|
||
|
||
def _juc700_if0_bulk_eps(dev: usb.core.Device) -> tuple[list[int], list[int]]:
|
||
"""All bulk OUT/IN on JUC700 IF0 (Mac↔Win may bridge different pairs)."""
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return [], []
|
||
for intf in cfg:
|
||
if int(intf.bInterfaceNumber) != 0:
|
||
continue
|
||
outs, ins, _ = _bulk_endpoints(intf)
|
||
return sorted(set(outs)), sorted(set(ins))
|
||
return [], []
|
||
|
||
|
||
def _juc700_mac_if0_usable(dev: usb.core.Device) -> bool:
|
||
"""True when 700x IF0 has at least one bulk IN/OUT pair (Mac Duet Bus)."""
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return False
|
||
for intf in cfg:
|
||
if int(intf.bInterfaceNumber) != 0:
|
||
continue
|
||
outs, ins, _ = _bulk_endpoints(intf)
|
||
if outs and ins:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _detect_juc700_if0_layout(dev: usb.core.Device) -> Optional[CableLayout]:
|
||
"""Map JUC700 IF0 bulk endpoints for Mac↔Mac (and Win fallback without IF5)."""
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return default_juc700_if0_layout()
|
||
for intf in cfg:
|
||
if int(intf.bInterfaceNumber) != 0:
|
||
continue
|
||
outs, ins, max_pkt = _bulk_endpoints(intf)
|
||
if not outs or not ins:
|
||
continue
|
||
out_set, in_set = set(outs), set(ins)
|
||
for data_out, data_in, ctrl_out, ctrl_in in _JUC700_IF0_EP_PRESETS:
|
||
if (
|
||
data_out in out_set
|
||
and data_in in in_set
|
||
and ctrl_out in out_set
|
||
and ctrl_in in in_set
|
||
):
|
||
return CableLayout(
|
||
0, data_out, data_in, ctrl_out, ctrl_in, max(max_pkt, 512)
|
||
)
|
||
# Single pair — reuse for data+ctrl (keepalive shares data pipe).
|
||
data_out, data_in = outs[0], ins[0]
|
||
ctrl_out = outs[1] if len(outs) > 1 else data_out
|
||
ctrl_in = ins[1] if len(ins) > 1 else data_in
|
||
return CableLayout(0, data_out, data_in, ctrl_out, ctrl_in, max(max_pkt, 512))
|
||
return None
|
||
|
||
|
||
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):
|
||
# WinUSB (Zadig MI_05): libusb_get_configuration is often NOT_SUPPORTED.
|
||
# Fall back to the static configuration descriptor (dev[0]) — enough for
|
||
# endpoint discovery without an active-config query.
|
||
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_juc700_pid(pid: int) -> bool:
|
||
return PID_JUC700_MIN <= int(pid) <= PID_JUC700_MAX
|
||
|
||
|
||
def _win_default_if5_pid(pid: int) -> bool:
|
||
"""PIDs where Windows WinUSB uses IF5 without reading config (7500 + JUC700 composite)."""
|
||
return int(pid) == PID_JUC500 or _is_juc700_pid(pid)
|
||
|
||
|
||
def supports_file_transfer(info: DeviceInfo) -> bool:
|
||
"""True for 7500/IF5; SmartKMLink IF5; JUC700 IF5 or IF0."""
|
||
vid = int(getattr(info, "vid", VID) or VID)
|
||
pid = int(info.pid)
|
||
if _is_smartkmlink(vid, pid) or info.model == CABLE_MODEL_SMARTKMLINK:
|
||
return info.if_num == IF_VENDOR
|
||
if _is_juc700_pid(pid):
|
||
# IF5 preferred; IF0 when peer is Mac / no Smart Data Link sibling.
|
||
return info.if_num in (IF_VENDOR, 0)
|
||
if pid == PID_JUC500:
|
||
return info.if_num == IF_VENDOR or info.if_num == 0
|
||
return info.if_num == IF_VENDOR
|
||
|
||
|
||
def device_supports_file_transfer(dev: usb.core.Device) -> bool:
|
||
"""Raw device check — 7500/SmartKMLink; Mac 700x IF0; Win 700x when no 7500 sibling."""
|
||
vid = int(dev.idVendor)
|
||
pid = int(dev.idProduct)
|
||
if _is_smartkmlink(vid, pid):
|
||
return True
|
||
if pid == PID_JUC500 and vid == VID:
|
||
return True
|
||
if _is_juc700_pid(pid) and vid == VID:
|
||
try:
|
||
peers = find_juc500_devices()
|
||
except Exception:
|
||
peers = []
|
||
# Prefer separate Smart Data Link (7500) when both nodes exist.
|
||
if _find_7500_sibling(peers, bus=int(dev.bus)) is not None:
|
||
return False
|
||
if _IS_WIN:
|
||
return True
|
||
return _juc700_mac_if0_usable(dev)
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return False
|
||
for intf in cfg:
|
||
if int(intf.bInterfaceNumber) != IF_VENDOR:
|
||
continue
|
||
outs, ins, _ = _bulk_endpoints(intf)
|
||
if 0x08 in outs and 0x89 in ins:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _juc700_winusb_hint() -> str:
|
||
return (
|
||
"JUC700 Mac↔Win: 공식 Wormhole 종료 → Zadig → List All Devices → "
|
||
"0711:700x → Interface 0 (MI_00) → WinUSB. "
|
||
"(Win↔Win 공식용 MI_05와 다름 — Mac 상대는 IF0가 필요합니다.) "
|
||
"Windows에서도 이 juc500 앱을 실행하세요."
|
||
)
|
||
|
||
|
||
def _smartkmlink_winusb_hint() -> str:
|
||
return (
|
||
"SmartKMLink (0EA0:7301): 공식 Wormhole 종료 → Zadig(관리자) → "
|
||
"Options → List All Devices → SmartKMLink / USB ID 0EA0:7301 → "
|
||
"Interface 5 (MI_05) → Driver: WinUSB → Replace Driver → "
|
||
"케이블 재연결 후 start.bat. "
|
||
"(부모 장치가 아니라 MI_05 만 WinUSB로 바꿔야 합니다.)"
|
||
)
|
||
|
||
|
||
def _expand_open_candidates(
|
||
chosen: usb.core.Device,
|
||
matches: list[usb.core.Device],
|
||
) -> list[usb.core.Device]:
|
||
"""Prefer 0711:7500 sibling when present, else the chosen 700x/7500 node."""
|
||
seen: set[tuple[int, int]] = set()
|
||
out: list[usb.core.Device] = []
|
||
|
||
def add(d: usb.core.Device) -> None:
|
||
key = (int(d.bus), int(d.address))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
out.append(d)
|
||
|
||
serial = _safe_str(chosen, getattr(chosen, "iSerialNumber", 0), "")
|
||
if _is_juc700_pid(int(chosen.idProduct)):
|
||
# Win↔Win may use 7500/IF5 sibling; Mac↔Win JUC700 needs 700x IF0 only.
|
||
if _IS_WIN:
|
||
sib = _find_7500_sibling(matches, bus=int(chosen.bus), serial=serial)
|
||
if sib is not None:
|
||
add(sib)
|
||
add(chosen)
|
||
# Also try other 700x nodes on the same bus (hotplug address churn).
|
||
if _is_juc700_pid(int(chosen.idProduct)):
|
||
for d in matches:
|
||
if _is_juc700_pid(int(d.idProduct)):
|
||
add(d)
|
||
return out
|
||
|
||
|
||
def _find_7500_sibling(
|
||
devices: list[usb.core.Device],
|
||
*,
|
||
bus: Optional[int] = None,
|
||
serial: str = "",
|
||
) -> Optional[usb.core.Device]:
|
||
"""Locate Smart Data Link (0711:7500) paired with a JUC700 DSS node."""
|
||
serial = (serial or "").strip()
|
||
for d in devices:
|
||
if int(d.idProduct) != PID_JUC500:
|
||
continue
|
||
if bus is not None and int(d.bus) != int(bus):
|
||
continue
|
||
if serial:
|
||
ds = _safe_str(d, getattr(d, "iSerialNumber", 0), "").strip()
|
||
if ds and ds != serial:
|
||
continue
|
||
return d
|
||
for d in devices:
|
||
if int(d.idProduct) == PID_JUC500:
|
||
return d
|
||
return None
|
||
|
||
|
||
def find_smart_data_link_device(
|
||
devices: list[usb.core.Device],
|
||
*,
|
||
bus: Optional[int] = None,
|
||
address: Optional[int] = None,
|
||
prefer_model: str | None = None,
|
||
) -> Optional[usb.core.Device]:
|
||
"""Resolve file-transfer node: SmartKMLink / 7500/IF5 / usable 700x."""
|
||
if bus is not None and address is not None:
|
||
picked = [
|
||
d
|
||
for d in devices
|
||
if int(d.bus) == int(bus) and int(d.address) == int(address)
|
||
]
|
||
if picked:
|
||
dev = picked[0]
|
||
vid = int(dev.idVendor)
|
||
pid = int(dev.idProduct)
|
||
if _is_smartkmlink(vid, pid) or pid == PID_JUC500:
|
||
return dev
|
||
if PID_JUC700_MIN <= pid <= PID_JUC700_MAX:
|
||
serial = _safe_str(dev, getattr(dev, "iSerialNumber", 0), "")
|
||
# Mac↔Win: always 700x IF0 — 7500/IF5 does not bridge to Mac IF0.
|
||
if not _IS_WIN and device_supports_file_transfer(dev):
|
||
return dev
|
||
sib = _find_7500_sibling(devices, bus=int(dev.bus), serial=serial)
|
||
if sib is not None:
|
||
return sib
|
||
# Windows IF5 composite, or Mac Duet Bus IF0.
|
||
if device_supports_file_transfer(dev) or _IS_WIN:
|
||
return dev
|
||
return None
|
||
# Stale bus/address after hotplug — fall through to global search.
|
||
|
||
prefer = (prefer_model or "").strip()
|
||
prefer_key = prefer.upper().replace(" ", "")
|
||
if prefer_key in ("SMARTKMLINK", "KMLINK") or prefer == CABLE_MODEL_SMARTKMLINK:
|
||
for d in devices:
|
||
if _is_smartkmlink(int(d.idVendor), int(d.idProduct)):
|
||
return d
|
||
if prefer_key == CABLE_MODEL_JUC700 or prefer == CABLE_MODEL_JUC700:
|
||
if not _IS_WIN:
|
||
for d in devices:
|
||
if _is_juc700_pid(int(d.idProduct)) and device_supports_file_transfer(d):
|
||
return d
|
||
sib = _find_7500_sibling(devices)
|
||
if sib is not None:
|
||
return sib
|
||
for d in devices:
|
||
if _is_juc700_pid(int(d.idProduct)) and (
|
||
device_supports_file_transfer(d) or _IS_WIN
|
||
):
|
||
return d
|
||
|
||
for d in devices:
|
||
if _is_smartkmlink(int(d.idVendor), int(d.idProduct)):
|
||
return d
|
||
for d in devices:
|
||
if int(d.idProduct) == PID_JUC500 and int(d.idVendor) == VID:
|
||
return d
|
||
for d in devices:
|
||
if _is_juc700_pid(int(d.idProduct)) and (
|
||
device_supports_file_transfer(d) or _IS_WIN
|
||
):
|
||
return d
|
||
return None
|
||
|
||
|
||
def data_link_devices(devices: list[DeviceInfo]) -> list[DeviceInfo]:
|
||
"""USB nodes usable for J5FX file transfer (7500/IF5 or Mac JUC700 IF0)."""
|
||
links = [d for d in devices if supports_file_transfer(d)]
|
||
if not _IS_WIN:
|
||
# Mac↔Win JUC700: Duet Bus IF0 only — sibling 7500/IF5 is not bridged.
|
||
j700_if0 = [
|
||
d
|
||
for d in links
|
||
if d.model == CABLE_MODEL_JUC700 and int(getattr(d, "if_num", 5)) == 0
|
||
]
|
||
if j700_if0:
|
||
return j700_if0
|
||
# Prefer 7500 over 700x when both are present on the same dongle (Win↔Win).
|
||
if any(int(d.pid) == PID_JUC500 for d in links):
|
||
links = [d for d in links if not _is_juc700_pid(int(d.pid))]
|
||
return links
|
||
|
||
|
||
def available_cable_models(devices: list[DeviceInfo] | None = None) -> tuple[str, ...]:
|
||
"""Cable types to show in the UI (one dongle vs JUC500+JUC700 pair)."""
|
||
if devices is None:
|
||
try:
|
||
devices = list_devices()
|
||
except Exception:
|
||
return (CABLE_MODEL_JUC500,)
|
||
links = data_link_devices(devices)
|
||
models: list[str] = []
|
||
if any(d.model == CABLE_MODEL_SMARTKMLINK for d in links):
|
||
models.append(CABLE_MODEL_SMARTKMLINK)
|
||
if _has_juc700_marker(devices):
|
||
if len([d for d in links if d.model != CABLE_MODEL_SMARTKMLINK]) >= 2:
|
||
models.extend([CABLE_MODEL_JUC500, CABLE_MODEL_JUC700])
|
||
else:
|
||
models.append(CABLE_MODEL_JUC700)
|
||
elif any(d.model == CABLE_MODEL_JUC500 for d in links) or (
|
||
links and not any(d.model == CABLE_MODEL_SMARTKMLINK for d in links)
|
||
):
|
||
if CABLE_MODEL_JUC500 not in models:
|
||
models.append(CABLE_MODEL_JUC500)
|
||
# Dedupe preserve order
|
||
out: list[str] = []
|
||
for m in models:
|
||
if m not in out:
|
||
out.append(m)
|
||
return tuple(out)
|
||
|
||
|
||
def detect_cable_layout(dev: usb.core.Device) -> CableLayout:
|
||
"""Pick bulk endpoints: JUC700 prefers IF0 (Mac↔Win); else IF5 (0x08/0x89)."""
|
||
pid = int(dev.idProduct)
|
||
# Mac↔Win JUC700: both ends must use Duet Bus IF0. Win↔Win classic IF5 is
|
||
# secondary — IF5 does not bridge to Mac IF0 HELLO.
|
||
if _IS_WIN and _is_juc700_pid(pid):
|
||
if0 = _detect_juc700_if0_layout(dev)
|
||
if if0 is not None:
|
||
return if0
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return default_if5_layout()
|
||
for intf in cfg:
|
||
ifnum = int(intf.bInterfaceNumber)
|
||
outs, ins, max_pkt = _bulk_endpoints(intf)
|
||
if not outs or not ins:
|
||
continue
|
||
if ifnum == IF_VENDOR and 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)
|
||
)
|
||
return default_if5_layout()
|
||
|
||
if _IS_WIN and (
|
||
_win_default_if5_pid(pid)
|
||
or _is_smartkmlink(int(dev.idVendor), pid)
|
||
):
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
return default_if5_layout()
|
||
for intf in cfg:
|
||
ifnum = int(intf.bInterfaceNumber)
|
||
outs, ins, max_pkt = _bulk_endpoints(intf)
|
||
if not outs or not ins:
|
||
continue
|
||
if ifnum == IF_VENDOR and 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)
|
||
)
|
||
return default_if5_layout()
|
||
|
||
try:
|
||
cfg = _get_device_configuration(dev)
|
||
except Exception:
|
||
if _IS_WIN and (
|
||
_win_default_if5_pid(int(dev.idProduct))
|
||
or _is_smartkmlink(int(dev.idVendor), int(dev.idProduct))
|
||
):
|
||
return default_if5_layout()
|
||
if _is_juc700_pid(pid):
|
||
return default_juc700_if0_layout()
|
||
if _is_smartkmlink(int(dev.idVendor), pid):
|
||
return default_if5_layout()
|
||
raise
|
||
|
||
for intf in cfg:
|
||
ifnum = int(intf.bInterfaceNumber)
|
||
outs, ins, max_pkt = _bulk_endpoints(intf)
|
||
if not outs or not ins:
|
||
continue
|
||
# Smart Data Link — IF5 (JUC500 + JUC700 file xfer when present)
|
||
if ifnum == IF_VENDOR and 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_juc700_pid(pid):
|
||
if0 = _detect_juc700_if0_layout(dev)
|
||
if if0 is not None:
|
||
return if0
|
||
if _IS_WIN:
|
||
return default_if5_layout()
|
||
raise RuntimeError(
|
||
"JUC700 IF0 bulk 엔드포인트를 찾지 못했습니다. "
|
||
"케이블을 다시 연결하거나 다른 USB 포트를 사용하세요."
|
||
)
|
||
|
||
if _IS_WIN and (
|
||
int(dev.idProduct) == PID_JUC500
|
||
or _is_smartkmlink(int(dev.idVendor), int(dev.idProduct))
|
||
):
|
||
return default_if5_layout()
|
||
|
||
raise RuntimeError(
|
||
"Smart Data Link IF5(0x08/0x89)를 찾지 못했습니다 — "
|
||
"JUC500(0711:7500) 또는 SmartKMLink(0ea0:7301) 케이블인지 확인하세요."
|
||
)
|
||
|
||
|
||
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))
|
||
# walk up a few levels looking for vendor/
|
||
for base in list(roots):
|
||
p = base
|
||
for _ in range(4):
|
||
roots.append(p)
|
||
if (p / "vendor").is_dir():
|
||
break
|
||
p = p.parent
|
||
# unique preserve order
|
||
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:
|
||
# venv Scripts / beside python
|
||
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") # PATH search
|
||
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"]
|
||
|
||
# dedupe
|
||
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:
|
||
# smoke: enumerate (may be empty)
|
||
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:
|
||
"""Device unplugged, stale handle, or bulk EP missing (libusb NOT_FOUND)."""
|
||
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:
|
||
"""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:
|
||
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):
|
||
mi = f"Interface {ifnum} (MI_{ifnum:02d})"
|
||
if ifnum == IF_VENDOR:
|
||
mi = "Interface 5 (MI_05)"
|
||
return (
|
||
f"{base}\n\n"
|
||
f"WinUSB가 해당 libusb API를 지원하지 않습니다. {mi}에 "
|
||
"WinUSB가 설치되어 있는지 Zadig로 확인하세요.\n"
|
||
f"{_smartkmlink_winusb_hint()}\n"
|
||
"JUC500도 동일하게 Interface 5(MI_05) → WinUSB 입니다."
|
||
)
|
||
return (
|
||
f"{base}\n"
|
||
f"Zadig에서 Smart Data Link → Interface {ifnum} (MI_0{ifnum}) 에 WinUSB를 설치하세요."
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class DeviceInfo:
|
||
bus: int
|
||
address: int
|
||
serial: str
|
||
product: str
|
||
manufacturer: str
|
||
vid: int = VID
|
||
pid: int = PID
|
||
model: str = CABLE_MODEL_GENERIC
|
||
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 _is_smartkmlink(int(dev.idVendor), int(dev.idProduct)):
|
||
return True
|
||
if int(dev.idVendor) != VID:
|
||
return False
|
||
if int(dev.idProduct) in ALT_PIDS or int(dev.idProduct) == PID:
|
||
return True
|
||
p = (product or "").lower()
|
||
return any(h in p for h in _PRODUCT_HINTS)
|
||
|
||
|
||
def _device_sort_key(dev: usb.core.Device) -> tuple[int, int]:
|
||
"""Prefer JUC700, then JUC500/SmartKMLink, then other siblings."""
|
||
vid = int(dev.idVendor)
|
||
pid = int(dev.idProduct)
|
||
if vid == VID and PID_JUC700_MIN <= pid <= PID_JUC700_MAX:
|
||
return (0, pid)
|
||
if vid == VID and pid == PID_JUC500:
|
||
return (1, pid)
|
||
if _is_smartkmlink(vid, pid):
|
||
return (1, pid)
|
||
return (2, pid)
|
||
|
||
|
||
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 JUC / SmartKMLink candidates. Avoid get_string on unrelated devices."""
|
||
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_smartkmlink(vid, pid):
|
||
found.append(dev)
|
||
continue
|
||
if vid != VID:
|
||
continue
|
||
# Known PIDs — no string descriptor I/O needed for matching.
|
||
if pid in ALT_PIDS or pid == PID or _is_juc700_pid(pid):
|
||
found.append(dev)
|
||
continue
|
||
product = _safe_str(dev, getattr(dev, "iProduct", 0), "")
|
||
if _looks_like_juc500(dev, product):
|
||
found.append(dev)
|
||
found.sort(key=_device_sort_key)
|
||
return found
|
||
|
||
|
||
def order_devices(
|
||
devices: list[DeviceInfo],
|
||
*,
|
||
serial: str | None = None,
|
||
model: str | None = None,
|
||
) -> list[DeviceInfo]:
|
||
"""Pick connection order when multiple cables are plugged in."""
|
||
if not devices:
|
||
return []
|
||
preferred = (
|
||
model
|
||
if model in KNOWN_CABLE_MODELS
|
||
else DEFAULT_CABLE_MODEL
|
||
)
|
||
|
||
def rank(d: DeviceInfo) -> tuple[int, int]:
|
||
r_serial = 0 if serial and d.serial == serial.strip() else 1
|
||
r_model = 0 if d.model == preferred else 1
|
||
return (r_serial, r_model)
|
||
|
||
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]:
|
||
raw: list[dict[str, Any]] = []
|
||
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)
|
||
serial = _safe_str(dev, getattr(dev, "iSerialNumber", 0))
|
||
if PID_JUC700_MIN <= pid <= PID_JUC700_MAX:
|
||
# Prefer real layout: Win↔Win often IF5; Mac peer may expose IF0 only.
|
||
if_num = IF_VENDOR if _IS_WIN else 0
|
||
try:
|
||
if_num = int(detect_cable_layout(dev).if_num)
|
||
except Exception:
|
||
pass
|
||
elif device_supports_file_transfer(dev):
|
||
try:
|
||
layout = detect_cable_layout(dev)
|
||
if_num = layout.if_num
|
||
except Exception:
|
||
if_num = (
|
||
IF_VENDOR
|
||
if pid == PID_JUC500 or _is_smartkmlink(vid, pid)
|
||
else 0
|
||
)
|
||
else:
|
||
if_num = 0
|
||
raw.append(
|
||
{
|
||
"bus": int(dev.bus),
|
||
"address": int(dev.address),
|
||
"serial": serial,
|
||
"product": product,
|
||
"manufacturer": _safe_str(dev, getattr(dev, "iManufacturer", 0)),
|
||
"vid": vid,
|
||
"pid": pid,
|
||
"if_num": if_num,
|
||
}
|
||
)
|
||
|
||
stubs = [
|
||
DeviceInfo(
|
||
bus=r["bus"],
|
||
address=r["address"],
|
||
serial=r["serial"],
|
||
product=r["product"],
|
||
manufacturer=r["manufacturer"],
|
||
vid=r["vid"],
|
||
pid=r["pid"],
|
||
model=cable_model_from(r["vid"], r["pid"], r["product"]),
|
||
if_num=r["if_num"],
|
||
)
|
||
for r in raw
|
||
]
|
||
|
||
out: list[DeviceInfo] = []
|
||
for r in raw:
|
||
model = resolve_cable_model(
|
||
r["vid"],
|
||
r["pid"],
|
||
r["product"],
|
||
serial=r["serial"],
|
||
bus=r["bus"],
|
||
peers=stubs,
|
||
)
|
||
out.append(
|
||
DeviceInfo(
|
||
bus=r["bus"],
|
||
address=r["address"],
|
||
serial=r["serial"],
|
||
product=r["product"],
|
||
manufacturer=r["manufacturer"],
|
||
vid=r["vid"],
|
||
pid=r["pid"],
|
||
model=model,
|
||
if_num=r["if_num"],
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _macos_ioreg_vid0711() -> list[dict[str, str]]:
|
||
"""Fallback: IORegistry may show JUC cable 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 not in (VID, VID_SMARTKMLINK):
|
||
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)
|
||
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
|
||
has_700 = any(_is_juc700_pid(int(x.get("pid", 0))) for x in report.juc500)
|
||
if _IS_WIN and has_700:
|
||
report.hints.append(_juc700_winusb_hint())
|
||
report.hints.append(
|
||
"케이블이 libusb로 보입니다. 「연결」을 진행하세요. "
|
||
"(JUC500/JUC700 파일 전송은 둘 다 Interface 5)"
|
||
)
|
||
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). "
|
||
"JUC500/JUC700 파일 전송 모두 IF5를 사용합니다. "
|
||
"공식 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에는 케이블이 보입니다 ({names}) — libusb만 못 찾는 상태입니다. "
|
||
"USB 허브/포트를 바꾸고 케이블을 다시 연결한 뒤 GUI를 재시작하세요."
|
||
)
|
||
else:
|
||
report.hints.append(
|
||
"이 PC에서 JUC500/JUC700(0711) 또는 SmartKMLink(0ea0:7301)이 보이지 않습니다. "
|
||
"케이블을 이 PC USB 포트에 직접 연결했는지, "
|
||
"반대쪽은 상대 PC에 꽂혀 있는지 확인하세요. "
|
||
"(양쪽 끝이 각각 다른 PC에 연결되어야 합니다.)"
|
||
)
|
||
if _IS_WIN:
|
||
report.hints.append(
|
||
"장치 관리자에서 'Smart Data Link' / '알 수 없는 장치' / "
|
||
"WORMHOLE CD 롬이 있는지 확인하세요."
|
||
)
|
||
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)
|
||
):
|
||
# Zadig WinUSB on MI_05: claim often NOT_SUPPORTED; bulk still works.
|
||
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 (JUC500 + JUC700)."""
|
||
|
||
def __init__(
|
||
self,
|
||
serial: Optional[str] = None,
|
||
*,
|
||
bus: Optional[int] = None,
|
||
address: Optional[int] = None,
|
||
model: 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._pick_bus = bus
|
||
self._pick_address = address
|
||
self._display_model = (
|
||
model
|
||
if model in KNOWN_CABLE_MODELS
|
||
else None
|
||
)
|
||
self._layout = CableLayout(
|
||
IF_VENDOR, EP_DATA_OUT, EP_DATA_IN, EP_CTRL_OUT, EP_CTRL_IN, MAX_PACKET
|
||
)
|
||
# Bytes drained during OUT recovery must not be discarded — RX loop reads here first.
|
||
self._rx_stash = bytearray()
|
||
self._rx_stash_lock = threading.Lock()
|
||
# Mac JUC700 IF0: poll/broadcast all bulk EPs (Win bridge may use 0x01/0x81 or 0x0D/0x8D).
|
||
self._if0_all_outs: list[int] = []
|
||
self._if0_all_ins: list[int] = []
|
||
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/JUC700(Smart Data Link / 0711:7500·700x) 장치를 찾지 못했습니다.\n"
|
||
+ hints
|
||
)
|
||
|
||
chosen_dev = find_smart_data_link_device(
|
||
matches,
|
||
bus=self._pick_bus,
|
||
address=self._pick_address,
|
||
prefer_model=self._display_model,
|
||
)
|
||
if chosen_dev is None:
|
||
has_700 = any(
|
||
PID_JUC700_MIN <= int(d.idProduct) <= PID_JUC700_MAX for d in matches
|
||
)
|
||
want_700 = self._display_model == CABLE_MODEL_JUC700
|
||
has_km = any(
|
||
_is_smartkmlink(int(d.idVendor), int(d.idProduct)) for d in matches
|
||
)
|
||
want_km = self._display_model == CABLE_MODEL_SMARTKMLINK
|
||
if has_km or want_km:
|
||
hint = (
|
||
"SmartKMLink(0ea0:7301) IF5를 열 수 없습니다. "
|
||
"케이블을 다시 꽂거나 (Windows) Zadig로 MI_05 → WinUSB를 확인하세요."
|
||
)
|
||
elif _IS_WIN and (has_700 or want_700):
|
||
hint = (
|
||
"JUC700: Zadig에서 Smart Data Link → Interface 5 (MI_05) → WinUSB 설치 후 "
|
||
"케이블을 다시 연결하세요."
|
||
)
|
||
elif has_700 or want_700:
|
||
hint = (
|
||
"JUC700(0711:700x) IF0 bulk 장치를 열 수 없습니다. "
|
||
"케이블을 다시 꽂거나 다른 USB-C 포트에 연결하세요."
|
||
)
|
||
else:
|
||
hint = (
|
||
"JUC500(0711:7500) 또는 SmartKMLink(0ea0:7301) 케이블을 연결하세요."
|
||
)
|
||
raise RuntimeError(
|
||
"파일 전송용 USB 장치를 찾지 못했습니다. " + hint
|
||
)
|
||
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 = _expand_open_candidates(chosen_dev, matches)
|
||
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:
|
||
expanded: list[usb.core.Device] = []
|
||
seen: set[tuple[int, int]] = set()
|
||
for sm in serial_matches:
|
||
for d in _expand_open_candidates(sm, matches):
|
||
key = (int(d.bus), int(d.address))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
expanded.append(d)
|
||
candidates = expanded or candidates
|
||
elif _safe_str(chosen_dev, getattr(chosen_dev, "iSerialNumber", 0)) != self._serial.strip():
|
||
raise RuntimeError(
|
||
f"시리얼 '{self._serial}' 에 해당하는 Smart Data Link 케이블이 없습니다."
|
||
)
|
||
|
||
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 ""
|
||
model = self._display_model or CABLE_MODEL_GENERIC
|
||
if candidates and model == CABLE_MODEL_GENERIC:
|
||
d0 = candidates[0]
|
||
model = cable_model_from(
|
||
int(d0.idVendor),
|
||
int(d0.idProduct),
|
||
_safe_str(d0, getattr(d0, "iProduct", 0), ""),
|
||
)
|
||
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 and (
|
||
model == CABLE_MODEL_SMARTKMLINK
|
||
or self._display_model == CABLE_MODEL_SMARTKMLINK
|
||
or any(
|
||
_is_smartkmlink(int(d.idVendor), int(d.idProduct))
|
||
for d in candidates
|
||
)
|
||
):
|
||
detail += f"\n\n{_smartkmlink_winusb_hint()}"
|
||
elif _IS_WIN and (
|
||
model == CABLE_MODEL_JUC700
|
||
or self._display_model == CABLE_MODEL_JUC700
|
||
or any(_is_juc700_pid(int(d.idProduct)) for d in candidates)
|
||
):
|
||
detail += f"\n\n{_juc700_winusb_hint()}"
|
||
elif _IS_WIN and self._layout.if_num == IF_VENDOR:
|
||
detail += (
|
||
"\n\nZadig → List All Devices → 해당 케이블 → "
|
||
"Interface 5 (MI_05) → WinUSB → Replace Driver"
|
||
)
|
||
raise RuntimeError(
|
||
f"{model} USB Interface {self._layout.if_num} 를 열 수 없습니다."
|
||
+ detail
|
||
) from last_err
|
||
|
||
# Optional CDC only for classic Smart Data Link (IF5). JUC700 Mac uses IF0.
|
||
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 self._display_model is None:
|
||
try:
|
||
peers = list_devices()
|
||
except Exception:
|
||
peers = []
|
||
self._display_model = resolve_cable_model(
|
||
int(chosen.idVendor),
|
||
int(chosen.idProduct),
|
||
_safe_str(chosen, getattr(chosen, "iProduct", 0), ""),
|
||
serial=_safe_str(chosen, getattr(chosen, "iSerialNumber", 0)),
|
||
bus=int(chosen.bus),
|
||
peers=peers,
|
||
)
|
||
if _IS_WIN:
|
||
self._configure_windows()
|
||
else:
|
||
self._configure_posix()
|
||
self._claim(self._layout.if_num, required=True)
|
||
# Force backend path readiness (Win claims IF here; then clear IF0 halts)
|
||
self._ensure_backend_ready()
|
||
self._prepare_juc700_if0()
|
||
if self._is_juc700_if0() and not _IS_WIN:
|
||
self._select_juc700_if0_preset()
|
||
self._probe_link()
|
||
|
||
def _refresh_juc700_if0_eps(self) -> None:
|
||
"""Cache IF0 bulk EP lists for multi-IN poll / HELLO broadcast."""
|
||
self._if0_all_outs = []
|
||
self._if0_all_ins = []
|
||
if not self._is_juc700_if0() or self._dev is None:
|
||
return
|
||
outs, ins = _juc700_if0_bulk_eps(self._dev)
|
||
self._if0_all_outs = outs or [self._layout.ep_data_out]
|
||
self._if0_all_ins = ins or [self._layout.ep_data_in]
|
||
|
||
def _select_juc700_if0_preset(self) -> None:
|
||
"""Mac↔Win: fixed 0x01/0x81 data path; HELLO uses all OUT EPs."""
|
||
if not self._is_juc700_if0() or self._dev is None:
|
||
return
|
||
max_pkt = self._layout.max_packet
|
||
self._layout = default_juc700_if0_layout(max_pkt)
|
||
self._prepare_juc700_if0()
|
||
self._refresh_juc700_if0_eps()
|
||
|
||
def _is_juc700_if0(self) -> bool:
|
||
if self._dev is None:
|
||
return False
|
||
return self._layout.if_num == 0 and _is_juc700_pid(int(self._dev.idProduct))
|
||
|
||
def _is_smartkmlink_dev(self) -> bool:
|
||
if self._dev is None:
|
||
return False
|
||
return _is_smartkmlink(int(self._dev.idVendor), int(self._dev.idProduct))
|
||
|
||
def _prepare_juc700_if0(self) -> None:
|
||
"""Clear stale halts on Duet Bus IF0 after claim (Mac and Win IF0 fallback)."""
|
||
if not self._is_juc700_if0() or self._dev is None:
|
||
return
|
||
for ep in (
|
||
self._layout.ep_data_out,
|
||
self._layout.ep_data_in,
|
||
self._layout.ep_ctrl_out,
|
||
self._layout.ep_ctrl_in,
|
||
):
|
||
try:
|
||
self._dev.clear_halt(ep)
|
||
except Exception:
|
||
pass
|
||
|
||
def wake_bridge(self) -> None:
|
||
"""Best-effort ctrl/data poke so the peer host starts draining IN (HELLO path).
|
||
|
||
JUC700 Mac↔Win: data OUT often stalls until the other end claims IN.
|
||
Soft ctrl ALIVE + short IN drain does not fail open; wait_peer keeps
|
||
retrying HELLO until the peer juc500 listen side is ready.
|
||
"""
|
||
try:
|
||
self.write_ctrl(b"\x00ALIVE", timeout_ms=120)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.read_ctrl(MAX_PACKET, timeout_ms=60)
|
||
except Exception:
|
||
pass
|
||
if self._is_juc700_if0():
|
||
size = min(512, self._layout.max_packet)
|
||
for ep in self._if0_all_ins or [self._layout.ep_data_in]:
|
||
try:
|
||
self._bulk_read(ep, size, 40)
|
||
except Exception:
|
||
pass
|
||
|
||
def rotate_juc700_if0_eps(self) -> bool:
|
||
"""No-op — Mac IF0 uses fixed 0x01/0x81 + broadcast HELLO on all OUT EPs."""
|
||
return False
|
||
|
||
def _probe_link(self) -> None:
|
||
"""Fail fast if bulk OUT is dead (WinUSB MI_05 mismatch shows as Entity not found).
|
||
|
||
JUC700 IF0 / SmartKMLink / Win IF5: OUT often stalls or reports NOT_SUPPORTED
|
||
until the peer claims IN / WinUSB is bound — soft-wake only on those paths.
|
||
"""
|
||
if (
|
||
self._is_juc700_if0()
|
||
or self._is_smartkmlink_dev()
|
||
or (_IS_WIN and self._layout.if_num == IF_VENDOR)
|
||
):
|
||
self.wake_bridge()
|
||
return
|
||
try:
|
||
self._backend_bulk_write(
|
||
self._layout.ep_ctrl_out, b"\x00ALIVE", 250
|
||
)
|
||
except Exception as exc:
|
||
if _IS_WIN and (
|
||
_is_not_supported(exc) or isinstance(exc, NotImplementedError)
|
||
):
|
||
# Driver bound but claim/probe API unsupported — try runtime HELLO.
|
||
self.wake_bridge()
|
||
return
|
||
hint = ""
|
||
if _IS_WIN and (
|
||
self._display_model == CABLE_MODEL_JUC700
|
||
or (
|
||
self._dev is not None
|
||
and _is_juc700_pid(int(self._dev.idProduct))
|
||
)
|
||
):
|
||
hint = f"\n{_juc700_winusb_hint()}"
|
||
elif _IS_WIN and self._is_smartkmlink_dev():
|
||
hint = f"\n{_smartkmlink_winusb_hint()}"
|
||
raise RuntimeError(
|
||
f"USB bulk 경로 검증 실패: {exc}{hint}"
|
||
) from exc
|
||
|
||
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))
|
||
model = self._display_model
|
||
self._force_close()
|
||
matches = find_juc500_devices()
|
||
if not matches:
|
||
raise RuntimeError("USB 장치를 다시 찾지 못했습니다. 케이블을 확인하세요.")
|
||
chosen = find_smart_data_link_device(
|
||
matches,
|
||
bus=bus,
|
||
address=addr,
|
||
prefer_model=model,
|
||
)
|
||
if chosen is None:
|
||
seed = next((x for x in matches if serial and _safe_str(x, getattr(x, "iSerialNumber", 0)) == serial), None)
|
||
if seed is None:
|
||
seed = next((x for x in matches if int(x.bus) == bus), matches[0])
|
||
for d in _expand_open_candidates(seed, matches):
|
||
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:
|
||
"""WinUSB: never call get_configuration / prefer skipping set_configuration.
|
||
|
||
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._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 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:
|
||
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:
|
||
# WinUSB: some hosts report NOT_SUPPORTED on open/claim — bulk may still work.
|
||
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)
|
||
):
|
||
# Per-interface WinUSB: claim_interface is often unimplemented; proceed.
|
||
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):
|
||
# Re-wrap SmartKMLink NOT_SUPPORTED with Zadig steps.
|
||
if _IS_WIN and _is_not_supported(exc):
|
||
raise RuntimeError(
|
||
f"{exc}\n\n{_smartkmlink_winusb_hint()}"
|
||
) from exc
|
||
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:
|
||
# pyusb maps LIBUSB_ERROR_NOT_SUPPORTED → NotImplementedError
|
||
# (CDC IF0/IF1 on Windows often has NDIS/WinUSB mismatch).
|
||
if not required and (_is_not_supported(exc) or isinstance(exc, NotImplementedError)):
|
||
return
|
||
# Official Wormhole often holds IF5 — kill and retry once.
|
||
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:
|
||
assert self._dev is not None
|
||
if self._display_model in KNOWN_CABLE_MODELS:
|
||
return self._display_model
|
||
return cable_model_from(
|
||
int(self._dev.idVendor), int(self._dev.idProduct), self.product
|
||
)
|
||
|
||
@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:
|
||
if self._dev is None:
|
||
return 0
|
||
if broadcast and self._is_juc700_if0() and self._if0_all_outs:
|
||
to = self.timeout_ms if timeout_ms is None else int(timeout_ms)
|
||
per = max(80, to // max(1, len(self._if0_all_outs)))
|
||
sent = 0
|
||
for ep in self._if0_all_outs:
|
||
try:
|
||
sent = max(sent, self._bulk_write(ep, payload, per))
|
||
except Exception:
|
||
pass
|
||
if sent:
|
||
return sent
|
||
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
|
||
ins = self._if0_all_ins if self._is_juc700_if0() and self._if0_all_ins else [
|
||
self._layout.ep_data_in
|
||
]
|
||
if len(ins) <= 1:
|
||
return self._bulk_read(ins[0], size, timeout_ms)
|
||
to = self.timeout_ms if timeout_ms is None else int(timeout_ms)
|
||
per = max(25, to // len(ins))
|
||
for ep in ins:
|
||
chunk = self._bulk_read(ep, size, per)
|
||
if chunk:
|
||
return chunk
|
||
return b""
|
||
|
||
def _stash_rx(self, data: bytes) -> None:
|
||
if not data:
|
||
return
|
||
with self._rx_stash_lock:
|
||
self._rx_stash.extend(data)
|
||
# Cap runaway buffer (corrupted stream); RX FrameReader also bounds frames.
|
||
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:
|
||
"""Bulk IN via libusb backend (skips pyusb get_configuration on WinUSB)."""
|
||
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):
|
||
ifn = self._layout.if_num
|
||
hint = (
|
||
"Interface 5(MI_05)에 WinUSB"
|
||
if ifn == IF_VENDOR
|
||
else f"Interface {ifn}(MI_{ifn:02d})에 WinUSB"
|
||
)
|
||
raise RuntimeError(
|
||
f"USB 읽기 실패(EP 0x{ep:02x}): {exc2}\n"
|
||
f"Windows: {hint}가 잡혀 있는지 Zadig로 확인하세요."
|
||
) from exc2
|
||
return b""
|
||
return b""
|
||
if _is_not_supported(exc):
|
||
ifn = self._layout.if_num
|
||
hint = (
|
||
"Interface 5(MI_05)에 WinUSB"
|
||
if ifn == IF_VENDOR
|
||
else f"Interface {ifn}(MI_{ifn:02d})에 WinUSB"
|
||
)
|
||
raise RuntimeError(
|
||
f"USB 읽기 실패(EP 0x{ep:02x}): {exc}\n"
|
||
f"Windows: {hint}가 잡혀 있는지 Zadig로 확인하세요."
|
||
) from exc
|
||
return b""
|
||
|
||
def _backend_bulk_write(self, ep: int, data: bytes, timeout_ms: int) -> int:
|
||
"""Bulk OUT without holding ctx.lock for the wait (allows concurrent IN)."""
|
||
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:
|
||
"""Bulk IN without holding ctx.lock for the wait (allows concurrent OUT)."""
|
||
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:
|
||
"""Bulk OUT via libusb backend (skips pyusb get_configuration on WinUSB)."""
|
||
assert self._dev is not None
|
||
to = self.timeout_ms if timeout_ms is None else timeout_ms
|
||
# Scale timeout with transfer size (64KB URB can need more than 2s under load).
|
||
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()
|