| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- """按测点及时间窗口取原始曲线，保存为本轮可引用的可信输入。"""
- import asyncio
- import config
- from extensions.main_fan import gateway_queries
- CURVE_PATH = '/aqjk/alarmcurve/getCurveByNodeId'
- def _normalize(body, node_id, start, end):
- if not isinstance(body, dict) or body.get('success') is False or str(body.get('code')) not in {'0', '200'}:
- raise ValueError('曲线业务响应未成功，请检查网关及上游返回')
- data = body.get('data')
- if data == []:
- rows, metadata = [], []
- elif (isinstance(data, list) and len(data) == 2
- and isinstance(data[0], list) and isinstance(data[1], list)):
- rows, metadata = data
- else:
- raise ValueError('曲线响应应包含 [采样记录数组, 测点信息数组]，不能将未知结构当成空数据')
- maximum = config.get_int('MAIN_FAN_SERIES_MAX_SAMPLES', 200000)
- if maximum <= 0 or len(rows) > maximum:
- raise ValueError(f'单次返回最多接受 {maximum} 条采样，请缩小查询时间范围')
- for item in [*rows, *metadata]:
- if not isinstance(item, dict) or item.get('NodeID') != node_id:
- raise ValueError('曲线返回的测点编号与请求不一致，不能混用其他测点')
- samples, times = [], []
- for row in rows:
- if 'RealData' not in row or 'DataTime' not in row:
- raise ValueError('曲线采样缺少 RealData 或 DataTime，请核对实际接口字段')
- observed = gateway_queries.parse_time(row['DataTime'])
- times.append(observed)
- samples.append({'time': observed.isoformat(), 'value': row['RealData'],
- 'value_label': row.get('KgDesc'), 'status_code': row.get('StateID'),
- 'status_label': row.get('stateName')})
- # 原顺序、重复、缺值和窗口外记录全部保留；只提供质量计数，不替 Agent 选择分析样本。
- quality = {'duplicate_time_count': len(times) - len(set(times)),
- 'out_of_order_count': sum(right < left for left, right in zip(times, times[1:])),
- 'out_of_range_count': sum(not start <= value <= end for value in times),
- 'null_value_count': sum(row['value'] is None for row in samples)}
- return {'metadata': metadata, 'samples': samples}, {
- 'sample_count': len(samples), 'first_time': min(times).isoformat() if times else None,
- 'last_time': max(times).isoformat() if times else None, 'quality': quality,
- 'preview': samples[:3] + (samples[-1:] if len(samples) > 3 else [])}
- async def query(identifier, node_id, dept_id, start_time, end_time, user, session):
- params, start, end = gateway_queries.curve_parameters(node_id, dept_id, start_time, end_time)
- workflow, request, body = await gateway_queries.fetch(identifier, user, session, CURVE_PATH, params)
- data, summary = await asyncio.to_thread(_normalize, body, params['nodeIds'], start, end)
- return await asyncio.to_thread(gateway_queries.archive, workflow, user, session, request, body,
- {'timezone': 'Asia/Shanghai', **data}, summary, kind='sensor_timeseries', collection='timeseries_queries',
- operation='查询测点历史时序', filename='samples.json')
|
