pristontale-eu-linux/ptlinux/runtime.py

"""Portable game runtime selection and Proton/UMU preparation."""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime
import json
import os
from pathlib import Path
import platform
import shutil
import sys
from typing import Mapping, Sequence

from .command import CommandOutcome, CommandRunner, OutputLineFilter
from .gamebox import build_gamebox_command
from .paths import AppPaths
from .settings import Settings


WINDOWS_GAME_EXE = r"C:\Program Files (x86)\Pristontale EU\Game.exe"
NATIVE_D3D43_DLLS = ("d3dx9_43", "d3dcompiler_43")
DXVK_D3D9_OVERRIDE = "native,builtin"
DLL_OVERRIDES_KEY = r"HKCU\Software\Wine\DllOverrides"
WINETRICKS_CODEPAGE_PROBE = (
	'winetricks_early_wine_arch cmd.exe /c "chcp 65001 > nul & echo %$1%"'
)
WINETRICKS_PROTON7_PROBE = 'winetricks_early_wine_arch cmd.exe /c "echo %$1%"'
RUNTIME_DEFAULT_SYNC = "runtime-default"
CACHYOS_MARKER = "cachyos"


class RuntimeErrorPT(RuntimeError):
	"""Raised when a supported game runtime cannot be prepared."""


@dataclass(frozen=True)
class RuntimeSelection:
	kind: str
	executable: str
	proton_path: str | None
	steam_root: str | None


def select_runtime(
	settings: Settings,
	environment: Mapping[str, str] | None = None,
) -> RuntimeSelection:
	env = os.environ if environment is None else environment
	umu = shutil.which("umu-run", path=env.get("PATH"))
	if settings.proton_path:
		proton = Path(settings.proton_path).expanduser().resolve()
		if not proton.is_file():
			raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
		steam_root = _steam_root_for_proton(proton)
		if settings.runtime == "proton7-container":
			if not umu:
				raise RuntimeErrorPT(
					"containerized Proton 7 was requested but umu-run was not found"
				)
			return RuntimeSelection(
				"proton7-container",
				umu,
				str(proton),
				str(steam_root),
			)
		if settings.runtime == "auto" and umu:
			return RuntimeSelection(
				"proton7-container",
				umu,
				str(proton),
				str(steam_root),
			)
		return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
	if settings.runtime in {"auto", "proton7", "proton7-container"}:
		proton_result = find_steam_proton7(env)
		if proton_result is not None:
			proton, steam_root = proton_result
			if settings.runtime == "proton7-container" or (
				settings.runtime == "auto" and umu
			):
				if not umu:
					raise RuntimeErrorPT(
						"containerized Proton 7 was requested but umu-run was not found"
					)
				return RuntimeSelection(
					"proton7-container",
					umu,
					str(proton),
					str(steam_root),
				)
			return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
		if settings.runtime in {"proton7", "proton7-container"}:
			raise RuntimeErrorPT("Steam Proton 7.0 was requested but not found")
	if settings.runtime in {"auto", "umu"}:
		if umu:
			return RuntimeSelection("umu", umu, None, None)
		raise RuntimeErrorPT(
			"no supported game runtime found; install umu-launcher or Steam Proton 7.0"
		)
	raise RuntimeErrorPT(f"unsupported runtime selection: {settings.runtime}")


def find_steam_proton7(
	environment: Mapping[str, str] | None = None,
) -> tuple[Path, Path] | None:
	env = os.environ if environment is None else environment
	home_value = env.get("HOME")
	if not home_value:
		return None
	home = Path(home_value).expanduser()
	roots: list[Path] = []
	explicit = env.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
	if explicit:
		roots.append(Path(explicit).expanduser())
	roots.extend(
		(
			home / ".local/share/Steam",
			home / ".steam/steam",
			home / "snap/steam/common/.local/share/Steam",
			home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
		)
	)
	seen: set[Path] = set()
	for root in roots:
		resolved = root.resolve()
		if resolved in seen:
			continue
		seen.add(resolved)
		proton = resolved / "steamapps/common/Proton 7.0/proton"
		if proton.is_file():
			return proton, resolved
	return None


