vent_agent_2.0/extensions/main_fan/equipment.py

"""重大设备历史曲线取数；未确认的返回字段由 Agent 读取原始数据解释。"""
import asyncio

from extensions.main_fan import gateway_queries

CURVE_PATH = '/zdsb/sbnode/selectSbNodeCurve'


def describe_data(value, depth=0):
    """只展示实际JSON结构、数量和少量原值，不推断业务语义。"""
    if isinstance(value, dict):
        return {key: describe_data(item, depth + 1) for key, item in list(value.items())[:12]} if depth < 5 else {'type': 'object', 'keys': list(value)[:12]}
    if isinstance(value, list):
        return {'type': 'array', 'count': len(value),
                'first': describe_data(value[0], depth + 1) if value and depth < 5 else None}
    return {'type': type(value).__name__, 'example': value[:120] if isinstance(value, str) else value}


def _curve_data(body, node_id=None):
    if not isinstance(body, (dict, list)):
        raise ValueError('重大设备曲线未返回 JSON 对象或数组，请核对上游接口')
    data = body.get('data') if isinstance(body, dict) and 'data' in body else body
    series, known = [], data == []
    if isinstance(data, list) and data and all(isinstance(row, dict) and 'nodeId' in row
            and isinstance(row.get('sbNodeCurve'), list) for row in data):
        known = True
        try:
            for row in data:
                if node_id is not None and str(row['nodeId']) != node_id:
                    raise ValueError('重大设备曲线返回测点与本次请求不一致')
                samples = []
                for sample in row['sbNodeCurve']:
                    if not isinstance(sample, dict) or not {'dataTime', 'realData'} <= sample.keys():
                        known = False
                        break
                    samples.append({'time': gateway_queries.parse_time(sample['dataTime']).isoformat(),
                                    'value': sample['realData']})
                series.append({'node_id': str(row['nodeId']), 'node_place': row.get('nodePlace'), 'samples': samples})
        except ValueError as exc:
            if '返回测点' in str(exc):
                raise
            known = False
    summary = {'response_type': type(body).__name__,
               'response_keys': list(body) if isinstance(body, dict) else None,
               'data_type': type(data).__name__,
               'data_keys': list(data) if isinstance(data, dict) else None,
               'data_item_count': len(data) if isinstance(data, list) else None,
               'sample_count': sum(len(row['samples']) for row in series) if known else None,
               'normalization': 'sbNodeCurve' if known else 'unrecognized',
               'data_structure': describe_data(data),
               'next_action': '读取完整 data_path 的 payload 并核对实际曲线字段、测点、时间和单位；数组项数不等于已确认的采样点数。'}
    return {'payload': body, 'series': series if known else None}, summary


async def query_timeseries(identifier, node_id, dept_id, start_time, end_time, user, session):
    params, _, _ = gateway_queries.curve_parameters(node_id, dept_id, start_time, end_time, node_key='nodeId')
    workflow, request, body = await gateway_queries.fetch(identifier, user, session, CURVE_PATH, params)
    data, summary = await asyncio.to_thread(_curve_data, body, node_id)
    return await asyncio.to_thread(gateway_queries.archive, workflow, user, session, request, body, data, summary,
        kind='equipment_timeseries', collection='equipment_queries', operation='查询重大设备历史时序',
        filename='equipment-curve.json')