Przeglądaj źródła

摄像头并发 ping 检测,替代串行 ping 提升实时性

lsfoo 3 miesięcy temu
rodzic
commit
f3ae34eb4e
1 zmienionych plików z 10 dodań i 11 usunięć
  1. 10 11
      hookvision/main_window.py

+ 10 - 11
hookvision/main_window.py

@@ -1241,25 +1241,24 @@ class MainWindow(QMainWindow):
         ips = self._collect_camera_ips()
         if not ips:
             return
-        results = {}
-        for ip in ips:
+
+        def ping_one(ip: str) -> tuple:
             try:
                 if sys.platform == "win32":
                     r = subprocess.run(["ping", "-n", "1", ip], capture_output=True, timeout=5, text=True, creationflags=subprocess.CREATE_NO_WINDOW)
                 else:
                     r = subprocess.run(["ping", "-c", "1", ip], capture_output=True, timeout=5, text=True)
                 if r.returncode != 0:
-                    results[ip] = -1
-                else:
-                    m = re.search(r'time[=<]\s*(\d+(?:\.\d+)?)\s*ms', r.stdout, re.IGNORECASE)
-                    if m:
-                        results[ip] = float(m.group(1))
-                    else:
-                        results[ip] = 0
+                    return (ip, -1)
+                m = re.search(r'time[=<]\s*(\d+(?:\.\d+)?)\s*ms', r.stdout, re.IGNORECASE)
+                return (ip, float(m.group(1)) if m else 0)
             except subprocess.TimeoutExpired:
-                results[ip] = -1
+                return (ip, -1)
             except Exception:
-                results[ip] = -1
+                return (ip, -1)
+
+        with ThreadPoolExecutor(max_workers=len(ips)) as pool:
+            results = dict(pool.map(ping_one, ips))
         self.camera_signal_signal.emit(results)
 
     def _on_camera_signal(self, results: dict):