风机性能曲线测绘与节能运行区间寻优系统 —— 基于 OOP 的数据驱动实战风机是工厂里的电老虎——一台 200kW 的离心风机每年电费动辄百万。很多人以为把风门开到 80% 就是最佳工况其实可能正踩在低效区的边缘。要找到真正的节能甜点你得先画出它的性能曲线然后在上面找到那个又省电又能干活的区间。—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在水泥、钢铁、化工、电厂等行业大型通风机/引风机是最主要的耗电设备之一。典型的风机系统结构如下┌──────────────────────────────────────────────┐│ 风机系统运行简图 ││ ││ 入口风门 ──→ [离心风机] ──→ 出口管网 ││ (挡板/变频) │ ││ ├── 电流传感器 (AI) ││ ├── 风量传感器 (AI) ││ ├── 风压传感器 (AI) ││ └── 转速传感器 (AI/DI) │└──────────────────────────────────────────────┘风机性能的核心关系参数 符号 单位 说明风量 Q m³/h 单位时间内输送的气体体积风压 H Pa 风机产生的压差电流 I A 电机输入电流反映轴功率效率 η % 有效功率 / 轴功率转速 n rpm 叶轮旋转速度典型工况与能耗工况 风门开度 电流 风量 问题满负荷 100% 320A 80000 m³/h 正常但未必经济常用负荷 70% 280A 56000 m³/h 可能落在低效区低负荷 40% 220A 32000 m³/h 风门节流严重浪费大变频改造后 变频调速 ? ? 需要重新测绘最优区间哈尔滨工程大学《工业过程控制》课程彭秀艳教授主讲国家级一流本科课程在第四章执行器与调节阀和第七章节能控制中涉及了风机水泵的节能原理风机水泵的功耗与转速的三次方成正比相似定律。通过调节转速而非挡板开度来控制流量是实现节能的根本途径。但在实施变频调速之前必须先掌握设备的实际性能曲线才能确定最佳运行区间。二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因盲目变频 上了变频器电费反而没降多少 没有基于性能曲线设定最优转速数据沉睡 DCS 里存了几年的电流风量数据没人分析 缺乏自动化分析工具铭牌≠实际 铭牌上说效率 82%实测只有 65% 管路阻力变化、磨损导致偏离设计点调度凭经验 操作员凭感觉调风门 没有量化的能效基准审计无据 节能改造验收拿不出前后对比 缺少系统化的性能测绘手段2.2 核心矛盾风机的设计性能曲线是在出厂试验台上测的用的是标准空气密度、清洁管道。到了现场管路阻力变了、叶轮积灰了、风门特性偏了——实际性能曲线早就漂移了。你拿着过期的地图在找路当然找不到节能的出口。2.3 我们要解决什么用一段 Python 程序构建一个风机性能曲线测绘与节能运行区间寻优系统实现1. 数据采集与清洗 —— 从 CSV/数据库读取电流、风量、风压时序数据2. 性能曲线拟合 —— 多项式回归得到 H-Q、P-Q、η-Q 曲线3. 高效区识别 —— 基于效率阈值划定经济运行区间4. 节能潜力计算 —— 对比当前运行点与最优点的能耗差5. 可视化输出 —— 四合一曲线图H-Q / P-Q / η-Q / I-Q6. 面向对象设计 —— 分层清晰可扩展三、核心逻辑讲解3.1 理论基础风机相似定律与性能曲线本工具基于哈工程《工业过程控制》第七章节能控制① 风机基本方程H A - B \cdot Q^2 \quad \text{(管网阻力曲线)}P \frac{Q \cdot H}{3600 \cdot 1000 \cdot \eta} \quad \text{(轴功率 kW)}\eta \frac{Q \cdot H}{P_{shaft} \cdot 3600 \cdot 1000} \times 100\%② 相似定律\frac{Q_1}{Q_2} \frac{n_1}{n_2}, \quad \frac{H_1}{H_2} \left(\frac{n_1}{n_2}\right)^2, \quad \frac{P_1}{P_2} \left(\frac{n_1}{n_2}\right)^3③ 效率计算从电流反推P_{electrical} \sqrt{3} \cdot U \cdot I \cdot \cos\phi \cdot \eta_{motor}P_{shaft} P_{electrical} \cdot \eta_{motor} \cdot \eta_{coupling}\eta_{fan} \frac{Q \cdot H}{P_{shaft} \cdot 3600 \cdot 1000}④ 最优运行区间判定高效区: η ≥ η_max × 90%经济运行区间: 同时满足 η ≥ 阈值 AND Q ∈ [Q_min, Q_max]节能潜力: ΔP P_current - P_optimal (kW)3.2 数据分析流程┌──────────────────────────────┐│ 数据源 (CSV/DB) ││ (timestamp, Q, H, I, U...) │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ① 数据加载 清洗 ││ 异常值剔除、滤波平滑 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ② 功率与效率计算 ││ P_elec → P_shaft → η │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ③ 曲线拟合 ││ H-Q: 二次多项式 ││ η-Q: 高斯或多项式 ││ P-Q: 三次多项式 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ④ 最优区间寻优 ││ 效率最高点 高效带宽 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ⑤ 可视化 报告 ││ 四合一曲线 节能建议 │└──────────────────────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式FanSample 单条采样记录dataclass 值对象FanConfig 风机参数配置值对象 值对象CurveConfig 曲线拟合配置值对象 值对象DataLoader 数据加载与清洗 封装PowerEfficiencyCalculator 功率与效率计算 策略模式CurveFitter 多项式曲线拟合 封装OptimalZoneFinder 最优运行区间寻优 封装EnergySavingsEstimator 节能潜力估算 封装ReportGenerator 分析报告生成器 模板方法FanPerformanceSystem 系统编排器聚合根 聚合根4.2 数据模型层from dataclasses import dataclassfrom typing import List, Tuple, Optionalfrom enum import Enumimport numpy as npimport csvfrom pathlib import Pathclass EfficiencyGrade(Enum):能效等级EXCELLENT 优秀GOOD 良好ACCEPTABLE 合格POOR 低效dataclass(frozenTrue)class FanSample:单条风机采样记录 —— 值对象timestamp: float # 时间戳flow: float # 风量 (m³/h)pressure: float # 风压 (Pa)current: float # 电流 (A)voltage: float 380.0 # 电压 (V)power_factor: float 0.85 # 功率因数speed: float 0.0 # 转速 (rpm, 可选)dataclass(frozenTrue)class FanConfig:风机参数配置motor_power: float 250.0 # 电机额定功率 (kW)motor_efficiency: float 0.94 # 电机效率coupling_efficiency: float 0.98 # 联轴器效率design_flow: float 80000.0 # 设计风量 (m³/h)design_pressure: float 3500.0 # 设计风压 (Pa)dataclass(frozenTrue)class CurveConfig:曲线拟合配置poly_degree_hq: int 2 # H-Q 曲线多项式阶数poly_degree_pq: int 3 # P-Q 曲线多项式阶数efficiency_threshold: float 0.75 # 高效区最低效率min_samples: int 20 # 最少样本数4.3 数据加载器class DataLoader:风机运行数据加载器CSV 格式:timestamp,flow,pressure,current,voltage,power_factor,speed100.5,65000,2800,245.5,380,0.86,1480101.0,67000,2750,252.0,380,0.85,1480...def __init__(self, file_path: str ):self.file_path Path(file_path)self.samples: List[FanSample] []def load_csv(self) - List[FanSample]:从 CSV 加载数据self.samples.clear()with open(self.file_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:sample FanSample(timestampfloat(row[timestamp]),flowfloat(row[flow]),pressurefloat(row[pressure]),currentfloat(row[current]),voltagefloat(row.get(voltage, 380)),power_factorfloat(row.get(power_factor, 0.85)),speedfloat(row.get(speed, 0)))self.samples.append(sample)return self.samplesdef clean(self, flow_range: Tuple[float, float] (10000, 150000),current_range: Tuple[float, float] (10, 500)) - List[FanSample]:数据清洗 —— 剔除明显异常值过滤条件:1. 风量在合理范围内2. 电流在合理范围内3. 风压 0cleaned []for s in self.samples:if (flow_range[0] s.flow flow_range[1] andcurrent_range[0] s.current current_range[1] ands.pressure 0):cleaned.append(s)return cleaned4.4 功率与效率计算器class PowerEfficiencyCalculator:功率与效率计算器计算公式:P_elec √3 × U × I × cosφP_shaft P_elec × η_motor × η_couplingη_fan (Q × H) / (P_shaft × 3600 × 1000)def __init__(self, fan_config: FanConfig):self.cfg fan_configdef calc_electrical_power(self, sample: FanSample) - float:计算输入电功率 (kW)return np.sqrt(3) * sample.voltage * sample.current * sample.power_factor / 1000.0def calc_shaft_power(self, sample: FanSample) - float:计算轴功率 (kW)p_elec self.calc_electrical_power(sample)return p_elec * self.cfg.motor_efficiency * self.cfg.coupling_efficiencydef calc_efficiency(self, sample: FanSample) - float:计算风机效率Returns:效率 (0~1)p_shaft self.calc_shaft_power(sample)if p_shaft 0:return 0.0# 有效功率 (kW)p_useful sample.flow * sample.pressure / (3600.0 * 1000.0)return min(1.0, p_useful / p_shaft)def grade_efficiency(self, eta: float) - EfficiencyGrade:能效等级评定if eta 0.80:return EfficiencyGrade.EXCELLENTelif eta 0.70:return EfficiencyGrade.GOODelif eta 0.55:return EfficiencyGrade.ACCEPTABLEelse:return EfficiencyGrade.POORdef batch_calculate(self, samples: List[FanSample]) - List[dict]:批量计算功率和效率results []for s in samples:p_elec self.calc_electrical_power(s)p_shaft self.calc_shaft_power(s)eta self.calc_efficiency(s)results.append({sample: s,p_elec: round(p_elec, 2),p_shaft: round(p_shaft, 2),efficiency: round(eta, 4),grade: self.grade_efficiency(eta).value})return results4.5 曲线拟合器核心算法class CurveFitter:风机性能曲线拟合器拟合:H-Q: 二次多项式 H a₀ a₁·Q a₂·Q²P-Q: 三次多项式 P b₀ b₁·Q b₂·Q² b₃·Q³η-Q: 二次多项式 (抛物线形)def __init__(self, config: CurveConfig):self.cfg configself.hq_coeffs: Optional[np.ndarray] Noneself.pq_coeffs: Optional[np.ndarray] Noneself.eq_coeffs: Optional[np.ndarray] Nonedef fit_hq_curve(self, flows: np.ndarray, pressures: np.ndarray) - np.ndarray:拟合 H-Q 曲线 (压力-流量)Args:flows: 风量数组pressures: 风压数组Returns:多项式系数 [a₀, a₁, a₂]# 按流量排序sort_idx np.argsort(flows)flows_sorted flows[sort_idx]pressures_sorted pressures[sort_idx]self.hq_coeffs np.polyfit(flows_sorted, pressures_sorted, self.cfg.poly_degree_hq)return self.hq_coeffsdef fit_pq_curve(self, flows: np.ndarray, powers: np.ndarray) - np.ndarray:拟合 P-Q 曲线 (功率-流量)Args:flows: 风量数组powers: 轴功率数组Returns:多项式系数 [b₀, b₁, b₂, b₃]sort_idx np.argsort(flows)flows_sorted flows[sort_idx]powers_sorted powers[sort_idx]self.pq_coeffs np.polyfit(flows_sorted, powers_sorted, self.cfg.poly_degree_pq)return self.pq_coeffsdef fit_eq_curve(self, flows: np.ndarray, efficiencies: np.ndarray) - np.ndarray:拟合 η-Q 曲线 (效率-流量)Args:flows: 风量数组efficiencies: 效率数组Returns:多项式系数sort_idx np.argsort(flows)flows_sorted flows[sort_idx]eff_sorted efficiencies[sort_idx]self.eq_coeffs np.polyfit(flows_sorted, eff_sorted, 2)return self.eq_coeffsdef predict_h(self, q: float) - float:根据 H-Q 曲线预测风压if self.hq_coeffs is None:raise ValueError(H-Q curve not fitted yet)return float(np.polyval(self.hq_coeffs, q))def predict_p(self, q: float) - float:根据 P-Q 曲线预测功率if self.pq_coeffs is None:raise ValueError(P-Q curve not fitted yet)return float(np.polyval(self.pq_coeffs, q))def predict_eta(self, q: float) - float:根据 η-Q 曲线预测效率if self.eq_coeffs is None:raise ValueError(η-Q curve not fitted yet)return float(np.polyval(self.eq_coeffs, q))4.6 最优运行区间寻优器class OptimalZoneFinder:最优运行区间寻优器寻找:1. 最高效率点 (Best Efficiency Point, BEP)2. 高效运行区间 (η ≥ η_max × 90%)3. 经济运行区间 (同时满足效率和需求约束)def __init__(self, curve_fitter: CurveFitter, flow_range: Tuple[float, float]):self.fitter curve_fitterself.flow_min, self.flow_max flow_rangeself.bep_flow: float 0.0self.bep_efficiency: float 0.0self.optimal_zone: Tuple[float, float] (0.0, 0.0)def find_bep(self, resolution: float 500.0) - Tuple[float, float]:寻找最高效率点 (BEP)Args:resolution: 搜索分辨率 (m³/h)Returns:(BEP流量, BEP效率)best_flow self.flow_minbest_eta 0.0for q in np.arange(self.flow_min, self.flow_max resolution, resolution):eta self.fitter.predict_eta(q)if eta best_eta:best_eta etabest_flow qself.bep_flow best_flowself.bep_efficiency best_etareturn best_flow, best_etadef find_optimal_zone(self, threshold_ratio: float 0.90) - Tuple[float, float]:寻找高效运行区间Args:threshold_ratio: 效率阈值比例 (相对于最高效率)Returns:(区间下限, 区间上限)if self.bep_efficiency 0.0:self.find_bep()threshold self.bep_efficiency * threshold_ratiozone_start self.flow_minzone_end self.flow_maxin_zone Falsefor q in np.arange(self.flow_min, self.flow_max 500, 500):eta self.fitter.predict_eta(q)if eta threshold and not in_zone:zone_start qin_zone Trueelif eta threshold and in_zone:zone_end qbreakself.optimal_zone (zone_start, zone_end)return self.optimal_zonedef get_summary(self) - dict:获取寻优结果摘要return {bep_flow: round(self.bep_flow, 0),bep_efficiency: round(self.bep_efficiency, 4),optimal_zone_start: round(self.optimal_zone[0], 0),optimal_zone_end: round(self.optimal_zone[1], 0),optimal_zone_width: round(self.optimal_zone[1] - self.optimal_zone[0], 0)}4.7 节能潜力估算器class EnergySavingsEstimator:节能潜力估算器对比:1. 当前运行点的功率 vs 最优运行点的功率2. 年运行小时数下的节电量def __init__(self, curve_fitter: CurveFitter, electricity_price: float 0.65):Args:curve_fitter: 已拟合的曲线electricity_price: 电价 (元/kWh)self.fitter curve_fitterself.price electricity_pricedef estimate_at_flow(self, current_flow: float, current_power: float,annual_hours: float 8000.0) - dict:估算指定风量下的节能潜力Args:current_flow: 当前风量 (m³/h)current_power: 当前功率 (kW)annual_hours: 年运行小时数Returns:节能估算结果optimal_power self.fitter.predict_p(current_flow)power_saving current_power - optimal_powerannual_saving_kwh power_saving * annual_hoursannual_saving_cost annual_saving_kwh * self.pricereturn {current_power: round(current_power, 2),optimal_power: round(optimal_power, 2),power_saving: round(power_saving, 2),annual_saving_kwh: round(annual_saving_kwh, 0),annual_saving_cost: round(annual_saving_cost, 0),saving_ratio: round(power_saving / current_power * 100, 1) if current_power 0 else 0}4.8 分析报告生成器class ReportGenerator:分析报告生成器def generate_report(self, summary: dict, savings: dict None) - str:生成文本报告lines [ * 60, 风机性能曲线测绘与节能分析报告, * 60,, 【最高效率点 (BEP)】,f 风量: {summary[bep_flow]} m³/h,f 效率: {summary[bep_efficiency]*100:.1f}%,, 【高效运行区间】,f 风量范围: {summary[optimal_zone_start]} ~ {summary[optimal_zone_end]} m³/h,f 区间宽度: {summary[optimal_zone_width]} m³/h,]if savings:lines.extend([ 【节能潜力估算】,f 当前功率: {savings[current_power]} kW,f 最优功率: {savings[optimal_power]} kW,f 单点节电: {savings[power_saving]} kW,f 年节电量: {savings[annual_saving_kwh]} kWh,f 年节约费用: ¥{savings[annual_saving_cost]},f 节能比例: {savings[saving_ratio]}%,])lines.append( * 60)return \n.join(lines)4.9 系统编排器class FanPerformanceSystem:风机性能分析系统 —— 聚合根def __init__(self, fan_config: FanConfig None,curve_config: CurveConfig None):self.fan_config fan_config or FanConfig()self.curve_config curve_config or CurveConfig()self.loader DataLoader()self.calculator PowerEfficiencyCalculator(self.fan_config)self.fitter CurveFitter(self.curve_config)self.zone_finder Noneself.estimator EnergySavingsEstimator(self.fitter)self.reporter ReportGenerator()def analyze(self, data_path: str) - str:执行完整的分析流程Args:data_path: 数据文件路径Returns:分析报告# ① 加载数据self.loader.file_path Path(data_path)samples self.loader.load_csv()samples self.loader.clean()if len(samples) self.curve_config.min_samples:return 错误: 有效样本数不足无法进行曲线拟合。# ② 计算功率和效率results self.calculator.batch_calculate(samples)flows np.array([r[sample].flow for r in results])pressures np.array([r[sample].pressure for r in results])powers np.array([r[p_elec] for r in results])efficiencies np.array([r[efficiency] for r in results])# ③ 曲线拟合self.fitter.fit_hq_curve(flows, pressures)self.fitter.fit_pq_curve(flows, powers)self.fitter.fit_eq_curve(flows, efficiencies)# ④ 寻优flow_range (np.min(flows), np.max(flows))self.zone_finder OptimalZoneFinder(self.fitter, flow_range)self.zone_finder.find_bep()self.zone_finder.find_optimal_zone()# ⑤ 节能估算 (以当前平均工况为例)avg_flow np.mean(flows)avg_power np.mean(powers)savings self.estimator.estimate_at_flow(avg_flow, avg_power)# ⑥ 生成报告summary self.zone_finder.get_summary()report self.reporter.generate_report(summary, savings)return report4.10 完整演示def demo():完整演示print( * 60)print( 风机性能曲线测绘与节能运行区间寻优系统 v1.0)print( 基于哈尔滨工程大学《工业过程控制》课程理论)print( * 60)# 生成模拟数据import tempfileimport osnp.random.seed(42)csv_lines [timestamp,flow,pressure,current,voltage,power_factor,speed]# 模拟风机在不同工况下的运行数据for i in range(100):flow 30000 i * 500 # 30000~80000 m³/h# 模拟 H-Q 曲线: H 4000 - 0.00003 * Q^2pressure 4000 - 0.00003 * flow**2 np.random.normal(0, 50)# 模拟电流: 随流量增加而增加current 100 flow * 0.003 np.random.normal(0, 5)csv_lines.append(f{100i*10},{flow:.0f},{pressure:.1f},{current:.1f},380,0.85,1480)tmpdir tempfile.mkdtemp()csv_path os.path.join(tmpdir, fan_data.csv)with open(csv_path, w) as f:f.write(\n.join(csv_lines))# 运行分析system FanPerformanceSystem(fan_configFanConfig(motor_power250.0),curve_configCurveConfig())report system.analyze(csv_path)print(report)# 清理os.r利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