vent_agent_2.0/tests/test_main_fan_equivalence.py

"""双机范围、额定功率、受控插值、真实切换窗口及报告项目选择。"""
from copy import deepcopy
from datetime import datetime, date, timedelta
import json
from types import SimpleNamespace

import pytest

from algorithms.main_fan.models import (ScreeningInput, PerformanceReport, PerformancePoint, Evidence,
    OperatingSeries, SeriesSample, SwitchRecord, Device, FanStateRecord)
from algorithms.main_fan.equivalence import (normalize_and_validate_fan_data, compare_basic_parameters,
    analyze_fan_performance_curve, analyze_fan_switch_event, assess_fan_equivalent_capability,
    RulePolicy, rated_power, _interpolate)
from extensions.main_fan.pipeline import load_rules
from extensions.main_fan import service
from extensions.main_fan.pair_reporting import render_fan_assessment_report
from db import main_fan_store


def sample(*, angle=35, pressure_ratio=1.02):
    reports=[]
    for number in (1,2):
        evidence=Evidence(source=f"/workspace/fan{number}.pdf",page=1,excerpt=f"合成{number}号风机测试报告",method='synthetic')
        r=PerformanceReport(report_id=f'SYN-{number}',mine_code='synthetic',mine_name='双机测试矿',fan_name=f'{number}号通风机',
            model='FBCDZ-No28',motor_model='SYN-MOTOR',serial_number=f'SYN-SERIAL-{number}',rated_power_raw='2×185 kW',
            rated_speed_rpm=740,impeller_diameter_m=2.8,inspected_on=date(2026,1,1),blade_angle_deg=angle,evidence=evidence)
        for index,q in enumerate((10,20,30,40,50,60,70)):
            pressure=500*(pressure_ratio if number==2 else 1)
            power=50
            r.points.append(PerformancePoint(label='运行工况' if index==3 else f'测点{index}',frequency_hz=50,
                flow_m3_s=q,static_pressure_pa=pressure,shaft_power_kw=power,
                reported_efficiency_pct=q*pressure/(10*power),evidence=evidence))
        reports.append(r)
    return ScreeningInput(mine_code='synthetic',mine_name='双机测试矿',fan_id='fan-1',fan_name='1号通风机',
        fan_pair=['fan-1','fan-2'],evaluation_time=datetime(2026,10,14,14),reports=reports,
        devices=[Device(device_code=f'fan-{i}',name=f'{i}号通风机',mine_code='synthetic') for i in (1,2)])


def dataset(data):
    return normalize_and_validate_fan_data(data,load_rules())


def test_both_fans_are_selected_and_real_nameplate_power_is_not_shaft_power():
    data=sample()
    normalized=dataset(data)
    assert [f['fan_id'] for f in normalized.fans]==['fan-1','fan-2']
    basic=compare_basic_parameters(normalized)
    power=next(r for r in basic['rows'] if r['field']=='rated_power')
    assert power['values'][0]['components_kw']==[185,185] and power['values'][0]['total_kw']==370
    assert power['status']=='match' and basic['status']=='pass'
    assert data.reports[0].points[0].shaft_power_kw==50


@pytest.mark.parametrize('expression,expected',[('2×185 kW',370),('185+185kW',370),('370千瓦',370),('185',185),('型号185',None),('/',None),('0',None)])
def test_rated_power_expression_is_parsed_without_guessing(expression,expected):
    report=sample().reports[0].model_copy(update={'rated_power_raw':expression})
    value=rated_power(report)
    assert (value['total_kw'] if value else None)==expected


def test_model_or_power_mismatch_is_explicit_and_missing_values_do_not_pass():
    data=sample();data.reports[1].rated_power_raw='2×200kW'
    result=compare_basic_parameters(dataset(data))
    assert result['status']=='fail'
    data.reports[1].rated_power_raw=''
    assert compare_basic_parameters(dataset(data))['status']=='insufficient'
    data.reports[1].model='FBCDZ-No32'
    assert compare_basic_parameters(dataset(data))['status']=='fail'


