diff options
Diffstat (limited to 'py_modules/lsfg_vk/installation.py')
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 623 |
1 files changed, 218 insertions, 405 deletions
diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index ce47268..e7366ee 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -1,443 +1,256 @@ -""" -Installation service for lsfg-vk. -""" - import os -import platform import shutil -import traceback -import zipfile +import tarfile import tempfile -import json +import traceback from pathlib import Path -from typing import Dict, Any +from typing import Dict from .base_service import BaseService +from .config_schema import ConfigurationManager, ProfileData from .constants import ( - LIB_FILENAME, JSON_FILENAME, ZIP_FILENAME, BIN_DIR, - SO_EXT, JSON_EXT, ARM_LIB_FILENAME, ARMADA_DEVICE_ENV + ARCHIVE_FILENAME, + BIN_DIR, + CLI_FILENAME, + JSON_FILENAME, + JSON_X86_FILENAME, + LEGACY_JSON_FILENAME, + LEGACY_LIB_FILENAME, + LIB_FILENAME, + LIB_X86_FILENAME, + LOCAL_SHARE, + UI_DESKTOP_FILENAME, + UI_FILENAME, + UI_ICON_FILENAME, ) -from .config_schema import ConfigurationManager -from .types import InstallationResponse, UninstallationResponse, InstallationCheckResponse +from .runtime_service import RuntimeService +from .steam_service import SteamService +from .types import InstallationCheckResponse, InstallationResponse, UninstallationResponse class InstallationService(BaseService): - """Service for handling lsfg-vk installation and uninstallation""" - - def __init__(self, logger=None): + def __init__( + self, + logger=None, + runtime_service: RuntimeService = None, + steam_service: SteamService = None, + ): super().__init__(logger) - + self.runtime_service = runtime_service or RuntimeService(logger=self.log) + self.steam_service = steam_service or SteamService(logger=self.log) self.lib_file = self.local_lib_dir / LIB_FILENAME + self.lib_x86_file = self.local_lib_dir / LIB_X86_FILENAME self.json_file = self.local_share_dir / JSON_FILENAME - + self.json_x86_file = self.local_share_dir / JSON_X86_FILENAME + self.cli_file = self.local_bin_dir / CLI_FILENAME + self.legacy_lib_file = self.local_lib_dir / LEGACY_LIB_FILENAME + self.legacy_json_file = self.local_share_dir / LEGACY_JSON_FILENAME + def install(self) -> InstallationResponse: - """Install lsfg-vk by extracting the zip file to ~/.local - - Returns: - InstallationResponse with success status and message/error - """ try: - plugin_dir = Path(__file__).parent.parent.parent - zip_path = plugin_dir / BIN_DIR / ZIP_FILENAME - - if not zip_path.exists(): - error_msg = f"{ZIP_FILENAME} not found at {zip_path}" - self.log.error(error_msg) - return self._error_response(InstallationResponse, error_msg, message="") - + 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() - - self._extract_and_install_files(zip_path) - - # If on ARM, overwrite the .so with the ARM version - if self._is_arm_architecture(): - self.log.info("Detected ARM architecture, using ARM binary") - arm_so_path = plugin_dir / BIN_DIR / ARM_LIB_FILENAME - self._copy_plugin_file(arm_so_path, self.lib_file) - self.log.info(f"Overwrote with ARM binary: {self.lib_file}") + profile_data = self._prepare_config() + self._install_archive(archive_path) + 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)) - self._create_config_file() - - self._create_lsfg_launch_script() - - self.log.info("lsfg-vk installed successfully") - return self._success_response(InstallationResponse, "lsfg-vk installed successfully") - - except (OSError, zipfile.BadZipFile, shutil.Error) as e: - error_msg = f"Error installing lsfg-vk: {str(e)}" - self.log.error(error_msg) - return self._error_response(InstallationResponse, str(e), message="") - except Exception as e: - error_msg = f"Unexpected error installing lsfg-vk: {str(e)}" - self.log.error(error_msg) - return self._error_response(InstallationResponse, str(e), message="") - - def _is_arm_architecture(self) -> bool: - """Check if running on ARM architecture + def _payload_destinations(self) -> Dict[str, tuple[Path, int]]: + return { + f"bin/{CLI_FILENAME}": (self.cli_file, 0o755), + f"bin/{UI_FILENAME}": (self.local_bin_dir / UI_FILENAME, 0o755), + f"lib/{LIB_FILENAME}": (self.lib_file, 0o644), + f"lib/{LIB_X86_FILENAME}": (self.lib_x86_file, 0o644), + f"share/vulkan/implicit_layer.d/{JSON_FILENAME}": (self.json_file, 0o644), + f"share/vulkan/implicit_layer.d/{JSON_X86_FILENAME}": (self.json_x86_file, 0o644), + f"share/applications/{UI_DESKTOP_FILENAME}": ( + self.user_home / LOCAL_SHARE / "applications" / UI_DESKTOP_FILENAME, + 0o644, + ), + f"share/icons/hicolor/256x256/apps/{UI_ICON_FILENAME}": ( + self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps" / UI_ICON_FILENAME, + 0o644, + ), + } - Returns: - True if running on ARM (aarch64), False otherwise - """ - if platform.machine().lower() in ('aarch64', 'arm64'): - return True + def _install_archive(self, archive_path: Path) -> None: + destinations = self._payload_destinations() + found = set() + with tarfile.open(archive_path, "r:*") as archive: + members = { + member.name.removeprefix("./"): member + for member in archive.getmembers() + if member.isfile() + } + for source_path, (destination, mode) in destinations.items(): + member = members.get(source_path) + if member is None: + continue + source = archive.extractfile(member) + if source is None: + continue + destination.parent.mkdir(parents=True, exist_ok=True) + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=destination.parent, + prefix=f".{destination.name}.", + delete=False, + ) as temporary_file: + temporary_path = Path(temporary_file.name) + with source: + shutil.copyfileobj(source, temporary_file) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + temporary_path.chmod(mode) + os.replace(temporary_path, destination) + found.add(source_path) + except Exception: + 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)) - # Decky runs through FEX on Armada, so Python reports x86_64 even - # though the host is AArch64. Armada exposes this native helper only - # on its ARM image, including inside Decky's FEX rootfs. - if ARMADA_DEVICE_ENV.is_file(): - self.log.info("Detected native AArch64 Armada host through device-env") - return True + def _default_config(self) -> ProfileData: + defaults = ConfigurationManager.get_defaults() + return ProfileData( + profiles={}, + global_config={"dll": defaults["dll"], "no_fp16": defaults["no_fp16"]}, + ) - # Fall back to the native PID 1 ELF header. e_machine 183 is AArch64. + def _prepare_config(self) -> ProfileData: try: - with Path('/proc/1/exe').open('rb') as host_executable: - elf_header = host_executable.read(20) - if elf_header[:4] == b'\x7fELF' and elf_header[5] in (1, 2): - byte_order = 'little' if elf_header[5] == 1 else 'big' - if int.from_bytes(elf_header[18:20], byte_order) == 183: - self.log.info("Detected native AArch64 host through PID 1") - return True - except OSError as e: - self.log.debug(f"Could not inspect native host architecture: {e}") + 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 = 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 + def _resolve_dll_path(self, profile_data: ProfileData) -> bool: + 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 + return True return False - @staticmethod - def _copy_plugin_file(src_file: Path, dst_file: Path) -> None: - """Copy plugin content without preserving FEX-incompatible metadata.""" - shutil.copyfile(src_file, dst_file) - dst_file.chmod(0o644) + def _remove_legacy_layer_files(self) -> None: + for path in (self.legacy_lib_file, self.legacy_json_file): + self._remove_if_exists(path) - def _extract_and_install_files(self, zip_path: Path) -> None: - """Extract zip file and install files to appropriate locations - - Args: - zip_path: Path to the zip file to extract - - Raises: - zipfile.BadZipFile: If zip file is corrupted - OSError: If file operations fail - """ - # Destination mapping for file types - dest_map = { - SO_EXT: self.local_lib_dir, - JSON_EXT: self.local_share_dir - } - - with zipfile.ZipFile(zip_path, 'r') as zip_ref: - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - zip_ref.extractall(temp_path) - - # Process extracted files - for root, dirs, files in os.walk(temp_path): - root_path = Path(root) - for file in files: - src_file = root_path / file - file_path = Path(file) - - # Check if we know where this file type should go - dst_dir = dest_map.get(file_path.suffix) - if dst_dir: - dst_file = dst_dir / file - - # Special handling for JSON files - need to modify library_path - if file_path.suffix == JSON_EXT and file == JSON_FILENAME: - self._copy_and_fix_json_file(src_file, dst_file) - else: - self._copy_plugin_file(src_file, dst_file) - - self.log.info(f"Copied {file} to {dst_file}") - - def _copy_and_fix_json_file(self, src_file: Path, dst_file: Path) -> None: - """Copy JSON file and fix the library_path to use relative path - - Args: - src_file: Source JSON file path - dst_file: Destination JSON file path - """ - try: - # Read the JSON file - with open(src_file, 'r') as f: - json_data = json.load(f) - - # Fix the library_path from "liblsfg-vk.so" to "../../../lib/liblsfg-vk.so" - if 'layer' in json_data and 'library_path' in json_data['layer']: - current_path = json_data['layer']['library_path'] - if current_path == "liblsfg-vk.so": - json_data['layer']['library_path'] = "../../../lib/liblsfg-vk.so" - self.log.info(f"Fixed library_path from '{current_path}' to '../../../lib/liblsfg-vk.so'") - - # Write the modified JSON file - with open(dst_file, 'w') as f: - json.dump(json_data, f, indent=2) - - except (json.JSONDecodeError, KeyError, OSError) as e: - self.log.error(f"Error fixing JSON file {src_file}: {e}") - # Fallback to simple copy if JSON modification fails - self._copy_plugin_file(src_file, dst_file) - - def _create_config_file(self) -> None: - """Create or update the TOML config file in ~/.config/lsfg-vk with default configuration and detected DLL path - - If a config file already exists, preserve existing profiles and only update global settings like DLL path. - """ - # Import here to avoid circular imports - from .dll_detection import DllDetectionService - - # Try to detect DLL path - dll_service = DllDetectionService(self.log) - - # Check if config file already exists - if self.config_file_path.exists(): + def _prune_empty_directories(self) -> None: + # Only prune directories created by this plugin. Never remove a + # non-empty directory because the user's other tools may use it. + candidates = ( + self.config_dir, + self.local_bin_dir, + self.local_lib_dir, + self.local_share_dir, + self.user_home / LOCAL_SHARE / "applications", + self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps", + ) + for directory in candidates: try: - # Read existing config to preserve user profiles - content = self.config_file_path.read_text(encoding='utf-8') - existing_profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) - self.log.info(f"Found existing config file, preserving user profiles") - - # Create merged profile data that preserves user settings but adds any new fields - merged_profile_data = self._merge_config_with_defaults(existing_profile_data, dll_service) - - # Generate TOML content with merged profiles - toml_content = ConfigurationManager.generate_toml_content_multi_profile(merged_profile_data) - - except Exception as e: - self.log.warning(f"Failed to parse existing config file: {str(e)}, creating new one") - # Fall back to creating a new config file - config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - toml_content = ConfigurationManager.generate_toml_content(config) - else: - # No existing config file, create a new one with defaults - config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - toml_content = ConfigurationManager.generate_toml_content(config) - self.log.info(f"Creating new config file") - - # Write config file - self._write_file(self.config_file_path, toml_content, 0o644) - self.log.info(f"Created config file at {self.config_file_path}") - - # Log detected DLL path if found - USE GENERATED CONSTANTS - from .config_schema_generated import DLL - try: - # Try to parse the written content to get the DLL path - final_content = self.config_file_path.read_text(encoding='utf-8') - final_config = ConfigurationManager.parse_toml_content(final_content) - if final_config.get(DLL): - self.log.info(f"Configured DLL path: {final_config[DLL]}") - except (OSError, IOError, ValueError, KeyError) as e: - # Don't fail installation if we can't log the DLL path - self.log.debug(f"Could not log DLL path: {e}") - - def _create_lsfg_launch_script(self) -> None: - """Create the ~/lsfg launch script for easier game setup""" - # Use the default configuration for the initial script - from .config_schema import ConfigurationManager - default_config = ConfigurationManager.get_defaults() - - # Create configuration service to generate the script - from .configuration import ConfigurationService - config_service = ConfigurationService(logger=self.log) - config_service.user_home = self.user_home - config_service.lsfg_script_path = self.lsfg_launch_script_path - - # Generate script content with default configuration - script_content = config_service._generate_script_content(default_config) - - # Write the script file - self._write_file(self.lsfg_launch_script_path, script_content, 0o755) - self.log.info(f"Created lsfg launch script at {self.lsfg_launch_script_path}") - - def get_launch_script_path(self) -> str: - """Get the path to the lsfg launch script - - Returns: - String path to the launch script file - """ - return str(self.lsfg_launch_script_path) + if directory.is_dir() and not directory.is_symlink(): + directory.rmdir() + self.log.info(f"Removed empty directory {directory}") + except OSError: + continue def check_installation(self) -> InstallationCheckResponse: - """Check if lsfg-vk is already installed - - Returns: - InstallationCheckResponse with installation status and file paths - """ try: - lib_exists = self.lib_file.exists() - json_exists = self.json_file.exists() - config_exists = self.config_file_path.exists() - - self.log.info(f"Installation check: lib={lib_exists}, json={json_exists}, config={config_exists}") - + installation_error = None + try: + installed = self.runtime_service.is_healthy() + except Exception as error: + installed = False + installation_error = str(error) + lossless_scaling = self.runtime_service.check_lossless_scaling() return { - "installed": lib_exists and json_exists, - "lib_exists": lib_exists, - "json_exists": json_exists, - "script_exists": config_exists, # Keep script_exists for backward compatibility - "lib_path": str(self.lib_file), - "json_path": str(self.json_file), - "script_path": str(self.config_file_path), # Keep script_path for backward compatibility - "error": None + "installed": installed, + "lossless_scaling_installed": bool(lossless_scaling["installed"]), + "lossless_scaling_status": str(lossless_scaling["status"]), + "error": installation_error, } - - except Exception as e: - error_msg = f"Error checking lsfg-vk installation: {str(e)}" - self.log.error(error_msg) + except Exception as error: return { "installed": False, - "lib_exists": False, - "json_exists": False, - "script_exists": False, - "lib_path": str(self.lib_file), - "json_path": str(self.json_file), - "script_path": str(self.config_file_path), - "error": str(e) + "lossless_scaling_installed": False, + "lossless_scaling_status": str(error), + "error": str(error), } - + def uninstall(self) -> UninstallationResponse: - """Uninstall lsfg-vk by removing the installed files - - Note: The config file (conf.toml) is preserved to maintain user's custom profiles - - Returns: - UninstallationResponse with success status and removed files list - """ - try: - removed_files = [] - # Remove core lsfg-vk files, but preserve config file to maintain user's custom profiles - files_to_remove = [self.lib_file, self.json_file, self.lsfg_launch_script_path] - - for file_path in files_to_remove: - if self._remove_if_exists(file_path): - removed_files.append(str(file_path)) - - # Also try to remove the old script file if it exists (for backward compatibility) - if self._remove_if_exists(self.lsfg_script_path): - removed_files.append(str(self.lsfg_script_path)) - - # Don't remove config directory since we're preserving the config file - - if not removed_files: - return self._success_response(UninstallationResponse, - "No lsfg-vk files found to remove", - removed_files=None) - - self.log.info("lsfg-vk uninstalled successfully") - return self._success_response(UninstallationResponse, - f"lsfg-vk uninstalled successfully. Removed {len(removed_files)} files.", - removed_files=removed_files) - - except OSError as e: - error_msg = f"Error uninstalling lsfg-vk: {str(e)}" - self.log.error(error_msg) - return self._error_response(UninstallationResponse, str(e), - message="", removed_files=None) - - def cleanup_on_uninstall(self) -> None: - """Clean up lsfg-vk files when the plugin is uninstalled - - Note: The config file (conf.toml) is preserved to maintain user's custom profiles - """ try: - self.log.info("Checking for lsfg-vk files to clean up:") - self.log.info(f" Library file: {self.lib_file}") - self.log.info(f" JSON file: {self.json_file}") - self.log.info(f" Config file: {self.config_file_path} (preserved)") - self.log.info(f" Launch script: {self.lsfg_launch_script_path}") - self.log.info(f" Old script file: {self.lsfg_script_path}") - - removed_files = [] - # Remove core lsfg-vk files, but preserve config file to maintain user's custom profiles - files_to_remove = [self.lib_file, self.json_file, self.lsfg_launch_script_path, self.lsfg_script_path] - - for file_path in files_to_remove: - try: - if self._remove_if_exists(file_path): - removed_files.append(str(file_path)) - except OSError as e: - self.log.error(f"Failed to remove {file_path}: {e}") - - # Don't remove config directory since we're preserving the config file - - if removed_files: - self.log.info(f"Cleaned up {len(removed_files)} lsfg-vk files during plugin uninstall: {removed_files}") - else: - self.log.info("No lsfg-vk files found to clean up during plugin uninstall") - - except Exception as e: - self.log.error(f"Error cleaning up lsfg-vk files during uninstall: {str(e)}") - self.log.error(f"Traceback: {traceback.format_exc()}") + 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, + self.legacy_script_path, + self.config_file_path, + ) + if self._remove_if_exists(path) + ] + self._prune_empty_directories() + if not removed: + return self._success_response( + UninstallationResponse, + "No lsfg-vk files found to remove", + removed_files=None, + ) + return self._success_response( + UninstallationResponse, + f"lsfg-vk uninstalled successfully. Removed {len(removed)} files.", + removed_files=removed, + ) + except Exception as error: + return self._error_response( + UninstallationResponse, + str(error), + removed_files=None, + ) - def _merge_config_with_defaults(self, existing_profile_data, dll_service): - """Merge existing user config with current schema defaults - - This ensures that: - 1. User's custom profiles and values are preserved - 2. Any new fields added to the schema get their default values - 3. Global settings like DLL path are updated as needed - - Args: - existing_profile_data: The user's existing ProfileData - dll_service: DLL detection service for updating DLL path - - Returns: - ProfileData with merged configuration - """ - from .config_schema import ProfileData - - # Get current schema defaults - default_config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - default_global_config = { - "dll": default_config.get("dll", ""), - "no_fp16": False - } - - # Start with existing data - merged_data: ProfileData = { - "current_profile": existing_profile_data.get("current_profile", "decky-lsfg-vk"), - "global_config": existing_profile_data.get("global_config", {}).copy(), - "profiles": {} - } - - # Merge global config: preserve user values, add missing fields, update DLL - for key, default_value in default_global_config.items(): - if key not in merged_data["global_config"]: - merged_data["global_config"][key] = default_value - self.log.info(f"Added missing global field '{key}' with default value: {default_value}") - - # Update DLL path if detected - dll_result = dll_service.check_lossless_scaling_dll() - if dll_result.get("detected") and dll_result.get("path"): - old_dll = merged_data["global_config"].get("dll") - merged_data["global_config"]["dll"] = dll_result["path"] - if old_dll != dll_result["path"]: - self.log.info(f"Updated DLL path from '{old_dll}' to: {dll_result['path']}") - - # Merge each profile: preserve user values, add missing fields - existing_profiles = existing_profile_data.get("profiles", {}) - - for profile_name, existing_profile_config in existing_profiles.items(): - merged_profile_config = existing_profile_config.copy() - - # Add any missing fields from current schema with default values - added_fields = [] - for key, default_value in default_config.items(): - if key not in merged_profile_config and key not in ["dll", "no_fp16"]: # Skip global fields - merged_profile_config[key] = default_value - added_fields.append(key) - - if added_fields: - self.log.info(f"Profile '{profile_name}': Added missing fields {added_fields}") - - merged_data["profiles"][profile_name] = merged_profile_config - - # If no profiles exist, create the default one - if not merged_data["profiles"]: - merged_data["profiles"]["decky-lsfg-vk"] = { - k: v for k, v in default_config.items() - if k not in ["dll", "no_fp16"] # Exclude global fields - } - merged_data["current_profile"] = "decky-lsfg-vk" - self.log.info("No existing profiles found, created default profile") - - return merged_data + def cleanup_on_uninstall(self) -> bool: + try: + result = self.uninstall() + if not result.get("success"): + self.log.error(f"Error cleaning up lsfg-vk files during uninstall: {result.get('error')}") + return False + return True + except Exception as error: + self.log.error(f"Error cleaning up lsfg-vk files during uninstall: {error}") + self.log.error(traceback.format_exc()) + return False |