Add per-game config APIs and proxy persistence

1 files changed, 309 insertions, 30 deletions

diff --git a/main.py b/main.py
index 4407e03..91824e6 100644
--- a/main.py
+++ b/main.py
@@ -3,6 +3,7 @@ import os
 import re
 import shutil
 import subprocess
+from configparser import ConfigParser
 from pathlib import Path
 
 # Toggle to enable overwriting the upscaler DLL from the static remote binary.
@@ -62,6 +63,17 @@ LEGACY_FILES = [
     "OptiScaler.log",
 ]
 
+QUICK_SETTING_DEFINITIONS = [
+    {"id": "Dx12Upscaler", "section": None, "key": "Dx12Upscaler", "default": "auto"},
+    {"id": "FrameGen.Enabled", "section": "FrameGen", "key": "Enabled", "default": "auto"},
+    {"id": "FGInput", "section": None, "key": "FGInput", "default": "auto"},
+    {"id": "FGOutput", "section": None, "key": "FGOutput", "default": "auto"},
+    {"id": "Fsr4ForceCapable", "section": None, "key": "Fsr4ForceCapable", "default": "false"},
+    {"id": "Fsr4EnableWatermark", "section": None, "key": "Fsr4EnableWatermark", "default": "false"},
+    {"id": "UseHQFont", "section": None, "key": "UseHQFont", "default": "false"},
+    {"id": "Menu.Scale", "section": "Menu", "key": "Scale", "default": "1.000000"},
+]
+
 
 class Plugin:
     async def _main(self):
@@ -152,6 +164,202 @@ class Plugin:
 
         return data
 
