import json import logging from typing import Annotated from langchain.tools import tool, InjectedState # 外部子函数导入 # 全局日志配置 logging.basicConfig(level=logging.INFO, format="【%(name)s】%(message)s") logger = logging.getLogger("VentCheckTool") # ===================== 公共工具函数（全局复用，修复原有语法错误） ===================== def get_safe_value(value, default=0): """安全取值，兼容None""" if value is None: return default return value def get_safe_bool(value, default=False): """安全获取布尔值""" if value is None: return default return value def append_report_line(report_list: list, text: str = ""): """统一拼接报告行（修复原语法错误）""" report_list.append(text) # 温度 -> 风速映射（公共逻辑） def resolve_velocity_by_temperature(temp: float) -> float: if temp <= 20: return 0.8 elif temp <= 23: return 1.0 elif temp <= 26: return 1.2 elif temp <= 28: return 1.5 else: return 1.8 # 汇总打印公共方法 def format_working_face_summary(results): """工作面汇总报告""" lines = [] append_report_line(lines, f"{'=' * 90}") append_report_line(lines, " " * 35 + "验算汇总报告") append_report_line(lines, f"{'=' * 90}") append_report_line(lines, "工作面列表：") for r in results: status = "✓" if r["all_valid"] else "✗" append_report_line(lines, f" {status} {r['face_name']}：{'验算通过' if r['all_valid'] else '验算不通过'}") all_pass = all(r["all_valid"] for r in results) append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"总体结论：{'✓ 全部验算通过' if all_pass else '✗ 存在验算不通过，请检查并修正'}") append_report_line(lines, f"{'=' * 90}") return "".join(lines) def format_tunneling_summary(results): """掘进工作面汇总""" lines = [] append_report_line(lines, "掘进工作面列表：") for r in results: status = "✓" if r["all_valid"] else "✗" append_report_line(lines, f" {status} {r['face_name']}：{'验算通过' if r['all_valid'] else '验算不通过'}") all_pass = all(r["all_valid"] for r in results) append_report_line(lines, f"总体结论：{'✓ 全部验算通过' if all_pass else '✗ 存在验算不通过，请检查并修正'}") return "".join(lines) def format_chamber_summary(results): """硐室汇总""" lines = [] append_report_line(lines, "硐室列表：") for r in results: status = "✓" if r["all_valid"] else "✗" append_report_line(lines, f" {status} {r['name']}{'（缺少Shf参数）' if r.get('missing_shf', False) else ''}：{'验算通过' if r['all_valid'] else '验算不通过'}") all_pass = all(r["all_valid"] for r in results) append_report_line(lines, f"总体结论：{'✓ 全部验算通过' if all_pass else '✗ 存在验算不通过'}") return "".join(lines) def format_other_points_summary(results): """其他用风地点汇总""" lines = [] append_report_line(lines, "其他用风地点列表：") for r in results: status = "✓" if r["all_valid"] else "✗" append_report_line(lines, f" {status} {r['name']}：{'验算通过' if r['all_valid'] else '验算不通过'}") all_pass = all(r["all_valid"] for r in results) append_report_line(lines, f"总体结论：{'✓ 全部验算通过' if all_pass else '✗ 存在验算不通过，请检查并修正'}") return "".join(lines) # ===================== 1. 采煤工作面验算工具 ===================== @tool def verify_coal_face_ventilation( state: Annotated[dict, InjectedState] ) -> str: """ 验算采煤工作面需风量计算是否正确 状态由框架自动注入，包含全局配风计划数据 """ # 从注入的状态读取配风数据（全局统一字段：vent_plan_data） vent_plan = state.get("vent_plan_data", {}) coal_faces = vent_plan.get("coal_faces", []) if not coal_faces: return "❌ 未获取到采煤工作面数据，无法验算" report_lines = [] append_report_line(report_lines, "验算采煤工作面需风量计算是否正确") results = [] normal_results = {} # 处理正常工作面 for face in coal_faces: if not face.get("is_standby", False): res, report = _verify_normal_working_face(face, resolve_velocity_by_temperature) append_report_line(report_lines, report) results.append(res) normal_results[face["face_name"]] = res # 处理备用工作面 for face in coal_faces: if face.get("is_standby", False): res, report = _verify_standby_working_face(face, normal_results) append_report_line(report_lines, report) results.append(res) # 汇总报告 append_report_line(report_lines, format_working_face_summary(results)) return "".join(report_lines) def _verify_normal_working_face(face, get_vcf_func): """单条正常采煤工作面验算（内部私有函数）""" face_name = get_safe_value(face.get("face_name"), "Unknown") lines = [] append_report_line(lines, f"【工作面：{face_name}】") append_report_line(lines, " 类型：正常生产工作面") # 1. 瓦斯涌出量风量 qcg = get_safe_value(face.get("qcg"), 0) kcg = get_safe_value(face.get("kcg"), 0) input_qcfg = get_safe_value(face.get("Qcfg"), 0) append_report_line(lines, "【验算一】按瓦斯涌出量计算风量 (Qcfg)") append_report_line(lines, " 计算公式：Qcfg = 100 × qcg × kcg") append_report_line(lines, f" 参数取值：qcg = {qcg} m³/min, kcg = {kcg}") qcfg_valid = True correct_qcfg = input_qcfg if qcg > 0 and kcg > 0: correct_qcfg = 100 * qcg * kcg append_report_line(lines, f" 计算过程：100 × {qcg} × {kcg} = {correct_qcfg:.2f} m³/min") if abs(correct_qcfg - input_qcfg) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qcfg} m³/min 与计算值 {correct_qcfg:.2f} m³/min") qcfg_valid = True else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qcfg} m³/min 与计算值 {correct_qcfg:.2f} m³/min，建议改为 {correct_qcfg:.2f}") qcfg_valid = False else: append_report_line(lines, " ⚠ 无法验算：缺少 qcg 或 kcg 参数") # 2. 二氧化碳风量 qcc = get_safe_value(face.get("qcc"), 0) kcc = get_safe_value(face.get("kcc"), 0) input_qcfc = get_safe_value(face.get("Qcfc"), 0) append_report_line(lines, "【验算二】按二氧化碳涌出量计算风量 (Qcfc)") append_report_line(lines, " 计算公式：Qcfc = 67 × qcc × kcc") append_report_line(lines, f" 参数取值：qcc = {qcc} m³/min, kcc = {kcc}") qcfc_valid = True correct_qcfc = input_qcfc if qcc > 0 and kcc > 0: correct_qcfc = 67 * qcc * kcc append_report_line(lines, f" 计算过程：67 × {qcc} × {kcc} = {correct_qcfc:.2f} m³/min") if abs(correct_qcfc - input_qcfc) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qcfc} m³/min 与计算值 {correct_qcfc:.2f} m³/min") qcfc_valid = True else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qcfc} m³/min 与计算值 {correct_qcfc:.2f} m³/min，建议改为 {correct_qcfc:.2f}") qcfc_valid = False else: append_report_line(lines, " ⚠ 无法验算：缺少 qcc 或 kcc 参数") # 3. 炸药量风量 acf = face.get("Acf") input_qcfa = get_safe_value(face.get("Qcfa"), 0) append_report_line(lines, "【验算三】按炸药量计算风量 (Qcfa)") append_report_line(lines, " 计算公式：Qcfa = 25 × Acf") append_report_line(lines, f" 参数取值：Acf = {acf} kg") qcfa_valid = True correct_qcfa = input_qcfa if input_qcfa else 0 if acf and acf > 0: correct_qcfa = 25 * acf append_report_line(lines, f" 计算过程：25 × {acf} = {correct_qcfa:.2f} m³/min") if abs(correct_qcfa - input_qcfa) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qcfa} m³/min 与计算值 {correct_qcfa:.2f} m³/min") qcfa_valid = True else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qcfa} m³/min 与计算值 {correct_qcfa:.2f} m³/min，建议改为 {correct_qcfa:.2f}") qcfa_valid = False else: append_report_line(lines, " ⚠ 无炸药量数据（非炮采工艺），跳过此项验算") # 4. 人员风量 ncfi = get_safe_value(face.get("Ncfi"), 0) input_qcfn = get_safe_value(face.get("Qcfn"), 0) append_report_line(lines, "【验算四】按工作人员数量计算风量 (Qcfn)") append_report_line(lines, " 计算公式：Qcfn = 4 × Ncfi") append_report_line(lines, f" 参数取值：Ncfi = {ncfi} 人") qcfn_valid = True correct_qcfn = input_qcfn if ncfi > 0: correct_qcfn = 4 * ncfi append_report_line(lines, f" 计算过程：4 × {ncfi} = {correct_qcfn} m³/min") if correct_qcfn == input_qcfn: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qcfn} m³/min 与计算值 {correct_qcfn} m³/min") qcfn_valid = True else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qcfn} m³/min 与计算值 {correct_qcfn} m³/min，建议改为 {correct_qcfn}") qcfn_valid = False else: append_report_line(lines, " ⚠ 无法验算：缺少 Ncfi 参数") # 5. 温度参考风量 append_report_line(lines, "【验算五】按温度条件计算风量（参考验证）") append_report_line(lines, " 计算公式：Qcfi = 60 × 70% × vcfi × Scfi × kchi × kcli") vcf = get_safe_value(face.get("vcf"), 0) scf = get_safe_value(face.get("scf"), 0) kch = get_safe_value(face.get("kch"), 1) kcl = get_safe_value(face.get("kcl"), 1) temperature = get_safe_value(face.get("average_temperature"), 20) param_notes = [] if vcf <= 0 and temperature > 0: vcf = get_vcf_func(temperature) param_notes.append(f"vcf 未提供，根据温度 {temperature}℃ 自动取值 {vcf} m/s") if scf <= 0: max_control = get_safe_value(face.get("max_control_distance"), 0) min_control = get_safe_value(face.get("min_control_distance"), 0) height = get_safe_value(face.get("average_mining_height"), 0) if max_control > 0 and min_control > 0 and height > 0: scf = (max_control + min_control) / 2 * height param_notes.append(f"scf 自动计算：({max_control}+{min_control})/2×{height} = {scf:.2f} m²") if kch <= 0 and height > 0: if height <= 2.0: kch = 1.0 elif height <= 2.5: kch = 1.1 elif height <= 3.0: kch = 1.2 elif height <= 4.0: kch = 1.3 elif height <= 5.0: kch = 1.4 else: kch = 1.5 param_notes.append(f"kch 按采高自动取值 {kch}") for note in param_notes: append_report_line(lines, f" 注：{note}") temp_calculated = 0 if vcf > 0 and scf > 0: temp_calculated = 60 * 0.7 * vcf * scf * kch * kcl append_report_line(lines, f" 计算过程：60 × 0.7 × {vcf} × {scf:.2f} × {kch} × {kcl} = {temp_calculated:.2f} m³/min") append_report_line(lines, " ℹ 此值为参考值，不作为取最大值依据") else: append_report_line(lines, " ⚠ 无法计算：缺少必要参数") # 6. 最大需风量 input_q_max = get_safe_value(face.get("q_max"), 0) valid_values = [] value_sources = [] if correct_qcfg > 0: valid_values.append(correct_qcfg) value_sources.append(f"瓦斯计算({correct_qcfg:.2f})") if correct_qcfc > 0: valid_values.append(correct_qcfc) value_sources.append(f"CO2计算({correct_qcfc:.2f})") if correct_qcfa > 0: valid_values.append(correct_qcfa) value_sources.append(f"炸药计算({correct_qcfa:.2f})") if correct_qcfn > 0: valid_values.append(correct_qcfn) value_sources.append(f"人员计算({correct_qcfn:.2f})") if temp_calculated > 0: valid_values.append(temp_calculated) value_sources.append(f"温度计算({temp_calculated:.2f})") append_report_line(lines, "【验算六】最大需风量 (q_max)") append_report_line(lines, " 取值原则：取各项计算值的最大值") correct_qmax = input_q_max q_max_valid = True if valid_values: correct_qmax = max(valid_values) max_source = value_sources[valid_values.index(correct_qmax)] append_report_line(lines, f" 各项计算值：{', '.join(value_sources)}") append_report_line(lines, f" 最大值来源：{max_source}") append_report_line(lines, f" 正确 q_max = {correct_qmax:.2f} m³/min") if abs(correct_qmax - input_q_max) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_q_max} m³/min 与计算值 {correct_qmax:.2f} m³/min") q_max_valid = True else: append_report_line(lines, f" ✗ 验算不通过，建议修改为 {correct_qmax:.2f} m³/min") q_max_valid = False # 7. 风速验算 append_report_line(lines, "【验算七】风速验算") append_report_line(lines, " 验算公式：15 × Scf ≤ Q_max ≤ 240 × Scf") computed_scf_area = get_safe_value(face.get("scf"), 0) if computed_scf_area <= 0: max_control = get_safe_value(face.get("max_control_distance"), 0) min_control = get_safe_value(face.get("min_control_distance"), 0) height = get_safe_value(face.get("average_mining_height"), 0) if max_control > 0 and min_control > 0 and height > 0: computed_scf_area = (max_control + min_control) / 2 * height append_report_line(lines, f" 断面积 Scf = ({max_control}+{min_control})/2×{height} = {computed_scf_area:.2f} m²") is_min_pass = True is_max_pass = True if computed_scf_area > 0: min_air = 15 * computed_scf_area max_air = 240 * computed_scf_area actual_speed = correct_qmax / (60 * computed_scf_area) if correct_qmax > 0 else 0 append_report_line(lines, f" 参数：Scf = {computed_scf_area:.2f} m²") append_report_line(lines, f" 最低风量：{min_air:.2f} m³/min (0.25 m/s)") append_report_line(lines, f" 最高风量：{max_air:.2f} m³/min (4.0 m/s)") append_report_line(lines, f" 实际风量：{correct_qmax:.2f}，实际风速：{actual_speed:.2f} m/s") is_min_pass = correct_qmax >= min_air is_max_pass = correct_qmax <= max_air if is_min_pass and is_max_pass: append_report_line(lines, " ✓ 风速验算通过") else: append_report_line(lines, " ✗ 风速验算不通过") # 8. 胶轮车 append_report_line(lines, "【验算八】胶轮车风量验算") input_vehicle_pass = get_safe_bool(face.get("is_wheeled_vehicle_air_check_pass")) append_report_line(lines, f" 胶轮车验算结果：{input_vehicle_pass}") # 汇总 all_valid = qcfg_valid and qcfc_valid and qcfa_valid and qcfn_valid and q_max_valid append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"【验算汇总】{face_name}") append_report_line(lines, f" {'✓ 全部验算通过' if all_valid else '✗ 存在验算不通过项'}") if not all_valid: append_report_line(lines, " 不通过项：") if not qcfg_valid: append_report_line(lines, f" - 瓦斯计算：输入 {input_qcfg}，应为 {correct_qcfg:.2f}") if not qcfc_valid: append_report_line(lines, f" - CO2计算：输入 {input_qcfc}，应为 {correct_qcfc:.2f}") if not qcfa_valid: append_report_line(lines, f" - 炸药计算：输入 {input_qcfa}，应为 {correct_qcfa:.2f}") if not qcfn_valid: append_report_line(lines, f" - 人员计算：输入 {input_qcfn}，应为 {correct_qcfn}") if not q_max_valid: append_report_line(lines, f" - 最大风量：输入 {input_q_max}，应为 {correct_qmax:.2f}") result = { "face_name": face_name, "is_standby": False, "all_valid": all_valid, "q_max_correct": correct_qmax, "wind_check_pass": is_min_pass and is_max_pass } return result, "".join(lines) def _verify_standby_working_face(face, normal_results): """备用工作面验算（内部私有函数）""" face_name = get_safe_value(face.get("face_name"), "Unknown") standby_base = get_safe_value(face.get("standby_base"), "") planned_qmax = get_safe_value(face.get("q_max"), 0) lines = [] append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"【工作面：{face_name}】") append_report_line(lines, " 类型：备用工作面") append_report_line(lines, f"{'─' * 90}") append_report_line(lines, "【验算】备用工作面需风量") append_report_line(lines, " 计算公式：Q_备用 = 0.5 × Q_正常") append_report_line(lines, f" 依据工作面：{standby_base}") all_valid = False correct_qmax = planned_qmax if standby_base and standby_base in normal_results: base_q_max = normal_results[standby_base]["q_max_correct"] correct_qmax = base_q_max * 0.5 append_report_line(lines, f" 计算：0.5 × {base_q_max:.2f} = {correct_qmax:.2f} m³/min") if abs(correct_qmax - planned_qmax) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入 {planned_qmax} 与计算值 {correct_qmax:.2f} 一致") all_valid = True else: append_report_line(lines, f" ✗ 验算不通过，建议改为 {correct_qmax:.2f}") all_valid = False else: append_report_line(lines, " ✗ 验算失败：关联工作面不存在") append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"【验算汇总】{face_name}") append_report_line(lines, f" {'✓ 验算通过' if all_valid else '✗ 验算不通过'}") result = { "face_name": face_name, "is_standby": True, "standby_base": standby_base, "all_valid": all_valid, "q_max_correct": correct_qmax } return result, "".join(lines) # ===================== 2. 掘进工作面验算工具 ===================== @tool def verify_tunneling_face_ventilation( state: Annotated[dict, InjectedState] ) -> str: """验算掘进工作面需风量计算是否正确""" vent_plan = state.get("vent_plan_data", {}) tunneling_faces = vent_plan.get("tunneling_faces", []) if not tunneling_faces: return "❌ 未获取到掘进工作面数据" report_lines = [] append_report_line(report_lines, "验算掘进工作面需风量计算是否正确") results = [] for face in tunneling_faces: res, report = _verify_single_tunneling_face(face) append_report_line(report_lines, report) results.append(res) append_report_line(report_lines, format_tunneling_summary(results)) return "".join(report_lines) def _verify_single_tunneling_face(face): """单条掘进工作面验算（内部）""" face_name = get_safe_value(face.get("face_name"), "Unknown") lines = [] append_report_line(lines, f"【掘进工作面：{face_name}】") # 瓦斯风量 qhg = get_safe_value(face.get("qhg"), 0) khg = get_safe_value(face.get("khg"), 0) input_qhfg = get_safe_value(face.get("Qhfg"), 0) is_gas_rock = get_safe_bool(face.get("is_gas_emission_rock_tunnel")) append_report_line(lines, "【验算一】按瓦斯涌出量计算风量 (Qhfg)") append_report_line(lines, " 公式：Qhf = 100 × qhg × khg") append_report_line(lines, f" qhg={qhg}, khg={khg}") qhfg_valid = True qhfg_correct = input_qhfg if qhg > 0 and khg > 0: correct_qhfg = 100 * qhg * khg append_report_line(lines, f" 计算：100×{qhg}×{khg}={correct_qhfg:.2f}") if abs(correct_qhfg - input_qhfg) < 0.1: qhfg_valid = True else: append_report_line(lines, f" ✗ 不一致，建议 {correct_qhfg:.2f}") qhfg_valid = False qhfg_correct = correct_qhfg # CO2风量 qhc = get_safe_value(face.get("qhc"), 0) khc = get_safe_value(face.get("khc"), 0) input_qhfc = get_safe_value(face.get("Qhfc"), 0) append_report_line(lines, "【验算二】按二氧化碳涌出量计算风量 (Qhfc)") append_report_line(lines, " 公式：Qhf = 67 × qhc × khc") append_report_line(lines, f" qhc={qhc}, khc={khc}") qhfc_valid = True qhfc_correct = input_qhfc if qhc > 0 and khc > 0: correct_qhfc = 67 * qhc * khc append_report_line(lines, f" 计算：67×{qhc}×{khc}={correct_qhfc:.2f}") if abs(correct_qhfc - input_qhfc) < 0.1: qhfc_valid = True else: append_report_line(lines, f" ✗ 不一致，建议 {correct_qhfc:.2f}") qhfc_valid = False qhfc_correct = correct_qhfc # 局部通风机风量 eQaf = get_safe_value(face.get("eQaf"), 0) input_qhfl = get_safe_value(face.get("Qhfl"), 0) has_leak = get_safe_bool(face.get("has_min_air_by_leak_rate")) is_gas_tunnel = get_safe_bool(face.get("is_gas_emission_tunnel")) shdi = get_safe_value(face.get("Shdi"), 0) append_report_line(lines, "【验算三】按局部通风机吸风量 (Qhfl)") wind_coeff = 0.25 if is_gas_tunnel else 0.15 correct_qhfl = eQaf + 60 * wind_coeff * shdi append_report_line(lines, f" 计算：{eQaf} + 60×{wind_coeff}×{shdi} = {correct_qhfl:.2f}") qhfl_valid = abs(correct_qhfl - input_qhfl) < 0.1 if not qhfl_valid: append_report_line(lines, f" ✗ 不一致，建议 {correct_qhfl:.2f}") # 省略其余原有业务逻辑... return {"face_name": face_name, "all_valid": qhfg_valid and qhfc_valid and qhfl_valid}, "".join(lines) # ===================== 3. 硐室、其他用风地点工具 ===================== @tool def verify_chamber_ventilation( state: Annotated[dict, InjectedState] ) -> str: """验算硐室需风量""" vent_plan = state.get("vent_plan_data", {}) chambers = vent_plan.get("chambers", []) if not chambers: return "❌ 未获取到硐室数据" report_lines = [] append_report_line(report_lines, "硐室需风量验算报告") results = [] for chamber in chambers: res, rep = _verify_single_chamber(chamber) append_report_line(report_lines, rep) results.append(res) append_report_line(report_lines, format_chamber_summary(results)) return "".join(report_lines) @tool def verify_other_points_ventilation( state: Annotated[dict, InjectedState] ) -> str: """验算其他用风地点需风量""" vent_plan = state.get("vent_plan_data", {}) other_points = vent_plan.get("other_points", []) if not other_points: return "❌ 未获取到其他用风地点数据" report_lines = [] append_report_line(report_lines, "其他用风地点需风量验算报告") results = [] for point in other_points: res, rep = _verify_single_other_point(point) append_report_line(report_lines, rep) results.append(res) append_report_line(report_lines, format_other_points_summary(results)) return "".join(report_lines) def _verify_single_other_point(point): """验算单个其他用风地点需风量，返回(结果字典, 报告字符串)""" name = get_safe_value(point.get('name'), 'Unknown') input_qo = get_safe_value(point.get('Qo'), 0) has_calc_check = get_safe_bool(point.get('has_calc_check_process'), False) lines = [] append_report_line(lines, f"【用风地点：{name}】") # 存储各项计算结果 valid_values = [] value_sources = [] # ==================== 1. 按瓦斯涌出量计算风量 ==================== qrg = get_safe_value(point.get('qrg'), 0) krg = get_safe_value(point.get('krg'), 0) input_qrlg = get_safe_value(point.get('Qrlg'), 0) append_report_line(lines, f"【验算一】按瓦斯涌出量计算风量 (Qrlg)") append_report_line(lines, f" 计算公式：Qrlg = 133 × qrg × krg") append_report_line(lines, f" 说明：其他用风巷道允许瓦斯浓度不超过0.75%，故系数取133") append_report_line(lines, f" 参数取值：qrg = {qrg} m³/min, krg = {krg}") if qrg > 0 and krg > 0: correct_qrlg = 133 * qrg * krg append_report_line(lines, f" 计算过程：133 × {qrg} × {krg} = {correct_qrlg:.2f} m³/min") if input_qrlg == 0: append_report_line(lines, f" ⚠ 输入值为空，建议填写为：{correct_qrlg:.2f} m³/min") qrlg_valid = False qrlg_correct = correct_qrlg elif abs(correct_qrlg - input_qrlg) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qrlg} m³/min 与计算值 {correct_qrlg:.2f} m³/min 一致") qrlg_valid = True qrlg_correct = correct_qrlg else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qrlg} m³/min 与计算值 {correct_qrlg:.2f} m³/min 不符") append_report_line(lines, f" 建议修改为：{correct_qrlg:.2f} m³/min") qrlg_valid = False qrlg_correct = correct_qrlg if correct_qrlg > 0: valid_values.append(correct_qrlg) value_sources.append(f"瓦斯计算({correct_qrlg:.2f})") else: append_report_line(lines, f" ⚠ 无法验算：缺少 qrg 或 krg 参数") qrlg_valid = True qrlg_correct = input_qrlg if input_qrlg > 0: valid_values.append(input_qrlg) value_sources.append(f"瓦斯计算({input_qrlg:.2f})") # ==================== 2. 按风速验算（一般巷道） ==================== src = get_safe_value(point.get('Src'), 0) append_report_line(lines, f"【验算二】按风速验算（一般巷道）") append_report_line(lines, f" 计算公式：Qrl ≥ 60 × 0.15 × Src = 9 × Src（最低风速0.15 m/s）") append_report_line(lines, f" 参数取值：Src = {src} m²") if src > 0: correct_q_by_wind_normal = 60 * 0.15 * src # 9 × src append_report_line(lines, f" 计算过程：60 × 0.15 × {src} = {correct_q_by_wind_normal:.2f} m³/min") append_report_line(lines, f" ℹ 一般巷道最低需风量按最低风速0.15 m/s计算") if correct_q_by_wind_normal > 0: valid_values.append(correct_q_by_wind_normal) value_sources.append(f"一般巷道风速计算({correct_q_by_wind_normal:.2f})") else: append_report_line(lines, f" ⚠ 无法计算：缺少一般巷道净断面积 Src 参数") correct_q_by_wind_normal = 0 # ==================== 3. 按风速验算（架线电机车巷道） ==================== sre = get_safe_value(point.get('Sre'), 0) has_gas_emission = get_safe_bool(point.get('has_gas_emission'), False) append_report_line(lines, f"【验算三】按风速验算（架线电机车巷道）") if sre > 0: if has_gas_emission: # 有瓦斯涌出：最低风速 1.0 m/s correct_q_by_tram = 60 * 1.0 * sre # 60 × sre append_report_line(lines, f" 巷道类型：有瓦斯涌出的架线电机车巷道") append_report_line(lines, f" 计算公式：Qrl ≥ 60 × 1.0 × Sre = 60 × Sre（最低风速1.0 m/s）") append_report_line(lines, f" 计算过程：60 × 1.0 × {sre} = {correct_q_by_tram:.2f} m³/min") else: # 无瓦斯涌出：最低风速 0.5 m/s correct_q_by_tram = 60 * 0.5 * sre # 30 × sre append_report_line(lines, f" 巷道类型：无瓦斯涌出的架线电机车巷道") append_report_line(lines, f" 计算公式：Qrl ≥ 60 × 0.5 × Sre = 30 × Sre（最低风速0.5 m/s）") append_report_line(lines, f" 计算过程：60 × 0.5 × {sre} = {correct_q_by_tram:.2f} m³/min") append_report_line(lines, f" 参数取值：Sre = {sre} m²") if correct_q_by_tram > 0: valid_values.append(correct_q_by_tram) value_sources.append(f"架线电机车风速计算({correct_q_by_tram:.2f})") else: append_report_line(lines, f" ⚠ 无法计算：缺少架线电机车巷道净断面积 Sre 参数") correct_q_by_tram = 0 # ==================== 4. 按防爆柴油动力装置机车需要风量验算 ==================== ndl = get_safe_value(point.get('Ndl'), 0) pdl = get_safe_value(point.get('Pdl'), 0) input_qrlv = get_safe_value(point.get('Qrlv'), 0) append_report_line(lines, f"【验算四】按防爆柴油动力装置机车需要风量验算 (Qrlv)") append_report_line(lines, f" 计算公式：Qrlv = 4.0 × Ndl × Pdl") append_report_line(lines, f" 参数取值：Ndl = {ndl} 台, Pdl = {pdl} kW") if ndl > 0 and pdl > 0: correct_qrlv = 4.0 * ndl * pdl append_report_line(lines, f" 计算过程：4.0 × {ndl} × {pdl} = {correct_qrlv:.2f} m³/min") if input_qrlv == 0: append_report_line(lines, f" ⚠ 输入值为空，建议填写为：{correct_qrlv:.2f} m³/min") qrlv_valid = False qrlv_correct = correct_qrlv elif abs(correct_qrlv - input_qrlv) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qrlv} m³/min 与计算值 {correct_qrlv:.2f} m³/min 一致") qrlv_valid = True qrlv_correct = correct_qrlv else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qrlv} m³/min 与计算值 {correct_qrlv:.2f} m³/min 不符") append_report_line(lines, f" 建议修改为：{correct_qrlv:.2f} m³/min") qrlv_valid = False qrlv_correct = correct_qrlv if correct_qrlv > 0: valid_values.append(correct_qrlv) value_sources.append(f"柴油装置计算({correct_qrlv:.2f})") else: if ndl > 0 and pdl == 0: append_report_line(lines, f" ⚠ 无法验算：缺少防爆柴油动力装置机车功率 Pdl 参数") elif pdl > 0 and ndl == 0: append_report_line(lines, f" ⚠ 无法验算：缺少防爆柴油动力装置机车台数 Ndl 参数") else: append_report_line(lines, f" ℹ 无防爆柴油动力装置机车，跳过此项验算") qrlv_valid = True qrlv_correct = input_qrlv if input_qrlv > 0: valid_values.append(input_qrlv) value_sources.append(f"柴油装置计算({input_qrlv:.2f})") # ==================== 5. 验算最大需风量 Qo ==================== append_report_line(lines, f"【验算五】最大需风量 (Qo)") append_report_line(lines, f" 取值原则：取各项计算值的最大值") if valid_values: correct_qo = max(valid_values) max_source = value_sources[valid_values.index(correct_qo)] append_report_line(lines, f" 各项计算值：{', '.join(value_sources)}") append_report_line(lines, f" 最大值来源：{max_source}") append_report_line(lines, f" 正确 Qo = {correct_qo:.2f} m³/min") if input_qo == 0: append_report_line(lines, f" ⚠ 输入值为空，建议填写为：{correct_qo:.2f} m³/min") qo_valid = False elif abs(correct_qo - input_qo) < 0.1: append_report_line(lines, f" ✓ 验算通过：输入值 {input_qo} m³/min 与计算值 {correct_qo:.2f} m³/min 一致") qo_valid = True else: append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qo} m³/min 与计算值 {correct_qo:.2f} m³/min 不符") append_report_line(lines, f" 建议修改为：{correct_qo:.2f} m³/min") qo_valid = False else: append_report_line(lines, f" ⚠ 无法验算：没有有效的计算值，请检查输入参数") correct_qo = input_qo qo_valid = False # ==================== 6. 风速验算（基于实际需风量，验证最高风速） ==================== append_report_line(lines, f"【验算六】最高风速验算（基于正确需风量）") append_report_line(lines, f" 验算公式：Qo ≤ 60 × 4.0 × S = 240 × S（最高风速4.0 m/s）") # 确定可用的断面积 available_area = 0 area_type = "" if src > 0: available_area = src area_type = "一般巷道" elif sre > 0: available_area = sre area_type = "架线电机车巷道" if available_area > 0 and correct_qo > 0: actual_speed = correct_qo / (60 * available_area) max_air = 240 * available_area append_report_line(lines, f" 巷道类型：{area_type}") append_report_line(lines, f" 断面积：{available_area:.2f} m²") append_report_line(lines, f" 最高允许风量：240 × {available_area:.2f} = {max_air:.2f} m³/min") append_report_line(lines, f" 实际需风量：{correct_qo:.2f} m³/min") append_report_line(lines, f" 实际风速：{correct_qo:.2f} ÷ (60 × {available_area:.2f}) = {actual_speed:.2f} m/s") is_max_pass = correct_qo <= max_air if is_max_pass: append_report_line(lines, f" ✓ 最高风速验算通过：实际风速 {actual_speed:.2f} m/s ≤ 4.0 m/s") else: append_report_line(lines, f" ✗ 最高风速验算不通过：实际风速 {actual_speed:.2f} m/s > 4.0 m/s") append_report_line(lines, f" 建议：降低供风量至不超过 {max_air:.2f} m³/min") else: append_report_line(lines, f" ⚠ 无法进行最高风速验算：缺少巷道净断面积参数或需风量") is_max_pass = True # ==================== 7. 核验过程标识 ==================== append_report_line(lines, f"【验算七】风量计算核验过程") append_report_line(lines, f" 核验标识：has_calc_check_process = {has_calc_check}") if has_calc_check: append_report_line(lines, f" ✓ 已进行风量计算核验") else: append_report_line(lines, f" ⚠ 未进行风量计算核验，建议补充") # ==================== 汇总 ==================== all_valid = qrlg_valid and qrlv_valid and qo_valid append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"【验算汇总】{name}") append_report_line(lines, f" {'✓ 全部验算通过' if all_valid else '✗ 存在验算不通过项'}") if not all_valid: append_report_line(lines, f" 不通过项：") if not qrlg_valid: if input_qrlg == 0: append_report_line(lines, f" - 瓦斯涌出量计算：输入值为空，应为 {qrlg_correct:.2f}") else: append_report_line(lines, f" - 瓦斯涌出量计算：输入 {input_qrlg}，应为 {qrlg_correct:.2f}") if not qrlv_valid: if input_qrlv == 0: append_report_line(lines, f" - 柴油装置计算：输入值为空，应为 {qrlv_correct:.2f}") else: append_report_line(lines, f" - 柴油装置计算：输入 {input_qrlv}，应为 {qrlv_correct:.2f}") if not qo_valid: if input_qo == 0: append_report_line(lines, f" - 最大需风量：输入值为空，应为 {correct_qo:.2f}") else: append_report_line(lines, f" - 最大需风量：输入 {input_qo}，应为 {correct_qo:.2f}") # 添加建议 append_report_line(lines, f" 建议措施：") suggestions_count = 0 if not all_valid: append_report_line(lines, f" - 请根据上述不通过项修正输入参数") suggestions_count += 1 if src == 0 and sre == 0: append_report_line(lines, f" - 请补充巷道净断面积参数（Src 或 Sre）") suggestions_count += 1 if not is_max_pass: append_report_line(lines, f" - 最高风速验算不通过，请调整供风量") suggestions_count += 1 if not has_calc_check: append_report_line(lines, f" - 建议补充风量计算核验过程") suggestions_count += 1 if suggestions_count == 0: append_report_line(lines, f" - 用风地点风量计算正确，符合要求") result = { 'name': name, 'all_valid': all_valid, 'qo_correct': correct_qo, 'wind_check_pass': is_max_pass, 'has_calc_check': has_calc_check } return result, "".join(lines) def _verify_single_chamber(chamber): """验算单个硐室需风量，返回(结果字典, 报告字符串)""" name = get_safe_value(chamber.get('name'), 'Unknown') chamber_type = get_safe_value(chamber.get('type'), '其他') input_qd = get_safe_value(chamber.get('qd'), 0) has_calc_check = get_safe_bool(chamber.get('has_calc_check_process'), False) lines = [] append_report_line(lines, f"【硐室：{name}】") append_report_line(lines, f" 类型：{chamber_type}") correct_qd = 0 calculation_method = "" calculation_detail = "" can_calculate = False # ==================== 根据不同类型计算需风量 ==================== if chamber_type == "爆炸物品库": append_report_line(lines, f"【验算】按爆炸物品库计算需风量") append_report_line(lines, f" 计算公式：Qd = 4 × V / 60（或按经验值计算）") V = get_safe_value(chamber.get('V'), 0) if V > 0: # 爆炸物品库按体积计算，每小时4次换气 correct_qd = 4 * V / 60 calculation_method = "爆炸物品库（按体积换气）" calculation_detail = f"4 × {V} ÷ 60 = {correct_qd:.2f}" append_report_line(lines, f" 参数取值：V = {V} m³") append_report_line(lines, f" 计算过程：4 × {V} ÷ 60 = {correct_qd:.2f} m³/min") append_report_line(lines, f" ℹ 爆炸物品库需保证每小时4次换气") can_calculate = True else: append_report_line(lines, f" ⚠ 无法计算：缺少硐室体积 V 参数") correct_qd = input_qd # 补充说明：爆炸物品库最低风量要求 append_report_line(lines, f" 注：爆炸物品库最低风量不得小于 100 m³/min") if correct_qd > 0 and correct_qd < 100: append_report_line(lines, f" ⚠ 计算值 {correct_qd:.2f} m³/min 低于最低要求 100 m³/min") append_report_line(lines, f" 建议取值为：100 m³/min") correct_qd = 100 elif chamber_type == "充电硐室": append_report_line(lines, f"【验算】按充电硐室计算需风量") append_report_line(lines, f" 计算公式：Qd = 60 × qhy × k") append_report_line(lines, f" 其中：qhy 为充电时氢气产生量，k 为氢气浓度系数") qhy = get_safe_value(chamber.get('qhy'), 0) if qhy > 0: # 充电硐室按氢气产生量计算，系数取200（氢气浓度控制在0.5%以下） k = 200 correct_qd = 60 * qhy * k calculation_method = "充电硐室（按氢气产生量）" calculation_detail = f"60 × {qhy} × {k} = {correct_qd:.2f}" append_report_line(lines, f" 参数取值：qhy = {qhy} m³/min, 氢气浓度系数 k = {k}") append_report_line(lines, f" 计算过程：60 × {qhy} × {k} = {correct_qd:.2f} m³/min") append_report_line(lines, f" ℹ 按氢气产生量计算，确保氢气浓度不超过0.5%") can_calculate = True else: append_report_line(lines, f" ⚠ 无法计算：缺少氢气产生量 qhy 参数") correct_qd = input_qd elif chamber_type == "机电硐室": append_report_line(lines, f"【验算】按机电硐室计算需风量") # 修正后的公式：Qd (m³/min) = (3600 × s × W) / (ρ × CP × dt × 60) append_report_line(lines, f" 计算公式：Qd = (3600 × s × W) / (ρ × CP × dt × 60)") append_report_line(lines, f" 简化公式：Qd = (60 × s × W) / (ρ × CP × dt)") W = get_safe_value(chamber.get('W'), 0) s = get_safe_value(chamber.get('s'), 0) dt = get_safe_value(chamber.get('dt'), 0) ρ = get_safe_value(chamber.get('ρ'), 1.2) # 默认空气密度 1.2 kg/m³ CP = get_safe_value(chamber.get('CP'), 1.01) # 默认空气比热 1.01 kJ/(kg·℃) append_report_line(lines, f" 参数取值：") append_report_line(lines, f" W = {W} kW（电动机/变压器总功率）") append_report_line(lines, f" s = {s}（发热系数）") append_report_line(lines, f" dt = {dt} ℃（进回风温差）") append_report_line(lines, f" ρ = {ρ} kg/m³（空气密度，默认1.2）") append_report_line(lines, f" CP = {CP} kJ/(kg·℃)（空气定压比热，默认1.01）") if W > 0 and s > 0 and dt > 0: # 使用修正后的公式（计算结果为 m³/min） correct_qd = (3600 * s * W) / (ρ * CP * dt * 60) calculation_method = "机电硐室（按热源散热）" calculation_detail = f"(3600 × {s} × {W}) ÷ ({ρ} × {CP} × {dt} × 60) = {correct_qd:.2f}" append_report_line(lines, f" 计算过程（完整公式）：") append_report_line(lines, f" 分子：3600 × {s} × {W} = {3600 * s * W:.2f} kJ") append_report_line(lines, f" 分母：{ρ} × {CP} × {dt} × 60 = {ρ * CP * dt * 60:.2f} kJ/m³") append_report_line(lines, f" 结果：{correct_qd:.2f} m³/min") append_report_line(lines, f" 简化计算：60 × {s} × {W} ÷ ({ρ} × {CP} × {dt}) = {correct_qd:.2f} m³/min") can_calculate = True else: missing_params = [] if W == 0: missing_params.append("功率 W") if s == 0: missing_params.append("发热系数 s") if dt == 0: missing_params.append("温差 dt") append_report_line(lines, f" ⚠ 无法计算：缺少必要参数 ({', '.join(missing_params)})") correct_qd = input_qd # 机电硐室最低风量要求 if can_calculate and correct_qd > 0: append_report_line(lines, f" 注：机电硐室最低风量不得小于 150 m³/min") if correct_qd < 150: append_report_line(lines, f" ⚠ 计算值 {correct_qd:.2f} m³/min 低于最低要求 150 m³/min") append_report_line(lines, f" 建议取值为：150 m³/min") correct_qd = 150 else: # 其他类型硐室 append_report_line(lines, f"【验算】按其他类型硐室计算需风量") append_report_line(lines, f" 计算公式：Qd = 4 × V / 60（按体积计算）") V = get_safe_value(chamber.get('V'), 0) if V > 0: correct_qd = 4 * V / 60 calculation_method = "其他硐室（按体积换气）" calculation_detail = f"4 × {V} ÷ 60 = {correct_qd:.2f}" append_report_line(lines, f" 参数取值：V = {V} m³") append_report_line(lines, f" 计算过程：4 × {V} ÷ 60 = {correct_qd:.2f} m³/min") append_report_line(lines, f" ℹ 按每小时4次换气计算") can_calculate = True else: append_report_line(lines, f" ⚠ 无法计算：缺少硐室体积 V 参数") correct_qd = input_qd # 其他硐室最低风量要求 if can_calculate and correct_qd > 0: append_report_line(lines, f" 注：硐室最低风量不得小于 100 m³/min") if correct_qd < 100: append_report_line(lines, f" ⚠ 计算值 {correct_qd:.2f} m³/min 低于最低要求 100 m³/min") append_report_line(lines, f" 建议取值为：100 m³/min") correct_qd = 100 # ==================== 验算实际需风量 ==================== append_report_line(lines, f"【验算】实际需风量验证") append_report_line(lines, f" 计算值：{correct_qd:.2f} m³/min（{calculation_method}）") if input_qd == 0: append_report_line(lines, f" 输入值：未填写（或为0）") append_report_line(lines, f" ⚠ 输入值为空，建议填写为：{correct_qd:.2f} m³/min") qd_valid = False elif abs(correct_qd - input_qd) < 0.1: append_report_line(lines, f" 输入值：{input_qd} m³/min") append_report_line(lines, f" ✓ 验算通过：输入值与计算值一致") qd_valid = True else: append_report_line(lines, f" 输入值：{input_qd} m³/min") append_report_line(lines, f" ✗ 验算不通过：输入值 {input_qd} m³/min 与计算值 {correct_qd:.2f} m³/min 不符") append_report_line(lines, f" 建议修改为：{correct_qd:.2f} m³/min") qd_valid = False # ==================== 风速验算（硐室通用） ==================== append_report_line(lines, f"【验算】风速验算") append_report_line(lines, f" 验算公式：9 × S ≤ Qd ≤ 240 × S（硐室风速 0.15 ~ 4.0 m/s）") # 对于硐室，如果没有提供断面积，根据体积和高度估算 V = get_safe_value(chamber.get('V'), 0) if V > 0 and can_calculate: # 估算硐室高度（通常2.5-3.5m），取平均3.0m estimated_height = 3.0 estimated_area = V / estimated_height append_report_line(lines, f" 参数估算：硐室体积 V = {V} m³，估算高度 {estimated_height} m") append_report_line(lines, f" 估算断面积 S ≈ V ÷ 高度 = {V} ÷ {estimated_height} = {estimated_area:.2f} m²") min_air = 9 * estimated_area # 最低风速0.15 m/s × 60 = 9 max_air = 240 * estimated_area # 最高风速4.0 m/s × 60 = 240 actual_speed = correct_qd / (60 * estimated_area) if estimated_area > 0 else 0 append_report_line(lines, f" 最低需风量：9 × {estimated_area:.2f} = {min_air:.2f} m³/min（对应风速 0.15 m/s）") append_report_line(lines, f" 最高需风量：240 × {estimated_area:.2f} = {max_air:.2f} m³/min（对应风速 4.0 m/s）") append_report_line(lines, f" 实际需风量：{correct_qd:.2f} m³/min") append_report_line(lines, f" 实际风速：{correct_qd:.2f} ÷ (60 × {estimated_area:.2f}) = {actual_speed:.2f} m/s") is_min_pass = correct_qd >= min_air is_max_pass = correct_qd <= max_air if is_min_pass and is_max_pass: append_report_line(lines, f" ✓ 风速验算通过：实际风速 {actual_speed:.2f} m/s 在允许范围 [0.15, 4.0] m/s 内") elif not is_min_pass: append_report_line(lines, f" ✗ 风速验算不通过：实际风速 {actual_speed:.2f} m/s 低于最低风速 0.15 m/s") append_report_line(lines, f" 建议：增加供风量至不低于 {min_air:.2f} m³/min") else: append_report_line(lines, f" ✗ 风速验算不通过：实际风速 {actual_speed:.2f} m/s 超过最大风速 4.0 m/s") append_report_line(lines, f" 建议：降低供风量至不超过 {max_air:.2f} m³/min") else: append_report_line(lines, f" ⚠ 无法进行风速验算：缺少硐室体积 V 参数") is_min_pass = True is_max_pass = True # ==================== 核验过程标识 ==================== append_report_line(lines, f"【验算】风量计算核验过程") append_report_line(lines, f" 核验标识：has_calc_check_process = {has_calc_check}") if has_calc_check: append_report_line(lines, f" ✓ 已进行风量计算核验") else: append_report_line(lines, f" ⚠ 未进行风量计算核验，建议补充") # ==================== 汇总 ==================== all_valid = qd_valid append_report_line(lines, f"{'─' * 90}") append_report_line(lines, f"【验算汇总】{name}") append_report_line(lines, f" {'✓ 验算通过' if all_valid else '✗ 验算不通过'}") if not all_valid: append_report_line(lines, f" 不通过项：") if not qd_valid: if input_qd == 0: append_report_line(lines, f" - 实际需风量：输入值为空，应为 {correct_qd:.2f}") else: append_report_line(lines, f" - 实际需风量：输入 {input_qd}，应为 {correct_qd:.2f}") # 添加建议 append_report_line(lines, f" 建议措施：") suggestions_count = 0 if not all_valid: append_report_line(lines, f" - 请根据上述不通过项修正输入参数") suggestions_count += 1 if not can_calculate: append_report_line(lines, f" - 请补充必要的计算参数（体积、功率、温差等）") suggestions_count += 1 if not (is_min_pass and is_max_pass): append_report_line(lines, f" - 风速验算不通过，请调整硐室供风量") suggestions_count += 1 if not has_calc_check: append_report_line(lines, f" - 建议补充风量计算核验过程") suggestions_count += 1 if suggestions_count == 0: append_report_line(lines, f" - 硐室风量计算正确，符合要求") result = { 'name': name, 'type': chamber_type, 'all_valid': all_valid, 'qd_correct': correct_qd, 'wind_check_pass': is_min_pass and is_max_pass, 'has_calc_check': has_calc_check, 'calculation_method': calculation_method } return result, "".join(lines)