Python统计CSV某列达阈值前行数 计算设备故障周期

统计规则 #

• 故障判定标准：监测值≥5时判定设备触发故障
• 数据读取范围：读取CSV文件第一列的监测数值
• 计数逻辑：自动跳过开头所有≥5的初始数据，从首次出现小于5的监测值的位置开始逐行计数，直到碰到第一个≥5的监测值时停止，累计的行数就是故障前运行周期数。

示例计算说明：
给定示例数据前4行第一列值为68、25、15、5，均≥5，直接跳过；从第5行值2（<5）开始计数，到第11行值5.01（≥5）停止，覆盖第5到第11行共7行，统计结果为7个周期。

实现代码 #

代码用Python内置csv模块实现，无需安装第三方依赖，自动兼容空行、格式异常行等常见脏数据场景：

import csv

def count_pre_fault_cycles(csv_path: str, threshold: float = 5) -> int | str:
    """
    统计设备故障前的有效运行周期
    :param csv_path: 待读取的CSV文件路径
    :param threshold: 故障判定阈值，默认值为5
    :return: 故障前运行周期数，边界场景返回对应提示文本
    """
    cycle_count = 0
    start_flag = False  # 标记是否已找到首个小于阈值的起始计数点

    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        for row in reader:
            # 跳过空行
            if not row:
                continue
            # 解析第一列数值，跳过格式错误的行
            try:
                current_val = float(row[0].strip())
            except (IndexError, ValueError):
                continue
            
            # 未到起始计数点的逻辑
            if not start_flag:
                if current_val < threshold:
                    start_flag = True
                    cycle_count += 1
                continue
            
            # 已开始计数的逻辑
            cycle_count += 1
            if current_val >= threshold:
                return cycle_count
    
    # 遍历完文件的边界场景处理
    if start_flag:
        return f"遍历至文件末尾未触发故障，累计有效运行周期{cycle_count}个"
    else:
        return "数据中无小于阈值的有效起始点，无法完成统计"

# 调用示例
if __name__ == "__main__":
    res = count_pre_fault_cycles("device_monitor.csv")
    print(f"故障前运行周期统计结果：{res}")

边界场景适配 #

• 初始数据全部高于/等于阈值：自动向后遍历直到找到第一个小于阈值的点才开始计数，符合需求兼容要求
• 文件无有效数值、全是高于阈值的数据：返回明确提示，不会抛出无意义的错误
• 从起始点到文件末尾都没有触发故障：返回累计的有效运行周期数
• 自动跳过空行、第一列非数值的异常行，适配日常导出的不规范CSV文件

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