def test_curve_metrics_have_known_analytic_answers_and_never_extrapolate():
    result=analyze_fan_performance_curve(dataset(sample()))
    assert result['status']=='pass'
    row=result['results'][0]
    assert row['metrics']['mean_pressure_deviation_percent']==pytest.approx(2)
    assert row['metrics']['maximum_pressure_deviation_percent']==pytest.approx(2)
    assert row['metrics']['pressure_rmse_pa']==pytest.approx(10)
    assert row['metrics']['efficiency_difference_points']==pytest.approx(1.4)
    assert row['common_flow_range_m3_min']==[600,4200] and row['extrapolation_used'] is False
    with pytest.raises(ValueError,match='外推'):_interpolate([(10,(500,50,10)),(20,(500,50,20))],9)


def test_abs_relative_integral_handles_crossing_curves_exactly():
    data=sample(pressure_ratio=1)
    for point in data.reports[1].points:
        point.static_pressure_pa=500+(point.flow_m3_s-40)
    row=analyze_fan_performance_curve(dataset(data))['results'][0]
    assert row['metrics']['mean_pressure_deviation_percent']==pytest.approx(3)
    assert row['metrics']['pressure_rmse_pa']==pytest.approx((300)**.5)


def test_no_blade_angle_is_frequency_comparison_only_not_equivalence_pass():
    result=analyze_fan_performance_curve(dataset(sample(angle=None)))
    assert result['status']=='insufficient'
    assert result['results'][0]['comparability']=='frequency_only'
    assert result['results'][0]['status']=='informational'


def test_different_angles_or_frequency_are_not_combined():
    data=sample();data.reports[1].blade_angle_deg=40
    assert analyze_fan_performance_curve(dataset(data))['results']==[]
    data.reports[1].blade_angle_deg=35
    for point in data.reports[1].points:point.frequency_hz=40
    assert analyze_fan_performance_curve(dataset(data))['results']==[]


def test_duplicate_flow_conflict_or_insufficient_overlap_cannot_be_silently_fitted():
    data=sample()
    data.reports[0].points.append(data.reports[0].points[0].model_copy(update={'static_pressure_pa':700}))
    result=analyze_fan_performance_curve(dataset(data))
    assert result['status']=='insufficient' and not result['results'] and result['warnings']
    data=sample()
    for point in data.reports[1].points:point.flow_m3_s+=100
    assert not analyze_fan_performance_curve(dataset(data))['results']


def with_switch(data, *, after_flow=98, pressure=True):
    at=datetime(2026,10,14,12)
    data.switch_records=[SwitchRecord(mine_code=data.mine_code,event_id='synthetic-switch',switched_at=at,
        from_fan_id='fan-1',to_fan_id='fan-2',evidence=Evidence(source='synthetic-switch-log',method='synthetic'))]
    for fan_id,begin,value in [('fan-1',-35,100),('fan-2',5,after_flow)]:
        for metric,unit,level in [('flow','m3/s',value), *([('static_pressure','Pa',500)] if pressure else [])]:
            data.operating_series.append(OperatingSeries(mine_code=data.mine_code,fan_id=fan_id,metric=metric,unit=unit,
                sensor_id=fan_id+'-'+metric,mapping_confirmed=True,evidence=Evidence(source='synthetic-series',method='synthetic'),
                samples=[SeriesSample(observed_at=at+timedelta(minutes=begin+i),value=level) for i in range(70)]))
    return data


def test_real_switch_windows_compute_mean_changes_and_recovery():
    data=with_switch(sample())
    data.switch_records[0].from_stopped_at=data.switch_records[0].switched_at
    data.switch_records[0].to_started_at=data.switch_records[0].switched_at
    result=analyze_fan_switch_event(dataset(data))
    assert result['status']=='pass' and len(result['events'])==1
    event=result['events'][0]
    assert event['changes']['flow']['relative_change_percent']==pytest.approx(-2)
    assert event['changes']['flow']['absolute_change']==pytest.approx(-2)
    assert event['before_window']['flow']['sample_count']==30
    assert event['metrics']['recovery_time_minutes']==5


