电力工程师必看：手把手教你用Python解析COMTRADE文件（含CFG/DAT文件实战）
电力工程师必看：手把手教你用Python解析COMTRADE文件（含CFG/DAT文件实战）
在电力系统故障分析和保护测试领域，COMTRADE格式已成为行业标准的数据交换格式。作为电力工程师，我们经常需要从继电保护装置或故障录波器中获取这类数据文件，但如何高效提取其中的电压电流波形数据，却是许多同行面临的实操难题。本文将用Python代码演示如何从原始CFG/DAT文件中还原真实的波形值，解决二进制/ASCII格式差异、变换因子计算等典型问题。
1. COMTRADE文件结构解析实战
COMTRADE文件组通常包含四个文件：.CFG（配置文件）、.DAT（数据文件）、.HDR（头文件）和.INF（信息文件）。其中前两个是必须的：
典型的COMTRADE文件组示例 NORMAL_OPERATION.CFG # 配置文件（ASCII格式） NORMAL_OPERATION.DAT # 数据文件（ASCII或二进制） NORMAL_OPERATION.HDR # 可选的标题文件 NORMAL_OPERATION.INF # 可选的信息文件配置文件(.CFG)的关键字段解析：
- 第一行：厂站名称、设备ID、标准版本（如"1999"表示遵循IEEE Std C37.111-1999）
- 通道定义：包含变换因子(fCoefA/B)、量程范围(fMin/fMax)等关键参数
- 采样率：可能包含多个采样率阶段（如故障前高频采样+故障后低频采样）
2. Python解析CFG配置文件的完整方案
2.1 基础解析框架搭建
首先创建配置文件的解析类，处理多版本兼容问题：
class ComtradeConfig: def __init__(self, cfg_path): self.station_name = "" self.rec_dev_id = "" self.rev_year = 1991 # 默认使用1991版标准 self.analog_channels = [] self.digital_channels = [] self._parse(cfg_path) def _parse(self, cfg_path): with open(cfg_path, 'r') as f: lines = [line.strip() for line in f if line.strip()] # 解析第一行（版本信息） if len(lines[0].split(',')) >= 3: self.station_name, self.rec_dev_id, year_str = lines[0].split(',')[:3] self.rev_year = int(year_str) if year_str.strip() else 1991 # 解析通道数量（第二行） total, analog, digital = self._parse_channel_count(lines[1]) # 解析模拟通道配置 for i in range(2, 2 + analog): self.analog_channels.append(self._parse_analog_channel(lines[i])) # 解析数字通道配置（略） ...2.2 关键参数提取技巧
特别注意变换因子的处理——这直接关系到数据还原的准确性：
def _parse_analog_channel(self, line): parts = [p.strip() for p in line.split(',')] return { 'id': int(parts[0]), 'name': parts[1], 'phase': parts[2], 'unit': parts[4], 'a': float(parts[5]), # 变换因子A 'b': float(parts[6]), # 变换因子B 'min': float(parts[9]), 'max': float(parts[10]), 'primary': float(parts[11]), 'secondary': float(parts[12]), 'ps': parts[13].upper() # P/S标识 }注意：实际项目中经常遇到CFG中声明的fMin/fMax与DAT文件中实际值不符的情况，建议在解析时进行动态统计。
3. DAT数据文件的高效处理
3.1 ASCII格式解析
对于ASCII格式的.DAT文件，每行对应一个采样点：
def parse_ascii_dat(dat_path, config): data = [] with open(dat_path, 'r') as f: for line in f: if not line.strip(): continue parts = [p.strip() for p in line.split(',')] sample = { 'seq': int(parts[0]), 'timestamp': int(parts[1]), 'analog': [], 'digital': [] } # 解析模拟量（应用变换因子） for i, ch in enumerate(config.analog_channels): raw = float(parts[2+i]) value = ch['a'] * raw + ch['b'] sample['analog'].append(value) # 解析数字量（略） ... data.append(sample) return data3.2 二进制格式解析
二进制格式更紧凑但需要特殊处理：
import struct def parse_binary_dat(dat_path, config): data = [] with open(dat_path, 'rb') as f: while True: # 读取序号(4字节)和时间戳(4字节) header = f.read(8) if not header: break seq, timestamp = struct.unpack('2i', header) sample = {'seq': seq, 'timestamp': timestamp, 'analog': []} # 读取模拟量（每个2字节） for _ in range(len(config.analog_channels)): raw = struct.unpack('h', f.read(2))[0] # 应用变换公式... sample['analog'].append(...) # 读取数字量（每16位一组） ... data.append(sample) return data关键点：二进制格式中模拟量用16位有符号整数表示，数字量每16通道打包为2字节
4. 实战中的典型问题解决方案
4.1 多采样率数据处理
当CFG中定义多个采样率时（如故障前5000Hz，故障后10Hz），需要分段处理：
def process_multi_sample_rates(data, config): rate_sections = [] current_pos = 0 for rate in config.sample_rates: section = { 'rate': rate['samp'], 'data': data[current_pos : current_pos + rate['endsamp']] } rate_sections.append(section) current_pos += rate['endsamp'] return rate_sections4.2 数据可视化示例
使用Matplotlib绘制解析后的波形：
import matplotlib.pyplot as plt def plot_waveform(data, channels, title="COMTRADE波形"): plt.figure(figsize=(12, 6)) time = [d['timestamp']/1e6 for d in data] # 转换为秒 for i, ch in enumerate(channels): values = [d['analog'][i] for d in data] plt.plot(time, values, label=f"{ch['name']}({ch['unit']})") plt.xlabel('时间(s)') plt.ylabel('幅值') plt.title(title) plt.legend() plt.grid() plt.show()4.3 性能优化技巧
处理大型COMTRADE文件时的优化方案：
| 优化方法 | 实施手段 | 效果提升 |
|---|---|---|
| 内存映射 | 使用numpy.memmap | 减少内存占用 |
| 并行处理 | 多进程分块解析 | 加快处理速度 |
| 缓存机制 | 预处理后保存为HDF5 | 后续快速加载 |
# 使用numpy加速变换计算 import numpy as np def fast_convert(analog_data, a, b): """使用numpy向量化运算加速变换""" return np.array(analog_data) * a + b5. 完整工程实现建议
对于需要集成到生产环境的解析方案，建议采用以下架构：
核心解析层：封装为独立Python包
- 支持多版本COMTRADE标准
- 自动检测文件格式（ASCII/二进制）
数据预处理层：
class ComtradeProcessor: def __init__(self, cfg_path, dat_path): self.config = ComtradeConfig(cfg_path) self.raw_data = self._load_dat(dat_path) self._validate() def get_primary_values(self): """返回一次值（考虑PT/CT变比）""" ...扩展功能模块：
- 自动生成测试报告（PDF/Excel）
- 与SCADA系统集成接口
- 长期数据存储方案（数据库集成）
实际项目中遇到过CFG文件编码不规范的情况，特别是老式录波器生成的文件可能使用本地编码（如GB2312），建议在解析时增加编码检测：
import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: raw = f.read(1024) return chardet.detect(raw)['encoding']