refactor: offload configuration to lsfg-vk

1 files changed, 105 insertions, 271 deletions

diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py
index e39be54..4ae1f4c 100644
--- a/py_modules/lsfg_vk/config_schema.py
+++ b/py_modules/lsfg_vk/config_schema.py
@@ -1,49 +1,37 @@
+"""Small adapter for the upstream lsfg-vk v2 configuration format."""
+
 import json
-import re
-import shlex
 import sys
 import tomllib
-from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, TypedDict, Union, cast
+from typing import Any, Dict, TypedDict, cast
 
 sys.path.insert(0, str(Path(__file__).parent.parent.parent))
-from shared_config import CONFIG_SCHEMA_DEF, ConfigFieldType, get_defaults
-from .config_schema_generated import ConfigurationData, get_script_parsing_logic
-
-
-@dataclass
-class ConfigField:
-    name: str
-    field_type: ConfigFieldType
-    default: Union[bool, int, float, str]
-    description: str
-
 
-CONFIG_SCHEMA: Dict[str, ConfigField] = {
-    name: ConfigField(
-        name=definition["name"],
-        field_type=ConfigFieldType(definition["fieldType"]),
-        default=definition["default"],
-        description=definition["description"],
-    )
-    for name, definition in CONFIG_SCHEMA_DEF.items()
-}
-
-SCRIPT_ONLY_FIELDS = {
-    name
-    for name, definition in CONFIG_SCHEMA_DEF.items()
-    if definition["location"] == "script"
-}
-DEFAULT_PROFILE_NAME = "decky-lsfg-vk"
+DEFAULT_PROFILE_NAME = "default"
+ConfigurationData = Dict[str, Any]
 
 
 class ProfileData(TypedDict):
+    # Internal compatibility field; the public API no longer exposes a
+    # currently selected profile.
     current_profile: str
     profiles: Dict[str, Dict[str, Any]]
     global_config: Dict[str, Any]
 
 
+PROFILE_DEFAULTS: Dict[str, Any] = {
+    "active_in": [],
+    "pacing_mode": "vsync",
+    "multiplier": 2,
+    "flow_scale": 1.0,
+    "performance_mode": False,
+    "override_present_mode": True,
+    "preserve_swapchain_image_count": False,
+}
+GLOBAL_DEFAULTS: Dict[str, Any] = {"dll": "", "no_fp16": False}
+
+
 def _toml_value(value: Any) -> str:
     if isinstance(value, bool):
         return str(value).lower()
@@ -54,41 +42,52 @@ def _toml_value(value: Any) -> str:
     return str(value)
 
 
+def _normalize_active_in(value: Any) -> list[str]:
+    if value in (None, ""):
+        return []
+    if isinstance(value, str):
+        return [value]
+    if not isinstance(value, list):
+        raise ValueError("active_in must be a string or list of strings")
+    return [str(item) for item in value if str(item)]
+
+
 class ConfigurationManager:
     @staticmethod
-    def get_defaults() -> ConfigurationData:
-        return cast(ConfigurationData, dict(get_defaults()))
+    def get_defaults() -> Dict[str, Any]:
+        return {**GLOBAL_DEFAULTS, **PROFILE_DEFAULTS}
 
     @staticmethod
     def get_field_names() -> list[str]:
-        return list(CONFIG_SCHEMA)
+        return list(ConfigurationManager.get_defaults())
 
     @staticmethod
-    def get_field_types() -> Dict[str, ConfigFieldType]:
-        return {name: field.field_type for name, field in CONFIG_SCHEMA.items()}
+    def get_field_types() -> Dict[str, str]:
+        return {
+            "dll": "string", "no_fp16": "boolean", "active_in": "array",
+            "pacing_mode": "string", "multiplier": "integer", "flow_scale": "float",
+            "performance_mode": "boolean", "override_present_mode": "boolean",
+            "preserve_swapchain_image_count": "boolean",
+        }
 
     @staticmethod
-    def validate_config(config: Dict[str, Any]) -> ConfigurationData:
-        validated: Dict[str, Any] = {}
-        for name, field in CONFIG_SCHEMA.items():
-            value = config.get(name, field.default)
-            if field.field_type == ConfigFieldType.BOOLEAN:
-                value = value.lower() in {"true", "1", "yes", "on"} if isinstance(value, str) else bool(value)
-            elif field.field_type == ConfigFieldType.INTEGER:
-                value = int(value)
-            elif field.field_type == ConfigFieldType.FLOAT:
-                value = float(value)
-            else:
-                value = str(value)
-            validated[name] = value
-
-        if validated["multiplier"] < 1:
+    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()
+        if result["pacing_mode"] != "vsync":
+            raise ValueError("pacing_mode must be vsync")
+        result["multiplier"] = int(result["multiplier"])
+        if result["multiplier"] < 1:
             raise ValueError("multiplier must be 1 or greater")