def test_persistent_switch_change_is_reported_without_claiming_recovery():
    result=analyze_fan_switch_event(dataset(with_switch(sample(),after_flow=80)))
    assert result['status']=='fail'
    assert result['events'][0]['changes']['flow']['relative_change_percent']==pytest.approx(-20)
    assert result['events'][0]['metrics']['recovery_time_minutes'] is None


@pytest.mark.parametrize('kind',['unconfirmed','missing','conflict','transition','foreign','future'])
def test_bad_or_unrelated_series_cannot_make_valid_switch_assessment(kind):
    data=with_switch(sample())
    if kind=='unconfirmed':
        for series in data.operating_series:series.mapping_confirmed=False
    elif kind=='missing':
        for series in data.operating_series:series.samples=series.samples[:10]
    elif kind=='conflict':
        for series in data.operating_series:
            series.samples.extend(s.model_copy(update={'value':s.value+20}) for s in list(series.samples))
    elif kind=='transition':
        data.fan_state_records=[FanStateRecord(mine_code=data.mine_code,fan_id='fan-2',observed_at=data.switch_records[0].switched_at,
            state='running',event_type='frequency_change')]
    elif kind=='foreign':
        for series in data.operating_series:series.mine_code='another'
    elif kind=='future':data.evaluation_time=datetime(2026,10,14,11)
    assert analyze_fan_switch_event(dataset(data))['events']==[]


def test_whole_fan_state_sequence_discovers_switch_but_stage_snapshots_do_not():
    data=with_switch(sample());data.switch_records=[]
    at=datetime(2026,10,14,12)
    for delta,states in [(-1,('running','stopped')),(0,('stopped','running'))]:
        data.fan_state_records.extend(FanStateRecord(mine_code=data.mine_code,fan_id=f'fan-{i+1}',state=state,
            observed_at=at+timedelta(minutes=delta),complete_fan_state=True,evidence=Evidence(source='synthetic-state',method='synthetic')) for i,state in enumerate(states))
    assert len(analyze_fan_switch_event(dataset(data))['events'])==1
    for record in data.fan_state_records:record.complete_fan_state=False
    assert analyze_fan_switch_event(dataset(data))['events']==[]


def three_check_case(*, after_flow=98, gap=0, policy=None):
    data = with_switch(sample(), after_flow=after_flow)
    event = data.switch_records[0]
    event.from_stopped_at = event.switched_at
    event.to_started_at = event.switched_at + timedelta(minutes=gap)
    normalized = normalize_and_validate_fan_data(data, policy or load_rules())
    basic = compare_basic_parameters(normalized)
    curves = analyze_fan_performance_curve(normalized)
    switches = analyze_fan_switch_event(normalized)
    return normalized, basic, curves, switches


@pytest.mark.parametrize('flow,code', [(90,'NORMAL'), (89.9,'WARNING'), (85.001,'WARNING'), (85,'ABNORMAL'), (80,'ABNORMAL')])
def test_flow_severity_uses_agreed_fifteen_percent_inclusive_boundary(flow, code):
    normalized, basic, curves, switches = three_check_case(after_flow=flow)
    decision = assess_fan_equivalent_capability(normalized, basic, curves, switches, {})
    assert decision['result_code'] == code
    assert bool(decision['severe_findings']) == (code == 'ABNORMAL')


def test_no_implicit_severe_timing_limit_and_explicit_configuration_is_effective():
    normalized, basic, curves, switches = three_check_case(gap=20)
    assert assess_fan_equivalent_capability(normalized, basic, curves, switches, {})['result_code'] == 'WARNING'
    normalized.policy.severity.abnormal_switch_interval_minutes = 20
    decision = assess_fan_equivalent_capability(normalized, basic, curves, switches, {})
    assert decision['result_code'] == 'ABNORMAL'
    assert decision['severe_findings'][0]['evidence_ids']


