vent_agent_2.0/tests/test_main_fan_preparation.py

"""固定取数与预处理的少量回归：类型、分页、窗口和首次Agent调用前的数据。"""
from datetime import datetime, timedelta
import json
from types import SimpleNamespace

import pytest

from db import main_fan_store
from extensions.main_fan import discovery, gateway_queries, history, history_processing
from extensions.main_fan.batch import BatchRunner
from extensions.main_fan.batch_context import current_scope
from runtime.python_sandbox import workspace_file
from test_main_fan_batch import batch_workspace, bound


def realtime_rows(mine):
    identity = {'deptId': mine['dept_id'], 'mineId': mine['mine_id']}
    safety = [{**identity, 'nodeId': f'state{fan}{index}', 'nodePlace': f'主通{fan}{stage}级',
               'sensorTypeId': '1010', 'sensorName': '主通风机'} for fan in (1, 2) for index, stage in enumerate(('一', '二'), 1)]
    equipment = [{**identity, 'nodeId': f'flow{fan}', 'nodeTypeNo': '30101',
                  'nodeTypeName': name + '号风量', 'sensorUnit': 'm3/min'} for fan, name in ((1, '一'), (2, '二'))]
    equipment += [{**identity, 'nodeId': 'speed2', 'nodeTypeNo': '30102', 'nodeTypeName': '二号风速'},
                  {**identity, 'nodeId': 'equipment-state2', 'nodeTypeNo': '30007', 'nodeTypeName': '二号风机运行状态'}]
    return safety, equipment


def test_fixed_selection_keeps_stages_and_excludes_equipment_state_and_speed():
    mine = {'dept_id': '1237', 'mine_id': '610831018590'}
    safety, equipment = realtime_rows(mine)
    safety.append({'nodeId': 'vibration', 'sensorTypeId': '1022', 'nodePlace': '主通1振动'})
    plan = discovery.plan_from_rows(mine, safety, equipment)
    assert len(plan.shafts) == 1 and plan.shafts[0].shaft_id == 'unresolved-main'
    points = plan.shafts[0].points
    assert len(points) == 6
    assert {p.fan_id for p in points} == {'1号风机', '2号风机'}
    assert {p.node_id for p in points} == {'state11', 'state12', 'state21', 'state22', 'flow1', 'flow2'}
    assert all(not p.mapping_confirmed for p in points)


@pytest.mark.parametrize('label,fan', [
    ('风机房1号主扇一级电机&其它&其它', '1号风机'),
    ('风机房1号主扇二级电机&其它&其它', '1号风机'),
    ('风机房2号主扇一级电机&其它&其它', '2号风机'),
    ('风机房2号主扇二级电机&其它&其它', '2号风机'),
    ('主通1一级', '1号风机'), ('主通1二级', '1号风机'),
    ('主通2一级', '2号风机'), ('主通12二级', '12号风机'),
    ('主扇一级电机', ''), ('主扇二级电机', ''),
])
def test_fan_number_is_not_motor_stage(label, fan):
    assert discovery.fan_name({'nodePlace': label}) == fan


def test_fixed_selection_rejects_other_mine_and_conflicting_duplicate():
    mine = {'dept_id': '1237', 'mine_id': '610831018590'}
    safety, equipment = realtime_rows(mine)
    safety[0]['deptId'] = '1045'
    with pytest.raises(ValueError, match='deptId'):
        discovery.plan_from_rows(mine, safety, equipment)
    safety[0]['deptId'] = '1237'
    safety.append({**safety[0], 'nodePlace': '主通2一级'})
    with pytest.raises(ValueError, match='冲突'):
        discovery.plan_from_rows(mine, safety, equipment)


