如何用Python/Pandas自动分析计算大量CSV日志文件

针对你需要批量处理10000个机器日志文件，计算特定条件下压力积分并汇总结果的需求，这里提供一个基于Python+Pandas的可直接运行的脚本方案，逻辑简单易修改：

核心思路 #

• 精准匹配目标日志文件（按递增数字_日期_时间.csv格式）
• 逐个读取文件，仅处理所需的压力和时间数据，不合并源文件
• 按自定义条件筛选数据后，用数值积分法计算压力对时间的积分
• 汇总每个文件的结果到单独的CSV/Excel文件

完整代码示例 #

import os
import pandas as pd
from scipy.integrate import trapz

# -------------------------- 配置参数（根据你的实际情况修改） --------------------------
LOG_DIR = "./machine_logs"  # 日志文件所在目录
RESULT_FILE = "pressure_integral_results.csv"  # 结果保存路径
TIME_COL = "时间"  # 你的日志中时间列的名称
PRESSURE_COL = "压力"  # 你的日志中压力列的名称
# 自定义筛选条件：示例为温度>30℃且泵处于开启状态，替换成你的实际条件
FILTER_CONDITION = lambda df: (df["温度"] > 30) & (df["泵状态"] == "开启")

# -------------------------- 核心处理逻辑 --------------------------
def calculate_pressure_integral(file_path):
    try:
        # 读取文件，仅加载需要的列以节省内存
        df = pd.read_csv(file_path, usecols=[TIME_COL, PRESSURE_COL, "温度", "泵状态"])
        
        # 转换时间列为datetime格式，计算时间差（转成总秒数作为积分的x轴）
        df[TIME_COL] = pd.to_datetime(df[TIME_COL])
        df["时间差(秒)"] = (df[TIME_COL] - df[TIME_COL].iloc[0]).dt.total_seconds()
        
        # 应用筛选条件，剔除无效数据
        filtered_df = df[FILTER_CONDITION(df)].dropna(subset=[PRESSURE_COL, "时间差(秒)"])
        
        if len(filtered_df) < 2:
            # 数据点不足无法计算积分，返回0或标记为无效
            return 0.0
        
        # 用梯形法计算积分：压力对时间的积分
        integral_value = trapz(filtered_df[PRESSURE_COL], x=filtered_df["时间差(秒)"])
        return round(integral_value, 4)  # 保留4位小数
    except Exception as e:
        print(f"处理文件 {file_path} 出错: {str(e)}")
        return None

if __name__ == "__main__":
    # 匹配所有符合命名格式的日志文件
    log_files = [f for f in os.listdir(LOG_DIR) if f.endswith(".csv") and len(f.split("_")) >=3]
    # 也可以用glob更精准匹配：from glob import glob; log_files = glob(os.path.join(LOG_DIR, "*_*_*.csv"))
    
    results = []
    for filename in log_files:
        file_path = os.path.join(LOG_DIR, filename)
        integral = calculate_pressure_integral(file_path)
        if integral is not None:
            results.append({"文件名": filename, "压力积分值": integral})
    
    # 将结果写入CSV文件（如需Excel，替换为pd.DataFrame(results).to_excel("results.xlsx", index=False)，需先装openpyxl库）
    pd.DataFrame(results).to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")
    print(f"处理完成，结果已保存到 {RESULT_FILE}")

关键部分说明 #

1. 配置参数修改：

   • 把LOG_DIR改成你的日志文件实际存放路径
   • 替换TIME_COL、PRESSURE_COL为你日志文件中对应的列名
   • 修改FILTER_CONDITION为你的实际筛选规则，比如基于温度、通风口状态等

2. 内存优化：

   • 用usecols参数只加载需要的列，避免读取整个大文件，提升处理速度
   • 单个文件处理完后立即释放内存，不会累积大量数据

3. 积分计算：

   • 使用梯形法（trapz）是工程上常用的数值积分方法，适合处理离散的时间序列数据
   • 时间差转成秒数，确保积分结果的单位是「压力单位·秒」，符合物理意义

4. 异常处理：

   • 单个文件处理出错时会打印错误信息并跳过，不会导致整个批处理中断
   • 数据点不足时返回0，你可以根据需求改成标记为「无效」

5. 结果保存：

   • 默认保存为CSV，如需Excel，只需把to_csv改成to_excel，并提前安装openpyxl库（执行pip install openpyxl）

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