def test_unassessable_candidate_cannot_be_hidden_by_another_passing_switch():
    normalized, basic, curves, switches = three_check_case()
    switches['candidate_event_count'] = 2
    decision = assess_fan_equivalent_capability(normalized, basic, curves, switches, {})
    assert decision['result_code'] == 'UNDETERMINED'
    assert decision['items'] == {'parameters':'pass', 'switch_interval':'insufficient', 'flow_change':'insufficient'}


@pytest.mark.parametrize('field', ['abnormal_switch_interval_minutes', 'abnormal_flow_deviation_percent'])
def test_serious_limits_cannot_replace_the_normal_limit(field):
    policy = load_rules().model_dump()
    policy['severity'][field] = 10
    with pytest.raises(ValueError, match='严重'):
        RulePolicy.model_validate(policy)


@pytest.mark.asyncio
async def test_report_only_contains_supported_projects_and_backend_preserves_missing_states():
    result=await service.analyze(sample(angle=None).model_dump(mode='json'),'pair-u','pair-s')
    assert result['judgment_scope']=='main_fan_consistency'
    assert len(result['assessment']['fan_pair'])==2 and len(result['analysis']['calculations'])==14
    assert '1号通风机' in result['result'] and '2号通风机' in result['result']
    assert '基础参数一致性' in result['result'] and '性能测定数据比较' in result['result']
    for term in ('切换','备用风机状态','技术措施','需复核','缺少','请补充','实时监测','检验日期已过'):
        assert term not in result['result']
    assert result['assessment']['decision']['overall_status']=='insufficient_evidence'
    assert result['assessment']['stages'][3]['status']=='insufficient'
    assert result['missing_materials']==[]
    assert 'data:image/svg+xml;base64,' in result['result']


@pytest.mark.asyncio
async def test_switch_section_only_appears_when_data_is_present_and_metric_scope_is_preserved():
    result=await service.analyze(with_switch(sample(),pressure=False).model_dump(mode='json'),'flow-u','flow-s')
    assert '切换前后稳定工况比较' in result['result']
    event=result['assessment']['switch_comparison']['events'][0]
    assert event['assessed_metrics']==['flow']
    assert result['assessment']['switch_comparison']['status']=='insufficient'
    assert '风量（m³/min）' in result['result']


@pytest.mark.asyncio
async def test_phase_retry_reuses_actual_snapshot_and_enforces_ownership(monkeypatch):
    result=await service.analyze(sample().model_dump(mode='json'),'phase-u','phase-s')
    async def forbidden(*args,**kwargs):raise AssertionError('计算阶段重试不应重复识别')
    monkeypatch.setattr(service,'read_document',forbidden)
    original=main_fan_store.workflow(result['workflow_id'],'phase-u','phase-s')
    retried=await service.resume_assessment(result['workflow_id'],3,'phase-u','phase-s')
    after=main_fan_store.workflow(result['workflow_id'],'phase-u','phase-s')
    assert original['stages'][:2]==after['stages'][:2]
    assert result['result']==retried['result'] and result['analysis_id']!=retried['analysis_id']
    with pytest.raises(ValueError,match='当前会话'):
        await service.resume_assessment(result['workflow_id'],3,'another-u','phase-s')


@pytest.mark.asyncio
async def test_nine_stage_tools_only_return_saved_results_of_owned_workflow():
    from extensions.main_fan import tools
    result=await service.analyze(sample().model_dump(mode='json'),'stage-u','stage-s')
    runtime=SimpleNamespace(config={'configurable':{'workspace_user':'stage-u','workspace_session':'stage-s'}})
    for name in ('parse_fan_evidence_documents','query_fan_operating_data','normalize_and_validate_fan_data',
        'analyze_fan_performance_curve','analyze_fan_switch_event','match_fan_rules_and_knowledge',
        'assess_fan_equivalent_capability','trace_fan_evidence_and_review','render_fan_assessment_report'):
        output=getattr(tools,name)(result['workflow_id'],runtime)
        assert output['workflow_id']==result['workflow_id']
    wrong=SimpleNamespace(config={'configurable':{'workspace_user':'another','workspace_session':'stage-s'}})
    with pytest.raises(ValueError):tools.analyze_fan_performance_curve(result['workflow_id'],wrong)