""" Flatpak service for managing lsfg-vk Flatpak runtime extensions. """ import subprocess import os from pathlib import Path from typing import Dict, Any, List, Optional from .base_service import BaseService from .config_schema import ConfigurationManager from .types import BaseResponse class FlatpakExtensionStatus(BaseResponse): """Response for Flatpak extension status""" def __init__(self, success: bool = False, message: str = "", error: str = "", installed_23_08: bool = False, installed_24_08: bool = False, installed_25_08: bool = False): super().__init__(success, message, error) self.installed_23_08 = installed_23_08 self.installed_24_08 = installed_24_08 self.installed_25_08 = installed_25_08 class FlatpakAppInfo(BaseResponse): """Response for Flatpak app information""" def __init__(self, success: bool = False, message: str = "", error: str = "", apps: List[Dict[str, Any]] = None, total_apps: int = 0): super().__init__(success, message, error) self.apps = apps or [] self.total_apps = total_apps class FlatpakOverrideResponse(BaseResponse): """Response for Flatpak override operations""" def __init__(self, success: bool = False, message: str = "", error: str = "", app_id: str = "", operation: str = ""): super().__init__(success, message, error) self.app_id = app_id self.operation = operation class FlatpakService(BaseService): """Service for handling Flatpak runtime extensions and app overrides""" def __init__(self, logger=None): super().__init__(logger) self.extension_id_23_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/23.08" self.extension_id_24_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/24.08" self.extension_id_25_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/25.08" self.flatpak_command = None def _get_lsfg_paths(self) -> tuple[str, str]: """Return the v2 config directory and directory containing Lossless.dll.""" config_path = str(self.config_dir) dll_directory = str(self.user_home / ".local/share/Steam/steamapps/common") if not self.config_file_path.exists(): return config_path, dll_directory try: profile_data = ConfigurationManager.parse_toml_content_multi_profile( self.config_file_path.read_text(encoding="utf-8") ) configured_dll = profile_data["global_config"].get("dll", "") if configured_dll: dll_directory = str(Path(str(configured_dll)).parent) except (OSError, ValueError, KeyError, TypeError) as error: self.log.debug("Could not read configured DLL path for Flatpak override: %s", error) return config_path, dll_directory def _remove_legacy_app_overrides(self, app_id: str) -> list[str]: """Remove only the v1 overrides previously created by this plugin.""" legacy_dll_path = self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" legacy_overrides = [ ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], ["override", "--user", f"--nofilesystem={legacy_dll_path}", app_id], ["override", "--user", f"--nofilesystem={self.lsfg_launch_script_path}", app_id], ] errors = [] for args in legacy_overrides: result = self._run_flatpak_command(args, capture_output=True, text=True) if result.returncode != 0: errors.append(f"{' '.join(args[2:-1])}: {result.stderr}") return errors def _get_clean_env(self): """Get a clean environment without PyInstaller's bundled libraries""" env = os.environ.copy() if 'LD_LIBRARY_PATH' in env: del env['LD_LIBRARY_PATH'] standard_paths = ['/usr/bin', '/usr/local/bin', '/bin'] current_path = env.get('PATH', '') path_parts = current_path.split(':') if current_path else [] for std_path in standard_paths: if std_path not in path_parts: path_parts.insert(0, std_path) env['PATH'] = ':'.join(path_parts) return env def _run_flatpak_command(self, args: List[str], **kwargs): """Run flatpak command with clean environment to avoid library conflicts""" if self.flatpak_command is None: raise FileNotFoundError("Flatpak command not available") env = self._get_clean_env() self.log.info(f"Running flatpak with PATH: {env.get('PATH')}") self.log.info(f"LD_LIBRARY_PATH removed: {'LD_LIBRARY_PATH' not in env}") return subprocess.run([self.flatpak_command] + args, env=env, **kwargs) def check_flatpak_available(self) -> bool: """Check if flatpak command is available and store the working command""" self.log.info(f"PATH: {os.environ.get('PATH', 'Not set')}") self.log.info(f"HOME: {os.environ.get('HOME', 'Not set')}") self.log.info(f"USER: {os.environ.get('USER', 'Not set')}") flatpak_paths = [ "flatpak", "/usr/bin/flatpak", "/var/lib/flatpak/exports/bin/flatpak", "/home/deck/.local/bin/flatpak" ] for flatpak_path in flatpak_paths: try: result = subprocess.run([flatpak_path, "--version"], capture_output=True, check=True, text=True, env=self._get_clean_env()) self.log.info(f"Flatpak found at {flatpak_path}: {result.stdout.strip()}") self.flatpak_command = flatpak_path return True except (subprocess.CalledProcessError, FileNotFoundError): self.log.debug(f"Flatpak not found at {flatpak_path}") continue self.log.error("Flatpak command not found in any known locations") self.flatpak_command = None return False def get_extension_status(self) -> FlatpakExtensionStatus: """Check if lsfg-vk Flatpak extensions are installed""" try: if not self.check_flatpak_available(): error_msg = "Flatpak is not available on this system" if self.flatpak_command is None: error_msg += ". Command not found in PATH or common install locations." self.log.error(error_msg) return self._error_response(FlatpakExtensionStatus, error_msg, installed_23_08=False, installed_24_08=False, installed_25_08=False) result = self._run_flatpak_command( ["list", "--runtime"], capture_output=True, text=True, check=True ) installed_runtimes = result.stdout base_extension_name = "org.freedesktop.Platform.VulkanLayer.lsfgvk" installed_23_08 = False installed_24_08 = False installed_25_08 = False for line in installed_runtimes.split('\n'): if base_extension_name in line: if "23.08" in line: installed_23_08 = True elif "24.08" in line: installed_24_08 = True elif "25.08" in line: installed_25_08 = True status_msg = [] if installed_23_08: status_msg.append("23.08 runtime extension installed") if installed_24_08: status_msg.append("24.08 runtime extension installed") if installed_25_08: status_msg.append("25.08 runtime extension installed") if not status_msg: status_msg.append("No lsfg-vk runtime extensions installed") return self._success_response(FlatpakExtensionStatus, "; ".join(status_msg), installed_23_08=installed_23_08, installed_24_08=installed_24_08, installed_25_08=installed_25_08) except subprocess.CalledProcessError as e: error_msg = f"Error checking Flatpak extensions: {e.stderr if e.stderr else str(e)}" self.log.error(error_msg) return self._error_response(FlatpakExtensionStatus, error_msg, installed_23_08=False, installed_24_08=False, installed_25_08=False) def install_extension(self, version: str) -> BaseResponse: """Install a specific version of the lsfg-vk Flatpak extension""" try: if version not in ["23.08", "24.08", "25.08"]: return self._error_response(BaseResponse, "Invalid version. Must be '23.08', '24.08', or '25.08'") if not self.check_flatpak_available(): return self._error_response(BaseResponse, "Flatpak is not available on this system") if version == "23.08": return self._error_response( BaseResponse, "Flathub supplies the lsfg-vk v2 extension for runtimes 24.08 and 25.08; install 23.08 manually.", ) result = self._run_flatpak_command( ["install", "--user", "--noninteractive", "flathub", f"org.freedesktop.Platform.VulkanLayer.lsfgvk//{version}"], capture_output=True, text=True ) if result.returncode != 0: error_msg = f"Failed to install Flatpak extension: {result.stderr}" self.log.error(error_msg) return self._error_response(BaseResponse, error_msg) self.log.info(f"Successfully installed lsfg-vk Flatpak extension {version}") return self._success_response(BaseResponse, f"lsfg-vk {version} runtime extension installed successfully") except Exception as e: error_msg = f"Error installing Flatpak extension {version}: {str(e)}" self.log.error(error_msg) return self._error_response(BaseResponse, error_msg) def uninstall_extension(self, version: str) -> BaseResponse: """Uninstall a specific version of the lsfg-vk Flatpak extension""" try: if version not in ["23.08", "24.08", "25.08"]: return self._error_response(BaseResponse, "Invalid version. Must be '23.08', '24.08', or '25.08'") if not self.check_flatpak_available(): return self._error_response(BaseResponse, "Flatpak is not available on this system") if version == "23.08": extension_id = self.extension_id_23_08 elif version == "24.08": extension_id = self.extension_id_24_08 else: extension_id = self.extension_id_25_08 result = self._run_flatpak_command( ["uninstall", "--user", "--noninteractive", extension_id], capture_output=True, text=True ) if result.returncode != 0: error_msg = f"Failed to uninstall Flatpak extension: {result.stderr}" self.log.error(error_msg) return self._error_response(BaseResponse, error_msg) self.log.info(f"Successfully uninstalled lsfg-vk Flatpak extension {version}") return self._success_response(BaseResponse, f"lsfg-vk {version} runtime extension uninstalled successfully") except Exception as e: error_msg = f"Error uninstalling Flatpak extension {version}: {str(e)}" self.log.error(error_msg) return self._error_response(BaseResponse, error_msg) def get_flatpak_apps(self) -> FlatpakAppInfo: """Get list of installed Flatpak apps and their lsfg-vk override status""" try: if not self.check_flatpak_available(): error_msg = "Flatpak is not available on this system" if self.flatpak_command is None: error_msg += ". Command not found in PATH or common install locations." return self._error_response(FlatpakAppInfo, error_msg, apps=[], total_apps=0) result = self._run_flatpak_command( ["list", "--app"], capture_output=True, text=True, check=True ) apps = [] for line in result.stdout.strip().split('\n'): if not line.strip(): continue parts = line.split('\t') if len(parts) >= 2: app_name = parts[0].strip() app_id = parts[1].strip() # Check override status override_status = self._check_app_override_status(app_id) apps.append({ "app_id": app_id, "app_name": app_name, "has_filesystem_override": override_status["filesystem"], "has_env_override": override_status["env"] }) return self._success_response(FlatpakAppInfo, f"Found {len(apps)} Flatpak applications", apps=apps, total_apps=len(apps)) except subprocess.CalledProcessError as e: error_msg = f"Error getting Flatpak apps: {e.stderr if e.stderr else str(e)}" self.log.error(error_msg) return self._error_response(FlatpakAppInfo, error_msg, apps=[], total_apps=0) def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: """Check if an app has lsfg-vk overrides set""" try: result = self._run_flatpak_command( ["override", "--user", "--show", app_id], capture_output=True, text=True ) if result.returncode != 0: return {"filesystem": False, "env": False} output = result.stdout config_path, dll_directory = self._get_lsfg_paths() filesystem_section = "" in_context = False for line in output.split('\n'): line = line.strip() if line == "[Context]": in_context = True elif line.startswith("[") and line != "[Context]": in_context = False elif in_context and line.startswith("filesystems="): filesystem_section = line break has_config_fs = config_path in filesystem_section has_dll_fs = dll_directory in filesystem_section filesystem_override = has_config_fs and has_dll_fs env_override = False in_environment = False for line in output.split('\n'): line = line.strip() if line == "[Environment]": in_environment = True elif line.startswith("[") and line != "[Environment]": in_environment = False elif in_environment and line.startswith(f"LSFGVK_CONFIG={config_path}/conf.toml"): env_override = True break self.log.debug(f"Override status for {app_id}: filesystem={filesystem_override} ({has_config_fs}/{has_dll_fs}), env={env_override}") return {"filesystem": filesystem_override, "env": env_override} except Exception as e: self.log.error(f"Error checking override status for {app_id}: {e}") return {"filesystem": False, "env": False} def set_app_override(self, app_id: str) -> FlatpakOverrideResponse: """Set lsfg-vk overrides for a Flatpak app""" try: if not self.check_flatpak_available(): return self._error_response(FlatpakOverrideResponse, "Flatpak is not available on this system", app_id=app_id, operation="set") config_path, dll_directory = self._get_lsfg_paths() filesystem_overrides = [ f"--filesystem={dll_directory}:ro", f"--filesystem={config_path}:rw", ] for override in filesystem_overrides: result = self._run_flatpak_command( ["override", "--user", override, app_id], capture_output=True, text=True ) if result.returncode != 0: error_msg = f"Failed to set filesystem override {override}: {result.stderr}" return self._error_response(FlatpakOverrideResponse, error_msg, app_id=app_id, operation="set") result = self._run_flatpak_command( ["override", "--user", f"--env=LSFGVK_CONFIG={config_path}/conf.toml", app_id], capture_output=True, text=True ) if result.returncode != 0: error_msg = f"Failed to set environment override: {result.stderr}" return self._error_response(FlatpakOverrideResponse, error_msg, app_id=app_id, operation="set") legacy_errors = self._remove_legacy_app_overrides(app_id) if legacy_errors: self.log.warning( "Applied v2 overrides for %s but could not fully remove v1 overrides: %s", app_id, "; ".join(legacy_errors), ) self.log.info(f"Successfully set lsfg-vk overrides for {app_id}") return self._success_response(FlatpakOverrideResponse, f"lsfg-vk overrides set for {app_id}", app_id=app_id, operation="set") except Exception as e: error_msg = f"Error setting overrides for {app_id}: {str(e)}" self.log.error(error_msg) return self._error_response(FlatpakOverrideResponse, error_msg, app_id=app_id, operation="set") def remove_app_override(self, app_id: str) -> FlatpakOverrideResponse: """Remove lsfg-vk overrides for a Flatpak app""" try: if not self.check_flatpak_available(): return self._error_response(FlatpakOverrideResponse, "Flatpak is not available on this system", app_id=app_id, operation="remove") config_path, dll_directory = self._get_lsfg_paths() filesystem_overrides = [ f"--nofilesystem={dll_directory}", f"--nofilesystem={config_path}", ] removal_errors = [] # Remove filesystem overrides for override in filesystem_overrides: result = self._run_flatpak_command( ["override", "--user", override, app_id], capture_output=True, text=True ) if result.returncode != 0: removal_errors.append(f"{override}: {result.stderr}") result = self._run_flatpak_command( ["override", "--user", "--unset-env=LSFGVK_CONFIG", app_id], capture_output=True, text=True ) if result.returncode != 0: removal_errors.append(f"unset-env: {result.stderr}") removal_errors.extend(self._remove_legacy_app_overrides(app_id)) if removal_errors: self.log.warning(f"Some override removals had issues for {app_id}: {'; '.join(removal_errors)}") self.log.info(f"Completed override removal for {app_id}") return self._success_response(FlatpakOverrideResponse, f"lsfg-vk overrides removed for {app_id}", app_id=app_id, operation="remove") except Exception as e: error_msg = f"Error removing overrides for {app_id}: {str(e)}" self.log.error(error_msg) return self._error_response(FlatpakOverrideResponse, error_msg, app_id=app_id, operation="remove")