summaryrefslogtreecommitdiff
path: root/py_modules/lsfg_vk/steam_service.py
diff options
context:
space:
mode:
Diffstat (limited to 'py_modules/lsfg_vk/steam_service.py')
-rw-r--r--py_modules/lsfg_vk/steam_service.py301
1 files changed, 89 insertions, 212 deletions
diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py
index b3bdb69..2108071 100644
--- a/py_modules/lsfg_vk/steam_service.py
+++ b/py_modules/lsfg_vk/steam_service.py
@@ -1,116 +1,64 @@
import re
-import shlex
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
-
-
-_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$")
-
-
-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 classify_shortcut_transport(
- executable: Optional[str],
- launch_options: Optional[str] = None,
-) -> Dict[str, object]:
- """Classify only direct Flatpak invocations; leave shell launchers on host."""
- 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"}
-
- if executable_tokens[0] != "/usr/bin/flatpak":
- 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 == "--":
- continue
- if argument.startswith("-"):
- continue
- if _FLATPAK_APP_ID.fullmatch(argument):
- return {"kind": "flatpak", "flatpakAppId": argument}
- return {"kind": "host"}
- return {"kind": "host"}
+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"
- # 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)
@@ -127,16 +75,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
@@ -155,53 +99,20 @@ 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,
- )
- game: Dict[str, object] = {
- "appid": str(appid & 0xffffffff),
+ return {
+ "appid": str(appid & 0xFFFFFFFF),
"name": name,
"nonSteam": True,
- "transport": classify_shortcut_transport(executable, launch_options),
}
- 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
- 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():
@@ -210,25 +121,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]]:
@@ -238,7 +136,6 @@ class SteamService(BaseService):
)
if section is None:
return None
-
depth = 1
in_string = False
escaped = False
@@ -251,9 +148,7 @@ class SteamService(BaseService):
escaped = True
elif character == '"':
in_string = False
- continue
-
- if character == '"':
+ elif character == '"':
in_string = True
elif character == "{":
depth += 1
@@ -268,98 +163,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
@@ -370,12 +249,10 @@ 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"},
}
for game in self._shortcut_games():
games.setdefault(str(game["appid"]), game)
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-11 08:39:36 +0000