-        if not 0.25 <= validated["flow_scale"] <= 1.0:
+        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")
-        if validated["experimental_present_mode"] not in {"fifo", "mailbox"}:
-            raise ValueError("experimental_present_mode must be fifo or mailbox")
-        return cast(ConfigurationData, validated)
+        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 "")
+        return result
 
     @staticmethod
     def _migrate_dll_path(value: Any) -> str:
@@ -102,246 +101,81 @@ class ConfigurationManager:
 
     @staticmethod
     def _config_from_profile(profile: Dict[str, Any], global_config: Dict[str, Any]) -> Dict[str, Any]:
-        config: Dict[str, Any] = dict(ConfigurationManager.get_defaults())
-        for field in ("multiplier", "flow_scale", "performance_mode"):
-            if field in profile:
-                config[field] = profile[field]
-        config["experimental_present_mode"] = "fifo" if bool(profile.get("override_present_mode", True)) else "mailbox"
-        config["dll"] = global_config.get("dll", "")
-        config["no_fp16"] = global_config.get("no_fp16", False)
-        for field in ("active_in", "pacing", "preserve_swapchain_image_count"):
-            if field in profile:
-                config[field] = profile[field]
-        return {**config, **ConfigurationManager.validate_config(config)}
-
-    @staticmethod
-    def generate_toml_content(config: ConfigurationData) -> str:
-        profile_data: ProfileData = {
+        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(config: Dict[str, Any]) -> str:
+        data: ProfileData = {
             "current_profile": DEFAULT_PROFILE_NAME,
             "profiles": {DEFAULT_PROFILE_NAME: dict(config)},
-            "global_config": {
-                "dll": config.get("dll", ""),
-                "no_fp16": config.get("no_fp16", False),
-            },
+            "global_config": {"dll": config.get("dll", ""), "no_fp16": config.get("no_fp16", False)},
         }
-        return ConfigurationManager.generate_toml_content_multi_profile(profile_data)
+        return ConfigurationManager.generate_toml_content_multi_profile(data)
 
     @staticmethod
     def generate_toml_content_multi_profile(profile_data: ProfileData) -> str:
-        global_config = profile_data["global_config"]
+        global_config = {**GLOBAL_DEFAULTS, **profile_data.get("global_config", {})}
         lines = ["version = 2", "", "[global]"]
-        dll = ConfigurationManager._migrate_dll_path(global_config.get("dll", ""))
+        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)))}")
-        if global_config.get("log_level"):
-            lines.append(f"log_level = {_toml_value(global_config['log_level'])}")
-        if global_config.get("log_file"):
-            lines.append(f"log_file = {_toml_value(global_config['log_file'])}")
-
-        profiles = sorted(
-            profile_data["profiles"].items(),
-            key=lambda item: (item[0] != DEFAULT_PROFILE_NAME, item[0]),
-        )
-        for profile_name, raw_config in profiles:
-            config = ConfigurationManager.validate_config(raw_config)
-            lines.extend(["", "[[profile]]", f"name = {_toml_value(profile_name)}"])
-            active_in = raw_config.get("active_in")
-            if active_in not in (None, "", []):
-                lines.append(f"active_in = {_toml_value(active_in)}")
-            lines.extend(
-                [
-                    f"multiplier = {config['multiplier']}",
-                    f"flow_scale = {config['flow_scale']}",
-                    f"performance_mode = {_toml_value(config['performance_mode'])}",
-                    f"pacing = {_toml_value(raw_config.get('pacing', 'vsync'))}",
-                    f"override_present_mode = {_toml_value(config['experimental_present_mode'] == 'fifo')}",
-                    f"preserve_swapchain_image_count = {_toml_value(bool(raw_config.get('preserve_swapchain_image_count', False)))}",
-                ]
-            )
+        profiles = sorted(profile_data["profiles"].items(), key=lambda item: (item[0] != DEFAULT_PROFILE_NAME, item[0]))
+        for name, raw in profiles:
+            config = ConfigurationManager.validate_config({**raw, **global_config})
+            lines.extend(["", "[[profile]]", f"name = {_toml_value(name)}"])
+            if config["active_in"]:
+                lines.append(f"active_in = {_toml_value(config['active_in'])}")
+            lines.extend([
+                f"pacing_mode = {_toml_value(config['pacing_mode'])}",
+                f"multiplier = {config['multiplier']}",
+                f"flow_scale = {config['flow_scale']}",
+                f"performance_mode = {_toml_value(config['performance_mode'])}",
+                f"override_present_mode = {_toml_value(config['override_present_mode'])}",
+                f"preserve_swapchain_image_count = {_toml_value(config['preserve_swapchain_image_count'])}",
+            ])
         return "\n".join(lines) + "\n"
 
     @staticmethod
