diff options
Diffstat (limited to 'py_modules/lsfg_vk')
| -rw-r--r-- | py_modules/lsfg_vk/config_schema.py | 75 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/configuration.py | 76 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/constants.py | 3 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_profile_service.py | 381 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 931 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 94 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 255 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/steam_service.py | 301 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/types.py | 29 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/wrapper_service.py | 224 |
10 files changed, 1231 insertions, 1138 deletions
diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py index ce109f3..bf3e174 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) - if config["active_in"]: + for profile in data.get("profile", []): + name = str(profile.get("name", "")) + config = ConfigurationManager.validate_config({ + **profile, + **global_config, + }) + if name or 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/configuration.py b/py_modules/lsfg_vk/configuration.py index bec3828..17dcaf3 100644 --- a/py_modules/lsfg_vk/configuration.py +++ b/py_modules/lsfg_vk/configuration.py @@ -7,7 +7,7 @@ from .runtime_service import RuntimeService class ConfigurationService(BaseService): - """Controller-facing adapter over upstream lsfg-vk profiles.""" + FLATPAK_PROFILE_PREFIX = "flatpak:" def __init__(self, logger=None, runtime_service: RuntimeService = None): super().__init__(logger) @@ -47,6 +47,13 @@ class ConfigurationService(BaseService): (None, None), ) + @classmethod + def flatpak_profile_name(cls, app_id: str) -> str: + value = str(app_id).strip() + if not value: + raise ValueError("Flatpak application ID is required") + return f"{cls.FLATPAK_PROFILE_PREFIX}{value}" + @staticmethod def _public_config(config: Dict[str, Any]) -> Dict[str, Any]: return ConfigurationManager.validate_config(config) @@ -87,6 +94,64 @@ class ConfigurationService(BaseService): except Exception as error: return self._error_response(dict, str(error), appid=str(appid), config=None) + def get_flatpak_config(self, app_id: str) -> Dict[str, Any]: + try: + data = self._get_profile_data() + name = self.flatpak_profile_name(app_id) + raw = data["profiles"].get(name) + return self._success_response( + dict, + app_id=str(app_id), + profile=name, + exists=raw is not None, + config=self._public_config(raw) if raw is not None else None, + global_config=dict(data["global_config"]), + ) + except Exception as error: + return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False) + + def update_flatpak_config(self, app_id: str, config: Dict[str, Any]) -> Dict[str, Any]: + try: + data = self._get_profile_data() + name = self.flatpak_profile_name(app_id) + merged_config = {**data["global_config"], **config} + if not config.get("dll"): + merged_config["dll"] = data["global_config"].get("dll", "") + validated = self._public_config(merged_config) + validated["active_in"] = [] + data["global_config"] = { + "dll": validated["dll"], + "no_fp16": validated["no_fp16"], + } + data["profiles"][name] = validated + self._save_profile_data(data) + return self._success_response(dict, app_id=str(app_id), profile=name, exists=True, config=validated) + except Exception as error: + return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False) + + def reset_flatpak_config(self, app_id: str) -> Dict[str, Any]: + try: + data = self._get_profile_data() + name = self.flatpak_profile_name(app_id) + data["profiles"].pop(name, None) + self._save_profile_data(data) + return self._success_response(dict, app_id=str(app_id), profile=name, exists=False) + except Exception as error: + return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False) + + def reset_all_flatpak_configs(self) -> Dict[str, Any]: + try: + data = self._get_profile_data() + data["profiles"] = { + name: profile + for name, profile in data["profiles"].items() + if not name.startswith(self.FLATPAK_PROFILE_PREFIX) + } + self._save_profile_data(data) + return self._success_response(dict, global_config=dict(data["global_config"])) + except Exception as error: + return self._error_response(dict, str(error)) + def reset_game_config(self, appid: str) -> Dict[str, Any]: try: data = self._get_profile_data() @@ -101,7 +166,14 @@ class ConfigurationService(BaseService): def reset_all_game_configs(self) -> Dict[str, Any]: try: data = self._get_profile_data() - data["profiles"] = {} + data["profiles"] = { + name: profile + for name, profile in data["profiles"].items() + if not ( + len(profile.get("active_in", [])) == 1 + and re.fullmatch(r"-?[0-9]+", str(profile.get("active_in", [""])[0])) + ) + } self._save_profile_data(data) return self._success_response(dict, global_config=dict(data["global_config"]), games=[]) except Exception as error: diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py index 960d230..45ac4c5 100644 --- a/py_modules/lsfg_vk/constants.py +++ b/py_modules/lsfg_vk/constants.py @@ -16,9 +16,6 @@ CLI_FILENAME = "lsfg-vk-cli" UI_FILENAME = "lsfg-vk-ui" UI_DESKTOP_FILENAME = "gay.pancake.lsfg-vk-ui.desktop" UI_ICON_FILENAME = "gay.pancake.lsfg-vk-ui.png" -FLATPAK_23_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_23.08.flatpak" -FLATPAK_24_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_24.08.flatpak" -FLATPAK_25_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_25.08.flatpak" STEAM_LOSSLESS_SCALING_APP_ID = "993090" STEAM_LOSSLESS_SCALING_BRANCH = "lsfg-vk" diff --git a/py_modules/lsfg_vk/flatpak_profile_service.py b/py_modules/lsfg_vk/flatpak_profile_service.py new file mode 100644 index 0000000..ddec116 --- /dev/null +++ b/py_modules/lsfg_vk/flatpak_profile_service.py @@ -0,0 +1,381 @@ +from __future__ import annotations + +import re +from typing import Any, Dict + +from .configuration import ConfigurationService +from .flatpak_service import FlatpakService + + +class FlatpakProfileService: + STATE_FIELDS = ( + "dxvkFrameRate", + "disableGamescopeWsi", + "disableHdr", + "disableSteamdeckMode", + "disableVkbasalt", + "enableZink", + ) + BOOLEAN_FIELDS = STATE_FIELDS[1:] + DXVK_FRAME_RATE_SEGMENT = re.compile( + r"^(?:dxvk\.maxFrameRate|dxgi\.maxFrameRate|d3d9\.maxFrameRate)\s*=", + re.IGNORECASE, + ) + + def __init__( + self, + flatpak_service: FlatpakService, + configuration_service: ConfigurationService, + ): + self.flatpak_service = flatpak_service + self.configuration_service = configuration_service + + @classmethod + def default_state(cls) -> Dict[str, Any]: + return { + "dxvkFrameRate": 0, + "disableGamescopeWsi": True, + "disableHdr": True, + "disableSteamdeckMode": False, + "disableVkbasalt": False, + "enableZink": False, + } + + @classmethod + def _validate_state(cls, raw: Any) -> Dict[str, Any]: + if not isinstance(raw, dict): + raise ValueError("Workaround state must be an object") + missing = [field for field in cls.STATE_FIELDS if field not in raw] + if missing: + raise ValueError("Workaround state is missing: " + ", ".join(missing)) + state = {field: raw[field] for field in cls.STATE_FIELDS} + frame_rate = state["dxvkFrameRate"] + if isinstance(frame_rate, bool) or not isinstance(frame_rate, int) or not 0 <= frame_rate <= 60: + raise ValueError("Base FPS Cap must be an integer from 0 to 60") + for field in cls.BOOLEAN_FIELDS: + if type(state[field]) is not bool: + raise ValueError(f"{field} must be a boolean") + return state + + def _state_entry(self, app_id: str) -> tuple[Dict[str, object], Dict[str, Any]]: + state = self.flatpak_service._read_state() + entry = state["prepared_apps"].get(app_id) + if not isinstance(entry, dict): + raise RuntimeError("Flatpak application is not owned by this plugin") + return state, entry + + def _baseline_content(self, app_id: str, entry: Dict[str, Any]) -> str: + if not entry.get("override_existed"): + return "" + path = self.flatpak_service._backup_path(app_id) + if path.is_symlink() or not path.is_file(): + raise RuntimeError("Flatpak override backup is unavailable") + return path.read_text(encoding="utf-8") + + @staticmethod + def _environment_value(content: str, key: str) -> str: + section = None + for raw_line in content.splitlines(): + line = raw_line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1] + continue + if section != "Environment": + continue + name, separator, value = line.partition("=") + if separator and name == key: + return value + return "" + + @classmethod + def _dxvk_config(cls, baseline: str, frame_rate: int) -> str: + existing = cls._environment_value(baseline, "DXVK_CONFIG") + parts = [part.strip() for part in existing.split(";") if part.strip()] + parts = [part for part in parts if not cls.DXVK_FRAME_RATE_SEGMENT.match(part)] + if frame_rate > 0: + parts.append(f"dxvk.maxFrameRate = {frame_rate}") + return "; ".join(parts) + + def _restore_baseline(self, app_id: str, entry: Dict[str, Any]) -> None: + existed, current = self.flatpak_service._snapshot_override(app_id) + current_hash = self.flatpak_service._sha256(current) if existed else self.flatpak_service._sha256(b"") + if current_hash != entry.get("managed_sha256"): + raise RuntimeError("Flatpak override changed after preparation; refusing to overwrite unrelated settings") + path = self.flatpak_service._override_path(app_id) + if entry.get("override_existed"): + self.flatpak_service._write_file(path, self._baseline_content(app_id, entry)) + else: + path.unlink(missing_ok=True) + + def _apply_state(self, app_id: str, workaround_state: Dict[str, Any]) -> Dict[str, Any]: + workaround_state = self._validate_state(workaround_state) + _, entry = self._state_entry(app_id) + baseline = self._baseline_content(app_id, entry) + self._restore_baseline(app_id, entry) + prepared = self.flatpak_service.prepare_app(app_id) + if not prepared.get("success") or not prepared.get("owned"): + raise RuntimeError(prepared.get("error") or "Could not restore plugin-owned Flatpak preparation") + profile = self.configuration_service.flatpak_profile_name(app_id) + args = [ + "override", + "--user", + f"--env=LSFGVK_PROFILE={profile}", + "--unset-env=DISABLE_LSFGVK", + "--unset-env=DISABLE_LSFG", + ] + if workaround_state["disableGamescopeWsi"]: + args.extend(["--env=ENABLE_GAMESCOPE_WSI=0", "--unset-env=DISABLE_GAMESCOPE_WSI"]) + if workaround_state["disableHdr"]: + args.append("--env=DXVK_HDR=0") + if workaround_state["disableSteamdeckMode"]: + args.append("--env=SteamDeck=0") + if workaround_state["disableVkbasalt"]: + args.extend(["--env=DISABLE_VKBASALT=1", "--unset-env=ENABLE_VKBASALT"]) + if workaround_state["enableZink"]: + args.extend([ + "--env=__GLX_VENDOR_LIBRARY_NAME=mesa", + "--env=MESA_LOADER_DRIVER_OVERRIDE=zink", + "--env=GALLIUM_DRIVER=zink", + ]) + dxvk_config = self._dxvk_config(baseline, workaround_state["dxvkFrameRate"]) + if dxvk_config: + args.append(f"--env=DXVK_CONFIG={dxvk_config}") + args.append(app_id) + result = self.flatpak_service._run_flatpak_command(args, capture_output=True, text=True) + if result.returncode != 0: + raise OSError(result.stderr.strip() or f"Could not apply Flatpak workarounds for {app_id}") + existed, managed = self.flatpak_service._snapshot_override(app_id) + if not existed: + raise RuntimeError(f"Flatpak override for {app_id} was not created") + state = self.flatpak_service._read_state() + entry = state["prepared_apps"].get(app_id) + if not isinstance(entry, dict): + raise RuntimeError("Flatpak application ownership state disappeared") + entry["managed_sha256"] = self.flatpak_service._sha256(managed) + entry["workaround_state"] = workaround_state + self.flatpak_service._write_state(state) + return workaround_state + + def enable_app(self, app_id: str) -> Dict[str, Any]: + created_profile = False + newly_owned = False + try: + existing = self.configuration_service.get_flatpak_config(app_id) + before = self.flatpak_service._read_state() + was_owned = app_id in before["prepared_apps"] + prepared = self.flatpak_service.prepare_app(app_id) + if not prepared.get("success"): + raise RuntimeError(prepared.get("error") or "Could not prepare Flatpak application") + if not prepared.get("owned"): + raise RuntimeError("Flatpak application is prepared outside this plugin and cannot be managed safely") + newly_owned = not was_owned + if not existing.get("exists"): + config = { + **self.configuration_service._public_config({}), + **(existing.get("global_config") or {}), + } + config["active_in"] = [] + saved = self.configuration_service.update_flatpak_config(app_id, config) + if not saved.get("success"): + raise RuntimeError(saved.get("error") or "Could not create Flatpak profile") + created_profile = True + _, entry = self._state_entry(app_id) + workaround_state = self._validate_state(entry.get("workaround_state", self.default_state())) + self._apply_state(app_id, workaround_state) + return self.get_app(app_id) + except Exception as error: + if created_profile: + self.configuration_service.reset_flatpak_config(app_id) + if newly_owned: + self.flatpak_service.remove_app_override(app_id) + return { + "success": False, + "message": "", + "error": str(error), + "app_id": str(app_id), + "enabled": False, + } + + def update_config(self, app_id: str, config: Dict[str, Any]) -> Dict[str, Any]: + try: + self._state_entry(app_id) + result = self.configuration_service.update_flatpak_config(app_id, config) + if not result.get("success"): + raise RuntimeError(result.get("error") or "Could not update Flatpak profile") + return self.get_app(app_id) + except Exception as error: + return { + "success": False, + "message": "", + "error": str(error), + "app_id": str(app_id), + "enabled": False, + } + + def get_workaround_state(self, app_id: str) -> Dict[str, Any]: + try: + _, entry = self._state_entry(app_id) + state = self._validate_state(entry.get("workaround_state", self.default_state())) + return { + "success": True, + "message": "", + "error": None, + "app_id": app_id, + "state": state, + } + except Exception as error: + return { + "success": False, + "message": "", + "error": str(error), + "app_id": str(app_id), + "state": None, + } + + def set_workaround_state(self, app_id: str, workaround_state: Dict[str, Any]) -> Dict[str, Any]: + try: + state = self._apply_state(app_id, workaround_state) + return { + "success": True, + "message": "", + "error": None, + "app_id": app_id, + "state": state, + } + except Exception as error: + return { + "success": False, + "message": "", + "error": str(error), + "app_id": str(app_id), + "state": None, + } + + def remove_app(self, app_id: str) -> Dict[str, Any]: + try: + removed = self.flatpak_service.remove_app_override(app_id) + if not removed.get("success"): + raise RuntimeError(removed.get("error") or "Could not remove Flatpak preparation") + reset = self.configuration_service.reset_flatpak_config(app_id) + if not reset.get("success"): + raise RuntimeError(reset.get("error") or "Could not remove Flatpak profile") + return { + "success": True, + "message": "Flatpak profile removed", + "error": None, + "app_id": app_id, + "enabled": False, + } + except Exception as error: + return { + "success": False, + "message": "", + "error": str(error), + "app_id": str(app_id), + "enabled": True, + } + + def get_app(self, app_id: str) -> Dict[str, Any]: + apps = self.get_apps() + if not apps.get("success"): + return { + "success": False, + "message": "", + "error": apps.get("error"), + "app_id": str(app_id), + "enabled": False, + } + app = next((item for item in apps.get("apps", []) if item.get("app_id") == app_id), None) + if app is None: + return { + "success": False, + "message": "", + "error": "Flatpak application is not installed", + "app_id": str(app_id), + "enabled": False, + } + return {"success": True, "message": "", "error": None, **app} + + def get_apps(self) -> Dict[str, Any]: + try: + result = self.flatpak_service.get_flatpak_apps() + if not result.get("success"): + raise RuntimeError(result.get("error") or "Could not list Flatpak applications") + ownership = self.flatpak_service._read_state() + apps = [] + for item in result.get("apps", []): + app_id = item["app_id"] + config_result = self.configuration_service.get_flatpak_config(app_id) + entry = ownership["prepared_apps"].get(app_id) + workarounds = self.default_state() + if isinstance(entry, dict): + workarounds = self._validate_state(entry.get("workaround_state", workarounds)) + profile = self.configuration_service.flatpak_profile_name(app_id) + selector_ready = False + if item.get("prepared"): + shown = self.flatpak_service._run_flatpak_command( + ["override", "--user", "--show", app_id], + capture_output=True, + text=True, + ) + selector_ready = shown.returncode == 0 and f"LSFGVK_PROFILE={profile}" in shown.stdout.splitlines() + apps.append({ + **item, + "profile": profile, + "enabled": bool(item.get("owned") and config_result.get("exists") and selector_ready), + "config": config_result.get("config"), + "workarounds": workarounds, + }) + return { + "success": True, + "message": result.get("message", ""), + "error": None, + "apps": apps, + } + except Exception as error: + return {"success": False, "message": "", "error": str(error), "apps": []} + + def get_running_apps(self) -> Dict[str, Any]: + try: + state = self.flatpak_service._read_state() + enabled = set() + for app_id, entry in state["prepared_apps"].items(): + if not isinstance(entry, dict): + continue + config = self.configuration_service.get_flatpak_config(app_id) + if not config.get("exists"): + continue + existed, content = self.flatpak_service._snapshot_override(app_id) + if not existed or self.flatpak_service._sha256(content) != entry.get("managed_sha256"): + continue + profile = self.configuration_service.flatpak_profile_name(app_id) + try: + text = content.decode("utf-8") + except UnicodeDecodeError: + continue + if self._environment_value(text, "LSFGVK_PROFILE") == profile: + enabled.add(app_id) + if not enabled: + return {"success": True, "message": "", "error": None, "apps": []} + result = self.flatpak_service._run_flatpak_command( + ["ps", "--columns=application,active,pid"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Could not inspect running Flatpak applications") + running = [] + for line in result.stdout.splitlines(): + fields = line.split("\t") if "\t" in line else line.split() + if not fields or fields[0] not in enabled: + continue + active = len(fields) > 1 and fields[1].strip().lower() in {"1", "true", "yes", "active"} + running.append({ + "app_id": fields[0], + "active": active, + "pid": fields[2].strip() if len(fields) > 2 else "", + }) + running.sort(key=lambda item: (not item["active"], item["app_id"])) + return {"success": True, "message": "", "error": None, "apps": running} + except Exception as error: + return {"success": False, "message": "", "error": str(error), "apps": []} diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index c0a9c0c..4f04382 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,7 +1,6 @@ -"""Flatpak runtime-extension infrastructure for unified game targets.""" - from __future__ import annotations +import hashlib import json import os import pwd @@ -10,77 +9,68 @@ 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 ( - BIN_DIR, - FLATPAK_23_08_FILENAME, - FLATPAK_24_08_FILENAME, - FLATPAK_25_08_FILENAME, -) 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" - OWNERSHIP_VERSION = 1 + FLATHUB_REMOTE = "flathub" + SUPPORTED_RUNTIMES = ("24.08", "25.08") + DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"} + RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL" + OWNERSHIP_FILENAME = "flatpak_state.json" + OWNERSHIP_VERSION = 2 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) self.flatpak_command: Optional[str] = None + self._verified_branches: Set[str] = set() self._lock = threading.RLock() @property def ownership_path(self) -> Path: return self.config_dir / self.OWNERSHIP_FILENAME - def _get_clean_env(self) -> Dict[str, str]: + @property + def backup_dir(self) -> Path: + return self.config_dir / "flatpak-overrides" + + 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,150 +79,134 @@ 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("/") - 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) + def runtime_branch_from_metadata(cls, metadata: str) -> str: + section = None + versions = [] + for raw_line in metadata.splitlines() if isinstance(metadata, str) else []: + line = raw_line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1].strip() + continue + if section != cls.RUNTIME_METADATA_SECTION: + continue + key, separator, value = line.partition("=") + if separator and key.strip() == "versions": + versions.extend(part.strip() for part in value.split(";")) + for value in versions: + for branch in cls.SUPPORTED_RUNTIMES: + if value == branch or value.startswith(f"{branch}-"): + return branch + raise ValueError("Could not determine a supported Freedesktop base runtime from Flatpak metadata") @classmethod - def _bundle_filename(cls, version: str) -> str: - return { - "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) - ) + def _extension_ref(cls, branch: str) -> str: + return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}" + + def _installed_extension_branches(self, scope: Optional[str] = None) -> Set[str]: + scopes = ("user", "system") if scope is None else (scope,) + installed = set() + for item in scopes: + result = self._run_flatpak_command( + ["list", f"--{item}", "--runtime", "--columns=application,arch,branch"], + capture_output=True, + text=True, + check=True, + ) + for line in result.stdout.splitlines(): + 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 _installed_extension_branches(self) -> Set[str]: + def _user_extension_origin(self, branch: str) -> str: result = self._run_flatpak_command( - ["list", "--runtime", "--columns=application,arch,branch"], + ["info", "--user", "--show-origin", self._extension_ref(branch)], capture_output=True, text=True, - check=True, ) - installed: Set[str] = 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) - return installed + return result.stdout.strip() if result.returncode == 0 else "" - def _read_owned_branches(self) -> Tuple[Set[str], bool]: - """Read ownership without guessing when metadata is damaged.""" - 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 - 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 - except (OSError, json.JSONDecodeError, TypeError, ValueError) as error: - self.log.warning(f"Could not trust Flatpak ownership metadata: {error}") - return set(), True - - 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() - return - document = { + def _empty_state(self) -> Dict[str, object]: + return { "version": self.OWNERSHIP_VERSION, - "plugin_owned_branches": sorted(branches), + "plugin_owned_branches": [], + "prepared_apps": {}, } - self._write_file(self.ownership_path, json.dumps(document, indent=2) + "\n") - def get_extension_status(self) -> Dict[str, Any]: - """Return global extension inventory for Setup diagnostics.""" + def _read_state(self) -> Dict[str, object]: + if not self.ownership_path.exists(): + return self._empty_state() + if self.ownership_path.is_symlink() or not self.ownership_path.is_file(): + raise RuntimeError("Flatpak ownership metadata is not a regular file") 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=[], - owned_branches=[], - ownership_uncertain=False, - ) - installed = self._installed_extension_branches() - owned, uncertain = self._read_owned_branches() - return self._success_response( - dict, - "Flatpak runtime extension status retrieved", - available=True, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=sorted(installed), - owned_branches=sorted(owned), - ownership_uncertain=uncertain, - ) - except Exception as error: - return self._error_response( - dict, - str(error), - available=self.check_flatpak_available(), - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=[], - owned_branches=[], - ownership_uncertain=False, - ) + data = json.loads(self.ownership_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise RuntimeError(f"Could not read Flatpak ownership metadata: {error}") from error + if data.get("version") != self.OWNERSHIP_VERSION: + raise RuntimeError("Unsupported Flatpak ownership metadata version") + branches = data.get("plugin_owned_branches") + apps = data.get("prepared_apps") + if not isinstance(branches, list) or not isinstance(apps, dict): + raise RuntimeError("Invalid Flatpak ownership metadata") + for branch in branches: + self._validate_runtime(branch) + for app_id, entry in apps.items(): + self._validate_app_id(app_id) + if not isinstance(entry, dict): + raise RuntimeError("Invalid Flatpak app ownership metadata") + if type(entry.get("override_existed")) is not bool: + raise RuntimeError("Invalid Flatpak app ownership metadata") + if not isinstance(entry.get("managed_sha256"), str): + raise RuntimeError("Invalid Flatpak app ownership metadata") + return data + + def _write_state(self, state: Dict[str, object]) -> None: + branches = state.get("plugin_owned_branches", []) + apps = state.get("prepared_apps", {}) + if not branches and not apps: + self.ownership_path.unlink(missing_ok=True) + if self.backup_dir.exists() and not any(self.backup_dir.iterdir()): + self.backup_dir.rmdir() + return + self._write_file( + self.ownership_path, + json.dumps(state, indent=2, sort_keys=True) + "\n", + ) + + def _owned_branches(self, state: Optional[Dict[str, object]] = None) -> Set[str]: + current = state if state is not None else self._read_state() + return {self._validate_runtime(branch) for branch in current["plugin_owned_branches"]} - def get_flatpak_support_status(self) -> Dict[str, Any]: - return self.get_extension_status() + def _override_path(self, app_id: str) -> Path: + return self.user_home / ".local/share/flatpak/overrides" / self._validate_app_id(app_id) + + def _backup_path(self, app_id: str) -> Path: + return self.backup_dir / f"{self._validate_app_id(app_id)}.ini" + + @staticmethod + def _sha256(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + def _snapshot_override(self, app_id: str) -> tuple[bool, bytes]: + path = self._override_path(app_id) + if path.is_symlink(): + raise RuntimeError("Flatpak override path is a symlink") + if not path.exists(): + return False, b"" + if not path.is_file(): + raise RuntimeError("Flatpak override path is not a regular file") + return True, path.read_bytes() - def _resolve_runtime(self, app_id: str) -> Dict[str, Any]: + def _resolve_runtime(self, app_id: str) -> tuple[str, str]: self._validate_app_id(app_id) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") @@ -243,349 +217,416 @@ 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 "" + parts = runtime.split("/") + if len(parts) != 3: + raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") + if parts[0] == "org.freedesktop.Platform": + return runtime, self._validate_runtime(parts[2]) + if parts[0] not in self.DERIVED_RUNTIME_IDS: + raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") + metadata_result = self._run_flatpak_command( + ["info", "--show-metadata", runtime], + capture_output=True, + text=True, + ) + if metadata_result.returncode != 0: + raise OSError(metadata_result.stderr.strip() or f"Could not inspect Flatpak runtime {runtime}") + return runtime, self.runtime_branch_from_metadata(metadata_result.stdout) + + def _dll_directory(self) -> Path: + if self.config_file_path.exists(): + try: + content = self.config_file_path.read_text(encoding="utf-8") + match = re.search( + r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"', + content, + ) + if match: + configured_dll = json.loads('"' + match.group(1) + '"') + if configured_dll: + return Path(configured_dll).parent + except Exception: + pass + return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" + + def _filesystem_present(self, entries: str, host_path: Path) -> bool: + accepted = {str(host_path)} + try: + accepted.add(f"~/{host_path.relative_to(self.user_home).as_posix()}") + except ValueError: + pass + enabled = False + for raw in entries.split(";"): + value = raw.strip() + if not value: + continue + denied = value.startswith("!") + path = value[1:] if denied else value + path = path.split(":", 1)[0] + if path in accepted: + if denied: + return False + enabled = True + return enabled + + def _app_override_status(self, app_id: str) -> Dict[str, object]: + result = self._run_flatpak_command( + ["override", "--user", "--show", app_id], + capture_output=True, + text=True, + ) + output = result.stdout if result.returncode == 0 else "" + section = None + filesystems = "" + unset_environment = set() + environment = {} + for raw_line in output.splitlines(): + line = raw_line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1] + continue + key, separator, value = line.partition("=") + if not separator: + continue + if section == "Context" and key == "filesystems": + filesystems = value + elif section == "Context" and key == "unset-environment": + unset_environment.update(item for item in value.split(";") if item) + elif section == "Environment": + environment[key] = value + config_ready = self._filesystem_present(filesystems, self.config_dir) + dll_ready = self._filesystem_present(filesystems, self._dll_directory()) + env_ready = ( + environment.get("LSFGVK_CONFIG") == str(self.config_file_path) + and environment.get("LSFGVK_FLATPAK") == "1" + and "DISABLE_LSFGVK" in unset_environment + and "DISABLE_LSFG" in unset_environment + ) + return { + "filesystem_ready": config_ready and dll_ready, + "environment_ready": env_ready, + "prepared": config_ready and dll_ready and env_ready, + } - def resolve_app_support(self, app_id: str) -> Dict[str, Any]: - """Resolve the exact runtime branch required by one Flatpak app.""" + def get_extension_status(self): try: - app_id = self._validate_app_id(app_id) - resolved = self._resolve_runtime(app_id) - installed = self._installed_extension_branches() - branch = resolved["runtime_branch"] - ready = branch in installed + available = self.check_flatpak_available() + installed = self._installed_extension_branches() if available else set() 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}" - ), - flatpak_app_id=app_id, - runtime=resolved["runtime"], - runtime_branch=branch, - support_status="ready" if ready else "needs-runtime", - extension_installed=ready, + "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), ) - except ValueError as error: - return self._success_response( - dict, - str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="unsupported", - extension_installed=False, - installed_branches=[], - error=str(error), - ) except Exception as error: return self._error_response( dict, str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="error", - extension_installed=False, + available=False, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], ) - def install_extension(self, version: str) -> Dict[str, Any]: - """Install one missing branch and record ownership only after readback.""" + get_flatpak_support_status = get_extension_status + + 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: - owned, uncertain = self._read_owned_branches() - if uncertain: - raise RuntimeError( - "Flatpak ownership metadata is uncertain; refusing to install " - "until it is repaired" - ) - 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, - owned_by_plugin=version in owned, - preserved=version not in owned, - ) - 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._verified_branches: + return self._extension_result(branch, True, False, "is ready") + user_installed = self._installed_extension_branches("user") + system_installed = self._installed_extension_branches("system") + if branch in system_installed and branch not in user_installed: + return self._extension_result(branch, True, False, "is ready") + if branch in user_installed and self._user_extension_origin(branch) != self.FLATHUB_REMOTE: + 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 "Could not replace the existing Flatpak extension") result = self._run_flatpak_command( [ "install", "--user", "--noninteractive", "--or-update", - str(bundle_path), + self.FLATHUB_REMOTE, + f"{self.EXTENSION_ID}//{branch}", ], 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.add(version) - self._write_owned_branches(owned) - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension installed", - runtime_branch=version, - installed=True, - enabled=True, - owned_by_plugin=True, - preserved=False, - ) + if branch not in self._installed_extension_branches("user"): + raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards") + state = self._read_state() + owned = self._owned_branches(state) + owned.add(branch) + state["plugin_owned_branches"] = sorted(owned) + self._write_state(state) + self._verified_branches.add(branch) + return self._extension_result(branch, True, False, "installed") except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=version, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, - ) + return self._error_response(dict, str(error), runtime_branch=branch, 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", - ) - 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, - owned_by_plugin=version in status.get("owned_branches", []), - ) - return self.install_extension(version) - - def ensure_app_support(self, app_id: str) -> Dict[str, Any]: - """Provision only the branch returned by flatpak info for this app.""" - 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) - 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, - 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( + def _remove_extension(self, branch: str) -> bool: + if branch not in self._installed_extension_branches("user"): + 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") + return True + + def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str): + return self._success_response( dict, - final.get("error") or "Required Flatpak runtime extension could not be verified", - flatpak_app_id=app_id, - runtime=resolved.get("runtime"), + f"lsfg-vk {branch} runtime extension {verb}", runtime_branch=branch, - support_status="error", - extension_installed=False, + installed=installed, + enabled=installed, + removed=removed, ) - def uninstall_extension(self, version: str) -> Dict[str, Any]: - """Uninstall only when explicitly requested for a plugin-owned branch.""" + def uninstall_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: - owned, uncertain = self._read_owned_branches() - if uncertain: - raise RuntimeError( - "Flatpak ownership metadata is uncertain; refusing to uninstall" - ) - installed = self._installed_extension_branches() - if version not in owned: - if version in installed: - return self._success_response( - dict, - f"Preserved Flatpak extension {version}; it is not plugin-owned", - runtime_branch=version, - removed=False, - installed=True, - enabled=True, - owned_by_plugin=False, - preserved=True, - ) + state = self._read_state() + owned = self._owned_branches(state) + if branch not in owned: + installed = branch in self._installed_extension_branches() + return self._extension_result(branch, installed, False, "preserved (not plugin-owned)") + removed = self._remove_extension(branch) + owned.remove(branch) + state["plugin_owned_branches"] = sorted(owned) + self._write_state(state) + installed = branch in self._installed_extension_branches() + return self._extension_result(branch, installed, removed, "uninstalled") + except Exception as error: + return self._error_response(dict, str(error), runtime_branch=branch, removed=False, installed=False, enabled=False) + + def ensure_extension(self, branch: str): + return self.install_extension(branch) + + 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=branch, installed=False, enabled=False) + return self.install_extension(branch) if enabled else self.uninstall_extension(branch) + + def get_flatpak_apps(self): + try: + if not self.check_flatpak_available(): + raise FileNotFoundError("Flatpak is not available on this system") + installed_extensions = self._installed_extension_branches() + state = self._read_state() + owned_apps = state["prepared_apps"] + result = self._run_flatpak_command( + ["list", "--app", "--columns=name,application"], + capture_output=True, + text=True, + check=True, + ) + apps = [] + for line in result.stdout.splitlines(): + fields = line.split("\t") + if len(fields) < 2: + continue + name, app_id = fields[0].strip(), fields[1].strip() + if not app_id: + continue + item = { + "app_id": app_id, + "app_name": name or app_id, + "runtime": None, + "runtime_branch": None, + "runtime_ready": False, + "prepared": False, + "owned": app_id in owned_apps, + "error": None, + } + try: + runtime, branch = self._resolve_runtime(app_id) + status = self._app_override_status(app_id) + item.update({ + "runtime": runtime, + "runtime_branch": branch, + "runtime_ready": branch in installed_extensions, + "prepared": status["prepared"], + }) + except Exception as error: + item["error"] = str(error) + apps.append(item) + apps.sort(key=lambda item: str(item["app_name"]).lower()) + return self._success_response(dict, f"Found {len(apps)} Flatpak applications", apps=apps) + except Exception as error: + return self._error_response(dict, str(error), apps=[]) + + def prepare_app(self, app_id: str): + try: + app_id = self._validate_app_id(app_id) + with self._lock: + runtime, branch = self._resolve_runtime(app_id) + extension = self.ensure_extension(branch) + if not extension.get("success") or not extension.get("installed"): + raise RuntimeError(extension.get("error") or f"Could not install Flatpak runtime {branch}") + state = self._read_state() + apps = state["prepared_apps"] + status = self._app_override_status(app_id) + if status["prepared"] and app_id not in apps: return self._success_response( dict, - f"Flatpak extension {version} is already not installed", - runtime_branch=version, - removed=False, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, + "Flatpak application is already prepared outside this plugin", + app_id=app_id, + runtime=runtime, + runtime_branch=branch, + prepared=True, + owned=False, ) - if version in 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.remove(version) - self._write_owned_branches(owned) + if app_id not in apps: + existed, original = self._snapshot_override(app_id) + backup = self._backup_path(app_id) + if existed: + self._write_file(backup, original.decode("utf-8")) + else: + backup.unlink(missing_ok=True) + apps[app_id] = { + "override_existed": existed, + "managed_sha256": "", + } + result = self._run_flatpak_command( + [ + "override", + "--user", + f"--filesystem={self.config_dir}:ro", + f"--filesystem={self._dll_directory()}:ro", + f"--env=LSFGVK_CONFIG={self.config_file_path}", + "--env=LSFGVK_FLATPAK=1", + "--unset-env=DISABLE_LSFGVK", + "--unset-env=DISABLE_LSFG", + app_id, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or f"Could not prepare Flatpak app {app_id}") + status = self._app_override_status(app_id) + if not status["prepared"]: + raise RuntimeError(f"Flatpak preparation did not become visible for {app_id}") + existed, managed = self._snapshot_override(app_id) + if not existed: + raise RuntimeError(f"Flatpak override for {app_id} was not created") + apps[app_id]["managed_sha256"] = self._sha256(managed) + self._write_state(state) return self._success_response( dict, - f"Plugin-owned lsfg-vk {version} runtime extension removed", - runtime_branch=version, - removed=True, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, + "Flatpak application prepared for lsfg-vk", + app_id=app_id, + runtime=runtime, + runtime_branch=branch, + prepared=True, + owned=True, ) except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=version, - removed=False, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, - ) - - def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: - """Set one runtime branch to the requested state, safely and idempotently.""" - if type(enabled) is not bool: - return self._error_response( - dict, - "enabled must be a boolean", - runtime_branch=version, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, - ) - return self.install_extension(version) if enabled else self.uninstall_extension(version) + return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=False) - def remove_plugin_owned_extensions(self) -> Dict[str, Any]: - """Uninstall only branches recorded as installed by this plugin.""" + def remove_app_override(self, app_id: str): try: + app_id = self._validate_app_id(app_id) 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, - ) - if not owned: + state = self._read_state() + apps = state["prepared_apps"] + entry = apps.get(app_id) + if entry is None: return self._success_response( dict, - "No plugin-owned Flatpak extensions to remove", - removed_branches=[], - preserved_branches=[], - ownership_uncertain=False, + "Flatpak application is not plugin-owned; existing overrides were preserved", + app_id=app_id, + prepared=self._app_override_status(app_id)["prepared"], + owned=False, ) - if not self.check_flatpak_available(): + existed, current = self._snapshot_override(app_id) + current_hash = self._sha256(current) if existed else self._sha256(b"") + if current_hash != entry["managed_sha256"]: + raise RuntimeError( + "Flatpak override changed after preparation; refusing to overwrite unrelated settings" + ) + override_path = self._override_path(app_id) + backup_path = self._backup_path(app_id) + if entry["override_existed"]: + if not backup_path.is_file() or backup_path.is_symlink(): + raise RuntimeError("Flatpak override backup is unavailable") + self._write_file(override_path, backup_path.read_text(encoding="utf-8")) + else: + override_path.unlink(missing_ok=True) + backup_path.unlink(missing_ok=True) + apps.pop(app_id, None) + self._write_state(state) + return self._success_response( + dict, + "Plugin-owned Flatpak preparation removed", + app_id=app_id, + prepared=False, + owned=False, + ) + except Exception as error: + return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=True) + + def remove_plugin_owned_environment(self): + try: + with self._lock: + state = self._read_state() + failures = [] + removed_apps = [] + for app_id in list(state["prepared_apps"]): + result = self.remove_app_override(app_id) + if result.get("success"): + removed_apps.append(app_id) + else: + failures.append(f"{app_id}: {result.get('error')}") + if failures: return self._error_response( dict, - "Flatpak is not available; plugin-owned extension metadata was preserved", + "; ".join(failures), + removed_apps=removed_apps, removed_branches=[], - preserved_branches=sorted(owned), - ownership_uncertain=False, ) - removed: List[str] = [] - failures: List[str] = [] - 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" - ) - removed.append(branch) - except Exception as error: - failures.append(f"{branch}: {error}") - remaining = owned - set(removed) - self._write_owned_branches(remaining) + state = self._read_state() + removed_branches = [] + for branch in sorted(self._owned_branches(state)): + result = self.uninstall_extension(branch) + if result.get("success"): + removed_branches.append(branch) + else: + failures.append(f"{branch}: {result.get('error')}") if failures: return self._error_response( dict, "; ".join(failures), - removed_branches=removed, - preserved_branches=sorted(remaining), - ownership_uncertain=False, + removed_apps=removed_apps, + removed_branches=removed_branches, ) return self._success_response( dict, - "Plugin-owned Flatpak extensions removed", - removed_branches=removed, - preserved_branches=[], - ownership_uncertain=False, + "Plugin-owned Flatpak state removed", + removed_apps=removed_apps, + removed_branches=removed_branches, ) except Exception as error: - return self._error_response( - dict, - str(error), - removed_branches=[], - preserved_branches=[], - ownership_uncertain=False, - ) + return self._error_response(dict, str(error), removed_apps=[], removed_branches=[]) diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 8a3094d..a73706c 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,33 @@ 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 _default_config(self) -> ProfileData: + defaults = ConfigurationManager.get_defaults() + return ProfileData( + profiles={}, + global_config={"dll": defaults["dll"], "no_fp16": defaults["no_fp16"]}, + ) + 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) - else: - default = dict(ConfigurationManager.get_defaults()) - profile_data = ProfileData( - profiles={}, - global_config={ - "dll": default.get("dll", ""), - "no_fp16": default.get("no_fp16", False), - }, + try: + profile_data = ( + ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) + if self.config_file_path.exists() + else self._default_config() ) - + except ValueError: + profile_data = self._default_config() 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 +152,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 +170,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 +187,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 +218,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 76a4250..d20fabf 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -1,34 +1,19 @@ -""" -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 +from typing import Any, Dict import decky -from .installation import InstallationService from .configuration import ConfigurationService +from .flatpak_profile_service import FlatpakProfileService 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( @@ -37,198 +22,180 @@ class Plugin: ) self.configuration_service = ConfigurationService(runtime_service=self.runtime_service) self.flatpak_service = FlatpakService() + self.flatpak_profile_service = FlatpakProfileService( + self.flatpak_service, + self.configuration_service, + ) 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): + flatpak = self.flatpak_service.remove_plugin_owned_environment() + if not flatpak.get("success"): + return { + "success": False, + "message": "", + "error": flatpak.get("error") or "Could not clean up Flatpak support", + "removed_files": None, + } + self.configuration_service.reset_all_flatpak_configs() 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]: - result = self.steam_service.get_installed_games() - if not result.get("success"): - return result - - support_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": - continue - flatpak_app_id = transport.get("flatpakAppId") - if not isinstance(flatpak_app_id, str) or not flatpak_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] - return result - - async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]: + async def get_installed_games(self): + return self.steam_service.get_installed_games() + + 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( self, appid: str, state: Dict[str, Any], - 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, - shortcut_exe, - command_token_added, - transport, - ) + ): + return self.wrapper_service.set(appid, state, command_token_added) - 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_debug_file_contents(self): + files = ( + ("config", "LSFG-VK configuration", self.configuration_service.config_file_path), + ("workarounds", "Per-app workarounds", self.wrapper_service.sidecar_path), + ("wrapper", "Generated launch wrapper", self.wrapper_service.wrapper_path), + ("flatpak", "Flatpak ownership state", self.flatpak_service.ownership_path), + ) + contents = [] + for file_id, label, path in files: + item = { + "id": file_id, + "label": label, + "path": str(path), + "exists": False, + "content": None, + "error": None, + } + try: + if path.is_symlink(): + item["error"] = "Path is a symlink; refusing to read it" + elif not path.exists(): + item["error"] = "File does not exist" + elif not path.is_file(): + item["error"] = "Path is not a regular file" + else: + item["exists"] = True + item["content"] = path.read_text(encoding="utf-8") + except Exception as error: + item["error"] = f"Error reading file: {error}" + contents.append(item) + return { + "success": True, + "message": "Debug file contents retrieved", + "error": None, + "files": contents, + } + + 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]: - return self.flatpak_service.get_flatpak_support_status() + async def get_flatpak_apps(self): + return self.flatpak_profile_service.get_apps() + + async def enable_flatpak_app(self, flatpak_app_id: str): + return self.flatpak_profile_service.enable_app(flatpak_app_id) + + async def update_flatpak_config(self, flatpak_app_id: str, config: Dict[str, Any]): + return self.flatpak_profile_service.update_config(flatpak_app_id, config) + + async def get_flatpak_workaround_state(self, flatpak_app_id: str): + return self.flatpak_profile_service.get_workaround_state(flatpak_app_id) - async def ensure_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]: - return self.flatpak_service.ensure_app_support(flatpak_app_id) + async def set_flatpak_workaround_state(self, flatpak_app_id: str, state: Dict[str, Any]): + return self.flatpak_profile_service.set_workaround_state(flatpak_app_id, state) - async def repair_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]: - return self.flatpak_service.ensure_app_support(flatpak_app_id) + async def remove_flatpak_app(self, flatpak_app_id: str): + return self.flatpak_profile_service.remove_app(flatpak_app_id) - async def set_flatpak_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: - return self.flatpak_service.set_extension_enabled(version, enabled) + async def get_running_flatpak_apps(self): + return self.flatpak_profile_service.get_running_apps() - async def remove_plugin_owned_flatpak_extensions(self) -> Dict[str, Any]: - return self.flatpak_service.remove_plugin_owned_extensions() - 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"): + result = self.flatpak_service.remove_plugin_owned_environment() + if result.get("success"): + self.configuration_service.reset_all_flatpak_configs() + else: decky.logger.warning(result.get("error")) except Exception as error: decky.logger.error(f"Error during Flatpak cleanup: {error}") - + self.installation_service.cleanup_on_uninstall() 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 b3bdb69..2108071 100644 --- a/py_modules/lsfg_vk/steam_service.py +++ b/py_modules/lsfg_vk/steam_service.py @@ -1,116 +1,64 @@ import re -import shlex from pathlib import Path from typing import Dict, Optional, Tuple from .base_service import BaseService -from .constants import STEAM_LOSSLESS_SCALING_APP_ID, STEAM_LOSSLESS_SCALING_BRANCH - - -_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$") - - -def _split_command(value: Optional[str]) -> Optional[list[str]]: - if not isinstance(value, str) or not value.strip(): - return [] - try: - return shlex.split(value, posix=True) - except ValueError: - return None - - -def classify_shortcut_transport( - executable: Optional[str], - launch_options: Optional[str] = None, -) -> Dict[str, object]: - """Classify only direct Flatpak invocations; leave shell launchers on host.""" - 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"} - - if executable_tokens[0] != "/usr/bin/flatpak": - 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 == "--": - continue - if argument.startswith("-"): - continue - if _FLATPAK_APP_ID.fullmatch(argument): - return {"kind": "flatpak", "flatpakAppId": argument} - return {"kind": "host"} - return {"kind": "host"} +from .constants import ( + STEAM_LOSSLESS_SCALING_APP_ID, + STEAM_LOSSLESS_SCALING_BRANCH, +) 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]: end = data.index(b"\0", offset) @@ -127,16 +75,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 @@ -155,53 +99,20 @@ 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, - ) - game: Dict[str, object] = { - "appid": str(appid & 0xffffffff), + return { + "appid": str(appid & 0xFFFFFFFF), "name": name, "nonSteam": True, - "transport": classify_shortcut_transport(executable, launch_options), } - 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 - 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(): @@ -210,25 +121,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]]: @@ -238,7 +136,6 @@ class SteamService(BaseService): ) if section is None: return None - depth = 1 in_string = False escaped = False @@ -251,9 +148,7 @@ class SteamService(BaseService): escaped = True elif character == '"': in_string = False - continue - - if character == '"': + elif character == '"': in_string = True elif character == "{": depth += 1 @@ -268,98 +163,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<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"' - ) - 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<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"') + 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 @@ -370,12 +249,10 @@ 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"}, } for game in self._shortcut_games(): games.setdefault(str(game["appid"]), game) 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 diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py index dae565b..ebe9526 100644 --- a/py_modules/lsfg_vk/wrapper_service.py +++ b/py_modules/lsfg_vk/wrapper_service.py @@ -1,12 +1,9 @@ -"""Own the small per-AppID workaround dispatcher used by Steam launches.""" - from __future__ import annotations import json import re import shlex import threading -from pathlib import Path from typing import Any, Dict, Optional, Tuple from .base_service import BaseService @@ -14,8 +11,6 @@ from .constants import WRAPPER_FILENAME class WrapperService(BaseService): - """Persist workaround state and compile it into a safe POSIX wrapper.""" - LEGACY_FORMAT_VERSION = 1 FORMAT_VERSION = 2 LEGACY_MARKER = "# lsfg-vk-wrapper-format: 1" @@ -35,6 +30,8 @@ class WrapperService(BaseService): "DISABLE_GAMESCOPE_WSI", "DXVK_HDR", "SteamDeck", + "DISABLE_LSFGVK", + "DISABLE_LSFG", "DISABLE_VKBASALT", "ENABLE_VKBASALT", "MESA_LOADER_DRIVER_OVERRIDE", @@ -84,51 +81,15 @@ class WrapperService(BaseService): return state @classmethod - def _validate_transport(cls, raw: Any) -> Dict[str, Any]: - if raw is None: - return {"kind": "host"} - if not isinstance(raw, dict): - raise ValueError("Workaround transport must be an object") - kind = raw.get("kind") - if kind == "host": - return {"kind": "host"} - if kind == "flatpak": - app_id = raw.get("flatpakAppId") - if ( - not isinstance(app_id, str) - or not re.fullmatch( - r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$", - app_id, - ) - ): - raise ValueError("Flatpak transport requires a valid application ID") - return {"kind": "flatpak", "flatpakAppId": app_id} - raise ValueError("Workaround transport must be host or flatpak") - - @classmethod def _validate_entry(cls, raw: Any) -> Dict[str, Any]: if not isinstance(raw, dict): raise ValueError("Workaround AppID entry must be an object") entry = { "state": cls._validate_state(raw.get("state")), "command_token_added": raw.get("command_token_added", False), - # Version 1 entries had no transport field. They are preserved as - # host entries until the shortcut is explicitly repaired with the - # backend's classified transport. - "transport": cls._validate_transport(raw.get("transport")), } if type(entry["command_token_added"]) is not bool: raise ValueError("command_token_added must be a boolean") - if "shortcut_exe" in raw and raw["shortcut_exe"] is not None: - shortcut_exe = raw["shortcut_exe"] - if ( - not isinstance(shortcut_exe, str) - or not shortcut_exe.startswith("/") - or "\x00" in shortcut_exe - or not shortcut_exe.strip() - ): - raise ValueError("shortcut_exe must be an absolute executable path") - entry["shortcut_exe"] = shortcut_exe return entry @classmethod @@ -158,11 +119,11 @@ class WrapperService(BaseService): if self.sidecar_path.is_symlink() or not self.sidecar_path.is_file(): raise RuntimeError("Workaround state path is not a regular file") try: - raw = json.loads(self.sidecar_path.read_text(encoding="utf-8")) + content = self.sidecar_path.read_text(encoding="utf-8") + raw = json.loads(content) except (OSError, json.JSONDecodeError) as error: raise RuntimeError(f"Could not read workaround state: {error}") from error - document = self._validate_document(raw) - return document, True, self.sidecar_path.read_text(encoding="utf-8") + return self._validate_document(raw), True, content def _wrapper_marker(self) -> bool: if self.wrapper_path.is_symlink() or not self.wrapper_path.exists(): @@ -179,29 +140,21 @@ class WrapperService(BaseService): if not self.wrapper_path.exists() and not self.wrapper_path.is_symlink(): return False if self.wrapper_path.is_symlink() or not self._wrapper_marker(): - raise RuntimeError( - f"Refusing to replace unowned wrapper at {self.wrapper_path}" - ) + raise RuntimeError(f"Refusing to replace unowned wrapper at {self.wrapper_path}") return True @staticmethod def _shell(value: str) -> str: return shlex.quote(value) - @staticmethod - def _direct_flatpak_tokens(value: str) -> Optional[list[str]]: - """Parse the supported full executable form: /usr/bin/flatpak run APP.""" - try: - tokens = shlex.split(value, posix=True) - except ValueError: - return None - if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run": - return tokens - return None - - @classmethod - def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]: - lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)] + def _state_lines(self, state: Dict[str, Any]) -> list[str]: + lines = [" unset " + " ".join(self.MANAGED_ENV_KEYS)] + lines.extend([ + ' SteamAppId="$appid"', + " export SteamAppId", + f" LSFGVK_CONFIG={self._shell(str(self.config_file_path))}", + " export LSFGVK_CONFIG", + ]) if state["disableGamescopeWsi"]: lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"]) if state["disableHdr"]: @@ -233,72 +186,12 @@ class WrapperService(BaseService): " fi", " export DXVK_CONFIG", ]) - lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}") return lines - def _dll_directory(self) -> Path: - if self.config_file_path.exists(): - try: - content = self.config_file_path.read_text(encoding="utf-8") - match = re.search( - r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"', - content, - ) - if match: - configured_dll = json.loads('"' + match.group(1) + '"') - if configured_dll: - return Path(configured_dll).parent - except Exception: - pass - return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" - - def _flatpak_args(self, state: Dict[str, Any]) -> list[str]: - config_dir = str(self.config_dir) - config_file = str(self.config_file_path) - dll_dir = str(self._dll_directory()) - args = [ - self._shell(f"--filesystem={config_dir}:rw"), - self._shell(f"--filesystem={dll_dir}:ro"), - self._shell(f"--env=LSFGVK_CONFIG={config_file}"), - '"--env=LSFGVK_FLATPAK=1"', - '"--env=SteamAppId=$appid"', - '"--unset-env=DISABLE_GAMESCOPE_WSI"', - '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else - '"--env=ENABLE_GAMESCOPE_WSI=0"', - '"--unset-env=DXVK_HDR"' if not state["disableHdr"] else - '"--env=DXVK_HDR=0"', - '"--unset-env=SteamDeck"' if not state["disableSteamdeckMode"] else - '"--env=SteamDeck=0"', - '"--unset-env=DISABLE_VKBASALT" "--unset-env=ENABLE_VKBASALT"', - ] - if state["disableVkbasalt"]: - args.append('"--env=DISABLE_VKBASALT=1"') - args.extend([ - '"--unset-env=MESA_LOADER_DRIVER_OVERRIDE" "--unset-env=__GLX_VENDOR_LIBRARY_NAME" "--unset-env=GALLIUM_DRIVER"', - ]) - if state["enableZink"]: - args.extend([ - '"--env=__GLX_VENDOR_LIBRARY_NAME=mesa"', - '"--env=MESA_LOADER_DRIVER_OVERRIDE=zink"', - '"--env=GALLIUM_DRIVER=zink"', - ]) - args.extend([ - '"--unset-env=DXVK_FRAME_RATE"', - ]) - static_args = " ".join(args) - return [ - ' if [ -n "${DXVK_CONFIG+x}" ]; then', - f' set -- "$flatpak_command" {static_args} "--env=DXVK_CONFIG=$DXVK_CONFIG" "$@"', - " else", - f' set -- "$flatpak_command" {static_args} "$@"', - " fi", - ] - def _render_wrapper(self, document: Dict[str, Any]) -> str: lines = [ "#!/bin/sh", self.MARKER, - "# Generated by Decky LSFG-VK; edits will be rejected on the next update.", "", "appid=", 'case "${SteamAppId-}" in', @@ -317,77 +210,25 @@ class WrapperService(BaseService): ' *) appid="${STEAM_COMPAT_APP_ID}" ;;', " esac", "fi", - "shortcut_exe=", 'case "$appid" in', ] for appid in sorted(document["apps"], key=lambda value: int(value)): - entry = document["apps"][appid] lines.append(f" {appid})") - lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe"))) + lines.extend(self._state_lines(document["apps"][appid]["state"])) lines.append(" ;;") lines.extend([ "esac", - "", - 'if [ -n "$shortcut_exe" ]; then', - ]) - # The arguments are emitted per branch below so the values are static and - # the wrapper never needs a JSON parser or another helper executable. - lines.append(' case "$appid" in') - for appid in sorted(document["apps"], key=lambda value: int(value)): - entry = document["apps"][appid] - transport = entry.get("transport", {"kind": "host"}) - if transport.get("kind") != "flatpak": - continue - shortcut_exe = entry.get("shortcut_exe", "") - direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe) - if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak": - raise ValueError( - f"Flatpak target {appid} does not use a direct flatpak executable" - ) - lines.append(f" {appid})") - lines.extend([ - *( - [ - f" shortcut_exe={self._shell(direct_flatpak_tokens[0])}", - " set -- " - + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:]) - + ' "$@"', - ] - if direct_flatpak_tokens - else [] - ), - ' if [ "${1-}" != "run" ]; then', - ' echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2', - " exit 64", - " fi", - ' flatpak_command="$1"', - " shift", - " flatpak_target=", - ' for flatpak_arg in "$@"; do', - ' case "$flatpak_arg" in', - ' -*) ;;', - ' *) flatpak_target="$flatpak_arg"; break ;;', - " esac", - " done", - f' if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then', - ' echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2', - " exit 64", - " fi", - ]) - lines.extend(self._flatpak_args(entry["state"])) - lines.append(" ;;") - lines.extend([ - " esac", - ' exec "$shortcut_exe" "$@"', - "fi", 'exec "$@"', "", ]) return "\n".join(lines) def _write_document(self, document: Dict[str, Any]) -> None: - content = json.dumps(document, indent=2, sort_keys=True) + "\n" - self._write_file(self.sidecar_path, content, 0o644) + self._write_file( + self.sidecar_path, + json.dumps(document, indent=2, sort_keys=True) + "\n", + 0o644, + ) def _write_pair(self, document: Dict[str, Any]) -> None: old_sidecar_exists = self.sidecar_path.exists() @@ -421,9 +262,7 @@ class WrapperService(BaseService): "state": dict(entry["state"]) if entry else None, "wrapper_path": self.WRAPPER_TOKEN, "wrapper_owned": self._wrapper_marker() if document["apps"] else False, - "shortcut_exe": entry.get("shortcut_exe") if entry else None, "command_token_added": entry.get("command_token_added", False) if entry else False, - "transport": dict(entry.get("transport", {"kind": "host"})) if entry else None, } def get(self, appid: str) -> Dict[str, Any]: @@ -448,9 +287,7 @@ class WrapperService(BaseService): self, appid: str, state: Dict[str, Any], - shortcut_exe: Optional[str] = None, command_token_added: bool = False, - transport: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: try: normalized = self._valid_appid(appid) @@ -460,26 +297,10 @@ class WrapperService(BaseService): with self._lock: self._assert_wrapper_owned_or_absent() document, _, _ = self._read_document() - previous_entry = document["apps"].get(normalized) - selected_transport = self._validate_transport( - transport - if transport is not None - else ( - previous_entry.get("transport") - if previous_entry - else None - ) - ) - entry: Dict[str, Any] = { + document["apps"][normalized] = { "state": validated_state, - "command_token_added": bool(command_token_added), - "transport": selected_transport, + "command_token_added": command_token_added, } - if shortcut_exe is not None: - entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe}) - elif previous_entry and "shortcut_exe" in previous_entry: - entry["shortcut_exe"] = previous_entry["shortcut_exe"] - document["apps"][normalized] = entry self._write_pair(document) return self._response(document, normalized) except Exception as error: @@ -516,7 +337,6 @@ class WrapperService(BaseService): } def repair(self) -> Dict[str, Any]: - """Regenerate a missing owned wrapper without importing old global state.""" try: with self._lock: document, _, _ = self._read_document() |