{title}

"""只呈现本次资料支持的双机一致性分析，不把未开展项目写进报告。""" import base64 from datetime import datetime from html import escape import re from algorithms.main_fan.analysis import LOCAL_ZONE, local_time from algorithms.main_fan.equivalence import three_check_statuses from extensions.main_fan.reporting import _text from extensions.main_fan.result_output import judgment_result def _value(field, value): if field == "rated_power": parts = value["components_kw"] expression = f"{len(parts)}×{parts[0]:g}" if len(set(parts)) == 1 and len(parts) > 1 else "+".join(f"{p:g}" for p in parts) return f"{expression} kW（合计{value['total_kw']:g} kW）" if len(parts) > 1 else f"{value['total_kw']:g} kW" if isinstance(value, list): value = "～".join(f"{v:g}" for v in value) units = {"rated_speed_rpm": " r/min", "impeller_diameter_m": " m", "rated_flow_range_m3_min": " m³/min", "rated_pressure_range_pa": " Pa"} if isinstance(value, (int, float)): value = f"{value:g}" return _text(value) + units.get(field, "") def performance_chart(curves, names): """按物理量分别画双机曲线；与差异计算共用插值点，不混用纵轴。""" charts = [] metrics = ((0, "静压", "Pa", 0), (1, "轴功率", "kW", 1), (2, "静压效率", "%", 1)) # 多于三个频率或角度组合时继续分页绘图，不丢弃后续曲线。 for index, label, unit, decimals in metrics: for begin in range(0, len(curves), 3): panels = curves[begin:begin+3] charts.append(_performance_metric_chart(panels, names, index, label, unit, decimals)) return "\n\n".join(charts) def _performance_metric_chart(panels, names, index, label, unit, decimals): width, height, panel_width = 1080, 365, 1080/len(panels) title = f"两台主通风机风量—{label}对比曲线" body = [f'') encoded = base64.b64encode("".join(body).encode()).decode() return f"![{title}](data:image/svg+xml;base64,{encoded})" def switch_chart(event, names): """同一时间轴展示受控开停序列和风量，开停采用阶梯线。""" chart = event['chart_data'] state_rows, flow_rows = chart['state_curves'], chart['flow_curves'] rows = state_rows + flow_rows timestamps = [local_time(datetime.fromisoformat(p['observed_at'])).timestamp() for row in rows for p in row['samples']] if not timestamps or min(timestamps) == max(timestamps): return '' low, high = min(timestamps), max(timestamps) left, plot_width = 80, 770 flow_top = 225 if state_rows else 45 bottom = flow_top+145 if flow_rows else 175 height = bottom+75 x = lambda at: left+(at-low)/(high-low)*plot_width colors = {event['from_fan_id']:'#74b6ff',event['to_fan_id']:'#ffca76'} body = [f'') return '![主备风机开停与风量时序曲线](data:image/svg+xml;base64,'+base64.b64encode(''.join(body).encode()).decode()+')' def render_fan_assessment_report(dataset, assessment, analysis): fans, basic, curves, switches = dataset.fans, assessment["basic_comparison"], assessment["performance_comparison"], assessment["switch_comparison"] names = [f["fan_name"] for f in fans] reports = [f["report"] for f in fans if f["report"]] now = local_time(dataset.data.evaluation_time or datetime.now(LOCAL_ZONE)).date() lines = [f"**研判对象：** {_text(dataset.data.mine_name)} {'、'.join(_text(n) for n in names)}", f"**研判日期：** {now.isoformat()}", ""] section = 0 def heading(title): nonlocal section section += 1 lines.extend(["", f"## {'一二三四五六七八九十'[section-1]}、{title}", ""]) heading("综合研判结论") code, short_result = judgment_result(assessment, analysis) lines.extend([f"**研判结果：{code}**。{short_result}", ""]) items = three_check_statuses(basic, switches) summary = [] if basic["rows"]: mismatches = [r["label"] for r in basic["rows"] if r["status"] == "mismatch"] summary.append(("两台风机参数一致性", "、".join(mismatches)+"存在差异" if mismatches else "已比较的基础参数一致", "parameters")) events = switches["events"] if events: intervals = [f"{e['metrics']['switch_interval_minutes']:g}分钟（上限{e['timing']['limit_minutes']:g}分钟）" for e in events if e['metrics'].get('switch_interval_minutes') is not None] flows = [f"{e['metrics']['flow_deviation_percent']:.2f}%（上限{e['limits']['flow_deviation_percent']:g}%）" for e in events if e['metrics'].get('flow_deviation_percent') is not None] if intervals: summary.append(("停机至启动间隔", "；".join(intervals), "switch_interval")) if flows: summary.append(("切换前后稳定风量差异", "；".join(flows), "flow_change")) if summary: lines.extend(["| 判识项目 | 本次结果 | 分项结论 |", "| --- | --- | --- |"]) labels = {"pass": "通过", "fail": "不通过", "insufficient": "未形成有效对照"} for label, detail, key in summary: lines.append(f"| {label} | {detail} | {labels[items[key]]} |") lines.append("") compared = [r["label"] for r in basic["rows"]] if basic["rows"]: mismatches = [r["label"] for r in basic["rows"] if r["status"] == "mismatch"] sentence = f"本次对{'与'.join(_text(n) for n in names)}的检测报告进行联合研判，重点核对两台设备的报告参数及性能测定数据。" sentence += "基础参数比较发现"+"、".join(mismatches)+"存在差异。" if mismatches else "已比较的"+"、".join(compared)+"一致。" lines.append(sentence) elif reports: lines.append("本次依据已取得的主通风机检测报告整理设备身份及性能记录，研判范围以本次实际参与比较的数据为准。") elif switches["events"]: lines.append(f"本次依据{'、'.join(_text(n) for n in names)}的运行记录与同步监测序列，分析实际切换事件前后的稳定参数变化。") else: lines.append("本次已完成资料整理，尚未形成可发布的双机一致性比较结果。") if curves["results"]: frequencies = "、".join(f"{r['frequency_hz']:g} Hz" for r in curves["results"]) max_pressure = max(r["metrics"]["maximum_pressure_deviation_percent"] for r in curves["results"]) max_eta = max(r["metrics"]["efficiency_difference_points"] for r in curves["results"]) lines.extend(["", f"两份报告共同记录了{frequencies}现场测定数据。本次在各自共同风量区间内比较，最大静压相对偏差为{max_pressure:.2f}%，最大静压效率差为{max_eta:.2f}个百分点。偏差均以表列数据为依据，计算结果及运行工况比较见下文。"]) formal = [r for r in curves["results"] if r["comparability"] == "same_angle_frequency"] if formal: lines.extend(["", "相同叶片角度、相同频率下的可比曲线" + ("存在超过本次配置阈值的差异。" if any(r['status']=='fail' for r in formal) else "各项差异处于本次配置阈值内。")]) if switches["events"]: synthetic = any(e['chart_data']['synthetic'] for e in switches['events']) sentence = f"本次分析{len(switches['events'])}次{'模拟' if synthetic else '运行'}切换事件，分别判断停机至启动间隔和前后稳定风量差异。" sentence += '两项均符合本次规则。' if all(items[k]=='pass' for k in ('switch_interval','flow_change')) else '具体分项结果见切换工况分析。' lines.extend(['',sentence]) if reports: heading("研判资料与设备身份") lines.extend(["本次使用的检测资料按矿井和风机身份分别归集，两台设备的原始参数保持独立，比较过程中统一风量、静压和功率单位。", "", "| 设备 | 报告编号 | 出厂编号 | 检测日期 |", "| --- | --- | --- | --- |"]) for fan in fans: report = fan["report"] if report: lines.append(f"| {_text(fan['fan_name'])} | {_text(report.report_id)} | {_text(report.serial_number)} | {report.inspected_on.isoformat() if report.inspected_on else ''} |") measured = sum(p.condition == "measured" for r in reports for p in r.points) converted = sum(p.condition == "converted" for r in reports for p in r.points) lines.extend(["", f"两份报告共整理现场实测记录{measured}行、折算记录{converted}行。性能比较使用现场实测记录；折算记录单独保留，未作为现场实测工况参与两机差异计算。检测日期和出厂编号用于确认资料对象及所属检测批次，出厂编号不同表明资料分别对应两台设备。"]) if basic["rows"]: heading("报告基础参数一致性") lines.extend([f"| 比较项目 | {_text(names[0])} | {_text(names[1])} | 比较结果 |", "| --- | --- | --- | --- |"]) for row in basic["rows"]: values = [_value(row["field"], value) for value in row["values"]] lines.append(f"| {row['label']} | {values[0]} | {values[1]} | {'一致' if row['status']=='match' else '不一致'} |") power = next((r for r in basic["rows"] if r["field"] == "rated_power"), None) if power: lines.extend(["", "额定功率比较采用报告基本参数表中的电动机铭牌配置，保留电动机组成并统一计算总额定功率。组合表达式由计算工具解析，运行工况下的轴功率及电输入功率另行用于性能分析。"]) matching = [r["label"] for r in basic["rows"] if r["status"] == "match"] if matching: lines.extend(["", "上述一致项表明，两台设备在已载明的"+"、".join(matching)+"方面具有一致的报告参数。基础参数一致性以型号、额定功率及双方共同载明的设备参数为依据。"]) if basic["backup_availability"]: lines.extend(["", "运行记录所列备用风机状态为" + ("可用。" if basic['backup_availability']['status']=='available' else "不可用。")]) if curves["results"]: heading("报告性能测定数据比较") lines.extend(["性能测定数据作为补充分析，单独说明两机性能曲线差异，不增加综合判识的必过条件。本次按共同检测频率分别组织两台风机的风量—静压、风量—轴功率和风量—静压效率数据，在双方原始测量范围的交集内进行分段线性插值。比较范围覆盖率按共同风量跨度占各自测量跨度的比例计算，所有比较点均位于原始测量范围内。", "", "| 频率 | 共同风量区间（m³/min） | 平均静压偏差 | 最大静压偏差 | 最大效率差（百分点） |", "| --- | ---: | ---: | ---: | ---: |"]) for result in curves["results"]: m, q = result["metrics"], result["common_flow_range_m3_min"] lines.append(f"| {result['frequency_hz']:g} Hz | {q[0]:.2f}～{q[1]:.2f} | {m['mean_pressure_deviation_percent']:.2f}% | {m['maximum_pressure_deviation_percent']:.2f}% | {m['efficiency_difference_points']:.2f} |") details = "；".join(f"{r['frequency_hz']:g} Hz共同区间覆盖率{r['common_flow_coverage']*100:.2f}%、静压均方根误差{r['metrics']['pressure_rmse_pa']:.2f} Pa、最大轴功率偏差{r['metrics']['maximum_power_deviation_percent']:.2f}%" for r in curves['results']) lines.extend(["", details+"。相对偏差以表格第一台设备为基准；效率差按百分点表达，避免与相对百分比混用。", "", performance_chart(curves['results'], names)]) operating = [r for r in curves["results"] if r["operating_points"]] if operating: lines.extend(["", "**报告所列运行工况比较。** 以下为两份检测报告明示的运行工况，分别保留各自对应的风量、静压及功率。", "", "| 频率与设备 | 风量（m³/min） | 静压（Pa） | 轴功率（kW） | 静压效率（%） |", "| --- | ---: | ---: | ---: | ---: |"]) operating_notes = [] for result in operating: for name, point in zip(names, result['operating_points']): lines.append(f"| {result['frequency_hz']:g} Hz {_text(name)} | {point['flow_m3_min']:.2f} | {point['pressure_pa']:.2f} | {point['shaft_power_kw']:.2f} | {point['efficiency_pct']:.2f} |") m = result["metrics"] pieces = [] if m.get('operating_flow_deviation_percent') is not None: pieces.append(f"风量偏差{m['operating_flow_deviation_percent']:.2f}%") if m.get('operating_pressure_deviation_percent') is not None: pieces.append(f"静压偏差{m['operating_pressure_deviation_percent']:.2f}%") if pieces: operating_notes.append(f"{result['frequency_hz']:g} Hz所列运行工况的"+"、".join(pieces)+"。这些参数描述各自检测时的运行点，与公共风量区间内的曲线数据差异分别计算。") for note in operating_notes: lines.extend(["", note]) if switches["events"]: heading("切换前后稳定工况比较") lookup = {f['fan_id']: f['fan_name'] for f in fans} for event in switches['events']: if event['chart_data']['synthetic']: lines.extend(['本节采用明确标注的开停、风量模拟时序，验证切换间隔和风量差异两项业务规则。','']) lines.extend([f"**切换事件：** {event['switched_at']}，{_text(lookup[event['from_fan_id']])}切换至{_text(lookup[event['to_fan_id']])}。", ""]) timing = event['timing'] if timing['interval_minutes'] is not None: sentence = f"原运行风机于{timing['from_stopped_at']}停机，另一台风机于{timing['to_started_at']}启动，切换间隔为**{timing['interval_minutes']:.2f}分钟**。" sentence += f"按不超过{timing['limit_minutes']:g}分钟的规则，判定为**{'通过' if timing['status']=='pass' else '超限'}**。" if timing['overlap_minutes']: sentence += f"两机运行重叠{timing['overlap_minutes']:.2f}分钟，停机等待间隔按0分钟计。" lines.extend([sentence,'']) chart = switch_chart(event,lookup) if chart: lines.extend([chart,'']) if event['changes']: lines.extend(["| 监测指标 | 切换前稳定均值 | 切换后稳定均值 | 数值变化 | 相对变化 |", "| --- | ---: | ---: | ---: | ---: |"]) for metric, change in event['changes'].items(): flow = metric == 'flow' label, factor, unit = ('风量', 60, 'm³/min') if flow else ('静压' if metric=='static_pressure' else '风压', 1, 'Pa') lines.append(f"| {label}（{unit}） | {change['before']*factor:.2f} | {change['after']*factor:.2f} | {change['absolute_change']*factor:+.2f} | {change['relative_change_percent']:+.2f}% |") if event['changes']: window = next(iter(event['before_window'].values())) after = next(iter(event['after_window'].values())) lines.extend(["", f"前稳定窗口为{window['start']}至{window['end']}，后稳定窗口为{after['start']}至{after['end']}。统计采用有效样本，窗口完整率分别为{window['completeness']*100:.2f}%和{after['completeness']*100:.2f}%。"]) if 'flow' in event['changes']: change = event['changes']['flow'] limit = event['limits']['flow_deviation_percent'] lines.extend(['',f"风量差异按 |切换后均值－切换前均值| ÷ 切换前均值 × 100% 计算，即 |{change['after']*60:.2f}－{change['before']*60:.2f}| ÷ {change['before']*60:.2f} × 100%＝**{event['metrics']['flow_deviation_percent']:.2f}%**。按不超过{limit:g}%的规则，判定为**{'通过' if event['checks']['flow_change']=='pass' else '超限'}**。"]) if event['metrics']['recovery_time_minutes'] is not None: lines.extend(["", f"切换后在{event['metrics']['recovery_time_minutes']:.1f}分钟开始形成连续恢复窗口。"]) if all(v == 'pass' for v in event['checks'].values()): lines.extend(['','本次切换间隔与稳定风量差异均符合配置规则，支持该次工况下的主备切换及供风衔接结论。']) consistency = [f for f in analysis.findings if f.code in {'curve_table_disagreement', 'report_formula_inconsistent', 'curve_formula_inconsistent'}] if consistency: heading("检测资料一致性事项") lines.extend(["以下事项记录检测资料自身的表图或公式差异，独立于基础参数、切换间隔和稳定风量三项判识，不作为综合结论的否决项。", ""]) for fan in fans: report = fan['report'] if not report: continue for caption in report.curves: points = [p for p in report.points if p.condition==caption.condition and p.frequency_hz==caption.frequency_hz and '运行工况' in p.label and abs(p.flow_m3_s-caption.flow_m3_s)<=max(.001,caption.flow_m3_s*.005) and abs(p.shaft_power_kw-caption.shaft_power_kw)<=max(.001,caption.shaft_power_kw*.005)] if len(points)!=1: continue row = points[0] pressure_diff = abs(row.static_pressure_pa-caption.static_pressure_pa) eta_diff = abs(row.reported_efficiency_pct-caption.reported_efficiency_pct) if row.reported_efficiency_pct is not None and caption.reported_efficiency_pct is not None else None if pressure_diff <= max(1,row.static_pressure_pa*.005) and (eta_diff is None or eta_diff <= .5): continue text = f"{_text(fan['fan_name'])}{caption.frequency_hz:g} Hz报告的数据表静压为{row.static_pressure_pa:.2f} Pa，曲线图注为{caption.static_pressure_pa:.2f} Pa，相差{pressure_diff:.2f} Pa。" if eta_diff is not None: text += f"表列静压效率为{row.reported_efficiency_pct:.2f}%，图注为{caption.reported_efficiency_pct:.2f}%，相差{eta_diff:.2f}个百分点。" lines.extend([text+"本次两机性能数据比较采用数据表，图注数值独立保留。", ""]) formula = [f for f in consistency if f.code in {'report_formula_inconsistent','curve_formula_inconsistent'}] if formula: lines.append(f"本次发现{len(formula)}项性能数值与同一组参数的效率公式不能一致对应，已在原始计算记录中保留具体数值和原文位置。") lines.extend(["", "建议将上述表图差异交由检测机构说明，统一正式采用的报表与曲线参数口径，避免在后续两机性能比较中交叉使用不同来源数值。"]) historical = [c for c in analysis.calculations if c.scope=='historical_report'] if historical: differences = [c.difference_percentage_points for c in historical if c.difference_percentage_points is not None] if differences: lines.extend(["", f"本次同时检查了两台风机共{len(historical)}行历史表格数据的静压效率公式，计算值与表列效率的最大差值为{max(differences):.4f}个百分点。逐行原始数值、公式及文件位置已与报告一并保存。"]) return re.sub(r"(\*\*[^*\n]+\*\*)(?=\S)", r"\1 ", "\n".join(lines))