From 4fe21381b6c694d5eb615447c4827e4921593a52 Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:30:42 -0400 Subject: feat: migrate native runtime to lsfg-vk v2 --- py_modules/lsfg_vk/base_service.py | 138 ++--- py_modules/lsfg_vk/config_schema.py | 784 +++++++++----------------- py_modules/lsfg_vk/config_schema_generated.py | 4 +- py_modules/lsfg_vk/configuration.py | 545 ++++++------------ py_modules/lsfg_vk/constants.py | 27 +- py_modules/lsfg_vk/installation.py | 555 +++++++----------- 6 files changed, 687 insertions(+), 1366 deletions(-) (limited to 'py_modules/lsfg_vk') diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py index 262e2b0..c4978c6 100644 --- a/py_modules/lsfg_vk/base_service.py +++ b/py_modules/lsfg_vk/base_service.py @@ -1,132 +1,66 @@ -""" -Base service class with common functionality. -""" - -import logging import os -import shutil +import tempfile from pathlib import Path -from typing import Any, Optional, TypeVar, Dict +from typing import Any, Dict, Optional, TypeVar import decky -from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME +from .constants import CONFIG_DIR, CONFIG_FILENAME, LOCAL_BIN, LOCAL_LIB, SCRIPT_NAME, VULKAN_LAYER_DIR -ResponseType = TypeVar('ResponseType', bound=Dict[str, Any]) +ResponseType = TypeVar("ResponseType", bound=Dict[str, Any]) class BaseService: - """Base service class with common functionality""" - def __init__(self, logger: Optional[Any] = None): - """Initialize base service - - Args: - logger: Logger instance, defaults to decky.logger if None - """ - if logger is None: - self.log = decky.logger - else: - self.log = logger - + self.log = decky.logger if logger is None else logger self.user_home = Path.home() + self.local_bin_dir = self.user_home / LOCAL_BIN self.local_lib_dir = self.user_home / LOCAL_LIB self.local_share_dir = self.user_home / VULKAN_LAYER_DIR self.lsfg_script_path = self.user_home / SCRIPT_NAME self.lsfg_launch_script_path = self.user_home / SCRIPT_NAME self.config_dir = self.user_home / CONFIG_DIR self.config_file_path = self.config_dir / CONFIG_FILENAME - + def _ensure_directories(self) -> None: - """Create necessary directories if they don't exist""" - self.local_lib_dir.mkdir(parents=True, exist_ok=True) - self.local_share_dir.mkdir(parents=True, exist_ok=True) - self.config_dir.mkdir(parents=True, exist_ok=True) - self.log.info(f"Ensured directories exist: {self.local_lib_dir}, {self.local_share_dir}, {self.config_dir}") - + for directory in (self.local_bin_dir, self.local_lib_dir, self.local_share_dir, self.config_dir): + directory.mkdir(parents=True, exist_ok=True) + def _remove_if_exists(self, path: Path) -> bool: - """Remove a file if it exists - - Args: - path: Path to the file to remove - - Returns: - True if file was removed, False if it didn't exist - - Raises: - OSError: If removal fails - """ - if path.exists(): - try: - path.unlink() - self.log.info(f"Removed {path}") - return True - except OSError as e: - self.log.error(f"Failed to remove {path}: {e}") - raise - else: - self.log.info(f"File not found: {path}") + if not path.exists() and not path.is_symlink(): return False - + path.unlink() + self.log.info(f"Removed {path}") + return True + def _write_file(self, path: Path, content: str, mode: int = 0o644) -> None: - """Write content to a file - - Args: - path: Target file path - content: Content to write - mode: File permissions (default: 0o644) - - Raises: - OSError: If write fails - """ + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = None try: - with open(path, 'w', encoding='utf-8') as f: - f.write(content) - f.flush() - os.fsync(f.fileno()) - - path.chmod(mode) - self.log.info(f"Wrote to {path}") - - except (OSError, IOError, PermissionError) as e: - self.log.error(f"Failed to write to {path}: {e}") + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.", + delete=False, + ) as temporary_file: + temporary_path = Path(temporary_file.name) + temporary_file.write(content) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + temporary_path.chmod(mode) + os.replace(temporary_path, path) + except Exception: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) raise def _success_response(self, response_type: type, message: str = "", **kwargs) -> Any: - """Create a standardized success response - - Args: - response_type: The TypedDict response type to create - message: Success message - **kwargs: Additional response fields - - Returns: - Success response dict - """ - response = { - "success": True, - "message": message, - "error": None - } + response = {"success": True, "message": message, "error": None} response.update(kwargs) return response - + def _error_response(self, response_type: type, error: str, message: str = "", **kwargs) -> Any: - """Create a standardized error response - - Args: - response_type: The TypedDict response type to create - error: Error description - message: Optional message - **kwargs: Additional response fields - - Returns: - Error response dict - """ - response = { - "success": False, - "message": message, - "error": error - } + response = {"success": False, "message": message, "error": error} response.update(kwargs) return response diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py index f76b5f2..86e8be7 100644 --- a/py_modules/lsfg_vk/config_schema.py +++ b/py_modules/lsfg_vk/config_schema.py @@ -1,605 +1,357 @@ -""" -Centralized configuration schema for lsfg-vk. - -This module defines the complete configuration structure for lsfg-vk, managing TOML-based config files, including: -- Field definitions with types, defaults, and metadata -- TOML generation logic -- Validation rules -- Type definitions -""" - -import logging +import json import re +import shlex import sys -from typing import TypedDict, Dict, Any, Union, cast, List +import tomllib from dataclasses import dataclass -from enum import Enum from pathlib import Path +from typing import Any, Dict, TypedDict, Union, cast -# Import shared configuration constants sys.path.insert(0, str(Path(__file__).parent.parent.parent)) -from shared_config import CONFIG_SCHEMA_DEF, ConfigFieldType, get_field_names, get_defaults, get_field_types - -# Import auto-generated configuration components -from .config_schema_generated import ConfigurationData, get_script_parsing_logic, get_script_generation_logic +from shared_config import CONFIG_SCHEMA_DEF, ConfigFieldType, get_defaults +from .config_schema_generated import ConfigurationData, get_script_parsing_logic @dataclass class ConfigField: - """Configuration field definition""" name: str field_type: ConfigFieldType default: Union[bool, int, float, str] description: str - - def get_toml_value(self, value: Union[bool, int, float, str]) -> Union[bool, int, float, str]: - """Get the value for TOML output""" - return value -# Use shared configuration schema as source of truth CONFIG_SCHEMA: Dict[str, ConfigField] = { - field_name: ConfigField( - name=field_def["name"], - field_type=ConfigFieldType(field_def["fieldType"]), - default=field_def["default"], - description=field_def["description"] + name: ConfigField( + name=definition["name"], + field_type=ConfigFieldType(definition["fieldType"]), + default=definition["default"], + description=definition["description"], ) - for field_name, field_def in CONFIG_SCHEMA_DEF.items() + for name, definition in CONFIG_SCHEMA_DEF.items() } -# Override DLL default to empty (will be populated dynamically) -CONFIG_SCHEMA["dll"] = ConfigField( - name="dll", - field_type=ConfigFieldType.STRING, - default="", # Will be populated dynamically based on detection - description="specify where Lossless.dll is stored" -) - -# Get script-only fields dynamically from shared config SCRIPT_ONLY_FIELDS = { - field_name: ConfigField( - name=field_def["name"], - field_type=ConfigFieldType(field_def["fieldType"]), - default=field_def["default"], - description=field_def["description"] - ) - for field_name, field_def in CONFIG_SCHEMA_DEF.items() - if field_def.get("location") == "script" + name + for name, definition in CONFIG_SCHEMA_DEF.items() + if definition["location"] == "script" } - -# Complete configuration schema (TOML + script-only fields) -COMPLETE_CONFIG_SCHEMA = {**CONFIG_SCHEMA, **SCRIPT_ONLY_FIELDS} - -# Constants for profile management DEFAULT_PROFILE_NAME = "decky-lsfg-vk" -GLOBAL_SECTION_FIELDS = {"dll", "no_fp16"} - -# Note: ConfigurationData is now imported from generated file -# No need to manually maintain the TypedDict anymore! class ProfileData(TypedDict): - """Profile data with current profile tracking""" current_profile: str - profiles: Dict[str, ConfigurationData] # profile_name -> config - global_config: Dict[str, Any] # Global settings (dll, no_fp16) + profiles: Dict[str, Dict[str, Any]] + global_config: Dict[str, Any] + + +def _toml_value(value: Any) -> str: + if isinstance(value, bool): + return str(value).lower() + if isinstance(value, str): + return json.dumps(value) + if isinstance(value, list): + return "[ " + ", ".join(_toml_value(item) for item in value) + " ]" + return str(value) class ConfigurationManager: - """Centralized configuration management""" - @staticmethod def get_defaults() -> ConfigurationData: - """Get default configuration values""" - # Use shared defaults and add script-only fields - shared_defaults = get_defaults() - - # Add script-only fields that aren't in the shared schema - script_defaults = { - field.name: field.default - for field in SCRIPT_ONLY_FIELDS.values() - } - - return cast(ConfigurationData, {**shared_defaults, **script_defaults}) - + return cast(ConfigurationData, dict(get_defaults())) + @staticmethod def get_defaults_with_dll_detection(dll_detection_service=None) -> ConfigurationData: - """Get default configuration values with DLL path detection - - Args: - dll_detection_service: Optional DLL detection service instance - - Returns: - ConfigurationData with detected DLL path if available - """ defaults = ConfigurationManager.get_defaults() - - # Try to detect DLL path if service provided - if dll_detection_service: - try: - dll_result = dll_detection_service.check_lossless_scaling_dll() - if dll_result.get("detected") and dll_result.get("path"): - defaults["dll"] = dll_result["path"] - except (OSError, IOError, KeyError, TypeError) as e: - # If detection fails, keep empty default - logging.getLogger(__name__).debug(f"DLL detection failed: {e}") - - # If DLL path is still empty, use a reasonable fallback - if not defaults["dll"]: - defaults["dll"] = "/home/deck/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - + if dll_detection_service is not None: + result = dll_detection_service.check_lossless_scaling_dll() + if result.get("detected") and result.get("path"): + defaults["dll"] = result["path"] return defaults - + @staticmethod def get_field_names() -> list[str]: - """Get ordered list of configuration field names""" - # Use shared field names and add script-only fields - shared_names = get_field_names() - script_names = list(SCRIPT_ONLY_FIELDS.keys()) - return shared_names + script_names - + return list(CONFIG_SCHEMA) + @staticmethod def get_field_types() -> Dict[str, ConfigFieldType]: - """Get field type mapping""" - # Use shared field types and add script-only field types - shared_types = {name: ConfigFieldType(type_str) for name, type_str in get_field_types().items()} - script_types = {field.name: field.field_type for field in SCRIPT_ONLY_FIELDS.values()} - return {**shared_types, **script_types} - + return {name: field.field_type for name, field in CONFIG_SCHEMA.items()} + @staticmethod def validate_config(config: Dict[str, Any]) -> ConfigurationData: - """Validate and convert configuration data""" - validated = {} - - for field_name, field_def in COMPLETE_CONFIG_SCHEMA.items(): - value = config.get(field_name, field_def.default) - - # Type validation and conversion - if field_def.field_type == ConfigFieldType.BOOLEAN: - validated[field_name] = bool(value) - elif field_def.field_type == ConfigFieldType.INTEGER: - validated[field_name] = int(value) - elif field_def.field_type == ConfigFieldType.FLOAT: - validated[field_name] = float(value) - elif field_def.field_type == ConfigFieldType.STRING: - validated[field_name] = str(value) + validated: Dict[str, Any] = {} + for name, field in CONFIG_SCHEMA.items(): + value = config.get(name, field.default) + if field.field_type == ConfigFieldType.BOOLEAN: + value = value.lower() in {"true", "1", "yes", "on"} if isinstance(value, str) else bool(value) + elif field.field_type == ConfigFieldType.INTEGER: + value = int(value) + elif field.field_type == ConfigFieldType.FLOAT: + value = float(value) else: - validated[field_name] = value - + value = str(value) + validated[name] = value + + if validated["multiplier"] < 1: + raise ValueError("multiplier must be 1 or greater") + if not 0.25 <= validated["flow_scale"] <= 1.0: + raise ValueError("flow_scale must be between 0.25 and 1.0") + if validated["experimental_present_mode"] not in {"fifo", "mailbox"}: + raise ValueError("experimental_present_mode must be fifo or mailbox") return cast(ConfigurationData, validated) - + + @staticmethod + def _migrate_dll_path(value: Any) -> str: + path_value = str(value or "") + if not path_value: + return "" + path = Path(path_value) + if path.name.lower() != "lossless.dll": + return path_value + replacement = path.with_name("lsfg-vk.dll") + return str(replacement) if replacement.exists() else "" + + @staticmethod + def _config_from_profile(profile: Dict[str, Any], global_config: Dict[str, Any]) -> Dict[str, Any]: + config: Dict[str, Any] = dict(ConfigurationManager.get_defaults()) + for field in ("multiplier", "flow_scale", "performance_mode"): + if field in profile: + config[field] = profile[field] + config["experimental_present_mode"] = "fifo" if bool(profile.get("override_present_mode", True)) else "mailbox" + config["dll"] = global_config.get("dll", "") + config["no_fp16"] = global_config.get("no_fp16", False) + for field in ("active_in", "pacing", "preserve_swapchain_image_count"): + if field in profile: + config[field] = profile[field] + return {**config, **ConfigurationManager.validate_config(config)} + @staticmethod def generate_toml_content(config: ConfigurationData) -> str: - """Generate TOML configuration file content for single profile (backward compatibility)""" - # For backward compatibility, create a single profile structure profile_data: ProfileData = { "current_profile": DEFAULT_PROFILE_NAME, - "profiles": {DEFAULT_PROFILE_NAME: config}, + "profiles": {DEFAULT_PROFILE_NAME: dict(config)}, "global_config": { "dll": config.get("dll", ""), - "no_fp16": config.get("no_fp16", False) - } + "no_fp16": config.get("no_fp16", False), + }, } return ConfigurationManager.generate_toml_content_multi_profile(profile_data) - + @staticmethod def generate_toml_content_multi_profile(profile_data: ProfileData) -> str: - """Generate TOML configuration file content with multiple profiles""" - lines = ["version = 1"] - lines.append("") - - # Add global section with global fields - lines.append("[global]") - - # Add current_profile field - lines.append(f"# Currently selected profile") - lines.append(f'current_profile = "{profile_data["current_profile"]}"') - lines.append("") - - # Add dll field if specified - dll_path = profile_data["global_config"].get("dll", "") - if dll_path: - lines.append(f"# specify where Lossless.dll is stored") - lines.append(f'dll = "{dll_path}"') - lines.append("") - - lines.append(f"# FP16 acceleration") - no_fp16 = bool(profile_data["global_config"].get("no_fp16", False)) - lines.append(f"no_fp16 = {str(no_fp16).lower()}") - lines.append("") - - # Add game sections for each profile - # Sort profiles to ensure consistent order (default profile first) - sorted_profiles = sorted(profile_data["profiles"].items(), - key=lambda x: (x[0] != DEFAULT_PROFILE_NAME, x[0])) - - for profile_name, config in sorted_profiles: - lines.append("[[game]]") - if profile_name == DEFAULT_PROFILE_NAME: - lines.append("# Plugin-managed game entry (default profile)") - else: - lines.append(f"# Profile: {profile_name}") - lines.append(f'exe = "{profile_name}"') - lines.append("") - - # Add all configuration fields to the game section (excluding global fields) - for field_name, field_def in CONFIG_SCHEMA.items(): - # Skip global fields - they go in global section - if field_name in GLOBAL_SECTION_FIELDS: - continue - - value = config.get(field_name, field_def.default) - - # Add field description comment - lines.append(f"# {field_def.description}") - - # Format value based on type - if isinstance(value, bool): - lines.append(f"{field_name} = {str(value).lower()}") - elif isinstance(value, str) and value: # Only add non-empty strings - lines.append(f'{field_name} = "{value}"') - elif isinstance(value, (int, float)): # Always include numbers, even if 0 or 1 - lines.append(f"{field_name} = {value}") - - lines.append("") # Empty line for readability - - return "\n".join(lines) - + global_config = profile_data["global_config"] + lines = ["version = 2", "", "[global]"] + dll = ConfigurationManager._migrate_dll_path(global_config.get("dll", "")) + if dll: + lines.append(f"dll = {_toml_value(dll)}") + lines.append(f"allow_fp16 = {_toml_value(not bool(global_config.get('no_fp16', False)))}") + if global_config.get("log_level"): + lines.append(f"log_level = {_toml_value(global_config['log_level'])}") + if global_config.get("log_file"): + lines.append(f"log_file = {_toml_value(global_config['log_file'])}") + + profiles = sorted( + profile_data["profiles"].items(), + key=lambda item: (item[0] != DEFAULT_PROFILE_NAME, item[0]), + ) + for profile_name, raw_config in profiles: + config = ConfigurationManager.validate_config(raw_config) + lines.extend(["", "[[profile]]", f"name = {_toml_value(profile_name)}"]) + active_in = raw_config.get("active_in") + if active_in not in (None, "", []): + lines.append(f"active_in = {_toml_value(active_in)}") + lines.extend( + [ + f"multiplier = {config['multiplier']}", + f"flow_scale = {config['flow_scale']}", + f"performance_mode = {_toml_value(config['performance_mode'])}", + f"pacing = {_toml_value(raw_config.get('pacing', 'vsync'))}", + f"override_present_mode = {_toml_value(config['experimental_present_mode'] == 'fifo')}", + f"preserve_swapchain_image_count = {_toml_value(bool(raw_config.get('preserve_swapchain_image_count', False)))}", + ] + ) + return "\n".join(lines) + "\n" + @staticmethod - def parse_toml_content(content: str) -> ConfigurationData: - """Parse TOML content into configuration data for the currently selected profile (backward compatibility)""" - profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) - current_profile = profile_data["current_profile"] - - # Merge global config with current profile config - current_config = profile_data["profiles"].get(current_profile, ConfigurationManager.get_defaults()) - - # Add global fields to the config - for field_name in GLOBAL_SECTION_FIELDS: - if field_name in profile_data["global_config"]: - current_config[field_name] = profile_data["global_config"][field_name] - - return current_config - + def _profile_data_from_v1(data: Dict[str, Any]) -> ProfileData: + old_global = dict(data.get("global", {})) + global_config: Dict[str, Any] = { + "dll": ConfigurationManager._migrate_dll_path(old_global.get("dll", "")), + "no_fp16": bool(old_global.get("no_fp16", False)), + } + profiles: Dict[str, Dict[str, Any]] = {} + for game in data.get("game", []): + profile_name = str(game.get("exe", DEFAULT_PROFILE_NAME)) + config: Dict[str, Any] = dict(ConfigurationManager.get_defaults()) + for field in ("multiplier", "flow_scale", "performance_mode", "experimental_present_mode"): + if field in game: + config[field] = game[field] + config["dll"] = global_config["dll"] + config["no_fp16"] = global_config["no_fp16"] + profiles[profile_name] = dict(ConfigurationManager.validate_config(config)) + + if not profiles: + profiles[DEFAULT_PROFILE_NAME] = dict(ConfigurationManager.get_defaults()) + + current_profile = str(old_global.get("current_profile", DEFAULT_PROFILE_NAME)) + if current_profile not in profiles: + current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles)) + return ProfileData( + current_profile=current_profile, + profiles=profiles, + global_config=global_config, + ) + @staticmethod - def parse_toml_content_multi_profile(content: str) -> ProfileData: - """Parse TOML content into profile data structure""" - profiles: Dict[str, ConfigurationData] = {} - global_config: Dict[str, Any] = {} - current_profile = DEFAULT_PROFILE_NAME - + def is_legacy_v1(content: str) -> bool: try: - # Look for both [global] and [[game]] sections - lines = content.split('\n') - in_global_section = False - in_game_section = False - current_game_exe = None - current_game_config: Dict[str, Any] = {} - - for line in lines: - line = line.strip() - - # Skip comments and empty lines - if not line or line.startswith('#'): - continue - - # Check for section headers - if line.startswith('[') and line.endswith(']'): - # Save previous game section if we were in one - if in_game_section and current_game_exe: - # Validate and store the profile config - validated_config = ConfigurationManager.get_defaults() - for key, value in current_game_config.items(): - if key in CONFIG_SCHEMA: - field_def = CONFIG_SCHEMA[key] - try: - if field_def.field_type == ConfigFieldType.BOOLEAN: - validated_config[key] = value - elif field_def.field_type == ConfigFieldType.INTEGER: - validated_config[key] = int(value) if not isinstance(value, int) else value - elif field_def.field_type == ConfigFieldType.FLOAT: - validated_config[key] = float(value) if not isinstance(value, float) else value - elif field_def.field_type == ConfigFieldType.STRING: - validated_config[key] = str(value) - except (ValueError, TypeError): - # If conversion fails, keep default value - pass - profiles[current_game_exe] = validated_config - current_game_config = {} - - # Set new section state - if line == '[global]': - in_global_section = True - in_game_section = False - elif line == '[[game]]': - in_global_section = False - in_game_section = True - current_game_exe = None - else: - in_global_section = False - in_game_section = False - continue - - # Parse key = value lines - if '=' in line: - key, value = line.split('=', 1) - key = key.strip() - value = value.strip() - - # Remove quotes from string values - if value.startswith('"') and value.endswith('"'): - value = value[1:-1] - elif value.startswith("'") and value.endswith("'"): - value = value[1:-1] - - # Handle global section - if in_global_section: - if key == "current_profile": - current_profile = value - elif key == "dll": - global_config["dll"] = value - elif key == "no_fp16": - global_config["no_fp16"] = value.lower() in ('true', '1', 'yes', 'on') - - # Handle game section - elif in_game_section: - # Track the exe for this game section - if key == "exe": - current_game_exe = value - # Store config fields for current game - elif key in CONFIG_SCHEMA: - field_def = CONFIG_SCHEMA[key] - try: - if field_def.field_type == ConfigFieldType.BOOLEAN: - current_game_config[key] = value.lower() in ('true', '1', 'yes', 'on') - elif field_def.field_type == ConfigFieldType.INTEGER: - current_game_config[key] = int(value) - elif field_def.field_type == ConfigFieldType.FLOAT: - current_game_config[key] = float(value) - elif field_def.field_type == ConfigFieldType.STRING: - current_game_config[key] = value - except (ValueError, TypeError): - # If conversion fails, keep default value - pass - - # Handle final game section if we were in one - if in_game_section and current_game_exe: - validated_config = ConfigurationManager.get_defaults() - for key, value in current_game_config.items(): - if key in CONFIG_SCHEMA: - field_def = CONFIG_SCHEMA[key] - try: - if field_def.field_type == ConfigFieldType.BOOLEAN: - validated_config[key] = value - elif field_def.field_type == ConfigFieldType.INTEGER: - validated_config[key] = int(value) if not isinstance(value, int) else value - elif field_def.field_type == ConfigFieldType.FLOAT: - validated_config[key] = float(value) if not isinstance(value, float) else value - elif field_def.field_type == ConfigFieldType.STRING: - validated_config[key] = str(value) - except (ValueError, TypeError): - # If conversion fails, keep default value - pass - profiles[current_game_exe] = validated_config - - # Ensure we have at least the default profile - if not profiles: - profiles[DEFAULT_PROFILE_NAME] = ConfigurationManager.get_defaults() - - # Ensure current_profile exists in profiles - if current_profile not in profiles: - current_profile = DEFAULT_PROFILE_NAME - if DEFAULT_PROFILE_NAME not in profiles: - profiles[DEFAULT_PROFILE_NAME] = ConfigurationManager.get_defaults() - - return ProfileData( - current_profile=current_profile, - profiles=profiles, - global_config=global_config - ) - - except (ValueError, KeyError, TypeError, AttributeError) as e: - # If parsing fails completely, return default profile structure - logging.getLogger(__name__).warning(f"Failed to parse TOML profiles, using defaults: {e}") - return ProfileData( - current_profile=DEFAULT_PROFILE_NAME, - profiles={DEFAULT_PROFILE_NAME: ConfigurationManager.get_defaults()}, - global_config={} - ) - + return tomllib.loads(content).get("version") == 1 + except tomllib.TOMLDecodeError: + return False + + @staticmethod + def parse_toml_content_multi_profile(content: str) -> ProfileData: + data = tomllib.loads(content) + version = data.get("version") + if version == 1: + return ConfigurationManager._profile_data_from_v1(data) + if version != 2: + raise ValueError("unsupported lsfg-vk configuration version") + + raw_global = dict(data.get("global", {})) + global_config: Dict[str, Any] = { + "dll": ConfigurationManager._migrate_dll_path(raw_global.get("dll", "")), + "no_fp16": not bool(raw_global.get("allow_fp16", True)), + } + for field in ("log_level", "log_file"): + if field in raw_global: + global_config[field] = raw_global[field] + + profiles: Dict[str, Dict[str, Any]] = {} + for profile in data.get("profile", []): + profile_name = str(profile.get("name", DEFAULT_PROFILE_NAME)) + profiles[profile_name] = ConfigurationManager._config_from_profile(profile, global_config) + + if not profiles: + default = dict(ConfigurationManager.get_defaults()) + default["dll"] = global_config["dll"] + default["no_fp16"] = global_config["no_fp16"] + profiles[DEFAULT_PROFILE_NAME] = default + + current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles)) + return ProfileData( + current_profile=current_profile, + profiles=profiles, + global_config=global_config, + ) + + @staticmethod + def parse_toml_content(content: str) -> ConfigurationData: + profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) + return cast(ConfigurationData, profile_data["profiles"][profile_data["current_profile"]]) + @staticmethod def parse_script_content(script_content: str) -> Dict[str, Union[bool, int, str]]: - """Parse launch script content to extract environment variable values - - Args: - script_content: Content of the launch script file - - Returns: - Dict containing parsed script-only field values - """ - # Use auto-generated parsing logic - parse_script_values = get_script_parsing_logic() - return parse_script_values(script_content.split('\n')) - + return get_script_parsing_logic()(script_content.splitlines()) + + @staticmethod + def parse_profile_selection(script_content: str) -> str | None: + selected = None + for line in script_content.splitlines(): + try: + tokens = shlex.split(line) + except ValueError: + continue + if len(tokens) != 2 or tokens[0] != "export" or "=" not in tokens[1]: + continue + key, value = tokens[1].split("=", 1) + if key in {"LSFGVK_PROFILE", "LSFG_PROCESS"} and value: + selected = value + return selected + @staticmethod - def merge_config_with_script(toml_config: ConfigurationData, script_values: Dict[str, Union[bool, int, str]]) -> ConfigurationData: - """Merge TOML configuration with script environment variable values - - Args: - toml_config: Configuration loaded from TOML file - script_values: Environment variable values parsed from script - - Returns: - Complete configuration with script values overlaid on TOML config - """ - merged_config = dict(toml_config) - - # Update script-only fields with values from script - for field_name in SCRIPT_ONLY_FIELDS.keys(): - if field_name in script_values: - merged_config[field_name] = script_values[field_name] - - return cast(ConfigurationData, merged_config) + def merge_config_with_script( + toml_config: Dict[str, Any], + script_values: Dict[str, Union[bool, int, str]], + ) -> Dict[str, Any]: + merged = dict(toml_config) + for field in SCRIPT_ONLY_FIELDS: + if field in script_values: + merged[field] = script_values[field] + return merged @staticmethod def normalize_profile_name(profile_name: str) -> str: - """Normalize profile name by converting spaces to dashes and trimming - - This allows users to enter names with spaces, which are then safely - converted to dashes for storage and shell script compatibility. - - Args: - profile_name: The raw profile name from user input - - Returns: - Normalized profile name with spaces converted to dashes - """ - if not profile_name: - return profile_name - - # Trim whitespace and convert spaces to dashes - normalized = profile_name.strip().replace(' ', '-') - - # Collapse multiple consecutive dashes into one - while '--' in normalized: - normalized = normalized.replace('--', '-') - - # Remove leading/trailing dashes - normalized = normalized.strip('-') - - return normalized - + return re.sub(r"\s+", "-", profile_name.strip()).strip("-") + @staticmethod def validate_profile_name(profile_name: str) -> bool: - """Validate profile name for safety (after normalization)""" - if not profile_name: - return False - - # Normalize first - this converts spaces to dashes normalized = ConfigurationManager.normalize_profile_name(profile_name) - - if not normalized: - return False - - # Check for invalid characters that could cause issues in shell scripts or TOML - # Note: spaces are now allowed as input (they get converted to dashes) - invalid_chars = set('\t\n\r\'"\\/$|&;()<>{}[]`*?') - if any(char in invalid_chars for char in normalized): - return False - - # Check for reserved names - reserved_names = {'global', 'game', 'current_profile'} - if normalized.lower() in reserved_names: - return False - - return True - + invalid = '\t\n\r\'"\\/$|&;()<>{}[]' + "`" + '*?' + return ( + bool(normalized) + and not any(character in invalid for character in normalized) + and normalized.lower() not in {"global", "profile"} + ) + @staticmethod def create_profile(profile_data: ProfileData, profile_name: str, source_profile: str = None) -> ProfileData: - """Create a new profile by copying from source profile or defaults""" if not ConfigurationManager.validate_profile_name(profile_name): raise ValueError(f"Invalid profile name: {profile_name}") - - # Normalize the profile name (converts spaces to dashes) - profile_name = ConfigurationManager.normalize_profile_name(profile_name) - - if profile_name in profile_data["profiles"]: - raise ValueError(f"Profile '{profile_name}' already exists") - - # Copy from source profile or use defaults - if source_profile and source_profile in profile_data["profiles"]: - new_config = dict(profile_data["profiles"][source_profile]) - else: - new_config = ConfigurationManager.get_defaults() - - # Create new profile data structure - new_profile_data = ProfileData( + normalized = ConfigurationManager.normalize_profile_name(profile_name) + if normalized in profile_data["profiles"]: + raise ValueError(f"Profile '{normalized}' already exists") + source = source_profile if source_profile in profile_data["profiles"] else profile_data["current_profile"] + profiles = dict(profile_data["profiles"]) + profiles[normalized] = dict(profiles[source]) + return ProfileData( current_profile=profile_data["current_profile"], - profiles=dict(profile_data["profiles"]), - global_config=dict(profile_data["global_config"]) + profiles=profiles, + global_config=dict(profile_data["global_config"]), ) - new_profile_data["profiles"][profile_name] = new_config - - return new_profile_data - + @staticmethod def delete_profile(profile_data: ProfileData, profile_name: str) -> ProfileData: - """Delete a profile (cannot delete default profile)""" if profile_name == DEFAULT_PROFILE_NAME: - raise ValueError(f"Cannot delete default profile '{DEFAULT_PROFILE_NAME}'") - + raise ValueError("Cannot delete the default profile") if profile_name not in profile_data["profiles"]: raise ValueError(f"Profile '{profile_name}' does not exist") - - # Create new profile data structure - new_profile_data = ProfileData( - current_profile=profile_data["current_profile"], - profiles=dict(profile_data["profiles"]), - global_config=dict(profile_data["global_config"]) + profiles = dict(profile_data["profiles"]) + del profiles[profile_name] + current_profile = profile_data["current_profile"] + if current_profile == profile_name: + current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles)) + return ProfileData( + current_profile=current_profile, + profiles=profiles, + global_config=dict(profile_data["global_config"]), ) - - # Remove the profile - del new_profile_data["profiles"][profile_name] - - # If we deleted the current profile, switch to default - if new_profile_data["current_profile"] == profile_name: - new_profile_data["current_profile"] = DEFAULT_PROFILE_NAME - # Ensure default profile exists - if DEFAULT_PROFILE_NAME not in new_profile_data["profiles"]: - new_profile_data["profiles"][DEFAULT_PROFILE_NAME] = ConfigurationManager.get_defaults() - - return new_profile_data - + @staticmethod def rename_profile(profile_data: ProfileData, old_name: str, new_name: str) -> ProfileData: - """Rename a profile""" if old_name == DEFAULT_PROFILE_NAME: - raise ValueError(f"Cannot rename default profile '{DEFAULT_PROFILE_NAME}'") - - if not ConfigurationManager.validate_profile_name(new_name): - raise ValueError(f"Invalid profile name: {new_name}") - - # Normalize the new name (converts spaces to dashes) - new_name = ConfigurationManager.normalize_profile_name(new_name) - - if old_name not in profile_data["profiles"]: - raise ValueError(f"Profile '{old_name}' does not exist") - - if new_name in profile_data["profiles"]: - raise ValueError(f"Profile '{new_name}' already exists") - - # Create new profile data structure - new_profile_data = ProfileData( - current_profile=profile_data["current_profile"], - profiles={}, - global_config=dict(profile_data["global_config"]) + raise ValueError("Cannot rename the default profile") + if old_name not in profile_data["profiles"] or not ConfigurationManager.validate_profile_name(new_name): + raise ValueError("Invalid profile rename") + normalized = ConfigurationManager.normalize_profile_name(new_name) + if normalized in profile_data["profiles"]: + raise ValueError(f"Profile '{normalized}' already exists") + profiles = { + normalized if name == old_name else name: value + for name, value in profile_data["profiles"].items() + } + current_profile = normalized if profile_data["current_profile"] == old_name else profile_data["current_profile"] + return ProfileData( + current_profile=current_profile, + profiles=profiles, + global_config=dict(profile_data["global_config"]), ) - - # Copy profiles with new name - for profile_name, config in profile_data["profiles"].items(): - if profile_name == old_name: - new_profile_data["profiles"][new_name] = dict(config) - else: - new_profile_data["profiles"][profile_name] = dict(config) - - # Update current_profile if necessary - if new_profile_data["current_profile"] == old_name: - new_profile_data["current_profile"] = new_name - - return new_profile_data - + @staticmethod def set_current_profile(profile_data: ProfileData, profile_name: str) -> ProfileData: - """Set the current active profile""" if profile_name not in profile_data["profiles"]: raise ValueError(f"Profile '{profile_name}' does not exist") - - # Create new profile data structure - new_profile_data = ProfileData( + return ProfileData( current_profile=profile_name, profiles=dict(profile_data["profiles"]), - global_config=dict(profile_data["global_config"]) + global_config=dict(profile_data["global_config"]), ) - - return new_profile_data diff --git a/py_modules/lsfg_vk/config_schema_generated.py b/py_modules/lsfg_vk/config_schema_generated.py index b320a97..913609b 100644 --- a/py_modules/lsfg_vk/config_schema_generated.py +++ b/py_modules/lsfg_vk/config_schema_generated.py @@ -18,7 +18,6 @@ NO_FP16 = "no_fp16" MULTIPLIER = "multiplier" FLOW_SCALE = "flow_scale" PERFORMANCE_MODE = "performance_mode" -HDR_MODE = "hdr_mode" EXPERIMENTAL_PRESENT_MODE = "experimental_present_mode" DXVK_FRAME_RATE = "dxvk_frame_rate" ENABLE_WOW64 = "enable_wow64" @@ -37,7 +36,6 @@ class ConfigurationData(TypedDict): multiplier: int flow_scale: float performance_mode: bool - hdr_mode: bool experimental_present_mode: str dxvk_frame_rate: int enable_wow64: bool @@ -122,4 +120,4 @@ def get_script_generation_logic(): return generate_script_lines -ALL_FIELDS = ['dll', 'no_fp16', 'multiplier', 'flow_scale', 'performance_mode', 'hdr_mode', 'experimental_present_mode', 'dxvk_frame_rate', 'enable_wow64', 'disable_steamdeck_mode', 'mangohud_workaround', 'disable_vkbasalt', 'force_enable_vkbasalt', 'enable_wsi', 'enable_zink'] +ALL_FIELDS = ['dll', 'no_fp16', 'multiplier', 'flow_scale', 'performance_mode', 'experimental_present_mode', 'dxvk_frame_rate', 'enable_wow64', 'disable_steamdeck_mode', 'mangohud_workaround', 'disable_vkbasalt', 'force_enable_vkbasalt', 'enable_wsi', 'enable_zink'] diff --git a/py_modules/lsfg_vk/configuration.py b/py_modules/lsfg_vk/configuration.py index de11b48..9b4d536 100644 --- a/py_modules/lsfg_vk/configuration.py +++ b/py_modules/lsfg_vk/configuration.py @@ -1,431 +1,224 @@ -""" -Configuration service for TOML-based lsfg configuration management. -""" - -from pathlib import Path -from typing import Dict, Any +import shlex from .base_service import BaseService -from .config_schema import ConfigurationManager, CONFIG_SCHEMA, ProfileData, DEFAULT_PROFILE_NAME +from .config_schema import ConfigurationManager, DEFAULT_PROFILE_NAME, ProfileData from .config_schema_generated import ConfigurationData, get_script_generation_logic -from .types import ConfigurationResponse, ProfilesResponse, ProfileResponse +from .types import ConfigurationResponse, ProfileResponse, ProfilesResponse class ConfigurationService(BaseService): - """Service for managing TOML-based lsfg configuration""" - def get_config(self) -> ConfigurationResponse: - """Read current TOML configuration merged with launch script environment variables - - Returns: - ConfigurationResponse with current configuration or error - """ try: - if not self.config_file_path.exists(): - from .dll_detection import DllDetectionService - dll_service = DllDetectionService(self.log) - toml_config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - else: - content = self.config_file_path.read_text(encoding='utf-8') - toml_config = ConfigurationManager.parse_toml_content(content) - - script_values = {} - if self.lsfg_script_path.exists(): - try: - script_content = self.lsfg_script_path.read_text(encoding='utf-8') - script_values = ConfigurationManager.parse_script_content(script_content) - self.log.info(f"Parsed script values: {script_values}") - except Exception as e: - self.log.warning(f"Failed to parse launch script: {str(e)}") - - config = ConfigurationManager.merge_config_with_script(toml_config, script_values) - + profile_data = self._get_profile_data() + current_profile = profile_data["current_profile"] + config = profile_data["profiles"].get(current_profile, dict(ConfigurationManager.get_defaults())) return self._success_response(ConfigurationResponse, config=config) - - except (OSError, IOError) as e: - error_msg = f"Error reading lsfg config: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) - except Exception as e: - error_msg = f"Error parsing config file: {str(e)}" - self.log.error(error_msg) - from .dll_detection import DllDetectionService - dll_service = DllDetectionService(self.log) - config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - return self._success_response(ConfigurationResponse, - f"Using default configuration due to parse error: {str(e)}", - config=config) - + except Exception as error: + self.log.error(f"Error reading lsfg config: {error}") + return self._error_response(ConfigurationResponse, str(error), config=None) + def update_config_from_dict(self, config: ConfigurationData) -> ConfigurationResponse: - """Update TOML configuration from configuration dictionary (eliminates parameter duplication) - - Args: - config: Complete configuration data dictionary - - Returns: - ConfigurationResponse with success status - """ try: profile_data = self._get_profile_data() - current_profile = profile_data["current_profile"] - - return self.update_profile_config(current_profile, config) - - except (OSError, IOError) as e: - error_msg = f"Error updating lsfg config: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) - except ValueError as e: - error_msg = f"Invalid configuration arguments: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) - + return self.update_profile_config(profile_data["current_profile"], config) + except Exception as error: + self.log.error(f"Error updating lsfg config: {error}") + return self._error_response(ConfigurationResponse, str(error), config=None) + def update_lsfg_script(self, config: ConfigurationData) -> ConfigurationResponse: - """Update the ~/lsfg launch script with current configuration - - Args: - config: Configuration data to apply to the script - - Returns: - ConfigurationResponse indicating success or failure - """ try: - script_content = self._generate_script_content(config) - - self._write_file(self.lsfg_script_path, script_content, 0o755) - - self.log.info(f"Updated lsfg launch script at {self.lsfg_script_path}") - - return self._success_response(ConfigurationResponse, - "Launch script updated successfully", - config=config) - - except Exception as e: - error_msg = f"Error updating launch script: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) - - def _generate_script_content(self, config: ConfigurationData) -> str: - """Generate the content for the ~/lsfg launch script - - Args: - config: Configuration data to apply to the script - - Returns: - The complete script content as a string - """ - lines = [ - "#!/bin/bash", - "# lsfg-vk launch script generated by decky-lossless-scaling-vk plugin", - "# This script sets up the environment for lsfg-vk to work with the plugin configuration", - ] - - generate_script_lines = get_script_generation_logic() - lines.extend(generate_script_lines(config)) - - lines.extend([ - "export LSFG_PROCESS=decky-lsfg-vk", - 'exec "$@"' - ]) - - return "\n".join(lines) + "\n" - + profile_data: ProfileData = { + "current_profile": DEFAULT_PROFILE_NAME, + "profiles": {DEFAULT_PROFILE_NAME: dict(config)}, + "global_config": { + "dll": config.get("dll", ""), + "no_fp16": config.get("no_fp16", False), + }, + } + return self.update_lsfg_script_from_profile_data(profile_data) + except Exception as error: + return self._error_response(ConfigurationResponse, str(error), config=None) + def _generate_script_content_for_profile(self, profile_data: ProfileData) -> str: - """Generate the content for the ~/lsfg launch script with profile support - - Args: - profile_data: Profile data containing current profile and configurations - - Returns: - The complete script content as a string - """ current_profile = profile_data["current_profile"] - config = profile_data["profiles"].get(current_profile, ConfigurationManager.get_defaults()) - - merged_config = dict(config) - for field_name, value in profile_data["global_config"].items(): - merged_config[field_name] = value - - lines = [ - "#!/bin/bash", - f"# Current profile: {current_profile}", - ] - - generate_script_lines = get_script_generation_logic() - lines.extend(generate_script_lines(merged_config)) - - lines.extend([ - f"export LSFG_PROCESS={current_profile}", - 'exec "$@"' - ]) - + config = dict(profile_data["profiles"].get(current_profile, ConfigurationManager.get_defaults())) + config["dll"] = profile_data["global_config"].get("dll", config.get("dll", "")) + config["no_fp16"] = profile_data["global_config"].get("no_fp16", config.get("no_fp16", False)) + + lines = ["#!/bin/bash"] + lines.extend(get_script_generation_logic()(config)) + lines.extend( + [ + f"export LSFGVK_CONFIG={shlex.quote(str(self.config_file_path))}", + f"export LSFGVK_PROFILE={shlex.quote(current_profile)}", + 'exec "$@"', + ] + ) return "\n".join(lines) + "\n" + def _generate_script_content(self, config: ConfigurationData) -> str: + profile_data: ProfileData = { + "current_profile": DEFAULT_PROFILE_NAME, + "profiles": {DEFAULT_PROFILE_NAME: dict(config)}, + "global_config": { + "dll": config.get("dll", ""), + "no_fp16": config.get("no_fp16", False), + }, + } + return self._generate_script_content_for_profile(profile_data) + def _get_profile_data(self) -> ProfileData: - """Get current profile data from config file""" if not self.config_file_path.exists(): from .dll_detection import DllDetectionService - dll_service = DllDetectionService(self.log) - default_config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) + + default = ConfigurationManager.get_defaults_with_dll_detection(DllDetectionService(self.log)) return ProfileData( current_profile=DEFAULT_PROFILE_NAME, - profiles={DEFAULT_PROFILE_NAME: default_config}, + profiles={DEFAULT_PROFILE_NAME: dict(default)}, global_config={ - "dll": default_config.get("dll", ""), - "no_fp16": False - } + "dll": default.get("dll", ""), + "no_fp16": default.get("no_fp16", False), + }, + ) + + profile_data = ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) + if self.lsfg_script_path.exists(): + script_content = self.lsfg_script_path.read_text(encoding="utf-8") + selected = ConfigurationManager.parse_profile_selection(script_content) + if selected in profile_data["profiles"]: + profile_data["current_profile"] = selected + current_profile = profile_data["current_profile"] + profile_data["profiles"][current_profile] = ConfigurationManager.merge_config_with_script( + profile_data["profiles"][current_profile], + ConfigurationManager.parse_script_content(script_content), ) - - content = self.config_file_path.read_text(encoding='utf-8') - return ConfigurationManager.parse_toml_content_multi_profile(content) - + return profile_data + def _save_profile_data(self, profile_data: ProfileData) -> None: - """Save profile data to config file""" - toml_content = ConfigurationManager.generate_toml_content_multi_profile(profile_data) - - self.config_dir.mkdir(parents=True, exist_ok=True) - - self._write_file(self.config_file_path, toml_content, 0o644) - + self._write_file( + self.config_file_path, + ConfigurationManager.generate_toml_content_multi_profile(profile_data), + 0o644, + ) + def get_profiles(self) -> ProfilesResponse: - """Get list of all profiles and current profile - - Returns: - ProfilesResponse with profile list and current profile - """ try: profile_data = self._get_profile_data() - - return self._success_response(ProfilesResponse, - "Profiles retrieved successfully", - profiles=list(profile_data["profiles"].keys()), - current_profile=profile_data["current_profile"]) - - except Exception as e: - error_msg = f"Error getting profiles: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfilesResponse, str(e), - profiles=None, current_profile=None) - + return self._success_response( + ProfilesResponse, + "Profiles retrieved successfully", + profiles=list(profile_data["profiles"]), + current_profile=profile_data["current_profile"], + ) + except Exception as error: + return self._error_response( + ProfilesResponse, + str(error), + profiles=None, + current_profile=None, + ) + def create_profile(self, profile_name: str, source_profile: str = None) -> ProfileResponse: - """Create a new profile - - Args: - profile_name: Name for the new profile (spaces will be converted to dashes) - source_profile: Optional source profile to copy from (default: current profile) - - Returns: - ProfileResponse with success status and the normalized profile name - """ try: profile_data = self._get_profile_data() - - if not source_profile: - source_profile = profile_data["current_profile"] - - # Get the normalized name that will be used for storage - normalized_name = ConfigurationManager.normalize_profile_name(profile_name) - new_profile_data = ConfigurationManager.create_profile(profile_data, profile_name, source_profile) - self._save_profile_data(new_profile_data) - - self.log.info(f"Created profile '{normalized_name}' from '{source_profile}'") - - # Return the normalized name so frontend can use the actual stored name - return self._success_response(ProfileResponse, - f"Profile '{normalized_name}' created successfully", - profile_name=normalized_name) - - except ValueError as e: - error_msg = f"Invalid profile operation: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - except Exception as e: - error_msg = f"Error creating profile: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - + normalized = ConfigurationManager.normalize_profile_name(profile_name) + return self._success_response( + ProfileResponse, + f"Profile '{normalized}' created successfully", + profile_name=normalized, + ) + except Exception as error: + return self._error_response(ProfileResponse, str(error), profile_name=None) + def delete_profile(self, profile_name: str) -> ProfileResponse: - """Delete a profile - - Args: - profile_name: Name of the profile to delete - - Returns: - ProfileResponse with success status - """ try: - profile_data = self._get_profile_data() - - new_profile_data = ConfigurationManager.delete_profile(profile_data, profile_name) - - self._save_profile_data(new_profile_data) - - script_result = self.update_lsfg_script_from_profile_data(new_profile_data) + profile_data = ConfigurationManager.delete_profile(self._get_profile_data(), profile_name) + self._save_profile_data(profile_data) + script_result = self.update_lsfg_script_from_profile_data(profile_data) if not script_result["success"]: - self.log.warning(f"Failed to update launch script: {script_result['error']}") - - self.log.info(f"Deleted profile '{profile_name}'") - - return self._success_response(ProfileResponse, - f"Profile '{profile_name}' deleted successfully", - profile_name=profile_name) - - except ValueError as e: - error_msg = f"Invalid profile operation: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - except Exception as e: - error_msg = f"Error deleting profile: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - + raise OSError(script_result["error"]) + return self._success_response( + ProfileResponse, + f"Profile '{profile_name}' deleted successfully", + profile_name=profile_name, + ) + except Exception as error: + return self._error_response(ProfileResponse, str(error), profile_name=None) + def rename_profile(self, old_name: str, new_name: str) -> ProfileResponse: - """Rename a profile - - Args: - old_name: Current profile name - new_name: New profile name (spaces will be converted to dashes) - - Returns: - ProfileResponse with success status and the normalized profile name - """ try: - profile_data = self._get_profile_data() - - # Get the normalized name that will be used for storage - normalized_name = ConfigurationManager.normalize_profile_name(new_name) - - new_profile_data = ConfigurationManager.rename_profile(profile_data, old_name, new_name) - - self._save_profile_data(new_profile_data) - - script_result = self.update_lsfg_script_from_profile_data(new_profile_data) + profile_data = ConfigurationManager.rename_profile(self._get_profile_data(), old_name, new_name) + self._save_profile_data(profile_data) + script_result = self.update_lsfg_script_from_profile_data(profile_data) if not script_result["success"]: - self.log.warning(f"Failed to update launch script: {script_result['error']}") - - self.log.info(f"Renamed profile '{old_name}' to '{normalized_name}'") - - # Return the normalized name so frontend can use the actual stored name - return self._success_response(ProfileResponse, - f"Profile renamed from '{old_name}' to '{normalized_name}' successfully", - profile_name=normalized_name) - - except ValueError as e: - error_msg = f"Invalid profile operation: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - except Exception as e: - error_msg = f"Error renaming profile: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - + raise OSError(script_result["error"]) + normalized = ConfigurationManager.normalize_profile_name(new_name) + return self._success_response( + ProfileResponse, + f"Profile renamed to '{normalized}' successfully", + profile_name=normalized, + ) + except Exception as error: + return self._error_response(ProfileResponse, str(error), profile_name=None) + def set_current_profile(self, profile_name: str) -> ProfileResponse: - """Set the current active profile - - Args: - profile_name: Name of the profile to set as current - - Returns: - ProfileResponse with success status - """ try: - profile_data = self._get_profile_data() - - new_profile_data = ConfigurationManager.set_current_profile(profile_data, profile_name) - - self._save_profile_data(new_profile_data) - - script_result = self.update_lsfg_script_from_profile_data(new_profile_data) + profile_data = ConfigurationManager.set_current_profile(self._get_profile_data(), profile_name) + script_result = self.update_lsfg_script_from_profile_data(profile_data) if not script_result["success"]: - self.log.warning(f"Failed to update launch script: {script_result['error']}") - - self.log.info(f"Set current profile to '{profile_name}'") - - return self._success_response(ProfileResponse, - f"Current profile set to '{profile_name}' successfully", - profile_name=profile_name) - - except ValueError as e: - error_msg = f"Invalid profile operation: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - except Exception as e: - error_msg = f"Error setting current profile: {str(e)}" - self.log.error(error_msg) - return self._error_response(ProfileResponse, str(e), profile_name=None) - + raise OSError(script_result["error"]) + return self._success_response( + ProfileResponse, + f"Current profile set to '{profile_name}' successfully", + profile_name=profile_name, + ) + except Exception as error: + return self._error_response(ProfileResponse, str(error), profile_name=None) + def update_profile_config(self, profile_name: str, config: ConfigurationData) -> ConfigurationResponse: - """Update configuration for a specific profile - - Args: - profile_name: Name of the profile to update - config: Configuration data to apply - - Returns: - ConfigurationResponse with success status - """ try: profile_data = self._get_profile_data() - if profile_name not in profile_data["profiles"]: - return self._error_response(ConfigurationResponse, - f"Profile '{profile_name}' does not exist", - config=None) - - # Update the profile's config - profile_data["profiles"][profile_name] = config - - # Update global config fields if they're in the config - for field_name in ["dll", "no_fp16"]: - if field_name in config: - profile_data["global_config"][field_name] = config[field_name] - + raise ValueError(f"Profile '{profile_name}' does not exist") + + validated = ConfigurationManager.validate_config(config) + profile_data["profiles"][profile_name] = { + **profile_data["profiles"][profile_name], + **validated, + } + profile_data["global_config"]["dll"] = validated.get("dll", "") + profile_data["global_config"]["no_fp16"] = validated.get("no_fp16", False) self._save_profile_data(profile_data) - + if profile_name == profile_data["current_profile"]: script_result = self.update_lsfg_script_from_profile_data(profile_data) if not script_result["success"]: - self.log.warning(f"Failed to update launch script: {script_result['error']}") - - field_values = ", ".join(f"{k}={repr(v)}" for k, v in config.items()) - self.log.info(f"Updated profile '{profile_name}' configuration: {field_values}") - - return self._success_response(ConfigurationResponse, - f"Profile '{profile_name}' configuration updated successfully", - config=config) - - except Exception as e: - error_msg = f"Error updating profile configuration: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) - + raise OSError(script_result["error"]) + + return self._success_response( + ConfigurationResponse, + f"Profile '{profile_name}' configuration updated successfully", + config=validated, + ) + except Exception as error: + return self._error_response(ConfigurationResponse, str(error), config=None) + def update_lsfg_script_from_profile_data(self, profile_data: ProfileData) -> ConfigurationResponse: - """Update the ~/lsfg launch script from profile data - - Args: - profile_data: Profile data to apply to the script - - Returns: - ConfigurationResponse indicating success or failure - """ try: script_content = self._generate_script_content_for_profile(profile_data) - - # Write the script file self._write_file(self.lsfg_script_path, script_content, 0o755) - - self.log.info(f"Updated lsfg launch script at {self.lsfg_script_path} for profile '{profile_data['current_profile']}'") - - # Get current profile config for response - current_config = profile_data["profiles"].get(profile_data["current_profile"], ConfigurationManager.get_defaults()) - - return self._success_response(ConfigurationResponse, - "Launch script updated successfully", - config=current_config) - - except Exception as e: - error_msg = f"Error updating launch script: {str(e)}" - self.log.error(error_msg) - return self._error_response(ConfigurationResponse, str(e), config=None) + current_config = profile_data["profiles"].get( + profile_data["current_profile"], + dict(ConfigurationManager.get_defaults()), + ) + return self._success_response( + ConfigurationResponse, + "Launch script updated successfully", + config=current_config, + ) + except Exception as error: + return self._error_response(ConfigurationResponse, str(error), config=None) diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py index fe2febb..f41b044 100644 --- a/py_modules/lsfg_vk/constants.py +++ b/py_modules/lsfg_vk/constants.py @@ -1,34 +1,27 @@ -""" -Constants for the lsfg-vk plugin. -""" - from pathlib import Path +LOCAL_BIN = ".local/bin" LOCAL_LIB = ".local/lib" -LOCAL_SHARE_BASE = ".local/share" VULKAN_LAYER_DIR = ".local/share/vulkan/implicit_layer.d" CONFIG_DIR = ".config/lsfg-vk" SCRIPT_NAME = "lsfg" CONFIG_FILENAME = "conf.toml" -LIB_FILENAME = "liblsfg-vk.so" -JSON_FILENAME = "VkLayer_LS_frame_generation.json" -ZIP_FILENAME = "lsfg-vk_noui.zip" - -FLATPAK_23_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_23.08.flatpak" -FLATPAK_24_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_24.08.flatpak" -FLATPAK_25_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_25.08.flatpak" +ARCHIVE_FILENAME = "lsfg-vk-2.0.0.tar.xz" +LIB_FILENAME = "liblsfg-vk-layer.so" +LIB_X86_FILENAME = "liblsfg-vk-layer.x86.so" +JSON_FILENAME = "VkLayer_LSFGVK_frame_generation.json" +JSON_X86_FILENAME = "VkLayer_LSFGVK_frame_generation.x86.json" +CLI_FILENAME = "lsfg-vk-cli" -SO_EXT = ".so" -JSON_EXT = ".json" +LEGACY_LIB_FILENAME = "liblsfg-vk.so" +LEGACY_JSON_FILENAME = "VkLayer_LS_frame_generation.json" BIN_DIR = "bin" - STEAM_COMMON_PATH = Path("steamapps/common/Lossless Scaling") -LOSSLESS_DLL_NAME = "Lossless.dll" +LOSSLESS_DLL_NAME = "lsfg-vk.dll" ENV_LSFG_DLL_PATH = "LSFG_DLL_PATH" ENV_XDG_DATA_HOME = "XDG_DATA_HOME" ENV_HOME = "HOME" - diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 4329d49..5e0a9c1 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -1,244 +1,202 @@ -""" -Installation service for lsfg-vk. -""" - import os 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, DEFAULT_PROFILE_NAME, ProfileData from .constants import ( - LIB_FILENAME, JSON_FILENAME, ZIP_FILENAME, BIN_DIR, - SO_EXT, JSON_EXT + ARCHIVE_FILENAME, + BIN_DIR, + CLI_FILENAME, + JSON_FILENAME, + JSON_X86_FILENAME, + LEGACY_JSON_FILENAME, + LEGACY_LIB_FILENAME, + LIB_FILENAME, + LIB_X86_FILENAME, ) -from .config_schema import ConfigurationManager -from .types import InstallationResponse, UninstallationResponse, InstallationCheckResponse +from .types import InstallationCheckResponse, InstallationResponse, UninstallationResponse class InstallationService(BaseService): - """Service for handling lsfg-vk installation and uninstallation""" - def __init__(self, logger=None): super().__init__(logger) - 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 = plugin_dir / 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) - - 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 _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 + profile_data = self._prepare_config() + self._install_archive(archive_path) + self._write_file( + self.config_file_path, + ConfigurationManager.generate_toml_content_multi_profile(profile_data), + 0o644, + ) + self._create_lsfg_launch_script(profile_data) + 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), message="") + + def _payload_destinations(self) -> Dict[str, tuple[Path, int]]: + return { + f"bin/{CLI_FILENAME}": (self.cli_file, 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), } - - 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: - shutil.copy2(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 - shutil.copy2(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 + + def _install_archive(self, archive_path: Path) -> None: + destinations = self._payload_destinations() + found = set() + with tarfile.open(archive_path, "r:xz") 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)) + + def _prepare_config(self) -> ProfileData: if self.config_file_path.exists(): - 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) + content = self.config_file_path.read_text(encoding="utf-8") + legacy = ConfigurationManager.is_legacy_v1(content) + profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) + if legacy: + backup_path = self.config_file_path.with_name(f"{self.config_file_path.name}.v1.bak") + if not backup_path.exists(): + self._write_file(backup_path, content, 0o644) 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 + default = dict(ConfigurationManager.get_defaults()) + profile_data = ProfileData( + current_profile=DEFAULT_PROFILE_NAME, + profiles={DEFAULT_PROFILE_NAME: default}, + global_config={ + "dll": default.get("dll", ""), + "no_fp16": default.get("no_fp16", False), + }, + ) + + from .dll_detection import DllDetectionService + + dll_result = DllDetectionService(self.log).check_lossless_scaling_dll() + if dll_result.get("detected") and dll_result.get("path"): + profile_data["global_config"]["dll"] = dll_result["path"] + + defaults = dict(ConfigurationManager.get_defaults()) + for profile_name, raw_profile in list(profile_data["profiles"].items()): + validated = ConfigurationManager.validate_config({**defaults, **raw_profile}) + profile_data["profiles"][profile_name] = {**raw_profile, **validated} + + if self.lsfg_script_path.exists(): + script_content = self.lsfg_script_path.read_text(encoding="utf-8") + selected = ConfigurationManager.parse_profile_selection(script_content) + if selected in profile_data["profiles"]: + profile_data["current_profile"] = selected + current_profile = profile_data["current_profile"] + profile_data["profiles"][current_profile] = ConfigurationManager.merge_config_with_script( + profile_data["profiles"][current_profile], + ConfigurationManager.parse_script_content(script_content), + ) + + if profile_data["current_profile"] not in profile_data["profiles"]: + profile_data["current_profile"] = ( + DEFAULT_PROFILE_NAME + if DEFAULT_PROFILE_NAME in profile_data["profiles"] + else next(iter(profile_data["profiles"])) + ) + + for profile in profile_data["profiles"].values(): + profile["dll"] = profile_data["global_config"].get("dll", "") + profile["no_fp16"] = profile_data["global_config"].get("no_fp16", False) + return profile_data + + def _create_lsfg_launch_script(self, profile_data: ProfileData) -> None: 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}") - + + configuration_service = ConfigurationService(logger=self.log) + configuration_service.user_home = self.user_home + configuration_service.config_dir = self.config_dir + configuration_service.config_file_path = self.config_file_path + configuration_service.lsfg_script_path = self.lsfg_launch_script_path + self._write_file( + self.lsfg_launch_script_path, + configuration_service._generate_script_content_for_profile(profile_data), + 0o755, + ) + + def _remove_legacy_layer_files(self) -> None: + for path in (self.legacy_lib_file, self.legacy_json_file): + self._remove_if_exists(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) 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}") - + lib_exists = self.lib_file.exists() and self.lib_x86_file.exists() + json_exists = self.json_file.exists() and self.json_x86_file.exists() + script_exists = self.lsfg_launch_script_path.exists() 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 + "script_exists": script_exists, "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 + "script_path": str(self.lsfg_launch_script_path), + "error": None, } - - 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, @@ -246,154 +204,47 @@ class InstallationService(BaseService): "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) + "script_path": str(self.lsfg_launch_script_path), + "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) - + removed = [] + for path in ( + self.lib_file, + self.lib_x86_file, + self.json_file, + self.json_x86_file, + self.cli_file, + self.legacy_lib_file, + self.legacy_json_file, + self.lsfg_launch_script_path, + ): + if self._remove_if_exists(path): + removed.append(str(path)) + 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), + 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()}") - - 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 + self.uninstall() + except Exception as error: + self.log.error(f"Error cleaning up lsfg-vk files during uninstall: {error}") + self.log.error(traceback.format_exc()) -- cgit v1.2.3