summaryrefslogtreecommitdiff
path: root/py_modules/lsfg_vk
diff options
context:
space:
mode:
Diffstat (limited to 'py_modules/lsfg_vk')
-rw-r--r--py_modules/lsfg_vk/config_schema.py73
-rw-r--r--py_modules/lsfg_vk/flatpak_service.py460
-rw-r--r--py_modules/lsfg_vk/installation.py79
-rw-r--r--py_modules/lsfg_vk/plugin.py164
-rw-r--r--py_modules/lsfg_vk/steam_service.py285
-rw-r--r--py_modules/lsfg_vk/types.py29
6 files changed, 355 insertions, 735 deletions
diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py
index ce109f3..3816d88 100644
--- a/py_modules/lsfg_vk/config_schema.py
+++ b/py_modules/lsfg_vk/config_schema.py
@@ -1,13 +1,9 @@
"""Small adapter for the upstream lsfg-vk v2 configuration format."""
import json
-import sys
import tomllib
-from pathlib import Path
from typing import Any, Dict, TypedDict
-sys.path.insert(0, str(Path(__file__).parent.parent.parent))
-
ConfigurationData = Dict[str, Any]
@@ -57,8 +53,8 @@ class ConfigurationManager:
def validate_config(config: Dict[str, Any]) -> Dict[str, Any]:
result = {**PROFILE_DEFAULTS, **GLOBAL_DEFAULTS}
result.update({key: value for key, value in config.items() if key in result})
- result["active_in"] = _normalize_active_in(result.get("active_in"))
- result["pacing_mode"] = str(result.get("pacing_mode", "vsync")).lower()
+ result["active_in"] = _normalize_active_in(result["active_in"])
+ result["pacing_mode"] = str(result["pacing_mode"]).lower()
if result["pacing_mode"] != "vsync":
raise ValueError("pacing_mode must be vsync")
result["multiplier"] = int(result["multiplier"])
@@ -67,43 +63,24 @@ class ConfigurationManager:
result["flow_scale"] = float(result["flow_scale"])
if not 0.25 <= result["flow_scale"] <= 1.0:
raise ValueError("flow_scale must be between 0.25 and 1.0")
- for name in ("no_fp16", "performance_mode", "override_present_mode", "preserve_swapchain_image_count"):
+ for name in (
+ "no_fp16",
+ "performance_mode",
+ "override_present_mode",
+ "preserve_swapchain_image_count",
+ ):
result[name] = bool(result[name])
- result["dll"] = str(result.get("dll") or "")
+ result["dll"] = str(result["dll"] or "")
return result
@staticmethod
- def _migrate_dll_path(value: Any) -> str:
- path_value = str(value or "")
- if not path_value:
- return ""
- path = Path(path_value)
- if path.name.lower() in {"lossless.dll"}:
- return str(path.with_name("lsfg-vk.dll"))
- return path_value
-
- @staticmethod
- def _config_from_profile(profile: Dict[str, Any], global_config: Dict[str, Any]) -> Dict[str, Any]:
- raw = dict(profile)
- if "pacing_mode" not in raw and "pacing" in raw:
- raw["pacing_mode"] = raw["pacing"]
- if "override_present_mode" not in raw and "experimental_present_mode" in raw:
- raw["override_present_mode"] = raw["experimental_present_mode"] == "fifo"
- raw["dll"] = global_config.get("dll", "")
- raw["no_fp16"] = global_config.get("no_fp16", False)
- return ConfigurationManager.validate_config(raw)
-
- @staticmethod
def generate_toml_content_multi_profile(profile_data: ProfileData) -> str:
global_config = {**GLOBAL_DEFAULTS, **profile_data.get("global_config", {})}
lines = ["version = 2", "", "[global]"]
- dll = ConfigurationManager._migrate_dll_path(global_config.get("dll"))
- if dll:
- lines.append(f"dll = {_toml_value(dll)}")
- lines.append(f"allow_fp16 = {_toml_value(not bool(global_config.get('no_fp16', False)))}")
- profiles = sorted(profile_data["profiles"].items())
- if not profiles:
- profiles = [("", {})]
+ if global_config["dll"]:
+ lines.append(f"dll = {_toml_value(global_config['dll'])}")
+ lines.append(f"allow_fp16 = {_toml_value(not bool(global_config['no_fp16']))}")
+ profiles = sorted(profile_data["profiles"].items()) or [("", {})]
for name, raw in profiles:
config = ConfigurationManager.validate_config({**raw, **global_config})
lines.extend(["", "[[profile]]", f"name = {_toml_value(name)}"])
@@ -122,26 +99,20 @@ class ConfigurationManager:
@staticmethod
def parse_toml_content_multi_profile(content: str) -> ProfileData:
data = tomllib.loads(content)
- version = data.get("version")
- if version not in (1, 2):
+ if data.get("version") != 2:
raise ValueError("unsupported lsfg-vk configuration version")
- raw_global = dict(data.get("global", {}))
+ raw_global = data.get("global", {})
global_config = {
- "dll": ConfigurationManager._migrate_dll_path(raw_global.get("dll", "")),
+ "dll": str(raw_global.get("dll", "") or ""),
"no_fp16": not bool(raw_global.get("allow_fp16", True)),
}
profiles: Dict[str, Dict[str, Any]] = {}
- source_profiles = data.get("game", []) if version == 1 else data.get("profile", [])
- for profile in source_profiles:
- name = str(profile.get("exe" if version == 1 else "name", ""))
- config = ConfigurationManager._config_from_profile(profile, global_config)
+ for profile in data.get("profile", []):
+ name = str(profile.get("name", ""))
+ config = ConfigurationManager.validate_config({
+ **profile,
+ **global_config,
+ })
if config["active_in"]:
profiles[name] = config
return {"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
diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py
index 11f1a82..d2241ab 100644
--- a/py_modules/lsfg_vk/flatpak_service.py
+++ b/py_modules/lsfg_vk/flatpak_service.py
@@ -1,4 +1,4 @@
-"""Flatpak runtime-extension infrastructure for unified game targets."""
+"""Flatpak runtime support for classified Steam targets."""
from __future__ import annotations
@@ -10,7 +10,7 @@ import shutil
import subprocess
import threading
from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Tuple
+from typing import Dict, Optional, Set
from .base_service import BaseService
from .constants import (
@@ -22,14 +22,6 @@ from .constants import (
class FlatpakService(BaseService):
- """Resolve and provision only the runtime support a target actually needs.
-
- Flatpak application permissions are deliberately not persisted here. The
- generated per-AppID wrapper supplies the narrow launch-time permissions and
- environment instead, while this service owns only the shared Vulkan layer
- runtime extensions installed from the plugin bundle.
- """
-
EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk"
SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08")
OWNERSHIP_FILENAME = "flatpak_extensions.json"
@@ -37,7 +29,6 @@ class FlatpakService(BaseService):
APP_ID_PATTERN = re.compile(
r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$"
)
- BRANCH_PATTERN = re.compile(r"^[0-9]+\.[0-9]+$")
def __init__(self, logger=None):
super().__init__(logger)
@@ -48,39 +39,37 @@ class FlatpakService(BaseService):
def ownership_path(self) -> Path:
return self.config_dir / self.OWNERSHIP_FILENAME
- def _get_clean_env(self) -> Dict[str, str]:
+ def _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]
+ path = [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:
- path_entries.insert(0, entry)
- env["PATH"] = ":".join(path_entries)
+ if entry not in path:
+ path.insert(0, entry)
+ env["PATH"] = ":".join(path)
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()
+ env = self._clean_env()
self.flatpak_command = shutil.which("flatpak", path=env["PATH"])
return self.flatpak_command is not None
- def _run_flatpak_command(self, args: List[str], **kwargs):
+ def _run_flatpak_command(self, args, **kwargs):
if self.flatpak_command is None and not self.check_flatpak_available():
raise FileNotFoundError("Flatpak command not available")
+ env = self._clean_env()
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"])
+ try:
+ user = 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
+ if os.geteuid() != user.pw_uid:
+ runuser = shutil.which("runuser", path=env["PATH"])
if runuser is None:
raise FileNotFoundError("runuser command not available")
- command = [runuser, "--user", target_user.pw_name, "--", *command]
- return subprocess.run(command, env=self._get_clean_env(), **kwargs)
+ command = [runuser, "--user", user.pw_name, "--", *command]
+ return subprocess.run(command, env=env, **kwargs)
@classmethod
def _validate_app_id(cls, app_id: str) -> str:
@@ -89,45 +78,32 @@ class FlatpakService(BaseService):
return app_id
@classmethod
- def _validate_runtime(cls, version: str) -> str:
- if version not in cls.SUPPORTED_RUNTIMES:
+ def _validate_runtime(cls, branch: str) -> str:
+ if branch not in cls.SUPPORTED_RUNTIMES:
raise ValueError(
- f"Unsupported Flatpak runtime branch {version}; "
- f"supported branches are {', '.join(cls.SUPPORTED_RUNTIMES)}"
+ f"Unsupported Flatpak runtime branch {branch}; supported branches are "
+ + ", ".join(cls.SUPPORTED_RUNTIMES)
)
- return version
-
- @classmethod
- def _extension_ref(cls, version: str) -> str:
- return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(version)}"
+ return branch
@classmethod
def runtime_branch_from_ref(cls, runtime_ref: str) -> str:
- """Return the supported Freedesktop branch from a runtime ref."""
- if not isinstance(runtime_ref, str):
- raise ValueError("Flatpak did not return a runtime reference")
- parts = runtime_ref.strip().split("/")
+ parts = runtime_ref.strip().split("/") if isinstance(runtime_ref, str) else []
if len(parts) != 3 or parts[0] != "org.freedesktop.Platform":
raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}")
- branch = parts[2]
- if not cls.BRANCH_PATTERN.fullmatch(branch):
- raise ValueError(f"Unrecognized Flatpak runtime branch: {branch}")
- return cls._validate_runtime(branch)
+ return cls._validate_runtime(parts[2])
@classmethod
- def _bundle_filename(cls, version: str) -> str:
- return {
+ def _extension_ref(cls, branch: str) -> str:
+ return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}"
+
+ def _bundled_extension_path(self, branch: str) -> Path:
+ filename = {
"23.08": FLATPAK_23_08_FILENAME,
"24.08": FLATPAK_24_08_FILENAME,
"25.08": FLATPAK_25_08_FILENAME,
- }[cls._validate_runtime(version)]
-
- def _bundled_extension_path(self, version: str) -> Path:
- return (
- Path(__file__).resolve().parent.parent.parent
- / BIN_DIR
- / self._bundle_filename(version)
- )
+ }[self._validate_runtime(branch)]
+ return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename
def _installed_extension_branches(self) -> Set[str]:
result = self._run_flatpak_command(
@@ -136,78 +112,54 @@ class FlatpakService(BaseService):
text=True,
check=True,
)
- installed: Set[str] = set()
+ installed = set()
for line in result.stdout.splitlines():
- if not line.strip():
- continue
- fields = line.split("\t")
- if len(fields) < 3:
- fields = line.split()
- if len(fields) < 3:
- continue
- application, arch, branch = (field.strip() for field in fields[:3])
- if application == self.EXTENSION_ID and arch == "x86_64":
- installed.add(branch)
+ fields = line.split("\t") if "\t" in line else line.split()
+ if len(fields) >= 3 and fields[0] == self.EXTENSION_ID and fields[1] == "x86_64":
+ installed.add(fields[2])
return installed
- def _read_owned_branches(self) -> Tuple[Set[str], bool]:
- """Read ownership without guessing when metadata is damaged."""
+ def _owned_branches(self) -> Set[str]:
path = self.ownership_path
- if path.is_symlink():
- self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}")
- return set(), True
- if not path.exists():
- return set(), False
- if not path.is_file():
- self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}")
- return set(), True
+ if not path.exists() and not path.is_symlink():
+ return set()
+ if path.is_symlink() or not path.is_file():
+ raise RuntimeError("Flatpak ownership metadata is not a regular file")
try:
- raw = json.loads(path.read_text(encoding="utf-8"))
- if not isinstance(raw, dict) or raw.get("version") != self.OWNERSHIP_VERSION:
- raise ValueError("unsupported ownership metadata version")
- branches = raw.get("plugin_owned_branches")
- if not isinstance(branches, list):
- raise ValueError("plugin_owned_branches is not a list")
- normalized = {
- self._validate_runtime(branch)
- for branch in branches
- if isinstance(branch, str)
- }
- if len(normalized) != len(branches):
- raise ValueError("ownership metadata contains invalid branches")
- return normalized, False
+ data = json.loads(path.read_text(encoding="utf-8"))
+ branches = data.get("plugin_owned_branches")
+ if data.get("version") != self.OWNERSHIP_VERSION or not isinstance(branches, list):
+ raise ValueError("invalid ownership metadata")
+ owned = {self._validate_runtime(branch) for branch in branches}
+ if len(owned) != len(branches):
+ raise ValueError("invalid ownership metadata")
+ return owned
except (OSError, json.JSONDecodeError, TypeError, ValueError) as error:
- self.log.warning(f"Could not trust Flatpak ownership metadata: {error}")
- return set(), True
+ raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error
def _write_owned_branches(self, branches: Set[str]) -> None:
if not branches:
- if self.ownership_path.exists() or self.ownership_path.is_symlink():
- self.ownership_path.unlink()
+ self.ownership_path.unlink(missing_ok=True)
return
- document = {
- "version": self.OWNERSHIP_VERSION,
- "plugin_owned_branches": sorted(branches),
- }
- self._write_file(self.ownership_path, json.dumps(document, indent=2) + "\n")
+ self._write_file(
+ self.ownership_path,
+ json.dumps(
+ {
+ "version": self.OWNERSHIP_VERSION,
+ "plugin_owned_branches": sorted(branches),
+ },
+ indent=2,
+ ) + "\n",
+ )
- def get_extension_status(self) -> Dict[str, Any]:
- """Return global extension inventory for Setup diagnostics."""
+ def get_extension_status(self):
try:
- if not self.check_flatpak_available():
- return self._success_response(
- dict,
- "Flatpak is not available",
- available=False,
- extension_id=self.EXTENSION_ID,
- supported_branches=list(self.SUPPORTED_RUNTIMES),
- installed_branches=[],
- )
- installed = self._installed_extension_branches()
+ available = self.check_flatpak_available()
+ installed = self._installed_extension_branches() if available else set()
return self._success_response(
dict,
- "Flatpak runtime extension status retrieved",
- available=True,
+ "Flatpak runtime extension status retrieved" if available else "Flatpak is not available",
+ available=available,
extension_id=self.EXTENSION_ID,
supported_branches=list(self.SUPPORTED_RUNTIMES),
installed_branches=sorted(installed),
@@ -216,16 +168,15 @@ class FlatpakService(BaseService):
return self._error_response(
dict,
str(error),
- available=self.check_flatpak_available(),
+ available=False,
extension_id=self.EXTENSION_ID,
supported_branches=list(self.SUPPORTED_RUNTIMES),
installed_branches=[],
)
- def get_flatpak_support_status(self) -> Dict[str, Any]:
- return self.get_extension_status()
+ get_flatpak_support_status = get_extension_status
- def _resolve_runtime(self, app_id: str) -> Dict[str, Any]:
+ def _resolve_runtime(self, app_id: str):
self._validate_app_id(app_id)
if not self.check_flatpak_available():
raise FileNotFoundError("Flatpak is not available on this system")
@@ -236,27 +187,21 @@ class FlatpakService(BaseService):
)
if result.returncode != 0:
raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}")
- runtime_ref = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
- branch = self.runtime_branch_from_ref(runtime_ref)
- return {"runtime": runtime_ref, "runtime_branch": branch}
+ runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
+ return runtime, self.runtime_branch_from_ref(runtime)
- def resolve_app_support(self, app_id: str) -> Dict[str, Any]:
- """Resolve the exact runtime branch required by one Flatpak app."""
+ def resolve_app_support(self, app_id: str):
try:
app_id = self._validate_app_id(app_id)
- resolved = self._resolve_runtime(app_id)
+ runtime, branch = self._resolve_runtime(app_id)
installed = self._installed_extension_branches()
- branch = resolved["runtime_branch"]
ready = branch in installed
return self._success_response(
dict,
- (
- f"lsfg-vk support is ready for {app_id}"
- if ready
- else f"lsfg-vk runtime extension {branch} is required for {app_id}"
- ),
+ f"lsfg-vk support is ready for {app_id}" if ready
+ else f"lsfg-vk runtime extension {branch} is required for {app_id}",
flatpak_app_id=app_id,
- runtime=resolved["runtime"],
+ runtime=runtime,
runtime_branch=branch,
support_status="ready" if ready else "needs-runtime",
extension_installed=ready,
@@ -286,194 +231,114 @@ class FlatpakService(BaseService):
installed_branches=[],
)
- def install_extension(self, version: str) -> Dict[str, Any]:
- """Install one branch, treating an already-installed branch as success."""
+ def install_extension(self, branch: str):
try:
- version = self._validate_runtime(version)
+ branch = self._validate_runtime(branch)
if not self.check_flatpak_available():
raise FileNotFoundError("Flatpak is not available on this system")
with self._lock:
- installed_before = self._installed_extension_branches()
- if version in installed_before:
- return self._success_response(
- dict,
- f"lsfg-vk {version} runtime extension is already installed",
- runtime_branch=version,
- installed=True,
- enabled=True,
- )
- bundle_path = self._bundled_extension_path(version)
- if not bundle_path.is_file():
- raise FileNotFoundError(
- f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin"
- )
+ if branch in self._installed_extension_branches():
+ return self._extension_result(branch, True, False, "already installed")
+ bundle = self._bundled_extension_path(branch)
+ if not bundle.is_file():
+ raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin")
result = self._run_flatpak_command(
- [
- "install",
- "--user",
- "--noninteractive",
- "--or-update",
- str(bundle_path),
- ],
+ ["install", "--user", "--noninteractive", "--or-update", str(bundle)],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise OSError(result.stderr.strip() or "Flatpak installation failed")
- installed_after = self._installed_extension_branches()
- if version not in installed_after:
- raise RuntimeError(
- f"Flatpak install completed but {self._extension_ref(version)} "
- "was not visible afterwards"
- )
- owned, uncertain = self._read_owned_branches()
- if not uncertain:
- owned.add(version)
+ if branch not in self._installed_extension_branches():
+ raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards")
+ owned = self._owned_branches()
+ owned.add(branch)
+ self._write_owned_branches(owned)
+ return self._extension_result(branch, True, False, "installed")
+ except Exception as error:
+ return self._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False)
+
+ def _remove_extension(self, branch: str) -> bool:
+ if branch not in self._installed_extension_branches():
+ return False
+ result = self._run_flatpak_command(
+ ["uninstall", "--user", "--noninteractive", self._extension_ref(branch)],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise OSError(result.stderr.strip() or "Flatpak uninstall failed")
+ if branch in self._installed_extension_branches():
+ raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed")
+ return True
+
+ def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str):
+ return self._success_response(
+ dict,
+ f"lsfg-vk {branch} runtime extension {verb}",
+ runtime_branch=branch,
+ installed=installed,
+ enabled=installed,
+ removed=removed,
+ )
+
+ def uninstall_extension(self, branch: str):
+ try:
+ branch = self._validate_runtime(branch)
+ if not self.check_flatpak_available():
+ raise FileNotFoundError("Flatpak is not available on this system")
+ with self._lock:
+ removed = self._remove_extension(branch)
+ owned = self._owned_branches()
+ if branch in owned:
+ owned.remove(branch)
self._write_owned_branches(owned)
- return self._success_response(
- dict,
- f"lsfg-vk {version} runtime extension installed",
- runtime_branch=version,
- installed=True,
- enabled=True,
- )
+ return self._extension_result(branch, False, removed, "uninstalled")
except Exception as error:
return self._error_response(
dict,
str(error),
- runtime_branch=version,
+ runtime_branch=branch,
+ removed=False,
installed=False,
enabled=False,
)
- def ensure_extension(self, version: str) -> Dict[str, Any]:
- status = self.get_extension_status()
- if not status.get("success"):
- return status
- if not status.get("available"):
- return self._error_response(
- dict,
- "Flatpak is not available on this system",
- runtime_branch=version,
- support_status="error",
- )
+ def ensure_extension(self, branch: str):
try:
- version = self._validate_runtime(version)
- except ValueError as error:
- return self._error_response(dict, str(error), runtime_branch=version, support_status="unsupported")
- if version in status.get("installed_branches", []):
- return self._success_response(
- dict,
- f"lsfg-vk {version} runtime extension is ready",
- runtime_branch=version,
- installed=True,
- )
- return self.install_extension(version)
+ branch = self._validate_runtime(branch)
+ if branch in self._installed_extension_branches():
+ return self._extension_result(branch, True, False, "is ready")
+ except Exception as error:
+ return self._error_response(dict, str(error), runtime_branch=branch, support_status="error")
+ return self.install_extension(branch)
- def ensure_app_support(self, app_id: str) -> Dict[str, Any]:
- """Provision only the branch returned by flatpak info for this app."""
+ def ensure_app_support(self, app_id: str):
resolved = self.resolve_app_support(app_id)
if not resolved.get("success") or resolved.get("support_status") != "needs-runtime":
return resolved
- branch = resolved.get("runtime_branch")
- result = self.ensure_extension(branch)
+ result = self.ensure_extension(resolved["runtime_branch"])
if not result.get("success"):
return self._error_response(
dict,
result.get("error") or "Could not install the required Flatpak runtime extension",
flatpak_app_id=app_id,
runtime=resolved.get("runtime"),
- runtime_branch=branch,
+ runtime_branch=resolved.get("runtime_branch"),
support_status="error",
extension_installed=False,
)
- final = self.resolve_app_support(app_id)
- if final.get("success") and final.get("support_status") == "ready":
- return final
- return self._error_response(
- dict,
- final.get("error") or "Required Flatpak runtime extension could not be verified",
- flatpak_app_id=app_id,
- runtime=resolved.get("runtime"),
- runtime_branch=branch,
- support_status="error",
- extension_installed=False,
- )
-
- def uninstall_extension(self, version: str) -> Dict[str, Any]:
- """Uninstall one branch, treating an already-absent branch as success."""
- try:
- version = self._validate_runtime(version)
- if not self.check_flatpak_available():
- raise FileNotFoundError("Flatpak is not available on this system")
- with self._lock:
- installed = self._installed_extension_branches()
- was_installed = version in installed
- if was_installed:
- result = self._run_flatpak_command(
- [
- "uninstall",
- "--user",
- "--noninteractive",
- self._extension_ref(version),
- ],
- capture_output=True,
- text=True,
- )
- if result.returncode != 0:
- raise OSError(result.stderr.strip() or "Flatpak uninstall failed")
- if version in self._installed_extension_branches():
- raise RuntimeError(
- f"Flatpak uninstall completed but {self._extension_ref(version)} "
- "is still installed"
- )
- owned, uncertain = self._read_owned_branches()
- if not uncertain and version in owned:
- owned.remove(version)
- self._write_owned_branches(owned)
- return self._success_response(
- dict,
- f"lsfg-vk {version} runtime extension uninstalled",
- runtime_branch=version,
- removed=was_installed,
- installed=False,
- enabled=False,
- )
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- runtime_branch=version,
- removed=False,
- installed=False,
- enabled=False,
- )
+ return self.resolve_app_support(app_id)
- def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]:
- """Set one runtime branch to the requested state, safely and idempotently."""
+ def set_extension_enabled(self, branch: str, enabled: bool):
if type(enabled) is not bool:
- return self._error_response(
- dict,
- "enabled must be a boolean",
- runtime_branch=version,
- installed=False,
- enabled=False,
- )
- return self.install_extension(version) if enabled else self.uninstall_extension(version)
+ return self._error_response(dict, "enabled must be a boolean", runtime_branch=branch, installed=False, enabled=False)
+ return self.install_extension(branch) if enabled else self.uninstall_extension(branch)
- def remove_plugin_owned_extensions(self) -> Dict[str, Any]:
- """Uninstall only branches recorded as installed by this plugin."""
+ def remove_plugin_owned_extensions(self):
try:
with self._lock:
- owned, uncertain = self._read_owned_branches()
- if uncertain:
- return self._error_response(
- dict,
- "Flatpak ownership metadata is uncertain; no extensions were removed",
- removed_branches=[],
- preserved_branches=[],
- ownership_uncertain=True,
- )
+ owned = self._owned_branches()
if not owned:
return self._success_response(
dict,
@@ -483,36 +348,11 @@ class FlatpakService(BaseService):
ownership_uncertain=False,
)
if not self.check_flatpak_available():
- return self._error_response(
- dict,
- "Flatpak is not available; plugin-owned extension metadata was preserved",
- removed_branches=[],
- preserved_branches=sorted(owned),
- ownership_uncertain=False,
- )
- removed: List[str] = []
- failures: List[str] = []
+ raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved")
+ removed, failures = [], []
for branch in sorted(owned):
try:
- installed = self._installed_extension_branches()
- if branch in installed:
- result = self._run_flatpak_command(
- [
- "uninstall",
- "--user",
- "--noninteractive",
- self._extension_ref(branch),
- ],
- capture_output=True,
- text=True,
- )
- if result.returncode != 0:
- raise OSError(result.stderr.strip() or "Flatpak uninstall failed")
- if branch in self._installed_extension_branches():
- raise RuntimeError(
- f"Flatpak uninstall completed but {self._extension_ref(branch)} "
- "is still installed"
- )
+ self._remove_extension(branch)
removed.append(branch)
except Exception as error:
failures.append(f"{branch}: {error}")
@@ -539,5 +379,5 @@ class FlatpakService(BaseService):
str(error),
removed_branches=[],
preserved_branches=[],
- ownership_uncertain=False,
+ ownership_uncertain=True,
)
diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py
index 8a3094d..5a583f5 100644
--- a/py_modules/lsfg_vk/installation.py
+++ b/py_modules/lsfg_vk/installation.py
@@ -48,26 +48,20 @@ class InstallationService(BaseService):
def install(self) -> InstallationResponse:
try:
- plugin_dir = Path(__file__).parent.parent.parent
- archive_path = plugin_dir / BIN_DIR / ARCHIVE_FILENAME
+ archive_path = Path(__file__).parent.parent.parent / BIN_DIR / ARCHIVE_FILENAME
if not archive_path.exists():
raise FileNotFoundError(f"{ARCHIVE_FILENAME} not found at {archive_path}")
-
self._ensure_directories()
profile_data = self._prepare_config()
self._install_archive(archive_path)
- config_content = ConfigurationManager.generate_toml_content_multi_profile(profile_data)
- self.runtime_service.validate_config_content(config_content)
- self._write_file(
- self.config_file_path,
- config_content,
- 0o644,
- )
+ content = ConfigurationManager.generate_toml_content_multi_profile(profile_data)
+ self.runtime_service.validate_config_content(content)
+ self._write_file(self.config_file_path, content, 0o644)
self._remove_legacy_layer_files()
return self._success_response(InstallationResponse, "lsfg-vk 2.0.0 installed successfully")
except Exception as error:
self.log.error(f"Error installing lsfg-vk: {error}")
- return self._error_response(InstallationResponse, str(error), message="")
+ return self._error_response(InstallationResponse, str(error))
def _payload_destinations(self) -> Dict[str, tuple[Path, int]]:
return {
@@ -82,7 +76,7 @@ class InstallationService(BaseService):
0o644,
),
f"share/icons/hicolor/256x256/apps/{UI_ICON_FILENAME}": (
- self.user_home / LOCAL_SHARE / "icons" / "hicolor" / "256x256" / "apps" / UI_ICON_FILENAME,
+ self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps" / UI_ICON_FILENAME,
0o644,
),
}
@@ -124,35 +118,26 @@ class InstallationService(BaseService):
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
raise
-
missing = sorted(set(destinations) - found)
if missing:
raise OSError("Archive is missing required files: " + ", ".join(missing))
def _prepare_config(self) -> ProfileData:
if self.config_file_path.exists():
- content = self.config_file_path.read_text(encoding="utf-8")
- legacy = ConfigurationManager.is_legacy_v1(content)
- profile_data = ConfigurationManager.parse_toml_content_multi_profile(content)
- if legacy:
- backup_path = self.config_file_path.with_name(f"{self.config_file_path.name}.v1.bak")
- if not backup_path.exists():
- self._write_file(backup_path, content, 0o644)
+ profile_data = ConfigurationManager.parse_toml_content_multi_profile(
+ self.config_file_path.read_text(encoding="utf-8")
+ )
else:
- default = dict(ConfigurationManager.get_defaults())
+ defaults = ConfigurationManager.get_defaults()
profile_data = ProfileData(
profiles={},
- global_config={
- "dll": default.get("dll", ""),
- "no_fp16": default.get("no_fp16", False),
- },
+ global_config={"dll": defaults["dll"], "no_fp16": defaults["no_fp16"]},
)
-
self._resolve_dll_path(profile_data)
- defaults = dict(ConfigurationManager.get_defaults())
- for profile_name, raw_profile in list(profile_data["profiles"].items()):
- profile_data["profiles"][profile_name] = ConfigurationManager.validate_config(
- {**defaults, **raw_profile, **profile_data["global_config"]}
+ defaults = ConfigurationManager.get_defaults()
+ for name, profile in profile_data["profiles"].items():
+ profile_data["profiles"][name] = ConfigurationManager.validate_config(
+ {**defaults, **profile, **profile_data["global_config"]}
)
return profile_data
@@ -160,7 +145,6 @@ class InstallationService(BaseService):
current_path = str(profile_data["global_config"].get("dll") or "")
if current_path and Path(current_path).is_file():
return False
-
dll_path = self.steam_service.find_lsfg_vk_dll()
if dll_path and current_path != dll_path:
profile_data["global_config"]["dll"] = dll_path
@@ -179,7 +163,6 @@ class InstallationService(BaseService):
except Exception as error:
installed = False
installation_error = str(error)
-
lossless_scaling = self.runtime_service.check_lossless_scaling()
return {
"installed": installed,
@@ -197,21 +180,22 @@ class InstallationService(BaseService):
def uninstall(self) -> UninstallationResponse:
try:
- removed = []
- for path in (
- self.lib_file,
- self.lib_x86_file,
- self.json_file,
- self.json_x86_file,
- self.cli_file,
- self.local_bin_dir / UI_FILENAME,
- self.user_home / LOCAL_SHARE / "applications" / UI_DESKTOP_FILENAME,
- self.user_home / LOCAL_SHARE / "icons" / "hicolor" / "256x256" / "apps" / UI_ICON_FILENAME,
- self.legacy_lib_file,
- self.legacy_json_file,
- ):
- if self._remove_if_exists(path):
- removed.append(str(path))
+ removed = [
+ str(path)
+ for path in (
+ self.lib_file,
+ self.lib_x86_file,
+ self.json_file,
+ self.json_x86_file,
+ self.cli_file,
+ self.local_bin_dir / UI_FILENAME,
+ self.user_home / LOCAL_SHARE / "applications" / UI_DESKTOP_FILENAME,
+ self.user_home / LOCAL_SHARE / "icons/hicolor/256x256/apps" / UI_ICON_FILENAME,
+ self.legacy_lib_file,
+ self.legacy_json_file,
+ )
+ if self._remove_if_exists(path)
+ ]
if not removed:
return self._success_response(
UninstallationResponse,
@@ -227,7 +211,6 @@ class InstallationService(BaseService):
return self._error_response(
UninstallationResponse,
str(error),
- message="",
removed_files=None,
)
diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py
index c576cc2..071b19b 100644
--- a/py_modules/lsfg_vk/plugin.py
+++ b/py_modules/lsfg_vk/plugin.py
@@ -1,34 +1,18 @@
-"""
-Main plugin class for the lsfg-vk Decky Loader plugin.
-
-This plugin provides services for installing and managing the lsfg-vk
-Vulkan layer for frame generation on Steam Deck.
-"""
-
import os
from typing import Any, Dict, Optional
import decky
-from .installation import InstallationService
from .configuration import ConfigurationService
from .flatpak_service import FlatpakService
+from .installation import InstallationService
from .runtime_service import RuntimeService
from .steam_service import SteamService
from .wrapper_service import WrapperService
class Plugin:
- """
- Main plugin class for lsfg-vk management.
-
- This class provides a unified interface for installation, configuration,
- and Flatpak management services. It implements the Decky Loader plugin lifecycle
- methods (_main, _unload, _uninstall, _migration).
- """
-
def __init__(self):
- """Initialize the plugin with all necessary services"""
self.runtime_service = RuntimeService()
self.steam_service = SteamService()
self.installation_service = InstallationService(
@@ -39,63 +23,45 @@ class Plugin:
self.flatpak_service = FlatpakService()
self.wrapper_service = WrapperService()
- async def install_lsfg_vk(self) -> Dict[str, Any]:
- """Install the bundled lsfg-vk runtime to ~/.local
-
- Returns:
- InstallationResponse dict with success status and message/error
- """
+ async def install_lsfg_vk(self):
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
- """
+ async def check_lsfg_vk_installed(self):
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
- """
+ async def uninstall_lsfg_vk(self):
return self.installation_service.uninstall()
- async def get_game_configs(self) -> Dict[str, Any]:
+ async def get_game_configs(self):
return self.configuration_service.get_game_configs()
- async def get_installed_games(self) -> Dict[str, Any]:
+ async def get_installed_games(self):
result = self.steam_service.get_installed_games()
if not result.get("success"):
return result
-
- support_cache: Dict[str, Dict[str, Any]] = {}
+ cache: Dict[str, Dict[str, Any]] = {}
for game in result.get("games", []):
- transport = game.get("transport") if isinstance(game, dict) else None
- if not isinstance(transport, dict) or transport.get("kind") != "flatpak":
+ transport = game.get("transport", {})
+ if transport.get("kind") != "flatpak":
continue
- flatpak_app_id = transport.get("flatpakAppId")
- if not isinstance(flatpak_app_id, str) or not flatpak_app_id:
+ app_id = transport.get("flatpakAppId")
+ if not app_id:
continue
- if flatpak_app_id not in support_cache:
- support_cache[flatpak_app_id] = self.flatpak_service.resolve_app_support(
- flatpak_app_id
- )
- game["flatpakSupport"] = support_cache[flatpak_app_id]
+ if app_id not in cache:
+ cache[app_id] = self.flatpak_service.resolve_app_support(app_id)
+ game["flatpakSupport"] = cache[app_id]
return result
- async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]:
+ async def update_game_config(self, appid: str, game_name: str, config: 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]:
+ async def reset_game_config(self, appid: str):
return self.configuration_service.reset_game_config(appid)
- async def reset_all_game_configs(self) -> Dict[str, Any]:
+ async def reset_all_game_configs(self):
return self.configuration_service.reset_all_game_configs()
- async def get_workaround_state(self, appid: str) -> Dict[str, Any]:
+ async def get_workaround_state(self, appid: str):
return self.wrapper_service.get(appid)
async def set_workaround_state(
@@ -105,7 +71,7 @@ class Plugin:
shortcut_exe: Optional[str] = None,
command_token_added: bool = False,
transport: Optional[Dict[str, Any]] = None,
- ) -> Dict[str, Any]:
+ ):
return self.wrapper_service.set(
appid,
state,
@@ -114,118 +80,82 @@ class Plugin:
transport,
)
- async def remove_workaround_state(self, appid: str) -> Dict[str, Any]:
+ async def remove_workaround_state(self, appid: str):
return self.wrapper_service.remove(appid)
- async def get_config_file_content(self) -> Dict[str, Any]:
- """Get the current config file content
-
- Returns:
- Dict containing the config file content or error message
- """
+ async def get_config_file_content(self):
+ path = self.configuration_service.config_file_path
try:
- config_path = self.configuration_service.config_file_path
- if not config_path.exists():
+ if not path.exists():
return {
"success": False,
"content": None,
- "path": str(config_path),
- "error": "Config file does not exist"
+ "path": str(path),
+ "error": "Config file does not exist",
}
-
- content = config_path.read_text(encoding='utf-8')
return {
"success": True,
- "content": content,
- "path": str(config_path),
- "error": None
+ "content": path.read_text(encoding="utf-8"),
+ "path": str(path),
+ "error": None,
}
- except Exception as e:
+ except Exception as error:
return {
"success": False,
"content": None,
- "path": str(config_path) if 'config_path' in locals() else "unknown",
- "error": f"Error reading config file: {str(e)}"
+ "path": str(path),
+ "error": f"Error reading config file: {error}",
}
- async def get_lossless_scaling_branch_status(self) -> Dict[str, Any]:
+ async def get_lossless_scaling_branch_status(self):
return self.steam_service.get_branch_status()
- async def get_flatpak_support_status(self) -> Dict[str, Any]:
+ async def get_flatpak_support_status(self):
return self.flatpak_service.get_flatpak_support_status()
- async def ensure_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]:
+ async def ensure_flatpak_support(self, flatpak_app_id: str):
return self.flatpak_service.ensure_app_support(flatpak_app_id)
- async def repair_flatpak_support(self, flatpak_app_id: str) -> Dict[str, Any]:
+ async def repair_flatpak_support(self, flatpak_app_id: str):
return self.flatpak_service.ensure_app_support(flatpak_app_id)
- async def set_flatpak_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]:
+ async def set_flatpak_extension_enabled(self, version: str, enabled: bool):
return self.flatpak_service.set_extension_enabled(version, enabled)
-
+
async def _main(self):
- """
- Main entry point for the plugin.
-
- This method is called by Decky Loader when the plugin is loaded.
- Any initialization code should go here.
- """
repair = self.wrapper_service.repair()
if not repair.get("success"):
decky.logger.error(f"Could not repair lsfg workaround wrapper: {repair.get('error')}")
decky.logger.info("decky-lsfg-vk plugin loaded")
async def _unload(self):
- """
- Cleanup tasks when the plugin is unloaded.
-
- This method is called by Decky Loader when the plugin is being unloaded.
- Any cleanup code should go here.
- """
decky.logger.info("decky-lsfg-vk plugin unloaded")
async def _uninstall(self):
- """
- 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.
- """
decky.logger.info("decky-lsfg-vk plugin being uninstalled")
-
- # Clean up lsfg-vk files when the plugin is uninstalled
- # Launch integrations are removed with their profiles. Keep the
- # generated pass-through wrapper if it is still referenced elsewhere;
- # InstallationService only removes files owned by the runtime bundle.
self.installation_service.cleanup_on_uninstall()
-
try:
result = self.flatpak_service.remove_plugin_owned_extensions()
if not result.get("success"):
decky.logger.warning(result.get("error"))
except Exception as error:
decky.logger.error(f"Error during Flatpak cleanup: {error}")
-
decky.logger.info("decky-lsfg-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.
- """
decky.logger.info("Running decky-lsfg-vk plugin migrations")
-
- decky.migrate_logs(os.path.join(decky.DECKY_USER_HOME,
- ".config", "decky-lossless-scaling-vk", "lossless-scaling-vk.log"))
-
+ decky.migrate_logs(os.path.join(
+ decky.DECKY_USER_HOME,
+ ".config",
+ "decky-lossless-scaling-vk",
+ "lossless-scaling-vk.log",
+ ))
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"))
-
+ os.path.join(decky.DECKY_USER_HOME, ".config", "decky-lossless-scaling-vk"),
+ )
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"))
-
+ os.path.join(decky.DECKY_USER_HOME, ".local", "share", "decky-lossless-scaling-vk"),
+ )
decky.logger.info("decky-lsfg-vk plugin migrations completed")
diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py
index 50d722b..201441e 100644
--- a/py_modules/lsfg_vk/steam_service.py
+++ b/py_modules/lsfg_vk/steam_service.py
@@ -10,7 +10,6 @@ from .constants import (
WRAPPER_FILENAME,
)
-
_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$")
_WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}"
@@ -25,107 +24,89 @@ def _split_command(value: Optional[str]) -> Optional[list[str]]:
def _is_managed_wrapper(value: str) -> bool:
- """Recognize the wrapper Target while keeping arbitrary launchers as host games."""
if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}:
return True
path = Path(value)
return path.is_absolute() and path.name == WRAPPER_FILENAME
-def classify_shortcut_transport(
- executable: Optional[str],
- launch_options: Optional[str] = None,
-) -> Dict[str, object]:
- """Classify direct Flatpak invocations, including the managed wrapper Target."""
+def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]:
executable_tokens = _split_command(executable)
option_tokens = _split_command(launch_options)
if executable_tokens is None or option_tokens is None or not executable_tokens:
return {"kind": "host"}
-
direct_flatpak = executable_tokens[0] == "/usr/bin/flatpak"
managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0])
if not direct_flatpak and not managed_wrapper:
return {"kind": "host"}
-
arguments = [*executable_tokens[1:], *option_tokens]
if not arguments or arguments[0] != "run":
return {"kind": "host"}
-
for argument in arguments[1:]:
- if argument == "--":
+ if argument == "--" or argument.startswith("-"):
continue
- if argument.startswith("-"):
- continue
- if _FLATPAK_APP_ID.fullmatch(argument):
- return {"kind": "flatpak", "flatpakAppId": argument}
- return {"kind": "host"}
+ return (
+ {"kind": "flatpak", "flatpakAppId": argument}
+ if _FLATPAK_APP_ID.fullmatch(argument)
+ else {"kind": "host"}
+ )
return {"kind": "host"}
+def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]:
+ return next((values[key] for key in keys if isinstance(values.get(key), str)), None)
+
+
class SteamService(BaseService):
DEFAULT_BRANCH = "public"
MANIFEST_FILENAME = f"appmanifest_{STEAM_LOSSLESS_SCALING_APP_ID}.acf"
- # Valve compatibility tools, runtimes, Steamworks redistributables, and LSFG.
GAME_SELECTOR_EXCLUDED_APPIDS = {
- "858280", # Proton 3.7
- "961940", # Proton 3.16
- "1054830", # Proton 4.2
- "1113280", # Proton 4.11
- "1245040", # Proton 5.0
- "1420170", # Proton 5.13
- "1493710", # Proton Experimental
- "1580130", # Proton 6.3
- "1887720", # Proton 7
- "2180100", # Proton Hotfix
- "228980", # Steamworks Common Redistributables
- "2348590", # Proton 8
- "2805730", # Proton 9
- "3029110", # Lepton
- "3127680", # fex
- "3658110", # Proton 10
- "4183110", # Steam Linux Runtime 4.0
- "4185400", # Steam Linux Runtime 4.0 for arm64
- "4427310", # Proton Experimental (ARM64)
- "4628710", # Proton 11 / Proton Next
- "4628740", # Proton 11 (ARM64)
- "4690330", # Legacy Steam Runtime
- "993090", # Lossless Scaling
- "1070560", # Steam Linux Runtime 1.0
- "1391110", # Steam Linux Runtime 2.0
- "1628350", # Steam Linux Runtime 3.0
+ "858280", "961940", "1054830", "1113280", "1245040", "1420170",
+ "1493710", "1580130", "1887720", "2180100", "228980", "2348590",
+ "2805730", "3029110", "3127680", "3658110", "4183110", "4185400",
+ "4427310", "4628710", "4628740", "4690330", "993090", "1070560",
+ "1391110", "1628350",
}
def _steam_roots(self):
- candidates = (
+ seen = set()
+ for candidate in (
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)
def _steam_library_roots(self):
seen = set()
- for candidate in self._steam_roots():
- yield from self._unique_existing_root(candidate, seen)
-
+ for root in self._steam_roots():
+ yield from self._unique_existing_root(root, seen)
for library_file in (
- candidate / "steamapps/libraryfolders.vdf",
- candidate / "config/libraryfolders.vdf",
+ root / "steamapps/libraryfolders.vdf",
+ root / "config/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 not in seen:
+ seen.add(resolved)
+ yield path
+
+ @staticmethod
def _read_shortcuts(data: bytes) -> Dict[str, object]:
def read_string(offset: int) -> Tuple[str, int]:
end = data.index(b"\0", offset)
@@ -142,16 +123,12 @@ class SteamService(BaseService):
value, offset = read_object(offset)
elif value_type == 1:
value, offset = read_string(offset)
- elif value_type == 2:
- if offset + 4 > len(data):
+ elif value_type in (2, 7):
+ width = 4 if value_type == 2 else 8
+ if offset + width > len(data):
raise ValueError("truncated binary VDF integer")
- value = int.from_bytes(data[offset:offset + 4], "little", signed=True)
- offset += 4
- elif value_type == 7:
- if offset + 8 > len(data):
- raise ValueError("truncated binary VDF 64-bit integer")
- value = int.from_bytes(data[offset:offset + 8], "little", signed=True)
- offset += 8
+ value = int.from_bytes(data[offset:offset + width], "little", signed=True)
+ offset += width
else:
raise ValueError(f"unsupported binary VDF type {value_type}")
values[key] = value
@@ -170,53 +147,28 @@ class SteamService(BaseService):
name = shortcut.get("AppName") or shortcut.get("appname")
if not isinstance(appid, int) or appid == 0 or not isinstance(name, str) or not name:
return None
- executable = next(
- (
- shortcut.get(key)
- for key in ("Exe", "exe", "executable")
- if isinstance(shortcut.get(key), str)
- ),
- None,
- )
- launch_options = next(
- (
- shortcut.get(key)
- for key in ("LaunchOptions", "launchoptions", "launch_options", "arguments")
- if isinstance(shortcut.get(key), str)
- ),
- None,
- )
- start_dir = next(
- (
- shortcut.get(key)
- for key in ("StartDir", "startdir", "start_dir")
- if isinstance(shortcut.get(key), str)
- ),
- None,
- )
+ executable = _first_string(shortcut, "Exe", "exe", "executable")
+ arguments = _first_string(shortcut, "LaunchOptions", "launchoptions", "launch_options", "arguments")
+ start_dir = _first_string(shortcut, "StartDir", "startdir", "start_dir")
game: Dict[str, object] = {
- "appid": str(appid & 0xffffffff),
+ "appid": str(appid & 0xFFFFFFFF),
"name": name,
"nonSteam": True,
- "transport": classify_shortcut_transport(executable, launch_options),
+ "transport": classify_shortcut_transport(executable, arguments),
}
- if executable is not None:
- game["executable"] = executable
- if launch_options is not None:
- game["arguments"] = launch_options
- if start_dir is not None:
- game["startDir"] = start_dir
+ for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)):
+ if value is not None:
+ game[key] = value
return game
def _shortcut_games(self):
games = {}
- for steam_root in self._steam_roots():
- for shortcuts_file in sorted((steam_root / "userdata").glob("*/config/shortcuts.vdf")):
+ for root in self._steam_roots():
+ for path in sorted((root / "userdata").glob("*/config/shortcuts.vdf")):
try:
- root = self._read_shortcuts(shortcuts_file.read_bytes())
+ shortcuts = self._read_shortcuts(path.read_bytes()).get("shortcuts", {})
except (OSError, ValueError):
continue
- shortcuts = root.get("shortcuts", {})
if not isinstance(shortcuts, dict):
continue
for shortcut in shortcuts.values():
@@ -225,25 +177,12 @@ class SteamService(BaseService):
games.setdefault(game["appid"], game)
return list(games.values())
- @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
+ return next((
+ path
+ for root in self._steam_library_roots()
+ if (path := root / "steamapps" / self.MANIFEST_FILENAME).is_file()
+ ), None)
@staticmethod
def _section_bounds(content: str, section_name: str) -> Optional[Tuple[int, int, str]]:
@@ -253,7 +192,6 @@ class SteamService(BaseService):
)
if section is None:
return None
-
depth = 1
in_string = False
escaped = False
@@ -266,9 +204,7 @@ class SteamService(BaseService):
escaped = True
elif character == '"':
in_string = False
- continue
-
- if character == '"':
+ elif character == '"':
in_string = True
elif character == "{":
depth += 1
@@ -283,98 +219,82 @@ class SteamService(BaseService):
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
+ start, end, _ = bounds
+ pattern = re.compile(r'(?m)^[ \t]*"(?P<key>[^"]+)"[ \t]+"(?P<value>(?:\\.|[^"\\])*)"')
+ return next((
+ match.group("value")
+ for match in pattern.finditer(content, start, end)
+ if match.group("key") == key
+ ), None)
@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(
+ selected = self._branch_or_default(self._section_value(content, "UserConfig", "BetaKey"))
+ current = 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
- )
+ needs_switch = selected != STEAM_LOSSLESS_SCALING_BRANCH or current != STEAM_LOSSLESS_SCALING_BRANCH
return {
"installed": True,
"manifest_path": str(manifest_path),
- "selected_branch": selected_branch,
- "current_branch": current_branch,
+ "selected_branch": selected,
+ "current_branch": current,
"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
- ),
+ "restart_required": selected == STEAM_LOSSLESS_SCALING_BRANCH and current != STEAM_LOSSLESS_SCALING_BRANCH,
+ }
+
+ @staticmethod
+ def _missing_branch_fields() -> Dict[str, object]:
+ return {
+ "installed": False,
+ "manifest_path": None,
+ "selected_branch": None,
+ "current_branch": None,
+ "target_branch": STEAM_LOSSLESS_SCALING_BRANCH,
+ "needs_switch": False,
+ "restart_required": False,
}
def find_lsfg_vk_dll(self) -> Optional[str]:
- """Find the branch-specific upstream DLL in any Steam library."""
if self.get_branch_status().get("needs_switch"):
return None
- for library_root in self._steam_library_roots():
- dll_path = library_root / "steamapps/common/Lossless Scaling/lsfg-vk.dll"
- if dll_path.is_file():
- return str(dll_path)
- return None
+ return next((
+ str(path)
+ for root in self._steam_library_roots()
+ if (path := root / "steamapps/common/Lossless Scaling/lsfg-vk.dll").is_file()
+ ), None)
def get_branch_status(self) -> Dict[str, object]:
try:
- manifest_path = self._manifest_path()
- if manifest_path is None:
+ manifest = self._manifest_path()
+ if manifest 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,
+ **self._missing_branch_fields(),
)
-
- 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 = "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas"
+ fields = self._status_fields(manifest, manifest.read_text(encoding="utf-8"))
+ message = (
+ "Lossless Scaling is using the lsfg-vk Steam branch"
+ if not fields["needs_switch"]
+ else "lsfg-vk is selected; restart Steam to finish the branch switch"
+ if fields["restart_required"]
+ else "Select lsfg-vk in Lossless Scaling's Steam Properties > Betas"
+ )
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,
- )
+ return self._error_response(dict, str(error), **self._missing_branch_fields())
def get_installed_games(self) -> Dict[str, object]:
- """Return installed Steam app IDs and names for the Game Mode selector."""
try:
games: Dict[str, Dict[str, object]] = {}
- for library_root in self._steam_library_roots():
- for manifest in (library_root / "steamapps").glob("appmanifest_*.acf"):
+ for root in self._steam_library_roots():
+ for manifest in (root / "steamapps").glob("appmanifest_*.acf"):
match = re.fullmatch(r"appmanifest_(\d+)\.acf", manifest.name)
if not match:
continue
@@ -385,10 +305,9 @@ class SteamService(BaseService):
appid = match.group(1)
if appid in self.GAME_SELECTOR_EXCLUDED_APPIDS:
continue
- name = self._section_value(content, "AppState", "name") or f"App {appid}"
games[appid] = {
"appid": appid,
- "name": name,
+ "name": self._section_value(content, "AppState", "name") or f"App {appid}",
"nonSteam": False,
"transport": {"kind": "host"},
}
diff --git a/py_modules/lsfg_vk/types.py b/py_modules/lsfg_vk/types.py
index 7b85708..ce541ec 100644
--- a/py_modules/lsfg_vk/types.py
+++ b/py_modules/lsfg_vk/types.py
@@ -1,40 +1,17 @@
-"""
-Type definitions for the lsfg-vk plugin responses.
-"""
+from typing import List, Optional, TypedDict
-from typing import TypedDict, Optional, List
-
-class BaseResponse(TypedDict):
- """Base response structure"""
+class InstallationResponse(TypedDict):
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
+class UninstallationResponse(InstallationResponse):
removed_files: Optional[List[str]]
- error: Optional[str]
class InstallationCheckResponse(TypedDict):
- """Response for installation check"""
installed: bool
lossless_scaling_installed: bool
lossless_scaling_status: str
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-11 05:14:03 +0000