From cc5eb55dbe02d65910c5278a375d0d5640374208 Mon Sep 17 00:00:00 2001 From: xXJsonDeruloXx Date: Sat, 1 Aug 2026 15:14:09 -0400 Subject: fix package layout for Decky runtime modules --- backend/__init__.py | 1 - backend/assets.py | 184 ----------- backend/bundle.py | 522 ------------------------------ backend/config.py | 175 ---------- backend/files.py | 36 --- backend/patcher.py | 685 ---------------------------------------- backend/steam.py | 187 ----------- main.py | 10 +- py_modules/services/__init__.py | 1 + py_modules/services/assets.py | 184 +++++++++++ py_modules/services/bundle.py | 522 ++++++++++++++++++++++++++++++ py_modules/services/config.py | 175 ++++++++++ py_modules/services/files.py | 36 +++ py_modules/services/patcher.py | 685 ++++++++++++++++++++++++++++++++++++++++ py_modules/services/steam.py | 187 +++++++++++ tests/test_backend.py | 11 +- 16 files changed, 1802 insertions(+), 1799 deletions(-) delete mode 100644 backend/__init__.py delete mode 100644 backend/assets.py delete mode 100644 backend/bundle.py delete mode 100644 backend/config.py delete mode 100644 backend/files.py delete mode 100644 backend/patcher.py delete mode 100644 backend/steam.py create mode 100644 py_modules/services/__init__.py create mode 100644 py_modules/services/assets.py create mode 100644 py_modules/services/bundle.py create mode 100644 py_modules/services/config.py create mode 100644 py_modules/services/files.py create mode 100644 py_modules/services/patcher.py create mode 100644 py_modules/services/steam.py diff --git a/backend/__init__.py b/backend/__init__.py deleted file mode 100644 index 462b5df..0000000 --- a/backend/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Backend services for Decky Framegen.""" diff --git a/backend/assets.py b/backend/assets.py deleted file mode 100644 index 69f71ba..0000000 --- a/backend/assets.py +++ /dev/null @@ -1,184 +0,0 @@ -"""Static asset metadata and patch file inventories.""" - -OPTISCALER_ARCHIVE_ASSET = { - "name": "Optiscaler_0.9.4-final.20260718._MM.7z", - "sha256": "575cb4df866116093df75af607e37fd70e10f5163e0f23fd5c804142e80ef0ad", - "version": "0.9.4-final.20260718", -} - -FSR4_INT8_ASSET = { - "name": "amd_fidelityfx_upscaler_dx12.dll", - "sha256": "c7720bc16bede334f59a1a32cd22edbcbbb159685ed5240e61350a5fb0bc8a94", - "version": "4.0.2c", -} - -FSR4_OFFICIAL_411_ASSET = { - "name": "amdxcffx64.dll", - "sha256": "a2b136b6affd35a49b141a936be935f7d5ddc8d8f9b8c9afbe62ff9ddb2538a0", - "version": "4.1.1-official", -} - -FSR4_VALVE_411_ASSET = { - "name": "amdxcffx64_valve_2.3.0.2913.dll", - "sha256": "4e7dc37aebea3a90e3d3cc43e24cb2b54176b2535315f20dbe63b3b7cfc56b1e", - "version": "4.1.1-valve-2.3.0.2913", -} - -AMDXC64_RDNA2_ASSET = { - "name": "amdxc64.dll", - "sha256": "a0a0af61d475e30a70966b3459f3793df772faf8f26ae3261d10554ff592cbd5", - "version": "8.18.10.0474", -} - -OPTISCALER_PRE10_ASSET = { - "name": "OptiScaler_0.10.0-pre1.20260622_135940.dll", - "sha256": "b374b19081cc066365d0c6da4808d768e16469b0cbdfc478b6e95999947d5364", - "version": "0.10.0-pre1.20260622_135940", -} - -OPTIPATCHER_ASSET = { - "name": "OptiPatcher_rolling.asi", - "sha256": "88b9e1be3559737cd205fdf5f2c8550cf1923fb1def4c603e5bf03c3e84131b1", - "version": "rolling", -} - -FSR4_UPSCALER_FILENAME = "amd_fidelityfx_upscaler_dx12.dll" -FSR4_DRIVER_OVERRIDE_FILENAME = "amdxcffx64.dll" -INSTALL_MANIFEST_FILENAME = "install-manifest.json" -VERSION_FILENAME = "version.txt" -DEFAULT_FSR4_VARIANT = "rdna23-int8" -DEFAULT_FRAMEGEN_BACKEND = "auto" -FRAMEGEN_BACKENDS = { - "auto": "OptiScaler automatic selection", - "nukems": "Nukem's DLSSG → FSR3", -} - -FSR4_VARIANTS = { - "rdna23-int8": { - "label": "4.0.2c / RDNA2-3 compatibility", - "dir_name": "fsr4-rdna2-3", - "sha256": "c7720bc16bede334f59a1a32cd22edbcbbb159685ed5240e61350a5fb0bc8a94", - "source_asset_name": FSR4_INT8_ASSET["name"], - "source_version": FSR4_INT8_ASSET["version"], - "uses_archive_native": False, - "extra_files": [], - }, - "rdna4-native": { - "label": "4.1.1 SDK / RDNA3 dGPU + RDNA4", - "dir_name": "fsr4-rdna4", - "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", - "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], - "source_version": OPTISCALER_ARCHIVE_ASSET["version"], - "uses_archive_native": True, - "extra_files": [], - }, - "rdna34-official-411": { - "label": "4.1.1 driver override / RDNA3-4", - "dir_name": "fsr4-rdna3-4-official-411", - "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", - "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], - "source_version": OPTISCALER_ARCHIVE_ASSET["version"], - "uses_archive_native": True, - "extra_files": [ - { - "name": FSR4_DRIVER_OVERRIDE_FILENAME, - "sha256": FSR4_OFFICIAL_411_ASSET["sha256"], - "source_asset_name": FSR4_OFFICIAL_411_ASSET["name"], - "source_version": FSR4_OFFICIAL_411_ASSET["version"], - } - ], - "config_overrides": {}, - }, - "rdna2-valve-411-pre10": { - "label": "4.1.1 Valve RDNA2 compatibility", - "dir_name": "fsr4-rdna2-valve-411-pre10", - "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", - "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], - "source_version": OPTISCALER_ARCHIVE_ASSET["version"], - "uses_archive_native": True, - "injector": { - "name": "OptiScaler.dll", - "sha256": OPTISCALER_PRE10_ASSET["sha256"], - "source_asset_name": OPTISCALER_PRE10_ASSET["name"], - "source_version": OPTISCALER_PRE10_ASSET["version"], - }, - "extra_files": [ - { - "name": FSR4_DRIVER_OVERRIDE_FILENAME, - "sha256": FSR4_VALVE_411_ASSET["sha256"], - "source_asset_name": FSR4_VALVE_411_ASSET["name"], - "source_version": FSR4_VALVE_411_ASSET["version"], - }, - { - "name": "amdxc64.dll", - "sha256": AMDXC64_RDNA2_ASSET["sha256"], - "source_asset_name": AMDXC64_RDNA2_ASSET["name"], - "source_version": AMDXC64_RDNA2_ASSET["version"], - }, - ], - "config_overrides": { - "FSR.Fsr4ForceModel": "2", - "Plugins.LoadCustomAmdxc64OnRdna2": "true", - }, - }, -} - -VARIANT_EXTRA_FILENAMES = sorted( - { - extra_file["name"] - for variant in FSR4_VARIANTS.values() - for extra_file in variant.get("extra_files", []) - } -) - -PROXY_DLL_BACKUPS = [ - "dxgi.dll", - "winmm.dll", - "dbghelp.dll", - "version.dll", - "wininet.dll", - "winhttp.dll", - "OptiScaler.asi", -] -VALID_DLL_NAMES = set(PROXY_DLL_BACKUPS) - -INJECTOR_FILENAMES = [ - *PROXY_DLL_BACKUPS, - "nvngx.dll", - "_nvngx.dll", - "nvngx-wrapper.dll", - "dlss-enabler.dll", - "OptiScaler.dll", -] - -ORIGINAL_DLL_BACKUPS = [ - "d3dcompiler_47.dll", - "amd_fidelityfx_dx12.dll", - "amd_fidelityfx_framegeneration_dx12.dll", - FSR4_UPSCALER_FILENAME, - FSR4_DRIVER_OVERRIDE_FILENAME, - "amdxc64.dll", - "amd_fidelityfx_vk.dll", -] - -RESTORABLE_BACKUP_FILES = [ - *PROXY_DLL_BACKUPS, - *ORIGINAL_DLL_BACKUPS, -] - -SUPPORT_FILES = [ - "libxess.dll", - "libxess_dx11.dll", - "libxess_fg.dll", - "libxell.dll", - "amd_fidelityfx_dx12.dll", - "amd_fidelityfx_framegeneration_dx12.dll", - "amd_fidelityfx_vk.dll", - "dlssg_to_fsr3_amd_is_better.dll", - "fakenvapi.dll", - "fakenvapi.ini", -] - -MARKER_FILENAME = "FRAMEGEN_PATCH" -MANIFEST_SCHEMA_VERSION = 2 -BACKUP_DIRECTORY_NAME = ".framegen-backups" diff --git a/backend/bundle.py b/backend/bundle.py deleted file mode 100644 index a6ef9e4..0000000 --- a/backend/bundle.py +++ /dev/null @@ -1,522 +0,0 @@ -"""Management of the shared OptiScaler bundle in ``~/fgmod``.""" - -import os -import shutil -import subprocess -from datetime import datetime, timezone -from pathlib import Path -from typing import Callable - -from . import config -from .assets import ( - DEFAULT_FRAMEGEN_BACKEND, - DEFAULT_FSR4_VARIANT, - FRAMEGEN_BACKENDS, - FSR4_DRIVER_OVERRIDE_FILENAME, - FSR4_VARIANTS, - FSR4_UPSCALER_FILENAME, - INSTALL_MANIFEST_FILENAME, - MANIFEST_SCHEMA_VERSION, - OPTIPATCHER_ASSET, - OPTISCALER_ARCHIVE_ASSET, - OPTISCALER_PRE10_ASSET, - AMDXC64_RDNA2_ASSET, - FSR4_INT8_ASSET, - FSR4_OFFICIAL_411_ASSET, - FSR4_VALVE_411_ASSET, - VERSION_FILENAME, -) -from .files import file_sha256, read_json, write_json - - -Logger = Callable[[str], None] - - -class BundleManager: - def __init__(self, home: Path, plugin_dir: Path, logger: Logger): - self.home = home - self.plugin_dir = plugin_dir - self.logger = logger - - @property - def fgmod_path(self) -> Path: - return self.home / "fgmod" - - def _log(self, message: str) -> None: - self.logger(f"[Framegen] {message}") - - def _create_renamed_copies(self, source_file: Path, renames_dir: Path) -> bool: - try: - if not source_file.exists(): - self._log(f"Source file {source_file} does not exist") - return False - renames_dir.mkdir(parents=True, exist_ok=True) - for filename in ("dxgi.dll", "winmm.dll", "dbghelp.dll", "version.dll", "wininet.dll", "winhttp.dll", "OptiScaler.asi"): - shutil.copy2(source_file, renames_dir / filename) - return True - except OSError as exc: - self._log(f"Failed to create renamed copies: {exc}") - return False - - def _copy_launcher_scripts(self, assets_dir: Path, extract_path: Path) -> bool: - try: - scripts = { - "fgmod.sh": "fgmod", - "fgmod-uninstaller.sh": "fgmod-uninstaller.sh", - "update-optiscaler-config.py": "update-optiscaler-config.py", - } - for source_name, destination_name in scripts.items(): - source = assets_dir / source_name - if not source.exists(): - continue - destination = extract_path / destination_name - shutil.copy2(source, destination) - destination.chmod(0o755) - return True - except OSError as exc: - self._log(f"Failed to copy launcher scripts: {exc}") - return False - - def _extract_archive(self, archive_path: Path, output_dir: Path) -> None: - output_dir.mkdir(parents=True, exist_ok=True) - environment = os.environ.copy() - environment["LD_LIBRARY_PATH"] = "" - result = subprocess.run( - ["7z", "x", "-y", "-o" + str(output_dir), str(archive_path)], - capture_output=True, - text=True, - check=False, - env=environment, - ) - if result.returncode != 0: - raise RuntimeError(result.stderr or result.stdout or f"Failed to extract {archive_path.name}") - - def _verify(self, path: Path, expected_sha256: str, description: str) -> str: - actual = file_sha256(path) - if actual.lower() != expected_sha256.lower(): - raise RuntimeError(f"{description} hash mismatch: expected {expected_sha256}, got {actual}") - return actual - - def _manifest_path(self) -> Path: - return self.fgmod_path / INSTALL_MANIFEST_FILENAME - - def load_manifest(self) -> dict: - return read_json(self._manifest_path()) - - @staticmethod - def normalize_variant(value: str | None) -> str: - return value if value in FSR4_VARIANTS else DEFAULT_FSR4_VARIANT - - @staticmethod - def normalize_backend(value: str | None) -> str: - return value if value in FRAMEGEN_BACKENDS else DEFAULT_FRAMEGEN_BACKEND - - def selected_variant(self, requested: str | None = None) -> str: - if requested in FSR4_VARIANTS: - return requested - installed = self.load_manifest().get("selected_default_variant") - return self.normalize_variant(installed) - - def selected_backend(self, requested: str | None = None) -> str: - if requested in FRAMEGEN_BACKENDS: - return requested - manifest_backend = self.load_manifest().get("framegen_backend") - if manifest_backend in FRAMEGEN_BACKENDS: - return manifest_backend - return self.normalize_backend(config.detect_backend(self.fgmod_path / "OptiScaler.ini")) - - def variant_info(self, variant: str | None) -> dict: - return FSR4_VARIANTS[self.normalize_variant(variant)] - - def variant_dir(self, variant: str | None) -> Path: - return self.fgmod_path / self.variant_info(variant)["dir_name"] - - def variant_path(self, variant: str | None) -> Path: - return self.variant_dir(variant) / FSR4_UPSCALER_FILENAME - - def variant_extra_files(self, variant: str | None) -> list[dict]: - return list(self.variant_info(variant).get("extra_files") or []) - - def variant_extra_path(self, variant: str | None, filename: str) -> Path: - return self.variant_dir(variant) / filename - - def variant_injector_info(self, variant: str | None) -> dict | None: - injector = self.variant_info(variant).get("injector") - return injector if isinstance(injector, dict) else None - - def variant_injector_path(self, variant: str | None) -> Path | None: - injector = self.variant_injector_info(variant) - return self.variant_dir(variant) / injector["name"] if injector else None - - def variant_proxy_path(self, variant: str | None, dll_name: str) -> Path | None: - if not self.variant_injector_info(variant): - return None - return self.variant_dir(variant) / "renames" / dll_name - - def variant_config_overrides(self, variant: str | None) -> dict: - overrides = self.variant_info(variant).get("config_overrides") or {} - return dict(overrides) if isinstance(overrides, dict) else {} - - def fgmod_version(self) -> str | None: - manifest = self.load_manifest() - optiscaler = manifest.get("optiscaler") - if isinstance(optiscaler, dict) and optiscaler.get("version"): - return str(optiscaler["version"]) - try: - version = (self.fgmod_path / VERSION_FILENAME).read_text(encoding="utf-8").strip() - return version or None - except OSError: - return None - - def managed_support_candidates(self, filename: str) -> list[Path]: - candidates = [self.fgmod_path / filename] - if filename == FSR4_UPSCALER_FILENAME: - candidates = [self.fgmod_path / filename] + [self.variant_path(variant) for variant in FSR4_VARIANTS] - else: - for variant in FSR4_VARIANTS: - if any(extra.get("name") == filename for extra in self.variant_extra_files(variant)): - candidates.append(self.variant_extra_path(variant, filename)) - unique: list[Path] = [] - seen: set[str] = set() - for candidate in candidates: - if str(candidate) not in seen: - unique.append(candidate) - seen.add(str(candidate)) - return unique - - def is_managed_support_file(self, path: Path) -> bool: - return path.exists() and any( - candidate.is_file() and file_sha256(path) == file_sha256(candidate) - for candidate in self.managed_support_candidates(path.name) - ) - - def _sync_variant_root_extra_files(self, variant: str) -> None: - selected = {extra["name"]: extra for extra in self.variant_extra_files(variant)} - all_extra_names = { - extra["name"] - for item in FSR4_VARIANTS.values() - for extra in item.get("extra_files", []) - } - for filename in all_extra_names: - root_path = self.fgmod_path / filename - if filename not in selected: - if root_path.exists(): - root_path.unlink() - continue - source = self.variant_extra_path(variant, filename) - if not source.exists(): - raise FileNotFoundError(f"Prepared FSR4 variant extra file missing: {source}") - shutil.copy2(source, root_path) - - def activate_variant(self, variant: str | None) -> str: - variant_id = self.normalize_variant(variant) - source = self.variant_path(variant_id) - if not source.exists(): - raise FileNotFoundError(f"Prepared FSR4 variant missing: {source}") - shutil.copy2(source, self.fgmod_path / FSR4_UPSCALER_FILENAME) - self._sync_variant_root_extra_files(variant_id) - return variant_id - - def detect_variant(self, directory: Path, upscaler_sha256: str | None) -> str | None: - if not upscaler_sha256: - return None - normalized = upscaler_sha256.lower() - for variant_id, variant in FSR4_VARIANTS.items(): - if str(variant.get("sha256", "")).lower() != normalized: - continue - if all( - (directory / extra["name"]).exists() - and file_sha256(directory / extra["name"]).lower() == extra["sha256"].lower() - for extra in variant.get("extra_files", []) - ): - return variant_id - return None - - def extract_static_optiscaler( - self, - selected_default_variant: str = DEFAULT_FSR4_VARIANT, - framegen_backend: str = DEFAULT_FRAMEGEN_BACKEND, - ) -> dict: - """Prepare the shared bundle; replacement remains intentionally non-atomic for compatibility.""" - try: - selected_default_variant = self.normalize_variant(selected_default_variant) - framegen_backend = self.normalize_backend(framegen_backend) - bin_path = self.plugin_dir / "bin" - assets_dir = self.plugin_dir / "assets" - if not bin_path.exists(): - return {"status": "error", "message": f"Bin directory not found: {bin_path}"} - - asset_sources = { - "archive": (bin_path / OPTISCALER_ARCHIVE_ASSET["name"], OPTISCALER_ARCHIVE_ASSET), - "int8": (bin_path / FSR4_INT8_ASSET["name"], FSR4_INT8_ASSET), - "official": (bin_path / FSR4_OFFICIAL_411_ASSET["name"], FSR4_OFFICIAL_411_ASSET), - "valve": (bin_path / FSR4_VALVE_411_ASSET["name"], FSR4_VALVE_411_ASSET), - "amdxc64": (bin_path / AMDXC64_RDNA2_ASSET["name"], AMDXC64_RDNA2_ASSET), - "pre10": (bin_path / OPTISCALER_PRE10_ASSET["name"], OPTISCALER_PRE10_ASSET), - "optipatcher": (bin_path / OPTIPATCHER_ASSET["name"], OPTIPATCHER_ASSET), - } - for path, asset in asset_sources.values(): - if not path.exists(): - return {"status": "error", "message": f"Required bundled asset missing: {asset['name']}"} - self._verify(path, asset["sha256"], asset["name"]) - - if self.fgmod_path.exists(): - shutil.rmtree(self.fgmod_path) - self.fgmod_path.mkdir(parents=True, exist_ok=True) - self._extract_archive(asset_sources["archive"][0], self.fgmod_path) - - source = self.fgmod_path / "OptiScaler.dll" - if not self._create_renamed_copies(source, self.fgmod_path / "renames"): - return {"status": "error", "message": "Failed to prepare renamed OptiScaler proxies."} - if not self._copy_launcher_scripts(assets_dir, self.fgmod_path): - return {"status": "error", "message": "Failed to copy launcher scripts."} - - plugins_dir = self.fgmod_path / "plugins" - plugins_dir.mkdir(parents=True, exist_ok=True) - optipatcher_dst = plugins_dir / "OptiPatcher.asi" - shutil.copy2(asset_sources["optipatcher"][0], optipatcher_dst) - optipatcher_sha256 = self._verify(optipatcher_dst, OPTIPATCHER_ASSET["sha256"], "Prepared OptiPatcher plugin") - - ini_file = self.fgmod_path / "OptiScaler.ini" - config.configure(ini_file, framegen_backend, logger=self.logger) - - native_root = self.fgmod_path / FSR4_UPSCALER_FILENAME - native_sha256 = self._verify(native_root, FSR4_VARIANTS["rdna4-native"]["sha256"], "Archive-native FSR4 upscaler") - for variant_id in ("rdna4-native", "rdna34-official-411", "rdna2-valve-411-pre10"): - variant_dir = self.variant_dir(variant_id) - variant_dir.mkdir(parents=True, exist_ok=True) - prepared = variant_dir / FSR4_UPSCALER_FILENAME - shutil.copy2(native_root, prepared) - self._verify(prepared, FSR4_VARIANTS[variant_id]["sha256"], f"Prepared {variant_id} FSR4 upscaler") - - official_dir = self.variant_dir("rdna34-official-411") - shutil.copy2(asset_sources["official"][0], official_dir / FSR4_DRIVER_OVERRIDE_FILENAME) - self._verify(official_dir / FSR4_DRIVER_OVERRIDE_FILENAME, FSR4_OFFICIAL_411_ASSET["sha256"], "Prepared official driver override") - - valve_dir = self.variant_dir("rdna2-valve-411-pre10") - shutil.copy2(asset_sources["valve"][0], valve_dir / FSR4_DRIVER_OVERRIDE_FILENAME) - shutil.copy2(asset_sources["amdxc64"][0], valve_dir / "amdxc64.dll") - shutil.copy2(asset_sources["pre10"][0], valve_dir / "OptiScaler.dll") - self._verify(valve_dir / FSR4_DRIVER_OVERRIDE_FILENAME, FSR4_VALVE_411_ASSET["sha256"], "Prepared Valve driver override") - self._verify(valve_dir / "amdxc64.dll", AMDXC64_RDNA2_ASSET["sha256"], "Prepared RDNA2 driver override") - self._verify(valve_dir / "OptiScaler.dll", OPTISCALER_PRE10_ASSET["sha256"], "Prepared pre10 OptiScaler injector") - if not self._create_renamed_copies(valve_dir / "OptiScaler.dll", valve_dir / "renames"): - return {"status": "error", "message": "Failed to prepare renamed pre10 OptiScaler proxies."} - - int8_dir = self.variant_dir("rdna23-int8") - int8_dir.mkdir(parents=True, exist_ok=True) - shutil.copy2(asset_sources["int8"][0], int8_dir / FSR4_UPSCALER_FILENAME) - self._verify(int8_dir / FSR4_UPSCALER_FILENAME, FSR4_INT8_ASSET["sha256"], "Prepared RDNA2-3 FSR4 upscaler") - - selected_default_variant = self.activate_variant(selected_default_variant) - active_sha256 = file_sha256(self.fgmod_path / FSR4_UPSCALER_FILENAME) - (self.fgmod_path / VERSION_FILENAME).write_text(OPTISCALER_ARCHIVE_ASSET["version"], encoding="utf-8") - manifest = self._build_manifest( - selected_default_variant, - framegen_backend, - native_sha256, - optipatcher_sha256, - active_sha256, - ) - write_json(self._manifest_path(), manifest) - return { - "status": "success", - "message": f"Successfully extracted OptiScaler {OPTISCALER_ARCHIVE_ASSET['version']} to ~/fgmod", - "version": OPTISCALER_ARCHIVE_ASSET["version"], - "selected_default_variant": selected_default_variant, - "selected_default_variant_label": FSR4_VARIANTS[selected_default_variant]["label"], - "framegen_backend": framegen_backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], - } - except Exception as exc: - self._log(f"Extract failed: {exc}") - return {"status": "error", "message": f"Extract failed: {exc}"} - - def _build_manifest( - self, - selected_variant: str, - framegen_backend: str, - native_sha256: str, - optipatcher_sha256: str, - active_sha256: str, - ) -> dict: - variants = {} - for variant_id, variant in FSR4_VARIANTS.items(): - injector = variant.get("injector") - injector_manifest = None - if isinstance(injector, dict): - injector_manifest = { - key: injector[key] - for key in ("name", "sha256", "source_asset_name", "source_version") - } - injector_manifest["path"] = str(Path(variant["dir_name"]) / injector["name"]) - variants[variant_id] = { - "label": variant["label"], - "dir_name": variant["dir_name"], - "path": str(Path(variant["dir_name"]) / FSR4_UPSCALER_FILENAME), - "sha256": variant["sha256"], - "source_asset_name": variant["source_asset_name"], - "source_version": variant["source_version"], - "uses_archive_native": bool(variant["uses_archive_native"]), - "injector": injector_manifest, - "config_overrides": dict(variant.get("config_overrides") or {}), - "extra_files": [ - { - **{key: extra[key] for key in ("name", "sha256", "source_asset_name", "source_version")}, - "path": str(Path(variant["dir_name"]) / extra["name"]), - } - for extra in variant.get("extra_files", []) - ], - } - return { - "schema_version": MANIFEST_SCHEMA_VERSION, - "installed_at": datetime.now(timezone.utc).isoformat(), - "optiscaler": { - "asset_name": OPTISCALER_ARCHIVE_ASSET["name"], - "version": OPTISCALER_ARCHIVE_ASSET["version"], - "sha256": OPTISCALER_ARCHIVE_ASSET["sha256"], - "native_upscaler_sha256": native_sha256, - }, - "optipatcher": { - "asset_name": OPTIPATCHER_ASSET["name"], - "version": OPTIPATCHER_ASSET["version"], - "sha256": optipatcher_sha256, - "target_path": "plugins/OptiPatcher.asi", - }, - "fsr4_variants": variants, - "selected_default_variant": selected_variant, - "framegen_backend": framegen_backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], - "active_root_upscaler": { - "path": FSR4_UPSCALER_FILENAME, - "sha256": active_sha256, - "variant": selected_variant, - }, - } - - def run_install(self, selected_variant: str, framegen_backend: str) -> dict: - selected_variant = self.normalize_variant(selected_variant) - framegen_backend = self.normalize_backend(framegen_backend) - result = self.extract_static_optiscaler(selected_variant, framegen_backend) - if result["status"] != "success": - return {"status": "error", "message": f"OptiScaler extraction failed: {result.get('message', 'Unknown error')}"} - return { - "status": "success", - "output": f"Successfully installed OptiScaler {result.get('version', OPTISCALER_ARCHIVE_ASSET['version'])} with {result.get('selected_default_variant_label', FSR4_VARIANTS[selected_variant]['label'])}.", - "version": result.get("version", OPTISCALER_ARCHIVE_ASSET["version"]), - "selected_default_variant": result.get("selected_default_variant", selected_variant), - "selected_default_variant_label": result.get("selected_default_variant_label", FSR4_VARIANTS[selected_variant]["label"]), - "framegen_backend": result.get("framegen_backend", framegen_backend), - "framegen_backend_label": result.get("framegen_backend_label", FRAMEGEN_BACKENDS[framegen_backend]), - } - - def uninstall(self) -> dict: - try: - if not self.fgmod_path.exists(): - return {"status": "success", "output": "No fgmod directory found to remove"} - shutil.rmtree(self.fgmod_path) - return {"status": "success", "output": "Successfully removed fgmod directory"} - except OSError as exc: - self._log(f"Uninstall error: {exc}") - return {"status": "error", "message": f"Uninstall failed: {exc}", "output": str(exc)} - - def set_default_variant(self, selected_variant: str) -> dict: - try: - if not self.fgmod_path.exists(): - return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} - manifest = self.load_manifest() - if not manifest: - return {"status": "error", "message": "Install manifest missing. Reinstall OptiScaler."} - selected_variant = self.activate_variant(selected_variant) - active_sha256 = file_sha256(self.fgmod_path / FSR4_UPSCALER_FILENAME) - manifest["selected_default_variant"] = selected_variant - manifest["active_root_upscaler"] = { - "path": FSR4_UPSCALER_FILENAME, - "sha256": active_sha256, - "variant": selected_variant, - } - manifest["updated_at"] = datetime.now(timezone.utc).isoformat() - write_json(self._manifest_path(), manifest) - return { - "status": "success", - "output": f"Default FSR4 runtime switched to {FSR4_VARIANTS[selected_variant]['label']}.", - "version": self.fgmod_version(), - "selected_default_variant": selected_variant, - "selected_default_variant_label": FSR4_VARIANTS[selected_variant]["label"], - } - except Exception as exc: - self._log(f"Failed to switch default FSR4 runtime: {exc}") - return {"status": "error", "message": f"Failed to switch default FSR4 runtime: {exc}"} - - def set_framegen_backend(self, framegen_backend: str) -> dict: - try: - if not self.fgmod_path.exists(): - return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} - backend = self.normalize_backend(framegen_backend) - manifest = self.load_manifest() - if not manifest: - return {"status": "error", "message": "Install manifest missing. Reinstall OptiScaler."} - ini_file = self.fgmod_path / "OptiScaler.ini" - if not config.configure(ini_file, backend, logger=self.logger): - return {"status": "error", "message": "Could not update the shared OptiScaler.ini."} - manifest["framegen_backend"] = backend - manifest["framegen_backend_label"] = FRAMEGEN_BACKENDS[backend] - manifest["updated_at"] = datetime.now(timezone.utc).isoformat() - write_json(self._manifest_path(), manifest) - return { - "status": "success", - "framegen_backend": backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[backend], - "output": f"Frame generation backend set to {FRAMEGEN_BACKENDS[backend]}.", - } - except Exception as exc: - self._log(f"Failed to set frame generation backend: {exc}") - return {"status": "error", "message": f"Failed to set frame generation backend: {exc}"} - - def check(self) -> dict: - required_files = [ - "OptiScaler.dll", - "OptiScaler.ini", - "dlssg_to_fsr3_amd_is_better.dll", - "fakenvapi.dll", - "fakenvapi.ini", - "amd_fidelityfx_dx12.dll", - "amd_fidelityfx_framegeneration_dx12.dll", - FSR4_UPSCALER_FILENAME, - "amd_fidelityfx_vk.dll", - "libxess.dll", - "libxess_dx11.dll", - "libxess_fg.dll", - "libxell.dll", - "fgmod", - "fgmod-uninstaller.sh", - "update-optiscaler-config.py", - INSTALL_MANIFEST_FILENAME, - ] - if not self.fgmod_path.exists() or any(not (self.fgmod_path / filename).exists() for filename in required_files): - return {"exists": False} - plugins_dir = self.fgmod_path / "plugins" - if not (plugins_dir / "OptiPatcher.asi").exists(): - return {"exists": False} - for variant_id, variant in FSR4_VARIANTS.items(): - variant_dir = self.variant_dir(variant_id) - if not (variant_dir / FSR4_UPSCALER_FILENAME).exists(): - return {"exists": False} - injector = variant.get("injector") - if isinstance(injector, dict) and not (variant_dir / injector.get("name", "OptiScaler.dll")).exists(): - return {"exists": False} - for extra in variant.get("extra_files", []): - if not (variant_dir / extra["name"]).exists(): - return {"exists": False} - manifest = self.load_manifest() - variant = self.selected_variant() - backend = self.selected_backend() - return { - "exists": True, - "version": self.fgmod_version(), - "selected_fsr4_variant": variant, - "selected_fsr4_variant_label": FSR4_VARIANTS[variant]["label"], - "framegen_backend": backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[backend], - "install_manifest_present": bool(manifest), - } diff --git a/backend/config.py b/backend/config.py deleted file mode 100644 index 87476cc..0000000 --- a/backend/config.py +++ /dev/null @@ -1,175 +0,0 @@ -"""OptiScaler configuration migration and default handling.""" - -import re -from pathlib import Path -from typing import Callable - - -Logger = Callable[[str], None] - - -def _split_override_key(raw_key: str) -> tuple[str | None, str]: - key = str(raw_key).strip() - if "." in key: - section, section_key = key.split(".", 1) - section = section.strip() - section_key = section_key.strip() - if section and section_key: - return section, section_key - return None, key - - -def apply_overrides(ini_file: Path, overrides: dict, logger: Logger | None = None) -> bool: - """Upsert INI keys without disturbing unrelated game configuration.""" - if not overrides: - return True - try: - content = ini_file.read_text(encoding="utf-8", errors="replace") - newline = "\r\n" if "\r\n" in content else "\n" - lines = content.splitlines(keepends=True) - section_pattern = re.compile(r"^\s*\[(?P
[^\]]+)\]\s*$") - - def ensure_trailing_newline() -> None: - if lines and not lines[-1].endswith(("\n", "\r")): - lines[-1] += newline - - def upsert(section: str | None, key: str, value: str) -> None: - replacement = f"{key}={value}" - key_pattern = re.compile(rf"^(\s*{re.escape(key)}\s*)=.*$") - - if section is None: - for index, line in enumerate(lines): - if key_pattern.match(line): - ending = "\r\n" if line.endswith("\r\n") else ("\n" if line.endswith("\n") else newline) - lines[index] = f"{replacement}{ending}" - return - ensure_trailing_newline() - lines.append(f"{replacement}{newline}") - return - - in_section = False - insert_at = None - for index, line in enumerate(lines): - match = section_pattern.match(line.strip()) - if match: - if in_section: - insert_at = index - break - if match.group("section") == section: - in_section = True - continue - if in_section and key_pattern.match(line): - ending = "\r\n" if line.endswith("\r\n") else ("\n" if line.endswith("\n") else newline) - lines[index] = f"{replacement}{ending}" - return - - if in_section: - if insert_at is None: - ensure_trailing_newline() - insert_at = len(lines) - lines.insert(insert_at, f"{replacement}{newline}") - return - - ensure_trailing_newline() - if lines and lines[-1].strip(): - lines.append(newline) - lines.extend((f"[{section}]{newline}", f"{replacement}{newline}")) - - for raw_key, raw_value in overrides.items(): - section, key = _split_override_key(str(raw_key)) - if key: - upsert(section, key, str(raw_value).strip()) - ini_file.write_text("".join(lines), encoding="utf-8") - return True - except Exception as exc: - if logger: - logger(f"Failed to apply OptiScaler.ini overrides to {ini_file}: {exc}") - return False - - -def migrate(ini_file: Path, logger: Logger | None = None) -> bool: - """Migrate the pre-0.9 ``FGType`` setting to split input/output keys.""" - try: - if not ini_file.exists(): - return False - content = ini_file.read_text(encoding="utf-8", errors="replace") - match = re.search(r"^FGType\s*=\s*(\S+)", content, re.MULTILINE) - if not match: - return True - - value = match.group(1) - if re.search(r"^FGInput\s*=", content, re.MULTILINE): - updated = re.sub(r"^FGType\s*=\s*\S+\n?", "", content, flags=re.MULTILINE) - message = f"Removed stale FGType from {ini_file}" - else: - updated = re.sub( - r"^FGType\s*=\s*\S+", - f"FGInput={value}\nFGOutput={value}", - content, - flags=re.MULTILINE, - ) - message = f"Migrated FGType={value} in {ini_file}" - if updated != content: - ini_file.write_text(updated, encoding="utf-8") - if logger: - logger(message) - return True - except Exception as exc: - if logger: - logger(f"Failed to migrate OptiScaler.ini: {exc}") - return False - - -def disable_hq_font_auto(ini_file: Path, logger: Logger | None = None) -> bool: - """Avoid the HQ font auto mode, which is unreliable under Proton.""" - try: - if not ini_file.exists(): - return False - content = ini_file.read_text(encoding="utf-8", errors="replace") - updated = re.sub(r"UseHQFont\s*=\s*auto", "UseHQFont=false", content) - if updated != content: - ini_file.write_text(updated, encoding="utf-8") - return True - except Exception as exc: - if logger: - logger(f"Failed to update HQ font setting in {ini_file}: {exc}") - return False - - -def configure( - ini_file: Path, - framegen_backend: str, - variant_overrides: dict | None = None, - logger: Logger | None = None, -) -> bool: - """Apply safe defaults and the selected frame-generation backend.""" - if not ini_file.exists(): - return False - backend = framegen_backend if framegen_backend in {"auto", "nukems"} else "auto" - overrides = { - "FGInput": backend, - "FGOutput": backend, - "Fsr4Update": "true", - "LoadAsiPlugins": "true", - } - if variant_overrides: - overrides.update(variant_overrides) - migrated = migrate(ini_file, logger) - hq_updated = disable_hq_font_auto(ini_file, logger) - configured = apply_overrides(ini_file, overrides, logger) - return migrated and hq_updated and configured - - -def detect_backend(ini_file: Path) -> str: - """Return the configured backend when both FG keys agree.""" - try: - content = ini_file.read_text(encoding="utf-8", errors="replace") - except OSError: - return "auto" - values = [ - match.group(1).strip().lower() - for match in re.finditer(r"^FG(?:Input|Output)\s*=\s*(\S+)", content, re.MULTILINE) - ] - if values and all(value == "nukems" for value in values): - return "nukems" - return "auto" diff --git a/backend/files.py b/backend/files.py deleted file mode 100644 index c5ec839..0000000 --- a/backend/files.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Small filesystem helpers shared by backend services.""" - -import filecmp -import hashlib -import json -from pathlib import Path - - -def file_sha256(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as stream: - for chunk in iter(lambda: stream.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest() - - -def files_match(file_a: Path, file_b: Path) -> bool: - try: - return file_a.is_file() and file_b.is_file() and filecmp.cmp(file_a, file_b, shallow=False) - except OSError: - return False - - -def read_json(path: Path) -> dict: - try: - with path.open("r", encoding="utf-8") as stream: - payload = json.load(stream) - return payload if isinstance(payload, dict) else {} - except (OSError, ValueError, TypeError): - return {} - - -def write_json(path: Path, payload: dict) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("w", encoding="utf-8") as stream: - json.dump(payload, stream, indent=2) diff --git a/backend/patcher.py b/backend/patcher.py deleted file mode 100644 index 59d08f1..0000000 --- a/backend/patcher.py +++ /dev/null @@ -1,685 +0,0 @@ -"""Patch ownership, safe cleanup, and Steam-game operations.""" - -import os -import shutil -from datetime import datetime, timezone -from pathlib import Path -from typing import Callable - -from . import config -from .assets import ( - BACKUP_DIRECTORY_NAME, - FRAMEGEN_BACKENDS, - FSR4_UPSCALER_FILENAME, - INJECTOR_FILENAMES, - MARKER_FILENAME, - MANIFEST_SCHEMA_VERSION, - RESTORABLE_BACKUP_FILES, - SUPPORT_FILES, - VALID_DLL_NAMES, - VARIANT_EXTRA_FILENAMES, -) -from .bundle import BundleManager -from .files import file_sha256, files_match, read_json, write_json -from .steam import SteamLibrary - - -Logger = Callable[[str], None] - - -class PatchManager: - def __init__(self, home: Path, bundle: BundleManager, steam: SteamLibrary, logger: Logger): - self.home = home - self.bundle = bundle - self.steam = steam - self.logger = logger - - def _log(self, message: str) -> None: - self.logger(f"[Framegen] {message}") - - @staticmethod - def _relative_path(root: Path, path: Path) -> str: - return path.resolve().relative_to(root.resolve()).as_posix() - - def _safe_path(self, root: Path, value: str | Path) -> Path: - candidate = Path(value) - if candidate.is_absolute(): - resolved = candidate.resolve() - else: - resolved = (root / candidate).resolve() - try: - resolved.relative_to(root.resolve()) - except ValueError as exc: - raise ValueError(f"Path escapes patch directory: {value}") from exc - return resolved - - def _read_marker(self, marker_path: Path) -> dict: - return read_json(marker_path) - - def _find_marker(self, install_root: Path, appid: str | None = None) -> Path | None: - if not install_root.exists(): - return None - try: - markers = sorted(path for path in install_root.rglob(MARKER_FILENAME) if path.is_file()) - except OSError: - return None - for marker in markers: - metadata = self._read_marker(marker) - if appid is None or str(metadata.get("appid")) == str(appid): - return marker - return None - - def _marker_target(self, marker_path: Path, metadata: dict) -> Path: - target_value = metadata.get("target_relative") or metadata.get("target_dir") or str(marker_path.parent) - target = self._safe_path(marker_path.parent, target_value) - if not target.is_dir(): - raise FileNotFoundError(f"Patch target directory does not exist: {target}") - return target - - def _bundle_source_for_proxy(self, variant: str, dll_name: str) -> Path: - candidates = [ - self.bundle.variant_proxy_path(variant, dll_name), - self.bundle.variant_injector_path(variant), - self.bundle.fgmod_path / "renames" / dll_name, - self.bundle.fgmod_path / "OptiScaler.dll", - ] - for candidate in candidates: - if candidate and candidate.exists(): - return candidate - raise FileNotFoundError(f"No injector source found for {dll_name}") - - def _copy_plan(self, target_dir: Path, variant: str) -> list[tuple[str, Path]]: - plan: list[tuple[str, Path]] = [] - # The proxy is the only file selected by the user; all other files are bundle-owned. - plan.append(("__proxy__", self._bundle_source_for_proxy(variant, "dxgi.dll"))) - for filename in SUPPORT_FILES: - source = self.bundle.fgmod_path / filename - if source.is_file(): - plan.append((filename, source)) - upscaler = self.bundle.variant_path(variant) - if upscaler.is_file(): - plan.append((FSR4_UPSCALER_FILENAME, upscaler)) - for extra in self.bundle.variant_extra_files(variant): - source = self.bundle.variant_extra_path(variant, extra["name"]) - if source.is_file(): - plan.append((extra["name"], source)) - - for source_root, relative_root in ( - (self.bundle.fgmod_path / "plugins", Path("plugins")), - (self.bundle.fgmod_path / "D3D12_Optiscaler", Path("D3D12_Optiscaler")), - ): - if not source_root.is_dir(): - continue - for source in sorted(path for path in source_root.rglob("*") if path.is_file()): - plan.append(((relative_root / source.relative_to(source_root)).as_posix(), source)) - return plan - - def _backup_destination(self, target_dir: Path, relative: str, source: Path, force: bool = False) -> str | None: - destination = self._safe_path(target_dir, relative) - if not destination.exists(): - return None - if destination.is_dir(): - raise IsADirectoryError(f"Patch destination is a directory: {destination}") - if not force and files_match(destination, source): - return None - - if relative in RESTORABLE_BACKUP_FILES: - legacy_backup = destination.with_name(destination.name + ".b") - if not legacy_backup.exists(): - shutil.copy2(destination, legacy_backup) - elif not files_match(legacy_backup, destination): - raise RuntimeError(f"Refusing to overwrite an existing backup: {legacy_backup}") - return self._relative_path(target_dir, legacy_backup) - - backup = target_dir / BACKUP_DIRECTORY_NAME / relative - backup.parent.mkdir(parents=True, exist_ok=True) - if backup.exists(): - if not files_match(backup, destination): - raise RuntimeError(f"Refusing to overwrite an existing backup: {backup}") - else: - shutil.copy2(destination, backup) - return self._relative_path(target_dir, backup) - - def _cleanup_empty_directories(self, target_dir: Path, managed_files: list[str]) -> None: - directories = sorted( - {self._safe_path(target_dir, path).parent for path in managed_files}, - key=lambda path: len(path.parts), - reverse=True, - ) - for directory in directories: - if directory == target_dir: - continue - try: - directory.rmdir() - except OSError: - pass - backup_dir = target_dir / BACKUP_DIRECTORY_NAME - try: - backup_dir.rmdir() - except OSError: - pass - - def _rollback( - self, - target_dir: Path, - managed: list[dict], - backups: list[dict], - ) -> None: - for entry in reversed(managed): - try: - path = self._safe_path(target_dir, entry["path"]) - backup = next((item["backup"] for item in backups if item["path"] == entry["path"]), None) - if backup: - backup_path = self._safe_path(target_dir, backup) - if backup_path.exists(): - if path.exists(): - path.unlink() - shutil.move(backup_path, path) - elif path.exists() and entry.get("sha256") == file_sha256(path): - path.unlink() - except (OSError, ValueError): - continue - self._cleanup_empty_directories(target_dir, [entry["path"] for entry in managed]) - - def _patch_directory( - self, - target_dir: Path, - dll_name: str, - variant: str, - framegen_backend: str, - *, - appid: str, - game_name: str, - original_launch_options: str, - previous_variant: str | None = None, - ) -> dict: - if dll_name not in VALID_DLL_NAMES: - return {"status": "error", "message": f"Invalid proxy DLL name: {dll_name}"} - if not self.bundle.fgmod_path.exists() or not (self.bundle.fgmod_path / "OptiScaler.dll").exists(): - return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} - - variant = self.bundle.selected_variant(variant) - framegen_backend = self.bundle.selected_backend(framegen_backend) - previous_marker = target_dir / MARKER_FILENAME - if previous_marker.exists(): - return {"status": "error", "message": f"Patch marker already exists in {target_dir}; clean it up before patching."} - variant_overrides = self.bundle.variant_config_overrides(variant) - if previous_variant == "rdna2-valve-411-pre10" and variant != previous_variant: - variant_overrides.setdefault("FSR.Fsr4ForceModel", "auto") - variant_overrides.setdefault("Plugins.LoadCustomAmdxc64OnRdna2", "false") - - managed: list[dict] = [] - backups: list[dict] = [] - try: - plan = self._copy_plan(target_dir, variant) - proxy_source = self._bundle_source_for_proxy(variant, dll_name) - plan[0] = (dll_name, proxy_source) - config_path = target_dir / "OptiScaler.ini" - if config_path.exists(): - plan.append(("OptiScaler.ini", config_path)) - elif (self.bundle.fgmod_path / "OptiScaler.ini").exists(): - plan.append(("OptiScaler.ini", self.bundle.fgmod_path / "OptiScaler.ini")) - - for relative, source in plan: - destination = self._safe_path(target_dir, relative) - if not source.is_file(): - continue - backup = self._backup_destination( - target_dir, - relative, - source, - # Preserve every pre-existing destination. Identical bytes do - # not prove that a file belongs to this plugin. - force=destination.exists(), - ) - if backup: - backups.append({ - "path": relative, - "backup": backup, - "sha256": file_sha256(self._safe_path(target_dir, backup)), - }) - destination.parent.mkdir(parents=True, exist_ok=True) - if destination.resolve() != source.resolve(): - shutil.copy2(source, destination) - managed.append({ - "path": relative, - "sha256": file_sha256(destination), - "source_sha256": file_sha256(source), - }) - - if config_path.exists() and not config.configure( - config_path, - framegen_backend, - variant_overrides=variant_overrides, - logger=self.logger, - ): - raise RuntimeError("Could not update OptiScaler.ini") - config_entry = next((entry for entry in managed if entry["path"] == "OptiScaler.ini"), None) - if config_entry: - config_entry["sha256"] = file_sha256(config_path) - - payload = { - "schema_version": MANIFEST_SCHEMA_VERSION, - "appid": str(appid), - "game_name": game_name, - "dll_name": dll_name, - "target_dir": str(target_dir), - "target_relative": ".", - "original_launch_options": original_launch_options, - "backed_up_files": [item["path"] for item in backups], - "backups": backups, - "managed_files": managed, - "managed_directories": sorted({str(Path(item["path"]).parent) for item in managed if Path(item["path"]).parent != Path(".")}), - "optiscaler_version": self.bundle.fgmod_version(), - "fsr4_variant": variant, - "fsr4_variant_label": self.bundle.variant_info(variant)["label"], - "fsr4_upscaler_sha256": file_sha256(target_dir / FSR4_UPSCALER_FILENAME) if (target_dir / FSR4_UPSCALER_FILENAME).exists() else None, - "framegen_backend": framegen_backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], - "patched_at": datetime.now(timezone.utc).isoformat(), - } - # The marker is written last, so an interrupted copy is never advertised as a complete patch. - write_json(previous_marker, payload) - return { - "status": "success", - "message": f"OptiScaler files copied to {target_dir} using {self.bundle.variant_info(variant)['label']}", - "fsr4_variant": variant, - "fsr4_variant_label": self.bundle.variant_info(variant)["label"], - "fsr4_upscaler_sha256": payload["fsr4_upscaler_sha256"], - "optiscaler_version": payload["optiscaler_version"], - "framegen_backend": framegen_backend, - } - except (OSError, RuntimeError, ValueError) as exc: - self._rollback(target_dir, managed, backups) - self._log(f"Patch failed for {target_dir}: {exc}") - return {"status": "error", "message": f"Patch failed: {exc}"} - - def _managed_entries(self, target_dir: Path, metadata: dict) -> tuple[list[tuple[dict, Path]], list[str]]: - entries: list[tuple[dict, Path]] = [] - errors: list[str] = [] - for entry in metadata.get("managed_files", []): - if not isinstance(entry, dict) or not entry.get("path"): - errors.append("Patch marker contains an invalid managed file entry.") - continue - try: - path = self._safe_path(target_dir, entry["path"]) - except ValueError as exc: - errors.append(str(exc)) - continue - expected = str(entry.get("sha256") or "") - if not expected: - errors.append(f"Patch marker has no hash for {entry['path']}.") - elif path.exists() and (not path.is_file() or file_sha256(path).lower() != expected.lower()): - errors.append(f"Refusing to remove modified managed file: {path}") - entries.append((entry, path)) - return entries, errors - - def _restore_backups(self, target_dir: Path, metadata: dict) -> list[str]: - restored: list[str] = [] - for item in metadata.get("backups", []): - if not isinstance(item, dict) or not item.get("path") or not item.get("backup"): - raise ValueError("Patch marker contains an invalid backup entry.") - path = self._safe_path(target_dir, item["path"]) - backup = self._safe_path(target_dir, item["backup"]) - if not backup.exists(): - continue - if not backup.is_file(): - raise ValueError(f"Patch backup is not a file: {backup}") - expected = str(item.get("sha256") or "") - if expected and file_sha256(backup).lower() != expected.lower(): - raise RuntimeError(f"Refusing to restore a modified backup: {backup}") - if path.exists(): - raise RuntimeError(f"Refusing to overwrite a file while restoring backup: {path}") - path.parent.mkdir(parents=True, exist_ok=True) - shutil.move(backup, path) - restored.append(str(item["path"])) - - # Older releases used ``name.b`` backups and listed only their base names. - for raw_name in metadata.get("backed_up_files", []): - name = Path(str(raw_name)).name - if name not in RESTORABLE_BACKUP_FILES: - continue - original = self._safe_path(target_dir, name) - backup = self._safe_path(target_dir, name + ".b") - if backup.exists() and not original.exists(): - shutil.move(backup, original) - restored.append(name) - return restored - - def _validate_backups(self, target_dir: Path, metadata: dict) -> None: - for item in metadata.get("backups", []): - if not isinstance(item, dict) or not item.get("path") or not item.get("backup"): - raise ValueError("Patch marker contains an invalid backup entry.") - self._safe_path(target_dir, item["path"]) - backup = self._safe_path(target_dir, item["backup"]) - if not backup.exists(): - continue - if not backup.is_file(): - raise ValueError(f"Patch backup is not a file: {backup}") - expected = str(item.get("sha256") or "") - if expected and file_sha256(backup).lower() != expected.lower(): - raise RuntimeError(f"Refusing to restore a modified backup: {backup}") - - def _legacy_owned_paths(self, target_dir: Path, metadata: dict) -> list[Path]: - paths: list[Path] = [] - for entry in metadata.get("managed_files", []): - if isinstance(entry, dict) and entry.get("path"): - try: - paths.append(self._safe_path(target_dir, entry["path"])) - except ValueError: - continue - - variant = self.bundle.normalize_variant(metadata.get("fsr4_variant")) - proxy_sources = [ - self.bundle.fgmod_path / "OptiScaler.dll", - self.bundle.fgmod_path / "renames", - self.bundle.variant_dir(variant) / "renames", - ] - for filename in INJECTOR_FILENAMES: - target = target_dir / filename - if not target.is_file(): - continue - matches_bundle = False - for source in proxy_sources: - candidate = source if source.is_file() and source.name == filename else source / filename - if candidate.is_file() and files_match(target, candidate): - matches_bundle = True - break - if matches_bundle: - paths.append(target) - for filename in SUPPORT_FILES + VARIANT_EXTRA_FILENAMES + [FSR4_UPSCALER_FILENAME]: - target = target_dir / filename - if target.is_file() and self.bundle.is_managed_support_file(target): - paths.append(target) - for relative in (Path("plugins/OptiPatcher.asi"),): - target = target_dir / relative - source = self.bundle.fgmod_path / relative - if target.is_file() and source.is_file() and files_match(target, source): - paths.append(target) - d3d12_source = self.bundle.fgmod_path / "D3D12_Optiscaler" - d3d12_target = target_dir / "D3D12_Optiscaler" - if d3d12_source.is_dir() and d3d12_target.is_dir(): - for source in d3d12_source.rglob("*"): - if source.is_file(): - candidate = d3d12_target / source.relative_to(d3d12_source) - if candidate.is_file() and files_match(candidate, source): - paths.append(candidate) - unique: list[Path] = [] - seen: set[str] = set() - for path in paths: - if str(path) not in seen: - unique.append(path) - seen.add(str(path)) - return unique - - def _unpatch_target(self, target_dir: Path, marker_path: Path | None) -> dict: - metadata = self._read_marker(marker_path) if marker_path else { - "backed_up_files": [ - name for name in RESTORABLE_BACKUP_FILES - if (target_dir / f"{name}.b").is_file() - ], - } - try: - if marker_path and marker_path.exists() and not metadata: - return {"status": "error", "message": f"Patch marker is unreadable: {marker_path}"} - schema_version = int(metadata.get("schema_version") or 1) if metadata else 1 - if schema_version >= MANIFEST_SCHEMA_VERSION: - entries, errors = self._managed_entries(target_dir, metadata) - if errors: - return {"status": "error", "message": " ".join(errors)} - self._validate_backups(target_dir, metadata) - managed_paths = [path for _, path in entries] - for _, path in entries: - if path.exists(): - path.unlink() - restored = self._restore_backups(target_dir, metadata) - self._cleanup_empty_directories(target_dir, [str(path.relative_to(target_dir)) for path in managed_paths]) - else: - owned = self._legacy_owned_paths(target_dir, metadata) - entries, errors = self._managed_entries(target_dir, metadata) - if errors: - return {"status": "error", "message": " ".join(errors)} - owned.extend(path for _, path in entries if path not in owned) - for path in owned: - if path.is_file(): - path.unlink() - restored = self._restore_backups(target_dir, metadata) - self._cleanup_empty_directories(target_dir, [str(path.relative_to(target_dir)) for path in owned]) - if marker_path and marker_path.exists(): - marker_path.unlink() - return { - "status": "success", - "message": f"OptiScaler files removed from {target_dir}", - "restored_files": restored, - } - except (OSError, RuntimeError, ValueError) as exc: - self._log(f"Unpatch failed for {target_dir}: {exc}") - return {"status": "error", "message": f"Unpatch failed: {exc}"} - - def resolve_target_directory(self, directory: str) -> Path: - target = Path(directory).expanduser() - if not target.exists(): - raise FileNotFoundError(f"Target directory does not exist: {directory}") - if not target.is_dir(): - raise NotADirectoryError(f"Target path is not a directory: {directory}") - if not os.access(target, os.W_OK | os.X_OK): - raise PermissionError(f"Insufficient permissions for {directory}") - return target.resolve() - - def manual_patch_directory( - self, - directory: str, - dll_name: str, - fsr4_variant: str, - framegen_backend: str | None = None, - ) -> dict: - try: - target = self.resolve_target_directory(directory) - except (FileNotFoundError, NotADirectoryError, PermissionError) as exc: - return {"status": "error", "message": str(exc)} - marker = target / MARKER_FILENAME - previous_metadata = self._read_marker(marker) if marker.exists() else {} - previous_variant = previous_metadata.get("fsr4_variant") - if marker.exists(): - if str(previous_metadata.get("appid")) not in ("", "manual", "None"): - return {"status": "error", "message": "This directory is managed by an AppID patch; unpatch it there first."} - cleanup = self._unpatch_target(target, marker) - if cleanup["status"] != "success": - return cleanup - return self._patch_directory( - target, - dll_name, - self.bundle.selected_variant(fsr4_variant), - self.bundle.selected_backend(framegen_backend), - appid="manual", - game_name="Manual patch", - original_launch_options="", - previous_variant=previous_variant, - ) - - def manual_unpatch_directory(self, directory: str) -> dict: - try: - target = self.resolve_target_directory(directory) - except (FileNotFoundError, NotADirectoryError, PermissionError) as exc: - return {"status": "error", "message": str(exc)} - marker = target / MARKER_FILENAME - return self._unpatch_target(target, marker if marker.exists() else None) - - def list_installed_games(self) -> dict: - try: - return { - "status": "success", - "games": [ - { - "appid": str(game["appid"]), - "name": game["name"], - "install_found": Path(game["install_path"]).exists(), - } - for game in self.steam.find_installed_games() - ], - } - except Exception as exc: - self._log(str(exc)) - return {"status": "error", "message": str(exc)} - - def get_game_status(self, appid: str) -> dict: - try: - game_info = self.steam.game_record(str(appid)) - if not game_info: - return { - "status": "success", "appid": str(appid), "install_found": False, "patched": False, - "dll_name": None, "target_dir": None, "fsr4_variant": None, "fsr4_variant_label": None, - "framegen_backend": None, "message": "Game not found in Steam library.", - } - install_root = Path(game_info["install_path"]) - base = { - "status": "success", "appid": str(appid), "name": game_info["name"], - "install_found": install_root.exists(), "patched": False, "dll_name": None, - "target_dir": None, "fsr4_variant": None, "fsr4_variant_label": None, - "framegen_backend": None, - } - if not install_root.exists(): - return {**base, "message": "Game install directory not found."} - marker = self._find_marker(install_root, str(appid)) - if not marker: - return {**base, "message": "Not patched."} - metadata = self._read_marker(marker) - target_dir = self._marker_target(marker, metadata) - dll_name = str(metadata.get("dll_name") or "dxgi.dll") - upscaler = target_dir / FSR4_UPSCALER_FILENAME - upscaler_sha256 = file_sha256(upscaler) if upscaler.is_file() else None - variant = self.bundle.detect_variant(target_dir, upscaler_sha256) or self.bundle.normalize_variant(metadata.get("fsr4_variant")) - backend = self.bundle.normalize_backend(metadata.get("framegen_backend") or self.bundle.selected_backend()) - dll_present = (target_dir / dll_name).is_file() - return { - **base, - "patched": dll_present, - "dll_name": dll_name, - "target_dir": str(target_dir), - "patched_at": metadata.get("patched_at"), - "optiscaler_version": metadata.get("optiscaler_version"), - "fsr4_variant": variant, - "fsr4_variant_label": self.bundle.variant_info(variant)["label"], - "fsr4_upscaler_sha256": upscaler_sha256, - "framegen_backend": backend, - "framegen_backend_label": FRAMEGEN_BACKENDS[backend], - "message": f"Patched using {dll_name} with {self.bundle.variant_info(variant)['label']}." if dll_present else f"Marker found but {dll_name} is missing. Reinstall recommended.", - } - except Exception as exc: - self._log(f"get_game_status failed for {appid}: {exc}") - return {"status": "error", "message": str(exc)} - - @staticmethod - def build_managed_launch_options(dll_name: str) -> str: - if dll_name == "OptiScaler.asi": - return "SteamDeck=0 %command%" - base = dll_name.replace(".dll", "") - return f"WINEDLLOVERRIDES={base}=n,b SteamDeck=0 %command%" - - @staticmethod - def is_managed_launch_options(options: str) -> bool: - normalized = " ".join((options or "").replace('"', "").split()) - if not normalized: - return False - return any( - dll_name != "OptiScaler.asi" and f"WINEDLLOVERRIDES={dll_name.replace('.dll', '')}=n,b" in normalized - for dll_name in VALID_DLL_NAMES - ) or "fgmod/fgmod" in normalized - - def patch_game( - self, - appid: str, - dll_name: str, - current_launch_options: str, - fsr4_variant: str, - framegen_backend: str | None = None, - ) -> dict: - try: - if dll_name not in VALID_DLL_NAMES: - return {"status": "error", "message": f"Invalid proxy DLL name: {dll_name}"} - game_info = self.steam.game_record(str(appid)) - if not game_info: - return {"status": "error", "message": "Game not found in Steam library."} - install_root = Path(game_info["install_path"]) - if not install_root.exists(): - return {"status": "error", "message": "Game install directory does not exist."} - if self.steam.is_game_running(game_info): - return {"status": "error", "message": "Close the game before patching."} - if not self.bundle.fgmod_path.exists(): - return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} - - existing_marker = self._find_marker(install_root) - existing_metadata = self._read_marker(existing_marker) if existing_marker else {} - if existing_marker and str(existing_metadata.get("appid")) not in (str(appid), "manual", "None"): - return {"status": "error", "message": "Another Framegen patch marker was found in this game install."} - original_options = current_launch_options or "" - stored_options = str(existing_metadata.get("original_launch_options") or "") - if stored_options and not self.is_managed_launch_options(stored_options): - original_options = stored_options - if self.is_managed_launch_options(original_options): - original_options = "" - - if existing_marker: - old_target = self._marker_target(existing_marker, existing_metadata) - cleanup = self._unpatch_target(old_target, existing_marker) - if cleanup["status"] != "success": - return cleanup - target_dir, _ = self.steam.guess_patch_target(game_info) - result = self._patch_directory( - target_dir, - dll_name, - self.bundle.selected_variant(fsr4_variant), - self.bundle.selected_backend(framegen_backend), - appid=str(appid), - game_name=game_info["name"], - original_launch_options=original_options, - previous_variant=existing_metadata.get("fsr4_variant"), - ) - if result["status"] != "success": - return result - backend = result.get("framegen_backend") or self.bundle.selected_backend(framegen_backend) - return { - **result, - "appid": str(appid), - "name": game_info["name"], - "dll_name": dll_name, - "target_dir": str(target_dir), - "launch_options": self.build_managed_launch_options(dll_name), - "original_launch_options": original_options, - "framegen_backend": backend, - "message": f"Patched {game_info['name']} using {dll_name} with {result.get('fsr4_variant_label') or self.bundle.variant_info(fsr4_variant)['label']}.", - } - except Exception as exc: - self._log(f"patch_game failed for {appid}: {exc}") - return {"status": "error", "message": str(exc)} - - def unpatch_game(self, appid: str) -> dict: - try: - game_info = self.steam.game_record(str(appid)) - if not game_info: - return {"status": "error", "message": "Game not found in Steam library."} - install_root = Path(game_info["install_path"]) - if not install_root.exists(): - return {"status": "success", "appid": str(appid), "name": game_info["name"], "launch_options": "", "message": "Game install directory does not exist."} - if self.steam.is_game_running(game_info): - return {"status": "error", "message": "Close the game before unpatching."} - marker = self._find_marker(install_root, str(appid)) - if not marker: - return {"status": "success", "appid": str(appid), "name": game_info["name"], "launch_options": "", "message": "No Framegen patch found for this game."} - metadata = self._read_marker(marker) - target_dir = self._marker_target(marker, metadata) - original_options = str(metadata.get("original_launch_options") or "") - cleanup = self._unpatch_target(target_dir, marker) - if cleanup["status"] != "success": - return {**cleanup, "appid": str(appid), "name": game_info["name"]} - self._log(f"unpatch_game success: appid={appid} target={target_dir}") - return { - "status": "success", - "appid": str(appid), - "name": game_info["name"], - "launch_options": original_options, - "message": f"Unpatched {game_info['name']}.", - } - except Exception as exc: - self._log(f"unpatch_game failed for {appid}: {exc}") - return {"status": "error", "message": str(exc)} diff --git a/backend/steam.py b/backend/steam.py deleted file mode 100644 index c2e90b0..0000000 --- a/backend/steam.py +++ /dev/null @@ -1,187 +0,0 @@ -"""Steam library discovery and conservative executable selection.""" - -import re -import subprocess -from pathlib import Path -from typing import Callable - - -Logger = Callable[[str], None] - -BAD_EXE_SUBSTRINGS = [ - "crashreport", - "crashreportclient", - "eac", - "easyanticheat", - "beclient", - "eosbootstrap", - "benchmark", - "uninstall", - "setup", - "launcher", - "updater", - "bootstrap", - "_redist", - "prereq", -] - - -class SteamLibrary: - def __init__(self, home: Path, logger: Logger): - self.home = home - self.logger = logger - - def steam_common_path(self) -> Path: - return self.home / ".local" / "share" / "Steam" / "steamapps" / "common" - - def _steam_root_candidates(self) -> list[Path]: - candidates = [ - self.home / ".local" / "share" / "Steam", - self.home / ".steam" / "steam", - self.home / ".steam" / "root", - self.home / ".var" / "app" / "com.valvesoftware.Steam" / "home" / ".local" / "share" / "Steam", - self.home / ".var" / "app" / "com.valvesoftware.Steam" / "home" / ".steam" / "steam", - ] - unique: list[Path] = [] - seen: set[str] = set() - for candidate in candidates: - key = str(candidate) - if key not in seen: - unique.append(candidate) - seen.add(key) - return unique - - def _steam_library_paths(self) -> list[Path]: - library_paths: list[Path] = [] - seen: set[str] = set() - for steam_root in self._steam_root_candidates(): - if steam_root.exists() and str(steam_root) not in seen: - library_paths.append(steam_root) - seen.add(str(steam_root)) - library_file = steam_root / "steamapps" / "libraryfolders.vdf" - if not library_file.exists(): - continue - try: - for line in library_file.read_text(encoding="utf-8", errors="replace").splitlines(): - if '"path"' not in line: - continue - path = line.split('"path"', 1)[1].strip().strip('"').replace("\\\\", "/") - candidate = Path(path) - if str(candidate) not in seen: - library_paths.append(candidate) - seen.add(str(candidate)) - except OSError as exc: - self.logger(f"[Framegen] failed to parse libraryfolders: {library_file}: {exc}") - return library_paths - - def find_installed_games(self, appid: str | None = None) -> list[dict]: - games: list[dict] = [] - for library_path in self._steam_library_paths(): - steamapps_path = library_path / "steamapps" - if not steamapps_path.exists(): - continue - for appmanifest in steamapps_path.glob("appmanifest_*.acf"): - game_info: dict = { - "appid": "", - "name": "", - "library_path": str(library_path), - "install_path": "", - } - install_dir = "" - try: - for line in appmanifest.read_text(encoding="utf-8", errors="replace").splitlines(): - if '"appid"' in line: - game_info["appid"] = line.split('"appid"', 1)[1].strip().strip('"') - elif '"name"' in line: - game_info["name"] = line.split('"name"', 1)[1].strip().strip('"') - elif '"installdir"' in line: - install_dir = line.split('"installdir"', 1)[1].strip().strip('"') - except OSError as exc: - self.logger(f"[Framegen] skipping manifest {appmanifest}: {exc}") - continue - if not game_info["appid"] or not game_info["name"]: - continue - if "Proton" in game_info["name"] or "Steam Linux Runtime" in game_info["name"]: - continue - game_info["install_path"] = str(steamapps_path / "common" / install_dir) if install_dir else str(Path()) - if appid is None or str(game_info["appid"]) == str(appid): - games.append(game_info) - deduped = {str(game["appid"]): game for game in games} - return sorted(deduped.values(), key=lambda game: game["name"].lower()) - - def game_record(self, appid: str) -> dict | None: - matches = self.find_installed_games(appid) - return matches[0] if matches else None - - @staticmethod - def normalized_path_string(value: str) -> str: - normalized = value.lower().replace("\\", "/") - normalized = normalized.replace("z:/", "/") - return normalized.replace("//", "/") - - def candidate_executables(self, install_root: Path) -> list[Path]: - if not install_root.exists(): - return [] - try: - return [exe for exe in install_root.rglob("*.exe") if exe.is_file()] - except OSError as exc: - self.logger(f"[Framegen] exe scan failed for {install_root}: {exc}") - return [] - - def _exe_score(self, exe: Path, install_root: Path, game_name: str) -> int: - normalized = self.normalized_path_string(str(exe)) - name = exe.name.lower() - score = 0 - if normalized.endswith("-win64-shipping.exe"): - score += 300 - if "shipping.exe" in name: - score += 220 - if "/binaries/win64/" in normalized: - score += 200 - if "/win64/" in normalized: - score += 80 - if exe.parent == install_root: - score += 20 - sanitized_game = re.sub(r"[^a-z0-9]", "", game_name.lower()) - sanitized_name = re.sub(r"[^a-z0-9]", "", exe.stem.lower()) - sanitized_root = re.sub(r"[^a-z0-9]", "", install_root.name.lower()) - if sanitized_game and sanitized_game in sanitized_name: - score += 120 - if sanitized_root and sanitized_root in sanitized_name: - score += 90 - for bad in BAD_EXE_SUBSTRINGS: - if bad in normalized: - score -= 200 - return score - len(exe.parts) - - def best_running_executable(self, candidates: list[Path]) -> Path | None: - if not candidates: - return None - try: - result = subprocess.run(["ps", "-eo", "args="], capture_output=True, text=True, check=False) - process_lines = result.stdout.splitlines() - except OSError as exc: - self.logger(f"[Framegen] running exe scan failed: {exc}") - return None - normalized_candidates = [(exe, self.normalized_path_string(str(exe))) for exe in candidates] - matches: list[tuple[int, Path]] = [] - for line in process_lines: - normalized_line = self.normalized_path_string(line) - for exe, normalized_exe in normalized_candidates: - if normalized_exe in normalized_line: - matches.append((len(normalized_exe), exe)) - return max(matches, key=lambda item: item[0])[1] if matches else None - - def guess_patch_target(self, game_info: dict) -> tuple[Path, Path | None]: - install_root = Path(game_info["install_path"]) - candidates = self.candidate_executables(install_root) - if not candidates: - return install_root, None - running_exe = self.best_running_executable(candidates) - if running_exe: - return running_exe.parent, running_exe - best = max(candidates, key=lambda exe: self._exe_score(exe, install_root, game_info["name"])) - return best.parent, best - - def is_game_running(self, game_info: dict) -> bool: - return self.best_running_executable(self.candidate_executables(Path(game_info["install_path"]))) is not None diff --git a/main.py b/main.py index c5a3e01..a48c847 100644 --- a/main.py +++ b/main.py @@ -1,17 +1,17 @@ """Decky entry point for the Framegen plugin. Decky-facing methods live here; filesystem, Steam, configuration, and patch -ownership concerns are implemented in ``backend`` services. +ownership concerns are implemented in focused service modules. """ from pathlib import Path import decky -from backend.assets import DEFAULT_FRAMEGEN_BACKEND, DEFAULT_FSR4_VARIANT -from backend.bundle import BundleManager -from backend.patcher import PatchManager -from backend.steam import SteamLibrary +from services.assets import DEFAULT_FRAMEGEN_BACKEND, DEFAULT_FSR4_VARIANT +from services.bundle import BundleManager +from services.patcher import PatchManager +from services.steam import SteamLibrary class Plugin: diff --git a/py_modules/services/__init__.py b/py_modules/services/__init__.py new file mode 100644 index 0000000..737c993 --- /dev/null +++ b/py_modules/services/__init__.py @@ -0,0 +1 @@ +"""Focused services for Decky Framegen.""" diff --git a/py_modules/services/assets.py b/py_modules/services/assets.py new file mode 100644 index 0000000..69f71ba --- /dev/null +++ b/py_modules/services/assets.py @@ -0,0 +1,184 @@ +"""Static asset metadata and patch file inventories.""" + +OPTISCALER_ARCHIVE_ASSET = { + "name": "Optiscaler_0.9.4-final.20260718._MM.7z", + "sha256": "575cb4df866116093df75af607e37fd70e10f5163e0f23fd5c804142e80ef0ad", + "version": "0.9.4-final.20260718", +} + +FSR4_INT8_ASSET = { + "name": "amd_fidelityfx_upscaler_dx12.dll", + "sha256": "c7720bc16bede334f59a1a32cd22edbcbbb159685ed5240e61350a5fb0bc8a94", + "version": "4.0.2c", +} + +FSR4_OFFICIAL_411_ASSET = { + "name": "amdxcffx64.dll", + "sha256": "a2b136b6affd35a49b141a936be935f7d5ddc8d8f9b8c9afbe62ff9ddb2538a0", + "version": "4.1.1-official", +} + +FSR4_VALVE_411_ASSET = { + "name": "amdxcffx64_valve_2.3.0.2913.dll", + "sha256": "4e7dc37aebea3a90e3d3cc43e24cb2b54176b2535315f20dbe63b3b7cfc56b1e", + "version": "4.1.1-valve-2.3.0.2913", +} + +AMDXC64_RDNA2_ASSET = { + "name": "amdxc64.dll", + "sha256": "a0a0af61d475e30a70966b3459f3793df772faf8f26ae3261d10554ff592cbd5", + "version": "8.18.10.0474", +} + +OPTISCALER_PRE10_ASSET = { + "name": "OptiScaler_0.10.0-pre1.20260622_135940.dll", + "sha256": "b374b19081cc066365d0c6da4808d768e16469b0cbdfc478b6e95999947d5364", + "version": "0.10.0-pre1.20260622_135940", +} + +OPTIPATCHER_ASSET = { + "name": "OptiPatcher_rolling.asi", + "sha256": "88b9e1be3559737cd205fdf5f2c8550cf1923fb1def4c603e5bf03c3e84131b1", + "version": "rolling", +} + +FSR4_UPSCALER_FILENAME = "amd_fidelityfx_upscaler_dx12.dll" +FSR4_DRIVER_OVERRIDE_FILENAME = "amdxcffx64.dll" +INSTALL_MANIFEST_FILENAME = "install-manifest.json" +VERSION_FILENAME = "version.txt" +DEFAULT_FSR4_VARIANT = "rdna23-int8" +DEFAULT_FRAMEGEN_BACKEND = "auto" +FRAMEGEN_BACKENDS = { + "auto": "OptiScaler automatic selection", + "nukems": "Nukem's DLSSG → FSR3", +} + +FSR4_VARIANTS = { + "rdna23-int8": { + "label": "4.0.2c / RDNA2-3 compatibility", + "dir_name": "fsr4-rdna2-3", + "sha256": "c7720bc16bede334f59a1a32cd22edbcbbb159685ed5240e61350a5fb0bc8a94", + "source_asset_name": FSR4_INT8_ASSET["name"], + "source_version": FSR4_INT8_ASSET["version"], + "uses_archive_native": False, + "extra_files": [], + }, + "rdna4-native": { + "label": "4.1.1 SDK / RDNA3 dGPU + RDNA4", + "dir_name": "fsr4-rdna4", + "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", + "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], + "source_version": OPTISCALER_ARCHIVE_ASSET["version"], + "uses_archive_native": True, + "extra_files": [], + }, + "rdna34-official-411": { + "label": "4.1.1 driver override / RDNA3-4", + "dir_name": "fsr4-rdna3-4-official-411", + "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", + "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], + "source_version": OPTISCALER_ARCHIVE_ASSET["version"], + "uses_archive_native": True, + "extra_files": [ + { + "name": FSR4_DRIVER_OVERRIDE_FILENAME, + "sha256": FSR4_OFFICIAL_411_ASSET["sha256"], + "source_asset_name": FSR4_OFFICIAL_411_ASSET["name"], + "source_version": FSR4_OFFICIAL_411_ASSET["version"], + } + ], + "config_overrides": {}, + }, + "rdna2-valve-411-pre10": { + "label": "4.1.1 Valve RDNA2 compatibility", + "dir_name": "fsr4-rdna2-valve-411-pre10", + "sha256": "d0dcccc74a43c44ba435b7a369b456e0970d8a4464e4bd683119b374f2c9fb46", + "source_asset_name": OPTISCALER_ARCHIVE_ASSET["name"], + "source_version": OPTISCALER_ARCHIVE_ASSET["version"], + "uses_archive_native": True, + "injector": { + "name": "OptiScaler.dll", + "sha256": OPTISCALER_PRE10_ASSET["sha256"], + "source_asset_name": OPTISCALER_PRE10_ASSET["name"], + "source_version": OPTISCALER_PRE10_ASSET["version"], + }, + "extra_files": [ + { + "name": FSR4_DRIVER_OVERRIDE_FILENAME, + "sha256": FSR4_VALVE_411_ASSET["sha256"], + "source_asset_name": FSR4_VALVE_411_ASSET["name"], + "source_version": FSR4_VALVE_411_ASSET["version"], + }, + { + "name": "amdxc64.dll", + "sha256": AMDXC64_RDNA2_ASSET["sha256"], + "source_asset_name": AMDXC64_RDNA2_ASSET["name"], + "source_version": AMDXC64_RDNA2_ASSET["version"], + }, + ], + "config_overrides": { + "FSR.Fsr4ForceModel": "2", + "Plugins.LoadCustomAmdxc64OnRdna2": "true", + }, + }, +} + +VARIANT_EXTRA_FILENAMES = sorted( + { + extra_file["name"] + for variant in FSR4_VARIANTS.values() + for extra_file in variant.get("extra_files", []) + } +) + +PROXY_DLL_BACKUPS = [ + "dxgi.dll", + "winmm.dll", + "dbghelp.dll", + "version.dll", + "wininet.dll", + "winhttp.dll", + "OptiScaler.asi", +] +VALID_DLL_NAMES = set(PROXY_DLL_BACKUPS) + +INJECTOR_FILENAMES = [ + *PROXY_DLL_BACKUPS, + "nvngx.dll", + "_nvngx.dll", + "nvngx-wrapper.dll", + "dlss-enabler.dll", + "OptiScaler.dll", +] + +ORIGINAL_DLL_BACKUPS = [ + "d3dcompiler_47.dll", + "amd_fidelityfx_dx12.dll", + "amd_fidelityfx_framegeneration_dx12.dll", + FSR4_UPSCALER_FILENAME, + FSR4_DRIVER_OVERRIDE_FILENAME, + "amdxc64.dll", + "amd_fidelityfx_vk.dll", +] + +RESTORABLE_BACKUP_FILES = [ + *PROXY_DLL_BACKUPS, + *ORIGINAL_DLL_BACKUPS, +] + +SUPPORT_FILES = [ + "libxess.dll", + "libxess_dx11.dll", + "libxess_fg.dll", + "libxell.dll", + "amd_fidelityfx_dx12.dll", + "amd_fidelityfx_framegeneration_dx12.dll", + "amd_fidelityfx_vk.dll", + "dlssg_to_fsr3_amd_is_better.dll", + "fakenvapi.dll", + "fakenvapi.ini", +] + +MARKER_FILENAME = "FRAMEGEN_PATCH" +MANIFEST_SCHEMA_VERSION = 2 +BACKUP_DIRECTORY_NAME = ".framegen-backups" diff --git a/py_modules/services/bundle.py b/py_modules/services/bundle.py new file mode 100644 index 0000000..a6ef9e4 --- /dev/null +++ b/py_modules/services/bundle.py @@ -0,0 +1,522 @@ +"""Management of the shared OptiScaler bundle in ``~/fgmod``.""" + +import os +import shutil +import subprocess +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable + +from . import config +from .assets import ( + DEFAULT_FRAMEGEN_BACKEND, + DEFAULT_FSR4_VARIANT, + FRAMEGEN_BACKENDS, + FSR4_DRIVER_OVERRIDE_FILENAME, + FSR4_VARIANTS, + FSR4_UPSCALER_FILENAME, + INSTALL_MANIFEST_FILENAME, + MANIFEST_SCHEMA_VERSION, + OPTIPATCHER_ASSET, + OPTISCALER_ARCHIVE_ASSET, + OPTISCALER_PRE10_ASSET, + AMDXC64_RDNA2_ASSET, + FSR4_INT8_ASSET, + FSR4_OFFICIAL_411_ASSET, + FSR4_VALVE_411_ASSET, + VERSION_FILENAME, +) +from .files import file_sha256, read_json, write_json + + +Logger = Callable[[str], None] + + +class BundleManager: + def __init__(self, home: Path, plugin_dir: Path, logger: Logger): + self.home = home + self.plugin_dir = plugin_dir + self.logger = logger + + @property + def fgmod_path(self) -> Path: + return self.home / "fgmod" + + def _log(self, message: str) -> None: + self.logger(f"[Framegen] {message}") + + def _create_renamed_copies(self, source_file: Path, renames_dir: Path) -> bool: + try: + if not source_file.exists(): + self._log(f"Source file {source_file} does not exist") + return False + renames_dir.mkdir(parents=True, exist_ok=True) + for filename in ("dxgi.dll", "winmm.dll", "dbghelp.dll", "version.dll", "wininet.dll", "winhttp.dll", "OptiScaler.asi"): + shutil.copy2(source_file, renames_dir / filename) + return True + except OSError as exc: + self._log(f"Failed to create renamed copies: {exc}") + return False + + def _copy_launcher_scripts(self, assets_dir: Path, extract_path: Path) -> bool: + try: + scripts = { + "fgmod.sh": "fgmod", + "fgmod-uninstaller.sh": "fgmod-uninstaller.sh", + "update-optiscaler-config.py": "update-optiscaler-config.py", + } + for source_name, destination_name in scripts.items(): + source = assets_dir / source_name + if not source.exists(): + continue + destination = extract_path / destination_name + shutil.copy2(source, destination) + destination.chmod(0o755) + return True + except OSError as exc: + self._log(f"Failed to copy launcher scripts: {exc}") + return False + + def _extract_archive(self, archive_path: Path, output_dir: Path) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + environment = os.environ.copy() + environment["LD_LIBRARY_PATH"] = "" + result = subprocess.run( + ["7z", "x", "-y", "-o" + str(output_dir), str(archive_path)], + capture_output=True, + text=True, + check=False, + env=environment, + ) + if result.returncode != 0: + raise RuntimeError(result.stderr or result.stdout or f"Failed to extract {archive_path.name}") + + def _verify(self, path: Path, expected_sha256: str, description: str) -> str: + actual = file_sha256(path) + if actual.lower() != expected_sha256.lower(): + raise RuntimeError(f"{description} hash mismatch: expected {expected_sha256}, got {actual}") + return actual + + def _manifest_path(self) -> Path: + return self.fgmod_path / INSTALL_MANIFEST_FILENAME + + def load_manifest(self) -> dict: + return read_json(self._manifest_path()) + + @staticmethod + def normalize_variant(value: str | None) -> str: + return value if value in FSR4_VARIANTS else DEFAULT_FSR4_VARIANT + + @staticmethod + def normalize_backend(value: str | None) -> str: + return value if value in FRAMEGEN_BACKENDS else DEFAULT_FRAMEGEN_BACKEND + + def selected_variant(self, requested: str | None = None) -> str: + if requested in FSR4_VARIANTS: + return requested + installed = self.load_manifest().get("selected_default_variant") + return self.normalize_variant(installed) + + def selected_backend(self, requested: str | None = None) -> str: + if requested in FRAMEGEN_BACKENDS: + return requested + manifest_backend = self.load_manifest().get("framegen_backend") + if manifest_backend in FRAMEGEN_BACKENDS: + return manifest_backend + return self.normalize_backend(config.detect_backend(self.fgmod_path / "OptiScaler.ini")) + + def variant_info(self, variant: str | None) -> dict: + return FSR4_VARIANTS[self.normalize_variant(variant)] + + def variant_dir(self, variant: str | None) -> Path: + return self.fgmod_path / self.variant_info(variant)["dir_name"] + + def variant_path(self, variant: str | None) -> Path: + return self.variant_dir(variant) / FSR4_UPSCALER_FILENAME + + def variant_extra_files(self, variant: str | None) -> list[dict]: + return list(self.variant_info(variant).get("extra_files") or []) + + def variant_extra_path(self, variant: str | None, filename: str) -> Path: + return self.variant_dir(variant) / filename + + def variant_injector_info(self, variant: str | None) -> dict | None: + injector = self.variant_info(variant).get("injector") + return injector if isinstance(injector, dict) else None + + def variant_injector_path(self, variant: str | None) -> Path | None: + injector = self.variant_injector_info(variant) + return self.variant_dir(variant) / injector["name"] if injector else None + + def variant_proxy_path(self, variant: str | None, dll_name: str) -> Path | None: + if not self.variant_injector_info(variant): + return None + return self.variant_dir(variant) / "renames" / dll_name + + def variant_config_overrides(self, variant: str | None) -> dict: + overrides = self.variant_info(variant).get("config_overrides") or {} + return dict(overrides) if isinstance(overrides, dict) else {} + + def fgmod_version(self) -> str | None: + manifest = self.load_manifest() + optiscaler = manifest.get("optiscaler") + if isinstance(optiscaler, dict) and optiscaler.get("version"): + return str(optiscaler["version"]) + try: + version = (self.fgmod_path / VERSION_FILENAME).read_text(encoding="utf-8").strip() + return version or None + except OSError: + return None + + def managed_support_candidates(self, filename: str) -> list[Path]: + candidates = [self.fgmod_path / filename] + if filename == FSR4_UPSCALER_FILENAME: + candidates = [self.fgmod_path / filename] + [self.variant_path(variant) for variant in FSR4_VARIANTS] + else: + for variant in FSR4_VARIANTS: + if any(extra.get("name") == filename for extra in self.variant_extra_files(variant)): + candidates.append(self.variant_extra_path(variant, filename)) + unique: list[Path] = [] + seen: set[str] = set() + for candidate in candidates: + if str(candidate) not in seen: + unique.append(candidate) + seen.add(str(candidate)) + return unique + + def is_managed_support_file(self, path: Path) -> bool: + return path.exists() and any( + candidate.is_file() and file_sha256(path) == file_sha256(candidate) + for candidate in self.managed_support_candidates(path.name) + ) + + def _sync_variant_root_extra_files(self, variant: str) -> None: + selected = {extra["name"]: extra for extra in self.variant_extra_files(variant)} + all_extra_names = { + extra["name"] + for item in FSR4_VARIANTS.values() + for extra in item.get("extra_files", []) + } + for filename in all_extra_names: + root_path = self.fgmod_path / filename + if filename not in selected: + if root_path.exists(): + root_path.unlink() + continue + source = self.variant_extra_path(variant, filename) + if not source.exists(): + raise FileNotFoundError(f"Prepared FSR4 variant extra file missing: {source}") + shutil.copy2(source, root_path) + + def activate_variant(self, variant: str | None) -> str: + variant_id = self.normalize_variant(variant) + source = self.variant_path(variant_id) + if not source.exists(): + raise FileNotFoundError(f"Prepared FSR4 variant missing: {source}") + shutil.copy2(source, self.fgmod_path / FSR4_UPSCALER_FILENAME) + self._sync_variant_root_extra_files(variant_id) + return variant_id + + def detect_variant(self, directory: Path, upscaler_sha256: str | None) -> str | None: + if not upscaler_sha256: + return None + normalized = upscaler_sha256.lower() + for variant_id, variant in FSR4_VARIANTS.items(): + if str(variant.get("sha256", "")).lower() != normalized: + continue + if all( + (directory / extra["name"]).exists() + and file_sha256(directory / extra["name"]).lower() == extra["sha256"].lower() + for extra in variant.get("extra_files", []) + ): + return variant_id + return None + + def extract_static_optiscaler( + self, + selected_default_variant: str = DEFAULT_FSR4_VARIANT, + framegen_backend: str = DEFAULT_FRAMEGEN_BACKEND, + ) -> dict: + """Prepare the shared bundle; replacement remains intentionally non-atomic for compatibility.""" + try: + selected_default_variant = self.normalize_variant(selected_default_variant) + framegen_backend = self.normalize_backend(framegen_backend) + bin_path = self.plugin_dir / "bin" + assets_dir = self.plugin_dir / "assets" + if not bin_path.exists(): + return {"status": "error", "message": f"Bin directory not found: {bin_path}"} + + asset_sources = { + "archive": (bin_path / OPTISCALER_ARCHIVE_ASSET["name"], OPTISCALER_ARCHIVE_ASSET), + "int8": (bin_path / FSR4_INT8_ASSET["name"], FSR4_INT8_ASSET), + "official": (bin_path / FSR4_OFFICIAL_411_ASSET["name"], FSR4_OFFICIAL_411_ASSET), + "valve": (bin_path / FSR4_VALVE_411_ASSET["name"], FSR4_VALVE_411_ASSET), + "amdxc64": (bin_path / AMDXC64_RDNA2_ASSET["name"], AMDXC64_RDNA2_ASSET), + "pre10": (bin_path / OPTISCALER_PRE10_ASSET["name"], OPTISCALER_PRE10_ASSET), + "optipatcher": (bin_path / OPTIPATCHER_ASSET["name"], OPTIPATCHER_ASSET), + } + for path, asset in asset_sources.values(): + if not path.exists(): + return {"status": "error", "message": f"Required bundled asset missing: {asset['name']}"} + self._verify(path, asset["sha256"], asset["name"]) + + if self.fgmod_path.exists(): + shutil.rmtree(self.fgmod_path) + self.fgmod_path.mkdir(parents=True, exist_ok=True) + self._extract_archive(asset_sources["archive"][0], self.fgmod_path) + + source = self.fgmod_path / "OptiScaler.dll" + if not self._create_renamed_copies(source, self.fgmod_path / "renames"): + return {"status": "error", "message": "Failed to prepare renamed OptiScaler proxies."} + if not self._copy_launcher_scripts(assets_dir, self.fgmod_path): + return {"status": "error", "message": "Failed to copy launcher scripts."} + + plugins_dir = self.fgmod_path / "plugins" + plugins_dir.mkdir(parents=True, exist_ok=True) + optipatcher_dst = plugins_dir / "OptiPatcher.asi" + shutil.copy2(asset_sources["optipatcher"][0], optipatcher_dst) + optipatcher_sha256 = self._verify(optipatcher_dst, OPTIPATCHER_ASSET["sha256"], "Prepared OptiPatcher plugin") + + ini_file = self.fgmod_path / "OptiScaler.ini" + config.configure(ini_file, framegen_backend, logger=self.logger) + + native_root = self.fgmod_path / FSR4_UPSCALER_FILENAME + native_sha256 = self._verify(native_root, FSR4_VARIANTS["rdna4-native"]["sha256"], "Archive-native FSR4 upscaler") + for variant_id in ("rdna4-native", "rdna34-official-411", "rdna2-valve-411-pre10"): + variant_dir = self.variant_dir(variant_id) + variant_dir.mkdir(parents=True, exist_ok=True) + prepared = variant_dir / FSR4_UPSCALER_FILENAME + shutil.copy2(native_root, prepared) + self._verify(prepared, FSR4_VARIANTS[variant_id]["sha256"], f"Prepared {variant_id} FSR4 upscaler") + + official_dir = self.variant_dir("rdna34-official-411") + shutil.copy2(asset_sources["official"][0], official_dir / FSR4_DRIVER_OVERRIDE_FILENAME) + self._verify(official_dir / FSR4_DRIVER_OVERRIDE_FILENAME, FSR4_OFFICIAL_411_ASSET["sha256"], "Prepared official driver override") + + valve_dir = self.variant_dir("rdna2-valve-411-pre10") + shutil.copy2(asset_sources["valve"][0], valve_dir / FSR4_DRIVER_OVERRIDE_FILENAME) + shutil.copy2(asset_sources["amdxc64"][0], valve_dir / "amdxc64.dll") + shutil.copy2(asset_sources["pre10"][0], valve_dir / "OptiScaler.dll") + self._verify(valve_dir / FSR4_DRIVER_OVERRIDE_FILENAME, FSR4_VALVE_411_ASSET["sha256"], "Prepared Valve driver override") + self._verify(valve_dir / "amdxc64.dll", AMDXC64_RDNA2_ASSET["sha256"], "Prepared RDNA2 driver override") + self._verify(valve_dir / "OptiScaler.dll", OPTISCALER_PRE10_ASSET["sha256"], "Prepared pre10 OptiScaler injector") + if not self._create_renamed_copies(valve_dir / "OptiScaler.dll", valve_dir / "renames"): + return {"status": "error", "message": "Failed to prepare renamed pre10 OptiScaler proxies."} + + int8_dir = self.variant_dir("rdna23-int8") + int8_dir.mkdir(parents=True, exist_ok=True) + shutil.copy2(asset_sources["int8"][0], int8_dir / FSR4_UPSCALER_FILENAME) + self._verify(int8_dir / FSR4_UPSCALER_FILENAME, FSR4_INT8_ASSET["sha256"], "Prepared RDNA2-3 FSR4 upscaler") + + selected_default_variant = self.activate_variant(selected_default_variant) + active_sha256 = file_sha256(self.fgmod_path / FSR4_UPSCALER_FILENAME) + (self.fgmod_path / VERSION_FILENAME).write_text(OPTISCALER_ARCHIVE_ASSET["version"], encoding="utf-8") + manifest = self._build_manifest( + selected_default_variant, + framegen_backend, + native_sha256, + optipatcher_sha256, + active_sha256, + ) + write_json(self._manifest_path(), manifest) + return { + "status": "success", + "message": f"Successfully extracted OptiScaler {OPTISCALER_ARCHIVE_ASSET['version']} to ~/fgmod", + "version": OPTISCALER_ARCHIVE_ASSET["version"], + "selected_default_variant": selected_default_variant, + "selected_default_variant_label": FSR4_VARIANTS[selected_default_variant]["label"], + "framegen_backend": framegen_backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], + } + except Exception as exc: + self._log(f"Extract failed: {exc}") + return {"status": "error", "message": f"Extract failed: {exc}"} + + def _build_manifest( + self, + selected_variant: str, + framegen_backend: str, + native_sha256: str, + optipatcher_sha256: str, + active_sha256: str, + ) -> dict: + variants = {} + for variant_id, variant in FSR4_VARIANTS.items(): + injector = variant.get("injector") + injector_manifest = None + if isinstance(injector, dict): + injector_manifest = { + key: injector[key] + for key in ("name", "sha256", "source_asset_name", "source_version") + } + injector_manifest["path"] = str(Path(variant["dir_name"]) / injector["name"]) + variants[variant_id] = { + "label": variant["label"], + "dir_name": variant["dir_name"], + "path": str(Path(variant["dir_name"]) / FSR4_UPSCALER_FILENAME), + "sha256": variant["sha256"], + "source_asset_name": variant["source_asset_name"], + "source_version": variant["source_version"], + "uses_archive_native": bool(variant["uses_archive_native"]), + "injector": injector_manifest, + "config_overrides": dict(variant.get("config_overrides") or {}), + "extra_files": [ + { + **{key: extra[key] for key in ("name", "sha256", "source_asset_name", "source_version")}, + "path": str(Path(variant["dir_name"]) / extra["name"]), + } + for extra in variant.get("extra_files", []) + ], + } + return { + "schema_version": MANIFEST_SCHEMA_VERSION, + "installed_at": datetime.now(timezone.utc).isoformat(), + "optiscaler": { + "asset_name": OPTISCALER_ARCHIVE_ASSET["name"], + "version": OPTISCALER_ARCHIVE_ASSET["version"], + "sha256": OPTISCALER_ARCHIVE_ASSET["sha256"], + "native_upscaler_sha256": native_sha256, + }, + "optipatcher": { + "asset_name": OPTIPATCHER_ASSET["name"], + "version": OPTIPATCHER_ASSET["version"], + "sha256": optipatcher_sha256, + "target_path": "plugins/OptiPatcher.asi", + }, + "fsr4_variants": variants, + "selected_default_variant": selected_variant, + "framegen_backend": framegen_backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], + "active_root_upscaler": { + "path": FSR4_UPSCALER_FILENAME, + "sha256": active_sha256, + "variant": selected_variant, + }, + } + + def run_install(self, selected_variant: str, framegen_backend: str) -> dict: + selected_variant = self.normalize_variant(selected_variant) + framegen_backend = self.normalize_backend(framegen_backend) + result = self.extract_static_optiscaler(selected_variant, framegen_backend) + if result["status"] != "success": + return {"status": "error", "message": f"OptiScaler extraction failed: {result.get('message', 'Unknown error')}"} + return { + "status": "success", + "output": f"Successfully installed OptiScaler {result.get('version', OPTISCALER_ARCHIVE_ASSET['version'])} with {result.get('selected_default_variant_label', FSR4_VARIANTS[selected_variant]['label'])}.", + "version": result.get("version", OPTISCALER_ARCHIVE_ASSET["version"]), + "selected_default_variant": result.get("selected_default_variant", selected_variant), + "selected_default_variant_label": result.get("selected_default_variant_label", FSR4_VARIANTS[selected_variant]["label"]), + "framegen_backend": result.get("framegen_backend", framegen_backend), + "framegen_backend_label": result.get("framegen_backend_label", FRAMEGEN_BACKENDS[framegen_backend]), + } + + def uninstall(self) -> dict: + try: + if not self.fgmod_path.exists(): + return {"status": "success", "output": "No fgmod directory found to remove"} + shutil.rmtree(self.fgmod_path) + return {"status": "success", "output": "Successfully removed fgmod directory"} + except OSError as exc: + self._log(f"Uninstall error: {exc}") + return {"status": "error", "message": f"Uninstall failed: {exc}", "output": str(exc)} + + def set_default_variant(self, selected_variant: str) -> dict: + try: + if not self.fgmod_path.exists(): + return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} + manifest = self.load_manifest() + if not manifest: + return {"status": "error", "message": "Install manifest missing. Reinstall OptiScaler."} + selected_variant = self.activate_variant(selected_variant) + active_sha256 = file_sha256(self.fgmod_path / FSR4_UPSCALER_FILENAME) + manifest["selected_default_variant"] = selected_variant + manifest["active_root_upscaler"] = { + "path": FSR4_UPSCALER_FILENAME, + "sha256": active_sha256, + "variant": selected_variant, + } + manifest["updated_at"] = datetime.now(timezone.utc).isoformat() + write_json(self._manifest_path(), manifest) + return { + "status": "success", + "output": f"Default FSR4 runtime switched to {FSR4_VARIANTS[selected_variant]['label']}.", + "version": self.fgmod_version(), + "selected_default_variant": selected_variant, + "selected_default_variant_label": FSR4_VARIANTS[selected_variant]["label"], + } + except Exception as exc: + self._log(f"Failed to switch default FSR4 runtime: {exc}") + return {"status": "error", "message": f"Failed to switch default FSR4 runtime: {exc}"} + + def set_framegen_backend(self, framegen_backend: str) -> dict: + try: + if not self.fgmod_path.exists(): + return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} + backend = self.normalize_backend(framegen_backend) + manifest = self.load_manifest() + if not manifest: + return {"status": "error", "message": "Install manifest missing. Reinstall OptiScaler."} + ini_file = self.fgmod_path / "OptiScaler.ini" + if not config.configure(ini_file, backend, logger=self.logger): + return {"status": "error", "message": "Could not update the shared OptiScaler.ini."} + manifest["framegen_backend"] = backend + manifest["framegen_backend_label"] = FRAMEGEN_BACKENDS[backend] + manifest["updated_at"] = datetime.now(timezone.utc).isoformat() + write_json(self._manifest_path(), manifest) + return { + "status": "success", + "framegen_backend": backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[backend], + "output": f"Frame generation backend set to {FRAMEGEN_BACKENDS[backend]}.", + } + except Exception as exc: + self._log(f"Failed to set frame generation backend: {exc}") + return {"status": "error", "message": f"Failed to set frame generation backend: {exc}"} + + def check(self) -> dict: + required_files = [ + "OptiScaler.dll", + "OptiScaler.ini", + "dlssg_to_fsr3_amd_is_better.dll", + "fakenvapi.dll", + "fakenvapi.ini", + "amd_fidelityfx_dx12.dll", + "amd_fidelityfx_framegeneration_dx12.dll", + FSR4_UPSCALER_FILENAME, + "amd_fidelityfx_vk.dll", + "libxess.dll", + "libxess_dx11.dll", + "libxess_fg.dll", + "libxell.dll", + "fgmod", + "fgmod-uninstaller.sh", + "update-optiscaler-config.py", + INSTALL_MANIFEST_FILENAME, + ] + if not self.fgmod_path.exists() or any(not (self.fgmod_path / filename).exists() for filename in required_files): + return {"exists": False} + plugins_dir = self.fgmod_path / "plugins" + if not (plugins_dir / "OptiPatcher.asi").exists(): + return {"exists": False} + for variant_id, variant in FSR4_VARIANTS.items(): + variant_dir = self.variant_dir(variant_id) + if not (variant_dir / FSR4_UPSCALER_FILENAME).exists(): + return {"exists": False} + injector = variant.get("injector") + if isinstance(injector, dict) and not (variant_dir / injector.get("name", "OptiScaler.dll")).exists(): + return {"exists": False} + for extra in variant.get("extra_files", []): + if not (variant_dir / extra["name"]).exists(): + return {"exists": False} + manifest = self.load_manifest() + variant = self.selected_variant() + backend = self.selected_backend() + return { + "exists": True, + "version": self.fgmod_version(), + "selected_fsr4_variant": variant, + "selected_fsr4_variant_label": FSR4_VARIANTS[variant]["label"], + "framegen_backend": backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[backend], + "install_manifest_present": bool(manifest), + } diff --git a/py_modules/services/config.py b/py_modules/services/config.py new file mode 100644 index 0000000..87476cc --- /dev/null +++ b/py_modules/services/config.py @@ -0,0 +1,175 @@ +"""OptiScaler configuration migration and default handling.""" + +import re +from pathlib import Path +from typing import Callable + + +Logger = Callable[[str], None] + + +def _split_override_key(raw_key: str) -> tuple[str | None, str]: + key = str(raw_key).strip() + if "." in key: + section, section_key = key.split(".", 1) + section = section.strip() + section_key = section_key.strip() + if section and section_key: + return section, section_key + return None, key + + +def apply_overrides(ini_file: Path, overrides: dict, logger: Logger | None = None) -> bool: + """Upsert INI keys without disturbing unrelated game configuration.""" + if not overrides: + return True + try: + content = ini_file.read_text(encoding="utf-8", errors="replace") + newline = "\r\n" if "\r\n" in content else "\n" + lines = content.splitlines(keepends=True) + section_pattern = re.compile(r"^\s*\[(?P
[^\]]+)\]\s*$") + + def ensure_trailing_newline() -> None: + if lines and not lines[-1].endswith(("\n", "\r")): + lines[-1] += newline + + def upsert(section: str | None, key: str, value: str) -> None: + replacement = f"{key}={value}" + key_pattern = re.compile(rf"^(\s*{re.escape(key)}\s*)=.*$") + + if section is None: + for index, line in enumerate(lines): + if key_pattern.match(line): + ending = "\r\n" if line.endswith("\r\n") else ("\n" if line.endswith("\n") else newline) + lines[index] = f"{replacement}{ending}" + return + ensure_trailing_newline() + lines.append(f"{replacement}{newline}") + return + + in_section = False + insert_at = None + for index, line in enumerate(lines): + match = section_pattern.match(line.strip()) + if match: + if in_section: + insert_at = index + break + if match.group("section") == section: + in_section = True + continue + if in_section and key_pattern.match(line): + ending = "\r\n" if line.endswith("\r\n") else ("\n" if line.endswith("\n") else newline) + lines[index] = f"{replacement}{ending}" + return + + if in_section: + if insert_at is None: + ensure_trailing_newline() + insert_at = len(lines) + lines.insert(insert_at, f"{replacement}{newline}") + return + + ensure_trailing_newline() + if lines and lines[-1].strip(): + lines.append(newline) + lines.extend((f"[{section}]{newline}", f"{replacement}{newline}")) + + for raw_key, raw_value in overrides.items(): + section, key = _split_override_key(str(raw_key)) + if key: + upsert(section, key, str(raw_value).strip()) + ini_file.write_text("".join(lines), encoding="utf-8") + return True + except Exception as exc: + if logger: + logger(f"Failed to apply OptiScaler.ini overrides to {ini_file}: {exc}") + return False + + +def migrate(ini_file: Path, logger: Logger | None = None) -> bool: + """Migrate the pre-0.9 ``FGType`` setting to split input/output keys.""" + try: + if not ini_file.exists(): + return False + content = ini_file.read_text(encoding="utf-8", errors="replace") + match = re.search(r"^FGType\s*=\s*(\S+)", content, re.MULTILINE) + if not match: + return True + + value = match.group(1) + if re.search(r"^FGInput\s*=", content, re.MULTILINE): + updated = re.sub(r"^FGType\s*=\s*\S+\n?", "", content, flags=re.MULTILINE) + message = f"Removed stale FGType from {ini_file}" + else: + updated = re.sub( + r"^FGType\s*=\s*\S+", + f"FGInput={value}\nFGOutput={value}", + content, + flags=re.MULTILINE, + ) + message = f"Migrated FGType={value} in {ini_file}" + if updated != content: + ini_file.write_text(updated, encoding="utf-8") + if logger: + logger(message) + return True + except Exception as exc: + if logger: + logger(f"Failed to migrate OptiScaler.ini: {exc}") + return False + + +def disable_hq_font_auto(ini_file: Path, logger: Logger | None = None) -> bool: + """Avoid the HQ font auto mode, which is unreliable under Proton.""" + try: + if not ini_file.exists(): + return False + content = ini_file.read_text(encoding="utf-8", errors="replace") + updated = re.sub(r"UseHQFont\s*=\s*auto", "UseHQFont=false", content) + if updated != content: + ini_file.write_text(updated, encoding="utf-8") + return True + except Exception as exc: + if logger: + logger(f"Failed to update HQ font setting in {ini_file}: {exc}") + return False + + +def configure( + ini_file: Path, + framegen_backend: str, + variant_overrides: dict | None = None, + logger: Logger | None = None, +) -> bool: + """Apply safe defaults and the selected frame-generation backend.""" + if not ini_file.exists(): + return False + backend = framegen_backend if framegen_backend in {"auto", "nukems"} else "auto" + overrides = { + "FGInput": backend, + "FGOutput": backend, + "Fsr4Update": "true", + "LoadAsiPlugins": "true", + } + if variant_overrides: + overrides.update(variant_overrides) + migrated = migrate(ini_file, logger) + hq_updated = disable_hq_font_auto(ini_file, logger) + configured = apply_overrides(ini_file, overrides, logger) + return migrated and hq_updated and configured + + +def detect_backend(ini_file: Path) -> str: + """Return the configured backend when both FG keys agree.""" + try: + content = ini_file.read_text(encoding="utf-8", errors="replace") + except OSError: + return "auto" + values = [ + match.group(1).strip().lower() + for match in re.finditer(r"^FG(?:Input|Output)\s*=\s*(\S+)", content, re.MULTILINE) + ] + if values and all(value == "nukems" for value in values): + return "nukems" + return "auto" diff --git a/py_modules/services/files.py b/py_modules/services/files.py new file mode 100644 index 0000000..d8571ba --- /dev/null +++ b/py_modules/services/files.py @@ -0,0 +1,36 @@ +"""Small filesystem helpers shared by service modules.""" + +import filecmp +import hashlib +import json +from pathlib import Path + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def files_match(file_a: Path, file_b: Path) -> bool: + try: + return file_a.is_file() and file_b.is_file() and filecmp.cmp(file_a, file_b, shallow=False) + except OSError: + return False + + +def read_json(path: Path) -> dict: + try: + with path.open("r", encoding="utf-8") as stream: + payload = json.load(stream) + return payload if isinstance(payload, dict) else {} + except (OSError, ValueError, TypeError): + return {} + + +def write_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as stream: + json.dump(payload, stream, indent=2) diff --git a/py_modules/services/patcher.py b/py_modules/services/patcher.py new file mode 100644 index 0000000..59d08f1 --- /dev/null +++ b/py_modules/services/patcher.py @@ -0,0 +1,685 @@ +"""Patch ownership, safe cleanup, and Steam-game operations.""" + +import os +import shutil +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable + +from . import config +from .assets import ( + BACKUP_DIRECTORY_NAME, + FRAMEGEN_BACKENDS, + FSR4_UPSCALER_FILENAME, + INJECTOR_FILENAMES, + MARKER_FILENAME, + MANIFEST_SCHEMA_VERSION, + RESTORABLE_BACKUP_FILES, + SUPPORT_FILES, + VALID_DLL_NAMES, + VARIANT_EXTRA_FILENAMES, +) +from .bundle import BundleManager +from .files import file_sha256, files_match, read_json, write_json +from .steam import SteamLibrary + + +Logger = Callable[[str], None] + + +class PatchManager: + def __init__(self, home: Path, bundle: BundleManager, steam: SteamLibrary, logger: Logger): + self.home = home + self.bundle = bundle + self.steam = steam + self.logger = logger + + def _log(self, message: str) -> None: + self.logger(f"[Framegen] {message}") + + @staticmethod + def _relative_path(root: Path, path: Path) -> str: + return path.resolve().relative_to(root.resolve()).as_posix() + + def _safe_path(self, root: Path, value: str | Path) -> Path: + candidate = Path(value) + if candidate.is_absolute(): + resolved = candidate.resolve() + else: + resolved = (root / candidate).resolve() + try: + resolved.relative_to(root.resolve()) + except ValueError as exc: + raise ValueError(f"Path escapes patch directory: {value}") from exc + return resolved + + def _read_marker(self, marker_path: Path) -> dict: + return read_json(marker_path) + + def _find_marker(self, install_root: Path, appid: str | None = None) -> Path | None: + if not install_root.exists(): + return None + try: + markers = sorted(path for path in install_root.rglob(MARKER_FILENAME) if path.is_file()) + except OSError: + return None + for marker in markers: + metadata = self._read_marker(marker) + if appid is None or str(metadata.get("appid")) == str(appid): + return marker + return None + + def _marker_target(self, marker_path: Path, metadata: dict) -> Path: + target_value = metadata.get("target_relative") or metadata.get("target_dir") or str(marker_path.parent) + target = self._safe_path(marker_path.parent, target_value) + if not target.is_dir(): + raise FileNotFoundError(f"Patch target directory does not exist: {target}") + return target + + def _bundle_source_for_proxy(self, variant: str, dll_name: str) -> Path: + candidates = [ + self.bundle.variant_proxy_path(variant, dll_name), + self.bundle.variant_injector_path(variant), + self.bundle.fgmod_path / "renames" / dll_name, + self.bundle.fgmod_path / "OptiScaler.dll", + ] + for candidate in candidates: + if candidate and candidate.exists(): + return candidate + raise FileNotFoundError(f"No injector source found for {dll_name}") + + def _copy_plan(self, target_dir: Path, variant: str) -> list[tuple[str, Path]]: + plan: list[tuple[str, Path]] = [] + # The proxy is the only file selected by the user; all other files are bundle-owned. + plan.append(("__proxy__", self._bundle_source_for_proxy(variant, "dxgi.dll"))) + for filename in SUPPORT_FILES: + source = self.bundle.fgmod_path / filename + if source.is_file(): + plan.append((filename, source)) + upscaler = self.bundle.variant_path(variant) + if upscaler.is_file(): + plan.append((FSR4_UPSCALER_FILENAME, upscaler)) + for extra in self.bundle.variant_extra_files(variant): + source = self.bundle.variant_extra_path(variant, extra["name"]) + if source.is_file(): + plan.append((extra["name"], source)) + + for source_root, relative_root in ( + (self.bundle.fgmod_path / "plugins", Path("plugins")), + (self.bundle.fgmod_path / "D3D12_Optiscaler", Path("D3D12_Optiscaler")), + ): + if not source_root.is_dir(): + continue + for source in sorted(path for path in source_root.rglob("*") if path.is_file()): + plan.append(((relative_root / source.relative_to(source_root)).as_posix(), source)) + return plan + + def _backup_destination(self, target_dir: Path, relative: str, source: Path, force: bool = False) -> str | None: + destination = self._safe_path(target_dir, relative) + if not destination.exists(): + return None + if destination.is_dir(): + raise IsADirectoryError(f"Patch destination is a directory: {destination}") + if not force and files_match(destination, source): + return None + + if relative in RESTORABLE_BACKUP_FILES: + legacy_backup = destination.with_name(destination.name + ".b") + if not legacy_backup.exists(): + shutil.copy2(destination, legacy_backup) + elif not files_match(legacy_backup, destination): + raise RuntimeError(f"Refusing to overwrite an existing backup: {legacy_backup}") + return self._relative_path(target_dir, legacy_backup) + + backup = target_dir / BACKUP_DIRECTORY_NAME / relative + backup.parent.mkdir(parents=True, exist_ok=True) + if backup.exists(): + if not files_match(backup, destination): + raise RuntimeError(f"Refusing to overwrite an existing backup: {backup}") + else: + shutil.copy2(destination, backup) + return self._relative_path(target_dir, backup) + + def _cleanup_empty_directories(self, target_dir: Path, managed_files: list[str]) -> None: + directories = sorted( + {self._safe_path(target_dir, path).parent for path in managed_files}, + key=lambda path: len(path.parts), + reverse=True, + ) + for directory in directories: + if directory == target_dir: + continue + try: + directory.rmdir() + except OSError: + pass + backup_dir = target_dir / BACKUP_DIRECTORY_NAME + try: + backup_dir.rmdir() + except OSError: + pass + + def _rollback( + self, + target_dir: Path, + managed: list[dict], + backups: list[dict], + ) -> None: + for entry in reversed(managed): + try: + path = self._safe_path(target_dir, entry["path"]) + backup = next((item["backup"] for item in backups if item["path"] == entry["path"]), None) + if backup: + backup_path = self._safe_path(target_dir, backup) + if backup_path.exists(): + if path.exists(): + path.unlink() + shutil.move(backup_path, path) + elif path.exists() and entry.get("sha256") == file_sha256(path): + path.unlink() + except (OSError, ValueError): + continue + self._cleanup_empty_directories(target_dir, [entry["path"] for entry in managed]) + + def _patch_directory( + self, + target_dir: Path, + dll_name: str, + variant: str, + framegen_backend: str, + *, + appid: str, + game_name: str, + original_launch_options: str, + previous_variant: str | None = None, + ) -> dict: + if dll_name not in VALID_DLL_NAMES: + return {"status": "error", "message": f"Invalid proxy DLL name: {dll_name}"} + if not self.bundle.fgmod_path.exists() or not (self.bundle.fgmod_path / "OptiScaler.dll").exists(): + return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} + + variant = self.bundle.selected_variant(variant) + framegen_backend = self.bundle.selected_backend(framegen_backend) + previous_marker = target_dir / MARKER_FILENAME + if previous_marker.exists(): + return {"status": "error", "message": f"Patch marker already exists in {target_dir}; clean it up before patching."} + variant_overrides = self.bundle.variant_config_overrides(variant) + if previous_variant == "rdna2-valve-411-pre10" and variant != previous_variant: + variant_overrides.setdefault("FSR.Fsr4ForceModel", "auto") + variant_overrides.setdefault("Plugins.LoadCustomAmdxc64OnRdna2", "false") + + managed: list[dict] = [] + backups: list[dict] = [] + try: + plan = self._copy_plan(target_dir, variant) + proxy_source = self._bundle_source_for_proxy(variant, dll_name) + plan[0] = (dll_name, proxy_source) + config_path = target_dir / "OptiScaler.ini" + if config_path.exists(): + plan.append(("OptiScaler.ini", config_path)) + elif (self.bundle.fgmod_path / "OptiScaler.ini").exists(): + plan.append(("OptiScaler.ini", self.bundle.fgmod_path / "OptiScaler.ini")) + + for relative, source in plan: + destination = self._safe_path(target_dir, relative) + if not source.is_file(): + continue + backup = self._backup_destination( + target_dir, + relative, + source, + # Preserve every pre-existing destination. Identical bytes do + # not prove that a file belongs to this plugin. + force=destination.exists(), + ) + if backup: + backups.append({ + "path": relative, + "backup": backup, + "sha256": file_sha256(self._safe_path(target_dir, backup)), + }) + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.resolve() != source.resolve(): + shutil.copy2(source, destination) + managed.append({ + "path": relative, + "sha256": file_sha256(destination), + "source_sha256": file_sha256(source), + }) + + if config_path.exists() and not config.configure( + config_path, + framegen_backend, + variant_overrides=variant_overrides, + logger=self.logger, + ): + raise RuntimeError("Could not update OptiScaler.ini") + config_entry = next((entry for entry in managed if entry["path"] == "OptiScaler.ini"), None) + if config_entry: + config_entry["sha256"] = file_sha256(config_path) + + payload = { + "schema_version": MANIFEST_SCHEMA_VERSION, + "appid": str(appid), + "game_name": game_name, + "dll_name": dll_name, + "target_dir": str(target_dir), + "target_relative": ".", + "original_launch_options": original_launch_options, + "backed_up_files": [item["path"] for item in backups], + "backups": backups, + "managed_files": managed, + "managed_directories": sorted({str(Path(item["path"]).parent) for item in managed if Path(item["path"]).parent != Path(".")}), + "optiscaler_version": self.bundle.fgmod_version(), + "fsr4_variant": variant, + "fsr4_variant_label": self.bundle.variant_info(variant)["label"], + "fsr4_upscaler_sha256": file_sha256(target_dir / FSR4_UPSCALER_FILENAME) if (target_dir / FSR4_UPSCALER_FILENAME).exists() else None, + "framegen_backend": framegen_backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[framegen_backend], + "patched_at": datetime.now(timezone.utc).isoformat(), + } + # The marker is written last, so an interrupted copy is never advertised as a complete patch. + write_json(previous_marker, payload) + return { + "status": "success", + "message": f"OptiScaler files copied to {target_dir} using {self.bundle.variant_info(variant)['label']}", + "fsr4_variant": variant, + "fsr4_variant_label": self.bundle.variant_info(variant)["label"], + "fsr4_upscaler_sha256": payload["fsr4_upscaler_sha256"], + "optiscaler_version": payload["optiscaler_version"], + "framegen_backend": framegen_backend, + } + except (OSError, RuntimeError, ValueError) as exc: + self._rollback(target_dir, managed, backups) + self._log(f"Patch failed for {target_dir}: {exc}") + return {"status": "error", "message": f"Patch failed: {exc}"} + + def _managed_entries(self, target_dir: Path, metadata: dict) -> tuple[list[tuple[dict, Path]], list[str]]: + entries: list[tuple[dict, Path]] = [] + errors: list[str] = [] + for entry in metadata.get("managed_files", []): + if not isinstance(entry, dict) or not entry.get("path"): + errors.append("Patch marker contains an invalid managed file entry.") + continue + try: + path = self._safe_path(target_dir, entry["path"]) + except ValueError as exc: + errors.append(str(exc)) + continue + expected = str(entry.get("sha256") or "") + if not expected: + errors.append(f"Patch marker has no hash for {entry['path']}.") + elif path.exists() and (not path.is_file() or file_sha256(path).lower() != expected.lower()): + errors.append(f"Refusing to remove modified managed file: {path}") + entries.append((entry, path)) + return entries, errors + + def _restore_backups(self, target_dir: Path, metadata: dict) -> list[str]: + restored: list[str] = [] + for item in metadata.get("backups", []): + if not isinstance(item, dict) or not item.get("path") or not item.get("backup"): + raise ValueError("Patch marker contains an invalid backup entry.") + path = self._safe_path(target_dir, item["path"]) + backup = self._safe_path(target_dir, item["backup"]) + if not backup.exists(): + continue + if not backup.is_file(): + raise ValueError(f"Patch backup is not a file: {backup}") + expected = str(item.get("sha256") or "") + if expected and file_sha256(backup).lower() != expected.lower(): + raise RuntimeError(f"Refusing to restore a modified backup: {backup}") + if path.exists(): + raise RuntimeError(f"Refusing to overwrite a file while restoring backup: {path}") + path.parent.mkdir(parents=True, exist_ok=True) + shutil.move(backup, path) + restored.append(str(item["path"])) + + # Older releases used ``name.b`` backups and listed only their base names. + for raw_name in metadata.get("backed_up_files", []): + name = Path(str(raw_name)).name + if name not in RESTORABLE_BACKUP_FILES: + continue + original = self._safe_path(target_dir, name) + backup = self._safe_path(target_dir, name + ".b") + if backup.exists() and not original.exists(): + shutil.move(backup, original) + restored.append(name) + return restored + + def _validate_backups(self, target_dir: Path, metadata: dict) -> None: + for item in metadata.get("backups", []): + if not isinstance(item, dict) or not item.get("path") or not item.get("backup"): + raise ValueError("Patch marker contains an invalid backup entry.") + self._safe_path(target_dir, item["path"]) + backup = self._safe_path(target_dir, item["backup"]) + if not backup.exists(): + continue + if not backup.is_file(): + raise ValueError(f"Patch backup is not a file: {backup}") + expected = str(item.get("sha256") or "") + if expected and file_sha256(backup).lower() != expected.lower(): + raise RuntimeError(f"Refusing to restore a modified backup: {backup}") + + def _legacy_owned_paths(self, target_dir: Path, metadata: dict) -> list[Path]: + paths: list[Path] = [] + for entry in metadata.get("managed_files", []): + if isinstance(entry, dict) and entry.get("path"): + try: + paths.append(self._safe_path(target_dir, entry["path"])) + except ValueError: + continue + + variant = self.bundle.normalize_variant(metadata.get("fsr4_variant")) + proxy_sources = [ + self.bundle.fgmod_path / "OptiScaler.dll", + self.bundle.fgmod_path / "renames", + self.bundle.variant_dir(variant) / "renames", + ] + for filename in INJECTOR_FILENAMES: + target = target_dir / filename + if not target.is_file(): + continue + matches_bundle = False + for source in proxy_sources: + candidate = source if source.is_file() and source.name == filename else source / filename + if candidate.is_file() and files_match(target, candidate): + matches_bundle = True + break + if matches_bundle: + paths.append(target) + for filename in SUPPORT_FILES + VARIANT_EXTRA_FILENAMES + [FSR4_UPSCALER_FILENAME]: + target = target_dir / filename + if target.is_file() and self.bundle.is_managed_support_file(target): + paths.append(target) + for relative in (Path("plugins/OptiPatcher.asi"),): + target = target_dir / relative + source = self.bundle.fgmod_path / relative + if target.is_file() and source.is_file() and files_match(target, source): + paths.append(target) + d3d12_source = self.bundle.fgmod_path / "D3D12_Optiscaler" + d3d12_target = target_dir / "D3D12_Optiscaler" + if d3d12_source.is_dir() and d3d12_target.is_dir(): + for source in d3d12_source.rglob("*"): + if source.is_file(): + candidate = d3d12_target / source.relative_to(d3d12_source) + if candidate.is_file() and files_match(candidate, source): + paths.append(candidate) + unique: list[Path] = [] + seen: set[str] = set() + for path in paths: + if str(path) not in seen: + unique.append(path) + seen.add(str(path)) + return unique + + def _unpatch_target(self, target_dir: Path, marker_path: Path | None) -> dict: + metadata = self._read_marker(marker_path) if marker_path else { + "backed_up_files": [ + name for name in RESTORABLE_BACKUP_FILES + if (target_dir / f"{name}.b").is_file() + ], + } + try: + if marker_path and marker_path.exists() and not metadata: + return {"status": "error", "message": f"Patch marker is unreadable: {marker_path}"} + schema_version = int(metadata.get("schema_version") or 1) if metadata else 1 + if schema_version >= MANIFEST_SCHEMA_VERSION: + entries, errors = self._managed_entries(target_dir, metadata) + if errors: + return {"status": "error", "message": " ".join(errors)} + self._validate_backups(target_dir, metadata) + managed_paths = [path for _, path in entries] + for _, path in entries: + if path.exists(): + path.unlink() + restored = self._restore_backups(target_dir, metadata) + self._cleanup_empty_directories(target_dir, [str(path.relative_to(target_dir)) for path in managed_paths]) + else: + owned = self._legacy_owned_paths(target_dir, metadata) + entries, errors = self._managed_entries(target_dir, metadata) + if errors: + return {"status": "error", "message": " ".join(errors)} + owned.extend(path for _, path in entries if path not in owned) + for path in owned: + if path.is_file(): + path.unlink() + restored = self._restore_backups(target_dir, metadata) + self._cleanup_empty_directories(target_dir, [str(path.relative_to(target_dir)) for path in owned]) + if marker_path and marker_path.exists(): + marker_path.unlink() + return { + "status": "success", + "message": f"OptiScaler files removed from {target_dir}", + "restored_files": restored, + } + except (OSError, RuntimeError, ValueError) as exc: + self._log(f"Unpatch failed for {target_dir}: {exc}") + return {"status": "error", "message": f"Unpatch failed: {exc}"} + + def resolve_target_directory(self, directory: str) -> Path: + target = Path(directory).expanduser() + if not target.exists(): + raise FileNotFoundError(f"Target directory does not exist: {directory}") + if not target.is_dir(): + raise NotADirectoryError(f"Target path is not a directory: {directory}") + if not os.access(target, os.W_OK | os.X_OK): + raise PermissionError(f"Insufficient permissions for {directory}") + return target.resolve() + + def manual_patch_directory( + self, + directory: str, + dll_name: str, + fsr4_variant: str, + framegen_backend: str | None = None, + ) -> dict: + try: + target = self.resolve_target_directory(directory) + except (FileNotFoundError, NotADirectoryError, PermissionError) as exc: + return {"status": "error", "message": str(exc)} + marker = target / MARKER_FILENAME + previous_metadata = self._read_marker(marker) if marker.exists() else {} + previous_variant = previous_metadata.get("fsr4_variant") + if marker.exists(): + if str(previous_metadata.get("appid")) not in ("", "manual", "None"): + return {"status": "error", "message": "This directory is managed by an AppID patch; unpatch it there first."} + cleanup = self._unpatch_target(target, marker) + if cleanup["status"] != "success": + return cleanup + return self._patch_directory( + target, + dll_name, + self.bundle.selected_variant(fsr4_variant), + self.bundle.selected_backend(framegen_backend), + appid="manual", + game_name="Manual patch", + original_launch_options="", + previous_variant=previous_variant, + ) + + def manual_unpatch_directory(self, directory: str) -> dict: + try: + target = self.resolve_target_directory(directory) + except (FileNotFoundError, NotADirectoryError, PermissionError) as exc: + return {"status": "error", "message": str(exc)} + marker = target / MARKER_FILENAME + return self._unpatch_target(target, marker if marker.exists() else None) + + def list_installed_games(self) -> dict: + try: + return { + "status": "success", + "games": [ + { + "appid": str(game["appid"]), + "name": game["name"], + "install_found": Path(game["install_path"]).exists(), + } + for game in self.steam.find_installed_games() + ], + } + except Exception as exc: + self._log(str(exc)) + return {"status": "error", "message": str(exc)} + + def get_game_status(self, appid: str) -> dict: + try: + game_info = self.steam.game_record(str(appid)) + if not game_info: + return { + "status": "success", "appid": str(appid), "install_found": False, "patched": False, + "dll_name": None, "target_dir": None, "fsr4_variant": None, "fsr4_variant_label": None, + "framegen_backend": None, "message": "Game not found in Steam library.", + } + install_root = Path(game_info["install_path"]) + base = { + "status": "success", "appid": str(appid), "name": game_info["name"], + "install_found": install_root.exists(), "patched": False, "dll_name": None, + "target_dir": None, "fsr4_variant": None, "fsr4_variant_label": None, + "framegen_backend": None, + } + if not install_root.exists(): + return {**base, "message": "Game install directory not found."} + marker = self._find_marker(install_root, str(appid)) + if not marker: + return {**base, "message": "Not patched."} + metadata = self._read_marker(marker) + target_dir = self._marker_target(marker, metadata) + dll_name = str(metadata.get("dll_name") or "dxgi.dll") + upscaler = target_dir / FSR4_UPSCALER_FILENAME + upscaler_sha256 = file_sha256(upscaler) if upscaler.is_file() else None + variant = self.bundle.detect_variant(target_dir, upscaler_sha256) or self.bundle.normalize_variant(metadata.get("fsr4_variant")) + backend = self.bundle.normalize_backend(metadata.get("framegen_backend") or self.bundle.selected_backend()) + dll_present = (target_dir / dll_name).is_file() + return { + **base, + "patched": dll_present, + "dll_name": dll_name, + "target_dir": str(target_dir), + "patched_at": metadata.get("patched_at"), + "optiscaler_version": metadata.get("optiscaler_version"), + "fsr4_variant": variant, + "fsr4_variant_label": self.bundle.variant_info(variant)["label"], + "fsr4_upscaler_sha256": upscaler_sha256, + "framegen_backend": backend, + "framegen_backend_label": FRAMEGEN_BACKENDS[backend], + "message": f"Patched using {dll_name} with {self.bundle.variant_info(variant)['label']}." if dll_present else f"Marker found but {dll_name} is missing. Reinstall recommended.", + } + except Exception as exc: + self._log(f"get_game_status failed for {appid}: {exc}") + return {"status": "error", "message": str(exc)} + + @staticmethod + def build_managed_launch_options(dll_name: str) -> str: + if dll_name == "OptiScaler.asi": + return "SteamDeck=0 %command%" + base = dll_name.replace(".dll", "") + return f"WINEDLLOVERRIDES={base}=n,b SteamDeck=0 %command%" + + @staticmethod + def is_managed_launch_options(options: str) -> bool: + normalized = " ".join((options or "").replace('"', "").split()) + if not normalized: + return False + return any( + dll_name != "OptiScaler.asi" and f"WINEDLLOVERRIDES={dll_name.replace('.dll', '')}=n,b" in normalized + for dll_name in VALID_DLL_NAMES + ) or "fgmod/fgmod" in normalized + + def patch_game( + self, + appid: str, + dll_name: str, + current_launch_options: str, + fsr4_variant: str, + framegen_backend: str | None = None, + ) -> dict: + try: + if dll_name not in VALID_DLL_NAMES: + return {"status": "error", "message": f"Invalid proxy DLL name: {dll_name}"} + game_info = self.steam.game_record(str(appid)) + if not game_info: + return {"status": "error", "message": "Game not found in Steam library."} + install_root = Path(game_info["install_path"]) + if not install_root.exists(): + return {"status": "error", "message": "Game install directory does not exist."} + if self.steam.is_game_running(game_info): + return {"status": "error", "message": "Close the game before patching."} + if not self.bundle.fgmod_path.exists(): + return {"status": "error", "message": "OptiScaler bundle not installed. Run Install first."} + + existing_marker = self._find_marker(install_root) + existing_metadata = self._read_marker(existing_marker) if existing_marker else {} + if existing_marker and str(existing_metadata.get("appid")) not in (str(appid), "manual", "None"): + return {"status": "error", "message": "Another Framegen patch marker was found in this game install."} + original_options = current_launch_options or "" + stored_options = str(existing_metadata.get("original_launch_options") or "") + if stored_options and not self.is_managed_launch_options(stored_options): + original_options = stored_options + if self.is_managed_launch_options(original_options): + original_options = "" + + if existing_marker: + old_target = self._marker_target(existing_marker, existing_metadata) + cleanup = self._unpatch_target(old_target, existing_marker) + if cleanup["status"] != "success": + return cleanup + target_dir, _ = self.steam.guess_patch_target(game_info) + result = self._patch_directory( + target_dir, + dll_name, + self.bundle.selected_variant(fsr4_variant), + self.bundle.selected_backend(framegen_backend), + appid=str(appid), + game_name=game_info["name"], + original_launch_options=original_options, + previous_variant=existing_metadata.get("fsr4_variant"), + ) + if result["status"] != "success": + return result + backend = result.get("framegen_backend") or self.bundle.selected_backend(framegen_backend) + return { + **result, + "appid": str(appid), + "name": game_info["name"], + "dll_name": dll_name, + "target_dir": str(target_dir), + "launch_options": self.build_managed_launch_options(dll_name), + "original_launch_options": original_options, + "framegen_backend": backend, + "message": f"Patched {game_info['name']} using {dll_name} with {result.get('fsr4_variant_label') or self.bundle.variant_info(fsr4_variant)['label']}.", + } + except Exception as exc: + self._log(f"patch_game failed for {appid}: {exc}") + return {"status": "error", "message": str(exc)} + + def unpatch_game(self, appid: str) -> dict: + try: + game_info = self.steam.game_record(str(appid)) + if not game_info: + return {"status": "error", "message": "Game not found in Steam library."} + install_root = Path(game_info["install_path"]) + if not install_root.exists(): + return {"status": "success", "appid": str(appid), "name": game_info["name"], "launch_options": "", "message": "Game install directory does not exist."} + if self.steam.is_game_running(game_info): + return {"status": "error", "message": "Close the game before unpatching."} + marker = self._find_marker(install_root, str(appid)) + if not marker: + return {"status": "success", "appid": str(appid), "name": game_info["name"], "launch_options": "", "message": "No Framegen patch found for this game."} + metadata = self._read_marker(marker) + target_dir = self._marker_target(marker, metadata) + original_options = str(metadata.get("original_launch_options") or "") + cleanup = self._unpatch_target(target_dir, marker) + if cleanup["status"] != "success": + return {**cleanup, "appid": str(appid), "name": game_info["name"]} + self._log(f"unpatch_game success: appid={appid} target={target_dir}") + return { + "status": "success", + "appid": str(appid), + "name": game_info["name"], + "launch_options": original_options, + "message": f"Unpatched {game_info['name']}.", + } + except Exception as exc: + self._log(f"unpatch_game failed for {appid}: {exc}") + return {"status": "error", "message": str(exc)} diff --git a/py_modules/services/steam.py b/py_modules/services/steam.py new file mode 100644 index 0000000..c2e90b0 --- /dev/null +++ b/py_modules/services/steam.py @@ -0,0 +1,187 @@ +"""Steam library discovery and conservative executable selection.""" + +import re +import subprocess +from pathlib import Path +from typing import Callable + + +Logger = Callable[[str], None] + +BAD_EXE_SUBSTRINGS = [ + "crashreport", + "crashreportclient", + "eac", + "easyanticheat", + "beclient", + "eosbootstrap", + "benchmark", + "uninstall", + "setup", + "launcher", + "updater", + "bootstrap", + "_redist", + "prereq", +] + + +class SteamLibrary: + def __init__(self, home: Path, logger: Logger): + self.home = home + self.logger = logger + + def steam_common_path(self) -> Path: + return self.home / ".local" / "share" / "Steam" / "steamapps" / "common" + + def _steam_root_candidates(self) -> list[Path]: + candidates = [ + self.home / ".local" / "share" / "Steam", + self.home / ".steam" / "steam", + self.home / ".steam" / "root", + self.home / ".var" / "app" / "com.valvesoftware.Steam" / "home" / ".local" / "share" / "Steam", + self.home / ".var" / "app" / "com.valvesoftware.Steam" / "home" / ".steam" / "steam", + ] + unique: list[Path] = [] + seen: set[str] = set() + for candidate in candidates: + key = str(candidate) + if key not in seen: + unique.append(candidate) + seen.add(key) + return unique + + def _steam_library_paths(self) -> list[Path]: + library_paths: list[Path] = [] + seen: set[str] = set() + for steam_root in self._steam_root_candidates(): + if steam_root.exists() and str(steam_root) not in seen: + library_paths.append(steam_root) + seen.add(str(steam_root)) + library_file = steam_root / "steamapps" / "libraryfolders.vdf" + if not library_file.exists(): + continue + try: + for line in library_file.read_text(encoding="utf-8", errors="replace").splitlines(): + if '"path"' not in line: + continue + path = line.split('"path"', 1)[1].strip().strip('"').replace("\\\\", "/") + candidate = Path(path) + if str(candidate) not in seen: + library_paths.append(candidate) + seen.add(str(candidate)) + except OSError as exc: + self.logger(f"[Framegen] failed to parse libraryfolders: {library_file}: {exc}") + return library_paths + + def find_installed_games(self, appid: str | None = None) -> list[dict]: + games: list[dict] = [] + for library_path in self._steam_library_paths(): + steamapps_path = library_path / "steamapps" + if not steamapps_path.exists(): + continue + for appmanifest in steamapps_path.glob("appmanifest_*.acf"): + game_info: dict = { + "appid": "", + "name": "", + "library_path": str(library_path), + "install_path": "", + } + install_dir = "" + try: + for line in appmanifest.read_text(encoding="utf-8", errors="replace").splitlines(): + if '"appid"' in line: + game_info["appid"] = line.split('"appid"', 1)[1].strip().strip('"') + elif '"name"' in line: + game_info["name"] = line.split('"name"', 1)[1].strip().strip('"') + elif '"installdir"' in line: + install_dir = line.split('"installdir"', 1)[1].strip().strip('"') + except OSError as exc: + self.logger(f"[Framegen] skipping manifest {appmanifest}: {exc}") + continue + if not game_info["appid"] or not game_info["name"]: + continue + if "Proton" in game_info["name"] or "Steam Linux Runtime" in game_info["name"]: + continue + game_info["install_path"] = str(steamapps_path / "common" / install_dir) if install_dir else str(Path()) + if appid is None or str(game_info["appid"]) == str(appid): + games.append(game_info) + deduped = {str(game["appid"]): game for game in games} + return sorted(deduped.values(), key=lambda game: game["name"].lower()) + + def game_record(self, appid: str) -> dict | None: + matches = self.find_installed_games(appid) + return matches[0] if matches else None + + @staticmethod + def normalized_path_string(value: str) -> str: + normalized = value.lower().replace("\\", "/") + normalized = normalized.replace("z:/", "/") + return normalized.replace("//", "/") + + def candidate_executables(self, install_root: Path) -> list[Path]: + if not install_root.exists(): + return [] + try: + return [exe for exe in install_root.rglob("*.exe") if exe.is_file()] + except OSError as exc: + self.logger(f"[Framegen] exe scan failed for {install_root}: {exc}") + return [] + + def _exe_score(self, exe: Path, install_root: Path, game_name: str) -> int: + normalized = self.normalized_path_string(str(exe)) + name = exe.name.lower() + score = 0 + if normalized.endswith("-win64-shipping.exe"): + score += 300 + if "shipping.exe" in name: + score += 220 + if "/binaries/win64/" in normalized: + score += 200 + if "/win64/" in normalized: + score += 80 + if exe.parent == install_root: + score += 20 + sanitized_game = re.sub(r"[^a-z0-9]", "", game_name.lower()) + sanitized_name = re.sub(r"[^a-z0-9]", "", exe.stem.lower()) + sanitized_root = re.sub(r"[^a-z0-9]", "", install_root.name.lower()) + if sanitized_game and sanitized_game in sanitized_name: + score += 120 + if sanitized_root and sanitized_root in sanitized_name: + score += 90 + for bad in BAD_EXE_SUBSTRINGS: + if bad in normalized: + score -= 200 + return score - len(exe.parts) + + def best_running_executable(self, candidates: list[Path]) -> Path | None: + if not candidates: + return None + try: + result = subprocess.run(["ps", "-eo", "args="], capture_output=True, text=True, check=False) + process_lines = result.stdout.splitlines() + except OSError as exc: + self.logger(f"[Framegen] running exe scan failed: {exc}") + return None + normalized_candidates = [(exe, self.normalized_path_string(str(exe))) for exe in candidates] + matches: list[tuple[int, Path]] = [] + for line in process_lines: + normalized_line = self.normalized_path_string(line) + for exe, normalized_exe in normalized_candidates: + if normalized_exe in normalized_line: + matches.append((len(normalized_exe), exe)) + return max(matches, key=lambda item: item[0])[1] if matches else None + + def guess_patch_target(self, game_info: dict) -> tuple[Path, Path | None]: + install_root = Path(game_info["install_path"]) + candidates = self.candidate_executables(install_root) + if not candidates: + return install_root, None + running_exe = self.best_running_executable(candidates) + if running_exe: + return running_exe.parent, running_exe + best = max(candidates, key=lambda exe: self._exe_score(exe, install_root, game_info["name"])) + return best.parent, best + + def is_game_running(self, game_info: dict) -> bool: + return self.best_running_executable(self.candidate_executables(Path(game_info["install_path"]))) is not None diff --git a/tests/test_backend.py b/tests/test_backend.py index 69165f1..fafee68 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -1,11 +1,14 @@ +import sys import tempfile import unittest from pathlib import Path -from backend import config -from backend.bundle import BundleManager -from backend.patcher import PatchManager -from backend.steam import SteamLibrary +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "py_modules")) + +from services import config +from services.bundle import BundleManager +from services.patcher import PatchManager +from services.steam import SteamLibrary class BackendSafetyTests(unittest.TestCase): -- cgit v1.2.3