Fixing the fragmented state of the tool manager backend

src/toolManager/backends/base.py

@@ -0,0 +1,62 @@
+import sys as _sys
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Optional, Tuple
+from platform_utils import run_cmd_stream as _run_cmd_stream
+
+class Backend(ABC):
+    @property
+    @abstractmethod
+    def name(self) -> str: ...
+
+    @abstractmethod
+    def find_executable(self, tool_id: str, version: str) -> Optional[str]: ...
+
+    @abstractmethod
+    def get_tool_version(self, exe_path: str) -> Optional[str]: ...
+
+    @abstractmethod
+    def install_package(self, package: str, version: str, fallback_versions: Optional[list[str]] = None) -> bool: ...
+
+    @abstractmethod
+    def uninstall_package(self, package: str, version: str) -> bool: ...
+
+    @abstractmethod
+    def download_file(self, url: str, filename: str) -> Path: ...
+
+    @abstractmethod
+    def extract_zip(self, archive: Path, dest_dir: Path) -> bool: ...
+
+    @abstractmethod
+    def extract_7z(self, archive: Path, dest_dir: Path) -> bool: ...
+
+    @abstractmethod
+    def run_installer(self, exe_path: str, *args: str) -> bool: ...
+
+    @property
+    @abstractmethod
+    def base_dir(self) -> Path: ...
+
+    @property
+    @abstractmethod
+    def download_dir(self) -> Path: ...
+
+    def find_executable_with_version(self, tool_id: str, version: str) -> Tuple[Optional[str], Optional[str]]:
+        exe = self.find_executable(tool_id, version)
+        return (exe, self.get_tool_version(exe)) if exe else (None, None)
+
+    def run_cmd(self, cmd, timeout=30, cwd=None, env=None):
+        import subprocess
+        try:
+            return subprocess.run(cmd, capture_output=True, text=True, shell=False,
+                timeout=timeout, encoding="utf-8", errors="ignore", cwd=cwd, env=env)
+        except: return None
+
+    def run_stream(self, cmd, timeout=900, cwd=None, env=None):
+        return _run_cmd_stream(cmd, timeout=timeout, cwd=cwd, env=env)
+
+    @property
+    def msys2_bash(self) -> Optional[Path]: return None
+
+    @property
+    def msys2_mingw_bin(self) -> Optional[Path]: return None

src/toolManager/backends/linux/__init__.py

