"""Bounded, privacy-safe compatibility details for launcher telemetry.""" from __future__ import annotations from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone import json import os from pathlib import Path import platform import re import shutil import struct import subprocess from .paths import AppPaths PROBE_TIMEOUT_SECONDS = 3.0 MAX_MARKER_BYTES = 256 * 1024 MAX_DLL_SCAN_BYTES = 16 * 1024 * 1024 MAX_PROBE_OUTPUT_CHARACTERS = 64 * 1024 _VULKAN_MANIFEST_DIRECTORIES = ( Path("/etc/vulkan/icd.d"), Path("/usr/local/share/vulkan/icd.d"), Path("/usr/share/vulkan/icd.d"), ) _LIBRARY_ROOTS = ( Path("/lib"), Path("/usr/lib"), Path("/usr/local/lib"), ) _PCI_DATABASES = ( Path("/usr/share/hwdata/pci.ids"), Path("/usr/share/misc/pci.ids"), Path("/usr/share/pci.ids"), ) _GPU_VENDORS = { 0x1002: "AMD", 0x10DE: "NVIDIA", 0x8086: "Intel", } _DXVK_VERSION_PATTERN = re.compile( r"\bdxvk(?:\s+version|[- /:])?\s*v?(\d+\.\d+(?:\.\d+){0,2})\b", re.IGNORECASE, ) _VERSION_PATTERN = re.compile(r"\d+(?:\.\d+){1,3}") @dataclass(frozen=True) class ProbeResult: """Small result surface for bounded local compatibility probes.""" returncode: int stdout: str stderr: str ProbeRunner = Callable[[Sequence[str], Mapping[str, str], float], ProbeResult | None] def collect_extended_environment( paths: AppPaths, environment: Mapping[str, str], *, probe_runner: ProbeRunner | None = None, sysfs_drm: Path = Path("/sys/class/drm"), vulkan_manifest_directories: Sequence[Path] = _VULKAN_MANIFEST_DIRECTORIES, library_roots: Sequence[Path] = _LIBRARY_ROOTS, pci_databases: Sequence[Path] = _PCI_DATABASES, now: datetime | None = None, ) -> dict[str, object]: """Collect compatibility data once without returning paths or device IDs.""" run_probe = probe_runner or _run_probe marker = _read_json_object(paths.game_runtime_marker) prefix = _selected_prefix(paths, marker) runtime = _runtime_info( paths, marker, prefix, environment, run_probe, vulkan_manifest_directories=vulkan_manifest_directories, library_roots=library_roots, now=now or datetime.now(timezone.utc), ) gpu = _gpu_info( runtime.pop("_vulkan_summary", None), environment=environment, sysfs_drm=sysfs_drm, pci_databases=pci_databases, ) dependencies = _dependency_health( environment, prefix, runtime, ) result: dict[str, object] = { "runtime": runtime, "dependencies": dependencies, } display = _display_info(sysfs_drm) if display: result["display"] = display if gpu: result["gpu"] = gpu return result def _runtime_info( paths: AppPaths, marker: Mapping[str, object] | None, prefix: Path | None, environment: Mapping[str, str], probe_runner: ProbeRunner, *, vulkan_manifest_directories: Sequence[Path], library_roots: Sequence[Path], now: datetime, ) -> dict[str, object]: runtime_marker = _object_field(marker, "runtime") runtime_kind = _short_string(runtime_marker.get("kind")) if runtime_marker else None result: dict[str, object] = { "kind": runtime_kind or "unknown", "wine_vendor": _wine_vendor(runtime_kind, runtime_marker, paths), } wine_command = _wine_command(runtime_marker, environment) if wine_command is not None: wine_probe = probe_runner( (wine_command, "--version"), environment, PROBE_TIMEOUT_SECONDS, ) wine_build = _first_probe_line(wine_probe) if wine_build: result["wine_build"] = wine_build wine_version = _wine_version(wine_build) if wine_version: result["wine"] = wine_version if runtime_marker: proton_name = _proton_name(runtime_marker.get("proton_path")) if proton_name: result["proton_version"] = proton_name if prefix is not None: prefix_architecture = _wine_prefix_architecture(prefix) if prefix_architecture: result["prefix_arch"] = prefix_architecture configured_at = _marker_timestamp(marker) if configured_at is not None: result["prefix_created_at"] = configured_at.isoformat() result["prefix_age_days"] = max(0, (now - configured_at).days) result.update( _dxvk_info( prefix, packaged_version=_proton_dxvk_version(runtime_marker), ) ) else: result.update( { "dxvk_enabled": False, "dxvk_install_status": "missing", } ) result.update(_runtime_component_info(paths.launcher_prefix, prefix)) vulkan = _vulkan_info( environment, probe_runner, manifest_directories=vulkan_manifest_directories, library_roots=library_roots, ) result.update(vulkan) return result def _vulkan_info( environment: Mapping[str, str], probe_runner: ProbeRunner, *, manifest_directories: Sequence[Path], library_roots: Sequence[Path], ) -> dict[str, object]: loader_architectures = _vulkan_loader_architectures(library_roots) manifest_architectures: set[int] = set() api_versions: list[str] = [] for manifest in _vulkan_manifests(environment, manifest_directories): parsed = _read_json_object(manifest) icd = _object_field(parsed, "ICD") if icd is None: continue api_version = _short_string(icd.get("api_version")) if api_version and _VERSION_PATTERN.fullmatch(api_version): api_versions.append(api_version) library = _short_string(icd.get("library_path")) architecture = _manifest_library_architecture( manifest, library, library_roots, ) if architecture is not None: manifest_architectures.add(architecture) vulkaninfo = shutil.which("vulkaninfo", path=environment.get("PATH", "")) probe = ( probe_runner( (vulkaninfo, "--summary"), environment, PROBE_TIMEOUT_SECONDS, ) if vulkaninfo else None ) functional_64bit = probe is not None and probe.returncode == 0 summary = _probe_text(probe) if functional_64bit else "" if summary: api_version = _summary_value(summary, "apiVersion") if api_version: api_versions.append(api_version) instance_version = re.search( r"Vulkan\s+Instance\s+Version\s*:\s*(\S+)", summary, re.IGNORECASE, ) if instance_version: api_versions.append(instance_version.group(1)) available_64bit = functional_64bit or ( 64 in loader_architectures and 64 in manifest_architectures ) available_32bit = ( 32 in loader_architectures and 32 in manifest_architectures ) result: dict[str, object] = { "vulkan_available": available_64bit, "vulkan_64bit": available_64bit, "vulkan_32bit": available_32bit, "vulkan_64bit_detection": ( "vulkaninfo" if functional_64bit else "loader_and_icd" ), "vulkan_32bit_detection": "loader_and_icd", } best_api_version = _maximum_version(api_versions) if best_api_version: result["vulkan_api_version"] = best_api_version driver_info = _summary_value(summary, "driverInfo") if driver_info: result["vulkan_driver_version"] = _driver_version(driver_info) or driver_info if summary: result["_vulkan_summary"] = summary return result def _gpu_info( vulkan_summary: object, *, environment: Mapping[str, str], sysfs_drm: Path, pci_databases: Sequence[Path], ) -> dict[str, object]: summary = vulkan_summary if isinstance(vulkan_summary, str) else "" vulkan_gpus = _vulkan_gpu_summaries(summary) has_gpu_blocks = re.search(r"(?mi)^\s*GPU\d+\s*:\s*$", summary) is not None if vulkan_gpus: primary_summary = vulkan_gpus[0] elif has_gpu_blocks: primary_summary = "" else: primary_summary = summary vendor_id = _hex_summary_value(primary_summary, "vendorID") model = _summary_value(primary_summary, "deviceName") driver_name = _summary_value(primary_summary, "driverName") driver_info = _summary_value(primary_summary, "driverInfo") sysfs_gpus = _sysfs_gpus(sysfs_drm) sysfs = next( (gpu for gpu in sysfs_gpus if gpu.get("boot_vga") is True), sysfs_gpus[0] if sysfs_gpus else None, ) if vendor_id is None and sysfs is not None: vendor_id = sysfs.get("vendor_id") if model is None and sysfs is not None: model = _short_string(sysfs.get("model")) if model is None and sysfs is not None: device_id = sysfs.get("device_id") if isinstance(vendor_id, int) and isinstance(device_id, int): model = _pci_model(vendor_id, device_id, pci_databases) if driver_name is None and sysfs is not None: driver_name = _short_string(sysfs.get("driver")) if driver_info is None and sysfs is not None: driver_info = _short_string(sysfs.get("driver_version")) vendor = _GPU_VENDORS.get(vendor_id) if isinstance(vendor_id, int) else None if vendor is None: vendor = _vendor_from_model(model) driver = _friendly_driver_name(driver_name, driver_info) driver_version = _driver_version(driver_info) result: dict[str, object] = {} if vendor: result["vendor"] = vendor if model: result["model"] = model[:255] if driver: result["driver"] = driver[:128] if driver_version: result["driver_version"] = driver_version[:128] if sysfs is not None and isinstance(sysfs.get("vram_mb"), int): result["vram_mb"] = sysfs["vram_mb"] summary_gpu_count = len(vulkan_gpus) if summary_gpu_count == 0 and model: summary_gpu_count = 1 gpu_count = max(len(sysfs_gpus), summary_gpu_count) if gpu_count: result["count"] = gpu_count if vendor: result["primary_vendor"] = vendor if model: result["primary_model"] = model[:255] vendor_ids = { vendor for gpu in sysfs_gpus if isinstance((vendor := gpu.get("vendor_id")), int) } for gpu_summary in vulkan_gpus: vendor_ids.update(_hex_summary_values(gpu_summary, "vendorID")) device_types: set[str] = set() for gpu_summary in vulkan_gpus: device_type = _summary_value(gpu_summary, "deviceType") if isinstance(device_type, str): device_types.add(device_type.casefold()) integrated_and_discrete = ( any("integrated" in value for value in device_types) and any("discrete" in value for value in device_types) ) result["hybrid_graphics"] = gpu_count > 1 and ( len(vendor_ids) > 1 or integrated_and_discrete ) result["offload_detected"] = _gpu_offload_detected(environment) return result def _dependency_health( environment: Mapping[str, str], prefix: Path | None, runtime: Mapping[str, object], ) -> dict[str, object]: tools: dict[str, bool] = {} required: list[str] = [] missing: list[str] = [] container_ready: bool | None = None try: from .doctor import inspect_host report = inspect_host(environment) for item in report.tools: available = item.path is not None tools[item.name] = available if item.required: required.append(item.name) if not available: missing.append(item.name) container_ready = report.container_host.ready report_ok: bool | None = report.ok except Exception: report_ok = None wine_available = tools.get("wine") if wine_available is None: wine_available = _wine_command(None, environment) is not None wine_32bit = _wine_32bit_available(prefix) vulkan_64bit = _optional_bool(runtime.get("vulkan_64bit")) vulkan_32bit = _optional_bool(runtime.get("vulkan_32bit")) font_dependencies = _font_dependencies_present(prefix) archive_tools = tools.get("cabextract") if archive_tools is None: archive_tools = shutil.which( "cabextract", path=environment.get("PATH", ""), ) is not None shared_libraries = ( vulkan_64bit and vulkan_32bit if platform.machine().casefold() in {"x86_64", "amd64"} else vulkan_64bit ) healthy = report_ok if healthy is not None: healthy = bool(healthy and shared_libraries) result: dict[str, object] = { "healthy": healthy, "required_count": len(required), "available_required_count": len(required) - len(missing), "missing_required": missing, "tools": tools, "wine_available": wine_available, "wine_32bit_available": wine_32bit, "vulkan_available": bool(runtime.get("vulkan_available")), "vulkan_64bit_available": vulkan_64bit, "vulkan_32bit_available": vulkan_32bit, "font_dependencies_present": font_dependencies, "dotnet_present": bool(runtime.get("dotnet_present")), "archive_tools_present": archive_tools, "required_shared_libraries_present": shared_libraries, "directx_runtime_present": bool(runtime.get("directx_runtime_present")), } if container_ready is not None: result["container_ready"] = container_ready return result def _selected_prefix( paths: AppPaths, marker: Mapping[str, object] | None, ) -> Path | None: runtime = _object_field(marker, "runtime") kind = _short_string(runtime.get("kind")) if runtime else None preferred = paths.umu_prefix if kind == "umu" else paths.proton_prefix for candidate in (preferred, paths.proton_prefix, paths.umu_prefix): if (candidate / "system.reg").is_file() or ( candidate / "drive_c/windows" ).is_dir(): return candidate return None def _wine_command( runtime: Mapping[str, object] | None, environment: Mapping[str, str], ) -> str | None: if runtime: proton_path = _short_string(runtime.get("proton_path")) if proton_path: root = Path(proton_path).expanduser().parent for relative in ( "files/bin/wine64", "files/bin/wine", "dist/bin/wine64", "dist/bin/wine", ): candidate = root / relative if candidate.is_file() and os.access(candidate, os.X_OK): return str(candidate) executable = _short_string(runtime.get("executable")) if executable and Path(executable).name.casefold() in {"wine", "wine64"}: return executable return shutil.which("wine", path=environment.get("PATH", "")) def _wine_vendor( runtime_kind: str | None, runtime: Mapping[str, object] | None, paths: AppPaths, ) -> str: if runtime_kind in {"proton7", "proton7-container", "umu"}: return "proton" if runtime_kind != "umu" else "umu" if runtime: executable = _short_string(runtime.get("executable")) if executable: lowered = executable.casefold() if "lutris" in lowered: return "lutris" try: if Path(executable).resolve().is_relative_to(paths.data.resolve()): return "pristontale-runtime" except (OSError, RuntimeError): pass return "system" def _wine_version(build: str) -> str | None: match = re.search(r"\bwine(?:64)?[- ]v?(\d+(?:\.\d+){1,3})", build, re.IGNORECASE) if match: return match.group(1) match = _VERSION_PATTERN.search(build) return match.group(0) if match else None def _proton_name(value: object) -> str | None: path = _short_string(value) if not path: return None name = Path(path).parent.name.strip() if "proton" not in name.casefold(): return None version = re.search(r"\d+(?:[._-]\d+)+(?:[A-Za-z0-9._-]*)?", name) return version.group(0)[:64] if version else None def _wine_prefix_architecture(prefix: Path) -> str | None: try: with (prefix / "system.reg").open("r", encoding="utf-8", errors="replace") as handle: text = handle.read(4096) except OSError: return None match = re.search(r"(?m)^#arch=(win32|win64)\s*$", text, re.IGNORECASE) return match.group(1).casefold() if match else None def _marker_timestamp(marker: Mapping[str, object] | None) -> datetime | None: if marker is None: return None value = _short_string(marker.get("configured_at")) if not value: return None try: parsed = datetime.fromisoformat(value) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _dxvk_info( prefix: Path, *, packaged_version: str | None = None, ) -> dict[str, object]: components: dict[str, str | None] = {} identified_components: set[str] = set() files_present = 0 dxvk_components = 0 invalid = 0 for name in ("d3d9", "d3d11", "dxgi"): versions: list[str] = [] component_present = False component_is_dxvk = False for system_directory in ("syswow64", "system32"): dll = prefix / "drive_c/windows" / system_directory / f"{name}.dll" if dll.is_symlink() and not dll.exists(): invalid += 1 continue if not dll.is_file(): continue component_present = True is_dxvk, version = _dxvk_dll_details(dll) component_is_dxvk = component_is_dxvk or is_dxvk if version: versions.append(version) if component_present: files_present += 1 if component_is_dxvk: dxvk_components += 1 identified_components.add(name) components[name] = _maximum_version(versions) if invalid: status = "invalid" elif files_present == 0: status = "missing" elif dxvk_components == 0: status = "unknown" elif dxvk_components < len(components): status = "partial" else: status = "installed" result: dict[str, object] = { "dxvk_enabled": dxvk_components > 0 and status != "invalid", "dxvk_install_status": status, } all_versions = [ version for version in components.values() if isinstance(version, str) ] dxvk_version = _maximum_version(all_versions) or packaged_version if dxvk_version: result["dxvk_version"] = dxvk_version for name, version in components.items(): if version is None and name in identified_components: version = packaged_version if version: result[f"dxvk_{name}_version"] = version return result def _proton_dxvk_version(runtime: Mapping[str, object] | None) -> str | None: if runtime is None: return None proton_path = _short_string(runtime.get("proton_path")) if not proton_path: return None root = Path(proton_path).expanduser().parent for relative in ( "dist/lib/wine/dxvk/version", "dist/lib64/wine/dxvk/version", "files/lib/wine/dxvk/version", "files/lib64/wine/dxvk/version", ): value = _read_short_text(root / relative) if not value: continue match = re.search( r"\bdxvk\s+\(v([0-9][0-9A-Za-z.+-]{1,63})\)", value, re.IGNORECASE, ) if match: return match.group(1) return None def _dxvk_dll_details(path: Path) -> tuple[bool, str | None]: try: with path.open("rb") as handle: content = handle.read(MAX_DLL_SCAN_BYTES + 1) except OSError: return False, None if len(content) > MAX_DLL_SCAN_BYTES: content = content[:MAX_DLL_SCAN_BYTES] texts = ( content.decode("latin-1", errors="ignore"), content.decode("utf-16-le", errors="ignore"), ) versions: list[str] = [] is_dxvk = False for text in texts: is_dxvk = is_dxvk or "dxvk" in text.casefold() versions.extend(match.group(1) for match in _DXVK_VERSION_PATTERN.finditer(text)) version = _maximum_version(versions) if version is None and is_dxvk: version = _pe_fixed_file_version(content) return is_dxvk, version def _pe_fixed_file_version( content: bytes, *, maximum_part: int = 10_000, ) -> str | None: """Read the numeric VS_FIXEDFILEINFO version without a PE dependency.""" signature = b"\xbd\x04\xef\xfe" offset = content.find(signature) while offset >= 0 and offset + 52 <= len(content): try: values = struct.unpack_from("<13I", content, offset) except struct.error: return None structure_version = values[1] major = values[2] >> 16 minor = values[2] & 0xFFFF build = values[3] >> 16 revision = values[3] & 0xFFFF parts = [major, minor, build, revision] if structure_version >> 16 == 1 and 0 < major < 100 and all( part < maximum_part for part in parts ): while len(parts) > 2 and parts[-1] == 0: parts.pop() return ".".join(str(part) for part in parts) offset = content.find(signature, offset + len(signature)) return None def _directx_runtime_present(prefix: Path) -> bool: directory = prefix / "drive_c/windows/syswow64" return all((directory / f"{name}.dll").is_file() for name in ("d3dx9_43", "d3dcompiler_43")) def _runtime_component_info( launcher_prefix: Path, game_prefix: Path | None, ) -> dict[str, object]: prefixes = tuple( dict.fromkeys( prefix for prefix in (game_prefix, launcher_prefix) if prefix is not None ) ) dotnet_runtime_versions: list[str] = [] dotnet_desktop_versions: list[str] = [] dotnet_framework_versions: list[str] = [] for prefix in prefixes: dotnet_root = prefix / "drive_c/Program Files/dotnet/shared" dotnet_runtime_versions.extend( _directory_versions(dotnet_root / "Microsoft.NETCore.App") ) dotnet_desktop_versions.extend( _directory_versions(dotnet_root / "Microsoft.WindowsDesktop.App") ) for framework_name in ("Framework", "Framework64"): dotnet_framework_versions.extend( _directory_versions( prefix / "drive_c/windows/Microsoft.NET" / framework_name ) ) dotnet_runtime = _maximum_version(dotnet_runtime_versions) dotnet_desktop = _maximum_version(dotnet_desktop_versions) dotnet_framework = _maximum_version(dotnet_framework_versions) vc_present, vc_version = _vc_runtime_info(prefixes) directx_present = ( _directx_runtime_present(game_prefix) if game_prefix is not None else False ) winetricks_components: list[str] = [] if game_prefix is not None: system = game_prefix / "drive_c/windows/syswow64" for name in ("d3dcompiler_43", "d3dx9_43"): if (system / f"{name}.dll").is_file(): winetricks_components.append(name) if dotnet_desktop: winetricks_components.append( "dotnetdesktop6" if dotnet_desktop.startswith("6.") else "dotnetdesktop" ) if vc_present: winetricks_components.append("vcrun2015-2022") runtime_components: dict[str, object] = { "directx": directx_present, "vc_runtime": vc_present, } if dotnet_runtime or dotnet_framework: runtime_components["dotnet_runtime"] = dotnet_runtime or dotnet_framework if dotnet_desktop: runtime_components["dotnet_desktop"] = dotnet_desktop if vc_version: runtime_components["vc_runtime_version"] = vc_version result: dict[str, object] = { "dotnet_present": bool( dotnet_runtime or dotnet_desktop or dotnet_framework or (game_prefix is not None and _dotnet_present(game_prefix)) ), "vc_runtime_present": vc_present, "directx_runtime_present": directx_present, "winetricks_components": winetricks_components, "runtime_components": runtime_components, } if dotnet_runtime or dotnet_framework: result["dotnet_runtime_version"] = dotnet_runtime or dotnet_framework if dotnet_desktop: result["dotnet_desktop_runtime_version"] = dotnet_desktop if vc_version: result["vc_runtime_version"] = vc_version return result def _directory_versions(root: Path) -> tuple[str, ...]: try: iterator = root.iterdir() except OSError: return () versions: list[str] = [] try: for index, entry in enumerate(iterator): if index >= 64: break try: is_directory = entry.is_dir() except OSError: continue if not is_directory: continue version = entry.name.removeprefix("v") if re.fullmatch(r"\d+(?:\.\d+){0,3}", version): versions.append(version) except OSError: return tuple(versions) return tuple(versions) def _vc_runtime_info(prefixes: Sequence[Path]) -> tuple[bool, str | None]: versions: list[str] = [] present = False for prefix in prefixes: for system_directory in ("syswow64", "system32"): system = prefix / "drive_c/windows" / system_directory for name in ("vcruntime140.dll", "vcruntime140_1.dll", "msvcp140.dll"): dll = system / name if not dll.is_file(): continue present = True version = _dll_file_version(dll) if version: versions.append(version) return present, _maximum_version(versions) def _dll_file_version(path: Path) -> str | None: try: with path.open("rb") as handle: content = handle.read(MAX_DLL_SCAN_BYTES) except OSError: return None return _pe_fixed_file_version(content, maximum_part=1_000_000) def _dotnet_present(prefix: Path) -> bool: framework = prefix / "drive_c/windows/Microsoft.NET" try: return any(path.is_file() for path in framework.glob("Framework*/v*/mscorlib.dll")) except OSError: return False def _wine_32bit_available(prefix: Path | None) -> bool | None: if prefix is None: return None architecture = _wine_prefix_architecture(prefix) if architecture == "win32": return True return (prefix / "drive_c/windows/syswow64").is_dir() def _font_dependencies_present(prefix: Path | None) -> bool: candidates = [ Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"), Path("/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf"), ] if prefix is not None: fonts = prefix / "drive_c/windows/Fonts" candidates.extend((fonts / "arial.ttf", fonts / "tahoma.ttf")) return any(candidate.is_file() for candidate in candidates) def _vulkan_manifests( environment: Mapping[str, str], directories: Sequence[Path], ) -> tuple[Path, ...]: candidates: list[Path] = [] explicit = environment.get("VK_ICD_FILENAMES", "") for value in explicit.split(os.pathsep): if value: candidates.append(Path(value).expanduser()) for directory in directories: try: candidates.extend(sorted(directory.glob("*.json"))) except OSError: continue seen: set[Path] = set() result: list[Path] = [] for candidate in candidates: try: resolved = candidate.resolve() except (OSError, RuntimeError): continue if resolved in seen or not resolved.is_file(): continue seen.add(resolved) result.append(resolved) return tuple(result) def _vulkan_loader_architectures(library_roots: Sequence[Path]) -> set[int]: architectures: set[int] = set() for root in library_roots: for candidate in _library_candidates(root, "libvulkan.so"): architecture = _elf_architecture(candidate) if architecture is not None: architectures.add(architecture) return architectures def _manifest_library_architecture( manifest: Path, library: str | None, library_roots: Sequence[Path], ) -> int | None: if library: path = Path(library) if path.is_absolute(): architecture = _elf_architecture(path) if architecture is not None: return architecture else: candidates = [manifest.parent / path] for root in library_roots: candidates.extend(_library_candidates(root, path.name)) for candidate in candidates: architecture = _elf_architecture(candidate) if architecture is not None: return architecture lowered = manifest.name.casefold() if any(token in lowered for token in ("i386", "i486", "i586", "i686", "x86_32")): return 32 if any(token in lowered for token in ("x86_64", "amd64")): return 64 return None def _library_candidates(root: Path, prefix: str) -> tuple[Path, ...]: try: direct = tuple(root.glob(f"{prefix}*")) children = tuple(root.glob(f"*/{prefix}*")) except OSError: return () return direct + children def _elf_architecture(path: Path) -> int | None: try: with path.open("rb") as handle: header = handle.read(5) except OSError: return None if len(header) < 5 or header[:4] != b"\x7fELF": return None if header[4] == 1: return 32 if header[4] == 2: return 64 return None def _sysfs_gpus(sysfs_drm: Path) -> tuple[dict[str, object], ...]: try: cards = sorted( path for path in sysfs_drm.glob("card[0-9]*") if path.name[4:].isdigit() ) except OSError: return () result_gpus: list[dict[str, object]] = [] for card in cards[:16]: device = card / "device" vendor_id = _read_hex(device / "vendor") device_id = _read_hex(device / "device") if vendor_id is None or device_id is None: continue result: dict[str, object] = { "vendor_id": vendor_id, "device_id": device_id, } if _read_decimal(device / "boot_vga") == 1: result["boot_vga"] = True for name in ("product_name", "label"): value = _read_short_text(device / name) if value: result["model"] = value break try: driver = (device / "driver").resolve(strict=True).name except (OSError, RuntimeError): driver = None if driver: result["driver"] = driver version = _read_short_text(Path("/sys/module") / driver / "version") if version: result["driver_version"] = version vram_bytes = _read_decimal(device / "mem_info_vram_total") if vram_bytes is not None and vram_bytes > 0: result["vram_mb"] = round(vram_bytes / (1024 * 1024)) result_gpus.append(result) return tuple(result_gpus) def _display_info(sysfs_drm: Path) -> dict[str, object]: monitor_count = _connected_monitor_count(sysfs_drm) return {"monitor_count": monitor_count} if monitor_count is not None else {} def _connected_monitor_count(sysfs_drm: Path) -> int | None: try: connectors = sorted(sysfs_drm.glob("card[0-9]*-*/status"))[:32] except OSError: return None if not connectors: return None connected = 0 for status in connectors: value = _read_short_text(status) if value and value.casefold() == "connected": connected += 1 return connected def _pci_model( vendor_id: int, device_id: int, databases: Sequence[Path], ) -> str | None: vendor_key = f"{vendor_id:04x}" device_key = f"{device_id:04x}" for database in databases: try: lines = database.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: continue inside_vendor = False for line in lines: if not line or line.startswith("#"): continue if not line.startswith("\t"): parts = line.split(None, 1) inside_vendor = bool(parts and parts[0].casefold() == vendor_key) continue if not inside_vendor or line.startswith("\t\t"): continue parts = line.strip().split(None, 1) if len(parts) == 2 and parts[0].casefold() == device_key: return parts[1].strip()[:255] return None def _friendly_driver_name(name: str | None, info: str | None) -> str | None: lowered = (name or "").casefold() combined = f"{name or ''} {info or ''}".casefold() if "radv" in combined: return "Mesa RADV" if "anv" in combined or "intel open-source mesa" in combined: return "Mesa ANV" if "nvidia" in combined: return "NVIDIA" if "nouveau" in combined: return "Mesa NVK" if "nvk" in combined else "Nouveau" if lowered in {"amdgpu", "i915", "xe"}: return lowered return name.strip() if name and name.strip() else None def _driver_version(info: str | None) -> str | None: if not info: return None mesa = re.search(r"\bMesa\s+(\d+(?:\.\d+){1,3})", info, re.IGNORECASE) if mesa: return mesa.group(1) match = _VERSION_PATTERN.search(info) return match.group(0) if match else None def _vendor_from_model(model: str | None) -> str | None: lowered = (model or "").casefold() if "amd" in lowered or "radeon" in lowered: return "AMD" if "nvidia" in lowered or "geforce" in lowered: return "NVIDIA" if "intel" in lowered: return "Intel" return None def _summary_value(summary: str, key: str) -> str | None: match = re.search( rf"(?mi)^\s*{re.escape(key)}\s*=\s*([^\r\n]+?)\s*$", summary, ) return match.group(1).strip()[:255] if match else None def _vulkan_gpu_summaries(summary: str) -> tuple[str, ...]: headers = tuple(re.finditer(r"(?mi)^\s*GPU\d+\s*:\s*$", summary)) result: list[str] = [] for index, header in enumerate(headers): end = headers[index + 1].start() if index + 1 < len(headers) else len(summary) block = summary[header.start():end] device_type = _summary_value(block, "deviceType") if device_type and "cpu" in device_type.casefold(): continue if _summary_value(block, "deviceName"): result.append(block) return tuple(result) def _hex_summary_value(summary: str, key: str) -> int | None: value = _summary_value(summary, key) if not value: return None match = re.search(r"0x([0-9a-f]+)", value, re.IGNORECASE) if not match: return None try: return int(match.group(1), 16) except ValueError: return None def _hex_summary_values(summary: str, key: str) -> set[int]: values: set[int] = set() for match in re.finditer( rf"(?mi)^\s*{re.escape(key)}\s*=\s*0x([0-9a-f]+)\b", summary, ): try: values.add(int(match.group(1), 16)) except ValueError: continue return values def _gpu_offload_detected(environment: Mapping[str, str]) -> bool: dri_prime = environment.get("DRI_PRIME", "").casefold().strip() nvidia_offload = ( environment.get("__NV_PRIME_RENDER_OFFLOAD", "").casefold().strip() ) glx_vendor = environment.get("__GLX_VENDOR_LIBRARY_NAME", "").casefold().strip() nvidia_optimus = environment.get("__VK_LAYER_NV_optimus", "").strip() return ( dri_prime not in {"", "0", "false"} or nvidia_offload not in {"", "0", "false"} or glx_vendor == "nvidia" or bool(nvidia_optimus) ) def _maximum_version(values: Sequence[str]) -> str | None: parsed: list[tuple[tuple[int, ...], str]] = [] for value in values: match = _VERSION_PATTERN.search(value) if not match: continue version = match.group(0) try: parts = tuple(int(part) for part in version.split(".")) except ValueError: continue parsed.append((parts, version)) if not parsed: return None return max(parsed, key=lambda item: item[0])[1] def _run_probe( argv: Sequence[str], environment: Mapping[str, str], timeout_seconds: float, ) -> ProbeResult | None: if not argv or timeout_seconds <= 0: return None probe_environment = os.environ.copy() probe_environment.update(environment) probe_environment["LC_ALL"] = "C" probe_environment.pop("VK_LOADER_DEBUG", None) try: completed = subprocess.run( [str(value) for value in argv], env=probe_environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=timeout_seconds, check=False, ) except (OSError, subprocess.TimeoutExpired): return None return ProbeResult( returncode=completed.returncode, stdout=completed.stdout[-MAX_PROBE_OUTPUT_CHARACTERS:], stderr=completed.stderr[-MAX_PROBE_OUTPUT_CHARACTERS:], ) def _probe_text(result: ProbeResult | None) -> str: if result is None: return "" return (result.stdout or result.stderr).strip() def _first_probe_line(result: ProbeResult | None) -> str | None: text = _probe_text(result) if not text: return None return text.splitlines()[0].strip()[:128] or None def _read_json_object(path: Path) -> dict[str, object] | None: try: with path.open("rb") as handle: content = handle.read(MAX_MARKER_BYTES + 1) if len(content) > MAX_MARKER_BYTES: return None value = json.loads(content.decode("utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return None return value if isinstance(value, dict) else None def _object_field( value: Mapping[str, object] | None, key: str, ) -> dict[str, object] | None: if value is None: return None field = value.get(key) return field if isinstance(field, dict) else None def _short_string(value: object) -> str | None: return value[:255] if isinstance(value, str) and value else None def _optional_bool(value: object) -> bool | None: return value if isinstance(value, bool) else None def _read_short_text(path: Path) -> str | None: try: value = path.read_text(encoding="utf-8", errors="replace").strip() except OSError: return None return value[:255] or None def _read_hex(path: Path) -> int | None: value = _read_short_text(path) if value is None: return None try: return int(value, 16) except ValueError: return None def _read_decimal(path: Path) -> int | None: value = _read_short_text(path) if value is None: return None try: return int(value, 10) except ValueError: return None