中广核消缺：故障分析功能逻辑修改

dataAnalysisBusiness/algorithm/faultAnalyst.py

@@ -1,121 +1,124 @@
 import pandas as pd
 import os
-import plotly.graph_objects as go
 from algorithmContract.confBusiness import *
 from algorithmContract.contract import Contract
 from behavior.analystNotFilter import AnalystNotFilter
-from plotly.subplots import make_subplots
 
 class FaultAnalyst(AnalystNotFilter):
     """
     风电机组故障分析
+    新逻辑：基于首发故障结束时间划分窗口， 以首发故障的结束时间为界，后续故障的开始时间如果 ≤ 这个结束时间，则视为连带故障；否则作为新的首发故障。
+
+    风场范围的故障分析
+    统计风场范围内 首发故障中各故障发生的次数和累计时长
     """
 
     def typeAnalyst(self):
         return "fault"
-    
-    def selectColumns(self):
-        # 这里的字段必须与数据库 _fault 表中的列名对应
-        return [Field_DeviceName, Field_FaultTime, Field_FaultDetail]
 
-    # 强制重写获取数据源类型的方法，防止因配置错误去查 _minute 表，强制代码去查 _fault 表
+    def selectColumns(self):
+        # 直接返回数据库表中的真实列名
+        return ["wind_turbine_name", "begin_time", "end_time", "fault_detail"]
+        #风机原始名称、故障开始时间、故障结束时间、故障描述
     def getTimeGranularitys(self, conf: Contract):
-        return ["fault"] 
+        return ["fault"]
 
     def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
-        # 1. 抓取数据
         dictionary = self.processTurbineData(turbineCodes, conf, self.selectColumns())
-        
-        # 2. 直接获取 "fault" key 的数据, 绕过父类的 userDataFrame 方法，因为它会读取错误的配置文件 ('minute')
         dataFrameMerge = dictionary.get("fault", pd.DataFrame())
-        
-        # 3. 增加空数据保护
         if dataFrameMerge.empty:
-            # print("Warning: No fault data found for the selected turbines.")
             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()
 
-        #---------------整个风场维度统计故障时长与次数---------------------------
-        # 统计各种类型故障出现的次数
-        if Field_FaultDetail in dataFrame.columns:
-            fault_detail_count = dataFrame[Field_FaultDetail].value_counts().reset_index()
-            fault_detail_count.columns = [Field_FaultDetail, 'count']
+        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"])
 
-            # 统计每个 fault_detail 的时长加和
-            fault_time_sum = dataFrame.groupby(Field_FaultDetail)[Field_FaultTime].sum().reset_index()
-            fault_time_sum.columns = [Field_FaultDetail, 'fault_time_sum']
+        if df.empty:
+            return pd.DataFrame()
 
-            # 合并两个 DataFrame
-            fault_summary = pd.merge(fault_detail_count, fault_time_sum, on=Field_FaultDetail, how='inner')
-            fault_summary_sorted = fault_summary.sort_values(by='fault_time_sum', ascending=False)
+        # 存储中间结果
+        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:
-            # 防御性代码：如果缺列
-            fault_summary_sorted = pd.DataFrame()
-
-        # -------------按风机分组统计故障情况------------------------------------------
-        # 确保有设备名称列
-        if Field_DeviceName not in dataFrame.columns:
-             # 有时 Field_DeviceName 没取到，尝试用 DeviceCode 或其他
-             groupby_col = dataFrame.columns[0] 
+            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:
-             groupby_col = Field_DeviceName
-
-        grouped = dataFrame.groupby(groupby_col)
-        results= []
+            fault_summary = pd.DataFrame()
 
-        for name, group in grouped:
-            turbine_fault_summary = pd.DataFrame({
-                Field_DeviceName: [name],
-                'count': [len(group)],
-                'fault_time': [group[Field_FaultTime].sum()]
-            })
-            results.append(turbine_fault_summary)
-
-        # 合并所有风机的故障统计结果
-        if results:
-            turbine_fault_summary = pd.concat(results, ignore_index=True)
-            turbine_fault_sorted = turbine_fault_summary.sort_values(by='fault_time', ascending=False)
-            # 故障类型前十名
-            # draw_results=turbine_fault_sorted.head(10) # 暂时没用到
-        else:
-            turbine_fault_sorted = pd.DataFrame()
-
-        # 保存结果
+        # 返回结果
         result_rows = []
-
-        if not turbine_fault_sorted.empty:
-            filePathOfturbinefault = os.path.join(outputAnalysisDir, f"turbine_fault_result{CSVSuffix}")
-            turbine_fault_sorted.to_csv(filePathOfturbinefault, index=False,encoding='utf-8-sig')
-
+        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: filePathOfturbinefault,
+                Field_MillTypeCode: "turbine_fault_result",
+                Field_Return_FilePath: turbine_file,
                 Field_Return_IsSaveDatabase: True
             })
-
-        if not fault_summary_sorted.empty:
-            filePathOftotalfault = os.path.join(outputAnalysisDir, f"total_fault_result{CSVSuffix}")
-            fault_summary_sorted.to_csv(filePathOftotalfault, index=False,encoding='utf-8-sig')
-
+        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: filePathOftotalfault,
+                Field_MillTypeCode: "total_fault_result",
+                Field_Return_FilePath: fault_file,
                 Field_Return_IsSaveDatabase: True
             })
-            
-        result_df = pd.DataFrame(result_rows)
-        return result_df
+        return pd.DataFrame(result_rows)