@@ -0,0 +1,119 @@
+import os, shutil, subprocess, threading
+from pathlib import Path
+from typing import Optional
+from ..base import Backend
+from . import file_ops as _file_ops, pm as _pm, search as _search
+from platform_utils import detect_package_manager
+
+_BASH_LOCK = threading.Lock()
+
+class LinuxBackend(Backend):
+    name = "linux"
+
+    def __init__(self, base_dir: Optional[Path] = None):
+        super().__init__()
+        self._base_dir = base_dir or Path(__file__).resolve().parent.parent.parent
+        self._download_dir = self._base_dir / "Download"
+        self._download_dir.mkdir(parents=True, exist_ok=True)
+        self._pm_name: Optional[str] = detect_package_manager()
+
+    def which(self, cmd):
+        return shutil.which(cmd)
+
+    def print_status(self, state, installed, target):
+        if installed: installed = str(installed).strip().replace("\n","").replace("\r","")
+        print(f"{state}|{installed}|{target}", flush=True)
+
+    def _search_fn(self, tool_id, version):
+        _S = {
+            "kicad": lambda: _search.find_kicad(version, self.run_cmd, self.which),
+            "ngspice": lambda: _search.find_ngspice(version, self.run_cmd, self.which),
+            "ghdl": lambda: _search.find_ghdl(version, self.run_cmd, self.which),
+            "verilator": lambda: _search.find_verilator(version, self.run_cmd, self.which),
+            "llvm": lambda: _search.find_llvm(version, self.run_cmd, self.which),
+            "esim": lambda: _search.find_tool("esim", version, self.run_cmd, self.which),
+            "chocolatey": lambda: (None, None),
+            "pyqt": lambda: self._find_pyqt(version),
+        }
+        return _S.get(tool_id)
+
+    def _find_pyqt(self, version):
+        result = self.run_cmd([os.sys.executable, "-c", "import PyQt6.QtCore; print(PyQt6.QtCore.PYQT_VERSION_STR)"], timeout=10)
+        if result and result.returncode == 0:
+            ver = (result.stdout or "").strip()
+            return ("PyQt6", ver) if ver else (None, None)
+        return (None, None)
+
+    def find_executable(self, tool_id, version):
+        fn = self._search_fn(tool_id, version)
+        return fn()[0] if fn else None
+
+    def find_executable_with_version(self, tool_id, version):
+        fn = self._search_fn(tool_id, version)
+        return fn() if fn else (None, None)
+
+    def get_tool_version(self, exe_path):
+        try:
+            result = self.run_cmd([exe_path, "--version"])
+            if result and result.returncode == 0:
+                parts = (result.stdout or "").split("\n")[0].strip().split()
+                return parts[-1] if parts else None
+        except: pass
+        return None
+
+    def install_package(self, package, version, fallback_versions=None):
+        if not self._pm_name: self.print_status("not_supported", "none", version); return False
+        versions = [version] + (fallback_versions or [])
+        for v in versions:
+            if v == "latest":
+                if _pm.pm_install(self._pm_name, package, self.run_cmd): return True
+            else:
+                spec = f"{package}={v}" if self._pm_name == "apt" else package
+                if _pm.pm_install(self._pm_name, spec if self._pm_name != "pacman" else package, self.run_cmd): return True
+        return False
+
+    def uninstall_package(self, package, version):
+        if not self._pm_name: self.print_status("not_supported", "none", version); return False
+        return _pm.pm_uninstall(self._pm_name, package, self.run_cmd)
+
+    def download_file(self, url, filename):
+        result = _file_ops.download_file(url, filename, self._download_dir)
+        if result is None: raise RuntimeError(f"Download failed: {url}")
+        return result
+
+    def extract_zip(self, archive, dest_dir): return _file_ops.extract_zip(archive, dest_dir)
+    def extract_7z(self, archive, dest_dir): return _file_ops.extract_7z(archive, dest_dir, run_cmd=self.run_cmd)
+    def run_installer(self, exe_path, *args): return _file_ops.run_installer(exe_path, list(args), run_cmd=self.run_cmd)
+
+    def sudo_is_cached(self):
+        result = self.run_cmd(["sudo", "-n", "true"])
+        return result is not None and result.returncode == 0
+
+    def ensure_sudo(self, password):
+        try:
+            proc = subprocess.Popen(["sudo","-S","-v"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+            proc.communicate(input=password + "\n", timeout=10)
+            return proc.returncode == 0
+        except: return False
+
+    def clean_package_locks(self):
+        locks = {"apt": ["/var/lib/dpkg/lock-frontend","/var/lib/dpkg/lock","/var/cache/apt/archives/lock"],
+                 "pacman": ["/var/lib/pacman/db.lck"],"dnf": ["/var/run/yum.pid"],"zypper": ["/var/run/zypp.pid"]}.get(self._pm_name, [])
+        for lock in locks:
+            self.run_cmd(f"sudo lsof {lock} 2>/dev/null || sudo rm -f {lock}".split(), timeout=10)
+
+    def run_bash_script(self, script_path, *args, cwd=None):
+        with _BASH_LOCK:
+            self.clean_package_locks()
+            script = Path(script_path)
+            if not script.is_absolute(): script = self._base_dir / script_path
+            if not script.exists(): self.print_status("missing_script", "none", script_path); return False
+            rc, _ = self.run_stream(["bash", str(script)] + list(args), timeout=900, cwd=cwd or self._base_dir)
+            if rc != 0: self.print_status("script_failed", str(rc), script_path)
+            return rc == 0
+
+    @property
+    def base_dir(self) -> Path: return self._base_dir
+
+    @property
+    def download_dir(self) -> Path: return self._download_dir

src/toolManager/backends/linux/file_ops.py

@@ -0,0 +1,46 @@
+import os, shutil, subprocess, urllib.request
+from pathlib import Path
+from typing import Callable, Optional
+
+def download_file(url, filename, download_dir, print_fn=None):
+    print_fn = print_fn or print
+    dest = download_dir / filename
+    if dest.exists(): print_fn(f"[LOCAL] Using cached: {filename}"); return dest
+    def hook(n, bs, total):
+        if total > 0:
+            pct = min(n * bs * 100 / total, 100)
+            mb = min(n * bs, total) / (1024 * 1024)
+            if n % 50 == 0: print_fn(f"[DOWNLOAD] {pct:.1f}% ({mb:.1f} MB / {total / (1024 * 1024):.1f} MB)", flush=True)
+    try:
+        urllib.request.urlretrieve(url, dest, hook)
+        return dest
+    except: return None
+
+def extract_zip(archive, dest_dir, print_fn=None):
+    print_fn = print_fn or print
+    import zipfile
+    try:
+        with zipfile.ZipFile(archive, 'r') as zf:
+            print_fn(f"Extracting {archive.name}...", flush=True)
+            zf.extractall(dest_dir)
+        return True
+    except: return False
+
+def extract_7z(archive, dest_dir, run_cmd, print_fn=None):
+    print_fn = print_fn or print
+    for cmd in ["7z", "7za", "7zr", "p7zip"]:
+        if Path(cmd).exists() or os.access(f"/usr/bin/{cmd}", os.X_OK) or shutil.which(cmd):
+            print_fn(f"Extracting {archive.name}...", flush=True)
+            result = run_cmd([cmd, "x", str(archive), f"-o{dest_dir}", "-y"], timeout=300)
+            return result is not None and result.returncode == 0
+    print_fn("[ERROR] No 7z extractor found"); return False
+
+def run_installer(exe_path, args, run_cmd, print_fn=None):
+    print_fn = print_fn or print
+    try:
+        os.chmod(exe_path, os.stat(exe_path).st_mode | 0o111)
+    except OSError as e:
+        print_fn(f"[ERROR] chmod failed: {e}"); return False
+    print_fn(f"Running installer: {exe_path}...", flush=True)
+    result = run_cmd([exe_path] + args, timeout=600)
+    return result is not None and result.returncode == 0

src/toolManager/backends/linux/pm.py

@@ -0,0 +1,55 @@
+import os
+from typing import Callable, Optional
+
+_PM_COMMANDS = {
+    "apt": {"check":"apt-get","install":["apt-get","install","-y"],"uninstall":["apt-get","autoremove","-y","--purge"],
+            "search":"apt-cache search {pkg}","update":["apt-get","update"],"version_parse":""},
+    "dnf": {"check":"dnf","install":["dnf","install","-y"],"uninstall":["dnf","remove","-y"],
+            "search":"dnf search {pkg}","update":["dnf","check-update"],"version_parse":""},
+    "yum": {"check":"yum","install":["yum","install","-y"],"uninstall":["yum","remove","-y"],
+            "search":"yum search {pkg}","update":["yum","check-update"],"version_parse":""},
+    "pacman": {"check":"pacman","install":["pacman","-S","--noconfirm"],"uninstall":["pacman","-R","--noconfirm"],
+               "search":"pacman -Ss {pkg}","update":["pacman","-Sy"],"version_parse":""},
+    "zypper": {"check":"zypper","install":["zypper","install","-y"],"uninstall":["zypper","remove","-y"],
+               "search":"zypper search {pkg}","update":["zypper","refresh"],"version_parse":""},
+    "pkg": {"check":"pkg","install":["pkg","install","-y"],"uninstall":["pkg","delete","-y"],
+            "search":"pkg search {pkg}","update":["pkg","update"],"version_parse":""},
+    "xbps": {"check":"xbps-install","install":["xbps-install","-y"],"uninstall":["xbps-remove","-y"],
+             "search":"xbps-query -Rs {pkg}","update":["xbps-install","-Su"],"version_parse":""},
+}
+
+def _sudo_prefix(cmd):
+    return cmd if os.geteuid() == 0 else ["sudo"] + cmd
+
+def pm_install(pm_name, package, run_cmd, use_sudo=True, timeout=120):
+    if pm_name not in _PM_COMMANDS: return False
+    cmd = _PM_COMMANDS[pm_name]["install"] + [package]
+    if use_sudo: cmd = _sudo_prefix(cmd)
+    result = run_cmd(cmd, timeout=timeout)
+    return result is not None and result.returncode == 0
+
+def pm_uninstall(pm_name, package, run_cmd, use_sudo=True, timeout=120):
+    if pm_name not in _PM_COMMANDS: return False
+    cmd = _PM_COMMANDS[pm_name]["uninstall"] + [package]
+    if use_sudo: cmd = _sudo_prefix(cmd)
+    result = run_cmd(cmd, timeout=timeout)
+    return result is not None and result.returncode == 0
+
+def pm_list_installed(pm_name, run_cmd, use_sudo=False, timeout=30):
+    if pm_name not in _PM_COMMANDS or "list" not in _PM_COMMANDS[pm_name]: return None
+    cmd = _PM_COMMANDS[pm_name].get("list", [f"{_PM_COMMANDS[pm_name]['check']} list --installed"])
+    if use_sudo: cmd = _sudo_prefix(cmd)
+    result = run_cmd(cmd if isinstance(cmd, list) else cmd.split(), timeout=timeout)
+    return result.stdout.split("\n") if result and result.returncode == 0 else None
+
+def pm_search(pm_name, package, run_cmd, timeout=30):
+    if pm_name not in _PM_COMMANDS: return False
+    result = run_cmd(_PM_COMMANDS[pm_name]["search"].format(pkg=package).split(), timeout=timeout)
+    return result is not None and result.returncode == 0
+
+def pm_update_index(pm_name, run_cmd, use_sudo=True, timeout=120):
+    if pm_name not in _PM_COMMANDS: return False
+    cmd = _PM_COMMANDS[pm_name]["update"]
+    if use_sudo: cmd = _sudo_prefix(cmd)
+    result = run_cmd(cmd, timeout=timeout)
+    return result is not None and result.returncode == 0

src/toolManager/backends/linux/search.py

@@ -0,0 +1,60 @@
+import os, shutil
+from pathlib import Path
+from typing import Callable, Optional
+
+_STANDARD_PREFIXES = [Path("/usr/local/bin"), Path("/usr/local/sbin"),
+    Path("/opt"), Path.home() / ".local" / "bin"]
+_SNAP_BIN = Path("/snap/bin")
+_FLATPAK_BIN = Path("/var/lib/flatpak/exports/bin")
+
+def find_tool(name, version, run_cmd, which_fn, extra_prefixes=None):
+    which_fn = which_fn or shutil.which
+    exe = which_fn(name)
+    if exe: return exe, _get_version(exe, run_cmd)
+    for pfx in (extra_prefixes or []):
+        r = _find_in_prefix(name, pfx, which_fn)
+        if r: return r, _get_version(r, run_cmd)
+    for pfx in _STANDARD_PREFIXES:
+        r = _find_in_prefix(name, pfx, which_fn)
+        if r: return r, _get_version(r, run_cmd)
+    for p in [_SNAP_BIN / name, _FLATPAK_BIN / name]:
+        if p.exists() and os.access(str(p), os.X_OK):
+            return str(p), _get_version(str(p), run_cmd)
+    return (None, None)
+
+def find_kicad(version, run_cmd, which_fn):
+    return find_tool("kicad", version, run_cmd, which_fn,
+        extra_prefixes=[Path("/usr/local/kicad"), Path("/opt/kicad"), Path("/usr/share/kicad")])
+
+def find_ngspice(version, run_cmd, which_fn):
+    return find_tool("ngspice", version, run_cmd, which_fn,
+        extra_prefixes=[Path("/usr/local/ngspice"), Path("/opt/ngspice")])
+
+def find_ghdl(version, run_cmd, which_fn):
+    return find_tool("ghdl", version, run_cmd, which_fn,
+        extra_prefixes=[Path("/usr/local/ghdl")])
+
+def find_verilator(version, run_cmd, which_fn):
+    return find_tool("verilator", version, run_cmd, which_fn,
+        extra_prefixes=[Path("/usr/local/verilator")])
+
+def find_llvm(version, run_cmd, which_fn):
+    result = find_tool("clang", version, run_cmd, which_fn,
+        extra_prefixes=[Path("/usr/local/llvm")])
+    if result[0]: return result
+    return find_tool("llvm-config", version, run_cmd, which_fn)
+
+def _find_in_prefix(exe_name, prefix, which_fn):
+    if not prefix.exists(): return None
+    for sub in ("", "bin", "sbin"):
+        c = prefix / sub / exe_name if sub else prefix / exe_name
+        if c.exists() and os.access(str(c), os.X_OK): return str(c)
+    return None
+
+def _get_version(exe_path, run_cmd):
+    if not run_cmd: return None
+    result = run_cmd([exe_path, "--version"], timeout=10)
+    if result and result.returncode == 0:
+        parts = (result.stdout or "").split("\n")[0].strip().split()
+        return parts[-1] if parts else None
+    return None

src/toolManager/backends/tools/__init__.py

@@ -0,0 +1,14 @@
+from .tools import TOOL_FUNCS
+
+def run(tool_id, cmd, version, backend, upgrade=False):
+    funcs = TOOL_FUNCS.get(tool_id)
+    if not funcs:
+        backend.print_status("not_supported", "none", version); return
+    if cmd == "check":
+        funcs["check"](version, backend)
+    elif cmd in ("install", "update"):
+        funcs["install"](version, cmd == "update" or upgrade, backend)
+    elif cmd == "uninstall":
+        fn = funcs.get("uninstall")
+        if fn: fn(version, backend)
+        else: backend.print_status("not_supported", "none", "none")