"""双机范围、额定功率、受控插值、真实切换窗口及报告项目选择。""" 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)