修改UCD连接需要重启错误
This commit is contained in:
@@ -78,8 +78,27 @@ class ConnectionController:
|
||||
|
||||
# -- UCD 连接 ------------------------------------------------
|
||||
|
||||
def connect_ucd(self, display: str) -> bool:
|
||||
def _refresh_ucd_selection(self, display: str) -> tuple[list[str], str | None]:
|
||||
"""Rescan devices and return (device list, valid selection or None)."""
|
||||
ucd_list = self.list_ucd_devices()
|
||||
if display in ucd_list:
|
||||
return ucd_list, display
|
||||
return ucd_list, (ucd_list[0] if ucd_list else None)
|
||||
|
||||
def _sync_ucd_combo(self, ucd_list: list[str], selected: str) -> None:
|
||||
app = self._app
|
||||
app.ucd_list_combo.config(values=ucd_list)
|
||||
app.ucd_list_var.set(selected)
|
||||
app.update_config()
|
||||
|
||||
def connect_ucd(self, display: str, *, rescan: bool = True) -> bool:
|
||||
"""打开指定 UCD 设备。成功返回 True。"""
|
||||
if rescan:
|
||||
_, display = self._refresh_ucd_selection(display)
|
||||
if not display:
|
||||
self._log("未检测到 UCD 设备,请确认电源已打开后重试", level="error")
|
||||
return False
|
||||
|
||||
if self._device.state.name != "CLOSED":
|
||||
try:
|
||||
self._device.close()
|
||||
@@ -158,7 +177,11 @@ class ConnectionController:
|
||||
|
||||
def worker():
|
||||
try:
|
||||
ucd_ok = self.connect_ucd(app.ucd_list_var.get())
|
||||
ucd_list, selected = self._refresh_ucd_selection(app.ucd_list_var.get())
|
||||
app._dispatch_ui(self._sync_ucd_combo, ucd_list, selected or "")
|
||||
ucd_ok = False
|
||||
if selected:
|
||||
ucd_ok = self.connect_ucd(selected, rescan=False)
|
||||
app._dispatch_ui(
|
||||
app.update_connection_indicator,
|
||||
app.ucd_status_indicator, ucd_ok,
|
||||
@@ -185,7 +208,19 @@ class ConnectionController:
|
||||
|
||||
def worker():
|
||||
try:
|
||||
ucd_ok = self.connect_ucd(app.ucd_list_var.get())
|
||||
ucd_list, selected = self._refresh_ucd_selection(app.ucd_list_var.get())
|
||||
app._dispatch_ui(self._sync_ucd_combo, ucd_list, selected or "")
|
||||
if not selected:
|
||||
app._dispatch_ui(
|
||||
app.update_connection_indicator,
|
||||
app.ucd_status_indicator,
|
||||
False,
|
||||
)
|
||||
app._dispatch_ui(app.status_var.set, "未检测到 UCD 设备")
|
||||
app._dispatch_ui(self._enable_widgets)
|
||||
return
|
||||
|
||||
ucd_ok = self.connect_ucd(selected, rescan=False)
|
||||
app._dispatch_ui(
|
||||
app.update_connection_indicator,
|
||||
app.ucd_status_indicator,
|
||||
|
||||
@@ -111,6 +111,12 @@ class _UcdSdkBackend:
|
||||
|
||||
def search_device(self):
|
||||
"""搜索可用设备"""
|
||||
self._ensure_tsi_lib()
|
||||
try:
|
||||
self.lUniTAP.rescan_devices()
|
||||
except Exception:
|
||||
log.exception("UcdSdk.search_device rescan failed; reinitializing TsiLib")
|
||||
self._reinit_tsi_lib()
|
||||
available_devices = self.lUniTAP.get_list_of_available_devices()
|
||||
return available_devices if available_devices else []
|
||||
|
||||
@@ -122,6 +128,13 @@ class _UcdSdkBackend:
|
||||
if self.dev is not None or self.status:
|
||||
self._force_cleanup()
|
||||
|
||||
self._ensure_tsi_lib()
|
||||
try:
|
||||
self.lUniTAP.rescan_devices()
|
||||
except Exception:
|
||||
log.exception("UcdSdk.open rescan failed; reinitializing TsiLib")
|
||||
self._reinit_tsi_lib()
|
||||
|
||||
device_id = int(device_name.split(":")[0])
|
||||
temp_dev = self.lUniTAP.open(device_id)
|
||||
|
||||
@@ -144,6 +157,7 @@ class _UcdSdkBackend:
|
||||
|
||||
except Exception as e:
|
||||
self._force_cleanup()
|
||||
self._reinit_tsi_lib()
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
@@ -157,30 +171,29 @@ class _UcdSdkBackend:
|
||||
self._close_device_object(self.dev)
|
||||
|
||||
self._reset_state()
|
||||
self.lUniTAP = None
|
||||
|
||||
for i in range(3):
|
||||
gc.collect()
|
||||
|
||||
time.sleep(2.0)
|
||||
|
||||
self.lUniTAP = UniTAP.TsiLib()
|
||||
|
||||
self._reinit_tsi_lib()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self._reset_state()
|
||||
|
||||
try:
|
||||
self.lUniTAP = None
|
||||
gc.collect()
|
||||
time.sleep(2.0)
|
||||
self.lUniTAP = UniTAP.TsiLib()
|
||||
except Exception as init_error:
|
||||
self._reinit_tsi_lib()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
def _ensure_tsi_lib(self) -> None:
|
||||
if self.lUniTAP is None:
|
||||
self.lUniTAP = UniTAP.TsiLib()
|
||||
|
||||
def _reinit_tsi_lib(self) -> None:
|
||||
"""Destroy and recreate TsiLib so the native driver rescans USB devices."""
|
||||
self.lUniTAP = None
|
||||
for _ in range(3):
|
||||
gc.collect()
|
||||
time.sleep(2.0)
|
||||
self.lUniTAP = UniTAP.TsiLib()
|
||||
|
||||
def _reset_state(self):
|
||||
"""重置所有运行时状态(不关闭设备句柄)"""
|
||||
self.dev = None
|
||||
|
||||
Reference in New Issue
Block a user