File size: 5,787 Bytes
c922f8b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""
Configuration utilities for GAIA implementation.

This module provides functions for loading, validating, and accessing configuration
settings from various sources like environment variables, files, and defaults.
"""

import os
import json
import logging
from typing import Dict, Any, Optional, List, Union
from pathlib import Path

from src.gaia.utils.common import get_nested_value, set_nested_value, load_json_file

logger = logging.getLogger("gaia_agent.utils.config_utils")

def load_from_env(prefix: str = "GAIA_") -> Dict[str, Any]:
    """
    Load configuration from environment variables.
    
    Args:
        prefix: Prefix for environment variables to load (e.g., 'GAIA_')
        
    Returns:
        Dictionary with environment variable values
    """
    config = {}
    
    for key, value in os.environ.items():
        if key.startswith(prefix):
            # Remove prefix and convert to lowercase
            config_key = key[len(prefix):].lower()
            
            # Handle nested keys (e.g., GAIA_API_BASE_URL becomes api.base_url)
            if '_' in config_key:
                parts = config_key.split('_')
                current = config
                
                # Navigate to the nested dictionary
                for part in parts[:-1]:
                    if part not in current:
                        current[part] = {}
                    current = current[part]
                    
                # Set the value in the nested dictionary
                current[parts[-1]] = value
            else:
                config[config_key] = value
    
    return config

def load_from_file(file_path: str) -> Optional[Dict[str, Any]]:
    """
    Load configuration from a JSON file.
    
    Args:
        file_path: Path to the configuration file
        
    Returns:
        Dictionary with configuration values, or None if loading failed
    """
    return load_json_file(file_path)

def load_from_env_file(env_path: Union[str, Path]) -> Dict[str, str]:
    """
    Load environment variables from a .env file.
    
    Args:
        env_path: Path to the .env file
        
    Returns:
        Dictionary with environment variables
    """
    env_vars = {}
    
    if not os.path.exists(env_path):
        return env_vars
        
    try:
        with open(env_path, 'r') as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                    
                # Handle export syntax (export KEY=value)
                if line.startswith('export '):
                    line = line[7:]
                    
                if '=' in line:
                    key, value = line.split('=', 1)
                    # Remove quotes if present
                    value = value.strip('\'"')
                    env_vars[key.strip()] = value
    except Exception as e:
        logger.error(f"Error loading .env file {env_path}: {str(e)}")
        
    return env_vars

def merge_configs(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
    """
    Merge two configuration dictionaries, with override values taking precedence.
    
    Args:
        base: Base configuration dictionary
        override: Override configuration dictionary
        
    Returns:
        Merged configuration dictionary
    """
    result = base.copy()
    
    for key, value in override.items():
        if isinstance(value, dict) and key in result and isinstance(result[key], dict):
            # Recursively merge nested dictionaries
            result[key] = merge_configs(result[key], value)
        else:
            # Simple override
            result[key] = value
            
    return result

def find_config_file(file_name: str, search_paths: List[str]) -> Optional[str]:
    """
    Find a configuration file in a list of search paths.
    
    Args:
        file_name: Name of the configuration file
        search_paths: List of paths to search
        
    Returns:
        Full path to the configuration file, or None if not found
    """
    for path in search_paths:
        full_path = os.path.join(path, file_name)
        if os.path.exists(full_path):
            return full_path
            
    return None

def get_config_value(config: Dict[str, Any], key: str, default: Any = None, env_var: Optional[str] = None) -> Any:
    """
    Get a configuration value from a configuration dictionary, with fallback to environment variable.
    
    Args:
        config: Configuration dictionary
        key: Configuration key (can use dot notation for nested keys)
        default: Default value if key is not found
        env_var: Environment variable to check if key is not found in config
        
    Returns:
        Configuration value or default
    """
    # First check the configuration dictionary
    value = get_nested_value(config, key, None)
    if value is not None:
        return value
        
    # Then check the environment variable if specified
    if env_var and env_var in os.environ:
        return os.environ[env_var]
        
    # Finally return the default value
    return default

def convert_value_type(value: str, target_type: type) -> Any:
    """
    Convert a string value to a specified type.
    
    Args:
        value: String value to convert
        target_type: Target type to convert to
        
    Returns:
        Converted value
    """
    if target_type == bool:
        return value.lower() in ('true', 'yes', 'y', '1', 'on')
    elif target_type == int:
        return int(value)
    elif target_type == float:
        return float(value)
    elif target_type == list:
        return [item.strip() for item in value.split(',')]
    else:
        return value