电力工程师必看手把手教你用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[2i]) 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, titleCOMTRADE波形): 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, labelf{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]
电力工程师必看：手把手教你用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[2i]) 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, titleCOMTRADE波形): 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, labelf{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]
