vent_agent_2.0/mock_gateway/operating_data.py

"""显式合成的两份时序样例；与原始检测报告及人工转录夹具独立。"""
import hashlib
import json
from pathlib import Path

from algorithms.main_fan.models import Device, Evidence, FanStateRecord, OperatingSeries


def load_operating_data(directory, *, base_url="http://127.0.0.1:8766"):
    result = {'devices':[], 'operating_series':[], 'fan_state_records':[], 'raw_curves':{}}
    identity = None
    for kind, filename in (('switch-states','mock_switch_states.json'),('airflow','mock_airflow_series.json')):
        path = Path(directory)/filename
        raw = json.loads(path.read_text(encoding='utf-8'))
        if raw.get('schema_version') != 1 or raw.get('data_kind') != 'synthetic':
            raise ValueError('模拟时序必须声明 schema_version=1、data_kind=synthetic')
        current = (raw['mine_code'],raw['mine_name'])
        if identity is not None and current != identity:
            raise ValueError('两份模拟时序的矿井身份不一致')
        identity = current
        result['raw_curves'][kind] = raw
        evidence = Evidence(source=base_url.rstrip('/')+'/mock/main-fan/curves/'+kind,
            label='主备风机整机开停模拟数据' if kind == 'switch-states' else '主备风机风量模拟数据',
            excerpt=raw['description'],sha256=hashlib.sha256(path.read_bytes()).hexdigest(),method='synthetic')
        common = {'mine_code':current[0],'mine_name':current[1],'evidence':evidence}
        for series in raw['series']:
            if kind == 'switch-states':
                if series['state_scope'] != 'whole_fan':
                    raise ValueError('开停样例必须是整机状态')
                result['devices'].append(Device(device_code=series['fan_id'],name=series['fan_name'],**common).model_dump(mode='json'))
                for sample in series['samples']:
                    if sample['value'] not in (0,1):
                        raise ValueError('整机开停状态只允许0或1')
                    result['fan_state_records'].append(FanStateRecord(fan_id=series['fan_id'],
                        observed_at=sample['observed_at'],state='running' if sample['value'] == 1 else 'stopped',
                        complete_fan_state=True,**common).model_dump(mode='json'))
            else:
                result['operating_series'].append(OperatingSeries(fan_id=series['fan_id'],metric='flow',unit=series['unit'],
                    sensor_id=series['sensor_id'],mapping_confirmed=series['mapping_confirmed'],samples=series['samples'],
                    **common).model_dump(mode='json'))
    if {d['device_code'] for d in result['devices']} != {s['fan_id'] for s in result['operating_series']}:
        raise ValueError('开停和风量样例的两台风机标识不一致')
    return result