Coverage for core / platform / config.py: 95.2%

1""" 

2Platform Config — Unified 3-layer configuration resolution. 

3 

4Resolves settings with precedence: environment variables > DB row > defaults. 

5Generalizes the compute_config.py pattern for use across all OS subsystems. 

6 

7Design decisions: 

8- Namespace-based: each subsystem defines its own config class 

9- TTL cache (30s default) — same pattern as compute_config.py 

10- Change notifications: on_change(key, callback) 

11- Typed: each key has a converter (int, float, str, bool, json) 

12- Falls back gracefully when DB unavailable 

13- Thread-safe via threading.Lock 

14 

15Usage: 

16 class DisplayConfig(PlatformConfig): 

17 _namespace = 'display' 

18 _defaults = {'scale': 1.0, 'brightness': 1.0} 

19 _env_map = {'scale': ('HART_DISPLAY_SCALE', float)} 

20 

21 display = DisplayConfig() 

22 scale = display.get('scale') # env > DB > default 

23 display.set('scale', 1.5) # persists + notifies 

24 display.on_change('scale', fn) # subscribe to changes 

25""" 

26 

27import json 

28import logging 

29import os 

30import threading 

31import time 

32from typing import Any, Callable, Dict, List, Optional, Tuple 

33 

34logger = logging.getLogger('hevolve.platform') 

35 

36 

37# ═══════════════════════════════════════════════════════════════ 

38# Type Converters 

39# ═══════════════════════════════════════════════════════════════ 

40 

41def _convert_bool(val: Any) -> bool: 

42 """Convert string/int/bool to bool.""" 

43 if isinstance(val, bool): 

44 return val 

45 if isinstance(val, str): 

46 return val.lower() in ('true', '1', 'yes', 'on') 

47 return bool(val) 

48 

49 

50def _convert_json(val: Any) -> Any: 

51 """Convert JSON string to Python object.""" 

52 if isinstance(val, str): 

53 return json.loads(val) 

54 return val 

55 

56 

57CONVERTERS = { 

58 int: int, 

59 float: float, 

60 str: str, 

61 bool: _convert_bool, 

62 'json': _convert_json, 

63} 

64 

65 

66# ═══════════════════════════════════════════════════════════════ 

67# Platform Config 

68# ═══════════════════════════════════════════════════════════════ 

69 

70class PlatformConfig: 

71 """3-layer config resolution: env vars > DB > defaults. 

72 

73 Subclass and set _namespace, _defaults, _env_map to create 

74 a config namespace for any subsystem. 

75 

76 Class attributes (set by subclasses): 

77 _namespace: str - Config namespace (e.g., 'display', 'audio') 

78 _defaults: dict - Default values for all keys 

79 _env_map: dict - Maps key → (ENV_VAR_NAME, converter) 

80 _cache_ttl: int - Cache TTL in seconds (default 30) 

81 """ 

82 

83 _namespace: str = '' 

84 _defaults: Dict[str, Any] = {} 

85 _env_map: Dict[str, Tuple[str, type]] = {} 

86 _cache_ttl: int = 30 

87 

88 def __init__(self, db_loader: Optional[Callable] = None): 

89 """Initialize config. 

90 

91 Args: 

92 db_loader: Optional callable(namespace, key) -> value or None. 

93 Used for layer 2 (DB) resolution. If None, DB layer 

94 is skipped (env + defaults only). 

95 """ 

96 self._db_loader = db_loader 

97 self._db_saver: Optional[Callable] = None 

98 self._cache: Dict[str, Any] = {} 

99 self._cache_ts: float = 0.0 

100 self._overrides: Dict[str, Any] = {} # in-memory set() values 

101 self._listeners: Dict[str, List[Callable]] = {} 

102 self._lock = threading.Lock() 

103 

104 def set_db_saver(self, saver: Callable) -> None: 

105 """Set a callable(namespace, key, value) for persisting to DB.""" 

106 self._db_saver = saver 

107 

108 def get(self, key: str, default: Any = None) -> Any: 

109 """Resolve a config value. Precedence: env > override > DB > defaults. 

110 

111 Args: 

112 key: Config key name. 

113 default: Fallback if key not found anywhere (overrides _defaults). 

114 

115 Returns: 

116 Resolved value, type-converted if env_map specifies a converter. 

117 """ 

118 # Layer 1: Environment variable (always highest priority) 

119 if key in self._env_map: 

120 env_name, converter = self._env_map[key] 

121 env_val = os.environ.get(env_name) 

122 if env_val is not None: 

123 try: 

124 conv = CONVERTERS.get(converter, converter) 

125 return conv(env_val) 

126 except (ValueError, TypeError, json.JSONDecodeError): 

127 logger.warning("Bad env value for %s=%s", env_name, env_val) 

128 

129 # Layer 2: In-memory override (from set()) 

130 with self._lock: 

