diff options
Diffstat (limited to 'py_modules')
| -rw-r--r-- | py_modules/lsfg_vk/base_service.py | 44 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/config_schema.py | 105 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/configuration.py | 78 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 34 |
4 files changed, 122 insertions, 139 deletions
diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py index b595b07..b684ec9 100644 --- a/py_modules/lsfg_vk/base_service.py +++ b/py_modules/lsfg_vk/base_service.py @@ -5,10 +5,13 @@ Base service class with common functionality. import os import shutil from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, TypeVar, Dict from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME, CONFIG_DIR, CONFIG_FILENAME +# Generic type for response dictionaries +ResponseType = TypeVar('ResponseType', bound=Dict[str, Any]) + class BaseService: """Base service class with common functionality""" @@ -90,3 +93,42 @@ class BaseService: except Exception: self.log.error(f"Failed to write to {path}") 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.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.update(kwargs) + return response diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py index 4f036ff..39593ab 100644 --- a/py_modules/lsfg_vk/config_schema.py +++ b/py_modules/lsfg_vk/config_schema.py @@ -9,18 +9,28 @@ This module defines the complete configuration structure for lsfg-vk, managing T """ import re +import sys from typing import TypedDict, Dict, Any, Union, cast from dataclasses import dataclass from enum import Enum from pathlib import Path +# 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 -class ConfigFieldType(Enum): - """Supported configuration field types""" - BOOLEAN = "boolean" - INTEGER = "integer" - FLOAT = "float" - STRING = "string" + +@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 @dataclass @@ -36,51 +46,25 @@ class ConfigField: return value -# Configuration schema definition +# Use shared configuration schema as source of truth CONFIG_SCHEMA: Dict[str, ConfigField] = { - "dll": ConfigField( - name="dll", - field_type=ConfigFieldType.STRING, - default="", # Will be populated dynamically based on detection - description="specify where Lossless.dll is stored" - ), - - "multiplier": ConfigField( - name="multiplier", - field_type=ConfigFieldType.INTEGER, - default=1, - description="change the fps multiplier" - ), - - "flow_scale": ConfigField( - name="flow_scale", - field_type=ConfigFieldType.FLOAT, - default=0.8, - description="change the flow scale" - ), - - "performance_mode": ConfigField( - name="performance_mode", - field_type=ConfigFieldType.BOOLEAN, - default=True, - description="toggle performance mode" - ), - - "hdr_mode": ConfigField( - name="hdr_mode", - field_type=ConfigFieldType.BOOLEAN, - default=False, - description="enable hdr mode" - ), - - "experimental_present_mode": ConfigField( - name="experimental_present_mode", - field_type=ConfigFieldType.STRING, - default="fifo", - description="experimental: override vulkan present mode (fifo/mailbox/immediate)" - ), + 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() } +# 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" +) + # Fields that should ONLY be in the lsfg script, not in TOML config SCRIPT_ONLY_FIELDS = { "dxvk_frame_rate": ConfigField( @@ -128,10 +112,16 @@ class ConfigurationManager: @staticmethod def get_defaults() -> ConfigurationData: """Get default configuration values""" - return cast(ConfigurationData, { + # 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 COMPLETE_CONFIG_SCHEMA.values() - }) + for field in SCRIPT_ONLY_FIELDS.values() + } + + return cast(ConfigurationData, {**shared_defaults, **script_defaults}) @staticmethod def get_defaults_with_dll_detection(dll_detection_service=None) -> ConfigurationData: @@ -164,15 +154,18 @@ class ConfigurationManager: @staticmethod def get_field_names() -> list[str]: """Get ordered list of configuration field names""" - return list(COMPLETE_CONFIG_SCHEMA.keys()) + # 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 @staticmethod def get_field_types() -> Dict[str, ConfigFieldType]: """Get field type mapping""" - return { - field.name: field.field_type - for field in CONFIG_SCHEMA.values() - } + # 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} @staticmethod def validate_config(config: Dict[str, Any]) -> ConfigurationData: diff --git a/py_modules/lsfg_vk/configuration.py b/py_modules/lsfg_vk/configuration.py index da765e0..47d0ebc 100644 --- a/py_modules/lsfg_vk/configuration.py +++ b/py_modules/lsfg_vk/configuration.py @@ -43,22 +43,12 @@ class ConfigurationService(BaseService): # Merge TOML config with script values config = ConfigurationManager.merge_config_with_script(toml_config, script_values) - return { - "success": True, - "config": config, - "message": None, - "error": None - } + 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 { - "success": False, - "config": None, - "message": None, - "error": str(e) - } + 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) @@ -66,12 +56,9 @@ class ConfigurationService(BaseService): from .dll_detection import DllDetectionService dll_service = DllDetectionService(self.log) config = ConfigurationManager.get_defaults_with_dll_detection(dll_service) - return { - "success": True, - "config": config, - "message": f"Using default configuration due to parse error: {str(e)}", - "error": None - } + return self._success_response(ConfigurationResponse, + f"Using default configuration due to parse error: {str(e)}", + config=config) def update_config(self, dll: str, multiplier: int, flow_scale: float, performance_mode: bool, hdr_mode: bool, @@ -123,31 +110,18 @@ class ConfigurationService(BaseService): f"dxvk_frame_rate={dxvk_frame_rate}, " f"enable_wow64={enable_wow64}, disable_steamdeck_mode={disable_steamdeck_mode}") - return { - "success": True, - "config": config, - "message": "lsfg configuration updated successfully", - "error": None - } + return self._success_response(ConfigurationResponse, + "lsfg configuration updated successfully", + config=config) except (OSError, IOError) as e: error_msg = f"Error updating lsfg config: {str(e)}" self.log.error(error_msg) - return { - "success": False, - "config": None, - "message": None, - "error": str(e) - } + 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 { - "success": False, - "config": None, - "message": None, - "error": str(e) - } + return self._error_response(ConfigurationResponse, str(e), config=None) def update_dll_path(self, dll_path: str) -> ConfigurationResponse: """Update just the DLL path in the configuration @@ -183,22 +157,14 @@ class ConfigurationService(BaseService): self.log.info(f"Updated DLL path in lsfg configuration: '{dll_path}'") - return { - "success": True, - "config": config, - "message": f"DLL path updated to: {dll_path}", - "error": None - } + return self._success_response(ConfigurationResponse, + f"DLL path updated to: {dll_path}", + config=config) except Exception as e: error_msg = f"Error updating DLL path: {str(e)}" self.log.error(error_msg) - return { - "success": False, - "config": None, - "message": None, - "error": str(e) - } + return self._error_response(ConfigurationResponse, str(e), config=None) def update_lsfg_script(self, config: ConfigurationData) -> ConfigurationResponse: """Update the ~/lsfg launch script with current configuration @@ -217,22 +183,14 @@ class ConfigurationService(BaseService): self.log.info(f"Updated lsfg launch script at {self.lsfg_script_path}") - return { - "success": True, - "config": config, - "message": "Launch script updated successfully", - "error": None - } + 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 { - "success": False, - "config": None, - "message": None, - "error": str(e) - } + 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 diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 5bfc88b..b340093 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -43,7 +43,7 @@ class InstallationService(BaseService): if not zip_path.exists(): error_msg = f"{ZIP_FILENAME} not found at {zip_path}" self.log.error(error_msg) - return {"success": False, "error": error_msg, "message": ""} + return self._error_response(InstallationResponse, error_msg, message="") # Create directories if they don't exist self._ensure_directories() @@ -58,17 +58,17 @@ class InstallationService(BaseService): self._create_lsfg_launch_script() self.log.info("lsfg-vk installed successfully") - return {"success": True, "message": "lsfg-vk installed successfully", "error": None} + 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 {"success": False, "error": str(e), "message": ""} + return self._error_response(InstallationResponse, str(e), message="") except Exception as e: # Catch unexpected errors but log them separately error_msg = f"Unexpected error installing lsfg-vk: {str(e)}" self.log.error(error_msg) - return {"success": False, "error": str(e), "message": ""} + 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 @@ -209,30 +209,20 @@ class InstallationService(BaseService): pass # Directory not empty or other error, ignore if not removed_files: - return { - "success": True, - "message": "No lsfg-vk files found to remove", - "removed_files": None, - "error": None - } + return self._success_response(UninstallationResponse, + "No lsfg-vk files found to remove", + removed_files=None) self.log.info("lsfg-vk uninstalled successfully") - return { - "success": True, - "message": f"lsfg-vk uninstalled successfully. Removed {len(removed_files)} files.", - "removed_files": removed_files, - "error": None - } + 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 { - "success": False, - "message": "", - "removed_files": None, - "error": str(e) - } + return self._error_response(UninstallationResponse, str(e), + message="", removed_files=None) def cleanup_on_uninstall(self) -> None: """Clean up lsfg-vk files when the plugin is uninstalled""" |