zhaoyuan_DataPreProcess/core/config_manager.py

import json
import os
from pathlib import Path
from typing import Dict, Any, Optional, Union, List
import yaml
from core.optimized_processor import get_base_path

# 全局配置字典
_CONFIG: Dict[str, Any] = {}
base_path = get_base_path()
print("base_path is", base_path)
# 默认配置
DEFAULT_CONFIG = {
    # 数据处理配置
    "processing": {
        "use_parallel": True,
        "max_workers": None,  # None表示使用CPU核心数
        "batch_size": 10000,
        "use_vectorized": True
    },
    # 数据库配置
    "database": {
        "batch_query_size": 10000,
        "use_parallel_query": True,
        "max_db_workers": None
    },
    # 插件配置
    "plugins": {
        "enabled": True,
        "auto_discover": True
    },
    # 日志配置
    "logging": {
        "level": "INFO",
        "log_to_file": True,
        "log_dir": "logs"
    },
    # 坐标系配置
    "coordinates": {
        "utm_zone": 51,
        "x_offset": 0.0,
        "y_offset": 0.0
    },
    # 文件路径配置
    "paths": {
        "resources_dir": "resources",
        "plugins_dir": "plugins",
        "output_dir": "output",
        "engine_path": f"{os.path.join(base_path, '_internal/engine')}",
        "map_path": f"{os.path.join(base_path, '_internal/data_map')}",
        "dbc_path": f"{os.path.join(base_path, '_internal/VBox.dbc')}"
        # "engine_path": '_internal/engine',
        # "map_path": '_internal/data_map',
        # "dbc_path": '_internal/VBox.dbc'
    }
}


def load_config(config_path: Optional[Union[str, Path]] = None) -> Dict[str, Any]:
    """加载配置文件

    Args:
        config_path: 配置文件路径，支持JSON和YAML格式

    Returns:
        配置字典
    """
    global _CONFIG

    # 首先使用默认配置
    _CONFIG = DEFAULT_CONFIG.copy()

    if config_path is None:
        # 尝试查找默认配置文件
        default_paths = [
            Path("config.json"),
            Path("config.yaml"),
            Path("config.yml"),
            Path("config/config.json"),
            Path("config/config.yaml"),
            Path("config/config.yml")
        ]

        for path in default_paths:
            if path.exists():
                config_path = path
                break

    if config_path is None:
        print("未找到配置文件，使用默认配置")
        return _CONFIG

    # 确保config_path是Path对象
    if isinstance(config_path, str):
        config_path = Path(config_path)

    if not config_path.exists():
        print(f"配置文件不存在: {config_path}，使用默认配置")
        return _CONFIG

    try:
        # 根据文件扩展名选择解析方法
        suffix = config_path.suffix.lower()

        if suffix in [".yaml", ".yml"]:
            with open(config_path, "r", encoding="utf-8") as f:
                user_config = yaml.safe_load(f)
        elif suffix == ".json":
            with open(config_path, "r", encoding="utf-8") as f:
                user_config = json.load(f)
        else:
            print(f"不支持的配置文件格式: {suffix}，使用默认配置")
            return _CONFIG

        # 递归合并配置
        _CONFIG = _merge_configs(_CONFIG, user_config)
        print(f"成功加载配置文件: {config_path}")

    except Exception as e:
        print(f"加载配置文件失败: {e}")

    return _CONFIG


def _merge_configs(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
    """递归合并配置字典

    Args:
        base: 基础配置
        override: 覆盖配置

    Returns:
        合并后的配置
    """
    result = base.copy()

    for key, value in override.items():
        # 如果两个都是字典，递归合并
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = _merge_configs(result[key], value)
        else:
            # 否则直接覆盖
            result[key] = value

    return result


def get_config(section: Optional[str] = None) -> Any:
    """获取配置

    Args:
        section: 配置节名称，如果为None则返回整个配置字典

    Returns:
        配置值或配置字典
    """
    global _CONFIG

    # 如果配置为空，加载默认配置
    if not _CONFIG:
        _CONFIG = DEFAULT_CONFIG.copy()

    if section is None:
        return _CONFIG

    # 支持点号分隔的多级配置访问，如 "database.batch_query_size"
    if "." in section:
        parts = section.split(".")
        current = _CONFIG
        for part in parts:
            if part in current:
                current = current[part]
            else:
                print(f"配置节不存在: {section}")
                return None
        return current

    # 单级配置访问
    return _CONFIG.get(section)


def set_config(section: str, value: Any) -> None:
    """设置配置

    Args:
        section: 配置节名称，支持点号分隔的多级配置
        value: 配置值
    """
    global _CONFIG

    # 如果配置为空，加载默认配置
    if not _CONFIG:
        _CONFIG = DEFAULT_CONFIG.copy()

    # 支持点号分隔的多级配置设置
    if "." in section:
        parts = section.split(".")
        current = _CONFIG

        # 遍历到倒数第二级
        for part in parts[:-1]:
            # 如果中间节点不存在或不是字典，创建新字典
            if part not in current or not isinstance(current[part], dict):
                current[part] = {}
            current = current[part]

        # 设置最后一级的值
        current[parts[-1]] = value
    else:
        # 单级配置设置
        _CONFIG[section] = value


def save_config(config_path: Union[str, Path], format: str = "json") -> bool:
    """保存配置到文件

    Args:
        config_path: 配置文件路径
        format: 文件格式，支持 "json" 和 "yaml"

    Returns:
        是否保存成功
    """
    global _CONFIG

    # 确保config_path是Path对象
    if isinstance(config_path, str):
        config_path = Path(config_path)

    # 确保目录存在
    config_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        # 根据格式选择保存方法
        if format.lower() == "yaml":
            with open(config_path, "w", encoding="utf-8") as f:
                yaml.dump(_CONFIG, f, default_flow_style=False, sort_keys=False)
        else:  # 默认使用JSON
            with open(config_path, "w", encoding="utf-8") as f:
                json.dump(_CONFIG, f, indent=2, ensure_ascii=False)

        print(f"配置已保存到: {config_path}")
        return True

    except Exception as e:
        print(f"保存配置失败: {e}")
        return False


def update_config(new_config: Dict[str, Any]) -> None:
    """更新全局配置

    Args:
        new_config: 新的配置字典，将与现有配置合并
    """
    global _CONFIG
    _CONFIG = _merge_configs(_CONFIG, new_config)
    print("配置已更新")


# 初始化时加载配置
load_config()