diff options
Diffstat (limited to 'py_modules')
| -rw-r--r-- | py_modules/lsfg_vk/constants.py | 2 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/plugin.py | 8 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/steam_service.py | 273 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/types.py | 17 |
4 files changed, 300 insertions, 0 deletions
diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py index 1f78a59..19df278 100644 --- a/py_modules/lsfg_vk/constants.py +++ b/py_modules/lsfg_vk/constants.py @@ -18,6 +18,8 @@ UI_ICON_FILENAME = "gay.pancake.lsfg-vk-ui.png" 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" +STEAM_LOSSLESS_SCALING_APP_ID = "993090" +STEAM_LOSSLESS_SCALING_BRANCH = "lsfg-vk" LEGACY_LIB_FILENAME = "liblsfg-vk.so" LEGACY_JSON_FILENAME = "VkLayer_LS_frame_generation.json" diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index 6a8dfc5..1676566 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -16,6 +16,7 @@ from .configuration import ConfigurationService from .config_schema import ConfigurationManager from .flatpak_service import FlatpakService from .runtime_service import RuntimeService +from .steam_service import SteamService class Plugin: @@ -33,6 +34,7 @@ class Plugin: self.installation_service = InstallationService(runtime_service=self.runtime_service) self.configuration_service = ConfigurationService(runtime_service=self.runtime_service) self.flatpak_service = FlatpakService() + self.steam_service = SteamService() async def install_lsfg_vk(self) -> Dict[str, Any]: """Install the bundled lsfg-vk runtime to ~/.local @@ -320,6 +322,12 @@ class Plugin: """ return self.flatpak_service.get_flatpak_apps() + async def get_lossless_scaling_branch_status(self) -> Dict[str, Any]: + return self.steam_service.get_branch_status() + + async def select_lossless_scaling_branch(self) -> Dict[str, Any]: + return self.steam_service.select_branch() + async def set_flatpak_app_override(self, app_id: str) -> Dict[str, Any]: """Set lsfg-vk overrides for a Flatpak app diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py new file mode 100644 index 0000000..3278f3c --- /dev/null +++ b/py_modules/lsfg_vk/steam_service.py @@ -0,0 +1,273 @@ +import os +import re +import tempfile +from pathlib import Path +from typing import Dict, Optional, Tuple + +from .base_service import BaseService +from .constants import STEAM_LOSSLESS_SCALING_APP_ID, STEAM_LOSSLESS_SCALING_BRANCH + + +class SteamService(BaseService): + DEFAULT_BRANCH = "public" + MANIFEST_FILENAME = f"appmanifest_{STEAM_LOSSLESS_SCALING_APP_ID}.acf" + + def _steam_library_roots(self): + candidates = ( + self.user_home / ".local/share/Steam", + self.user_home / ".steam/steam", + self.user_home / ".steam/root", + self.user_home / ".var/app/com.valvesoftware.Steam/.local/share/Steam", + ) + seen = set() + + for candidate in candidates: + yield from self._unique_existing_root(candidate, seen) + + library_file = candidate / "steamapps/libraryfolders.vdf" + try: + content = library_file.read_text(encoding="utf-8") + except OSError: + continue + + for raw_path in re.findall(r'(?m)^\s*"path"\s+"((?:\\.|[^"])*)"', content): + path = raw_path.replace(r'\"', '"').replace(r'\\', '\\') + yield from self._unique_existing_root(Path(path), seen) + + @staticmethod + def _unique_existing_root(path: Path, seen: set[str]): + if not path.exists(): + return + try: + resolved = str(path.resolve()) + except OSError: + resolved = str(path) + if resolved in seen: + return + seen.add(resolved) + yield path + + def _manifest_path(self) -> Optional[Path]: + for library_root in self._steam_library_roots(): + manifest = library_root / "steamapps" / self.MANIFEST_FILENAME + if manifest.is_file(): + return manifest + return None + + @staticmethod + def _section_bounds(content: str, section_name: str) -> Optional[Tuple[int, int, str]]: + section = re.search( + rf'(?m)^(?P<indent>[ \t]*)"{re.escape(section_name)}"[ \t\r\n]*\{{', + content, + ) + if section is None: + return None + + depth = 1 + in_string = False + escaped = False + for index in range(section.end(), len(content)): + character = content[index] + if in_string: + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == '"': + in_string = False + continue + + if character == '"': + in_string = True + elif character == "{": + depth += 1 + elif character == "}": + depth -= 1 + if depth == 0: + return section.end(), index, section.group("indent") + return None + + @classmethod + def _section_value(cls, content: str, section_name: str, key: str) -> Optional[str]: + bounds = cls._section_bounds(content, section_name) + if bounds is None: + return None + body_start, body_end, _ = bounds + pattern = re.compile( + r'(?m)^[ \t]*"(?P<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"' + ) + for match in pattern.finditer(content, body_start, body_end): + if match.group("key") == key: + return match.group("value") + return None + + @classmethod + def _set_section_value(cls, content: str, section_name: str, key: str, value: str) -> str: + bounds = cls._section_bounds(content, section_name) + if bounds is None: + if section_name != "UserConfig": + raise ValueError(f"Steam manifest is missing the {section_name} section") + app_state = cls._section_bounds(content, "AppState") + if app_state is None: + raise ValueError("Steam manifest is missing the AppState section") + _, app_state_end, app_state_indent = app_state + prefix = content[:app_state_end] + if not prefix.endswith(("\n", "\r")): + prefix += "\n" + entry_indent = app_state_indent + "\t" + section = ( + f'{entry_indent}"UserConfig"\n' + f'{entry_indent}{{\n' + f'{entry_indent}\t"{key}"\t"{value}"\n' + f'{entry_indent}}}\n' + ) + return prefix + section + content[app_state_end:] + + body_start, body_end, section_indent = bounds + pattern = re.compile( + rf'(?m)^[ \t]*"{re.escape(key)}"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"' + ) + match = pattern.search(content, body_start, body_end) + if match is not None: + return content[: match.start("value")] + value + content[match.end("value") :] + + prefix = content[:body_end] + if not prefix.endswith(("\n", "\r")): + prefix += "\n" + entry_indent = section_indent + "\t" + return prefix + f'{entry_indent}"{key}"\t"{value}"\n' + content[body_end:] + + @classmethod + def _branch_or_default(cls, branch: Optional[str]) -> str: + return branch or cls.DEFAULT_BRANCH + + def _status_fields(self, manifest_path: Path, content: str) -> Dict[str, object]: + selected_branch = self._branch_or_default( + self._section_value(content, "UserConfig", "BetaKey") + ) + current_branch = self._branch_or_default( + self._section_value(content, "MountedConfig", "BetaKey") + or self._section_value(content, "UserConfig", "BetaKey") + ) + needs_switch = ( + selected_branch != STEAM_LOSSLESS_SCALING_BRANCH + or current_branch != STEAM_LOSSLESS_SCALING_BRANCH + ) + return { + "installed": True, + "manifest_path": str(manifest_path), + "selected_branch": selected_branch, + "current_branch": current_branch, + "target_branch": STEAM_LOSSLESS_SCALING_BRANCH, + "needs_switch": needs_switch, + "restart_required": ( + selected_branch == STEAM_LOSSLESS_SCALING_BRANCH + and current_branch != STEAM_LOSSLESS_SCALING_BRANCH + ), + } + + def get_branch_status(self) -> Dict[str, object]: + try: + manifest_path = self._manifest_path() + if manifest_path is None: + return self._success_response( + dict, + "Lossless Scaling is not installed through Steam", + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) + + content = manifest_path.read_text(encoding="utf-8") + fields = self._status_fields(manifest_path, content) + if not fields["needs_switch"]: + message = "Lossless Scaling is using the lsfg-vk Steam branch" + elif fields["restart_required"]: + message = "lsfg-vk is selected; restart Steam to finish the branch switch" + else: + message = "Lossless Scaling is not using the lsfg-vk Steam branch" + return self._success_response(dict, message, **fields) + except Exception as error: + return self._error_response( + dict, + str(error), + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) + + def select_branch(self) -> Dict[str, object]: + try: + manifest_path = self._manifest_path() + if manifest_path is None: + raise FileNotFoundError("Lossless Scaling is not installed through Steam") + + content = manifest_path.read_text(encoding="utf-8") + fields = self._status_fields(manifest_path, content) + if not fields["needs_switch"]: + return self._success_response( + dict, + "Lossless Scaling is already using the lsfg-vk Steam branch", + changed=False, + **fields, + ) + + updated = self._set_section_value( + content, + "UserConfig", + "BetaKey", + STEAM_LOSSLESS_SCALING_BRANCH, + ) + if updated != content: + file_mode = manifest_path.stat().st_mode & 0o777 + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=manifest_path.parent, + prefix=f".{manifest_path.name}.", + delete=False, + ) as temporary_file: + temporary_path = Path(temporary_file.name) + temporary_file.write(updated) + temporary_file.flush() + os.fsync(temporary_file.fileno()) + temporary_path.chmod(file_mode) + os.replace(temporary_path, manifest_path) + except Exception: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + raise + + new_fields = dict(fields) + new_fields["selected_branch"] = STEAM_LOSSLESS_SCALING_BRANCH + new_fields["needs_switch"] = new_fields["current_branch"] != STEAM_LOSSLESS_SCALING_BRANCH + new_fields["restart_required"] = new_fields["needs_switch"] + return self._success_response( + dict, + "lsfg-vk selected for Lossless Scaling; restart Steam to download it", + changed=updated != content, + **new_fields, + ) + except Exception as error: + return self._error_response( + dict, + str(error), + changed=False, + installed=False, + manifest_path=None, + selected_branch=None, + current_branch=None, + target_branch=STEAM_LOSSLESS_SCALING_BRANCH, + needs_switch=False, + restart_required=False, + ) diff --git a/py_modules/lsfg_vk/types.py b/py_modules/lsfg_vk/types.py index 0f0428b..96ce23b 100644 --- a/py_modules/lsfg_vk/types.py +++ b/py_modules/lsfg_vk/types.py @@ -42,6 +42,23 @@ class InstallationCheckResponse(TypedDict): error: Optional[str] +class SteamBranchStatusResponse(TypedDict): + success: bool + message: str + error: Optional[str] + installed: bool + manifest_path: Optional[str] + selected_branch: Optional[str] + current_branch: Optional[str] + target_branch: str + needs_switch: bool + restart_required: bool + + +class SteamBranchOperationResponse(SteamBranchStatusResponse): + changed: bool + + class ConfigurationResponse(BaseResponse): """Response for configuration operations""" config: Optional[ConfigurationData] |