summaryrefslogtreecommitdiff
path: root/py_modules/lsfg_vk
diff options
context:
space:
mode:
authorxXJSONDeruloXx <danielhimebauch@gmail.com>2026-09-10 16:26:17 -0400
committerxXJSONDeruloXx <danielhimebauch@gmail.com>2026-09-10 16:26:17 -0400
commit485ebb748185391e28764f48c39322b3cbdac65c (patch)
tree9b03047ed842e14d9c800bcdfa16cad322733cf4 /py_modules/lsfg_vk
parentde74f0d2499159ed1cf8f628a166302146ae1f13 (diff)
downloaddecky-lsfg-vk-v2-flatpakless.tar.gz
decky-lsfg-vk-v2-flatpakless.zip
refactor: remove Flatpak supportv2-flatpakless
Diffstat (limited to 'py_modules/lsfg_vk')
-rw-r--r--py_modules/lsfg_vk/constants.py3
-rw-r--r--py_modules/lsfg_vk/flatpak_service.py431
-rw-r--r--py_modules/lsfg_vk/plugin.py49
-rw-r--r--py_modules/lsfg_vk/steam_service.py47
-rw-r--r--py_modules/lsfg_vk/wrapper_service.py193
5 files changed, 8 insertions, 715 deletions
diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py
index 960d230..45ac4c5 100644
--- a/py_modules/lsfg_vk/constants.py
+++ b/py_modules/lsfg_vk/constants.py
@@ -16,9 +16,6 @@ CLI_FILENAME = "lsfg-vk-cli"
UI_FILENAME = "lsfg-vk-ui"
UI_DESKTOP_FILENAME = "gay.pancake.lsfg-vk-ui.desktop"
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"
diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py
deleted file mode 100644
index f486b74..0000000
--- a/py_modules/lsfg_vk/flatpak_service.py
+++ /dev/null
@@ -1,431 +0,0 @@
-"""Flatpak runtime support for classified Steam targets."""
-
-from __future__ import annotations
-
-import json
-import os
-import pwd
-import re
-import shutil
-import subprocess
-import threading
-from pathlib import Path
-from typing import Dict, Optional, Set
-
-from .base_service import BaseService
-from .constants import (
- BIN_DIR,
- FLATPAK_23_08_FILENAME,
- FLATPAK_24_08_FILENAME,
- FLATPAK_25_08_FILENAME,
-)
-
-
-class FlatpakService(BaseService):
- EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk"
- SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08")
- DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"}
- RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL"
- OWNERSHIP_FILENAME = "flatpak_extensions.json"
- OWNERSHIP_VERSION = 1
- APP_ID_PATTERN = re.compile(
- r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$"
- )
-
- def __init__(self, logger=None):
- super().__init__(logger)
- self.flatpak_command: Optional[str] = None
- self._lock = threading.RLock()
-
- @property
- def ownership_path(self) -> Path:
- return self.config_dir / self.OWNERSHIP_FILENAME
-
- def _clean_env(self) -> Dict[str, str]:
- env = os.environ.copy()
- env.pop("LD_LIBRARY_PATH", None)
- env["HOME"] = str(self.user_home)
- 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:
- path.insert(0, entry)
- env["PATH"] = ":".join(path)
- return env
-
- def check_flatpak_available(self) -> bool:
- 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, **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]
- 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", user.pw_name, "--", *command]
- return subprocess.run(command, env=env, **kwargs)
-
- @classmethod
- def _validate_app_id(cls, app_id: str) -> str:
- if not isinstance(app_id, str) or not cls.APP_ID_PATTERN.fullmatch(app_id):
- raise ValueError("Invalid Flatpak application ID")
- return app_id
-
- @classmethod
- def _validate_runtime(cls, branch: str) -> str:
- if branch not in cls.SUPPORTED_RUNTIMES:
- raise ValueError(
- f"Unsupported Flatpak runtime branch {branch}; supported branches are "
- + ", ".join(cls.SUPPORTED_RUNTIMES)
- )
- return branch
-
- @classmethod
- def runtime_branch_from_ref(cls, runtime_ref: str) -> str:
- 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}")
- return cls._validate_runtime(parts[2])
-
- @classmethod
- def runtime_branch_from_metadata(cls, metadata: str) -> str:
- section = None
- versions = []
- for raw_line in metadata.splitlines() if isinstance(metadata, str) else []:
- line = raw_line.strip()
- if line.startswith("[") and line.endswith("]"):
- section = line[1:-1].strip()
- continue
- if section != cls.RUNTIME_METADATA_SECTION:
- continue
- key, separator, value = line.partition("=")
- if separator and key.strip() == "versions":
- versions.extend(part.strip() for part in value.split(";"))
- for value in versions:
- for branch in cls.SUPPORTED_RUNTIMES:
- if value == branch or value.startswith(f"{branch}-"):
- return branch
- raise ValueError("Could not determine a supported Freedesktop base runtime from Flatpak metadata")
-
- @classmethod
- 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,
- }[self._validate_runtime(branch)]
- return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename
-
- def _installed_extension_branches(self, scope: Optional[str] = None) -> Set[str]:
- scopes = ("user", "system") if scope is None else (scope,)
- if any(item not in ("user", "system") for item in scopes):
- raise ValueError("Flatpak installation scope must be user or system")
- installed = set()
- for item in scopes:
- result = self._run_flatpak_command(
- ["list", f"--{item}", "--runtime", "--columns=application,arch,branch"],
- capture_output=True,
- text=True,
- check=True,
- )
- for line in result.stdout.splitlines():
- 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 _owned_branches(self) -> Set[str]:
- path = self.ownership_path
- 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:
- 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:
- raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error
-
- def _write_owned_branches(self, branches: Set[str]) -> None:
- if not branches:
- self.ownership_path.unlink(missing_ok=True)
- return
- 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):
- try:
- available = self.check_flatpak_available()
- installed = self._installed_extension_branches() if available else set()
- return self._success_response(
- dict,
- "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),
- )
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- available=False,
- extension_id=self.EXTENSION_ID,
- supported_branches=list(self.SUPPORTED_RUNTIMES),
- installed_branches=[],
- )
-
- get_flatpak_support_status = get_extension_status
-
- 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")
- result = self._run_flatpak_command(
- ["info", "--show-runtime", app_id],
- capture_output=True,
- text=True,
- )
- if result.returncode != 0:
- raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}")
- runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
- parts = runtime.split("/")
- if len(parts) != 3:
- raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}")
- if parts[0] == "org.freedesktop.Platform":
- branch = self._validate_runtime(parts[2])
- elif parts[0] in self.DERIVED_RUNTIME_IDS:
- metadata_result = self._run_flatpak_command(
- ["info", "--show-metadata", runtime],
- capture_output=True,
- text=True,
- )
- if metadata_result.returncode != 0:
- raise OSError(
- metadata_result.stderr.strip()
- or f"Could not inspect Flatpak runtime {runtime}"
- )
- branch = self.runtime_branch_from_metadata(metadata_result.stdout)
- else:
- raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}")
- return runtime, branch
-
- def resolve_app_support(self, app_id: str):
- try:
- app_id = self._validate_app_id(app_id)
- runtime, branch = self._resolve_runtime(app_id)
- installed = self._installed_extension_branches()
- 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}",
- flatpak_app_id=app_id,
- runtime=runtime,
- runtime_branch=branch,
- support_status="ready" if ready else "needs-runtime",
- extension_installed=ready,
- installed_branches=sorted(installed),
- )
- except ValueError as error:
- return self._success_response(
- dict,
- str(error),
- flatpak_app_id=app_id,
- runtime=None,
- runtime_branch=None,
- support_status="unsupported",
- extension_installed=False,
- installed_branches=[],
- error=str(error),
- )
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- flatpak_app_id=app_id,
- runtime=None,
- runtime_branch=None,
- support_status="error",
- extension_installed=False,
- installed_branches=[],
- )
-
- def install_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:
- 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)],
- capture_output=True,
- text=True,
- )
- if result.returncode != 0:
- raise OSError(result.stderr.strip() or "Flatpak installation failed")
- if branch not in self._installed_extension_branches("user"):
- 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("user"):
- 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("user"):
- 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:
- owned = self._owned_branches()
- if branch not in owned:
- installed = branch in self._installed_extension_branches()
- return self._extension_result(branch, installed, False, "preserved (not plugin-owned)")
- removed = self._remove_extension(branch)
- owned.remove(branch)
- self._write_owned_branches(owned)
- installed = branch in self._installed_extension_branches()
- return self._extension_result(branch, installed, removed, "uninstalled")
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- runtime_branch=branch,
- removed=False,
- installed=False,
- enabled=False,
- )
-
- def ensure_extension(self, branch: str):
- try:
- 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):
- resolved = self.resolve_app_support(app_id)
- if not resolved.get("success") or resolved.get("support_status") != "needs-runtime":
- return resolved
- 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=resolved.get("runtime_branch"),
- support_status="error",
- extension_installed=False,
- )
- return self.resolve_app_support(app_id)
-
- 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=branch, installed=False, enabled=False)
- return self.install_extension(branch) if enabled else self.uninstall_extension(branch)
-
- def remove_plugin_owned_extensions(self):
- try:
- with self._lock:
- owned = self._owned_branches()
- if not owned:
- return self._success_response(
- dict,
- "No plugin-owned Flatpak extensions to remove",
- removed_branches=[],
- preserved_branches=[],
- ownership_uncertain=False,
- )
- if not self.check_flatpak_available():
- raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved")
- removed, failures = [], []
- for branch in sorted(owned):
- try:
- self._remove_extension(branch)
- removed.append(branch)
- except Exception as error:
- failures.append(f"{branch}: {error}")
- remaining = owned - set(removed)
- self._write_owned_branches(remaining)
- if failures:
- return self._error_response(
- dict,
- "; ".join(failures),
- removed_branches=removed,
- preserved_branches=sorted(remaining),
- ownership_uncertain=False,
- )
- return self._success_response(
- dict,
- "Plugin-owned Flatpak extensions removed",
- removed_branches=removed,
- preserved_branches=[],
- ownership_uncertain=False,
- )
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- removed_branches=[],
- preserved_branches=[],
- ownership_uncertain=True,
- )
diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py
index 13df7a4..b010db6 100644
--- a/py_modules/lsfg_vk/plugin.py
+++ b/py_modules/lsfg_vk/plugin.py
@@ -1,10 +1,9 @@
import os
-from typing import Any, Dict, Optional
+from typing import Any, Dict
import decky
from .configuration import ConfigurationService
-from .flatpak_service import FlatpakService
from .installation import InstallationService
from .runtime_service import RuntimeService
from .steam_service import SteamService
@@ -20,7 +19,6 @@ class Plugin:
steam_service=self.steam_service,
)
self.configuration_service = ConfigurationService(runtime_service=self.runtime_service)
- self.flatpak_service = FlatpakService()
self.wrapper_service = WrapperService()
async def install_lsfg_vk(self):
@@ -36,21 +34,7 @@ class Plugin:
return self.configuration_service.get_game_configs()
async def get_installed_games(self):
- result = self.steam_service.get_installed_games()
- if not result.get("success"):
- return result
- cache: Dict[str, Dict[str, Any]] = {}
- for game in result.get("games", []):
- transport = game.get("transport", {})
- if transport.get("kind") != "flatpak":
- continue
- app_id = transport.get("flatpakAppId")
- if not app_id:
- continue
- if app_id not in cache:
- cache[app_id] = self.flatpak_service.resolve_app_support(app_id)
- game["flatpakSupport"] = cache[app_id]
- return result
+ return self.steam_service.get_installed_games()
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)
@@ -68,17 +52,9 @@ class Plugin:
self,
appid: str,
state: Dict[str, Any],
- shortcut_exe: Optional[str] = None,
command_token_added: bool = False,
- transport: Optional[Dict[str, Any]] = None,
):
- return self.wrapper_service.set(
- appid,
- state,
- shortcut_exe,
- command_token_added,
- transport,
- )
+ return self.wrapper_service.set(appid, state, command_token_added)
async def remove_workaround_state(self, appid: str):
return self.wrapper_service.remove(appid)
@@ -112,7 +88,6 @@ class Plugin:
("config", "LSFG-VK configuration", self.configuration_service.config_file_path),
("workarounds", "Per-app workarounds", self.wrapper_service.sidecar_path),
("wrapper", "Generated launch wrapper", self.wrapper_service.wrapper_path),
- ("flatpak_extensions", "Flatpak extension ownership", self.flatpak_service.ownership_path),
)
contents = []
for file_id, label, path in files:
@@ -147,18 +122,6 @@ class Plugin:
async def get_lossless_scaling_branch_status(self):
return self.steam_service.get_branch_status()
- 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):
- return self.flatpak_service.ensure_app_support(flatpak_app_id)
-
- 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):
- return self.flatpak_service.set_extension_enabled(version, enabled)
-
async def _main(self):
repair = self.wrapper_service.repair()
if not repair.get("success"):
@@ -171,12 +134,6 @@ class Plugin:
async def _uninstall(self):
decky.logger.info("decky-lsfg-vk plugin being uninstalled")
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):
diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py
index a952fcb..53748de 100644
--- a/py_modules/lsfg_vk/steam_service.py
+++ b/py_modules/lsfg_vk/steam_service.py
@@ -1,5 +1,4 @@
import re
-import shlex
from pathlib import Path
from typing import Dict, Optional, Tuple
@@ -7,51 +6,8 @@ from .base_service import BaseService
from .constants import (
STEAM_LOSSLESS_SCALING_APP_ID,
STEAM_LOSSLESS_SCALING_BRANCH,
- 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}"
-
-
-def _split_command(value: Optional[str]) -> Optional[list[str]]:
- if not isinstance(value, str) or not value.strip():
- return []
- try:
- return shlex.split(value, posix=True)
- except ValueError:
- return None
-
-
-def _is_managed_wrapper(value: str) -> bool:
- 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]:
- 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] in {"flatpak", "/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 == "--" or argument.startswith("-"):
- continue
- 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)
@@ -74,7 +30,6 @@ class SteamService(BaseService):
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",
):
yield from self._unique_existing_root(candidate, seen)
@@ -154,7 +109,6 @@ class SteamService(BaseService):
"appid": str(appid & 0xFFFFFFFF),
"name": name,
"nonSteam": True,
- "transport": classify_shortcut_transport(executable, arguments),
}
for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)):
if value is not None:
@@ -309,7 +263,6 @@ class SteamService(BaseService):
"appid": appid,
"name": self._section_value(content, "AppState", "name") or f"App {appid}",
"nonSteam": False,
- "transport": {"kind": "host"},
}
for game in self._shortcut_games():
games.setdefault(str(game["appid"]), game)
diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py
index 980f7ae..69a3600 100644
--- a/py_modules/lsfg_vk/wrapper_service.py
+++ b/py_modules/lsfg_vk/wrapper_service.py
@@ -4,9 +4,7 @@ from __future__ import annotations
import json
import re
-import shlex
import threading
-from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from .base_service import BaseService
@@ -86,51 +84,18 @@ class WrapperService(BaseService):
return state
@classmethod
- def _validate_transport(cls, raw: Any) -> Dict[str, Any]:
- if raw is None:
- return {"kind": "host"}
- if not isinstance(raw, dict):
- raise ValueError("Workaround transport must be an object")
- kind = raw.get("kind")
- if kind == "host":
- return {"kind": "host"}
- if kind == "flatpak":
- app_id = raw.get("flatpakAppId")
- if (
- not isinstance(app_id, str)
- or not re.fullmatch(
- r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$",
- app_id,
- )
- ):
- raise ValueError("Flatpak transport requires a valid application ID")
- return {"kind": "flatpak", "flatpakAppId": app_id}
- raise ValueError("Workaround transport must be host or flatpak")
-
- @classmethod
def _validate_entry(cls, raw: Any) -> Dict[str, Any]:
if not isinstance(raw, dict):
raise ValueError("Workaround AppID entry must be an object")
+ unsupported = set(raw) - {"state", "command_token_added"}
+ if unsupported:
+ raise ValueError("Unsupported workaround entry fields: " + ", ".join(sorted(unsupported)))
entry = {
"state": cls._validate_state(raw.get("state")),
"command_token_added": raw.get("command_token_added", False),
- # Version 1 entries had no transport field. They are preserved as
- # host entries until the shortcut is explicitly repaired with the
- # backend's classified transport.
- "transport": cls._validate_transport(raw.get("transport")),
}
if type(entry["command_token_added"]) is not bool:
raise ValueError("command_token_added must be a boolean")
- if entry["transport"]["kind"] == "flatpak" and "shortcut_exe" in raw and raw["shortcut_exe"] is not None:
- shortcut_exe = raw["shortcut_exe"]
- if (
- not isinstance(shortcut_exe, str)
- or not shortcut_exe.startswith("/")
- or "\x00" in shortcut_exe
- or not shortcut_exe.strip()
- ):
- raise ValueError("shortcut_exe must be an absolute executable path")
- entry["shortcut_exe"] = shortcut_exe
return entry
@classmethod
@@ -186,23 +151,8 @@ class WrapperService(BaseService):
)
return True
- @staticmethod
- def _shell(value: str) -> str:
- return shlex.quote(value)
-
- @staticmethod
- def _direct_flatpak_tokens(value: str) -> Optional[list[str]]:
- """Parse the supported full executable form: /usr/bin/flatpak run APP."""
- try:
- tokens = shlex.split(value, posix=True)
- except ValueError:
- return None
- if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run":
- return tokens
- return None
-
@classmethod
- def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]:
+ def _state_lines(cls, state: Dict[str, Any]) -> list[str]:
lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)]
if state["disableGamescopeWsi"]:
lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"])
@@ -235,68 +185,8 @@ class WrapperService(BaseService):
" fi",
" export DXVK_CONFIG",
])
- lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}")
return lines
- def _dll_directory(self) -> Path:
- if self.config_file_path.exists():
- try:
- content = self.config_file_path.read_text(encoding="utf-8")
- match = re.search(
- r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"',
- content,
- )
- if match:
- configured_dll = json.loads('"' + match.group(1) + '"')
- if configured_dll:
- return Path(configured_dll).parent
- except Exception:
- pass
- return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling"
-
- def _flatpak_args(self, state: Dict[str, Any]) -> list[str]:
- config_dir = str(self.config_dir)
- config_file = str(self.config_file_path)
- dll_dir = str(self._dll_directory())
- args = [
- self._shell(f"--filesystem={config_dir}:rw"),
- self._shell(f"--filesystem={dll_dir}:ro"),
- self._shell(f"--env=LSFGVK_CONFIG={config_file}"),
- '"--env=LSFGVK_FLATPAK=1"',
- '"--env=SteamAppId=$appid"',
- '"--unset-env=DISABLE_LSFGVK" "--unset-env=DISABLE_LSFG"',
- '"--unset-env=DISABLE_GAMESCOPE_WSI"',
- '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else
- '"--env=ENABLE_GAMESCOPE_WSI=0"',
- '"--unset-env=DXVK_HDR"' if not state["disableHdr"] else
- '"--env=DXVK_HDR=0"',
- '"--unset-env=SteamDeck"' if not state["disableSteamdeckMode"] else
- '"--env=SteamDeck=0"',
- '"--unset-env=DISABLE_VKBASALT" "--unset-env=ENABLE_VKBASALT"',
- ]
- if state["disableVkbasalt"]:
- args.append('"--env=DISABLE_VKBASALT=1"')
- args.extend([
- '"--unset-env=MESA_LOADER_DRIVER_OVERRIDE" "--unset-env=__GLX_VENDOR_LIBRARY_NAME" "--unset-env=GALLIUM_DRIVER"',
- ])
- if state["enableZink"]:
- args.extend([
- '"--env=__GLX_VENDOR_LIBRARY_NAME=mesa"',
- '"--env=MESA_LOADER_DRIVER_OVERRIDE=zink"',
- '"--env=GALLIUM_DRIVER=zink"',
- ])
- args.extend([
- '"--unset-env=DXVK_FRAME_RATE"',
- ])
- static_args = " ".join(args)
- return [
- ' if [ -n "${DXVK_CONFIG+x}" ]; then',
- f' set -- "$flatpak_command" {static_args} "--env=DXVK_CONFIG=$DXVK_CONFIG" "$@"',
- " else",
- f' set -- "$flatpak_command" {static_args} "$@"',
- " fi",
- ]
-
def _render_wrapper(self, document: Dict[str, Any]) -> str:
lines = [
"#!/bin/sh",
@@ -320,69 +210,16 @@ class WrapperService(BaseService):
' *) appid="${STEAM_COMPAT_APP_ID}" ;;',
" esac",
"fi",
- "shortcut_exe=",
'case "$appid" in',
]
for appid in sorted(document["apps"], key=lambda value: int(value)):
entry = document["apps"][appid]
lines.append(f" {appid})")
- lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe")))
+ lines.extend(self._state_lines(entry["state"]))
lines.append(" ;;")
lines.extend([
"esac",
"",
- 'if [ -n "$shortcut_exe" ]; then',
- ])
- # The arguments are emitted per branch below so the values are static and
- # the wrapper never needs a JSON parser or another helper executable.
- lines.append(' case "$appid" in')
- for appid in sorted(document["apps"], key=lambda value: int(value)):
- entry = document["apps"][appid]
- transport = entry.get("transport", {"kind": "host"})
- if transport.get("kind") != "flatpak":
- continue
- shortcut_exe = entry.get("shortcut_exe", "")
- direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe)
- if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak":
- raise ValueError(
- f"Flatpak target {appid} does not use a direct flatpak executable"
- )
- lines.append(f" {appid})")
- lines.extend([
- *(
- [
- f" shortcut_exe={self._shell(direct_flatpak_tokens[0])}",
- " set -- "
- + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:])
- + ' "$@"',
- ]
- if direct_flatpak_tokens
- else []
- ),
- ' if [ "${1-}" != "run" ]; then',
- ' echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2',
- " exit 64",
- " fi",
- ' flatpak_command="$1"',
- " shift",
- " flatpak_target=",
- ' for flatpak_arg in "$@"; do',
- ' case "$flatpak_arg" in',
- ' -*) ;;',
- ' *) flatpak_target="$flatpak_arg"; break ;;',
- " esac",
- " done",
- f' if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then',
- ' echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2',
- " exit 64",
- " fi",
- ])
- lines.extend(self._flatpak_args(entry["state"]))
- lines.append(" ;;")
- lines.extend([
- " esac",
- ' exec "$shortcut_exe" "$@"',
- "fi",
'exec "$@"',
"",
])
@@ -424,9 +261,7 @@ class WrapperService(BaseService):
"state": dict(entry["state"]) if entry else None,
"wrapper_path": self.WRAPPER_TOKEN,
"wrapper_owned": self._wrapper_marker() if document["apps"] else False,
- "shortcut_exe": entry.get("shortcut_exe") if entry else None,
"command_token_added": entry.get("command_token_added", False) if entry else False,
- "transport": dict(entry.get("transport", {"kind": "host"})) if entry else None,
}
def get(self, appid: str) -> Dict[str, Any]:
@@ -451,9 +286,7 @@ class WrapperService(BaseService):
self,
appid: str,
state: Dict[str, Any],
- shortcut_exe: Optional[str] = None,
command_token_added: bool = False,
- transport: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
try:
normalized = self._valid_appid(appid)
@@ -463,26 +296,10 @@ class WrapperService(BaseService):
with self._lock:
self._assert_wrapper_owned_or_absent()
document, _, _ = self._read_document()
- previous_entry = document["apps"].get(normalized)
- selected_transport = self._validate_transport(
- transport
- if transport is not None
- else (
- previous_entry.get("transport")
- if previous_entry
- else None
- )
- )
entry: Dict[str, Any] = {
"state": validated_state,
"command_token_added": bool(command_token_added),
- "transport": selected_transport,
}
- if selected_transport["kind"] == "flatpak":
- if shortcut_exe is not None:
- entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe})
- elif previous_entry and "shortcut_exe" in previous_entry:
- entry["shortcut_exe"] = previous_entry["shortcut_exe"]
document["apps"][normalized] = entry
self._write_pair(document)
return self._response(document, normalized)