pristontale-eu-linux/ptlinux/telemetry_environment.py

"""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