diff options
Diffstat (limited to 'py_modules/lsfg_vk')
| -rw-r--r-- | py_modules/lsfg_vk/base_service.py | 6 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/configuration.py | 64 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 21 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 23 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 48 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/steam_service.py | 77 |
6 files changed, 129 insertions, 110 deletions
diff --git a/py_modules/lsfg_vk/base_service.py b/py_modules/lsfg_vk/base_service.py index c4978c6..036dfc6 100644 --- a/py_modules/lsfg_vk/base_service.py +++ b/py_modules/lsfg_vk/base_service.py @@ -13,12 +13,12 @@ ResponseType = TypeVar("ResponseType", bound=Dict[str, Any]) class BaseService: def __init__(self, logger: Optional[Any] = None): self.log = decky.logger if logger is None else logger - self.user_home = Path.home() + decky_user_home = getattr(decky, "DECKY_USER_HOME", None) + self.user_home = Path(decky_user_home) if decky_user_home else Path.home() self.local_bin_dir = self.user_home / LOCAL_BIN self.local_lib_dir = self.user_home / LOCAL_LIB self.local_share_dir = self.user_home / VULKAN_LAYER_DIR - self.lsfg_script_path = self.user_home / SCRIPT_NAME - self.lsfg_launch_script_path = self.user_home / SCRIPT_NAME + self.legacy_script_path = self.user_home / SCRIPT_NAME self.config_dir = self.user_home / CONFIG_DIR self.config_file_path = self.config_dir / CONFIG_FILENAME diff --git a/py_modules/lsfg_vk/configuration.py b/py_modules/lsfg_vk/configuration.py index 2626a66..a4ee2cc 100644 --- a/py_modules/lsfg_vk/configuration.py +++ b/py_modules/lsfg_vk/configuration.py @@ -1,5 +1,4 @@ import re -import shlex from typing import Any, Dict from .base_service import BaseService @@ -29,10 +28,27 @@ class ConfigurationService(BaseService): self._write_file(self.config_file_path, content, 0o644) @staticmethod - def _game_profile_name(appid: str) -> str: - if not re.fullmatch(r"[0-9]+", str(appid)): - raise ValueError("appid must be numeric") - return f"game-{appid}" + def _profile_name(data: ProfileData, appid: str, game_name: str) -> str: + name = str(game_name).strip() + if not name: + raise ValueError("game name is required") + if name == DEFAULT_PROFILE_NAME: + name = f"{name} ({appid})" + existing = data["profiles"].get(name) + if existing is not None and str(appid) not in existing.get("active_in", []): + name = f"{name} ({appid})" + return name + + @staticmethod + def _profile_for_appid(data: ProfileData, appid: str): + return next( + ( + (name, profile) + for name, profile in data["profiles"].items() + if str(appid) in profile.get("active_in", []) + ), + (None, None), + ) @staticmethod def _public_config(config: Dict[str, Any]) -> Dict[str, Any]: @@ -52,7 +68,7 @@ class ConfigurationService(BaseService): games = [] for name, raw in data["profiles"].items(): active_in = raw.get("active_in", []) - if len(active_in) != 1 or not str(active_in[0]).isdigit(): + if len(active_in) != 1 or not re.fullmatch(r"-?[0-9]+", str(active_in[0])): continue games.append({"appid": str(active_in[0]), "profile": name, "config": self._public_config(raw)}) return self._success_response(dict, default=self._public_config(data["profiles"][DEFAULT_PROFILE_NAME]), games=games) @@ -63,20 +79,20 @@ class ConfigurationService(BaseService): def get_game_config(self, appid: str) -> Dict[str, Any]: try: data = self._get_profile_data() - name = self._game_profile_name(appid) - profile = data["profiles"].get(name) - if profile is None: - profile = next((value for value in data["profiles"].values() if str(appid) in value.get("active_in", [])), None) + _, profile = self._profile_for_appid(data, appid) return self._success_response(dict, appid=str(appid), exists=profile is not None, config=self._public_config(profile or data["profiles"][DEFAULT_PROFILE_NAME])) except Exception as error: return self._error_response(dict, str(error), appid=str(appid), exists=False, config=None) - def update_game_config(self, appid: str, config: Dict[str, Any]) -> Dict[str, Any]: + def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]: try: data = self._get_profile_data() - name = self._game_profile_name(appid) + old_name, _ = self._profile_for_appid(data, appid) + name = self._profile_name(data, appid, game_name) validated = self._public_config(config) validated["active_in"] = [str(appid)] + if old_name and old_name != name: + data["profiles"].pop(old_name, None) data["profiles"][name] = validated self._save_profile_data(data) return self._success_response(dict, appid=str(appid), config=validated) @@ -86,11 +102,9 @@ class ConfigurationService(BaseService): def reset_game_config(self, appid: str) -> Dict[str, Any]: try: data = self._get_profile_data() - name = self._game_profile_name(appid) - data["profiles"].pop(name, None) - for profile_name, profile in list(data["profiles"].items()): - if profile_name != DEFAULT_PROFILE_NAME and str(appid) in profile.get("active_in", []): - data["profiles"].pop(profile_name) + name, _ = self._profile_for_appid(data, appid) + if name: + data["profiles"].pop(name, None) self._save_profile_data(data) return self._success_response(dict, appid=str(appid), config=self._public_config(data["profiles"][DEFAULT_PROFILE_NAME])) except Exception as error: @@ -116,19 +130,3 @@ class ConfigurationService(BaseService): return self._success_response(dict, config=validated) except Exception as error: return self._error_response(dict, str(error), config=None) - - def update_lsfg_script(self, config: Dict[str, Any]) -> Dict[str, Any]: - return self.update_config_from_dict(config) - - def _generate_script_content_for_profile(self, profile_data: ProfileData) -> str: - return "#!/bin/bash\n" f"export LSFGVK_CONFIG={shlex.quote(str(self.config_file_path))}\n" 'exec "$@"\n' - - def _generate_script_content(self, config: Dict[str, Any]) -> str: - return self._generate_script_content_for_profile(self._default_data()) - - def update_lsfg_script_from_profile_data(self, profile_data: ProfileData) -> Dict[str, Any]: - try: - self._write_file(self.lsfg_script_path, self._generate_script_content_for_profile(profile_data), 0o755) - return self._success_response(dict) - except Exception as error: - return self._error_response(dict, str(error)) diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 302efc7..6f8a596 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,4 +1,5 @@ import os +import pwd import shutil import subprocess from pathlib import Path @@ -26,6 +27,7 @@ class FlatpakService(BaseService): def _get_clean_env(self) -> Dict[str, str]: env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) + env["HOME"] = str(self.user_home) path_entries = [entry for entry in env.get("PATH", "").split(":") if entry] for entry in ("/usr/bin", "/usr/local/bin", "/bin"): if entry not in path_entries: @@ -33,6 +35,12 @@ class FlatpakService(BaseService): env["PATH"] = ":".join(path_entries) return env + def _flatpak_user(self) -> pwd.struct_passwd: + try: + return pwd.getpwuid(self.user_home.stat().st_uid) + except (KeyError, OSError) as error: + raise RuntimeError(f"Unable to resolve Flatpak user for {self.user_home}") from error + def check_flatpak_available(self) -> bool: env = self._get_clean_env() self.flatpak_command = shutil.which("flatpak", path=env["PATH"]) @@ -41,8 +49,15 @@ class FlatpakService(BaseService): def _run_flatpak_command(self, args: List[str], **kwargs): if self.flatpak_command is None and not self.check_flatpak_available(): raise FileNotFoundError("Flatpak command not available") + command = [self.flatpak_command, *args] + target_user = self._flatpak_user() + if os.geteuid() != target_user.pw_uid: + runuser = shutil.which("runuser", path=self._get_clean_env()["PATH"]) + if runuser is None: + raise FileNotFoundError("runuser command not available") + command = [runuser, "--user", target_user.pw_name, "--", *command] return subprocess.run( - [self.flatpak_command, *args], + command, env=self._get_clean_env(), **kwargs, ) @@ -181,7 +196,7 @@ class FlatpakService(BaseService): self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" ), - "legacy_script": str(self.lsfg_launch_script_path), + "legacy_script": str(self.legacy_script_path), } def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: @@ -200,7 +215,7 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["list", "--user", "--app", "--columns=name,application"], + ["list", "--app", "--columns=name,application"], capture_output=True, text=True, check=True, diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index f7bfdaf..0a728c7 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -63,7 +63,6 @@ class InstallationService(BaseService): config_content, 0o644, ) - self._create_lsfg_launch_script(profile_data) self._remove_legacy_layer_files() return self._success_response(InstallationResponse, "lsfg-vk 2.0.0 installed successfully") except Exception as error: @@ -170,20 +169,6 @@ class InstallationService(BaseService): return True return False - def _create_lsfg_launch_script(self, profile_data: ProfileData) -> None: - from .configuration import ConfigurationService - - configuration_service = ConfigurationService(logger=self.log) - configuration_service.user_home = self.user_home - configuration_service.config_dir = self.config_dir - configuration_service.config_file_path = self.config_file_path - configuration_service.lsfg_script_path = self.lsfg_launch_script_path - self._write_file( - self.lsfg_launch_script_path, - configuration_service._generate_script_content_for_profile(profile_data), - 0o755, - ) - def _remove_legacy_layer_files(self) -> None: for path in (self.legacy_lib_file, self.legacy_json_file): self._remove_if_exists(path) @@ -212,15 +197,11 @@ class InstallationService(BaseService): except Exception: return True - def get_launch_script_path(self) -> str: - return str(self.lsfg_launch_script_path) - def check_installation(self) -> InstallationCheckResponse: try: - script_exists = self.lsfg_launch_script_path.exists() installation_error = None try: - installed = script_exists and self.runtime_service.is_healthy() + installed = self.runtime_service.is_healthy() except Exception as error: installed = False installation_error = str(error) @@ -254,7 +235,7 @@ class InstallationService(BaseService): self.user_home / LOCAL_SHARE / "icons" / "hicolor" / "256x256" / "apps" / UI_ICON_FILENAME, self.legacy_lib_file, self.legacy_json_file, - self.lsfg_launch_script_path, + self.legacy_script_path, ): if self._remove_if_exists(path): removed.append(str(path)) diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index 3b9a97f..8c788a8 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -103,8 +103,8 @@ class Plugin: async def get_game_config(self, appid: str) -> Dict[str, Any]: return self.configuration_service.get_game_config(appid) - async def update_game_config(self, appid: str, config: Dict[str, Any]) -> Dict[str, Any]: - return self.configuration_service.update_game_config(appid, config) + async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]: + return self.configuration_service.update_game_config(appid, game_name, config) async def reset_game_config(self, appid: str) -> Dict[str, Any]: return self.configuration_service.reset_game_config(appid) @@ -178,18 +178,6 @@ class Plugin: """ return {"success": False, "error": "Use update_game_config with a Steam AppID"} - async def get_launch_option(self) -> Dict[str, Any]: - """Get the launch option that users need to set for their games - - Returns: - Dict containing the launch option string and instructions - """ - return { - "launch_option": "~/lsfg %command%", - "instructions": "Add this to your game's launch options in Steam Properties", - "explanation": "The lsfg script points games at the upstream configuration; profiles are selected by Steam AppID" - } - async def get_config_file_content(self) -> Dict[str, Any]: """Get the current config file content @@ -221,38 +209,6 @@ class Plugin: "error": f"Error reading config file: {str(e)}" } - async def get_launch_script_content(self) -> Dict[str, Any]: - """Get the content of the launch script file - - Returns: - FileContentResponse dict with file content or error information - """ - try: - script_path = self.installation_service.get_launch_script_path() - - if not os.path.exists(script_path): - return { - "success": False, - "error": f"Launch script not found at {script_path}", - "path": str(script_path) - } - - with open(script_path, 'r') as file: - content = file.read() - - return { - "success": True, - "content": content, - "path": str(script_path) - } - - except Exception as e: - decky.logger.error(f"Error reading launch script: {e}") - return { - "success": False, - "error": str(e) - } - async def check_fgmod_directory(self) -> Dict[str, Any]: """Check if the fgmod directory exists in the home directory diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py index 9b69806..54ac359 100644 --- a/py_modules/lsfg_vk/steam_service.py +++ b/py_modules/lsfg_vk/steam_service.py @@ -39,7 +39,7 @@ class SteamService(BaseService): "1628350", # Steam Linux Runtime 3.0 } - def _steam_library_roots(self): + def _steam_roots(self): candidates = ( self.user_home / ".local/share/Steam", self.user_home / ".steam/steam", @@ -51,6 +51,11 @@ class SteamService(BaseService): for candidate in candidates: yield from self._unique_existing_root(candidate, seen) + def _steam_library_roots(self): + seen = set() + for candidate in self._steam_roots(): + yield from self._unique_existing_root(candidate, seen) + library_file = candidate / "steamapps/libraryfolders.vdf" try: content = library_file.read_text(encoding="utf-8") @@ -62,6 +67,65 @@ class SteamService(BaseService): yield from self._unique_existing_root(Path(path), seen) @staticmethod + def _read_shortcuts(data: bytes) -> Dict[str, object]: + def read_string(offset: int) -> Tuple[str, int]: + end = data.index(b"\0", offset) + return data[offset:end].decode("utf-8", errors="replace"), end + 1 + + def read_object(offset: int = 0) -> Tuple[Dict[str, object], int]: + values = {} + while offset < len(data): + value_type, offset = data[offset], offset + 1 + if value_type == 8: + return values, offset + key, offset = read_string(offset) + if value_type == 0: + value, offset = read_object(offset) + elif value_type == 1: + value, offset = read_string(offset) + elif value_type == 2: + if offset + 4 > len(data): + raise ValueError("truncated binary VDF integer") + value = int.from_bytes(data[offset:offset + 4], "little", signed=True) + offset += 4 + else: + raise ValueError(f"unsupported binary VDF type {value_type}") + values[key] = value + raise ValueError("unterminated binary VDF object") + + values, offset = read_object() + if offset != len(data): + raise ValueError("trailing binary VDF data") + return values + + @staticmethod + def _shortcut_game(shortcut: object) -> Optional[Dict[str, object]]: + if not isinstance(shortcut, dict): + return None + appid = shortcut.get("appid") + name = shortcut.get("AppName") + if not isinstance(appid, int) or appid == 0 or not isinstance(name, str) or not name: + return None + return {"appid": str(appid), "name": name, "nonSteam": True} + + def _shortcut_games(self): + games = {} + for steam_root in self._steam_roots(): + for shortcuts_file in sorted((steam_root / "userdata").glob("*/config/shortcuts.vdf")): + try: + root = self._read_shortcuts(shortcuts_file.read_bytes()) + except (OSError, ValueError): + continue + shortcuts = root.get("shortcuts", {}) + if not isinstance(shortcuts, dict): + continue + for shortcut in shortcuts.values(): + game = self._shortcut_game(shortcut) + if game and game["appid"] not in self.GAME_SELECTOR_EXCLUDED_APPIDS: + games.setdefault(game["appid"], game) + return list(games.values()) + + @staticmethod def _unique_existing_root(path: Path, seen: set[str]): if not path.exists(): return @@ -208,7 +272,7 @@ class SteamService(BaseService): def get_installed_games(self) -> Dict[str, object]: """Return installed Steam app IDs and names for the Game Mode selector.""" try: - games = {} + games: Dict[str, Dict[str, object]] = {} for library_root in self._steam_library_roots(): for manifest in (library_root / "steamapps").glob("appmanifest_*.acf"): match = re.fullmatch(r"appmanifest_(\d+)\.acf", manifest.name) @@ -222,7 +286,12 @@ class SteamService(BaseService): if appid in self.GAME_SELECTOR_EXCLUDED_APPIDS: continue name = self._section_value(content, "AppState", "name") or f"App {appid}" - games[appid] = name - return self._success_response(dict, games=[{"appid": appid, "name": name} for appid, name in sorted(games.items(), key=lambda item: item[1].lower())]) + games[appid] = {"appid": appid, "name": name, "nonSteam": False} + for game in self._shortcut_games(): + games.setdefault(str(game["appid"]), game) + return self._success_response( + dict, + games=sorted(games.values(), key=lambda game: str(game["name"]).lower()), + ) except Exception as error: return self._error_response(dict, str(error), games=[]) |