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)

    # 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)

    # 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)
        )
        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)

    return config