diff options
Diffstat (limited to 'py_modules')
| -rw-r--r-- | py_modules/lsfg_vk/base_service.py | 27 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/config_schema.py | 738 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/config_schema_generated.py | 31 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/configuration.py | 35 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/constants.py | 24 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 111 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 610 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 35 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/types.py | 3 |
9 files changed, 611 insertions, 1003 deletions
diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py index 262e2b0..15e14f7 100644 --- a/py_modules/lsfg_vk/base_service.py +++ b/py_modules/lsfg_vk/base_service.py @@ -5,12 +5,13 @@ 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 import decky -from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME +from .constants import LOCAL_LIB, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME ResponseType = TypeVar('ResponseType', bound=Dict[str, Any]) @@ -79,16 +80,24 @@ class BaseService: Raises: OSError: If write fails """ + temp_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}") - + with tempfile.NamedTemporaryFile( + mode='w', encoding='utf-8', dir=path.parent, + prefix=f'.{path.name}.', delete=False, + ) as temp_file: + temp_path = Path(temp_file.name) + temp_file.write(content) + temp_file.flush() + os.fsync(temp_file.fileno()) + + temp_path.chmod(mode) + os.replace(temp_path, path) + self.log.info(f"Atomically wrote {path}") + except (OSError, IOError, PermissionError) as e: + if temp_path is not None: + temp_path.unlink(missing_ok=True) self.log.error(f"Failed to write to {path}: {e}") raise diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py index 3a82bbd..f8a9fe6 100644 --- a/py_modules/lsfg_vk/config_schema.py +++ b/py_modules/lsfg_vk/config_schema.py @@ -1,605 +1,317 @@ -""" -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 -""" +"""lsfg-vk v2 configuration and Decky profile management.""" +import json import logging import re 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 +GLOBAL_SECTION_FIELDS = {"dll", "allow_fp16"} 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 +PROFILE_TOML_FIELDS = {"active_in", "gpu", "multiplier", "flow_scale", "performance_mode", "pacing"} 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! +CURRENT_PROFILE_COMMENT = re.compile(r'^\s*#\s*decky-current-profile\s*=\s*"([^"]+)"\s*$') 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, ConfigurationData] + global_config: Dict[str, Any] + + +def _toml_string(value: str) -> str: + return json.dumps(value) class ConfigurationManager: - """Centralized configuration management""" - + """Read both legacy v1 and upstream v2 configs, and write only v2.""" + @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" - + result = dll_detection_service.check_lossless_scaling_dll() + if result.get("detected") and result.get("path"): + defaults["dll"] = result["path"] + except (OSError, IOError, KeyError, TypeError) as error: + logging.getLogger(__name__).debug("DLL detection failed: %s", error) 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"] < 2: + raise ValueError("multiplier must be 2 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["pacing"] != "none": + raise ValueError("only pacing = 'none' is currently available") return cast(ConfigurationData, validated) - + + @staticmethod + def _config_from_profile(profile: Dict[str, Any], global_config: Dict[str, Any]) -> ConfigurationData: + config = ConfigurationManager.get_defaults() + for field in PROFILE_TOML_FIELDS | SCRIPT_ONLY_FIELDS: + if field not in profile: + continue + value = profile[field] + if field == "active_in" and isinstance(value, list): + value = ", ".join(str(item) for item in value) + config[field] = value + for field in GLOBAL_SECTION_FIELDS: + if field in global_config: + config[field] = global_config[field] + return 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}, - "global_config": { - "dll": config.get("dll", ""), - "no_fp16": False # Always enabled even if previously set - } + "global_config": {field: config[field] for field in GLOBAL_SECTION_FIELDS}, } 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") - lines.append(f"no_fp16 = false") - 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", + f"# decky-current-profile = {_toml_string(profile_data['current_profile'])}", + "", + "[global]", + ] + if global_config.get("dll"): + lines.append(f"dll = {_toml_string(str(global_config['dll']))}") + lines.append(f"allow_fp16 = {str(bool(global_config.get('allow_fp16', True))).lower()}") + + for profile_name, raw_config in profile_data["profiles"].items(): + config = ConfigurationManager.validate_config({**raw_config, **global_config}) + lines.extend(["", "[[profile]]", f"name = {_toml_string(profile_name)}"]) + active_in = [entry.strip() for entry in config["active_in"].split(",") if entry.strip()] + if len(active_in) == 1: + lines.append(f"active_in = {_toml_string(active_in[0])}") + elif active_in: + lines.append("active_in = [" + ", ".join(_toml_string(entry) for entry in active_in) + "]") + if config["gpu"]: + lines.append(f"gpu = {_toml_string(config['gpu'])}") + lines.extend([ + f"multiplier = {config['multiplier']}", + f"flow_scale = {config['flow_scale']}", + f"performance_mode = {str(config['performance_mode']).lower()}", + "pacing = 'none'", + ]) + 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 = data.get("global", {}) + global_config = { + "dll": str(old_global.get("dll", "")), + "allow_fp16": not bool(old_global.get("no_fp16", False)), + } + profiles: Dict[str, ConfigurationData] = {} + for game in data.get("game", []): + name = str(game.get("exe", DEFAULT_PROFILE_NAME)) + migrated_profile = dict(game) + if int(migrated_profile.get("multiplier", 2)) <= 1: + migrated_profile["disable_lsfgvk"] = True + migrated_profile["multiplier"] = max(2, int(migrated_profile.get("multiplier", 2))) + profiles[name] = ConfigurationManager._config_from_profile(migrated_profile, global_config) + if not profiles: + profiles[DEFAULT_PROFILE_NAME] = 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 is_legacy_v1(content: str) -> bool: + try: + return tomllib.loads(content).get("version") == 1 + except tomllib.TOMLDecodeError: + return False + @staticmethod def parse_toml_content_multi_profile(content: str) -> ProfileData: - """Parse TOML content into profile data structure""" + data = tomllib.loads(content) + if data.get("version") == 1: + return ConfigurationManager._profile_data_from_v1(data) + if data.get("version") != 2: + raise ValueError("unsupported lsfg-vk configuration version") + + raw_global = dict(data.get("global", {})) + global_config = { + "dll": str(raw_global.get("dll", "")), + "allow_fp16": bool(raw_global.get("allow_fp16", True)), + } profiles: Dict[str, ConfigurationData] = {} - global_config: Dict[str, Any] = {} - current_profile = DEFAULT_PROFILE_NAME - - 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": - # Always enforce FP16 to be enabled (no_fp16 = false) - global_config["no_fp16"] = False - - # 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={} - ) - + for profile in data.get("profile", []): + name = str(profile.get("name", DEFAULT_PROFILE_NAME)) + profiles[name] = ConfigurationManager._config_from_profile(profile, global_config) + if not profiles: + profiles[DEFAULT_PROFILE_NAME] = ConfigurationManager.get_defaults() + + current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles)) + for line in content.splitlines(): + match = CURRENT_PROFILE_COMMENT.match(line) + if match and match.group(1) in profiles: + current_profile = match.group(1) + break + 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 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 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: ConfigurationData, + script_values: Dict[str, Union[bool, int, str]], + ) -> ConfigurationData: + merged = dict(toml_config) + for field in SCRIPT_ONLY_FIELDS: + if field in script_values: + merged[field] = script_values[field] + return cast(ConfigurationData, 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(char in invalid for char 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"] + new_profile = dict(profile_data["profiles"][source]) + # A cloned profile may inherit Active In values that collide with its source. + # Force the newly selected profile until the user explicitly enables native matching. + new_profile["use_native_matching"] = False + return ProfileData( current_profile=profile_data["current_profile"], - profiles=dict(profile_data["profiles"]), - global_config=dict(profile_data["global_config"]) + profiles={**profile_data["profiles"], normalized: new_profile}, + 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..9cfda51 100644 --- a/py_modules/lsfg_vk/config_schema_generated.py +++ b/py_modules/lsfg_vk/config_schema_generated.py @@ -14,12 +14,15 @@ from shared_config import CONFIG_SCHEMA_DEF, ConfigFieldType # Field name constants for type-safe access DLL = "dll" -NO_FP16 = "no_fp16" +ALLOW_FP16 = "allow_fp16" MULTIPLIER = "multiplier" FLOW_SCALE = "flow_scale" PERFORMANCE_MODE = "performance_mode" -HDR_MODE = "hdr_mode" -EXPERIMENTAL_PRESENT_MODE = "experimental_present_mode" +PACING = "pacing" +ACTIVE_IN = "active_in" +USE_NATIVE_MATCHING = "use_native_matching" +GPU = "gpu" +DISABLE_LSFGVK = "disable_lsfgvk" DXVK_FRAME_RATE = "dxvk_frame_rate" ENABLE_WOW64 = "enable_wow64" DISABLE_STEAMDECK_MODE = "disable_steamdeck_mode" @@ -33,12 +36,15 @@ ENABLE_ZINK = "enable_zink" class ConfigurationData(TypedDict): """Type-safe configuration data structure - AUTO-GENERATED""" dll: str - no_fp16: bool + allow_fp16: bool multiplier: int flow_scale: float performance_mode: bool - hdr_mode: bool - experimental_present_mode: str + pacing: str + active_in: str + use_native_matching: bool + gpu: str + disable_lsfgvk: bool dxvk_frame_rate: int enable_wow64: bool disable_steamdeck_mode: bool @@ -64,6 +70,10 @@ def get_script_parsing_logic(): value = value.strip() # Auto-generated parsing logic: + if key == "DECKY_LSFGVK_AUTO_PROFILE": + script_values["use_native_matching"] = value == "1" + if key == "DISABLE_LSFGVK": + script_values["disable_lsfgvk"] = value == "1" if key == "DXVK_FRAME_RATE": try: script_values["dxvk_frame_rate"] = int(value) @@ -81,8 +91,6 @@ def get_script_parsing_logic(): script_values["force_enable_vkbasalt"] = value == "1" if key == "ENABLE_GAMESCOPE_WSI": script_values["enable_wsi"] = value != "0" - if key == "DXVK_HDR": - script_values["enable_wsi"] = value != "0" if key == "__GLX_VENDOR_LIBRARY_NAME" and value == "mesa": script_values["enable_zink"] = True if key == "MESA_LOADER_DRIVER_OVERRIDE" and value == "zink": @@ -98,6 +106,10 @@ def get_script_generation_logic(): """Return the script generation logic as a callable""" def generate_script_lines(config): lines = [] + if config.get("use_native_matching", False): + lines.append("export DECKY_LSFGVK_AUTO_PROFILE=1") + if config.get("disable_lsfgvk", False): + lines.append("export DISABLE_LSFGVK=1") dxvk_frame_rate = config.get("dxvk_frame_rate", 0) if dxvk_frame_rate > 0: lines.append(f"export DXVK_FRAME_RATE={dxvk_frame_rate}") @@ -113,7 +125,6 @@ def get_script_generation_logic(): lines.append("export ENABLE_VKBASALT=1") if not config.get("enable_wsi", False): lines.append("export ENABLE_GAMESCOPE_WSI=0") - lines.append("export DXVK_HDR=0") if config.get("enable_zink", False): lines.append("export __GLX_VENDOR_LIBRARY_NAME=mesa") lines.append("export MESA_LOADER_DRIVER_OVERRIDE=zink") @@ -122,4 +133,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', 'allow_fp16', 'multiplier', 'flow_scale', 'performance_mode', 'pacing', 'active_in', 'use_native_matching', 'gpu', 'disable_lsfgvk', '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 fce3738..6511745 100644 --- a/py_modules/lsfg_vk/configuration.py +++ b/py_modules/lsfg_vk/configuration.py @@ -3,7 +3,8 @@ Configuration service for TOML-based lsfg configuration management. """ from pathlib import Path -from typing import Dict, Any +import shlex +from typing import Dict, Any, cast from .base_service import BaseService from .config_schema import ConfigurationManager, CONFIG_SCHEMA, ProfileData, DEFAULT_PROFILE_NAME @@ -14,6 +15,13 @@ from .types import ConfigurationResponse, ProfilesResponse, ProfileResponse class ConfigurationService(BaseService): """Service for managing TOML-based lsfg configuration""" + + @staticmethod + def _profile_selection_lines(profile_name: str, config: ConfigurationData) -> list[str]: + """Force the selected profile unless the user explicitly opts into native matching.""" + if config.get("use_native_matching", False): + return ["# lsfg-vk will select a matching profile from Active In."] + return [f"export LSFGVK_PROFILE={shlex.quote(profile_name)}"] def get_config(self) -> ConfigurationResponse: """Read current TOML configuration merged with launch script environment variables @@ -124,7 +132,10 @@ class ConfigurationService(BaseService): generate_script_lines = get_script_generation_logic() lines.extend(generate_script_lines(config)) - lines.append("export LSFG_PROCESS=decky-lsfg-vk") + lines.extend([ + f"export LSFGVK_CONFIG={shlex.quote(str(self.config_file_path))}", + *self._profile_selection_lines(DEFAULT_PROFILE_NAME, config), + ]) lines.extend(self._generate_game_launch_lines()) return "\n".join(lines) + "\n" @@ -153,7 +164,10 @@ class ConfigurationService(BaseService): generate_script_lines = get_script_generation_logic() lines.extend(generate_script_lines(merged_config)) - lines.append(f"export LSFG_PROCESS={current_profile}") + lines.extend([ + f"export LSFGVK_CONFIG={shlex.quote(str(self.config_file_path))}", + *self._profile_selection_lines(current_profile, cast(ConfigurationData, merged_config)), + ]) lines.extend(self._generate_game_launch_lines()) return "\n".join(lines) + "\n" @@ -187,12 +201,21 @@ class ConfigurationService(BaseService): profiles={DEFAULT_PROFILE_NAME: default_config}, global_config={ "dll": default_config.get("dll", ""), - "no_fp16": False + "allow_fp16": default_config.get("allow_fp16", True) } ) content = self.config_file_path.read_text(encoding='utf-8') - return ConfigurationManager.parse_toml_content_multi_profile(content) + profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) + if self.lsfg_script_path.exists(): + script_values = ConfigurationManager.parse_script_content( + self.lsfg_script_path.read_text(encoding='utf-8') + ) + current_profile = profile_data["current_profile"] + profile_data["profiles"][current_profile] = ConfigurationManager.merge_config_with_script( + profile_data["profiles"][current_profile], script_values + ) + return profile_data def _save_profile_data(self, profile_data: ProfileData) -> None: """Save profile data to config file""" @@ -393,7 +416,7 @@ class ConfigurationService(BaseService): profile_data["profiles"][profile_name] = config # Update global config fields if they're in the config - for field_name in ["dll", "no_fp16"]: + for field_name in ["dll", "allow_fp16"]: if field_name in config: profile_data["global_config"][field_name] = config[field_name] diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py index 023894f..ad3fb32 100644 --- a/py_modules/lsfg_vk/constants.py +++ b/py_modules/lsfg_vk/constants.py @@ -1,28 +1,26 @@ -""" -Constants for the lsfg-vk plugin. -""" +"""Constants for the lsfg-vk Decky plugin.""" from pathlib import Path 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" -ARM_LIB_FILENAME = "liblsfg-vk-arm64.so" +LIB_FILENAME = "liblsfg-vk-layer.so" +JSON_FILENAME = "VkLayer_LSFGVK_frame_generation.json" +LAYER_ARCHIVE_X86_64 = "lsfg-vk-layer-x86_64.tar.xz" +LAYER_ARCHIVE_AARCH64 = "lsfg-vk-layer-aarch64.tar.xz" +CLI_FILENAME = "lsfg-vk-cli" + +LEGACY_LIB_FILENAME = "liblsfg-vk.so" +LEGACY_JSON_FILENAME = "VkLayer_LS_frame_generation.json" 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" -SO_EXT = ".so" -JSON_EXT = ".json" - BIN_DIR = "bin" ARMADA_DEVICE_ENV = Path("/usr/libexec/armada/device-env") @@ -35,3 +33,7 @@ ENV_LSFG_DLL_PATH = "LSFG_DLL_PATH" ENV_XDG_DATA_HOME = "XDG_DATA_HOME" ENV_HOME = "HOME" + +def get_layer_archive_filename(is_arm: bool) -> str: + """Return the v2 layer archive appropriate for the native game host.""" + return LAYER_ARCHIVE_AARCH64 if is_arm else LAYER_ARCHIVE_X86_64 diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index c9be0ec..80dcf5d 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -8,9 +8,7 @@ from pathlib import Path from typing import Dict, Any, List, Optional from .base_service import BaseService -from .constants import ( - FLATPAK_23_08_FILENAME, FLATPAK_24_08_FILENAME, FLATPAK_25_08_FILENAME, BIN_DIR, CONFIG_DIR -) +from .config_schema import ConfigurationManager from .types import BaseResponse @@ -52,6 +50,38 @@ class FlatpakService(BaseService): self.extension_id_25_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/25.08" self.flatpak_command = None + def _get_lsfg_paths(self) -> tuple[str, str]: + """Return the v2 config directory and directory containing Lossless.dll.""" + config_path = str(self.config_dir) + dll_directory = str(self.user_home / ".local/share/Steam/steamapps/common") + if not self.config_file_path.exists(): + return config_path, dll_directory + try: + profile_data = ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) + configured_dll = profile_data["global_config"].get("dll", "") + if configured_dll: + dll_directory = str(Path(str(configured_dll)).parent) + except (OSError, ValueError, KeyError, TypeError) as error: + self.log.debug("Could not read configured DLL path for Flatpak override: %s", error) + return config_path, dll_directory + + def _remove_legacy_app_overrides(self, app_id: str) -> list[str]: + """Remove only the v1 overrides previously created by this plugin.""" + legacy_dll_path = self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" + legacy_overrides = [ + ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], + ["override", "--user", f"--nofilesystem={legacy_dll_path}", app_id], + ["override", "--user", f"--nofilesystem={self.lsfg_launch_script_path}", app_id], + ] + errors = [] + for args in legacy_overrides: + result = self._run_flatpak_command(args, capture_output=True, text=True) + if result.returncode != 0: + errors.append(f"{' '.join(args[2:-1])}: {result.stderr}") + return errors + def _get_clean_env(self): """Get a clean environment without PyInstaller's bundled libraries""" env = os.environ.copy() @@ -177,20 +207,14 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): return self._error_response(BaseResponse, "Flatpak is not available on this system") - plugin_dir = Path(__file__).parent.parent.parent if version == "23.08": - filename = FLATPAK_23_08_FILENAME - elif version == "24.08": - filename = FLATPAK_24_08_FILENAME - else: - filename = FLATPAK_25_08_FILENAME - flatpak_path = plugin_dir / BIN_DIR / filename - - if not flatpak_path.exists(): - return self._error_response(BaseResponse, f"Flatpak file not found: {flatpak_path}") + return self._error_response( + BaseResponse, + "Flathub supplies the lsfg-vk v2 extension for runtimes 24.08 and 25.08; install 23.08 manually.", + ) result = self._run_flatpak_command( - ["install", "--user", "--noninteractive", str(flatpak_path)], + ["install", "--user", "--noninteractive", "flathub", f"org.freedesktop.Platform.VulkanLayer.lsfgvk//{version}"], capture_output=True, text=True ) @@ -298,10 +322,7 @@ class FlatpakService(BaseService): return {"filesystem": False, "env": False} output = result.stdout - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" + config_path, dll_directory = self._get_lsfg_paths() filesystem_section = "" in_context = False @@ -317,10 +338,9 @@ class FlatpakService(BaseService): break has_config_fs = config_path in filesystem_section - has_dll_fs = dll_path in filesystem_section - has_lsfg_fs = lsfg_path in filesystem_section + has_dll_fs = dll_directory in filesystem_section - filesystem_override = has_config_fs and has_dll_fs and has_lsfg_fs + filesystem_override = has_config_fs and has_dll_fs env_override = False in_environment = False @@ -331,11 +351,11 @@ class FlatpakService(BaseService): in_environment = True elif line.startswith("[") and line != "[Environment]": in_environment = False - elif in_environment and line.startswith(f"LSFG_CONFIG={config_path}/conf.toml"): + elif in_environment and line.startswith(f"LSFGVK_CONFIG={config_path}/conf.toml"): env_override = True break - self.log.debug(f"Override status for {app_id}: filesystem={filesystem_override} ({has_config_fs}/{has_dll_fs}/{has_lsfg_fs}), env={env_override}") + self.log.debug(f"Override status for {app_id}: filesystem={filesystem_override} ({has_config_fs}/{has_dll_fs}), env={env_override}") return {"filesystem": filesystem_override, "env": env_override} @@ -351,15 +371,11 @@ class FlatpakService(BaseService): "Flatpak is not available on this system", app_id=app_id, operation="set") - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" + config_path, dll_directory = self._get_lsfg_paths() filesystem_overrides = [ - f"--filesystem={dll_path}", - f"--filesystem={config_path}:rw", - f"--filesystem={lsfg_path}:rw" + f"--filesystem={dll_directory}:ro", + f"--filesystem={config_path}:rw", ] for override in filesystem_overrides: @@ -373,7 +389,7 @@ class FlatpakService(BaseService): app_id=app_id, operation="set") result = self._run_flatpak_command( - ["override", "--user", f"--env=LSFG_CONFIG={config_path}/conf.toml", app_id], + ["override", "--user", f"--env=LSFGVK_CONFIG={config_path}/conf.toml", app_id], capture_output=True, text=True ) @@ -382,6 +398,14 @@ class FlatpakService(BaseService): return self._error_response(FlatpakOverrideResponse, error_msg, app_id=app_id, operation="set") + legacy_errors = self._remove_legacy_app_overrides(app_id) + if legacy_errors: + self.log.warning( + "Applied v2 overrides for %s but could not fully remove v1 overrides: %s", + app_id, + "; ".join(legacy_errors), + ) + self.log.info(f"Successfully set lsfg-vk overrides for {app_id}") return self._success_response(FlatpakOverrideResponse, f"lsfg-vk overrides set for {app_id}", @@ -401,28 +425,11 @@ class FlatpakService(BaseService): "Flatpak is not available on this system", app_id=app_id, operation="remove") - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" + config_path, dll_directory = self._get_lsfg_paths() - reset_result = self._run_flatpak_command( - ["override", "--user", "--reset", app_id], - capture_output=True, text=True - ) - - if reset_result.returncode == 0: - self.log.info(f"Successfully reset all overrides for {app_id}") - return self._success_response(FlatpakOverrideResponse, - f"All overrides reset for {app_id}", - app_id=app_id, operation="remove") - - self.log.debug(f"Reset failed, trying individual removal: {reset_result.stderr}") - filesystem_overrides = [ - f"--nofilesystem={dll_path}", + f"--nofilesystem={dll_directory}", f"--nofilesystem={config_path}", - f"--nofilesystem={lsfg_path}" ] removal_errors = [] @@ -437,13 +444,15 @@ class FlatpakService(BaseService): removal_errors.append(f"{override}: {result.stderr}") result = self._run_flatpak_command( - ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], + ["override", "--user", "--unset-env=LSFGVK_CONFIG", app_id], capture_output=True, text=True ) if result.returncode != 0: removal_errors.append(f"unset-env: {result.stderr}") + removal_errors.extend(self._remove_legacy_app_overrides(app_id)) + if removal_errors: self.log.warning(f"Some override removals had issues for {app_id}: {'; '.join(removal_errors)}") diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 49ed711..7eee848 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -1,443 +1,311 @@ -""" -Installation service for lsfg-vk. -""" +"""Installation and in-place migration service for lsfg-vk v2.""" -import os +import json 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 Any, 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 + ARMADA_DEVICE_ENV, + BIN_DIR, + JSON_FILENAME, + LEGACY_JSON_FILENAME, + LEGACY_LIB_FILENAME, + LIB_FILENAME, + get_layer_archive_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""" - + """Install architecture-matched v2 assets and migrate the public config.""" + def __init__(self, logger=None): super().__init__(logger) - self.lib_file = self.local_lib_dir / LIB_FILENAME self.json_file = self.local_share_dir / JSON_FILENAME - + self.legacy_lib_file = self.local_lib_dir / LEGACY_LIB_FILENAME + self.legacy_json_file = self.local_share_dir / LEGACY_JSON_FILENAME + self._config_recovery_backup: Path | None = None + 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}" + archive_name = get_layer_archive_filename(self._is_arm_architecture()) + archive_path = plugin_dir / BIN_DIR / archive_name + if not archive_path.exists(): + error_msg = f"{archive_name} not found at {archive_path}" self.log.error(error_msg) return self._error_response(InstallationResponse, error_msg, message="") - + 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}") - - 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="") - + self._extract_and_install_files(archive_path) + profile_data = self._create_config_file() + self._create_lsfg_launch_script(profile_data) + self._remove_legacy_layer_files() + + message = "lsfg-vk v2 installed successfully" + if self._config_recovery_backup is not None: + message += f"; unsupported config backed up to {self._config_recovery_backup.name}" + self.log.info("lsfg-vk v2 installed successfully from %s", archive_name) + return self._success_response(InstallationResponse, message) + except (OSError, tarfile.TarError, shutil.Error) as error: + self.log.error("Error installing lsfg-vk: %s", error) + return self._error_response(InstallationResponse, str(error), message="") + except Exception as error: + self.log.error("Unexpected error installing lsfg-vk: %s", error) + return self._error_response(InstallationResponse, str(error), message="") + def _is_arm_architecture(self) -> bool: - """Check if running on ARM architecture - - Returns: - True if running on ARM (aarch64), False otherwise - """ - if platform.machine().lower() in ('aarch64', 'arm64'): + """Detect the native game host, including Decky running under FEX on Armada.""" + if platform.machine().lower() in ("aarch64", "arm64"): return True - - # 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 - - # Fall back to the native PID 1 ELF header. e_machine 183 is AArch64. try: - with Path('/proc/1/exe').open('rb') as host_executable: + 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 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}") - + except OSError as error: + self.log.debug("Could not inspect native host architecture: %s", error) 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 _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 + def _copy_plugin_file(source: Path, destination: Path, mode: int = 0o644) -> None: + """Copy payload content without preserving FEX-incompatible metadata.""" + shutil.copyfile(source, destination) + destination.chmod(mode) + + def _extract_and_install_files(self, archive_path: Path) -> None: + """Extract just the v2 library and manifest from an owner release archive.""" + destinations = { + LIB_FILENAME: self.lib_file, + JSON_FILENAME: self.json_file, } - - with zipfile.ZipFile(zip_path, 'r') as zip_ref: + found = set() + with tarfile.open(archive_path, "r:xz") as archive: 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 - """ + temp_root = Path(temp_dir) + for member in archive.getmembers(): + if not member.isfile(): + continue + filename = Path(member.name).name + destination = destinations.get(filename) + if destination is None: + continue + source = archive.extractfile(member) + if source is None: + continue + temporary_file = temp_root / filename + with source, temporary_file.open("wb") as output: + shutil.copyfileobj(source, output) + destination.parent.mkdir(parents=True, exist_ok=True) + if filename == JSON_FILENAME: + self._copy_and_fix_json_file(temporary_file, destination) + else: + self._copy_plugin_file(temporary_file, destination) + found.add(filename) + self.log.info("Installed %s to %s", filename, destination) + + missing = [name for name in destinations if name not in found] + if missing: + raise OSError("Archive did not contain required lsfg-vk files: " + ", ".join(missing)) + + def _copy_and_fix_json_file(self, source: Path, destination: Path) -> None: + """Point the manifest at ~/.local/lib after Decky installs it.""" 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 + data = json.loads(source.read_text(encoding="utf-8")) + data["layer"]["library_path"] = "../../../lib/liblsfg-vk-layer.so" + destination.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + destination.chmod(0o644) + except (json.JSONDecodeError, KeyError, OSError) as error: + self.log.error("Error fixing JSON file %s: %s", source, error) + self._copy_plugin_file(source, destination) + + def _create_config_file(self) -> ProfileData: + """Migrate v1 once, normalize v2, and retain Decky launch workarounds.""" from .dll_detection import DllDetectionService - - # Try to detect DLL path + dll_service = DllDetectionService(self.log) - - # Check if config file already exists + profile_data: ProfileData + was_legacy = False + self._config_recovery_backup = None if self.config_file_path.exists(): + content = self.config_file_path.read_text(encoding="utf-8") + was_legacy = ConfigurationManager.is_legacy_v1(content) 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) + profile_data = ConfigurationManager.parse_toml_content_multi_profile(content) + except (ValueError, KeyError, TypeError) as error: + self._config_recovery_backup = self._backup_config(content, "unrecognized") + self.log.warning( + "Could not parse existing config; saved it to %s and using defaults: %s", + self._config_recovery_backup, + error, + ) + default = ConfigurationManager.get_defaults_with_dll_detection(dll_service) + profile_data = ProfileData( + current_profile="decky-lsfg-vk", + profiles={"decky-lsfg-vk": default}, + global_config={"dll": default["dll"], "allow_fp16": default["allow_fp16"]}, + ) + if was_legacy: + self._backup_legacy_config(content) 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 + default = ConfigurationManager.get_defaults_with_dll_detection(dll_service) + profile_data = ProfileData( + current_profile="decky-lsfg-vk", + profiles={"decky-lsfg-vk": default}, + global_config={"dll": default["dll"], "allow_fp16": default["allow_fp16"]}, + ) + + profile_data = self._merge_config_with_defaults(profile_data, dll_service) + script_values = self._read_script_values() + current_profile = profile_data["current_profile"] + profile_data["profiles"][current_profile] = ConfigurationManager.merge_config_with_script( + profile_data["profiles"][current_profile], script_values + ) + self._write_file( + self.config_file_path, + ConfigurationManager.generate_toml_content_multi_profile(profile_data), + 0o644, + ) + return profile_data + + def _backup_config(self, content: str, label: str) -> Path: + backup_path = self.config_file_path.with_name(f"{self.config_file_path.name}.{label}.bak") + if not backup_path.exists(): + self._write_file(backup_path, content, 0o644) + self.log.info("Backed up %s configuration to %s", label, backup_path) + return backup_path + + def _backup_legacy_config(self, content: str) -> None: + self._backup_config(content, "v1") + + def _read_script_values(self) -> Dict[str, Any]: + if not self.lsfg_launch_script_path.exists(): + return {} 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 + return ConfigurationManager.parse_script_content( + self.lsfg_launch_script_path.read_text(encoding="utf-8") + ) + except OSError as error: + self.log.warning("Could not preserve launch-script settings: %s", error) + return {} + + def _create_lsfg_launch_script(self, profile_data: ProfileData = None) -> None: + """Write the v2 launcher after config migration has completed.""" 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}") - + + if profile_data is None: + profile_data = ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) + 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 + script = configuration_service._generate_script_content_for_profile(profile_data) + self._write_file(self.lsfg_launch_script_path, script, 0o755) + + def _merge_config_with_defaults(self, existing: ProfileData, dll_service) -> ProfileData: + defaults = ConfigurationManager.get_defaults_with_dll_detection(dll_service) + global_config = dict(existing.get("global_config", {})) + global_config.setdefault("dll", defaults["dll"]) + global_config.setdefault("allow_fp16", defaults["allow_fp16"]) + + profiles: Dict[str, Any] = {} + for name, existing_config in existing.get("profiles", {}).items(): + profiles[name] = ConfigurationManager.validate_config({ + **defaults, + **existing_config, + **global_config, + }) + if not profiles: + profiles["decky-lsfg-vk"] = ConfigurationManager.validate_config(defaults) + + current_profile = existing.get("current_profile", "decky-lsfg-vk") + if current_profile not in profiles: + current_profile = "decky-lsfg-vk" if "decky-lsfg-vk" in profiles else next(iter(profiles)) + return ProfileData( + current_profile=current_profile, + profiles=profiles, + global_config=global_config, + ) + + def _remove_legacy_layer_files(self) -> None: + for path in (self.legacy_lib_file, self.legacy_json_file): + if self._remove_if_exists(path): + self.log.info("Removed superseded v1 layer file %s", 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}") - + legacy_installed = self.legacy_lib_file.exists() or self.legacy_json_file.exists() return { "installed": lib_exists and json_exists, + "legacy_installed": legacy_installed, "lib_exists": lib_exists, "json_exists": json_exists, - "script_exists": config_exists, # Keep script_exists for backward compatibility + "script_exists": self.config_file_path.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.config_file_path), + "error": None, } - - except Exception as e: - error_msg = f"Error checking lsfg-vk installation: {str(e)}" - self.log.error(error_msg) + except OSError as error: return { "installed": False, + "legacy_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) + "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.json_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 OSError 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("Error cleaning up lsfg-vk files during uninstall: %s", error) + self.log.error("Traceback: %s", traceback.format_exc()) diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index cb59b4f..aefec3e 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -433,43 +433,16 @@ class Plugin: Called when the plugin is uninstalled. This method is called by Decky Loader when the plugin is being uninstalled. - Performs cleanup of plugin files and flatpak extensions. + Removes plugin-owned layer files while leaving shared Flatpak extensions installed. """ decky.logger.info("decky-lsfg-vk plugin being uninstalled") # Clean up lsfg-vk files when the plugin is uninstalled self.installation_service.cleanup_on_uninstall() - # Also clean up flatpak extensions if they are installed - try: - decky.logger.info("Checking for flatpak extensions to uninstall") - - extension_status = self.flatpak_service.get_extension_status() - - if extension_status.get("success"): - if extension_status.get("installed_23_08"): - decky.logger.info("Uninstalling lsfg-vk flatpak runtime 23.08") - result = self.flatpak_service.uninstall_extension("23.08") - if result.get("success"): - decky.logger.info("Successfully uninstalled flatpak runtime 23.08") - else: - decky.logger.warning(f"Failed to uninstall flatpak runtime 23.08: {result.get('error')}") - - if extension_status.get("installed_24_08"): - decky.logger.info("Uninstalling lsfg-vk flatpak runtime 24.08") - result = self.flatpak_service.uninstall_extension("24.08") - if result.get("success"): - decky.logger.info("Successfully uninstalled flatpak runtime 24.08") - else: - decky.logger.warning(f"Failed to uninstall flatpak runtime 24.08: {result.get('error')}") - - decky.logger.info("Flatpak extension cleanup completed") - else: - decky.logger.info(f"Could not check flatpak status for cleanup: {extension_status.get('error')}") - - except Exception as e: - decky.logger.error(f"Error during flatpak cleanup: {e}") - + # Runtime extensions are shared by all Flatpak lsfg-vk installations. + # Do not remove them automatically: another application or plugin may use them. + decky.logger.info("Leaving shared Flatpak runtime extensions installed") decky.logger.info("decky-lsfg-vk plugin uninstall cleanup completed") async def _migration(self): diff --git a/py_modules/lsfg_vk/types.py b/py_modules/lsfg_vk/types.py index 7b7ca2b..6f3cd05 100644 --- a/py_modules/lsfg_vk/types.py +++ b/py_modules/lsfg_vk/types.py @@ -37,6 +37,7 @@ class UninstallationResponse(BaseResponse): class InstallationCheckResponse(TypedDict): """Response for installation check""" installed: bool + legacy_installed: bool lib_exists: bool json_exists: bool script_exists: bool @@ -64,7 +65,7 @@ class ConfigurationResponse(BaseResponse): class ProfileConfig(TypedDict): """Configuration for a single profile""" - exe: str + name: str config: ConfigurationData |