WTOAAM/dataAnalysisBusiness/algorithm/faultAnalyst.py

import pandas as pd
import os
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystNotFilter import AnalystNotFilter

class FaultAnalyst(AnalystNotFilter):
    """
    风电机组故障分析
    新逻辑：基于首发故障结束时间划分窗口， 以首发故障的结束时间为界，后续故障的开始时间如果 ≤ 这个结束时间，则视为连带故障；否则作为新的首发故障。

    风场范围的故障分析
    统计风场范围内 首发故障中各故障发生的次数和累计时长
    """

    def typeAnalyst(self):
        return "fault"

    def selectColumns(self):
        # 直接返回数据库表中的真实列名
        return ["wind_turbine_name", "begin_time", "end_time", "fault_detail"]
        #风机原始名称、故障开始时间、故障结束时间、故障描述
    def getTimeGranularitys(self, conf: Contract):
        return ["fault"]

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf, self.selectColumns())
        dataFrameMerge = dictionary.get("fault", pd.DataFrame())
        if dataFrameMerge.empty:
            return pd.DataFrame()
        return self.get_result(dataFrameMerge, outputAnalysisDir, conf)

    def get_result(self, dataFrame: pd.DataFrame, outputAnalysisDir: str, conf: Contract):
        if dataFrame.empty:
            return pd.DataFrame()

        df = dataFrame.copy()
        
        # 转换时间列
        df["begin_time"] = pd.to_datetime(df["begin_time"])
        df["end_time"] = pd.to_datetime(df["end_time"])
        df = df.dropna(subset=["begin_time", "end_time"])

        if df.empty:
            return pd.DataFrame()

        # 存储中间结果
        turbine_events = []
        fault_events = []

        # 按风机分组处理
        for turbine, group in df.groupby("wind_turbine_name"):
            group = group.sort_values("begin_time").reset_index(drop=True)
            i = 0
            n = len(group)
            while i < n:
                primary = group.iloc[i]
                primary_start = primary["begin_time"]
                primary_end = primary["end_time"]
                duration_sec = (primary_end - primary_start).total_seconds()

                turbine_events.append({
                    "turbine": turbine,
                    "duration_sec": duration_sec
                })
                fault_events.append({
                    "fault_detail": primary["fault_detail"],
                    "duration_sec": duration_sec
                })

                # 跳过连带故障
                j = i + 1
                while j < n and group.iloc[j]["begin_time"] <= primary_end:
                    j += 1
                i = j

        # 聚合风机维度
        if turbine_events:
            turbine_df = pd.DataFrame(turbine_events)
            turbine_summary = turbine_df.groupby("turbine").agg(
                count=("duration_sec", "size"),
                fault_time=("duration_sec", "sum")
            ).reset_index()
            turbine_summary = turbine_summary.rename(columns={"turbine": "wind_turbine_name"})
            turbine_file = os.path.join(outputAnalysisDir, f"turbine_fault_result{CSVSuffix}")
            turbine_summary.to_csv(turbine_file, index=False, encoding='utf-8-sig')
        else:
            turbine_summary = pd.DataFrame()

        # 聚合故障类型维度
        if fault_events:
            fault_df = pd.DataFrame(fault_events)
            fault_summary = fault_df.groupby("fault_detail").agg(
                count=("duration_sec", "size"),
                fault_time_sum=("duration_sec", "sum")
            ).reset_index()
            fault_file = os.path.join(outputAnalysisDir, f"total_fault_result{CSVSuffix}")
            fault_summary.to_csv(fault_file, index=False, encoding='utf-8-sig')
        else:
            fault_summary = pd.DataFrame()

        # 返回结果
        result_rows = []
        if not turbine_summary.empty:
            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: "total",
                Field_MillTypeCode: "turbine_fault_result",
                Field_Return_FilePath: turbine_file,
                Field_Return_IsSaveDatabase: True
            })
        if not fault_summary.empty:
            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: "total",
                Field_MillTypeCode: "total_fault_result",
                Field_Return_FilePath: fault_file,
                Field_Return_IsSaveDatabase: True
            })
        return pd.DataFrame(result_rows)