218 lines
6.6 KiB
Python
218 lines
6.6 KiB
Python
"""Read-only state snapshots for the launcher's hidden debug view."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from .import_state import game_import_requires_verification
|
|
from .paths import AppPaths
|
|
from .runtime import (
|
|
RuntimeErrorPT,
|
|
RuntimeSelection,
|
|
resolve_proton_sync_mode,
|
|
select_runtime,
|
|
)
|
|
from .settings import Settings
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimeDebugInfo:
|
|
"""Configured, selected, and last-prepared runtime state."""
|
|
|
|
requested_kind: str
|
|
selected_kind: str | None
|
|
executable: str | None
|
|
proton_path: str | None
|
|
prefix: str | None
|
|
sync_configured: str
|
|
sync_effective: str | None
|
|
prepared_kind: str | None
|
|
prepared_proton_path: str | None
|
|
prepared_at: str | None
|
|
marker_matches_selection: bool | None
|
|
dxvk_d3d9_override: str | None
|
|
error: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VerificationDebugInfo:
|
|
"""Cached trust evidence from completed depot and import operations."""
|
|
|
|
import_state: str
|
|
import_verified_at: str | None
|
|
import_build: int | None
|
|
last_sync_at: str | None
|
|
last_full_verification_at: str | None
|
|
build: int | None
|
|
version: str | None
|
|
mode: str | None
|
|
files_total: int | None
|
|
files_downloaded: int | None
|
|
files_deleted: int | None
|
|
files_created: int | None
|
|
files_quarantined: int | None
|
|
|
|
|
|
RuntimeSelector = Callable[[Settings], RuntimeSelection]
|
|
SyncResolver = Callable[[str], str]
|
|
|
|
|
|
def inspect_runtime_debug(
|
|
paths: AppPaths,
|
|
settings: Settings,
|
|
*,
|
|
selector: RuntimeSelector | None = None,
|
|
sync_resolver: SyncResolver = resolve_proton_sync_mode,
|
|
) -> RuntimeDebugInfo:
|
|
"""Resolve the next runtime and compare it with the preparation marker."""
|
|
|
|
marker = _read_json_object(paths.game_runtime_marker)
|
|
prepared_runtime: dict[str, object] | None = None
|
|
if marker is not None:
|
|
raw_prepared_runtime = marker.get("runtime")
|
|
if isinstance(raw_prepared_runtime, dict):
|
|
prepared_runtime = raw_prepared_runtime
|
|
prepared_kind = _text(prepared_runtime.get("kind")) if prepared_runtime else None
|
|
prepared_proton_path = (
|
|
_text(prepared_runtime.get("proton_path")) if prepared_runtime else None
|
|
)
|
|
prepared_at = _text(marker.get("configured_at")) if marker else None
|
|
dxvk_override = _text(marker.get("dxvk_d3d9_override")) if marker else None
|
|
selection: RuntimeSelection | None = None
|
|
error: str | None = None
|
|
try:
|
|
selection = (
|
|
select_runtime(settings, paths=paths)
|
|
if selector is None
|
|
else selector(settings)
|
|
)
|
|
except (OSError, RuntimeErrorPT, ValueError) as exc:
|
|
error = str(exc).strip() or type(exc).__name__
|
|
selected_kind = selection.kind if selection is not None else None
|
|
prefix_kind = selected_kind or prepared_kind
|
|
prefix = None
|
|
if prefix_kind == "umu":
|
|
prefix = str(paths.umu_prefix)
|
|
elif prefix_kind in {"proton7", "proton7-container"}:
|
|
prefix = str(paths.proton_prefix)
|
|
sync_effective = None
|
|
if prefix_kind in {"proton7", "proton7-container"}:
|
|
try:
|
|
sync_effective = sync_resolver(settings.proton_sync)
|
|
except ValueError as exc:
|
|
error = error or (str(exc).strip() or type(exc).__name__)
|
|
marker_matches = (
|
|
prepared_runtime == asdict(selection)
|
|
if prepared_runtime is not None and selection is not None
|
|
else None
|
|
)
|
|
return RuntimeDebugInfo(
|
|
requested_kind=settings.runtime,
|
|
selected_kind=selected_kind,
|
|
executable=selection.executable if selection is not None else None,
|
|
proton_path=selection.proton_path if selection is not None else None,
|
|
prefix=prefix,
|
|
sync_configured=settings.proton_sync,
|
|
sync_effective=sync_effective,
|
|
prepared_kind=prepared_kind,
|
|
prepared_proton_path=prepared_proton_path,
|
|
prepared_at=prepared_at,
|
|
marker_matches_selection=marker_matches,
|
|
dxvk_d3d9_override=dxvk_override,
|
|
error=error,
|
|
)
|
|
|
|
|
|
def read_verification_debug(paths: AppPaths) -> VerificationDebugInfo:
|
|
"""Read cached verification state without hashing or changing game files."""
|
|
|
|
depot = _read_json_object(paths.native_depot_state)
|
|
if depot is not None and depot.get("dry_run") is not False:
|
|
depot = None
|
|
import_manifest_exists = paths.game_import_manifest.exists()
|
|
import_manifest = _read_json_object(paths.game_import_manifest)
|
|
if not import_manifest_exists:
|
|
import_state = "not applicable"
|
|
elif game_import_requires_verification(paths):
|
|
import_state = "pending"
|
|
else:
|
|
import_state = "verified"
|
|
import_verified_at = (
|
|
_text(import_manifest.get("verification_completed_at"))
|
|
if import_state == "verified" and import_manifest is not None
|
|
else None
|
|
)
|
|
import_build = (
|
|
_non_negative_int(import_manifest.get("verification_build"))
|
|
if import_state == "verified" and import_manifest is not None
|
|
else None
|
|
)
|
|
last_sync_at = _text(depot.get("completed_at")) if depot else None
|
|
full_verifications = [
|
|
value
|
|
for value in (
|
|
last_sync_at if depot is not None and depot.get("repair") is True else None,
|
|
import_verified_at,
|
|
)
|
|
if value is not None
|
|
]
|
|
last_full_verification_at = (
|
|
max(full_verifications, key=_iso_timestamp)
|
|
if full_verifications
|
|
else None
|
|
)
|
|
return VerificationDebugInfo(
|
|
import_state=import_state,
|
|
import_verified_at=import_verified_at,
|
|
import_build=import_build,
|
|
last_sync_at=last_sync_at,
|
|
last_full_verification_at=last_full_verification_at,
|
|
build=_non_negative_int(depot.get("build")) if depot else None,
|
|
version=_text(depot.get("version")) if depot else None,
|
|
mode=(
|
|
"full repair"
|
|
if depot is not None and depot.get("repair") is True
|
|
else "update"
|
|
if depot is not None and depot.get("repair") is False
|
|
else None
|
|
),
|
|
files_total=_non_negative_int(depot.get("files_total")) if depot else None,
|
|
files_downloaded=(
|
|
_non_negative_int(depot.get("files_downloaded")) if depot else None
|
|
),
|
|
files_deleted=_non_negative_int(depot.get("files_deleted")) if depot else None,
|
|
files_created=_non_negative_int(depot.get("files_created")) if depot else None,
|
|
files_quarantined=(
|
|
_non_negative_int(depot.get("files_quarantined")) if depot else None
|
|
),
|
|
)
|
|
|
|
|
|
def _read_json_object(path: Path) -> dict[str, object] | None:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
return None
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
def _text(value: object) -> str | None:
|
|
return value.strip() if isinstance(value, str) and value.strip() else None
|
|
|
|
|
|
def _non_negative_int(value: object) -> int | None:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
return None
|
|
return value
|
|
|
|
|
|
def _iso_timestamp(value: str) -> float:
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
|
|
except (OverflowError, ValueError):
|
|
return float("-inf")
|