131 if key in self._overrides: 

132 return self._overrides[key] 

133 

134 # Layer 3: DB (cached with TTL) 

135 db_val = self._get_from_db(key) 

136 if db_val is not None: 

137 return db_val 

138 

139 # Layer 4: Defaults 

140 if key in self._defaults: 

141 return self._defaults[key] 

142 

143 return default 

144 

145 def get_all(self) -> Dict[str, Any]: 

146 """Return all config values (resolved).""" 

147 result = {} 

148 for key in self._defaults: 

149 result[key] = self.get(key) 

150 return result 

151 

152 def set(self, key: str, value: Any) -> None: 

153 """Set a config value. Persists to DB if saver is configured. 

154 

155 Also triggers change notifications for registered listeners. 

156 """ 

157 old_val = self.get(key) 

158 

159 with self._lock: 

160 self._overrides[key] = value 

161 

162 # Persist to DB 

163 if self._db_saver: 

164 try: 

165 self._db_saver(self._namespace, key, value) 

166 except Exception as e: 

167 logger.warning("DB save failed for %s.%s: %s", 

168 self._namespace, key, e) 

169 

170 # Invalidate cache 

171 self._invalidate_cache() 

172 

173 # Notify listeners 

174 if old_val != value: 

175 self._notify(key, old_val, value) 

176 

177 def reset(self, key: str) -> None: 

178 """Reset a key to its default value.""" 

179 with self._lock: 

180 self._overrides.pop(key, None) 

181 self._invalidate_cache() 

182 default_val = self._defaults.get(key) 

183 self._notify(key, None, default_val) 

184 

185 def reset_all(self) -> None: 

186 """Reset all overrides.""" 

187 with self._lock: 

188 self._overrides.clear() 

189 self._invalidate_cache() 

190 

191 # ── Change Notifications ────────────────────────────────── 

192 

193 def on_change(self, key: str, callback: Callable) -> None: 

194 """Subscribe to changes for a specific key. 

195 

196 Callback receives (key, old_value, new_value). 

197 """ 

198 with self._lock: 

199 if key not in self._listeners: 

200 self._listeners[key] = [] 

201 self._listeners[key].append(callback) 

202 

203 def off_change(self, key: str, callback: Callable) -> None: 

204 """Unsubscribe from changes.""" 

205 with self._lock: 

206 if key in self._listeners: 

207 try: 

208 self._listeners[key].remove(callback) 

209 except ValueError: 

210 pass 

211 

212 # ── Internal ────────────────────────────────────────────── 

213 

214 def _get_from_db(self, key: str) -> Optional[Any]: 

215 """Load from DB with TTL cache.""" 

216 if not self._db_loader: 

217 return None 

218 

219 now = time.time() 

220 with self._lock: 

221 if now - self._cache_ts < self._cache_ttl and key in self._cache: 

222 return self._cache[key] 

223 

224 try: 

225 val = self._db_loader(self._namespace, key) 

226 if val is not None: 

227 with self._lock: 

228 self._cache[key] = val 

229 self._cache_ts = now 

230 return val 

231 except Exception as e: 

232 logger.debug("DB load failed for %s.%s: %s", 

233 self._namespace, key, e) 

234 return None 

235 

236 def _invalidate_cache(self) -> None: 

237 """Clear the TTL cache.""" 

238 with self._lock: 

239 self._cache.clear() 

240 self._cache_ts = 0.0 

241 

242 def _notify(self, key: str, old_val: Any, new_val: Any) -> None: 

243 """Dispatch change notifications to listeners.""" 

244 with self._lock: 

245 listeners = list(self._listeners.get(key, [])) 

246 for cb in listeners: 

247 try: 

248 cb(key, old_val, new_val) 

249 except Exception as e: 

250 logger.warning("Config listener error for %s.%s: %s", 

251 self._namespace, key, e) 

252 

253 # ── Settings Export / Import (for sync) ───────────────── 

254 

255 def export_settings(self) -> Dict[str, Any]: 

256 """Export all config values as a JSON-serializable dict. 

257 

258 Used for cross-device settings sync via federation. 

259 """ 

260 return { 

261 'namespace': self._namespace, 

262 'values': self.get_all(), 

263 'exported_at': time.time(), 

264 } 

265 

266 def import_settings(self, data: Dict[str, Any]) -> int: 

267 """Import settings from an exported dict. 

268 

269 Returns the number of keys updated. 

270 """ 

271 values = data.get('values', {}) 

272 count = 0 

273 for key, val in values.items(): 

274 if key in self._defaults: 

275 self.set(key, val) 

276 count += 1 

277 return count 

278 

279 @property 

280 def namespace(self) -> str: 

281 """Return the config namespace.""" 

282 return self._namespace 

283 

284 def __repr__(self) -> str: 

285 return f"PlatformConfig(namespace={self._namespace!r})"