| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869 |
- """合成测试输入，与扫描报告的自动识别事实独立。"""
- from copy import deepcopy
- from datetime import datetime, timedelta
- from algorithms.main_fan.models import ScreeningInput
- PRESETS = (
- {'id':'normal','name':'主备运行正常','description':'参数一致，切换6分钟，风量变化5%。','expected':'NORMAL','tag':'规则样例'},
- {'id':'warning','name':'风量轻度偏差','description':'参数一致，切换6分钟，风量变化12%。','expected':'WARNING','tag':'规则样例'},
- {'id':'abnormal','name':'风量明显偏差','description':'参数一致，切换6分钟，风量变化20%。','expected':'ABNORMAL','tag':'规则样例'},
- {'id':'insufficient','name':'缺少切换记录','description':'保留双机报告参数，未提供有效切换时序。','expected':'UNDETERMINED','tag':'规则样例'},
- {'id':'documents','name':'PDF资料研判','description':'选择原始PDF或上传资料，配合主备开停与风量曲线研判。','expected':None,'tag':'原件识别'},
- {'id':'gateway','name':'网关取数联调','description':'从当前已配置网关取得资料，运行真实取数与研判流程。','expected':None,'tag':'网关资料'},
- )
- def preset(identifier):
- try:
- return deepcopy(next(p for p in PRESETS if p['id']==identifier))
- except StopIteration:
- raise ValueError('测试样例不存在') from None
- def main_fan_input(identifier):
- preset(identifier)
- at = datetime.fromisoformat('2026-10-14T10:00:00+08:00')
- common = {'mine_code':'mock-zy','mine_name':'陕西众源煤业集团有限公司'}
- evidence = {'source':'test-lab://main-fan/'+identifier,'method':'synthetic','label':'智能体测试样例','excerpt':'合成数据，用于验证主备风机判识流程'}
- ids = ['mock-report-fan-1','mock-report-fan-2']
- data = {**common,'fan_id':ids[0],'fan_name':'1号通风机','fan_pair':ids,
- 'evaluation_time':'2026-10-14T12:00:00+08:00','use_gateway':identifier=='gateway',
- 'document_cache_policy':'auto','paired_assessment':True,'reports':[],
- 'devices':[{'device_code':key,'name':f'{i+1}号通风机',**common,'evidence':evidence} for i,key in enumerate(ids)],
- 'fan_state_records':[],'operating_series':[]}
- if identifier=='gateway':
- data['devices']=[]
- if identifier not in {'documents','gateway'}:
- for i in range(2):
- points = []
- for frequency in (30,40,50):
- factor = frequency/30
- for j in range(7):
- flow, pressure, power = (42+j*5)*factor, (920-j*95)*factor**2, (55-j)*factor**3
- if i:
- pressure *= .99
- power *= .995
- points.append({'frequency_hz':frequency,'flow_m3_s':flow,'static_pressure_pa':pressure,
- 'shaft_power_kw':power,'reported_efficiency_pct':flow*pressure/(10*power),
- 'label':f'测点{j+1}','evidence':evidence})
- data['reports'].append({**common,'report_id':f'TEST-LAB-{i+1}','fan_name':f'{i+1}号通风机',
- 'serial_number':f'TEST-{i+1}','model':'FBCDZ-№24','motor_model':'YBF2355L1-8',
- 'rated_power_raw':'2×185 kW','rated_speed_rpm':740,'impeller_diameter_m':2.4,
- 'blade_angle_deg':35,'inspected_on':'2026-10-01','points':points,'evidence':evidence})
- if identifier not in {'insufficient','gateway'}:
- after = {'warning':.88,'abnormal':.8}.get(identifier,.95)*3800
- values = [[],[]]
- for minute in range(-40,61):
- time = at+timedelta(minutes=minute)
- for i,fan_id in enumerate(ids):
- running = minute<0 if i==0 else minute>=6
- data['fan_state_records'].append({**common,'fan_id':fan_id,'observed_at':time.isoformat(),
- 'state':'running' if running else 'stopped','complete_fan_state':True,'evidence':evidence})
- values[i].append({'observed_at':time.isoformat(),'value':(3800 if i==0 else after) if running else 0})
- data['operating_series']=[{**common,'fan_id':fan_id,'metric':'flow','unit':'m3/min',
- 'sensor_id':f'test-flow-{i+1}','mapping_confirmed':True,'samples':values[i],'evidence':evidence} for i,fan_id in enumerate(ids)]
- # 预先验证样例，禁止无效或隐含人工转录的输入进入工作台。
- core={k:v for k,v in data.items() if k in ScreeningInput.model_fields}
- ScreeningInput.model_validate(core)
- return data
|