class FixedGateway:
    def __init__(self, mine):
        self.safety, self.equipment = realtime_rows(mine)
        # 测点ID本身不含中文，名称保留一级/二级。
        for index, row in enumerate(self.safety):
            row['nodeId'] = 'state' + str(index + 1)
        self.calls = []

    async def proxy(self, agent, path, *, params):
        params = dict(params)
        self.calls.append((path, params))
        if path.endswith('getNodeRealtimeDetails') or path.endswith('selectSbNodeList'):
            rows = self.safety if path.endswith('getNodeRealtimeDetails') else self.equipment
            page, size = int(params['pageNum']), int(params['pageSize'])
            return {'code': 200, 'total': len(rows), 'rows': rows[(page - 1) * size:page * size]}
        if 'nodeIds' in params:
            return {'code': 200, 'data': [[{'NodeID': params['nodeIds'], 'DataTime': params['startTime'],
                'RealData': 1, 'KgDesc': '开', 'StateID': 0, 'stateName': '正常'}], []]}
        return {'code': 200, 'data': [{'nodeId': params['nodeId'], 'sbNodeCurve': [
            {'dataTime': params['startTime'], 'realData': '0.00'}]}]}


@pytest.mark.asyncio
async def test_fixed_discovery_gets_all_pages_without_model(batch_workspace, monkeypatch):
    gateway = FixedGateway(batch_workspace.mine)
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
    monkeypatch.setenv('MAIN_FAN_REALTIME_MAX_PAGE_SIZE', '2')
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    with bound(batch_workspace, days=1) as (scope, identifier):
        plan = await discovery.discover(identifier, scope.user, scope.session)
    assert len(plan.shafts[0].points) == 6 and len(gateway.calls) == 4
    assert [params['sensorTypeList[0]'] for path, params in gateway.calls if '/aqjk/' in path] == ['1010', '1010']
    assert [params['devOfSystem'] for path, params in gateway.calls if '/zdsb/' in path] == ['30', '30']


def test_explicit_two_shaft_names_are_grouped_without_splitting_by_fan_number():
    mine = {'dept_id': '1237', 'mine_id': '610831018590'}
    safety, equipment = realtime_rows(mine)
    for row in safety[:2] + equipment[:1]:
        row['shaftName'] = '北风井'
    for row in safety[2:] + equipment[1:2]:
        row['shaftName'] = '南风井'
    plan = discovery.plan_from_rows(mine, safety, equipment)
    assert {shaft.shaft_name for shaft in plan.shafts} == {'北风井', '南风井'}


def test_preprocessing_keeps_zero_and_sampling_bounds_for_single_point_comparison():
    def state(fan, old, new, by):
        return {'node_id': 'state' + fan, 'system': 'safety', 'role': 'fan_state', 'fan_id': fan + '号风机',
            'state_intervals': [], 'state_value_semantics': '接口标签', 'state_changes': [
                {'from_state': old, 'to_state': new, 'after': '2026-10-03T15:00:00+08:00',
                 'by': by, 'intervening_unknown_or_gap': False, 'evidence': []}]}
    points = [state('2', '开', '停', '2026-10-03T15:00:10+08:00'),
              state('1', '停', '开', '2026-10-03T15:01:00+08:00')]
    points += [{'node_id': 'flow' + fan, 'system': 'equipment', 'role': 'flow',
        'fan_id': fan + '号风机', 'unit': 'm3/min', 'values': {}} for fan in ('1', '2')]
    details = {'points': points}
    segments = [{'key': point['node_id'], 'system': 'equipment', 'node_id': point['node_id'], 'status': 'completed',
        'start_time': '2026-10-03T00:00:00+08:00', 'end_time': '2026-10-04T00:00:00+08:00'} for point in points[2:]]
    class Reader:
        def data(self, record):
            if record['node_id'] == 'flow1':
                samples = [{'time': '2026-10-03T15:10:00+08:00', 'value': 0}]
            else:
                start = datetime.fromisoformat('2026-10-03T14:40:00+08:00')
                samples = [{'time': (start + timedelta(seconds=i)).isoformat(), 'value': 5000} for i in range(30)]
            return {'series': [{'samples': samples}]}
    rules = {'switch': {'maximum_switch_interval_minutes': 10, 'transition_minutes': 5, 'window_minutes': 30,
        'sample_interval_seconds': 60, 'minimum_completeness': .9, 'maximum_cv': .05}}
    compact = history_processing.prepare(details, {'segments': segments}, Reader(), rules, '/workspace/details.json')
    event = compact['preprocessing']['switch_candidates'][0]
    assert event['from_fan'] == '2号风机' and event['to_fan'] == '1号风机'
    assert event['sampling_localization_seconds'] == 60 and event['interval_seconds'] == 50
    assert event['interval_minutes'] == pytest.approx(50 / 60)
    assert event['interval_status'] == 'estimated'
    assert event['interval_bounds_minutes'] == {'lower': 0, 'upper': 1}
    old = next(row for row in event['flow_context'] if row['fan_id'] == '2号风机')
    assert old['before_reference']['value'] == 5000 and old['before_reference']['valid']
    new = next(row for row in event['flow_context'] if row['fan_id'] == '1号风机')
    assert new['after_reference']['value'] == 0 and new['after_reference']['valid']
    assert event['flow_comparison']['relative_change_percent'] == 100
    # 默认输入省略缺口字段，原观测信息保留；零值、查询失败和比较口径不变。
    compact.update(cutoff='2026-10-21T09:42:02+08:00', lookback_days=40,
                   transport={'coverage_complete': False})
    compact['points'][0]['state_periods'] = [{'state': '开', 'first_observed_at': '2026-10-03T14:00:00+08:00',
        'last_observed_at': '2026-10-03T15:00:00+08:00', 'observed_parts': 2}]
    slim = history_processing.agent_input(compact)
    assert 'gaps' not in slim['states'][0]['periods'][0]
    assert compact['points'][0]['state_periods'][0]['observed_parts'] == 2
    slim_event = slim['switch_candidates'][0]
    assert slim_event['comparison']['change_percent'] == 100
    assert slim_event['duration_minutes'] == pytest.approx(50 / 60)
    assert slim_event['duration_status'] == 'estimated'
    assert slim_event['from_stopped_at'] == '2026-10-03 15:00:10'
    assert slim_event['to_started_at'] == '2026-10-03 15:01:00'
    assert slim_event['comparison']['before_value'] == 5000 and slim_event['comparison']['after_value'] == 0
    assert not {'before_mean', 'after_mean', 'before_samples', 'after_samples'} & slim_event['comparison'].keys()
    assert next(row for row in slim_event['flow'] if row['fan'] == '1号风机')['after'][1] == 0
    assert any('日段取数失败' in note for note in slim['notes'])