-    def _profile_data_from_v1(data: Dict[str, Any]) -> ProfileData:
-        old_global = dict(data.get("global", {}))
-        global_config: Dict[str, Any] = {
-            "dll": ConfigurationManager._migrate_dll_path(old_global.get("dll", "")),
-            "no_fp16": bool(old_global.get("no_fp16", False)),
-        }
-        profiles: Dict[str, Dict[str, Any]] = {}
-        for game in data.get("game", []):
-            profile_name = str(game.get("exe", DEFAULT_PROFILE_NAME))
-            config: Dict[str, Any] = dict(ConfigurationManager.get_defaults())
-            for field in ("multiplier", "flow_scale", "performance_mode", "experimental_present_mode"):
-                if field in game:
-                    config[field] = game[field]
-            config["dll"] = global_config["dll"]
-            config["no_fp16"] = global_config["no_fp16"]
-            profiles[profile_name] = dict(ConfigurationManager.validate_config(config))
-
-        if not profiles:
-            profiles[DEFAULT_PROFILE_NAME] = dict(ConfigurationManager.get_defaults())
-
-        current_profile = str(old_global.get("current_profile", DEFAULT_PROFILE_NAME))
-        if current_profile not in profiles:
-            current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles))
-        return ProfileData(
-            current_profile=current_profile,
-            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
-
-    @staticmethod
     def parse_toml_content_multi_profile(content: str) -> ProfileData:
         data = tomllib.loads(content)
         version = data.get("version")
-        if version == 1:
-            return ConfigurationManager._profile_data_from_v1(data)
-        if version != 2:
+        if version not in (1, 2):
             raise ValueError("unsupported lsfg-vk configuration version")
-
         raw_global = dict(data.get("global", {}))
-        global_config: Dict[str, Any] = {
+        global_config = {
             "dll": ConfigurationManager._migrate_dll_path(raw_global.get("dll", "")),
             "no_fp16": not bool(raw_global.get("allow_fp16", True)),
         }
-        for field in ("log_level", "log_file"):
-            if field in raw_global:
-                global_config[field] = raw_global[field]
-
         profiles: Dict[str, Dict[str, Any]] = {}
-        for profile in data.get("profile", []):
-            profile_name = str(profile.get("name", DEFAULT_PROFILE_NAME))
-            profiles[profile_name] = ConfigurationManager._config_from_profile(profile, global_config)
-
+        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", DEFAULT_PROFILE_NAME))
+            profiles[name] = ConfigurationManager._config_from_profile(profile, global_config)
         if not profiles:
-            default = dict(ConfigurationManager.get_defaults())
-            default["dll"] = global_config["dll"]
-            default["no_fp16"] = global_config["no_fp16"]
-            profiles[DEFAULT_PROFILE_NAME] = default
-
-        current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles))
-        return ProfileData(
-            current_profile=current_profile,
-            profiles=profiles,
-            global_config=global_config,
-        )
-
-    @staticmethod
-    def parse_toml_content(content: str) -> ConfigurationData:
-        profile_data = ConfigurationManager.parse_toml_content_multi_profile(content)
-        return cast(ConfigurationData, profile_data["profiles"][profile_data["current_profile"]])
-
-    @staticmethod
-    def parse_script_content(script_content: str) -> Dict[str, Union[bool, int, str]]:
-        return get_script_parsing_logic()(script_content.splitlines())
-
-    @staticmethod
-    def parse_profile_selection(script_content: str) -> str | None:
-        selected = None
-        for line in script_content.splitlines():
-            try:
-                tokens = shlex.split(line)
-            except ValueError:
-                continue
-            if len(tokens) != 2 or tokens[0] != "export" or "=" not in tokens[1]:
-                continue
-            key, value = tokens[1].split("=", 1)
-            if key in {"LSFGVK_PROFILE", "LSFG_PROCESS"} and value:
-                selected = value
-        return selected
-
-    @staticmethod
-    def merge_config_with_script(
-        toml_config: Dict[str, Any],
-        script_values: Dict[str, Union[bool, int, str]],
-    ) -> Dict[str, Any]:
-        merged = dict(toml_config)
-        for field in SCRIPT_ONLY_FIELDS:
-            if field in script_values:
-                merged[field] = script_values[field]
-        return merged
+            profiles[DEFAULT_PROFILE_NAME] = ConfigurationManager.validate_config(global_config)
+        elif DEFAULT_PROFILE_NAME not in profiles:
+            source = profiles.get("decky-lsfg-vk", next(iter(profiles.values())))
+            profiles[DEFAULT_PROFILE_NAME] = {**source, "active_in": []}
+            if profiles.get("decky-lsfg-vk", {}).get("active_in", []) == []:
+                profiles.pop("decky-lsfg-vk", None)
+        return {"current_profile": DEFAULT_PROFILE_NAME, "profiles": profiles, "global_config": global_config}
 
     @staticmethod
