diff options
Diffstat (limited to 'py_modules/lsfg_vk/plugin.py')
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 605 |
1 files changed, 166 insertions, 439 deletions
diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index cb59b4f..c63e3a9 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -1,495 +1,222 @@ -""" -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 -import subprocess -import hashlib -from typing import Dict, Any -from pathlib import Path +from typing import Any, Dict import decky -from .installation import InstallationService -from .dll_detection import DllDetectionService from .configuration import ConfigurationService -from .config_schema import ConfigurationManager +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 DLL detection services. It implements the Decky Loader plugin lifecycle - methods (_main, _unload, _uninstall, _migration). - """ - def __init__(self): - """Initialize the plugin with all necessary services""" - self.installation_service = InstallationService() - self.dll_detection_service = DllDetectionService() - self.configuration_service = ConfigurationService() + self.runtime_service = RuntimeService() + self.steam_service = SteamService() + self.installation_service = InstallationService( + runtime_service=self.runtime_service, + steam_service=self.steam_service, + ) + 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 lsfg-vk by extracting the zip file 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 - """ - return self.installation_service.uninstall() - - async def check_lossless_scaling_dll(self) -> Dict[str, Any]: - """Check if Lossless Scaling DLL is available at the expected paths - - Returns: - DllDetectionResponse dict with detection status and path info - """ - return self.dll_detection_service.check_lossless_scaling_dll() - - async def get_dll_stats(self) -> Dict[str, Any]: - """Get detailed statistics about the detected DLL - - Returns: - Dict containing DLL path, SHA256 hash, and other stats - """ - try: - dll_result = self.dll_detection_service.check_lossless_scaling_dll() - - if not dll_result.get("detected") or not dll_result.get("path"): - return { - "success": False, - "error": "DLL not detected", - "dll_path": None, - "dll_sha256": None - } - - dll_path = dll_result["path"] - if dll_path is None: - return { - "success": False, - "error": "DLL path is None", - "dll_path": None, - "dll_sha256": None - } - - dll_path_obj = Path(dll_path) - - sha256_hash = hashlib.sha256() - try: - with open(dll_path_obj, "rb") as f: - for chunk in iter(lambda: f.read(4096), b""): - sha256_hash.update(chunk) - dll_sha256 = sha256_hash.hexdigest() - except Exception as e: - return { - "success": False, - "error": f"Failed to calculate SHA256: {str(e)}", - "dll_path": dll_path, - "dll_sha256": None - } - - return { - "success": True, - "dll_path": dll_path, - "dll_sha256": dll_sha256, - "dll_source": dll_result.get("source"), - "error": None - } - - except Exception as e: + 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, - "error": f"Failed to get DLL stats: {str(e)}", - "dll_path": None, - "dll_sha256": None + "message": "", + "error": error, + "removed_files": None, } + return self.installation_service.uninstall() - async def get_lsfg_config(self) -> Dict[str, Any]: - """Read current lsfg script configuration - - Returns: - ConfigurationResponse dict with current configuration or error - """ - return self.configuration_service.get_config() - - async def get_config_schema(self) -> Dict[str, Any]: - """Get configuration schema information for frontend - - Returns: - Dict with field names, types, defaults, and profile information - """ - try: - profiles_response = self.configuration_service.get_profiles() - - schema_data = { - "field_names": ConfigurationManager.get_field_names(), - "field_types": {name: field_type.value for name, field_type in ConfigurationManager.get_field_types().items()}, - "defaults": ConfigurationManager.get_defaults() - } - - if profiles_response.get("success"): - schema_data["profiles"] = profiles_response.get("profiles", []) - schema_data["current_profile"] = profiles_response.get("current_profile") - else: - schema_data["profiles"] = ["decky-lsfg-vk"] - schema_data["current_profile"] = "decky-lsfg-vk" - - return schema_data - - except (ValueError, KeyError, AttributeError) as e: - self.configuration_service.log.warning(f"Failed to get full schema, using fallback: {e}") - return { - "field_names": ConfigurationManager.get_field_names(), - "field_types": {name: field_type.value for name, field_type in ConfigurationManager.get_field_types().items()}, - "defaults": ConfigurationManager.get_defaults(), - "profiles": ["decky-lsfg-vk"], - "current_profile": "decky-lsfg-vk" - } + async def get_game_configs(self): + return self.configuration_service.get_game_configs() - async def update_lsfg_config(self, config: Dict[str, Any]) -> Dict[str, Any]: - """Update lsfg TOML configuration using object-based API (single source of truth) - - Args: - config: Configuration data dictionary containing all settings - - Returns: - ConfigurationResponse dict with success status - """ - validated_config = ConfigurationManager.validate_config(config) - - return self.configuration_service.update_config_from_dict(validated_config) - - async def get_profiles(self) -> Dict[str, Any]: - """Get list of all profiles and current profile - - Returns: - ProfilesResponse dict with profile list and current profile - """ - return self.configuration_service.get_profiles() - - async def create_profile(self, profile_name: str, source_profile: str = None) -> Dict[str, Any]: - """Create a new profile - - Args: - profile_name: Name for the new profile - source_profile: Optional source profile to copy from (default: current profile) - - Returns: - ProfileResponse dict with success status - """ - return self.configuration_service.create_profile(profile_name, source_profile) - - async def delete_profile(self, profile_name: str) -> Dict[str, Any]: - """Delete a profile - - Args: - profile_name: Name of the profile to delete - - Returns: - ProfileResponse dict with success status - """ - return self.configuration_service.delete_profile(profile_name) - - async def rename_profile(self, old_name: str, new_name: str) -> Dict[str, Any]: - """Rename a profile - - Args: - old_name: Current profile name - new_name: New profile name - - Returns: - ProfileResponse dict with success status - """ - return self.configuration_service.rename_profile(old_name, new_name) - - async def set_current_profile(self, profile_name: str) -> Dict[str, Any]: - """Set the current active profile - - Args: - profile_name: Name of the profile to set as current - - Returns: - ProfileResponse dict with success status - """ - return self.configuration_service.set_current_profile(profile_name) - - async def update_profile_config(self, profile_name: str, config: Dict[str, Any]) -> Dict[str, Any]: - """Update configuration for a specific profile - - Args: - profile_name: Name of the profile to update - config: Configuration data dictionary containing settings - - Returns: - ConfigurationResponse dict with success status - """ - validated_config = ConfigurationManager.validate_config(config) - - return self.configuration_service.update_profile_config(profile_name, validated_config) - - async def get_launch_option(self) -> Dict[str, Any]: - """Get the launch option that users need to set for their games - - Returns: - Dict containing the launch option string and instructions - """ - return { - "launch_option": "~/lsfg %command%", - "instructions": "Add this to your game's launch options in Steam Properties", - "explanation": "The lsfg script is created during installation and sets up the environment for the plugin" - } + 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): + return self.configuration_service.reset_game_config(appid) + + async def reset_game_configs(self, appids): + return self.configuration_service.reset_game_configs(appids) + + async def reset_all_game_configs(self): + return self.configuration_service.reset_all_game_configs() + + 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 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 set_workaround_state( + self, + appid: str, + state: Dict[str, Any], + command_token_added: bool = False, + non_steam: bool = False, + ): + return self.wrapper_service.set(appid, state, command_token_added, non_steam) + + async def remove_workaround_state(self, appid: str): + return self.wrapper_service.remove(appid) + + 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_launch_script_content(self) -> Dict[str, Any]: - """Get the content of the launch script file - - Returns: - FileContentResponse dict with file content or error information - """ - try: - script_path = self.installation_service.get_launch_script_path() - - if not os.path.exists(script_path): - return { - "success": False, - "error": f"Launch script not found at {script_path}", - "path": str(script_path) - } - - with open(script_path, 'r') as file: - content = file.read() - - return { - "success": True, - "content": content, - "path": str(script_path) - } - - except Exception as e: - decky.logger.error(f"Error reading launch script: {e}") - return { - "success": False, - "error": str(e) - } - - async def check_fgmod_directory(self) -> Dict[str, Any]: - """Check if the fgmod directory exists in the home directory - - Returns: - Dict with exists status and directory path - """ - try: - home_path = Path(decky.DECKY_USER_HOME) - fgmod_path = home_path / "fgmod" - - exists = fgmod_path.exists() and fgmod_path.is_dir() - - return { - "success": True, - "exists": exists, - "path": str(fgmod_path) - } - - except Exception as e: - decky.logger.error(f"Error checking fgmod directory: {e}") - return { - "success": False, + 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, - "error": str(e) + "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_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 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 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 check_flatpak_extension_status(self) -> Dict[str, Any]: - """Check status of lsfg-vk Flatpak runtime extensions - - Returns: - FlatpakExtensionStatus dict with installation status for both runtime versions - """ - return self.flatpak_service.get_extension_status() - - async def install_flatpak_extension(self, version: str) -> Dict[str, Any]: - """Install lsfg-vk Flatpak runtime extension - - Args: - version: Runtime version to install ("23.08" or "24.08") - - Returns: - BaseResponse dict with success status and message/error - """ - return self.flatpak_service.install_extension(version) - - async def uninstall_flatpak_extension(self, version: str) -> Dict[str, Any]: - """Uninstall lsfg-vk Flatpak runtime extension - - Args: - version: Runtime version to uninstall ("23.08" or "24.08") - - Returns: - BaseResponse dict with success status and message/error - """ - return self.flatpak_service.uninstall_extension(version) - - async def get_flatpak_apps(self) -> Dict[str, Any]: - """Get list of installed Flatpak apps and their lsfg-vk override status - - Returns: - FlatpakAppInfo dict with apps list and override status - """ - return self.flatpak_service.get_flatpak_apps() - - async def set_flatpak_app_override(self, app_id: str) -> Dict[str, Any]: - """Set lsfg-vk overrides for a Flatpak app - - Args: - app_id: Flatpak application ID - - Returns: - FlatpakOverrideResponse dict with operation result - """ - return self.flatpak_service.set_app_override(app_id) - - async def remove_flatpak_app_override(self, app_id: str) -> Dict[str, Any]: - """Remove lsfg-vk overrides for a Flatpak app - - Args: - app_id: Flatpak application ID - - Returns: - FlatpakOverrideResponse dict with operation result - """ - return self.flatpak_service.remove_app_override(app_id) - 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 - self.installation_service.cleanup_on_uninstall() - - # Also clean up flatpak extensions if they are installed try: - decky.logger.info("Checking for flatpak extensions to uninstall") - - extension_status = self.flatpak_service.get_extension_status() - - if extension_status.get("success"): - if extension_status.get("installed_23_08"): - decky.logger.info("Uninstalling lsfg-vk flatpak runtime 23.08") - result = self.flatpak_service.uninstall_extension("23.08") - if result.get("success"): - decky.logger.info("Successfully uninstalled flatpak runtime 23.08") - else: - decky.logger.warning(f"Failed to uninstall flatpak runtime 23.08: {result.get('error')}") - - if extension_status.get("installed_24_08"): - decky.logger.info("Uninstalling lsfg-vk flatpak runtime 24.08") - result = self.flatpak_service.uninstall_extension("24.08") - if result.get("success"): - decky.logger.info("Successfully uninstalled flatpak runtime 24.08") - else: - decky.logger.warning(f"Failed to uninstall flatpak runtime 24.08: {result.get('error')}") - - decky.logger.info("Flatpak extension cleanup completed") - else: - decky.logger.info(f"Could not check flatpak status for cleanup: {extension_status.get('error')}") - - except Exception as e: - decky.logger.error(f"Error during flatpak cleanup: {e}") - + 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 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") |