def prepare_game_runtime(
	paths: AppPaths,
	selection: RuntimeSelection,
	runner: CommandRunner,
) -> list[CommandOutcome]:
	if not runner.dry_run:
		paths.create_base_directories()
	results: list[CommandOutcome] = []
	if selection.kind == "umu":
		environment = _umu_environment(paths)
		if not native_d3d_ready(paths.umu_prefix):
			results.append(
				runner.run(
					[selection.executable, "winetricks", "-q", "d3dx9_43", "d3dcompiler_43"],
					env_overrides=environment,
					timeout_seconds=1800.0,
				)
			)
	elif selection.kind in {"proton7", "proton7-container"}:
		if not selection.proton_path or not selection.steam_root:
			raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
		if not runner.dry_run:
			paths.compat_data.mkdir(parents=True, exist_ok=True)
		proton = Path(selection.proton_path)
		environment = _proton_environment(paths, selection)
		prefix_verb = "runinprefix" if _proton_prefix_ready(paths.proton_prefix) else "run"
		results.append(
			runner.run(
				[
					sys.executable,
					"-m",
					"ptlinux.proton_compat",
					str(proton),
					prefix_verb,
					"cmd",
					"/c",
					"exit",
				],
				env_overrides=environment,
				timeout_seconds=600.0,
			)
		)
		if not runner.dry_run:
			_ensure_proton_game_link(paths)
		results.append(
			runner.run(
				[
					sys.executable,
					"-m",
					"ptlinux.proton_compat",
					str(proton),
					"runinprefix",
					"reg.exe",
					"add",
					DLL_OVERRIDES_KEY,
					"/v",
					"d3d9",
					"/t",
					"REG_SZ",
					"/d",
					DXVK_D3D9_OVERRIDE,
					"/f",
				],
				env_overrides=environment,
				timeout_seconds=60.0,
			)
		)
		if not native_d3d_ready(paths.proton_prefix):
			wine_tools = _find_proton_wine_tools(proton)
			winetricks = shutil.which("winetricks")
			if wine_tools is None or not winetricks:
				raise RuntimeErrorPT("Proton Winetricks helpers are unavailable")
			wine, wineserver = wine_tools
			compatible_winetricks = prepare_proton7_winetricks(
				Path(winetricks),
				paths.cache / "compat/winetricks-proton7",
				dry_run=runner.dry_run,
			)
			results.append(
				runner.run(
					[
						str(compatible_winetricks),
						"-q",
						"d3dx9_43",
						"d3dcompiler_43",
					],
					env_overrides={
						"WINEPREFIX": str(paths.proton_prefix),
						"WINE": str(wine),
						"WINESERVER": str(wineserver),
						"WINEDEBUG": "-all",
					},
					timeout_seconds=1800.0,
				)
			)
	else:
		raise RuntimeErrorPT(f"unsupported runtime kind: {selection.kind}")
	if not runner.dry_run:
		_write_json(
			paths.state / "game-runtime.json",
			{
				"configured_at": datetime.now().astimezone().isoformat(),
				"native_d3d43": True,
				"dxvk_d3d9_override": DXVK_D3D9_OVERRIDE,
				"runtime": asdict(selection),
			},
		)
	return results


def run_game(
	paths: AppPaths,
	settings: Settings,
	selection: RuntimeSelection,
	runner: CommandRunner,
	*,
	use_gamebox: bool = True,
	timeout_seconds: float = 12 * 60 * 60,
	check: bool = True,
	extra_environment: Mapping[str, str] | None = None,
	output_line_filter: OutputLineFilter | None = None,
) -> CommandOutcome:
	if not paths.runtime_view.is_dir() and not runner.dry_run:
		raise RuntimeErrorPT("runtime view is missing; prepare the game first")
	if selection.kind == "umu":
		game_command = [
			selection.executable,
			str(paths.runtime_view / "Game.exe"),
			"PTEU_REBORN",
		]
		environment = _umu_environment(paths)
	elif selection.kind == "proton7-container":
		if not selection.proton_path:
			raise RuntimeErrorPT("Proton script path is missing")
		game_command = [
			selection.executable,
			str(paths.runtime_view / "Game.exe"),
			"PTEU_REBORN",
		]
		environment = _umu_proton7_environment(
			paths,
			selection,
			synchronization=resolve_proton_sync_mode(settings.proton_sync),
		)
	else:
		if not selection.proton_path:
			raise RuntimeErrorPT("Proton script path is missing")
		game_command = [
			sys.executable,
			"-m",
			"ptlinux.proton_compat",
			selection.proton_path,
			"runinprefix",
			WINDOWS_GAME_EXE,
			"PTEU_REBORN",
		]
		environment = _proton_environment(
			paths,
			selection,
			synchronization=resolve_proton_sync_mode(settings.proton_sync),
		)
	if use_gamebox:
		game_command = build_gamebox_command(game_command, settings)
	if extra_environment:
		environment.update(extra_environment)
	environment["PYTHONPATH"] = _runtime_python_path(
		os.environ.get("PYTHONPATH"),
	)
	return runner.run(
		game_command,
		cwd=paths.runtime_view,
		env_overrides=environment,
		timeout_seconds=timeout_seconds,
		check=check,
		stream_output=True,
		output_line_filter=output_line_filter,
	)


