Coverage for integrations / agent_engine / theme_service.py: 87.1%
202 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2HART OS Theme Service — OS-wide theme management.
4Manages glass shell + Conky + GTK appearance from a single source of truth.
5Theme presets live as JSON files; the active theme is persisted to disk
6and propagated to all visual layers (LiquidUI CSS vars, Conky Lua, GTK gsettings).
8Agent-driven customization: agents can call apply_theme() or update_custom()
9to change the OS appearance on voice command ("make it darker", "bigger fonts").
10"""
12import json
13import logging
14import os
15import subprocess
16from typing import Dict, List, Optional
18logger = logging.getLogger('hevolve.theme_service')
20# ── Paths ────────────────────────────────────────────────────────
22_DATA_DIR = os.environ.get('HEVOLVE_DATA_DIR', os.environ.get(
23 'HART_DATA_DIR', os.path.join(os.path.dirname(os.path.dirname(
24 os.path.dirname(os.path.abspath(__file__)))), 'agent_data')))
26_THEME_DIR = os.environ.get('HART_THEME_DIR', os.path.join(
27 os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
28 'nixos', 'assets', 'conky-themes'))
30_ACTIVE_THEME_PATH = os.path.join(_DATA_DIR, 'active_theme.json')
31_CUSTOM_OVERRIDES_PATH = os.path.join(_DATA_DIR, 'theme_custom.json')
34class ThemeService:
35 """OS-wide theme management — Glass Shell + Conky + GTK."""
37 # ── Preset Management ────────────────────────────────────────
39 @staticmethod
40 def list_presets() -> List[dict]:
41 """Return all available theme presets."""
42 presets = []
43 if not os.path.isdir(_THEME_DIR):
44 logger.warning("Theme directory not found: %s", _THEME_DIR)
45 return presets
47 for fname in sorted(os.listdir(_THEME_DIR)):
48 if not fname.endswith('.json'):
49 continue
50 path = os.path.join(_THEME_DIR, fname)
51 try:
52 with open(path, 'r', encoding='utf-8') as f:
53 preset = json.load(f)
54 presets.append({
55 'id': preset.get('id', fname.replace('.json', '')),
56 'name': preset.get('name', ''),
57 'description': preset.get('description', ''),
58 'category': preset.get('category', 'dark'),
59 'accent': preset.get('colors', {}).get('accent', ''),
60 'background': preset.get('colors', {}).get('background', ''),
61 })
62 except (json.JSONDecodeError, OSError) as e:
63 logger.warning("Failed to load theme %s: %s", fname, e)
64 return presets
66 @staticmethod
67 def get_preset(theme_id: str) -> Optional[dict]:
68 """Load a full theme preset by ID."""
69 path = os.path.join(_THEME_DIR, f'{theme_id}.json')
70 if not os.path.isfile(path):
71 return None
72 try:
73 with open(path, 'r', encoding='utf-8') as f:
74 return json.load(f)
75 except (json.JSONDecodeError, OSError) as e:
76 logger.warning("Failed to load theme %s: %s", theme_id, e)
77 return None
79 # ── Active Theme ─────────────────────────────────────────────
81 @staticmethod
82 def get_active_theme() -> dict:
83 """Return the currently active theme (preset + any custom overrides).
85 On first call with no active theme, auto-detects hardware and
86 selects appropriate theme (potato for low-end, default for standard+).
87 """
88 theme = ThemeService._load_active_file()
89 if theme:
90 overrides = ThemeService._load_custom_overrides()
91 if overrides:
92 theme = ThemeService._deep_merge(theme, overrides)
93 return theme
95 # Auto-detect on first access (writes active_theme.json)
96 recommended = ThemeService.detect_performance_tier()
97 if recommended:
98 preset = ThemeService.get_preset(recommended)
99 if preset:
100 return preset
102 # Fallback: hart-default
103 default = ThemeService.get_preset('hart-default')
104 if default:
105 return default
107 # Ultimate fallback: minimal inline theme
108 return {
109 'id': 'hart-default',
110 'name': 'HART Default',
111 'category': 'dark',
112 'colors': {
113 'background': '0F0E17', 'accent': '6C63FF',
114 'active': '00e676', 'text': 'e0e0e0',
115 'heading': '6C63FF', 'glass_bg': 'rgba(15,14,23,0.65)',
116 'glass_border': 'rgba(108,99,255,0.15)',
117 'muted': '78909c', 'surface': '1a1a2e',
118 },
119 'font': {'family': 'JetBrains Mono', 'size': 13,
120 'heading_size': 18, 'weight': 400, 'heading_weight': 600},
121 'shell': {'blur_radius': 20, 'saturation': 180,
122 'border_radius': 16, 'panel_opacity': 0.65},
123 'conky': {'heading': '6C63FF', 'active': '00e676',
124 'muted': '78909c', 'default_text': 'b0b0b0'},
125 'gtk_prefer_dark': True,
126 }
128 @staticmethod
129 def apply_theme(theme_id: str) -> dict:
130 """Apply a theme OS-wide. Returns the applied theme or error."""
131 preset = ThemeService.get_preset(theme_id)
132 if not preset:
133 return {'error': f'Unknown theme: {theme_id}'}
135 # 1. Persist active theme file (read by Conky Lua every 5s)
136 try:
137 os.makedirs(os.path.dirname(_ACTIVE_THEME_PATH), exist_ok=True)
138 with open(_ACTIVE_THEME_PATH, 'w', encoding='utf-8') as f:
139 json.dump(preset, f, indent=2)
140 except OSError as e:
141 logger.error("Failed to write active theme: %s", e)
142 return {'error': str(e)}
144 # 2. Clear custom overrides (new preset = fresh start)
145 if os.path.isfile(_CUSTOM_OVERRIDES_PATH):
146 try:
147 os.remove(_CUSTOM_OVERRIDES_PATH)
148 except OSError:
149 pass
151 # 3. Apply GTK theme via gsettings (Linux only, non-blocking)
152 ThemeService._apply_gtk(preset)
154 logger.info("Theme applied: %s", theme_id)
156 # Single notification path: EventBus → WAMP → all subsystems
157 # LiquidUI subscribes to 'theme.changed' on the EventBus
158 try:
159 from core.platform.events import emit_event
160 emit_event('theme.changed', {'theme_id': theme_id, 'preset': preset})
161 except Exception:
162 pass
164 return {'status': 'applied', 'theme_id': theme_id, 'theme': preset}
166 # ── Agent-Driven Customization ───────────────────────────────
168 @staticmethod
169 def update_custom(overrides: dict) -> dict:
170 """Apply partial customization on top of the active theme.
172 Agents use this for voice-driven tweaks:
173 "make fonts bigger" → update_custom({'font': {'size': 16}})
174 "more transparency" → update_custom({'shell': {'panel_opacity': 0.5}})
175 "change accent to red" → update_custom({'colors': {'accent': 'f44336'}})
176 """
177 current = ThemeService._load_custom_overrides() or {}
178 merged = ThemeService._deep_merge(current, overrides)
180 try:
181 os.makedirs(os.path.dirname(_CUSTOM_OVERRIDES_PATH), exist_ok=True)
182 with open(_CUSTOM_OVERRIDES_PATH, 'w', encoding='utf-8') as f:
183 json.dump(merged, f, indent=2)
184 except OSError as e:
185 logger.error("Failed to write custom overrides: %s", e)
186 return {'error': str(e)}
188 # Re-write active theme file with overrides applied
189 base = ThemeService._load_active_file()
190 if base:
191 combined = ThemeService._deep_merge(base, merged)
192 try:
193 with open(_ACTIVE_THEME_PATH, 'w', encoding='utf-8') as f:
194 json.dump(combined, f, indent=2)
195 except OSError:
196 pass
198 try:
199 from core.platform.events import emit_event
200 emit_event('theme.custom_updated', {'overrides': merged})
201 except Exception:
202 pass
204 return {'status': 'customized', 'overrides': merged}
206 @staticmethod
207 def get_font_options() -> List[dict]:
208 """Available font families for customization."""
209 return [
210 {'family': 'JetBrains Mono', 'category': 'monospace'},
211 {'family': 'Inter', 'category': 'sans-serif'},
212 {'family': 'Fira Code', 'category': 'monospace'},
213 {'family': 'IBM Plex Sans', 'category': 'sans-serif'},
214 {'family': 'Roboto', 'category': 'sans-serif'},
215 {'family': 'Source Code Pro', 'category': 'monospace'},
216 {'family': 'Noto Sans', 'category': 'sans-serif'},
217 {'family': 'Ubuntu', 'category': 'sans-serif'},
218 ]
220 # ── Performance Auto-Detection ──────────────────────────────
222 @staticmethod
223 def detect_performance_tier() -> str:
224 """Detect hardware tier and return recommended theme.
226 Returns theme ID: 'potato' for OBSERVER/EMBEDDED,
227 'minimal' for LITE, None for STANDARD+.
228 """
229 try:
230 from security.system_requirements import get_tier, NodeTierLevel
231 tier = get_tier()
232 if tier in (NodeTierLevel.EMBEDDED, NodeTierLevel.OBSERVER):
233 return 'potato'
234 if tier == NodeTierLevel.LITE:
235 return 'minimal'
236 except Exception:
237 pass
239 # Fallback: direct hardware check (no system_requirements module)
240 try:
241 import os
242 cores = os.cpu_count() or 1
243 # Try psutil for RAM
244 try:
245 import psutil
246 ram_gb = psutil.virtual_memory().total / (1024 ** 3)
247 except ImportError:
248 ram_gb = 4.0 # conservative default
250 if cores <= 2 and ram_gb < 4:
251 return 'potato'
252 if cores <= 2 or ram_gb < 6:
253 return 'minimal'
254 except Exception:
255 pass
257 return None # Standard+ hardware, use default theme
259 @staticmethod
260 def auto_select_theme() -> dict:
261 """Auto-select theme based on hardware on first boot.
263 Only acts when no active_theme.json exists yet.
264 Returns the result of apply_theme() or None if no action needed.
265 """
266 # Don't override existing theme choice
267 if os.path.isfile(_ACTIVE_THEME_PATH):
268 return None
270 recommended = ThemeService.detect_performance_tier()
271 if recommended:
272 logger.info("Auto-selecting theme '%s' for hardware", recommended)
273 return ThemeService.apply_theme(recommended)
275 # Default to hart-default for capable hardware
276 return ThemeService.apply_theme('hart-default')
278 # ── Conky Integration ────────────────────────────────────────
280 @staticmethod
281 def get_conky_color_overrides() -> dict:
282 """Return Conky-specific colors from the active theme."""
283 theme = ThemeService.get_active_theme()
284 return theme.get('conky', {})
286 # ── CSS Variables Export ─────────────────────────────────────
288 @staticmethod
289 def get_css_variables() -> str:
290 """Export active theme as CSS custom properties for the shell."""
291 theme = ThemeService.get_active_theme()
292 colors = theme.get('colors', {})
293 font = theme.get('font', {})
294 shell = theme.get('shell', {})
296 lines = [':root {']
297 # Colors
298 for key, val in colors.items():
299 css_key = key.replace('_', '-')
300 if val.startswith('rgba') or val.startswith('#'):
301 lines.append(f' --hart-{css_key}: {val};')
302 else:
303 lines.append(f' --hart-{css_key}: #{val};')
304 # Font
305 lines.append(f' --hart-font-family: "{font.get("family", "JetBrains Mono")}";')
306 lines.append(f' --hart-font-size: {font.get("size", 13)}px;')
307 lines.append(f' --hart-heading-size: {font.get("heading_size", 18)}px;')
308 lines.append(f' --hart-font-weight: {font.get("weight", 400)};')
309 lines.append(f' --hart-heading-weight: {font.get("heading_weight", 600)};')
310 # Shell
311 lines.append(f' --hart-blur: {shell.get("blur_radius", 20)}px;')
312 lines.append(f' --hart-saturation: {shell.get("saturation", 180)}%;')
313 lines.append(f' --hart-radius: {shell.get("border_radius", 16)}px;')
314 lines.append(f' --hart-panel-opacity: {shell.get("panel_opacity", 0.65)};')
315 lines.append(f' --hart-topbar-height: {shell.get("topbar_height", 40)}px;')
316 lines.append(f' --hart-icon-size: {shell.get("icon_size", 20)}px;')
317 lines.append(f' --hart-titlebar-height: {shell.get("panel_titlebar_height", 32)}px;')
318 lines.append(f' --hart-anim-speed: {shell.get("animation_speed_ms", 200)}ms;')
319 lines.append('}')
320 return '\n'.join(lines)
322 # ── Internal ─────────────────────────────────────────────────
324 @staticmethod
325 def _load_active_file() -> Optional[dict]:
326 if not os.path.isfile(_ACTIVE_THEME_PATH):
327 return None
328 try:
329 with open(_ACTIVE_THEME_PATH, 'r', encoding='utf-8') as f:
330 return json.load(f)
331 except (json.JSONDecodeError, OSError):
332 return None
334 @staticmethod
335 def _load_custom_overrides() -> Optional[dict]:
336 if not os.path.isfile(_CUSTOM_OVERRIDES_PATH):
337 return None
338 try:
339 with open(_CUSTOM_OVERRIDES_PATH, 'r', encoding='utf-8') as f:
340 return json.load(f)
341 except (json.JSONDecodeError, OSError):
342 return None
344 @staticmethod
345 def _deep_merge(base: dict, override: dict) -> dict:
346 """Recursively merge override into base."""
347 result = dict(base)
348 for key, val in override.items():
349 if key in result and isinstance(result[key], dict) and isinstance(val, dict):
350 result[key] = ThemeService._deep_merge(result[key], val)
351 else:
352 result[key] = val
353 return result
355 @staticmethod
356 def _apply_gtk(theme: dict):
357 """Apply GTK dark/light preference via gsettings (Linux only)."""
358 try:
359 dark = theme.get('gtk_prefer_dark', True)
360 scheme = 'prefer-dark' if dark else 'default'
361 subprocess.Popen(
362 ['gsettings', 'set', 'org.gnome.desktop.interface',
363 'color-scheme', scheme],
364 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
365 )
366 except FileNotFoundError:
367 pass # gsettings not available (Windows dev, etc.)
369 @staticmethod
370 def _notify_liquid_ui(theme: dict):
371 """Push theme update to LiquidUI Flask server."""
372 try:
373 from core.http_pool import pooled_post
374 port = os.environ.get('HART_LIQUID_UI_PORT', '6800')
375 pooled_post(
376 f'http://localhost:{port}/api/theme',
377 json={'theme': theme},
378 timeout=2,
379 )
380 except Exception:
381 pass # LiquidUI not running or requests unavailable