import os import sys import yaml DEFAULT_FEATURE_COLS = ["Temperature", "Humidity", "Pressure", "Wind Speed", "Rain"] # Resolve project root current_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.abspath(os.path.join(current_dir, "../../")) if project_root not in sys.path: sys.path.append(project_root) def _resolve_config_path() -> str: """Resolve config path with optional env override.""" env_path = os.environ.get("FSL_CONFIG_PATH", "").strip() path = env_path if env_path else os.path.join(project_root, "config.yaml") if not os.path.isabs(path): path = os.path.join(project_root, path) return os.path.abspath(path) def _load_config_or_raise(path: str) -> dict: """Load YAML config and fail fast on invalid/missing config.""" if not os.path.exists(path): raise FileNotFoundError(f"Config file not found: {path}") with open(path, "r", encoding="utf-8") as f: loaded = yaml.safe_load(f) or {} if not isinstance(loaded, dict): raise ValueError(f"Config root must be a mapping/dict: {path}") return loaded CONFIG_PATH = _resolve_config_path() cfg = _load_config_or_raise(CONFIG_PATH) def get_config(): """Return the loaded configuration dict.""" return cfg def reload_config(): """Reload configuration from disk into the module-level `cfg`.""" global cfg, CONFIG_PATH CONFIG_PATH = _resolve_config_path() cfg = _load_config_or_raise(CONFIG_PATH) return cfg def get_config_path() -> str: """Return the active config path for this process.""" return CONFIG_PATH def get_nested(config: dict | None, keys: tuple[str, ...], default=None): """Safely fetch a nested key path from a dict-like config.""" current = config if isinstance(config, dict) else {} for key in keys: if not isinstance(current, dict) or key not in current: return default current = current[key] return current def feature_cols_from_cfg(config: dict | None = None) -> list[str]: """ Resolve feature columns with backward-compatible fallback: 1) data.feature_cols (preferred) 2) data_download.feature_cols (legacy) 3) hardcoded defaults """ source = cfg if config is None else config for key_path in (("data", "feature_cols"), ("data_download", "feature_cols")): cols = get_nested(source, key_path, None) if isinstance(cols, list) and cols: return [str(col) for col in cols] return DEFAULT_FEATURE_COLS.copy()