Python实现memory_前缀txt结构化数据写入预定义Pandas DataFrame

实现方案说明 #

• 规避性能问题：循环调用pd.DataFrame.append()是官方已弃用的低效率写法，我们将所有文件的解析结果先存为字典列表，最后一次性合并为DataFrame，执行效率会有非常明显的提升
• 处理重复列名：你定义的列名中存在多个重复的mode字段，解析时会自动给重复列加下标区分，避免匹配错误
• 键值对兼容逻辑：逐行读取txt文件时按:最多分割1次，避免值内部含冒号时解析出错，自动过滤空行、无冒号的无效行，未匹配到的列值默认留空

完整实现代码 #

import os
import pandas as pd
from collections import defaultdict

# 设置os.walk遍历的根目录
rootdir = 'K:/Retouren' 

# 预定义列名
column_names = ["Memory dump", "Serialnr", "Date/time", "mode", "Hardware release", "Software release", "Rom test 1 checksum", "Rom test 2 checksum", 
    "mode", "Line power connected (hours)", "Line power disconnected (number of times)", "Ch function(hours)", "Dhw function(hours)", "Burnerstarts (number of times)", 
    "Ignition failed (number of times)", "Flame lost (number of times)", "Reset (number of times)", "Gasmeter_ch", "Gasmeter_dhw", "Watermeter", "Burnerstarts_dhw", 
    "mode", "T1", "T2", "T3", "T4", "T5", "T6", "Temp_set", "Fanspeed_set", "Fanspeed", "Fan_pwm", "Opentherm", "Roomtherm", "Tap_switch", "Gp_switch", "Pump", "Dwk", 
    "Gasvalve", "Io_signal", "Spark", "Io_curr", "Displ_code", "Ch_pressure", "Rf_rth_bound", "Rf_rth_communication", "Rf_rth_battery_info", "Rf_rth_battery_ok", 
    "Bc_tapflow", "Pump_pwm", "Room_override_zone1", "Room_set_zone1", "Room_temp_zone1", "Room_override_zone2", "Room_set_zone2", "Room_temp_zone2", "Outside_temp", 
    "Ot_master_member_id", "Ot_therm_prod_version", "Ot_therm_prod_type", "mode", "Node nr", "Cloud id0", "Cloud id1", "Cloud id2", "Rf cloud nr", "Rssi_lower", 
    "Rssi_upper", "Rssi_wait", "Attention_period", "Attention_number", "Info10", "Info11", "Info12", "Info13", "Info14", "Info15", "Info16", "Info17", "Info18", 
    "Ramses_thermostat_idh", "Ramses_thermostat_idm", "Ramses_thermostat_idl", "Ramses_boiler_idh", "Ramses_boiler_idm", "Ramses_boiler_idl", "Prod. token", 
    "Year", "Month", "Line number", "Serial1", "Serial2", "Serial3", "mode", "Id_dongle0", "Id_dongle1", "Id_dongle2", "Id_dongle3", "Id_lan0", "Id_lan1", 
    "Id_lan2", "Id_lan3", "Info2_7", "Info2_8", "Info2_9", "Info2_10", "Info2_11", "Info2_12", "Info2_13", "Info2_14", "mode", "Interrupt_time", 
    "Interrupt_load (%)", "Main_load (%)", "Net fequency (hz)", "Voltage ref. (v)", "Checksum1", "Checksum2", "nmode", "Fault 0", "Fault 1", "Fault 2", 
    "Fault 3", "Fault 4", "Fault 5", "Fault 6", "Fault 7", "Fault 8", "Fault 9", "Fault 10", "Fault 11", "Fault 12", "Fault 13", "Fault 14", "Fault 15", 
    "Fault 16", "Fault 17", "Fault 18", "Fault 19", "Fault 20", "Fault 21", "Fault 22", "Fault 23", "Fault 24", "Fault 25", "Fault 26", "Fault 27", "Fault 28", 
    "Fault 29", "Fault 30", "Fault 31", "mode", "Heater_on", "Comfort_mode", "Ch_set_max", "Dhw_set", "Eco_days", "Comfort_set", "Dhw_at_night", "Ch_at_night", 
    "Parameter 1", "Parameter 2", "Parameter 3", "Parameter 4", "Parameter 5", "Parameter 6", "Parameter 7", "Parameter 8", "Parameter 9", "Parameter a", 
    "Parameter b", "Parameter c", "Parameter c", "Parameter d", "Parameter e", "Parameter e.", "Parameter f", "Parameter h", "Parameter n", "Parameter o", 
    "Parameter p", "Parameter r", "Parameter f.", "mode", "Param31", "Param32", "Param33", "Param34", "Param35", "Param36", "Param37", "Param38", "Param39", 
    "Param40", "Param41", "Param42", "Param43", "Param44", "Param45", "Param46", "Param47", "Param48", "Param49", "Param50", "Param51", "Param52", "Param53", 
    "Param54", "Param55", "Param56", "Param57", "Param58", "Param59", "Param60", "Param61", "Param62", "Param63"]

# 给重复列名加下标避免冲突
count = defaultdict(int)
unique_columns = []
for col in column_names:
    if count[col] > 0:
        unique_columns.append(f"{col}_{count[col]}")
    else:
        unique_columns.append(col)
    count[col] += 1

# 存储所有行数据的列表
rows_list = []

for subdir, dirs, files in os.walk(rootdir):
    for file in files:
        if file.startswith('memory_') and os.path.splitext(file)[1] == '.txt':
            filepath = os.path.join(subdir, file)
            # 初始化当前行字典，默认值为空
            row_dict = {col: None for col in unique_columns}
            # 将文件路径存入Memory dump列，可根据需求修改为文件名
            row_dict["Memory dump"] = filepath
            # mode计数器，匹配多个重复的mode列
            mode_counter = 0
            
            with open(filepath, "r", encoding="utf-8", errors="ignore") as curfile:
                for line in curfile:
                    line = line.strip()
                    if not line or ":" not in line:
                        continue
                    # 最多分割一次，避免值内包含冒号出错
                    key, value = line.split(":", 1)
                    key = key.strip()
                    value = value.strip()
                    
                    # 处理mode字段匹配
                    if key == "mode":
                        if mode_counter == 0:
                            row_dict["mode"] = value
                        else:
                            row_dict[f"mode_{mode_counter}"] = value
                        mode_counter += 1
                    else:
                        # 其他字段直接匹配，不存在的键自动忽略
                        if key in row_dict:
                            row_dict[key] = value
            # 加入行列表
            rows_list.append(row_dict)

# 一次性生成DataFrame
data = pd.DataFrame(rows_list, columns=unique_columns)
# 若需要恢复原始无下标的列名，取消注释下一行即可
# data.columns = column_names

注意事项 #

• 代码中添加了encoding="utf-8", errors="ignore"参数避免不同编码的文件读取报错，如果你的文件为GBK等其他编码，可自行修改encoding参数
• 若需要将数值类字段自动转为数字类型，可在生成DataFrame后添加data = data.apply(pd.to_numeric, errors='ignore')处理

内容的提问来源于stack exchange，提问作者Martijn