vent_agent_2.0/runtime/test_lab/main_fan_simulation.py

"""从三个明确的测试输入组装本次主备风机任务，不预置报告事实或结论。"""
from datetime import datetime, timedelta
from math import ceil
from typing import Literal
from zoneinfo import ZoneInfo

from pydantic import BaseModel, ConfigDict, Field, model_validator

from algorithms.main_fan.models import ScreeningInput

LOCAL_ZONE = ZoneInfo('Asia/Shanghai')
FAN_IDS = ('mock-report-fan-1', 'mock-report-fan-2')


class MainFanSimulation(BaseModel):
    model_config = ConfigDict(extra='forbid', allow_inf_nan=False)
    mine_code: str = Field(default='mock-zy', min_length=1, max_length=128)
    mine_name: str = Field(default='陕西众源煤业集团有限公司', min_length=1, max_length=256)
    direction: Literal['1_to_2', '2_to_1'] = '1_to_2'
    stopped_at: datetime = Field(default_factory=lambda: datetime.now(LOCAL_ZONE).replace(hour=10, minute=0, second=0, microsecond=0))
    switch_interval_minutes: float = Field(default=6, ge=0, le=120)
    before_flow_m3_min: float = Field(default=3800, gt=0, le=1_000_000)
    after_flow_m3_min: float = Field(default=3610, ge=0, le=1_000_000)
    switch_source: Literal['simulation', 'file'] = 'simulation'
    flow_source: Literal['simulation', 'file'] = 'simulation'
    switch_asset_id: str | None = Field(default=None, min_length=1, max_length=128)
    flow_asset_id: str | None = Field(default=None, min_length=1, max_length=128)

    @model_validator(mode='after')
    def timezone_required(self):
        if self.stopped_at.utcoffset() is None:
            raise ValueError('停机时间必须包含时区')
        for slot in ('switch', 'flow'):
            if (getattr(self, slot+'_source') == 'file') != bool(getattr(self, slot+'_asset_id')):
                raise ValueError('选择JSON输入时须指定该项文件，模拟输入时请移除对应文件')
        return self


def build_input(settings: MainFanSimulation):
    at = settings.stopped_at
    started = at + timedelta(minutes=settings.switch_interval_minutes)
    common = {'mine_code': settings.mine_code, 'mine_name': settings.mine_name}
    evidence = {'source': 'test-lab://main-fan/three-inputs', 'method': 'synthetic',
        'label': '测试页生成的运行时序', 'excerpt': '根据本次填写的切换间隔及前后风量生成，不含报告识别事实'}
    origin, destination = (0, 1) if settings.direction == '1_to_2' else (1, 0)
    data = {**common, 'fan_id': FAN_IDS[0], 'fan_name': '1号通风机', 'fan_pair': list(FAN_IDS),
        'evaluation_time': (at + timedelta(minutes=ceil(settings.switch_interval_minutes)+90)).isoformat(),
        'use_gateway': False, 'paired_assessment': True, 'document_cache_policy': 'off', 'reports': [],
        'devices': [{'device_code': key, 'name': f'{i+1}号通风机', **common, 'evidence': evidence}
            for i, key in enumerate(FAN_IDS)], 'fan_state_records': [], 'operating_series': []}
    # 同步整机状态按分钟采样，并插入准确的启动时刻，保留小数分钟而不向上取整。
    times = {at+timedelta(minutes=m) for m in range(-40, ceil(settings.switch_interval_minutes)+61)}
    times.add(started)
    samples = [[], []]
    for time in sorted(times):
        for i, key in enumerate(FAN_IDS):
            running = time < at if i == origin else time >= started
            data['fan_state_records'].append({**common, 'fan_id': key, 'observed_at': time.isoformat(),
                'state': 'running' if running else 'stopped', 'complete_fan_state': True, 'evidence': evidence})
            value = settings.before_flow_m3_min if i == origin else settings.after_flow_m3_min
            samples[i].append({'observed_at': time.isoformat(), 'value': value if running else 0})
    data['operating_series'] = [{**common, 'fan_id': key, 'metric': 'flow', 'unit': 'm3/min',
        'sensor_id': f'test-flow-{i+1}', 'mapping_confirmed': True, 'samples': samples[i], 'evidence': evidence}
        for i, key in enumerate(FAN_IDS)]
    ScreeningInput.model_validate({k: v for k, v in data.items() if k in ScreeningInput.model_fields})
    return data