diff options
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_profile_service.py | 358 |
1 files changed, 358 insertions, 0 deletions
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..29751d4 --- /dev/null +++ b/py_modules/lsfg_vk/flatpak_profile_service.py @@ -0,0 +1,358 @@ +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"): + baseline = self._baseline_content(app_id, entry) + self.flatpak_service._write_file(path, baseline) + 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) + 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]: + try: + existing = self.configuration_service.get_flatpak_config(app_id) + 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") + 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") + state, 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: + 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: + apps = self.get_apps() + if not apps.get("success"): + raise RuntimeError(apps.get("error") or "Could not list Flatpak applications") + enabled = {item["app_id"]: item for item in apps.get("apps", []) if item.get("enabled")} + 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 len(fields) < 1: + continue + app_id = fields[0] + if app_id not in enabled: + continue + active = len(fields) > 1 and fields[1].strip().lower() in {"1", "true", "yes", "active"} + pid = fields[2].strip() if len(fields) > 2 else "" + running.append({**enabled[app_id], "active": active, "pid": pid}) + running.sort(key=lambda item: (not item.get("active", False), str(item.get("app_name", "")).lower())) + return {"success": True, "message": "", "error": None, "apps": running} + except Exception as error: + return {"success": False, "message": "", "error": str(error), "apps": []} |