def run_proton_prefix_command(
	paths: AppPaths,
	selection: RuntimeSelection,
	runner: CommandRunner,
	arguments: Sequence[str],
	*,
	check: bool = True,
	timeout_seconds: float = 60.0,
) -> CommandOutcome:
	"""Run a maintenance command in the pinned Proton 7 prefix."""

	if selection.kind not in {"proton7", "proton7-container"}:
		raise RuntimeErrorPT("the login trace requires the pinned Proton 7 runtime")
	if not selection.proton_path or not selection.steam_root:
		raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
	return runner.run(
		[
			sys.executable,
			"-m",
			"ptlinux.proton_compat",
			selection.proton_path,
			"runinprefix",
			*arguments,
		],
		env_overrides=_proton_environment(paths, selection),
		timeout_seconds=timeout_seconds,
		check=check,
	)


def _runtime_python_path(existing: str | None) -> str:
	"""Make the installed package importable after changing into runtime_view."""
	package_root = str(Path(__file__).resolve().parent.parent)
	if not existing:
		return package_root
	return os.pathsep.join((package_root, existing))


def _umu_environment(paths: AppPaths) -> dict[str, str]:
	return {
		"WINEPREFIX": str(paths.umu_prefix),
		"GAMEID": "umu-pristontale-eu",
		"STORE": "none",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
		"WINEDEBUG": "-all",
	}


def _umu_proton7_environment(
	paths: AppPaths,
	selection: RuntimeSelection,
	*,
	synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]:
	"""Run the retained Proton 7 build inside its Steam Runtime container."""

	if not selection.proton_path:
		raise RuntimeErrorPT("Proton script path is missing")
	environment = {
		# UMU expects the Proton compatibility-data root and manages its pfx child.
		# Pointing it at proton_prefix would create a nested pfx -> . compatibility
		# symlink; compat_data preserves the exact prefix prepared by Proton 7.
		"WINEPREFIX": str(paths.compat_data),
		"GAMEID": "umu-pristontale-eu",
		"STORE": "none",
		"PROTONPATH": str(Path(selection.proton_path).parent),
		"PROTON_VERB": "waitforexitandrun",
		"PROTONFIXES_DISABLE": "1",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
		"WINEDEBUG": "-all",
	}
	environment.update(_proton_sync_environment(synchronization))
	return environment


def _proton_environment(
	paths: AppPaths,
	selection: RuntimeSelection,
	*,
	synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]:
	if not selection.steam_root:
		raise RuntimeErrorPT("Steam root is missing for Proton")
	environment = {
		"STEAM_COMPAT_CLIENT_INSTALL_PATH": selection.steam_root,
		"STEAM_COMPAT_DATA_PATH": str(paths.compat_data),
		"STEAM_COMPAT_INSTALL_PATH": str(paths.runtime_view),
		"SteamAppId": "0",
		"SteamGameId": "0",
		"WINEPREFIX": str(paths.proton_prefix),
		"WINEDEBUG": "-all",
		"PROTON_LOG": "0",
		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
	}
	environment.update(_proton_sync_environment(synchronization))
	return environment


def detect_distribution_id() -> str:
	"""Return the freedesktop distribution ID without invoking distro tools."""

	try:
		value = platform.freedesktop_os_release().get("ID", "")
	except OSError:
		return ""
	return value.strip().casefold()


