hookvision-rk/hookvision/config_loader.py

import configparser
import re
from .models import *


def load_config(config_path: str) -> Config:
    cfg = configparser.ConfigParser()
    cfg.read(config_path, encoding='utf-8')

    config = Config()
    config.app_name = cfg.get("App", "name", fallback="钢包工位视频监测系统")
    config.copyright = cfg.get("App", "copyright", fallback="恒莱达")
    config.screen = cfg.getint("App", "screen", fallback=0)

    module_type_str = cfg.get("Hardware", "module_type", fallback="RFID")
    config.module_type = ModuleType.RFID if module_type_str == "RFID" else ModuleType.HD3200

    if cfg.has_section("Video"):
        config.video.rtsp_timeout = cfg.getfloat("Video", "rtsp_timeout", fallback=5.0)
        config.video.ratio_16_9 = cfg.getboolean("Video", "ratio_16_9", fallback=True)

    # RFID
    if config.module_type == ModuleType.RFID:
        config.rfid = SocketConfig(
            ip=cfg.get("RFID", "ip"),
            port=cfg.getint("RFID", "port"),
            stability_time=cfg.getint("RFID", "stability_time", fallback=20),
            stability_count=cfg.getint("RFID", "stability_count", fallback=3)
        )

    # Modbus
    if config.module_type == ModuleType.HD3200:
        config.modbus = ModbusConfig(
            ip=cfg.get("Modbus", "ip"),
            port=cfg.getint("Modbus", "port", fallback=502),
            frequency=cfg.getint("Modbus", "frequency", fallback=100),
            stability_count=cfg.getint("Modbus", "stability_count", fallback=3)
        )

    # AI
    config.ai.enabled = cfg.getboolean("AI", "enabled", fallback=False)
    config.ai.tts_enabled = cfg.getboolean("AI", "tts_enabled", fallback=False)
    config.ai.debug_trace = cfg.getboolean("AI", "debug_trace", fallback=True)
    config.ai.offset_ratio_tolerance = cfg.getfloat("AI", "offset_ratio_tolerance", fallback=0.25)
    config.ai.ready_debounce = cfg.getint("AI", "ready_debounce", fallback=3)
    config.ai.auto_calib_ready_frames = cfg.getint("AI", "auto_calib_ready_frames", fallback=3)
    config.ai.auto_calib_duration_secs = cfg.getfloat("AI", "auto_calib_duration_secs", fallback=1.0)
    config.ai.auto_calib_min_samples = cfg.getint("AI", "auto_calib_min_samples", fallback=2)

    # AI Modules (自动扫描 AI:N 段)
    for section in sorted(
        (s for s in cfg.sections() if re.match(r'^AI:\d+$', s)),
        key=lambda s: int(s.split(':')[1])
    ):
        idx = int(section.split(':')[1])
        topics_str = cfg.get(section, "topics", fallback="")
        config.ai_modules.append(AIModuleModel(
            id=idx,
            broker=cfg.get(section, "broker"),
            port=cfg.getint(section, "port", fallback=1883),
            topics=[t.strip() for t in topics_str.split('\n') if t.strip()]
        ))

    # Buckets (自动扫描 Bucket:N 段)
    bucket_sections = sorted(
        (s for s in cfg.sections() if re.match(r'^Bucket:\d+$', s)),
        key=lambda s: int(s.split(':')[1])
    )
    for section in bucket_sections:
        bucket = BucketModel(
            id=cfg.getint(section, "id"),
            description=cfg.get(section, "description"),
            code_x=cfg.get(section, "code_x"),
            code_y=cfg.get(section, "code_y", fallback=""),
            ai_module_id=cfg.getint(section, "ai_module", fallback=None),
            hook_lug_x_offset_ratio_left=cfg.getfloat(section, "hook_lug_x_offset_ratio_left", fallback=None),
            hook_lug_y_offset_ratio_left=cfg.getfloat(section, "hook_lug_y_offset_ratio_left", fallback=None),
            hook_lug_x_offset_ratio_right=cfg.getfloat(section, "hook_lug_x_offset_ratio_right", fallback=None),
            hook_lug_y_offset_ratio_right=cfg.getfloat(section, "hook_lug_y_offset_ratio_right", fallback=None),
        )
        left_rtsp = cfg.get(section, "left_rtsp")
        right_rtsp = cfg.get(section, "right_rtsp")
        left_topic = cfg.get(section, "left_topic", fallback="")
        right_topic = cfg.get(section, "right_topic", fallback="")

        cameras = []
        if left_rtsp:
            cameras.append(CameraModel(position=CameraPosition.LEFT, rtsp_url=left_rtsp, topic=left_topic))
        if right_rtsp:
            cameras.append(CameraModel(position=CameraPosition.RIGHT, rtsp_url=right_rtsp, topic=right_topic))
        bucket.cameras = cameras
        config.buckets.append(bucket)

    _ensure_auto_calib_defaults(cfg, config_path)

    # AutoRestart
    if cfg.has_section("AutoRestart"):
        ar = config.auto_restart
        ar.enabled = cfg.getboolean("AutoRestart", "enabled", fallback=False)
        ar.interval = RestartInterval(cfg.get("AutoRestart", "interval", fallback="day"))
        ar.hour = cfg.getint("AutoRestart", "hour", fallback=3)
        ar.minute = cfg.getint("AutoRestart", "minute", fallback=0)
        ar.target = RestartTarget(cfg.get("AutoRestart", "target", fallback="app"))
        ar.weekdays = cfg.getint("AutoRestart", "weekdays", fallback=0)
        ar.day_of_month = cfg.getint("AutoRestart", "day_of_month", fallback=1)
        ar.countdown = cfg.getint("AutoRestart", "countdown", fallback=60)

    return config


_AUTO_CALIB_DEFAULTS = {
    "auto_calib_ready_frames": "3",
    "auto_calib_duration_secs": "1.0",
    "auto_calib_min_samples": "2",
}


def _ensure_auto_calib_defaults(cfg, config_path: str):
    """配置文件缺少自动标定参数时自动补写默认值（仅在文件存在且发生变更时写回）。"""
    import os
    if not os.path.exists(config_path):
        return
    if not cfg.has_section("AI"):
        cfg.add_section("AI")
    changed = False
    for key, val in _AUTO_CALIB_DEFAULTS.items():
        if not cfg.has_option("AI", key):
            cfg.set("AI", key, val)
            changed = True
    if changed:
        with open(config_path, "w", encoding="utf-8") as f:
            cfg.write(f)