diff options
Diffstat (limited to 'backend')
| -rw-r--r-- | backend/__init__.py | 1 | ||||
| -rw-r--r-- | backend/assets.py | 184 | ||||
| -rw-r--r-- | backend/bundle.py | 522 | ||||
| -rw-r--r-- | backend/config.py | 175 | ||||
| -rw-r--r-- | backend/files.py | 36 | ||||
| -rw-r--r-- | backend/patcher.py | 685 | ||||
| -rw-r--r-- | backend/steam.py | 187 |
7 files changed, 0 insertions, 1790 deletions
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<section>[^\]]+)\]\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 |