vent_agent_2.0/scripts/preview_main_fan_data.py

#!/usr/bin/env python3
"""用已有矿级原件回放固定发现及预处理，生成真实Agent输入；不访问模型或网关。"""
import argparse
import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from extensions.main_fan import discovery, history_compaction, history_processing
from extensions.main_fan.source_policy import source_policy


def read(path):
    return json.loads(path.read_text(encoding='utf-8'))


def write(path, value):
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + '\n', encoding='utf-8')


def preview(capture, output):
    capture, output = Path(capture).resolve(), Path(output).resolve()
    source = read(capture / 'history-manifest.json')
    mine = {key: source[key] for key in ('dept_id', 'mine_id', 'mine_name')}
    def rows(name):
        body = read(capture / name)
        return body['rows'] if 'rows' in body else body['data']['rows']
    plan = discovery.plan_from_rows(mine, rows('safety-realtime.json'), rows('equipment-realtime.json'))
    points = [point.model_dump(mode='json') for shaft in plan.shafts for point in shaft.points]
    selected = {(point['system'], point['node_id']) for point in points}
    segments = [row for row in source['segments'] if (row['system'], row['node_id']) in selected]
    expected = len(points) * source['lookback_days']
    completed = sum(row['status'] == 'completed' for row in segments)
    manifest = {**source, 'points': points, 'segments': segments, 'point_count': len(points),
        'expected_segments': expected, 'completed_segments': completed,
        'failed_segments': len(segments) - completed, 'coverage_complete': completed == expected,
        'source_policy': source_policy()}
    output.mkdir(parents=True, exist_ok=True)
    def resolve(virtual):
        if not virtual.startswith('/workspace/'):
            raise ValueError('证据路径必须位于/workspace/')
        path = (capture / virtual.removeprefix('/workspace/')).resolve()
        if not path.is_relative_to(capture):
            raise ValueError('证据路径越出回放目录')
        return path
    with history_compaction.RecordReader(resolve) as reader:
        details = history_compaction.summarize(manifest, reader)
        details.update(mine=mine, source_policy=source_policy(), storage=source['storage'])
        data = history_processing.prepare(details, manifest, reader,
            read(ROOT / 'examples/main_fan/equivalence_rules.json'), str(output / '开停与候选详情.json'))
    details['discovery'] = {'mode': 'fixed_code', 'shafts': [{key: shaft.model_dump(mode='json')[key]
        for key in ('shaft_id', 'shaft_name', 'identity_status', 'limitations')} for shaft in plan.shafts]}
    details['data_preparation'] = {'mode': 'fixed_code', 'validation': 'existing_real_capture_replay',
        'agent_needs_monitoring_queries': False, 'new_gateway_requests': 0, 'model_calls': 0,
        'note': '仅回放本矿已有40天原件；此示例为监测数据输入，PDF由正式批次另外提供。'}
    details['evidence'] = {'archive_local_path': str(capture / 'history-evidence.zip'),
        'original_manifest_local_path': str(capture / 'history-manifest.json')}
    data = history_processing.agent_input(data)
    write(output / '固定测点计划.json', plan.model_dump(mode='json'))
    write(output / '开停与候选详情.json', details)
    write(output / '给Agent的数据.json', data)
    events = details['preprocessing']['switch_candidates']
    summary = ['# 固定代码给Agent的数据', '',
        '矿井：' + mine['mine_name'] + '；deptId：' + mine['dept_id'] + '。',
        '回查' + str(source['lookback_days']) + '天；' + str(len(points)) + '个测点；' + str(completed) + '/' + str(expected) + '个成功日段。',
        '本次使用已有真实原件验证，没有新增网关请求或模型调用。正式批次在调用Agent前自动执行相同取数及处理流程。', '',
        '固定程序完成：筛矿 → 分页获取1010开停点及30101风量点 → 按天回查 → 开停区间整理 → 切换候选 → 前后单点风量对比。',
        '原机以切换范围起点前1小时为目标，新机以终点后1小时为目标，分别取对应一侧最接近目标的一条采样；不求平均，不使用分钟覆盖率或CV门槛，零值保留。',
        'Agent直接读取同目录的“给Agent的数据.json”。默认只保留矿井、时段、开停区间、风量概况、切换候选和必要限制。',
        '调试计数、测点ID、质量明细、规则及计算过程在“开停与候选详情.json”，原始采样仍在原目录的ZIP。', '',
        '## 本例切换候选', '']
    for event in events:
        summary.extend([event['event_id'] + '：' + event['from_fan'] + ' → ' + event['to_fan'],
            '采样定位范围：(' + event['after'] + ', ' + event['by'] + ']。该范围不是切换耗时。', '',
            '| 对象 | 目标时间 | 实际采样时间 | 风量 |', '|---|---|---|---:|'])
        for flow in event['flow_context']:
            side = 'before' if flow['fan_id'] == event['from_fan'] else 'after' if flow['fan_id'] == event['to_fan'] else None
            if side:
                sample = flow[side + '_reference']
                summary.append('| ' + flow['fan_id'] + ('切换前' if side == 'before' else '切换后') +
                    ' | ' + sample['target_time'] + ' | ' + str(sample['time']) + ' | ' + str(sample['value']) + ' |')
        change = event['flow_comparison']['relative_change_percent']
        summary.extend(['', '按原编号的单点读数变化：' + (str(change) + '%' if change is not None else '不可计算') +
            '。该数值不证明连续稳定工况或实际故障；物理对应待确认的说明保留。', ''])
    (output / '说明.md').write_text('\n'.join(summary) + '\n', encoding='utf-8')
    return {'output': str(output), 'point_count': len(points), 'completed_segments': completed,
        'switch_candidates': len(events), 'agent_data_bytes': (output / '给Agent的数据.json').stat().st_size,
        'new_gateway_requests': 0, 'model_calls': 0}


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--capture', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    print(json.dumps(preview(args.capture, args.output), ensure_ascii=False))