-    def normalize_profile_name(profile_name: str) -> str:
-        return re.sub(r"\s+", "-", profile_name.strip()).strip("-")
-
-    @staticmethod
-    def validate_profile_name(profile_name: str) -> bool:
-        normalized = ConfigurationManager.normalize_profile_name(profile_name)
-        invalid = '\t\n\r\'"\\/$|&;()<>{}[]' + "`" + '*?'
-        return (
-            bool(normalized)
-            and not any(character in invalid for character in normalized)
-            and normalized.lower() not in {"global", "profile"}
-        )
-
-    @staticmethod
-    def create_profile(profile_data: ProfileData, profile_name: str, source_profile: str = None) -> ProfileData:
-        if not ConfigurationManager.validate_profile_name(profile_name):
-            raise ValueError(f"Invalid profile name: {profile_name}")
-        normalized = ConfigurationManager.normalize_profile_name(profile_name)
-        if normalized in profile_data["profiles"]:
-            raise ValueError(f"Profile '{normalized}' already exists")
-        source = source_profile if source_profile in profile_data["profiles"] else profile_data["current_profile"]
-        profiles = dict(profile_data["profiles"])
-        profiles[normalized] = dict(profiles[source])
-        return ProfileData(
-            current_profile=profile_data["current_profile"],
-            profiles=profiles,
-            global_config=dict(profile_data["global_config"]),
-        )
-
-    @staticmethod
-    def delete_profile(profile_data: ProfileData, profile_name: str) -> ProfileData:
-        if profile_name == DEFAULT_PROFILE_NAME:
-            raise ValueError("Cannot delete the default profile")
-        if profile_name not in profile_data["profiles"]:
-            raise ValueError(f"Profile '{profile_name}' does not exist")
-        profiles = dict(profile_data["profiles"])
-        del profiles[profile_name]
-        current_profile = profile_data["current_profile"]
-        if current_profile == profile_name:
-            current_profile = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in profiles else next(iter(profiles))
-        return ProfileData(
-            current_profile=current_profile,
-            profiles=profiles,
-            global_config=dict(profile_data["global_config"]),
-        )
-
-    @staticmethod
-    def rename_profile(profile_data: ProfileData, old_name: str, new_name: str) -> ProfileData:
-        if old_name == DEFAULT_PROFILE_NAME:
-            raise ValueError("Cannot rename the default profile")
-        if old_name not in profile_data["profiles"] or not ConfigurationManager.validate_profile_name(new_name):
-            raise ValueError("Invalid profile rename")
-        normalized = ConfigurationManager.normalize_profile_name(new_name)
-        if normalized in profile_data["profiles"]:
-            raise ValueError(f"Profile '{normalized}' already exists")
-        profiles = {
-            normalized if name == old_name else name: value
-            for name, value in profile_data["profiles"].items()
-        }
-        current_profile = normalized if profile_data["current_profile"] == old_name else profile_data["current_profile"]
-        return ProfileData(
-            current_profile=current_profile,
-            profiles=profiles,
-            global_config=dict(profile_data["global_config"]),
-        )
+    def is_legacy_v1(content: str) -> bool:
+        try:
+            return tomllib.loads(content).get("version") == 1
+        except tomllib.TOMLDecodeError:
+            return False
 
     @staticmethod
-    def set_current_profile(profile_data: ProfileData, profile_name: str) -> ProfileData:
-        if profile_name not in profile_data["profiles"]:
-            raise ValueError(f"Profile '{profile_name}' does not exist")
-        return ProfileData(
-            current_profile=profile_name,
-            profiles=dict(profile_data["profiles"]),
-            global_config=dict(profile_data["global_config"]),
-        )
+    def parse_toml_content(content: str) -> Dict[str, Any]:
+        data = ConfigurationManager.parse_toml_content_multi_profile(content)
+        return cast(Dict[str, Any], data["profiles"][DEFAULT_PROFILE_NAME])