def resolve_proton_sync_mode(
	configured: str,
	*,
	distribution_id: str | None = None,
	kernel_release: str | None = None,
) -> str:
	"""Resolve the configured mode, applying the narrow CachyOS compatibility fallback."""

	if configured not in {"auto", "fsync", "esync"}:
		raise ValueError(f"unsupported Proton synchronization mode: {configured}")
	if configured != "auto":
		return configured
	distribution = detect_distribution_id() if distribution_id is None else distribution_id
	kernel = platform.release() if kernel_release is None else kernel_release
	if CACHYOS_MARKER in distribution.casefold() or CACHYOS_MARKER in kernel.casefold():
		return "esync"
	return RUNTIME_DEFAULT_SYNC


def _proton_sync_environment(synchronization: str) -> dict[str, str]:
	if synchronization == RUNTIME_DEFAULT_SYNC:
		return {}
	if synchronization == "esync":
		return {
			"PROTON_NO_FSYNC": "1",
			"WINEFSYNC": "0",
			"PROTON_NO_ESYNC": "0",
			"WINEESYNC": "1",
		}
	if synchronization == "fsync":
		return {
			"PROTON_NO_FSYNC": "0",
			"WINEFSYNC": "1",
			"PROTON_NO_ESYNC": "0",
			"WINEESYNC": "1",
		}
	raise ValueError(f"unsupported resolved Proton synchronization mode: {synchronization}")


def _ensure_proton_game_link(paths: AppPaths) -> None:
	link = paths.proton_prefix / "drive_c/Program Files (x86)/Pristontale EU"
	link.parent.mkdir(parents=True, exist_ok=True)
	if link.is_symlink():
		if link.resolve() == paths.runtime_view.resolve():
			return
		raise RuntimeErrorPT(f"Proton game link points somewhere unexpected: {link}")
	if link.exists():
		if link.is_dir() and not any(link.iterdir()):
			link.rmdir()
		else:
			raise RuntimeErrorPT(f"refusing to replace existing Proton game path: {link}")
	link.symlink_to(paths.runtime_view, target_is_directory=True)


def _steam_root_for_proton(proton: Path) -> Path:
	parts = proton.parts
	try:
		index = parts.index("steamapps")
	except ValueError as exc:
		raise RuntimeErrorPT(
			"configured Proton path must be inside a Steam steamapps directory"
		) from exc
	return Path(*parts[:index])


def native_d3d_ready(prefix: Path) -> bool:
	directory = prefix / "drive_c/windows/syswow64"
	if not all((directory / f"{name}.dll").is_file() for name in NATIVE_D3D43_DLLS):
		return False
	try:
		registry = (prefix / "user.reg").read_text(
			encoding="utf-8",
			errors="replace",
		).casefold()
	except OSError:
		return False
	return all(_native_override_ready(registry, name) for name in NATIVE_D3D43_DLLS)


def _native_override_ready(registry: str, name: str) -> bool:
	return any(
		entry in registry
		for entry in (
			f'"{name}"="native,builtin"',
			f'"{name}"="native"',
			f'"*{name}"="native"',
		)
	)


def prepare_proton7_winetricks(
	source: Path,
	destination: Path,
	*,
	dry_run: bool = False,
) -> Path:
	try:
		content = source.read_text(encoding="utf-8")
	except OSError as exc:
		raise RuntimeErrorPT(f"could not read Winetricks script {source}: {exc}") from exc
	if WINETRICKS_CODEPAGE_PROBE not in content:
		return source
	if dry_run:
		return destination
	destination.parent.mkdir(parents=True, exist_ok=True)
	patched = content.replace(
		WINETRICKS_CODEPAGE_PROBE,
		WINETRICKS_PROTON7_PROBE,
		1,
	)
	temporary = destination.with_name(f".{destination.name}.{os.getpid()}.tmp")
	temporary.write_text(patched, encoding="utf-8")
	temporary.chmod(source.stat().st_mode & 0o777)
	temporary.replace(destination)
	return destination


def _proton_prefix_ready(prefix: Path) -> bool:
	return (prefix / "user.reg").is_file() and (
		prefix / "drive_c/windows/system32/cmd.exe"
	).exists()


def _find_proton_wine_tools(proton: Path) -> tuple[Path, Path] | None:
	for directory in (proton.parent / "files/bin", proton.parent / "dist/bin"):
		wine = directory / "wine"
		wineserver = directory / "wineserver"
		if wine.is_file() and wineserver.is_file():
			return wine, wineserver
	return None


def _write_json(path: Path, value: object) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
	temporary.replace(path)