"""Configuration manager for Rex AI Assistant. Separates secrets (.env) from runtime configuration (rex_config.json). Provides migration from legacy environment-based configuration. """ from __future__ import annotations # ruff: noqa: I001, UP006, UP015, UP035, UP045 import copy import json import os import shutil import tempfile from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from rex.logging_utils import get_logger from rex.runtime_paths import config_path as resolve_config_path from rex.runtime_paths import env_path as resolve_env_path logger = get_logger(__name__) # Default configuration structure DEFAULT_CONFIG: Dict[str, Any] = { "active_profile": "default", "profiles_dir": "profiles", "audio": { "input_device_index": None, "output_device_index": None, "sample_rate": 16000, }, "wakeword": { "backend": "openwakeword", "wakeword": "rex", "keyword": None, "model_path": None, "embedding_path": None, "fallback_to_builtin": True, "fallback_keyword": "hey jarvis", "threshold": 0.5, "window": 1.0, "poll_interval": 0.01, "wake_sound_path": "assets/wake_acknowledgment.wav", }, "models": { "llm_provider": "transformers", "llm_backend": None, "llm_model": "sshleifer/tiny-gpt2", "llm_max_tokens": 120, "llm_temperature": 0.7, "autonomy_mode": "manual", "llm_top_p": 0.9, "llm_top_k": 50, "llm_seed": 42, "stt_model": "base", "stt_device": "cpu", "tts_provider": "xtts", "tts_model": None, "tts_voice": None, "tts_speed": 1.08, "windows_tts_voice_index": None, }, "runtime": { "log_level": "INFO", "file_logging_enabled": False, "memory_max_bytes": 131072, "warm_runtime_max_cost_mb": 6144.0, "warm_runtime_idle_timeout_s": 900.0, "transcripts_enabled": True, "transcripts_dir": "transcripts", "active_user": None, "user_id": "default", "memory_max_turns": 50, "command_duration": 5.0, "capture_seconds": 5.0, "detection_frame_seconds": 1.0, "conversation_export": True, "speak_language": "en", }, "search": { "providers": "serpapi,brave,duckduckgo,google", }, "home_assistant": { "base_url": None, "verify_ssl": True, "timeout": 10.0, }, "ollama": { "base_url": "http://localhost:11434", "use_cloud": False, }, "openai": { "model": None, "base_url": None, }, "api": { "rate_limit": "30/minute", "allowed_origins": ["*"], }, "ui": { "start_minimized": False, }, "conversation": { "followups": { "enabled": True, "max_per_session": 1, "lookback_hours": 72, "expire_hours": 168, }, }, "location": { "default_location": None, "default_timezone": None, }, } # Map of legacy environment variable names to config paths ENV_TO_CONFIG_MAPPING: Dict[str, str] = { # Audio settings "REX_SAMPLE_RATE": "audio.sample_rate", "REX_INPUT_DEVICE": "audio.input_device_index", "REX_OUTPUT_DEVICE": "audio.output_device_index", "REX_AUDIO_INPUT_DEVICE": "audio.input_device_index", "REX_AUDIO_OUTPUT_DEVICE": "audio.output_device_index", "REX_DEVICE": "audio.input_device_index", # Wake word settings (canonical key: wakeword) # Note: REX_WAKEWORD_THRESHOLD was removed in US-307; use wakeword.threshold in rex_config.json "REX_WAKEWORD_BACKEND": "wakeword.backend", "REX_WAKEWORD": "wakeword.wakeword", "REX_WAKEWORD_KEYWORD": "wakeword.keyword", "REX_WAKEWORD_WINDOW": "wakeword.window", "REX_WAKEWORD_POLL_INTERVAL": "wakeword.poll_interval", "REX_WAKE_SOUND": "wakeword.wake_sound_path", # Model settings "REX_LLM_PROVIDER": "models.llm_provider", "REX_LLM_BACKEND": "models.llm_backend", "REX_LLM_MODEL": "models.llm_model", "REX_LLM_MAX_TOKENS": "models.llm_max_tokens", "REX_LLM_TEMPERATURE": "models.llm_temperature", "REX_LLM_TOP_P": "models.llm_top_p", "REX_LLM_TOP_K": "models.llm_top_k", "REX_LLM_SEED": "models.llm_seed", "REX_WHISPER_MODEL": "models.stt_model", "REX_WHISPER_DEVICE": "models.stt_device", "REX_TTS_PROVIDER": "models.tts_provider", "REX_TTS_MODEL": "models.tts_model", "REX_TTS_VOICE": "models.tts_voice", "REX_WINDOWS_TTS_VOICE_INDEX": "models.windows_tts_voice_index", # Runtime settings "REX_LOG_LEVEL": "runtime.log_level", "REX_FILE_LOGGING_ENABLED": "runtime.file_logging_enabled", "REX_MEMORY_MAX_BYTES": "runtime.memory_max_bytes", "REX_TRANSCRIPTS_ENABLED": "runtime.transcripts_enabled", "REX_TRANSCRIPTS_DIR": "runtime.transcripts_dir", "REX_ACTIVE_USER": "runtime.active_user", "REX_USER_ID": "runtime.user_id", "REX_MEMORY_MAX_TURNS": "runtime.memory_max_turns", "REX_COMMAND_DURATION": "runtime.command_duration", "REX_CAPTURE_SECONDS": "runtime.capture_seconds", "REX_DETECTION_FRAME_SECONDS": "runtime.detection_frame_seconds", "REX_CONVERSATION_EXPORT": "runtime.conversation_export", "REX_SPEAK_LANGUAGE": "runtime.speak_language", "REX_DEBUG_LOGGING": "runtime.log_level", # Special: maps "true" to "DEBUG" # Search settings "REX_SEARCH_PROVIDERS": "search.providers", # Home Assistant "HA_BASE_URL": "home_assistant.base_url", "HA_VERIFY_SSL": "home_assistant.verify_ssl", "HA_TIMEOUT": "home_assistant.timeout", # Ollama "OLLAMA_HOST": "ollama.base_url", "OLLAMA_BASE_URL": "ollama.base_url", "OLLAMA_USE_CLOUD": "ollama.use_cloud", # OpenAI "OPENAI_MODEL": "openai.model", "OPENAI_BASE_URL": "openai.base_url", # API settings "REX_RATE_LIMIT": "api.rate_limit", "REX_ALLOWED_ORIGINS": "api.allowed_origins", } # Secret environment variables that should stay in .env SECRET_ENV_VARS = { "OPENAI_API_KEY", "BRAVE_API_KEY", "SERPAPI_KEY", "GOOGLE_API_KEY", "BROWSERLESS_API_KEY", "REX_PROXY_TOKEN", "REX_SPEAK_API_KEY", "HASS_SECRET", "HA_SECRET", "HA_TOKEN", "FLASK_LIMITER_STORAGE_URI", "REX_SPEAK_STORAGE_URI", "OPENWEATHERMAP_API_KEY", "SERPAPI_URL", "BRAVE_URL", "GOOGLE_URL", "BROWSERLESS_URL", "GOOGLE_CSE_ID", "OLLAMA_API_KEY", } def _get_nested(data: Dict[str, Any], path: str, default: Any = None) -> Any: """Get value from nested dict using dot notation path.""" keys = path.split(".") value: Any = data for key in keys: if not isinstance(value, dict): return default if key not in value: return default value = value[key] return value def _set_nested(data: Dict[str, Any], path: str, value: Any) -> None: """Set value in nested dict using dot notation path.""" keys = path.split(".") cursor: Any = data for key in keys[:-1]: if not isinstance(cursor, dict): raise TypeError(f"Cannot set nested key on non-dict at '{key}' for path '{path}'") if key not in cursor or not isinstance(cursor[key], dict): cursor[key] = {} cursor = cursor[key] if not isinstance(cursor, dict): raise TypeError(f"Cannot set nested key on non-dict for path '{path}'") cursor[keys[-1]] = value def _parse_bool(value: str) -> bool: """Parse boolean from string.""" return value.strip().lower() in ("true", "1", "yes", "on") def _parse_int(value: str) -> Optional[int]: """Parse integer from string, handling float-formatted strings.""" try: float_val = float(value) if float_val.is_integer(): return int(float_val) except (ValueError, OverflowError): pass return None def _parse_float(value: str) -> Optional[float]: """Parse float from string.""" try: return float(value) except (ValueError, OverflowError): return None def load_config(path: str | Path | None = None) -> Dict[str, Any]: """Load configuration from JSON file. Args: path: Path to config file (default: config/rex_config.json) Returns: Configuration dict with all settings Behavior: - Creates directory if missing - Creates default file if missing - If invalid JSON, renames to .invalid..json and recreates defaults """ config_path = resolve_config_path(path) # Ensure config directory exists config_path.parent.mkdir(parents=True, exist_ok=True) # If file doesn't exist, create it with defaults if not config_path.exists(): logger.info(f"Config file not found, creating default: {config_path}") save_config(copy.deepcopy(DEFAULT_CONFIG), path) return copy.deepcopy(DEFAULT_CONFIG) # Try to load existing config try: with open(config_path, "r", encoding="utf-8") as f: config = json.load(f) # Merge with defaults to ensure all keys exist _canonicalize_wakeword_config(config) merged = _deep_merge(copy.deepcopy(DEFAULT_CONFIG), config) _normalize_wake_word_config(merged) logger.debug(f"Loaded config from {config_path}") return merged except json.JSONDecodeError as exc: # Invalid JSON - backup and recreate timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = config_path.with_suffix(f".invalid.{timestamp}.json") shutil.copy2(config_path, backup_path) logger.error( f"Invalid JSON in config file: {exc}. " f"Backed up to {backup_path} and recreating with defaults." ) save_config(copy.deepcopy(DEFAULT_CONFIG), path) return copy.deepcopy(DEFAULT_CONFIG) except Exception as exc: logger.error(f"Failed to load config from {config_path}: {exc}") logger.info("Using default configuration") return copy.deepcopy(DEFAULT_CONFIG) def save_config(config: Dict[str, Any], path: str | Path | None = None) -> None: """Save configuration to JSON file. Args: config: Configuration dict to save path: Path to config file (default: config/rex_config.json) Output format: - Pretty printed JSON with 2-space indentation - Sorted keys for stable ordering - Newline at end of file """ config_path = resolve_config_path(path) # Ensure config directory exists config_path.parent.mkdir(parents=True, exist_ok=True) _normalize_wake_word_config(config) temp_path: Optional[Path] = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=str(config_path.parent), delete=False, ) as f: temp_path = Path(f.name) json.dump(config, f, indent=2, sort_keys=True, ensure_ascii=False) f.write("\n") # Newline at EOF f.flush() os.fsync(f.fileno()) os.replace(temp_path, config_path) logger.info(f"Saved config to {config_path}") except Exception as exc: logger.error(f"Failed to save config to {config_path}: {exc}") raise finally: if temp_path and temp_path.exists(): try: temp_path.unlink() except OSError: pass def _deep_merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]: """Deep merge two dicts, with overlay taking precedence.""" result: Dict[str, Any] = base.copy() for key, value in overlay.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = _deep_merge(result[key], value) else: result[key] = value return result def _canonicalize_wakeword_config(config: Dict[str, Any]) -> None: legacy = config.pop("wake_word", None) if not isinstance(legacy, dict): return canonical = config.get("wakeword") if isinstance(canonical, dict): merged = legacy.copy() merged.update(canonical) config["wakeword"] = merged else: config["wakeword"] = legacy def _normalize_wake_word_config(config: Dict[str, Any]) -> None: _canonicalize_wakeword_config(config) wakeword_section = config.get("wakeword") if not isinstance(wakeword_section, dict): return keyword = wakeword_section.get("keyword") wakeword = wakeword_section.get("wakeword") if ( (not isinstance(keyword, str) or not keyword.strip()) and isinstance(wakeword, str) and wakeword.strip() ): wakeword_section["keyword"] = wakeword.strip() def migrate_legacy_env_to_config( env_path: str | Path | None = None, config_path: str | Path | None = None, *, dry_run: bool = False, ) -> List[str]: """Migrate legacy non-secret environment variables to rex_config.json. Args: env_path: Path to .env file config_path: Path to rex_config.json dry_run: If True, compute and return planned actions without writing any files. Returns: List of migration notes and warnings. In dry_run mode the notes use the prefix "[dry-run] Would migrate" so callers can distinguish them while still comparing env_key, config path, and parsed value. Behavior: - Reads the .env file directly (not the process environment) for both dry-run and real migration, so both modes see the same source data. - Only migrates non-secret settings from .env. - Only migrates if config value is null or default. - Never migrates secrets (they stay in .env). - In dry_run=True mode: returns planned actions without writing files. - In dry_run=False mode: writes updated config and returns action notes. """ env_file = resolve_env_path(env_path) if not env_file.exists(): return ["No .env file found, no migration needed"] notes: List[str] = [] # Load current config. In dry_run mode we must not create the file if it # does not exist, so we short-circuit to DEFAULT_CONFIG in that case. cfg_path = resolve_config_path(config_path) if dry_run and not cfg_path.exists(): config = _deep_merge(copy.deepcopy(DEFAULT_CONFIG), {}) else: config = load_config(cfg_path) # Read .env file directly — same source for both dry-run and real migration. env_vars: Dict[str, str] = {} try: with open(env_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue if "=" in line: key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") env_vars[key] = value except Exception as exc: notes.append(f"Failed to read .env file: {exc}") return notes # Check for legacy non-secret variables applying the same mapping and coercion # rules as the real migration. migrated_count = 0 for env_key, config_path_str in ENV_TO_CONFIG_MAPPING.items(): if env_key in env_vars and env_key not in SECRET_ENV_VARS: env_value = env_vars[env_key] current_value = _get_nested(config, config_path_str) # Get default value for comparison default_value = _get_nested(DEFAULT_CONFIG, config_path_str) # Only migrate if current config value is null or matches default should_migrate = current_value is None or current_value == default_value if should_migrate and env_value: # Parse value based on type — identical rules for both modes. parsed_value: Any = env_value effective_path = config_path_str # Special handling for specific keys if env_key == "REX_DEBUG_LOGGING": if _parse_bool(env_value): parsed_value = "DEBUG" effective_path = "runtime.log_level" else: continue # Don't migrate if false elif any( x in env_key for x in ["THRESHOLD", "WINDOW", "INTERVAL", "DURATION", "TIMEOUT"] ): parsed = _parse_float(env_value) parsed_value = parsed if parsed is not None else parsed_value elif ("DEVICE" in env_key and "INDEX" in env_key) or env_key in [ "REX_INPUT_DEVICE", "REX_OUTPUT_DEVICE", "REX_DEVICE", "REX_AUDIO_INPUT_DEVICE", "REX_AUDIO_OUTPUT_DEVICE", ]: parsed = _parse_int(env_value) parsed_value = parsed if parsed is not None else parsed_value elif any( x in env_key for x in ["MAX_TOKENS", "TOP_K", "SEED", "MAX_TURNS", "SAMPLE_RATE"] ): parsed = _parse_int(env_value) parsed_value = parsed if parsed is not None else parsed_value elif any(x in env_key for x in ["ENABLED", "EXPORT", "USE_CLOUD", "VERIFY_SSL"]): parsed_value = _parse_bool(env_value) elif env_key == "REX_ALLOWED_ORIGINS": parsed_value = [x.strip() for x in env_value.split(",") if x.strip()] if dry_run: notes.append( f"[dry-run] Would migrate {env_key} -> {effective_path} = {parsed_value!r}" ) else: # Apply to in-memory config; write once at the end. _set_nested(config, effective_path, parsed_value) notes.append(f"Migrated {env_key} -> {effective_path} = {parsed_value}") migrated_count += 1 elif not should_migrate: if dry_run: notes.append( f"[dry-run] Skipped {env_key} -> {config_path_str}" f" (config already has non-default value: {current_value!r})" ) if dry_run: if migrated_count > 0: notes.append( f"\nDry run complete — {migrated_count} setting(s) would be written to {config_path}" ) notes.append("Run without --dry-run to apply these changes.") else: notes.append("No legacy environment variables found that need migration.") return notes # Save config if anything was migrated if migrated_count > 0: save_config(config, cfg_path) notes.append(f"\nMigrated {migrated_count} settings from .env to {cfg_path}") notes.append( "These environment variables are now ignored. Use rex_config.json for runtime settings." ) else: notes.append("No legacy environment variables found that needed migration") return notes def get_legacy_env_warnings() -> List[str]: """Check for legacy non-secret environment variables and return warnings. Returns: List of warning messages about legacy env vars that should be in config """ warnings: List[str] = [] for env_key in ENV_TO_CONFIG_MAPPING.keys(): if env_key not in SECRET_ENV_VARS and os.getenv(env_key): config_path = ENV_TO_CONFIG_MAPPING[env_key] warnings.append( f"Legacy env var {env_key} is set. " f"This is ignored at runtime; use {config_path} in rex_config.json instead. " f"Run 'rex-config migrate-legacy-env' to migrate automatically." ) return warnings __all__ = [ "DEFAULT_CONFIG", "load_config", "save_config", "migrate_legacy_env_to_config", "get_legacy_env_warnings", "ENV_TO_CONFIG_MAPPING", "SECRET_ENV_VARS", ]