diff options
Diffstat (limited to 'py_modules/lsfg_vk/plugin.py')
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 277 |
1 files changed, 131 insertions, 146 deletions
diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index 76a4250..a1fd40f 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,198 @@ 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 - """ + def _cleanup_runtime_state(self, preserve_wrapper: bool = False): + flatpak = self.flatpak_service.remove_plugin_owned_environment() + if not flatpak.get("success"): + return flatpak.get("error") or "Could not clean up Flatpak support" + profiles = self.configuration_service.reset_all_flatpak_configs() + if not profiles.get("success"): + return profiles.get("error") or "Could not remove Flatpak profiles" + wrapper = self.wrapper_service.neutralize() if preserve_wrapper else self.wrapper_service.purge() + if not wrapper.get("success"): + return wrapper.get("error") or "Could not remove workaround state" + return None + + async def uninstall_lsfg_vk(self): + error = self._cleanup_runtime_state() + if error: + return { + "success": False, + "message": "", + "error": error, + "removed_files": None, + } 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 update_global_config(self, config: Dict[str, Any]): + return self.configuration_service.update_global_config(config) + + 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 get_workaround_apps(self): + return self.wrapper_service.list_apps() + 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, - ) + non_steam: bool = False, + ): + return self.wrapper_service.set(appid, state, command_token_added, non_steam) - 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 ensure_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]: - return self.flatpak_service.ensure_app_support(flatpak_app_id) + async def get_flatpak_workaround_state(self, flatpak_app_id: str): + return self.flatpak_profile_service.get_workaround_state(flatpak_app_id) - 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 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 set_flatpak_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: - return self.flatpak_service.set_extension_enabled(version, enabled) + async def remove_flatpak_app(self, flatpak_app_id: str): + return self.flatpak_profile_service.remove_app(flatpak_app_id) + + 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"): - decky.logger.warning(result.get("error")) + error = self._cleanup_runtime_state(preserve_wrapper=True) + if error: + decky.logger.warning(f"Preserving lsfg-vk files because uninstall cleanup failed: {error}") + return except Exception as error: - decky.logger.error(f"Error during Flatpak cleanup: {error}") - + decky.logger.error(f"Error during lsfg-vk cleanup: {error}") + return + 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") |