def timing_point(node, fan, changes):
    return {'node_id': node, 'label': node, 'system': 'safety', 'role': 'fan_state',
        'fan_id': fan, 'state_changes': [{'from_state': old, 'to_state': new,
            'after': '2026-10-03T' + after + '+08:00', 'by': '2026-10-03T' + by + '+08:00',
            'intervening_unknown_or_gap': False} for old, new, after, by in changes]}


def test_switch_interval_longer_than_normal_limit_is_not_lost_by_grouping():
    points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00')]),
              timing_point('new', '1号风机', [('停', '开', '15:14:55', '15:15:00')])]
    events = history_processing.switch_candidates(points, 10)
    assert len(events) == 1
    event = events[0]
    assert event['from_fan'] == '2号风机' and event['to_fan'] == '1号风机'
    assert event['interval_minutes'] == 15 and event['interval_status'] == 'estimated'
    assert event['interval_bounds_minutes']['lower'] == pytest.approx(895 / 60)
    assert event['interval_bounds_minutes']['upper'] == pytest.approx(905 / 60)
    assert not event['interval_has_gaps']


def test_switch_interval_uses_final_start_after_jitter_and_last_participating_stage():
    points = [timing_point('old1', '2号风机', [('开', '停', '15:00:10', '15:00:15')]),
              timing_point('old2', '2号风机', [('开', '停', '15:00:15', '15:00:20')]),
              timing_point('new1', '1号风机', [('停', '开', '15:00:00', '15:00:05')]),
              timing_point('new2', '1号风机', [('停', '开', '15:00:05', '15:00:10'),
                  ('开', '停', '15:00:10', '15:00:12'), ('停', '开', '15:00:45', '15:00:50')])]
    event = history_processing.switch_candidates(points, 10)[0]
    assert event['from_stopped_at'] == '2026-10-03T15:00:20+08:00'
    assert event['to_started_at'] == '2026-10-03T15:00:50+08:00'
    assert event['interval_seconds'] == 30 and event['overlap_seconds'] == 0
    assert event['interval_bounds_minutes'] == {'lower': 25 / 60, 'upper': 35 / 60}
    assert len(event['interval_sources']) == 4


