diff options
Diffstat (limited to 'py_modules/lsfg_vk')
| -rw-r--r-- | py_modules/lsfg_vk/__init__.py | 14 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/base_service.py | 103 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/configuration.py | 175 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/constants.py | 54 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/dll_detection.py | 120 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 225 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 159 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/types.py | 71 |
8 files changed, 921 insertions, 0 deletions
diff --git a/py_modules/lsfg_vk/__init__.py b/py_modules/lsfg_vk/__init__.py new file mode 100644 index 0000000..8343853 --- /dev/null +++ b/py_modules/lsfg_vk/__init__.py @@ -0,0 +1,14 @@ +""" +lsfg-vk plugin package for Decky Loader. + +This package provides services for installing and managing the lsfg-vk +Vulkan layer for Lossless Scaling frame generation. +""" + +# Import will be available once plugin.py exists +try: + from .plugin import Plugin + __all__ = ['Plugin'] +except ImportError: + # During development, plugin may not exist yet + __all__ = [] diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py new file mode 100644 index 0000000..b547759 --- /dev/null +++ b/py_modules/lsfg_vk/base_service.py @@ -0,0 +1,103 @@ +""" +Base service class with common functionality. +""" + +import os +import shutil +import tempfile +from pathlib import Path +from typing import Any, Optional + +from .constants import LOCAL_LIB, LOCAL_SHARE_BASE, VULKAN_LAYER_DIR, SCRIPT_NAME + + +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: + import decky + self.log = decky.logger + else: + self.log = logger + + # Initialize common paths using pathlib + self.user_home = Path.home() + 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 + + 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.log.info(f"Ensured directories exist: {self.local_lib_dir}, {self.local_share_dir}") + + 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}") + return False + + def _atomic_write(self, path: Path, content: str, mode: int = 0o644) -> None: + """Write content to a file atomically + + Args: + path: Target file path + content: Content to write + mode: File permissions (default: 0o644) + + Raises: + OSError: If write fails + """ + # Create temporary file in the same directory to ensure atomic move + temp_path = None + try: + with tempfile.NamedTemporaryFile( + mode='w', + dir=path.parent, + delete=False, + prefix=f'.{path.name}.', + suffix='.tmp' + ) as temp_file: + temp_file.write(content) + temp_path = Path(temp_file.name) + + # Set permissions before moving + temp_path.chmod(mode) + + # Atomic move + temp_path.replace(path) + self.log.info(f"Atomically wrote to {path}") + + except Exception: + # Clean up temp file if something went wrong + if temp_path and temp_path.exists(): + try: + temp_path.unlink() + except OSError: + pass # Best effort cleanup + raise diff --git a/py_modules/lsfg_vk/configuration.py b/py_modules/lsfg_vk/configuration.py new file mode 100644 index 0000000..f5e2981 --- /dev/null +++ b/py_modules/lsfg_vk/configuration.py @@ -0,0 +1,175 @@ +""" +Configuration service for lsfg script management. +""" + +import re +from pathlib import Path +from typing import Dict, Any + +from .base_service import BaseService +from .constants import LSFG_SCRIPT_TEMPLATE +from .types import ConfigurationResponse, ConfigurationData + + +class ConfigurationService(BaseService): + """Service for managing lsfg script configuration""" + + def get_config(self) -> ConfigurationResponse: + """Read current lsfg script configuration + + Returns: + ConfigurationResponse with current configuration or error + """ + try: + if not self.lsfg_script_path.exists(): + return { + "success": False, + "config": None, + "message": None, + "error": "lsfg script not found" + } + + content = self.lsfg_script_path.read_text() + config = self._parse_script_content(content) + + self.log.info(f"Parsed lsfg config: {config}") + + return { + "success": True, + "config": config, + "message": None, + "error": None + } + + 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) + } + + def _parse_script_content(self, content: str) -> ConfigurationData: + """Parse script content to extract configuration values + + Args: + content: Script file content + + Returns: + ConfigurationData with parsed values + """ + config: ConfigurationData = { + "enable_lsfg": False, + "multiplier": 2, + "flow_scale": 1.0, + "hdr": False, + "perf_mode": False, + "immediate_mode": False + } + + lines = content.split('\n') + for line in lines: + line = line.strip() + + # Parse ENABLE_LSFG + if match := re.match(r'^(#\s*)?export\s+ENABLE_LSFG=(\d+)', line): + config["enable_lsfg"] = not bool(match.group(1)) and match.group(2) == '1' + + # Parse LSFG_MULTIPLIER + elif match := re.match(r'^export\s+LSFG_MULTIPLIER=(\d+)', line): + try: + config["multiplier"] = int(match.group(1)) + except ValueError: + pass + + # Parse LSFG_FLOW_SCALE + elif match := re.match(r'^export\s+LSFG_FLOW_SCALE=([0-9]*\.?[0-9]+)', line): + try: + config["flow_scale"] = float(match.group(1)) + except ValueError: + pass + + # Parse LSFG_HDR + elif match := re.match(r'^(#\s*)?export\s+LSFG_HDR=(\d+)', line): + config["hdr"] = not bool(match.group(1)) and match.group(2) == '1' + + # Parse LSFG_PERF_MODE + elif match := re.match(r'^(#\s*)?export\s+LSFG_PERF_MODE=(\d+)', line): + config["perf_mode"] = not bool(match.group(1)) and match.group(2) == '1' + + # Parse MESA_VK_WSI_PRESENT_MODE + elif match := re.match(r'^(#\s*)?export\s+MESA_VK_WSI_PRESENT_MODE=([^\s#]+)', line): + config["immediate_mode"] = not bool(match.group(1)) and match.group(2) == 'immediate' + + return config + + def update_config(self, enable_lsfg: bool, multiplier: int, flow_scale: float, + hdr: bool, perf_mode: bool, immediate_mode: bool) -> ConfigurationResponse: + """Update lsfg script configuration + + Args: + enable_lsfg: Whether to enable LSFG + multiplier: LSFG multiplier value + flow_scale: LSFG flow scale value + hdr: Whether to enable HDR + perf_mode: Whether to enable performance mode + immediate_mode: Whether to enable immediate present mode (disable vsync) + + Returns: + ConfigurationResponse with success status + """ + try: + # Generate script content using template + script_content = self._generate_script_content( + enable_lsfg, multiplier, flow_scale, hdr, perf_mode, immediate_mode + ) + + # Write the updated script atomically + self._atomic_write(self.lsfg_script_path, script_content, 0o755) + + self.log.info(f"Updated lsfg script configuration: enable={enable_lsfg}, " + f"multiplier={multiplier}, flow_scale={flow_scale}, hdr={hdr}, " + f"perf_mode={perf_mode}, immediate_mode={immediate_mode}") + + return { + "success": True, + "config": None, + "message": "lsfg configuration updated successfully", + "error": None + } + + 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) + } + + def _generate_script_content(self, enable_lsfg: bool, multiplier: int, flow_scale: float, + hdr: bool, perf_mode: bool, immediate_mode: bool) -> str: + """Generate script content from configuration parameters + + Args: + enable_lsfg: Whether to enable LSFG + multiplier: LSFG multiplier value + flow_scale: LSFG flow scale value + hdr: Whether to enable HDR + perf_mode: Whether to enable performance mode + immediate_mode: Whether to enable immediate present mode + + Returns: + Generated script content + """ + return LSFG_SCRIPT_TEMPLATE.format( + enable_lsfg="export ENABLE_LSFG=1" if enable_lsfg else "# export ENABLE_LSFG=1", + multiplier=multiplier, + flow_scale=flow_scale, + hdr="export LSFG_HDR=1" if hdr else "# export LSFG_HDR=1", + perf_mode="export LSFG_PERF_MODE=1" if perf_mode else "# export LSFG_PERF_MODE=1", + immediate_mode="export MESA_VK_WSI_PRESENT_MODE=immediate # - disable vsync" if immediate_mode else "# export MESA_VK_WSI_PRESENT_MODE=immediate # - disable vsync" + ) diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py new file mode 100644 index 0000000..28246c2 --- /dev/null +++ b/py_modules/lsfg_vk/constants.py @@ -0,0 +1,54 @@ +""" +Constants for the lsfg-vk plugin. +""" + +from pathlib import Path + +# Directory paths +LOCAL_LIB = ".local/lib" +LOCAL_SHARE_BASE = ".local/share" +VULKAN_LAYER_DIR = ".local/share/vulkan/implicit_layer.d" + +# File names +SCRIPT_NAME = "lsfg" +LIB_FILENAME = "liblsfg-vk.so" +JSON_FILENAME = "VkLayer_LS_frame_generation.json" +ZIP_FILENAME = "lsfg-vk_archlinux.zip" + +# File extensions +SO_EXT = ".so" +JSON_EXT = ".json" + +# Directory for the zip file +BIN_DIR = "bin" + +# Lossless Scaling paths +STEAM_COMMON_PATH = Path("steamapps/common/Lossless Scaling") +LOSSLESS_DLL_NAME = "Lossless.dll" + +# Script template +LSFG_SCRIPT_TEMPLATE = """#!/bin/bash + +{enable_lsfg} +export LSFG_MULTIPLIER={multiplier} +export LSFG_FLOW_SCALE={flow_scale} +{hdr} +{perf_mode} +{immediate_mode} + +# Execute the passed command with the environment variables set +exec "$@" +""" + +# Environment variable names +ENV_LSFG_DLL_PATH = "LSFG_DLL_PATH" +ENV_XDG_DATA_HOME = "XDG_DATA_HOME" +ENV_HOME = "HOME" + +# Default configuration values +DEFAULT_MULTIPLIER = 2 +DEFAULT_FLOW_SCALE = 1.0 +DEFAULT_ENABLE_LSFG = True +DEFAULT_HDR = False +DEFAULT_PERF_MODE = False +DEFAULT_IMMEDIATE_MODE = False diff --git a/py_modules/lsfg_vk/dll_detection.py b/py_modules/lsfg_vk/dll_detection.py new file mode 100644 index 0000000..f1dace9 --- /dev/null +++ b/py_modules/lsfg_vk/dll_detection.py @@ -0,0 +1,120 @@ +""" +DLL detection service for Lossless Scaling. +""" + +import os +from pathlib import Path +from typing import Dict, Any + +from .base_service import BaseService +from .constants import ( + ENV_LSFG_DLL_PATH, ENV_XDG_DATA_HOME, ENV_HOME, + STEAM_COMMON_PATH, LOSSLESS_DLL_NAME +) +from .types import DllDetectionResponse + + +class DllDetectionService(BaseService): + """Service for detecting Lossless Scaling DLL""" + + def check_lossless_scaling_dll(self) -> DllDetectionResponse: + """Check if Lossless Scaling DLL is available at the expected paths + + Returns: + DllDetectionResponse with detection status and path information + """ + try: + # Check environment variable first + dll_path = self._check_env_dll_path() + if dll_path: + return dll_path + + # Check XDG_DATA_HOME path + xdg_path = self._check_xdg_data_home() + if xdg_path: + return xdg_path + + # Check HOME/.local/share path + home_path = self._check_home_local_share() + if home_path: + return home_path + + # DLL not found in any expected location + return { + "detected": False, + "path": None, + "source": None, + "message": "Lossless Scaling DLL not found in expected locations", + "error": None + } + + except Exception as e: + error_msg = f"Error checking Lossless Scaling DLL: {str(e)}" + self.log.error(error_msg) + return { + "detected": False, + "path": None, + "source": None, + "message": None, + "error": str(e) + } + + def _check_env_dll_path(self) -> DllDetectionResponse | None: + """Check LSFG_DLL_PATH environment variable + + Returns: + DllDetectionResponse if found, None otherwise + """ + dll_path = os.getenv(ENV_LSFG_DLL_PATH) + if dll_path and dll_path.strip(): + dll_path_obj = Path(dll_path.strip()) + if dll_path_obj.exists(): + self.log.info(f"Found DLL via {ENV_LSFG_DLL_PATH}: {dll_path_obj}") + return { + "detected": True, + "path": str(dll_path_obj), + "source": f"{ENV_LSFG_DLL_PATH} environment variable", + "message": None, + "error": None + } + return None + + def _check_xdg_data_home(self) -> DllDetectionResponse | None: + """Check XDG_DATA_HOME Steam directory + + Returns: + DllDetectionResponse if found, None otherwise + """ + data_dir = os.getenv(ENV_XDG_DATA_HOME) + if data_dir and data_dir.strip(): + dll_path = Path(data_dir.strip()) / "Steam" / STEAM_COMMON_PATH / LOSSLESS_DLL_NAME + if dll_path.exists(): + self.log.info(f"Found DLL via {ENV_XDG_DATA_HOME}: {dll_path}") + return { + "detected": True, + "path": str(dll_path), + "source": f"{ENV_XDG_DATA_HOME} Steam directory", + "message": None, + "error": None + } + return None + + def _check_home_local_share(self) -> DllDetectionResponse | None: + """Check HOME/.local/share Steam directory + + Returns: + DllDetectionResponse if found, None otherwise + """ + home_dir = os.getenv(ENV_HOME) + if home_dir and home_dir.strip(): + dll_path = Path(home_dir.strip()) / ".local" / "share" / "Steam" / STEAM_COMMON_PATH / LOSSLESS_DLL_NAME + if dll_path.exists(): + self.log.info(f"Found DLL via {ENV_HOME}/.local/share: {dll_path}") + return { + "detected": True, + "path": str(dll_path), + "source": f"{ENV_HOME}/.local/share Steam directory", + "message": None, + "error": None + } + return None diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py new file mode 100644 index 0000000..1d0e96f --- /dev/null +++ b/py_modules/lsfg_vk/installation.py @@ -0,0 +1,225 @@ +""" +Installation service for lsfg-vk. +""" + +import os +import shutil +import zipfile +import tempfile +from pathlib import Path +from typing import Dict, Any + +from .base_service import BaseService +from .constants import ( + LIB_FILENAME, JSON_FILENAME, ZIP_FILENAME, BIN_DIR, + SO_EXT, JSON_EXT, LSFG_SCRIPT_TEMPLATE, + DEFAULT_MULTIPLIER, DEFAULT_FLOW_SCALE, DEFAULT_ENABLE_LSFG, + DEFAULT_HDR, DEFAULT_PERF_MODE, DEFAULT_IMMEDIATE_MODE +) +from .types import InstallationResponse, UninstallationResponse, InstallationCheckResponse + + +class InstallationService(BaseService): + """Service for handling lsfg-vk installation and uninstallation""" + + def __init__(self, logger=None): + super().__init__(logger) + + # File paths using constants + self.lib_file = self.local_lib_dir / LIB_FILENAME + self.json_file = self.local_share_dir / 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: + # Get the path to the zip file - need to go up to plugin root from py_modules/lsfg_vk/ + plugin_dir = Path(__file__).parent.parent.parent + zip_path = plugin_dir / BIN_DIR / ZIP_FILENAME + + # Check if the zip file exists + 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": ""} + + # Create directories if they don't exist + self._ensure_directories() + + # Extract and install files + self._extract_and_install_files(zip_path) + + # Create the lsfg script + self._create_lsfg_script() + + self.log.info("lsfg-vk installed successfully") + return {"success": True, "message": "lsfg-vk installed successfully", "error": None} + + 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": ""} + 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": ""} + + def _extract_and_install_files(self, zip_path: Path) -> None: + """Extract zip file and install files to appropriate locations + + Args: + zip_path: Path to the zip file to extract + + Raises: + zipfile.BadZipFile: If zip file is corrupted + OSError: If file operations fail + """ + # Destination mapping for file types + dest_map = { + SO_EXT: self.local_lib_dir, + JSON_EXT: self.local_share_dir + } + + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + with tempfile.TemporaryDirectory() as temp_dir: + temp_path = Path(temp_dir) + zip_ref.extractall(temp_path) + + # Process extracted files + for root, dirs, files in os.walk(temp_path): + root_path = Path(root) + for file in files: + src_file = root_path / file + file_path = Path(file) + + # Check if we know where this file type should go + dst_dir = dest_map.get(file_path.suffix) + if dst_dir: + dst_file = dst_dir / file + shutil.copy2(src_file, dst_file) + self.log.info(f"Copied {file} to {dst_file}") + + def _create_lsfg_script(self) -> None: + """Create the lsfg script in home directory with default configuration""" + script_content = LSFG_SCRIPT_TEMPLATE.format( + enable_lsfg="export ENABLE_LSFG=1" if DEFAULT_ENABLE_LSFG else "# export ENABLE_LSFG=1", + multiplier=DEFAULT_MULTIPLIER, + flow_scale=DEFAULT_FLOW_SCALE, + hdr="export LSFG_HDR=1" if DEFAULT_HDR else "# export LSFG_HDR=1", + perf_mode="export LSFG_PERF_MODE=1" if DEFAULT_PERF_MODE else "# export LSFG_PERF_MODE=1", + immediate_mode="export MESA_VK_WSI_PRESENT_MODE=immediate # - disable vsync" if DEFAULT_IMMEDIATE_MODE else "# export MESA_VK_WSI_PRESENT_MODE=immediate # - disable vsync" + ) + + # Use atomic write to prevent corruption + self._atomic_write(self.lsfg_script_path, script_content, 0o755) + self.log.info(f"Created executable lsfg script at {self.lsfg_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() + script_exists = self.lsfg_script_path.exists() + + self.log.info(f"Installation check: lib={lib_exists}, json={json_exists}, script={script_exists}") + + return { + "installed": lib_exists and json_exists, + "lib_exists": lib_exists, + "json_exists": json_exists, + "script_exists": script_exists, + "lib_path": str(self.lib_file), + "json_path": str(self.json_file), + "script_path": str(self.lsfg_script_path), + "error": None + } + + except Exception as e: + error_msg = f"Error checking lsfg-vk installation: {str(e)}" + self.log.error(error_msg) + return { + "installed": False, + "lib_exists": False, + "json_exists": False, + "script_exists": False, + "lib_path": str(self.lib_file), + "json_path": str(self.json_file), + "script_path": str(self.lsfg_script_path), + "error": str(e) + } + + def uninstall(self) -> UninstallationResponse: + """Uninstall lsfg-vk by removing the installed files + + Returns: + UninstallationResponse with success status and removed files list + """ + try: + removed_files = [] + files_to_remove = [self.lib_file, self.json_file, self.lsfg_script_path] + + for file_path in files_to_remove: + if self._remove_if_exists(file_path): + removed_files.append(str(file_path)) + + if not removed_files: + return { + "success": True, + "message": "No lsfg-vk files found to remove", + "removed_files": None, + "error": 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 + } + + 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) + } + + def cleanup_on_uninstall(self) -> None: + """Clean up lsfg-vk files when the plugin is uninstalled""" + 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" lsfg script: {self.lsfg_script_path}") + + removed_files = [] + files_to_remove = [self.lib_file, self.json_file, 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}") + + 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)}") + import traceback + self.log.error(f"Traceback: {traceback.format_exc()}") diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py new file mode 100644 index 0000000..000830f --- /dev/null +++ b/py_modules/lsfg_vk/plugin.py @@ -0,0 +1,159 @@ +""" +Main plugin class for the lsfg-vk Decky Loader plugin. + +This plugin provides services for installing and managing the lsfg-vk +Vulkan layer for Lossless Scaling frame generation on Steam Deck. +""" + +import os +from typing import Dict, Any + +from .installation import InstallationService +from .dll_detection import DllDetectionService +from .configuration import ConfigurationService + + +class Plugin: + """ + Main plugin class for lsfg-vk management. + + This class provides a unified interface for installation, configuration, + and DLL detection services. It implements the Decky Loader plugin lifecycle + methods (_main, _unload, _uninstall, _migration). + """ + + def __init__(self): + """Initialize the plugin with all necessary services""" + # Initialize services - they will use decky.logger by default + self.installation_service = InstallationService() + self.dll_detection_service = DllDetectionService() + self.configuration_service = ConfigurationService() + + # Installation methods + async def install_lsfg_vk(self) -> Dict[str, Any]: + """Install lsfg-vk by extracting the zip file to ~/.local + + Returns: + InstallationResponse dict with success status and message/error + """ + return self.installation_service.install() + + async def check_lsfg_vk_installed(self) -> Dict[str, Any]: + """Check if lsfg-vk is already installed + + Returns: + InstallationCheckResponse dict with installation status and paths + """ + return self.installation_service.check_installation() + + async def uninstall_lsfg_vk(self) -> Dict[str, Any]: + """Uninstall lsfg-vk by removing the installed files + + Returns: + UninstallationResponse dict with success status and removed files + """ + return self.installation_service.uninstall() + + # DLL detection methods + async def check_lossless_scaling_dll(self) -> Dict[str, Any]: + """Check if Lossless Scaling DLL is available at the expected paths + + Returns: + DllDetectionResponse dict with detection status and path info + """ + return self.dll_detection_service.check_lossless_scaling_dll() + + # Configuration methods + async def get_lsfg_config(self) -> Dict[str, Any]: + """Read current lsfg script configuration + + Returns: + ConfigurationResponse dict with current configuration or error + """ + return self.configuration_service.get_config() + + async def update_lsfg_config(self, enable_lsfg: bool, multiplier: int, flow_scale: float, + hdr: bool, perf_mode: bool, immediate_mode: bool) -> Dict[str, Any]: + """Update lsfg script configuration + + Args: + enable_lsfg: Whether to enable LSFG + multiplier: LSFG multiplier value (typically 2-4) + flow_scale: LSFG flow scale value (typically 0.5-2.0) + hdr: Whether to enable HDR + perf_mode: Whether to enable performance mode + immediate_mode: Whether to enable immediate present mode (disable vsync) + + Returns: + ConfigurationResponse dict with success status + """ + return self.configuration_service.update_config( + enable_lsfg, multiplier, flow_scale, hdr, perf_mode, immediate_mode + ) + + # Plugin lifecycle methods + async def _main(self): + """ + Asyncio-compatible long-running code, executed in a task when the plugin is loaded. + + This method is called by Decky Loader when the plugin starts up. + Currently just logs that the plugin has loaded successfully. + """ + import decky + decky.logger.info("Lossless Scaling VK plugin loaded!") + + async def _unload(self): + """ + Function called first during the unload process. + + This method is called by Decky Loader when the plugin is being unloaded. + Use this for cleanup that should happen when the plugin stops. + """ + import decky + decky.logger.info("Lossless Scaling VK plugin unloading") + + async def _uninstall(self): + """ + Function called after `_unload` during uninstall. + + This method is called by Decky Loader when the plugin is being uninstalled. + It automatically cleans up any lsfg-vk files that were installed. + """ + import decky + decky.logger.info("Lossless Scaling VK plugin uninstalled - starting cleanup") + + # Clean up lsfg-vk files when the plugin is uninstalled + self.installation_service.cleanup_on_uninstall() + + decky.logger.info("Lossless Scaling VK plugin uninstall cleanup completed") + + async def _migration(self): + """ + Migrations that should be performed before entering `_main()`. + + This method is called by Decky Loader for plugin migrations. + Currently migrates logs, settings, and runtime data from old locations. + """ + import decky + decky.logger.info("Running Lossless Scaling VK plugin migrations") + + # Migrate logs from old location + # ~/.config/decky-lossless-scaling-vk/lossless-scaling-vk.log -> decky.DECKY_LOG_DIR/lossless-scaling-vk.log + decky.migrate_logs(os.path.join(decky.DECKY_USER_HOME, + ".config", "decky-lossless-scaling-vk", "lossless-scaling-vk.log")) + + # Migrate settings from old locations + # ~/homebrew/settings/lossless-scaling-vk.json -> decky.DECKY_SETTINGS_DIR/lossless-scaling-vk.json + # ~/.config/decky-lossless-scaling-vk/ -> decky.DECKY_SETTINGS_DIR/ + decky.migrate_settings( + os.path.join(decky.DECKY_HOME, "settings", "lossless-scaling-vk.json"), + os.path.join(decky.DECKY_USER_HOME, ".config", "decky-lossless-scaling-vk")) + + # Migrate runtime data from old locations + # ~/homebrew/lossless-scaling-vk/ -> decky.DECKY_RUNTIME_DIR/ + # ~/.local/share/decky-lossless-scaling-vk/ -> decky.DECKY_RUNTIME_DIR/ + decky.migrate_runtime( + os.path.join(decky.DECKY_HOME, "lossless-scaling-vk"), + os.path.join(decky.DECKY_USER_HOME, ".local", "share", "decky-lossless-scaling-vk")) + + decky.logger.info("Lossless Scaling VK plugin migrations completed") diff --git a/py_modules/lsfg_vk/types.py b/py_modules/lsfg_vk/types.py new file mode 100644 index 0000000..f0ec892 --- /dev/null +++ b/py_modules/lsfg_vk/types.py @@ -0,0 +1,71 @@ +""" +Type definitions for the lsfg-vk plugin responses. +""" + +from typing import TypedDict, Optional, List + + +class BaseResponse(TypedDict): + """Base response structure""" + success: bool + + +class ErrorResponse(BaseResponse): + """Response structure for errors""" + error: str + + +class MessageResponse(BaseResponse): + """Response structure with message""" + message: str + + +class InstallationResponse(BaseResponse): + """Response for installation operations""" + message: str + error: Optional[str] + + +class UninstallationResponse(BaseResponse): + """Response for uninstallation operations""" + message: str + removed_files: Optional[List[str]] + error: Optional[str] + + +class InstallationCheckResponse(TypedDict): + """Response for installation check""" + installed: bool + lib_exists: bool + json_exists: bool + script_exists: bool + lib_path: str + json_path: str + script_path: str + error: Optional[str] + + +class DllDetectionResponse(TypedDict): + """Response for DLL detection""" + detected: bool + path: Optional[str] + source: Optional[str] + message: Optional[str] + error: Optional[str] + + +class ConfigurationData(TypedDict): + """Configuration data structure""" + enable_lsfg: bool + multiplier: int + flow_scale: float + hdr: bool + perf_mode: bool + immediate_mode: bool + + +class ConfigurationResponse(BaseResponse): + """Response for configuration operations""" + config: Optional[ConfigurationData] + message: Optional[str] + error: Optional[str] |