1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
import re
from typing import Any, Dict
from .base_service import BaseService
from .config_schema import ConfigurationManager, ProfileData
from .runtime_service import RuntimeService
class ConfigurationService(BaseService):
"""Controller-facing adapter over upstream lsfg-vk profiles."""
def __init__(self, logger=None, runtime_service: RuntimeService = None):
super().__init__(logger)
self.runtime_service = runtime_service or RuntimeService(logger=self.log)
def _default_data(self) -> ProfileData:
return {"profiles": {}, "global_config": {"dll": "", "no_fp16": False}}
def _get_profile_data(self) -> ProfileData:
if not self.config_file_path.exists():
return self._default_data()
return ConfigurationManager.parse_toml_content_multi_profile(self.config_file_path.read_text(encoding="utf-8"))
def _save_profile_data(self, data: ProfileData) -> None:
content = ConfigurationManager.generate_toml_content_multi_profile(data)
self.runtime_service.validate_config_content(content)
self._write_file(self.config_file_path, content, 0o644)
@staticmethod
def _profile_name(data: ProfileData, appid: str, game_name: str) -> str:
name = str(game_name).strip()
if not name:
raise ValueError("game name is required")
existing = data["profiles"].get(name)
if existing is not None and str(appid) not in existing.get("active_in", []):
name = f"{name} ({appid})"
return name
@staticmethod
def _profile_for_appid(data: ProfileData, appid: str):
return next(
(
(name, profile)
for name, profile in data["profiles"].items()
if str(appid) in profile.get("active_in", [])
),
(None, None),
)
@staticmethod
def _public_config(config: Dict[str, Any]) -> Dict[str, Any]:
return ConfigurationManager.validate_config(config)
def get_game_configs(self) -> Dict[str, Any]:
try:
data = self._get_profile_data()
games = []
for name, raw in data["profiles"].items():
active_in = raw.get("active_in", [])
if len(active_in) != 1 or not re.fullmatch(r"-?[0-9]+", str(active_in[0])):
continue
games.append({"appid": str(active_in[0]), "profile": name, "config": self._public_config(raw)})
return self._success_response(dict, global_config=dict(data["global_config"]), games=games)
except Exception as error:
self.log.error(f"Error reading game configs: {error}")
return self._error_response(dict, str(error), games=[])
def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]) -> Dict[str, Any]:
try:
data = self._get_profile_data()
old_name, _ = self._profile_for_appid(data, appid)
name = self._profile_name(data, appid, game_name)
merged_config = {**data["global_config"], **config}
if not config.get("dll"):
merged_config["dll"] = data["global_config"].get("dll", "")
validated = self._public_config(merged_config)
validated["active_in"] = [str(appid)]
data["global_config"] = {
"dll": validated["dll"],
"no_fp16": validated["no_fp16"],
}
if old_name and old_name != name:
data["profiles"].pop(old_name, None)
data["profiles"][name] = validated
self._save_profile_data(data)
return self._success_response(dict, appid=str(appid), config=validated)
except Exception as error:
return self._error_response(dict, str(error), appid=str(appid), config=None)
def reset_game_config(self, appid: str) -> Dict[str, Any]:
try:
data = self._get_profile_data()
name, _ = self._profile_for_appid(data, appid)
if name:
data["profiles"].pop(name, None)
self._save_profile_data(data)
return self._success_response(dict, appid=str(appid), exists=False)
except Exception as error:
return self._error_response(dict, str(error), appid=str(appid), config=None)
def reset_all_game_configs(self) -> Dict[str, Any]:
try:
data = self._get_profile_data()
data["profiles"] = {}
self._save_profile_data(data)
return self._success_response(dict, global_config=dict(data["global_config"]), games=[])
except Exception as error:
return self._error_response(dict, str(error), games=[])
|
