From 450d3e5e6612d467a00bb937538fded640c66ecb Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:24:05 -0400 Subject: refactor: simplify migration implementation --- py_modules/lsfg_vk/config_schema.py | 73 ++---- py_modules/lsfg_vk/flatpak_service.py | 460 +++++++++++----------------------- py_modules/lsfg_vk/installation.py | 79 +++--- py_modules/lsfg_vk/plugin.py | 164 ++++-------- py_modules/lsfg_vk/steam_service.py | 285 ++++++++------------- py_modules/lsfg_vk/types.py | 29 +-- 6 files changed, 355 insertions(+), 735 deletions(-) (limited to 'py_modules/lsfg_vk') diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py index ce109f3..3816d88 100644 --- a/py_modules/lsfg_vk/config_schema.py +++ b/py_modules/lsfg_vk/config_schema.py @@ -1,13 +1,9 @@ """Small adapter for the upstream lsfg-vk v2 configuration format.""" import json -import sys import tomllib -from pathlib import Path from typing import Any, Dict, TypedDict -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - ConfigurationData = Dict[str, Any] @@ -57,8 +53,8 @@ class ConfigurationManager: def validate_config(config: Dict[str, Any]) -> Dict[str, Any]: result = {**PROFILE_DEFAULTS, **GLOBAL_DEFAULTS} result.update({key: value for key, value in config.items() if key in result}) - result["active_in"] = _normalize_active_in(result.get("active_in")) - result["pacing_mode"] = str(result.get("pacing_mode", "vsync")).lower() + result["active_in"] = _normalize_active_in(result["active_in"]) + result["pacing_mode"] = str(result["pacing_mode"]).lower() if result["pacing_mode"] != "vsync": raise ValueError("pacing_mode must be vsync") result["multiplier"] = int(result["multiplier"]) @@ -67,43 +63,24 @@ class ConfigurationManager: result["flow_scale"] = float(result["flow_scale"]) if not 0.25 <= result["flow_scale"] <= 1.0: raise ValueError("flow_scale must be between 0.25 and 1.0") - for name in ("no_fp16", "performance_mode", "override_present_mode", "preserve_swapchain_image_count"): + for name in ( + "no_fp16", + "performance_mode", + "override_present_mode", + "preserve_swapchain_image_count", + ): result[name] = bool(result[name]) - result["dll"] = str(result.get("dll") or "") + result["dll"] = str(result["dll"] or "") return result - @staticmethod - def _migrate_dll_path(value: Any) -> str: - path_value = str(value or "") - if not path_value: - return "" - path = Path(path_value) - if path.name.lower() in {"lossless.dll"}: - return str(path.with_name("lsfg-vk.dll")) - return path_value - - @staticmethod - def _config_from_profile(profile: Dict[str, Any], global_config: Dict[str, Any]) -> Dict[str, Any]: - raw = dict(profile) - if "pacing_mode" not in raw and "pacing" in raw: - raw["pacing_mode"] = raw["pacing"] - if "override_present_mode" not in raw and "experimental_present_mode" in raw: - raw["override_present_mode"] = raw["experimental_present_mode"] == "fifo" - raw["dll"] = global_config.get("dll", "") - raw["no_fp16"] = global_config.get("no_fp16", False) - return ConfigurationManager.validate_config(raw) - @staticmethod def generate_toml_content_multi_profile(profile_data: ProfileData) -> str: global_config = {**GLOBAL_DEFAULTS, **profile_data.get("global_config", {})} lines = ["version = 2", "", "[global]"] - dll = ConfigurationManager._migrate_dll_path(global_config.get("dll")) - if dll: - lines.append(f"dll = {_toml_value(dll)}") - lines.append(f"allow_fp16 = {_toml_value(not bool(global_config.get('no_fp16', False)))}") - profiles = sorted(profile_data["profiles"].items()) - if not profiles: - profiles = [("", {})] + if global_config["dll"]: + lines.append(f"dll = {_toml_value(global_config['dll'])}") + lines.append(f"allow_fp16 = {_toml_value(not bool(global_config['no_fp16']))}") + profiles = sorted(profile_data["profiles"].items()) or [("", {})] for name, raw in profiles: config = ConfigurationManager.validate_config({**raw, **global_config}) lines.extend(["", "[[profile]]", f"name = {_toml_value(name)}"]) @@ -122,26 +99,20 @@ class ConfigurationManager: @staticmethod def parse_toml_content_multi_profile(content: str) -> ProfileData: data = tomllib.loads(content) - version = data.get("version") - if version not in (1, 2): + if data.get("version") != 2: raise ValueError("unsupported lsfg-vk configuration version") - raw_global = dict(data.get("global", {})) + raw_global = data.get("global", {}) global_config = { - "dll": ConfigurationManager._migrate_dll_path(raw_global.get("dll", "")), + "dll": str(raw_global.get("dll", "") or ""), "no_fp16": not bool(raw_global.get("allow_fp16", True)), } profiles: Dict[str, Dict[str, Any]] = {} - source_profiles = data.get("game", []) if version == 1 else data.get("profile", []) - for profile in source_profiles: - name = str(profile.get("exe" if version == 1 else "name", "")) - config = ConfigurationManager._config_from_profile(profile, global_config) + for profile in data.get("profile", []): + name = str(profile.get("name", "")) + config = ConfigurationManager.validate_config({ + **profile, + **global_config, + }) if config["active_in"]: profiles[name] = config return {"profiles": profiles, "global_config": global_config} - - @staticmethod - def is_legacy_v1(content: str) -> bool: - try: - return tomllib.loads(content).get("version") == 1 - except tomllib.TOMLDecodeError: - return False diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 11f1a82..d2241ab 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,4 +1,4 @@ -"""Flatpak runtime-extension infrastructure for unified game targets.""" +"""Flatpak runtime support for classified Steam targets.""" from __future__ import annotations @@ -10,7 +10,7 @@ import shutil import subprocess import threading from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Dict, Optional, Set from .base_service import BaseService from .constants import ( @@ -22,14 +22,6 @@ from .constants import ( class FlatpakService(BaseService): - """Resolve and provision only the runtime support a target actually needs. - - Flatpak application permissions are deliberately not persisted here. The - generated per-AppID wrapper supplies the narrow launch-time permissions and - environment instead, while this service owns only the shared Vulkan layer - runtime extensions installed from the plugin bundle. - """ - EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") OWNERSHIP_FILENAME = "flatpak_extensions.json" @@ -37,7 +29,6 @@ class FlatpakService(BaseService): APP_ID_PATTERN = re.compile( r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$" ) - BRANCH_PATTERN = re.compile(r"^[0-9]+\.[0-9]+$") def __init__(self, logger=None): super().__init__(logger) @@ -48,39 +39,37 @@ class FlatpakService(BaseService): def ownership_path(self) -> Path: return self.config_dir / self.OWNERSHIP_FILENAME - def _get_clean_env(self) -> Dict[str, str]: + def _clean_env(self) -> Dict[str, str]: env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) env["HOME"] = str(self.user_home) - path_entries = [entry for entry in env.get("PATH", "").split(":") if entry] + path = [entry for entry in env.get("PATH", "").split(":") if entry] for entry in ("/usr/bin", "/usr/local/bin", "/bin"): - if entry not in path_entries: - path_entries.insert(0, entry) - env["PATH"] = ":".join(path_entries) + if entry not in path: + path.insert(0, entry) + env["PATH"] = ":".join(path) return env - def _flatpak_user(self) -> pwd.struct_passwd: - try: - return pwd.getpwuid(self.user_home.stat().st_uid) - except (KeyError, OSError) as error: - raise RuntimeError(f"Unable to resolve Flatpak user for {self.user_home}") from error - def check_flatpak_available(self) -> bool: - env = self._get_clean_env() + env = self._clean_env() self.flatpak_command = shutil.which("flatpak", path=env["PATH"]) return self.flatpak_command is not None - def _run_flatpak_command(self, args: List[str], **kwargs): + def _run_flatpak_command(self, args, **kwargs): if self.flatpak_command is None and not self.check_flatpak_available(): raise FileNotFoundError("Flatpak command not available") + env = self._clean_env() command = [self.flatpak_command, *args] - target_user = self._flatpak_user() - if os.geteuid() != target_user.pw_uid: - runuser = shutil.which("runuser", path=self._get_clean_env()["PATH"]) + try: + user = pwd.getpwuid(self.user_home.stat().st_uid) + except (KeyError, OSError) as error: + raise RuntimeError(f"Unable to resolve Flatpak user for {self.user_home}") from error + if os.geteuid() != user.pw_uid: + runuser = shutil.which("runuser", path=env["PATH"]) if runuser is None: raise FileNotFoundError("runuser command not available") - command = [runuser, "--user", target_user.pw_name, "--", *command] - return subprocess.run(command, env=self._get_clean_env(), **kwargs) + command = [runuser, "--user", user.pw_name, "--", *command] + return subprocess.run(command, env=env, **kwargs) @classmethod def _validate_app_id(cls, app_id: str) -> str: @@ -89,45 +78,32 @@ class FlatpakService(BaseService): return app_id @classmethod - def _validate_runtime(cls, version: str) -> str: - if version not in cls.SUPPORTED_RUNTIMES: + def _validate_runtime(cls, branch: str) -> str: + if branch not in cls.SUPPORTED_RUNTIMES: raise ValueError( - f"Unsupported Flatpak runtime branch {version}; " - f"supported branches are {', '.join(cls.SUPPORTED_RUNTIMES)}" + f"Unsupported Flatpak runtime branch {branch}; supported branches are " + + ", ".join(cls.SUPPORTED_RUNTIMES) ) - return version - - @classmethod - def _extension_ref(cls, version: str) -> str: - return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(version)}" + return branch @classmethod def runtime_branch_from_ref(cls, runtime_ref: str) -> str: - """Return the supported Freedesktop branch from a runtime ref.""" - if not isinstance(runtime_ref, str): - raise ValueError("Flatpak did not return a runtime reference") - parts = runtime_ref.strip().split("/") + parts = runtime_ref.strip().split("/") if isinstance(runtime_ref, str) else [] if len(parts) != 3 or parts[0] != "org.freedesktop.Platform": raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}") - branch = parts[2] - if not cls.BRANCH_PATTERN.fullmatch(branch): - raise ValueError(f"Unrecognized Flatpak runtime branch: {branch}") - return cls._validate_runtime(branch) + return cls._validate_runtime(parts[2]) @classmethod - def _bundle_filename(cls, version: str) -> str: - return { + def _extension_ref(cls, branch: str) -> str: + return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}" + + def _bundled_extension_path(self, branch: str) -> Path: + filename = { "23.08": FLATPAK_23_08_FILENAME, "24.08": FLATPAK_24_08_FILENAME, "25.08": FLATPAK_25_08_FILENAME, - }[cls._validate_runtime(version)] - - def _bundled_extension_path(self, version: str) -> Path: - return ( - Path(__file__).resolve().parent.parent.parent - / BIN_DIR - / self._bundle_filename(version) - ) + }[self._validate_runtime(branch)] + return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename def _installed_extension_branches(self) -> Set[str]: result = self._run_flatpak_command( @@ -136,78 +112,54 @@ class FlatpakService(BaseService): text=True, check=True, ) - installed: Set[str] = set() + installed = set() for line in result.stdout.splitlines(): - if not line.strip(): - continue - fields = line.split("\t") - if len(fields) < 3: - fields = line.split() - if len(fields) < 3: - continue - application, arch, branch = (field.strip() for field in fields[:3]) - if application == self.EXTENSION_ID and arch == "x86_64": - installed.add(branch) + fields = line.split("\t") if "\t" in line else line.split() + if len(fields) >= 3 and fields[0] == self.EXTENSION_ID and fields[1] == "x86_64": + installed.add(fields[2]) return installed - def _read_owned_branches(self) -> Tuple[Set[str], bool]: - """Read ownership without guessing when metadata is damaged.""" + def _owned_branches(self) -> Set[str]: path = self.ownership_path - if path.is_symlink(): - self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") - return set(), True - if not path.exists(): - return set(), False - if not path.is_file(): - self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") - return set(), True + if not path.exists() and not path.is_symlink(): + return set() + if path.is_symlink() or not path.is_file(): + raise RuntimeError("Flatpak ownership metadata is not a regular file") try: - raw = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(raw, dict) or raw.get("version") != self.OWNERSHIP_VERSION: - raise ValueError("unsupported ownership metadata version") - branches = raw.get("plugin_owned_branches") - if not isinstance(branches, list): - raise ValueError("plugin_owned_branches is not a list") - normalized = { - self._validate_runtime(branch) - for branch in branches - if isinstance(branch, str) - } - if len(normalized) != len(branches): - raise ValueError("ownership metadata contains invalid branches") - return normalized, False + data = json.loads(path.read_text(encoding="utf-8")) + branches = data.get("plugin_owned_branches") + if data.get("version") != self.OWNERSHIP_VERSION or not isinstance(branches, list): + raise ValueError("invalid ownership metadata") + owned = {self._validate_runtime(branch) for branch in branches} + if len(owned) != len(branches): + raise ValueError("invalid ownership metadata") + return owned except (OSError, json.JSONDecodeError, TypeError, ValueError) as error: - self.log.warning(f"Could not trust Flatpak ownership metadata: {error}") - return set(), True + raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error def _write_owned_branches(self, branches: Set[str]) -> None: if not branches: - if self.ownership_path.exists() or self.ownership_path.is_symlink(): - self.ownership_path.unlink() + self.ownership_path.unlink(missing_ok=True) return - document = { - "version": self.OWNERSHIP_VERSION, - "plugin_owned_branches": sorted(branches), - } - self._write_file(self.ownership_path, json.dumps(document, indent=2) + "\n") + self._write_file( + self.ownership_path, + json.dumps( + { + "version": self.OWNERSHIP_VERSION, + "plugin_owned_branches": sorted(branches), + }, + indent=2, + ) + "\n", + ) - def get_extension_status(self) -> Dict[str, Any]: - """Return global extension inventory for Setup diagnostics.""" + def get_extension_status(self): try: - if not self.check_flatpak_available(): - return self._success_response( - dict, - "Flatpak is not available", - available=False, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=[], - ) - installed = self._installed_extension_branches() + available = self.check_flatpak_available() + installed = self._installed_extension_branches() if available else set() return self._success_response( dict, - "Flatpak runtime extension status retrieved", - available=True, + "Flatpak runtime extension status retrieved" if available else "Flatpak is not available", + available=available, extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=sorted(installed), @@ -216,16 +168,15 @@ class FlatpakService(BaseService): return self._error_response( dict, str(error), - available=self.check_flatpak_available(), + available=False, extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], ) - def get_flatpak_support_status(self) -> Dict[str, Any]: - return self.get_extension_status() + get_flatpak_support_status = get_extension_status - def _resolve_runtime(self, app_id: str) -> Dict[str, Any]: + def _resolve_runtime(self, app_id: str): self._validate_app_id(app_id) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") @@ -236,27 +187,21 @@ class FlatpakService(BaseService): ) if result.returncode != 0: raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}") - runtime_ref = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" - branch = self.runtime_branch_from_ref(runtime_ref) - return {"runtime": runtime_ref, "runtime_branch": branch} + runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" + return runtime, self.runtime_branch_from_ref(runtime) - def resolve_app_support(self, app_id: str) -> Dict[str, Any]: - """Resolve the exact runtime branch required by one Flatpak app.""" + def resolve_app_support(self, app_id: str): try: app_id = self._validate_app_id(app_id) - resolved = self._resolve_runtime(app_id) + runtime, branch = self._resolve_runtime(app_id) installed = self._installed_extension_branches() - branch = resolved["runtime_branch"] ready = branch in installed return self._success_response( dict, - ( - f"lsfg-vk support is ready for {app_id}" - if ready - else f"lsfg-vk runtime extension {branch} is required for {app_id}" - ), + f"lsfg-vk support is ready for {app_id}" if ready + else f"lsfg-vk runtime extension {branch} is required for {app_id}", flatpak_app_id=app_id, - runtime=resolved["runtime"], + runtime=runtime, runtime_branch=branch, support_status="ready" if ready else "needs-runtime", extension_installed=ready, @@ -286,194 +231,114 @@ class FlatpakService(BaseService): installed_branches=[], ) - def install_extension(self, version: str) -> Dict[str, Any]: - """Install one branch, treating an already-installed branch as success.""" + def install_extension(self, branch: str): try: - version = self._validate_runtime(version) + branch = self._validate_runtime(branch) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - installed_before = self._installed_extension_branches() - if version in installed_before: - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension is already installed", - runtime_branch=version, - installed=True, - enabled=True, - ) - bundle_path = self._bundled_extension_path(version) - if not bundle_path.is_file(): - raise FileNotFoundError( - f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" - ) + if branch in self._installed_extension_branches(): + return self._extension_result(branch, True, False, "already installed") + bundle = self._bundled_extension_path(branch) + if not bundle.is_file(): + raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin") result = self._run_flatpak_command( - [ - "install", - "--user", - "--noninteractive", - "--or-update", - str(bundle_path), - ], + ["install", "--user", "--noninteractive", "--or-update", str(bundle)], capture_output=True, text=True, ) if result.returncode != 0: raise OSError(result.stderr.strip() or "Flatpak installation failed") - installed_after = self._installed_extension_branches() - if version not in installed_after: - raise RuntimeError( - f"Flatpak install completed but {self._extension_ref(version)} " - "was not visible afterwards" - ) - owned, uncertain = self._read_owned_branches() - if not uncertain: - owned.add(version) + if branch not in self._installed_extension_branches(): + raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards") + owned = self._owned_branches() + owned.add(branch) + self._write_owned_branches(owned) + return self._extension_result(branch, True, False, "installed") + except Exception as error: + return self._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False) + + def _remove_extension(self, branch: str) -> bool: + if branch not in self._installed_extension_branches(): + return False + result = self._run_flatpak_command( + ["uninstall", "--user", "--noninteractive", self._extension_ref(branch)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Flatpak uninstall failed") + if branch in self._installed_extension_branches(): + raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed") + return True + + def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str): + return self._success_response( + dict, + f"lsfg-vk {branch} runtime extension {verb}", + runtime_branch=branch, + installed=installed, + enabled=installed, + removed=removed, + ) + + def uninstall_extension(self, branch: str): + try: + branch = self._validate_runtime(branch) + if not self.check_flatpak_available(): + raise FileNotFoundError("Flatpak is not available on this system") + with self._lock: + removed = self._remove_extension(branch) + owned = self._owned_branches() + if branch in owned: + owned.remove(branch) self._write_owned_branches(owned) - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension installed", - runtime_branch=version, - installed=True, - enabled=True, - ) + return self._extension_result(branch, False, removed, "uninstalled") except Exception as error: return self._error_response( dict, str(error), - runtime_branch=version, + runtime_branch=branch, + removed=False, installed=False, enabled=False, ) - def ensure_extension(self, version: str) -> Dict[str, Any]: - status = self.get_extension_status() - if not status.get("success"): - return status - if not status.get("available"): - return self._error_response( - dict, - "Flatpak is not available on this system", - runtime_branch=version, - support_status="error", - ) + def ensure_extension(self, branch: str): try: - version = self._validate_runtime(version) - except ValueError as error: - return self._error_response(dict, str(error), runtime_branch=version, support_status="unsupported") - if version in status.get("installed_branches", []): - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension is ready", - runtime_branch=version, - installed=True, - ) - return self.install_extension(version) + branch = self._validate_runtime(branch) + if branch in self._installed_extension_branches(): + return self._extension_result(branch, True, False, "is ready") + except Exception as error: + return self._error_response(dict, str(error), runtime_branch=branch, support_status="error") + return self.install_extension(branch) - def ensure_app_support(self, app_id: str) -> Dict[str, Any]: - """Provision only the branch returned by flatpak info for this app.""" + def ensure_app_support(self, app_id: str): resolved = self.resolve_app_support(app_id) if not resolved.get("success") or resolved.get("support_status") != "needs-runtime": return resolved - branch = resolved.get("runtime_branch") - result = self.ensure_extension(branch) + result = self.ensure_extension(resolved["runtime_branch"]) if not result.get("success"): return self._error_response( dict, result.get("error") or "Could not install the required Flatpak runtime extension", flatpak_app_id=app_id, runtime=resolved.get("runtime"), - runtime_branch=branch, + runtime_branch=resolved.get("runtime_branch"), support_status="error", extension_installed=False, ) - final = self.resolve_app_support(app_id) - if final.get("success") and final.get("support_status") == "ready": - return final - return self._error_response( - dict, - final.get("error") or "Required Flatpak runtime extension could not be verified", - flatpak_app_id=app_id, - runtime=resolved.get("runtime"), - runtime_branch=branch, - support_status="error", - extension_installed=False, - ) - - def uninstall_extension(self, version: str) -> Dict[str, Any]: - """Uninstall one branch, treating an already-absent branch as success.""" - try: - version = self._validate_runtime(version) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - installed = self._installed_extension_branches() - was_installed = version in installed - if was_installed: - result = self._run_flatpak_command( - [ - "uninstall", - "--user", - "--noninteractive", - self._extension_ref(version), - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - if version in self._installed_extension_branches(): - raise RuntimeError( - f"Flatpak uninstall completed but {self._extension_ref(version)} " - "is still installed" - ) - owned, uncertain = self._read_owned_branches() - if not uncertain and version in owned: - owned.remove(version) - self._write_owned_branches(owned) - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension uninstalled", - runtime_branch=version, - removed=was_installed, - installed=False, - enabled=False, - ) - except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=version, - removed=False, - installed=False, - enabled=False, - ) + return self.resolve_app_support(app_id) - def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: - """Set one runtime branch to the requested state, safely and idempotently.""" + def set_extension_enabled(self, branch: str, enabled: bool): if type(enabled) is not bool: - return self._error_response( - dict, - "enabled must be a boolean", - runtime_branch=version, - installed=False, - enabled=False, - ) - return self.install_extension(version) if enabled else self.uninstall_extension(version) + return self._error_response(dict, "enabled must be a boolean", runtime_branch=branch, installed=False, enabled=False) + return self.install_extension(branch) if enabled else self.uninstall_extension(branch) - def remove_plugin_owned_extensions(self) -> Dict[str, Any]: - """Uninstall only branches recorded as installed by this plugin.""" + def remove_plugin_owned_extensions(self): try: with self._lock: - owned, uncertain = self._read_owned_branches() - if uncertain: - return self._error_response( - dict, - "Flatpak ownership metadata is uncertain; no extensions were removed", - removed_branches=[], - preserved_branches=[], - ownership_uncertain=True, - ) + owned = self._owned_branches() if not owned: return self._success_response( dict, @@ -483,36 +348,11 @@ class FlatpakService(BaseService): ownership_uncertain=False, ) if not self.check_flatpak_available(): - return self._error_response( - dict, - "Flatpak is not available; plugin-owned extension metadata was preserved", - removed_branches=[], - preserved_branches=sorted(owned), - ownership_uncertain=False, - ) - removed: List[str] = [] - failures: List[str] = [] + raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved") + removed, failures = [], [] for branch in sorted(owned): try: - installed = self._installed_extension_branches() - if branch in installed: - result = self._run_flatpak_command( - [ - "uninstall", - "--user", - "--noninteractive", - self._extension_ref(branch), - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - if branch in self._installed_extension_branches(): - raise RuntimeError( - f"Flatpak uninstall completed but {self._extension_ref(branch)} " - "is still installed" - ) + self._remove_extension(branch) removed.append(branch) except Exception as error: failures.append(f"{branch}: {error}") @@ -539,5 +379,5 @@ class FlatpakService(BaseService): str(error), removed_branches=[], preserved_branches=[], - ownership_uncertain=False, + ownership_uncertain=True, ) diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 8a3094d..5a583f5 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -48,26 +48,20 @@ class InstallationService(BaseService): def install(self) -> InstallationResponse: try: - plugin_dir = Path(__file__).parent.parent.parent - archive_path = plugin_dir / BIN_DIR / ARCHIVE_FILENAME + archive_path = Path(__file__).parent.parent.parent / BIN_DIR / ARCHIVE_FILENAME if not archive_path.exists(): raise FileNotFoundError(f"{ARCHIVE_FILENAME} not found at {archive_path}") - self._ensure_directories() profile_data = self._prepare_config() self._install_archive(archive_path) - config_content = ConfigurationManager.generate_toml_content_multi_profile(profile_data) - self.runtime_service.validate_config_content(config_content) - self._write_file( - self.config_file_path, - config_content, - 0o644, - ) + content = ConfigurationManager.generate_toml_content_multi_profile(profile_data) + self.runtime_service.validate_config_content(content) + self._write_file(self.config_file_path, content, 0o644) self._remove_legacy_layer_files() return self._success_response(InstallationResponse, "lsfg-vk 2.0.0 installed successfully") except Exception as error: self.log.error(f"Error installing lsfg-vk: {error}") - return self._error_response(InstallationResponse, str(error), message="") + return self._error_response(InstallationResponse, str(error)) def _payload_destinations(self) -> Dict[str, tuple[Path, int]]: return { @@ -82,7 +76,7 @@ class InstallationService(BaseService): 0o644, ), f"share/icons/hicolor/256x256/apps/{UI_ICON_FILENAME}": ( - self.user_home / LOCAL_SHARE / "icons" / "hicolor" / "256x256" / "apps" / UI_ICON_FILENAME, + self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps" / UI_ICON_FILENAME, 0o644, ), } @@ -124,35 +118,26 @@ class InstallationService(BaseService): if temporary_path is not None: temporary_path.unlink(missing_ok=True) raise - missing = sorted(set(destinations) - found) if missing: raise OSError("Archive is missing required files: " + ", ".join(missing)) def _prepare_config(self) -> ProfileData: if self.config_file_path.exists(): - content = self.config_file_path.read_text(encoding="utf-8") - legacy = ConfigurationManager.is_legacy_v1(content) - profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) - if legacy: - backup_path = self.config_file_path.with_name(f"{self.config_file_path.name}.v1.bak") - if not backup_path.exists(): - self._write_file(backup_path, content, 0o644) + profile_data = ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) else: - default = dict(ConfigurationManager.get_defaults()) + defaults = ConfigurationManager.get_defaults() profile_data = ProfileData( profiles={}, - global_config={ - "dll": default.get("dll", ""), - "no_fp16": default.get("no_fp16", False), - }, + global_config={"dll": defaults["dll"], "no_fp16": defaults["no_fp16"]}, ) - self._resolve_dll_path(profile_data) - defaults = dict(ConfigurationManager.get_defaults()) - for profile_name, raw_profile in list(profile_data["profiles"].items()): - profile_data["profiles"][profile_name] = ConfigurationManager.validate_config( - {**defaults, **raw_profile, **profile_data["global_config"]} + defaults = ConfigurationManager.get_defaults() + for name, profile in profile_data["profiles"].items(): + profile_data["profiles"][name] = ConfigurationManager.validate_config( + {**defaults, **profile, **profile_data["global_config"]} ) return profile_data @@ -160,7 +145,6 @@ class InstallationService(BaseService): current_path = str(profile_data["global_config"].get("dll") or "") if current_path and Path(current_path).is_file(): return False - dll_path = self.steam_service.find_lsfg_vk_dll() if dll_path and current_path != dll_path: profile_data["global_config"]["dll"] = dll_path @@ -179,7 +163,6 @@ class InstallationService(BaseService): except Exception as error: installed = False installation_error = str(error) - lossless_scaling = self.runtime_service.check_lossless_scaling() return { "installed": installed, @@ -197,21 +180,22 @@ class InstallationService(BaseService): def uninstall(self) -> UninstallationResponse: try: - removed = [] - for path in ( - self.lib_file, - self.lib_x86_file, - self.json_file, - self.json_x86_file, - self.cli_file, - self.local_bin_dir / UI_FILENAME, - self.user_home / LOCAL_SHARE / "applications" / UI_DESKTOP_FILENAME, - self.user_home / LOCAL_SHARE / "icons" / "hicolor" / "256x256" / "apps" / UI_ICON_FILENAME, - self.legacy_lib_file, - self.legacy_json_file, - ): - if self._remove_if_exists(path): - removed.append(str(path)) + removed = [ + str(path) + for path in ( + self.lib_file, + self.lib_x86_file, + self.json_file, + self.json_x86_file, + self.cli_file, + self.local_bin_dir / UI_FILENAME, + self.user_home / LOCAL_SHARE / "applications" / UI_DESKTOP_FILENAME, + self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps" / UI_ICON_FILENAME, + self.legacy_lib_file, + self.legacy_json_file, + ) + if self._remove_if_exists(path) + ] if not removed: return self._success_response( UninstallationResponse, @@ -227,7 +211,6 @@ class InstallationService(BaseService): return self._error_response( UninstallationResponse, str(error), - message="", removed_files=None, ) diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index c576cc2..071b19b 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -1,34 +1,18 @@ -""" -Main plugin class for the lsfg-vk Decky Loader plugin. - -This plugin provides services for installing and managing the lsfg-vk -Vulkan layer for frame generation on Steam Deck. -""" - import os from typing import Any, Dict, Optional import decky -from .installation import InstallationService from .configuration import ConfigurationService from .flatpak_service import FlatpakService +from .installation import InstallationService from .runtime_service import RuntimeService from .steam_service import SteamService from .wrapper_service import WrapperService class Plugin: - """ - Main plugin class for lsfg-vk management. - - This class provides a unified interface for installation, configuration, - and Flatpak management services. It implements the Decky Loader plugin lifecycle - methods (_main, _unload, _uninstall, _migration). - """ - def __init__(self): - """Initialize the plugin with all necessary services""" self.runtime_service = RuntimeService() self.steam_service = SteamService() self.installation_service = InstallationService( @@ -39,63 +23,45 @@ class Plugin: self.flatpak_service = FlatpakService() self.wrapper_service = WrapperService() - async def install_lsfg_vk(self) -> Dict[str, Any]: - """Install the bundled lsfg-vk runtime to ~/.local - - Returns: - InstallationResponse dict with success status and message/error - """ + async def install_lsfg_vk(self): return self.installation_service.install() - async def check_lsfg_vk_installed(self) -> Dict[str, Any]: - """Check if lsfg-vk is already installed - - Returns: - InstallationCheckResponse dict with installation status and paths - """ + async def check_lsfg_vk_installed(self): return self.installation_service.check_installation() - async def uninstall_lsfg_vk(self) -> Dict[str, Any]: - """Uninstall lsfg-vk by removing the installed files - - Returns: - UninstallationResponse dict with success status and removed files - """ + async def uninstall_lsfg_vk(self): return self.installation_service.uninstall() - async def get_game_configs(self) -> Dict[str, Any]: + async def get_game_configs(self): return self.configuration_service.get_game_configs() - async def get_installed_games(self) -> Dict[str, Any]: + async def get_installed_games(self): result = self.steam_service.get_installed_games() if not result.get("success"): return result - - support_cache: Dict[str, Dict[str, Any]] = {} + cache: Dict[str, Dict[str, Any]] = {} for game in result.get("games", []): - transport = game.get("transport") if isinstance(game, dict) else None - if not isinstance(transport, dict) or transport.get("kind") != "flatpak": + transport = game.get("transport", {}) + if transport.get("kind") != "flatpak": continue - flatpak_app_id = transport.get("flatpakAppId") - if not isinstance(flatpak_app_id, str) or not flatpak_app_id: + app_id = transport.get("flatpakAppId") + if not app_id: continue - if flatpak_app_id not in support_cache: - support_cache[flatpak_app_id] = self.flatpak_service.resolve_app_support( - flatpak_app_id - ) - game["flatpakSupport"] = support_cache[flatpak_app_id] + if app_id not in cache: + cache[app_id] = self.flatpak_service.resolve_app_support(app_id) + game["flatpakSupport"] = cache[app_id] return result - async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]: + async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]): return self.configuration_service.update_game_config(appid, game_name, config) - async def reset_game_config(self, appid: str) -> Dict[str, Any]: + async def reset_game_config(self, appid: str): return self.configuration_service.reset_game_config(appid) - async def reset_all_game_configs(self) -> Dict[str, Any]: + async def reset_all_game_configs(self): return self.configuration_service.reset_all_game_configs() - async def get_workaround_state(self, appid: str) -> Dict[str, Any]: + async def get_workaround_state(self, appid: str): return self.wrapper_service.get(appid) async def set_workaround_state( @@ -105,7 +71,7 @@ class Plugin: shortcut_exe: Optional[str] = None, command_token_added: bool = False, transport: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: + ): return self.wrapper_service.set( appid, state, @@ -114,118 +80,82 @@ class Plugin: transport, ) - async def remove_workaround_state(self, appid: str) -> Dict[str, Any]: + async def remove_workaround_state(self, appid: str): return self.wrapper_service.remove(appid) - async def get_config_file_content(self) -> Dict[str, Any]: - """Get the current config file content - - Returns: - Dict containing the config file content or error message - """ + async def get_config_file_content(self): + path = self.configuration_service.config_file_path try: - config_path = self.configuration_service.config_file_path - if not config_path.exists(): + if not path.exists(): return { "success": False, "content": None, - "path": str(config_path), - "error": "Config file does not exist" + "path": str(path), + "error": "Config file does not exist", } - - content = config_path.read_text(encoding='utf-8') return { "success": True, - "content": content, - "path": str(config_path), - "error": None + "content": path.read_text(encoding="utf-8"), + "path": str(path), + "error": None, } - except Exception as e: + except Exception as error: return { "success": False, "content": None, - "path": str(config_path) if 'config_path' in locals() else "unknown", - "error": f"Error reading config file: {str(e)}" + "path": str(path), + "error": f"Error reading config file: {error}", } - async def get_lossless_scaling_branch_status(self) -> Dict[str, Any]: + async def get_lossless_scaling_branch_status(self): return self.steam_service.get_branch_status() - async def get_flatpak_support_status(self) -> Dict[str, Any]: + async def get_flatpak_support_status(self): return self.flatpak_service.get_flatpak_support_status() - async def ensure_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]: + async def ensure_flatpak_support(self, flatpak_app_id: str): return self.flatpak_service.ensure_app_support(flatpak_app_id) - async def repair_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]: + async def repair_flatpak_support(self, flatpak_app_id: str): return self.flatpak_service.ensure_app_support(flatpak_app_id) - async def set_flatpak_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: + async def set_flatpak_extension_enabled(self, version: str, enabled: bool): return self.flatpak_service.set_extension_enabled(version, enabled) - + async def _main(self): - """ - Main entry point for the plugin. - - This method is called by Decky Loader when the plugin is loaded. - Any initialization code should go here. - """ repair = self.wrapper_service.repair() if not repair.get("success"): decky.logger.error(f"Could not repair lsfg workaround wrapper: {repair.get('error')}") decky.logger.info("decky-lsfg-vk plugin loaded") async def _unload(self): - """ - Cleanup tasks when the plugin is unloaded. - - This method is called by Decky Loader when the plugin is being unloaded. - Any cleanup code should go here. - """ decky.logger.info("decky-lsfg-vk plugin unloaded") async def _uninstall(self): - """ - Called when the plugin is uninstalled. - - This method is called by Decky Loader when the plugin is being uninstalled. - Performs cleanup of plugin files and flatpak extensions. - """ decky.logger.info("decky-lsfg-vk plugin being uninstalled") - - # Clean up lsfg-vk files when the plugin is uninstalled - # Launch integrations are removed with their profiles. Keep the - # generated pass-through wrapper if it is still referenced elsewhere; - # InstallationService only removes files owned by the runtime bundle. self.installation_service.cleanup_on_uninstall() - try: result = self.flatpak_service.remove_plugin_owned_extensions() if not result.get("success"): decky.logger.warning(result.get("error")) except Exception as error: decky.logger.error(f"Error during Flatpak cleanup: {error}") - decky.logger.info("decky-lsfg-vk plugin uninstall cleanup completed") async def _migration(self): - """ - Migrations that should be performed before entering `_main()`. - - This method is called by Decky Loader for plugin migrations. - Currently migrates logs, settings, and runtime data from old locations. - """ decky.logger.info("Running decky-lsfg-vk plugin migrations") - - decky.migrate_logs(os.path.join(decky.DECKY_USER_HOME, - ".config", "decky-lossless-scaling-vk", "lossless-scaling-vk.log")) - + decky.migrate_logs(os.path.join( + decky.DECKY_USER_HOME, + ".config", + "decky-lossless-scaling-vk", + "lossless-scaling-vk.log", + )) decky.migrate_settings( os.path.join(decky.DECKY_HOME, "settings", "lossless-scaling-vk.json"), - os.path.join(decky.DECKY_USER_HOME, ".config", "decky-lossless-scaling-vk")) - + os.path.join(decky.DECKY_USER_HOME, ".config", "decky-lossless-scaling-vk"), + ) decky.migrate_runtime( os.path.join(decky.DECKY_HOME, "lossless-scaling-vk"), - os.path.join(decky.DECKY_USER_HOME, ".local", "share", "decky-lossless-scaling-vk")) - + os.path.join(decky.DECKY_USER_HOME, ".local", "share", "decky-lossless-scaling-vk"), + ) decky.logger.info("decky-lsfg-vk plugin migrations completed") diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py index 50d722b..201441e 100644 --- a/py_modules/lsfg_vk/steam_service.py +++ b/py_modules/lsfg_vk/steam_service.py @@ -10,7 +10,6 @@ from .constants import ( WRAPPER_FILENAME, ) - _FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$") _WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}" @@ -25,106 +24,88 @@ def _split_command(value: Optional[str]) -> Optional[list[str]]: def _is_managed_wrapper(value: str) -> bool: - """Recognize the wrapper Target while keeping arbitrary launchers as host games.""" if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}: return True path = Path(value) return path.is_absolute() and path.name == WRAPPER_FILENAME -def classify_shortcut_transport( - executable: Optional[str], - launch_options: Optional[str] = None, -) -> Dict[str, object]: - """Classify direct Flatpak invocations, including the managed wrapper Target.""" +def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]: executable_tokens = _split_command(executable) option_tokens = _split_command(launch_options) if executable_tokens is None or option_tokens is None or not executable_tokens: return {"kind": "host"} - direct_flatpak = executable_tokens[0] == "/usr/bin/flatpak" managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0]) if not direct_flatpak and not managed_wrapper: return {"kind": "host"} - arguments = [*executable_tokens[1:], *option_tokens] if not arguments or arguments[0] != "run": return {"kind": "host"} - for argument in arguments[1:]: - if argument == "--": + if argument == "--" or argument.startswith("-"): continue - if argument.startswith("-"): - continue - if _FLATPAK_APP_ID.fullmatch(argument): - return {"kind": "flatpak", "flatpakAppId": argument} - return {"kind": "host"} + return ( + {"kind": "flatpak", "flatpakAppId": argument} + if _FLATPAK_APP_ID.fullmatch(argument) + else {"kind": "host"} + ) return {"kind": "host"} +def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]: + return next((values[key] for key in keys if isinstance(values.get(key), str)), None) + + class SteamService(BaseService): DEFAULT_BRANCH = "public" MANIFEST_FILENAME = f"appmanifest_{STEAM_LOSSLESS_SCALING_APP_ID}.acf" - # Valve compatibility tools, runtimes, Steamworks redistributables, and LSFG. GAME_SELECTOR_EXCLUDED_APPIDS = { - "858280", # Proton 3.7 - "961940", # Proton 3.16 - "1054830", # Proton 4.2 - "1113280", # Proton 4.11 - "1245040", # Proton 5.0 - "1420170", # Proton 5.13 - "1493710", # Proton Experimental - "1580130", # Proton 6.3 - "1887720", # Proton 7 - "2180100", # Proton Hotfix - "228980", # Steamworks Common Redistributables - "2348590", # Proton 8 - "2805730", # Proton 9 - "3029110", # Lepton - "3127680", # fex - "3658110", # Proton 10 - "4183110", # Steam Linux Runtime 4.0 - "4185400", # Steam Linux Runtime 4.0 for arm64 - "4427310", # Proton Experimental (ARM64) - "4628710", # Proton 11 / Proton Next - "4628740", # Proton 11 (ARM64) - "4690330", # Legacy Steam Runtime - "993090", # Lossless Scaling - "1070560", # Steam Linux Runtime 1.0 - "1391110", # Steam Linux Runtime 2.0 - "1628350", # Steam Linux Runtime 3.0 + "858280", "961940", "1054830", "1113280", "1245040", "1420170", + "1493710", "1580130", "1887720", "2180100", "228980", "2348590", + "2805730", "3029110", "3127680", "3658110", "4183110", "4185400", + "4427310", "4628710", "4628740", "4690330", "993090", "1070560", + "1391110", "1628350", } def _steam_roots(self): - candidates = ( + seen = set() + for candidate in ( self.user_home / ".local/share/Steam", self.user_home / ".steam/steam", self.user_home / ".steam/root", self.user_home / ".var/app/com.valvesoftware.Steam/.local/share/Steam", - ) - seen = set() - - for candidate in candidates: + ): yield from self._unique_existing_root(candidate, seen) def _steam_library_roots(self): seen = set() - for candidate in self._steam_roots(): - yield from self._unique_existing_root(candidate, seen) - + for root in self._steam_roots(): + yield from self._unique_existing_root(root, seen) for library_file in ( - candidate / "steamapps/libraryfolders.vdf", - candidate / "config/libraryfolders.vdf", + root / "steamapps/libraryfolders.vdf", + root / "config/libraryfolders.vdf", ): try: content = library_file.read_text(encoding="utf-8") except OSError: continue - for raw_path in re.findall(r'(?m)^\s*"path"\s+"((?:\\.|[^"])*)"', content): path = raw_path.replace(r'\"', '"').replace(r'\\', '\\') yield from self._unique_existing_root(Path(path), seen) + @staticmethod + def _unique_existing_root(path: Path, seen: set[str]): + if not path.exists(): + return + try: + resolved = str(path.resolve()) + except OSError: + resolved = str(path) + if resolved not in seen: + seen.add(resolved) + yield path + @staticmethod def _read_shortcuts(data: bytes) -> Dict[str, object]: def read_string(offset: int) -> Tuple[str, int]: @@ -142,16 +123,12 @@ class SteamService(BaseService): value, offset = read_object(offset) elif value_type == 1: value, offset = read_string(offset) - elif value_type == 2: - if offset + 4 > len(data): + elif value_type in (2, 7): + width = 4 if value_type == 2 else 8 + if offset + width > len(data): raise ValueError("truncated binary VDF integer") - value = int.from_bytes(data[offset:offset + 4], "little", signed=True) - offset += 4 - elif value_type == 7: - if offset + 8 > len(data): - raise ValueError("truncated binary VDF 64-bit integer") - value = int.from_bytes(data[offset:offset + 8], "little", signed=True) - offset += 8 + value = int.from_bytes(data[offset:offset + width], "little", signed=True) + offset += width else: raise ValueError(f"unsupported binary VDF type {value_type}") values[key] = value @@ -170,53 +147,28 @@ class SteamService(BaseService): name = shortcut.get("AppName") or shortcut.get("appname") if not isinstance(appid, int) or appid == 0 or not isinstance(name, str) or not name: return None - executable = next( - ( - shortcut.get(key) - for key in ("Exe", "exe", "executable") - if isinstance(shortcut.get(key), str) - ), - None, - ) - launch_options = next( - ( - shortcut.get(key) - for key in ("LaunchOptions", "launchoptions", "launch_options", "arguments") - if isinstance(shortcut.get(key), str) - ), - None, - ) - start_dir = next( - ( - shortcut.get(key) - for key in ("StartDir", "startdir", "start_dir") - if isinstance(shortcut.get(key), str) - ), - None, - ) + executable = _first_string(shortcut, "Exe", "exe", "executable") + arguments = _first_string(shortcut, "LaunchOptions", "launchoptions", "launch_options", "arguments") + start_dir = _first_string(shortcut, "StartDir", "startdir", "start_dir") game: Dict[str, object] = { - "appid": str(appid & 0xffffffff), + "appid": str(appid & 0xFFFFFFFF), "name": name, "nonSteam": True, - "transport": classify_shortcut_transport(executable, launch_options), + "transport": classify_shortcut_transport(executable, arguments), } - if executable is not None: - game["executable"] = executable - if launch_options is not None: - game["arguments"] = launch_options - if start_dir is not None: - game["startDir"] = start_dir + for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)): + if value is not None: + game[key] = value return game def _shortcut_games(self): games = {} - for steam_root in self._steam_roots(): - for shortcuts_file in sorted((steam_root / "userdata").glob("*/config/shortcuts.vdf")): + for root in self._steam_roots(): + for path in sorted((root / "userdata").glob("*/config/shortcuts.vdf")): try: - root = self._read_shortcuts(shortcuts_file.read_bytes()) + shortcuts = self._read_shortcuts(path.read_bytes()).get("shortcuts", {}) except (OSError, ValueError): continue - shortcuts = root.get("shortcuts", {}) if not isinstance(shortcuts, dict): continue for shortcut in shortcuts.values(): @@ -225,25 +177,12 @@ class SteamService(BaseService): games.setdefault(game["appid"], game) return list(games.values()) - @staticmethod - def _unique_existing_root(path: Path, seen: set[str]): - if not path.exists(): - return - try: - resolved = str(path.resolve()) - except OSError: - resolved = str(path) - if resolved in seen: - return - seen.add(resolved) - yield path - def _manifest_path(self) -> Optional[Path]: - for library_root in self._steam_library_roots(): - manifest = library_root / "steamapps" / self.MANIFEST_FILENAME - if manifest.is_file(): - return manifest - return None + return next(( + path + for root in self._steam_library_roots() + if (path := root / "steamapps" / self.MANIFEST_FILENAME).is_file() + ), None) @staticmethod def _section_bounds(content: str, section_name: str) -> Optional[Tuple[int, int, str]]: @@ -253,7 +192,6 @@ class SteamService(BaseService): ) if section is None: return None - depth = 1 in_string = False escaped = False @@ -266,9 +204,7 @@ class SteamService(BaseService): escaped = True elif character == '"': in_string = False - continue - - if character == '"': + elif character == '"': in_string = True elif character == "{": depth += 1 @@ -283,98 +219,82 @@ class SteamService(BaseService): bounds = cls._section_bounds(content, section_name) if bounds is None: return None - body_start, body_end, _ = bounds - pattern = re.compile( - r'(?m)^[ \t]*"(?P[^"]+)"[ \t]+"(?P(?:\\.|[^"\\])*)"' - ) - for match in pattern.finditer(content, body_start, body_end): - if match.group("key") == key: - return match.group("value") - return None + start, end, _ = bounds + pattern = re.compile(r'(?m)^[ \t]*"(?P[^"]+)"[ \t]+"(?P(?:\\.|[^"\\])*)"') + return next(( + match.group("value") + for match in pattern.finditer(content, start, end) + if match.group("key") == key + ), None) @classmethod def _branch_or_default(cls, branch: Optional[str]) -> str: return branch or cls.DEFAULT_BRANCH def _status_fields(self, manifest_path: Path, content: str) -> Dict[str, object]: - selected_branch = self._branch_or_default( - self._section_value(content, "UserConfig", "BetaKey") - ) - current_branch = self._branch_or_default( + selected = self._branch_or_default(self._section_value(content, "UserConfig", "BetaKey")) + current = self._branch_or_default( self._section_value(content, "MountedConfig", "BetaKey") or self._section_value(content, "UserConfig", "BetaKey") ) - needs_switch = ( - selected_branch != STEAM_LOSSLESS_SCALING_BRANCH - or current_branch != STEAM_LOSSLESS_SCALING_BRANCH - ) + needs_switch = selected != STEAM_LOSSLESS_SCALING_BRANCH or current != STEAM_LOSSLESS_SCALING_BRANCH return { "installed": True, "manifest_path": str(manifest_path), - "selected_branch": selected_branch, - "current_branch": current_branch, + "selected_branch": selected, + "current_branch": current, "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, "needs_switch": needs_switch, - "restart_required": ( - selected_branch == STEAM_LOSSLESS_SCALING_BRANCH - and current_branch != STEAM_LOSSLESS_SCALING_BRANCH - ), + "restart_required": selected == STEAM_LOSSLESS_SCALING_BRANCH and current != STEAM_LOSSLESS_SCALING_BRANCH, + } + + @staticmethod + def _missing_branch_fields() -> Dict[str, object]: + return { + "installed": False, + "manifest_path": None, + "selected_branch": None, + "current_branch": None, + "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, + "needs_switch": False, + "restart_required": False, } def find_lsfg_vk_dll(self) -> Optional[str]: - """Find the branch-specific upstream DLL in any Steam library.""" if self.get_branch_status().get("needs_switch"): return None - for library_root in self._steam_library_roots(): - dll_path = library_root / "steamapps/common/Lossless Scaling/lsfg-vk.dll" - if dll_path.is_file(): - return str(dll_path) - return None + return next(( + str(path) + for root in self._steam_library_roots() + if (path := root / "steamapps/common/Lossless Scaling/lsfg-vk.dll").is_file() + ), None) def get_branch_status(self) -> Dict[str, object]: try: - manifest_path = self._manifest_path() - if manifest_path is None: + manifest = self._manifest_path() + if manifest is None: return self._success_response( dict, "Lossless Scaling is not installed through Steam", - installed=False, - manifest_path=None, - selected_branch=None, - current_branch=None, - target_branch=STEAM_LOSSLESS_SCALING_BRANCH, - needs_switch=False, - restart_required=False, + **self._missing_branch_fields(), ) - - content = manifest_path.read_text(encoding="utf-8") - fields = self._status_fields(manifest_path, content) - if not fields["needs_switch"]: - message = "Lossless Scaling is using the lsfg-vk Steam branch" - elif fields["restart_required"]: - message = "lsfg-vk is selected; restart Steam to finish the branch switch" - else: - message = "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas" + fields = self._status_fields(manifest, manifest.read_text(encoding="utf-8")) + message = ( + "Lossless Scaling is using the lsfg-vk Steam branch" + if not fields["needs_switch"] + else "lsfg-vk is selected; restart Steam to finish the branch switch" + if fields["restart_required"] + else "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas" + ) return self._success_response(dict, message, **fields) except Exception as error: - return self._error_response( - dict, - str(error), - installed=False, - manifest_path=None, - selected_branch=None, - current_branch=None, - target_branch=STEAM_LOSSLESS_SCALING_BRANCH, - needs_switch=False, - restart_required=False, - ) + return self._error_response(dict, str(error), **self._missing_branch_fields()) def get_installed_games(self) -> Dict[str, object]: - """Return installed Steam app IDs and names for the Game Mode selector.""" try: games: Dict[str, Dict[str, object]] = {} - for library_root in self._steam_library_roots(): - for manifest in (library_root / "steamapps").glob("appmanifest_*.acf"): + for root in self._steam_library_roots(): + for manifest in (root / "steamapps").glob("appmanifest_*.acf"): match = re.fullmatch(r"appmanifest_(\d+)\.acf", manifest.name) if not match: continue @@ -385,10 +305,9 @@ class SteamService(BaseService): appid = match.group(1) if appid in self.GAME_SELECTOR_EXCLUDED_APPIDS: continue - name = self._section_value(content, "AppState", "name") or f"App {appid}" games[appid] = { "appid": appid, - "name": name, + "name": self._section_value(content, "AppState", "name") or f"App {appid}", "nonSteam": False, "transport": {"kind": "host"}, } diff --git a/py_modules/lsfg_vk/types.py b/py_modules/lsfg_vk/types.py index 7b85708..ce541ec 100644 --- a/py_modules/lsfg_vk/types.py +++ b/py_modules/lsfg_vk/types.py @@ -1,40 +1,17 @@ -""" -Type definitions for the lsfg-vk plugin responses. -""" +from typing import List, Optional, TypedDict -from typing import TypedDict, Optional, List - -class BaseResponse(TypedDict): - """Base response structure""" +class InstallationResponse(TypedDict): success: bool - - -class ErrorResponse(BaseResponse): - """Response structure for errors""" - error: str - - -class MessageResponse(BaseResponse): - """Response structure with message""" - message: str - - -class InstallationResponse(BaseResponse): - """Response for installation operations""" message: str error: Optional[str] -class UninstallationResponse(BaseResponse): - """Response for uninstallation operations""" - message: str +class UninstallationResponse(InstallationResponse): removed_files: Optional[List[str]] - error: Optional[str] class InstallationCheckResponse(TypedDict): - """Response for installation check""" installed: bool lossless_scaling_installed: bool lossless_scaling_status: str -- cgit v1.2.3