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

from .command import CommandOutcome, CommandRunner
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
	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)
		return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
	if settings.runtime in {"auto", "proton7"}:
		proton_result = find_steam_proton7(env)
		if proton_result is not None:
			proton, steam_root = proton_result
			return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
		if settings.runtime == "proton7":
			raise RuntimeErrorPT("Steam Proton 7.0 was requested but not found")
	if settings.runtime in {"auto", "umu"}:
		umu = shutil.which("umu-run")
		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 == "proton7":
		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,
) -> 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)
	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)
	environment["PYTHONPATH"] = _runtime_python_path(
		os.environ.get("PYTHONPATH"),
	)
	return runner.run(
		game_command,
		cwd=paths.runtime_view,
		env_overrides=environment,
		timeout_seconds=12 * 60 * 60,
		stream_output=True,
	)


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 _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)