def test_start_before_stop_retains_overlap_instead_of_negative_gap():
    points = [timing_point('old', '2号风机', [('开', '停', '15:00:25', '15:00:30')]),
              timing_point('new', '1号风机', [('停', '开', '15:00:05', '15:00:10')])]
    event = history_processing.switch_candidates(points, 10)[0]
    assert event['interval_seconds'] == 0 and event['interval_minutes'] == 0
    assert event['overlap_seconds'] == 20
    assert event['interval_bounds_minutes'] == {'lower': 0, 'upper': 0}


def test_original_fan_restart_cancels_pending_handover():
    points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00'),
                  ('停', '开', '15:14:55', '15:15:00')]),
              timing_point('new', '1号风机', [('停', '开', '15:29:55', '15:30:00')])]
    assert not history_processing.switch_candidates(points, 10)


def test_timing_across_unknown_samples_is_computed_but_marked_as_limited():
    points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00')]),
              timing_point('new', '1号风机', [('停', '开', '15:14:55', '15:15:00')])]
    points[1]['state_changes'][0]['intervening_unknown_or_gap'] = True
    event = history_processing.switch_candidates(points, 10)[0]
    assert event['interval_minutes'] == 15 and event['interval_has_gaps']


def test_nearest_point_uses_one_hour_target_and_never_crosses_switch():
    boundary = datetime.fromisoformat('2026-10-03T15:00:00+08:00')
    samples = [(boundary + timedelta(minutes=offset), value) for offset, value in
               [(-120, 100), (-70, 200), (-61, 300), (-45, 400), (-1, 500), (1, 9999)]]
    result = history_processing.reference_sample(samples, boundary, 'before')
    assert result['value'] == 300
    assert result['time'] == '2026-10-03T13:59:00+08:00'
    assert result['target_time'] == '2026-10-03T14:00:00+08:00'
    after = history_processing.reference_sample(
        [(boundary + timedelta(minutes=offset), 0) for offset in (-1, 2, 28, 59, 80)], boundary, 'after')
    assert after['valid'] and after['value'] == 0
    assert after['time'] == (boundary + timedelta(minutes=59)).isoformat()
    missing = history_processing.reference_sample([(boundary + timedelta(minutes=1), 9000)], boundary, 'before')
    assert not missing['valid'] and missing['time'] is None and missing['value'] is None


@pytest.mark.asyncio
async def test_runner_supplies_processed_data_before_workflow_call(batch_workspace, monkeypatch):
    gateway = FixedGateway(batch_workspace.mine)
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    calls = []
    async def executor(**kwargs):
        scope = current_scope.get()
        workflow = main_fan_store.workflow(scope.active_workflow_id, scope.user, scope.session)
        prepared = workflow['context']['prepared_data']
        data = json.loads(workspace_file(scope.user, scope.session, prepared['agent_data_path']).read_text())
        assert prepared['mode'] == 'fixed_code'
        assert len(data['states']) == 4 and len(data['flows']) == 2
        assert not {'points', 'preprocessing', 'transport', 'source_policy', 'data_preparation'} & data.keys()
        assert data['reports'][0]['mode'] == 'analyze'
        assert kwargs['workflow_id'] == scope.active_workflow_id
        assert 'message' not in kwargs and 'graph_config' not in kwargs
        calls.append(workflow['context']['task_input']['measurements']['batch']['phase'])
        return SimpleNamespace(status='success', run_id='test', error='', output='', usage={}, structured_output={
            'result': '# 测试报告', 'analysis_id': 'test', 'workflow_id': scope.active_workflow_id,
            'platform_payload': {'taskId': 'test-task', 'judgment': {'resultCode': 'UNDETERMINED'}}})
    output = batch_workspace.root / 'prepared-output'
    result = await BatchRunner(output, batch_workspace.pdf, executor=executor).run(
        {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 1},
        snapshot={'snapshot_id': 'c' * 32, 'mines': [batch_workspace.mine]})
    assert result['status'] == 'completed', result
    assert calls == ['assessment']
    assert len(gateway.calls) == 8, '两个实时列表和六个日段全部先由固定代码获取'
    assert (output / '1219/unresolved-main/给Agent的数据.json').is_file()