+    def _write_manifest_env(self, manifest_path: Path, updates: dict) -> dict:
+        data = self._parse_manifest_env(manifest_path)
+        for key, value in updates.items():
+            if value is None:
+                continue
+            data[str(key)] = str(value)
+
+        manifest_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(manifest_path, "w", encoding="utf-8") as manifest:
+            for key in sorted(data.keys()):
+                manifest.write(f'{key}="{data[key]}"\n')
+        return data
+
+    def _normalize_proxy(self, proxy: str | None) -> str:
+        normalized = (proxy or "winmm").replace(".dll", "").strip().lower()
+        if normalized not in SUPPORTED_PROXIES:
+            raise ValueError(f"Unsupported proxy '{proxy}'")
+        return normalized
+
+    def _find_installed_game(self, appid: str | None = None) -> list[dict]:
+        games = []
+        for library_path in self._steam_library_paths():
+            steamapps_path = library_path / "steamapps"
+            if not steamapps_path.exists():
+                continue
+
+            for appmanifest in steamapps_path.glob("appmanifest_*.acf"):
+                game_info = {"appid": "", "name": "", "library_path": str(library_path)}
+                try:
+                    with open(appmanifest, "r", encoding="utf-8", errors="replace") as file:
+                        for line in file:
+                            if '"appid"' in line:
+                                game_info["appid"] = line.split('"appid"', 1)[1].strip().strip('"')
+                            if '"name"' in line:
+                                game_info["name"] = line.split('"name"', 1)[1].strip().strip('"')
+                except Exception as exc:
+                    decky.logger.error(f"Skipping {appmanifest}: {exc}")
+
+                if not game_info["appid"] or not game_info["name"]:
+                    continue
+                if "Proton" in game_info["name"] or "Steam Linux Runtime" in game_info["name"]:
+                    continue
+                if appid is None or str(game_info["appid"]) == str(appid):
+                    games.append(game_info)
+
+        deduped = {}
+        for game in games:
+            deduped[str(game["appid"])] = game
+        return list(deduped.values())
+
+    def _compatdata_dir_for_appid(self, appid: str, create: bool = False) -> Path | None:
+        compatdata_dirs = self._compatdata_dirs_for_appid(str(appid))
+        if compatdata_dirs:
+            return compatdata_dirs[0]
+
+        if not create:
+            return None
+
+        installed_games = self._find_installed_game(appid)
+        if not installed_games:
+            return None
+
+        library_path = Path(installed_games[0]["library_path"])
+        compatdata_dir = library_path / "steamapps" / "compatdata" / str(appid)
+        compatdata_dir.mkdir(parents=True, exist_ok=True)
+        return compatdata_dir
+
+    def _managed_paths_for_appid(self, appid: str, create: bool = False) -> dict | None:
+        compatdata_dir = self._compatdata_dir_for_appid(str(appid), create=create)
+        if not compatdata_dir:
+            return None
+
+        managed_root = compatdata_dir / MANAGED_DIRNAME
+        system32 = compatdata_dir / "pfx" / "drive_c" / "windows" / "system32"
+        if create:
+            managed_root.mkdir(parents=True, exist_ok=True)
+
+        return {
+            "compatdata_dir": compatdata_dir,
+            "managed_root": managed_root,
+            "manifest_path": managed_root / MANIFEST_FILENAME,
+            "managed_ini": managed_root / "OptiScaler.ini",
+            "system32": system32,
+            "live_ini": system32 / "OptiScaler.ini",
+        }
+
+    def _ensure_managed_ini(self, appid: str) -> dict:
+        paths = self._managed_paths_for_appid(appid, create=True)
+        if not paths:
+            raise FileNotFoundError(f"Unable to resolve compatdata path for app {appid}")
+
+        managed_ini = paths["managed_ini"]
+        if managed_ini.exists():
+            return paths
+
+        bundle_ini = self._bundle_path() / "OptiScaler.ini"
+        if not bundle_ini.exists():
+            raise FileNotFoundError("OptiScaler runtime is not installed; missing bundled OptiScaler.ini")
+
+        shutil.copy2(bundle_ini, managed_ini)
+        self._modify_optiscaler_ini(managed_ini)
+        return paths
+
+    def _read_ini_config(self, ini_path: Path) -> ConfigParser:
+        config = ConfigParser(interpolation=None)
+        config.optionxform = str
+        if ini_path.exists():
+            config.read(ini_path, encoding="utf-8")
+        return config
+
+    def _find_sections_for_key(self, config: ConfigParser, key: str) -> list[str]:
+        return [section for section in config.sections() if config.has_option(section, key)]
+
+    def _resolve_setting_definition(self, setting_id: str) -> dict | None:
+        for definition in QUICK_SETTING_DEFINITIONS:
+            if definition["id"] == setting_id:
+                return definition
+        return None
+
+    def _resolve_setting_section(self, config: ConfigParser, definition: dict) -> str | None:
+        hinted_section = definition.get("section")
+        key = definition["key"]
+
+        if hinted_section and config.has_option(hinted_section, key):
+            return hinted_section
+
+        matching_sections = self._find_sections_for_key(config, key)
+        if len(matching_sections) == 1:
+            return matching_sections[0]
+        if hinted_section:
+            return hinted_section
+        if matching_sections:
+            return matching_sections[0]
+        return None
+
+    def _extract_quick_settings(self, ini_path: Path) -> dict:
+        config = self._read_ini_config(ini_path)
+        settings = {}
+        for definition in QUICK_SETTING_DEFINITIONS:
+            resolved_section = self._resolve_setting_section(config, definition)
+            value = definition["default"]
+            if resolved_section and config.has_option(resolved_section, definition["key"]):
+                value = config.get(resolved_section, definition["key"])
+            settings[definition["id"]] = value
+        return settings
+
+    def _replace_ini_values(self, ini_path: Path, resolved_updates: list[tuple[str, str, str]]) -> None:
+        lines = []
+        if ini_path.exists():
+            with open(ini_path, "r", encoding="utf-8", errors="replace") as ini_file:
+                lines = ini_file.readlines()
+
+        for section, key, value in resolved_updates:
+            section_pattern = re.compile(rf"^\s*\[{re.escape(section)}]\s*$")
+            any_section_pattern = re.compile(r"^\s*\[.*]\s*$")
+            key_pattern = re.compile(rf"^\s*{re.escape(key)}\s*=.*$")
+
+            in_target_section = False
+            section_found = False
+            inserted_or_updated = False
+            insert_index = len(lines)
+
+            for index, line in enumerate(lines):
+                if section_pattern.match(line):
+                    in_target_section = True
+                    section_found = True
+                    insert_index = index + 1
+                    continue
+
+                if in_target_section and any_section_pattern.match(line):
+                    insert_index = index
+                    break
+
+                if in_target_section:
+                    if key_pattern.match(line):
+                        newline = "\n" if line.endswith("\n") else ""
+                        lines[index] = f"{key}={value}{newline}"
+                        inserted_or_updated = True
+                        break
+                    insert_index = index + 1
+
+            if inserted_or_updated:
+                continue
+
+            if not section_found:
+                if lines and lines[-1].strip():
+                    lines.append("\n")
+                lines.append(f"[{section}]\n")
+                lines.append(f"{key}={value}\n")
+                continue
+
+            lines[insert_index:insert_index] = [f"{key}={value}\n"]
+
+        with open(ini_path, "w", encoding="utf-8") as ini_file:
+            ini_file.writelines(lines)
+
     def _disable_hq_font_auto(self, ini_file: Path) -> bool:
         try:
             if not ini_file.exists():
@@ -494,40 +702,111 @@ class Plugin:
 
         return {"status": "success", "message": "\n".join(cleanup_messages)}
 
-    async def list_installed_games(self) -> dict:
+    async def get_game_config(self, appid: str) -> dict:
         try:
-            games = []
-            for library_path in self._steam_library_paths():
-                steamapps_path = library_path / "steamapps"
-                if not steamapps_path.exists():
-                    continue
+            paths = self._ensure_managed_ini(str(appid))
+            manifest = self._parse_manifest_env(paths["manifest_path"])
+            installed_game = self._find_installed_game(str(appid))
+            game_name = installed_game[0]["name"] if installed_game else str(appid)
 
