-
Notifications
You must be signed in to change notification settings - Fork 40
Expand file tree
/
Copy pathserver_config.py
More file actions
218 lines (171 loc) · 7.18 KB
/
Copy pathserver_config.py
File metadata and controls
218 lines (171 loc) · 7.18 KB
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
"""
ServerConfig - Centralized configuration management for OpenArc.
This module handles all configuration file operations without any CLI/presentation logic.
"""
import json
import os
from pathlib import Path
from typing import Optional, Dict, Any, List, cast
from ..utils import get_config_file_path
class ServerConfig:
"""Manages OpenArc server and model configurations."""
def __init__(self, config_file: Optional[Path] = None):
"""
Initialize ServerConfig with a config file path.
Args:
config_file: Path to the config file. If None, OPENARC_CONFIG_FILE env var when set,
otherwise defaults to openarc_config.json in project root.
"""
self.config_file = get_config_file_path() if config_file is None else config_file
def load_config(self) -> Dict[str, Any]:
"""
Load full configuration from JSON config file.
Returns:
Configuration dictionary, or empty dict if file doesn't exist or is invalid.
"""
if not self.config_file.exists():
return {}
try:
with open(self.config_file, "r") as f:
config = json.load(f)
return config if config else {}
except (json.JSONDecodeError, FileNotFoundError):
return {}
def save_config(self, config: Dict[str, Any]) -> None:
"""
Save configuration to JSON config file. Does nothing if an on-disk
config file exists and its contents are identical to the provided
config, allowing the config to live on a read-only filesystem.
Args:
config: Configuration dictionary to save.
"""
if self.config_file.exists():
try:
with open(self.config_file, "r") as f:
existing_config = json.load(f)
if existing_config == config:
return # No changes, skip writing
except:
pass # Overwrite if the existing file is invalid or missing
with open(self.config_file, "w") as f:
json.dump(config, f, indent=2)
def save_server_config(self, host: str, port: int) -> Path:
"""
Save server configuration.
Args:
host: Server host address.
port: Server port number.
Returns:
Path to the saved config file.
"""
config = self.load_config()
config.update({
"server": {
"host": host,
"port": port
},
"created_by": "openarc-cli",
"version": "1.0"
})
self.save_config(config)
return self.config_file
def load_server_config(self) -> Dict[str, Any]:
"""
Load server configuration.
Returns:
Server configuration dict with 'host' and 'port' keys.
Returns defaults if not configured.
"""
config = self.load_config()
if config and "server" in config:
return config["server"]
return {"host": "localhost", "port": 8000}
def save_model_config(self, model_name: str, model_config: Dict[str, Any]) -> None:
"""
Save model configuration.
Args:
model_name: Name of the model.
model_config: Model configuration dictionary.
"""
config = self.load_config()
if "models" not in config:
config["models"] = {}
config["models"][model_name] = model_config
self.save_config(config)
def get_model_config(self, model_name: str) -> Optional[Dict[str, Any]]:
"""
Get model configuration by name.
Args:
model_name: Name of the model.
Returns:
Model configuration dict, or None if not found. Relative paths are resolved
against the config file's directory, allowing for configs to be packaged with models.
"""
config = self.load_config()
models = cast(Dict[str, Dict[str, Any]], config.get("models", {}))
model = models.get(model_name)
return self._resolve_model_paths(model) if model else None
def get_all_models(self) -> Dict[str, Dict[str, Any]]:
"""
Get all model configurations.
Returns:
Dictionary mapping model names to their configurations. Relative paths are resolved
against the config file's directory, allowing for configs to be packaged with models.
"""
config = self.load_config()
models = cast(Dict[str, Dict[str, Any]], config.get("models", {}))
return {name: self._resolve_model_paths(cfg) for name, cfg in models.items()}
def _resolve_model_paths(self, model_config: Dict[str, Any]) -> Dict[str, Any]:
"""Return a copy of model_config with relative model_path, draft_model_path,
and cache_dir made absolute by joining them onto the config file's directory."""
resolved = dict(model_config)
path = resolved.get("model_path")
if path and not Path(path).is_absolute():
resolved["model_path"] = str((self.config_file.parent / path).resolve())
draft_model_path = resolved.get("draft_model_path")
if draft_model_path and not Path(draft_model_path).is_absolute():
resolved["draft_model_path"] = str((self.config_file.parent / draft_model_path).resolve())
cache_dir = resolved.get("cache_dir")
if cache_dir and not Path(cache_dir).is_absolute():
resolved["cache_dir"] = str((self.config_file.parent / cache_dir).resolve())
return resolved
def remove_model_config(self, model_name: str) -> bool:
"""
Remove model configuration by name.
Args:
model_name: Name of the model to remove.
Returns:
True if model was removed, False if model was not found.
"""
config = self.load_config()
models = config.get("models", {})
if model_name not in models:
return False
del models[model_name]
config["models"] = models
self.save_config(config)
return True
def model_exists(self, model_name: str) -> bool:
"""
Check if a model configuration exists.
Args:
model_name: Name of the model.
Returns:
True if model exists, False otherwise.
"""
return self.get_model_config(model_name) is not None
def get_model_names(self) -> List[str]:
"""
Get list of all configured model names.
Returns:
List of model names.
"""
return list(self.get_all_models().keys())
def get_base_url(self) -> str:
"""
Get the base URL for the OpenArc server.
Returns:
Base URL string (e.g., 'http://localhost:8000').
"""
server_config = self.load_server_config()
return f"http://{server_config['host']}:{server_config['port']}"