fix: Replace Windows tasklist with Toolhelp32 API for reliable process detection (#21)

Replaces unreliable tasklist subprocess calls with proper Windows Toolhelp32 API.

- Uses CreateToolhelp32Snapshot/Process32FirstW/Process32NextW for process enumeration
- Adds pointer-size fallback for wintypes.ULONG_PTR missing in KiCad's embedded Python
- Fixes check_kicad_ui intermittent timeouts and false negatives on Windows
- Adds encoding/timeout fixes for 'where' command calls

Fixes #20

Co-authored-by: jbjardine <167578676+jbjardine@users.noreply.github.com>
This commit is contained in:
jbjardine
2025-12-28 17:29:09 +01:00
committed by GitHub
parent c65600049e
commit 834d05130c

View File

@@ -3,22 +3,77 @@ KiCAD Process Management Utilities
Detects if KiCAD is running and provides auto-launch functionality. Detects if KiCAD is running and provides auto-launch functionality.
""" """
import os import os
import subprocess import subprocess
import logging import logging
import platform import platform
import time import time
from pathlib import Path import ctypes
from ctypes import wintypes
from pathlib import Path
from typing import Optional, List from typing import Optional, List
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class KiCADProcessManager: class KiCADProcessManager:
"""Manages KiCAD process detection and launching""" """Manages KiCAD process detection and launching"""
@staticmethod @staticmethod
def is_running() -> bool: def _windows_list_processes() -> List[dict]:
"""List running processes on Windows using Toolhelp API."""
processes: List[dict] = []
try:
TH32CS_SNAPPROCESS = 0x00000002
try:
ulong_ptr = wintypes.ULONG_PTR # type: ignore[attr-defined]
except AttributeError:
ulong_ptr = ctypes.c_ulonglong if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_ulong
class PROCESSENTRY32W(ctypes.Structure):
_fields_ = [
("dwSize", wintypes.DWORD),
("cntUsage", wintypes.DWORD),
("th32ProcessID", wintypes.DWORD),
("th32DefaultHeapID", ulong_ptr),
("th32ModuleID", wintypes.DWORD),
("cntThreads", wintypes.DWORD),
("th32ParentProcessID", wintypes.DWORD),
("pcPriClassBase", wintypes.LONG),
("dwFlags", wintypes.DWORD),
("szExeFile", wintypes.WCHAR * wintypes.MAX_PATH),
]
CreateToolhelp32Snapshot = ctypes.windll.kernel32.CreateToolhelp32Snapshot
Process32FirstW = ctypes.windll.kernel32.Process32FirstW
Process32NextW = ctypes.windll.kernel32.Process32NextW
CloseHandle = ctypes.windll.kernel32.CloseHandle
snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
if snapshot == wintypes.HANDLE(-1).value:
return processes
entry = PROCESSENTRY32W()
entry.dwSize = ctypes.sizeof(PROCESSENTRY32W)
if Process32FirstW(snapshot, ctypes.byref(entry)):
while True:
processes.append({
"pid": str(entry.th32ProcessID),
"name": entry.szExeFile,
"command": entry.szExeFile
})
if not Process32NextW(snapshot, ctypes.byref(entry)):
break
CloseHandle(snapshot)
except Exception as e:
logger.error(f"Error listing Windows processes: {e}")
return processes
@staticmethod
def is_running() -> bool:
""" """
Check if KiCAD is currently running Check if KiCAD is currently running
@@ -68,13 +123,13 @@ class KiCADProcessManager:
) )
return result.returncode == 0 return result.returncode == 0
elif system == "Windows": elif system == "Windows":
result = subprocess.run( processes = KiCADProcessManager._windows_list_processes()
["tasklist", "/FI", "IMAGENAME eq pcbnew.exe"], for proc in processes:
capture_output=True, name = (proc.get("name") or "").lower()
text=True if name in ("pcbnew.exe", "kicad.exe"):
) return True
return "pcbnew.exe" in result.stdout return False
else: else:
logger.warning(f"Process detection not implemented for {system}") logger.warning(f"Process detection not implemented for {system}")
@@ -96,11 +151,14 @@ class KiCADProcessManager:
# Try to find executable in PATH first # Try to find executable in PATH first
for cmd in ["pcbnew", "kicad"]: for cmd in ["pcbnew", "kicad"]:
result = subprocess.run( result = subprocess.run(
["which", cmd] if system != "Windows" else ["where", cmd], ["which", cmd] if system != "Windows" else ["where", cmd],
capture_output=True, capture_output=True,
text=True text=True,
) encoding="mbcs" if system == "Windows" else None,
errors="ignore" if system == "Windows" else None,
timeout=5 if system == "Windows" else None
)
if result.returncode == 0: if result.returncode == 0:
path = result.stdout.strip().split("\n")[0] path = result.stdout.strip().split("\n")[0]
logger.info(f"Found KiCAD executable: {path}") logger.info(f"Found KiCAD executable: {path}")
@@ -235,29 +293,17 @@ class KiCADProcessManager:
"command": " ".join(parts[10:]) "command": " ".join(parts[10:])
}) })
elif system == "Windows": elif system == "Windows":
result = subprocess.run( for proc in KiCADProcessManager._windows_list_processes():
["tasklist", "/V", "/FO", "CSV"], name = (proc.get("name") or "").lower()
capture_output=True, if "pcbnew" in name or "kicad" in name:
text=True processes.append(proc)
)
import csv
reader = csv.reader(result.stdout.split("\n"))
for row in reader:
if row and len(row) > 0:
if "pcbnew" in row[0].lower() or "kicad" in row[0].lower():
processes.append({
"pid": row[1] if len(row) > 1 else "unknown",
"name": row[0],
"command": row[0]
})
except Exception as e: except Exception as e:
logger.error(f"Error getting process info: {e}") logger.error(f"Error getting process info: {e}")
return processes return processes
def check_and_launch_kicad(project_path: Optional[Path] = None, auto_launch: bool = True) -> dict: def check_and_launch_kicad(project_path: Optional[Path] = None, auto_launch: bool = True) -> dict:
""" """
Check if KiCAD is running and optionally launch it Check if KiCAD is running and optionally launch it