-                for appmanifest in steamapps_path.glob("appmanifest_*.acf"):
-                    game_info = {"appid": "", "name": ""}
-                    try:
-                        with open(appmanifest, "r", encoding="utf-8", errors="replace") as file:
-                            for line in file:
-                                if '"appid"' in line:
-                                    game_info["appid"] = line.split('"appid"', 1)[1].strip().strip('"')
-                                if '"name"' in line:
-                                    game_info["name"] = line.split('"name"', 1)[1].strip().strip('"')
-                    except Exception as exc:
-                        decky.logger.error(f"Skipping {appmanifest}: {exc}")
-
-                    if game_info["appid"] and game_info["name"]:
-                        games.append(game_info)
-
-            filtered_games = [
-                g
-                for g in games
-                if "Proton" not in g["name"] and "Steam Linux Runtime" not in g["name"]
-            ]
+            source_ini = paths["live_ini"] if paths["live_ini"].exists() else paths["managed_ini"]
+            with open(source_ini, "r", encoding="utf-8", errors="replace") as ini_file:
+                raw_ini = ini_file.read()
 
-            deduped = {}
-            for game in filtered_games:
-                deduped[str(game["appid"])] = game
+            preferred_proxy = manifest.get("PREFERRED_PROXY") or manifest.get("MANAGED_PROXY") or "winmm"
+            proxy = self._normalize_proxy(preferred_proxy)
+            settings = self._extract_quick_settings(source_ini)
 
-            return {"status": "success", "games": list(deduped.values())}
+            return {
+                "status": "success",
+                "appid": str(appid),
+                "name": game_name,
+                "proxy": proxy,
+                "settings": settings,
+                "raw_ini": raw_ini,
+                "managed_exists": paths["managed_ini"].exists(),
+                "live_available": paths["live_ini"].exists(),
+                "paths": {
+                    "compatdata": str(paths["compatdata_dir"]),
+                    "managed_root": str(paths["managed_root"]),
+                    "managed_ini": str(paths["managed_ini"]),
+                    "system32": str(paths["system32"]),
+                    "live_ini": str(paths["live_ini"]),
+                },
+            }
+        except Exception as exc:
+            decky.logger.error(f"get_game_config failed for {appid}: {exc}")
+            return {"status": "error", "message": str(exc)}
+
+    async def save_game_config(
+        self,
+        appid: str,
+        settings: dict | None = None,
+        proxy: str | None = None,
+        apply_live: bool = False,
+        raw_ini: str | None = None,
+    ) -> dict:
+        try:
+            paths = self._ensure_managed_ini(str(appid))
+            manifest_updates = {}
+            if proxy:
+                manifest_updates["PREFERRED_PROXY"] = self._normalize_proxy(proxy)
+            if manifest_updates:
+                self._write_manifest_env(paths["manifest_path"], manifest_updates)
+
+            managed_ini = paths["managed_ini"]
+            resolved_updates = []
+            settings = settings or {}
+
+            if raw_ini is not None:
+                with open(managed_ini, "w", encoding="utf-8") as ini_file:
+                    ini_file.write(raw_ini)
+            elif settings:
+                config = self._read_ini_config(managed_ini)
+                for setting_id, value in settings.items():
+                    definition = self._resolve_setting_definition(setting_id)
+                    if not definition:
+                        continue
+                    section = self._resolve_setting_section(config, definition) or definition.get("section") or "OptiScaler"
+                    resolved_updates.append((section, definition["key"], str(value)))
+                if resolved_updates:
+                    self._replace_ini_values(managed_ini, resolved_updates)
+
+            self._disable_hq_font_auto(managed_ini)
+
+            live_applied = False
+            if apply_live and paths["live_ini"].exists():
+                if raw_ini is not None:
+                    with open(paths["live_ini"], "w", encoding="utf-8") as ini_file:
+                        ini_file.write(raw_ini)
+                elif resolved_updates:
+                    self._replace_ini_values(paths["live_ini"], resolved_updates)
+                self._disable_hq_font_auto(paths["live_ini"])
+                live_applied = True
+
+            message = "Saved OptiScaler config"
+            if apply_live:
+                message = (
+                    "Saved OptiScaler config and applied it to the running game's live prefix copy"
+                    if live_applied
+                    else "Saved OptiScaler config, but no live prefix copy was available to update"
+                )
+
+            return {
+                "status": "success",
+                "message": message,
+                "live_applied": live_applied,
+                "proxy": self._parse_manifest_env(paths["manifest_path"]).get("PREFERRED_PROXY", proxy or "winmm"),
+            }
+        except Exception as exc:
+            decky.logger.error(f"save_game_config failed for {appid}: {exc}")
+            return {"status": "error", "message": str(exc)}
+
+    async def list_installed_games(self) -> dict:
+        try:
+            games = self._find_installed_game()
+            sanitized_games = [{"appid": game["appid"], "name": game["name"]} for game in games]
+            return {"status": "success", "games": sanitized_games}
         except Exception as exc:
             decky.logger.error(str(exc))
             return {"status": "error", "message": str(exc)}