WTOAAM/dataAnalysisBusiness/algorithm/powerCurveAnalyst.py

import os

import numpy as np
import pandas as pd
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodPoint import AnalystWithGoodPoint
from utils.jsonUtil import JsonUtil


class PowerCurveAnalyst(AnalystWithGoodPoint):
    """
    风电机组功率曲线散点分析。
    秒级scada数据运算太慢，建议使用分钟级scada数据
    """

    def typeAnalyst(self):
        return "power_curve"

    def turbinesAnalysis(self,  outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf, [
                                             Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower])
        dataFrameOfTurbines = self.userDataFrame(
            dictionary, conf.dataContract.configAnalysis, self)
        # 检查所需列是否存在
        required_columns = {Field_WindSpeed, Field_ActiverPower}
        if not required_columns.issubset(dataFrameOfTurbines.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")

        turbrineInfos = self.common.getTurbineInfos(
            conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)

        groupedOfTurbineModel = turbrineInfos.groupby(Field_MillTypeCode)

        returnDatas = []
        for turbineModelCode, group in groupedOfTurbineModel:
            currTurbineCodes = group[Field_CodeOfTurbine].unique().tolist()
            currTurbineModeInfo = self.common.getTurbineModelByCode(
                turbineModelCode, self.turbineModelInfo)

            dataFrameOfContractPowerCurve = self.dataFrameContractOfTurbine[
                self.dataFrameContractOfTurbine[Field_MillTypeCode] == turbineModelCode]
            currDataFrameOfTurbines = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine].isin(
                currTurbineCodes)]

            # 【新增筛选逻辑】 只在画图计算前清洗，剔除高风速下的低功率（停机）数据
            
            # +1. 获取额定风速。
            # 如果 confBusiness 中定义了 Field_RatedWindSpeed 且机型信息里有该字段，则使用；否则给一个默认值（例如 11m/s 或 12m/s），防止报错。
            rated_ws = 11 # 默认值，防报错
            if 'Field_RatedWindSpeed' in globals() and Field_RatedWindSpeed in currTurbineModeInfo:
                 rated_ws = currTurbineModeInfo[Field_RatedWindSpeed]

            # +2. 执行过滤：保留 (风速 <= 额定风速) 或者 (风速 > 额定风速 且 功率 >= 20) 的数据
            # 先做一个 .copy() 防止 SettingWithCopyWarning
            currDataFrameOfTurbines = currDataFrameOfTurbines.copy()
            
            mask_bad_data = (currDataFrameOfTurbines[Field_WindSpeed] > rated_ws) & (currDataFrameOfTurbines[Field_ActiverPower] < 20)
            currDataFrameOfTurbines = currDataFrameOfTurbines[~mask_bad_data]
            
            
            powerCurveDataOfTurbines = self.dataReprocess(
                currDataFrameOfTurbines, self.binsWindSpeed)

            # returnData = self.drawOfPowerCurve(
            #     powerCurveDataOfTurbines, outputAnalysisDir, conf, dataFrameOfContractPowerCurve, currTurbineModeInfo)
            # returnDatas.append(returnData)

            returnJsonData= self.outputPowerCurveData(conf,outputAnalysisDir,currTurbineModeInfo,powerCurveDataOfTurbines,dataFrameOfContractPowerCurve)
            returnDatas.append(returnJsonData)

        returnResult = pd.concat(returnDatas, ignore_index=True)

        return returnResult

    def outputPowerCurveData(self, conf: Contract, outputAnalysisDir: str, turbineModelInfo: pd.Series, powerCurveDataOfTurbines: pd.DataFrame, dataFrameOfContractPowerCurve: pd.DataFrame) -> pd.DataFrame:
        turbineCodes = powerCurveDataOfTurbines[Field_CodeOfTurbine].unique()
        jsonDictionary = self.convert2Json(turbineModelInfo,turbineCodes=turbineCodes,
                                       dataFrameOfTurbines=powerCurveDataOfTurbines, dataFrameOfContract=dataFrameOfContractPowerCurve)
        jsonFileName = f"power_curve-{turbineModelInfo[Field_MillTypeCode]}.json"
        jsonFilePath = os.path.join(outputAnalysisDir, jsonFileName)

        JsonUtil.write_json(jsonDictionary, file_path=jsonFilePath)
        result_rows = []
        result_rows.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: Const_Output_Total,
            Field_MillTypeCode:turbineModelInfo[Field_MillTypeCode],
            Field_Return_FilePath: jsonFilePath,
            Field_Return_IsSaveDatabase: True
        })

        for turbineCode in turbineCodes:
            data:pd.DataFrame=powerCurveDataOfTurbines[powerCurveDataOfTurbines[Field_CodeOfTurbine]==turbineCode]
            jsonFileName2 = f"power_curve-{data[Field_NameOfTurbine].iloc[0]}.json"
            jsonFilePath2 = os.path.join(outputAnalysisDir, jsonFileName2)
            JsonUtil.write_json(jsonDictionary, file_path=jsonFilePath2)
            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: turbineCode,
                Field_Return_FilePath: jsonFilePath2,
                Field_Return_IsSaveDatabase: True
            })

        returnDatas = pd.DataFrame(result_rows)
        return returnDatas

    def convert2Json(self, turbineModelInfo: pd.Series,turbineCodes, dataFrameOfTurbines: pd.DataFrame, dataFrameOfContract: pd.DataFrame):
        result = {
            "analysisTypeCode":"功率曲线分析",
            "engineTypeCode":  turbineModelInfo[Field_MillTypeCode] ,
            "engineTypeName": turbineModelInfo[Field_MachineTypeCode] ,
            "data": []
        }

        # 定义要替换的空值类型
        na_values = {pd.NA, float('nan')}

        # 从对象A提取数据
        for turbineCode in turbineCodes:
            data:pd.DataFrame=dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine]==turbineCode]
            engine_data = {
                "enginName": data[Field_NameOfTurbine].iloc[0],
                "enginCode": turbineCode,
                "xData": data[Field_WindSpeed].replace(na_values, None).tolist(),
                "yData": data[Field_ActiverPower].replace(na_values, None).tolist(),
                "zData": []
            }
            result["data"].append(engine_data)

        # 从对象B提取数据
        contract_curve = {
            "enginName": "合同功率曲线",
            "xData": dataFrameOfContract[Field_WindSpeed].replace(na_values, None).tolist(),
            "yData": dataFrameOfContract[Field_ActiverPower].replace(na_values, None).tolist(),
            "zData": []
        }
        result["data"].append(contract_curve)

        return result

    # def buildPowerCurveData(self, group: pd.DataFrame, fieldWindSpeed: str, fieldActivePower: str, bins) -> pd.DataFrame:
    #     """
    #     计算设备的功率曲线。
    #     """
    #     powerCut = group.groupby(pd.cut(group[fieldWindSpeed], bins, labels=np.arange(0, 25.5, 0.5))).agg({
    #         fieldActivePower: 'median',
    #         fieldWindSpeed: ['median', 'count']
    #     })
    #     wind_count = powerCut[fieldWindSpeed]['count'].tolist()
    #     line = powerCut[fieldActivePower]['median'].round(decimals=2).tolist()
    #     act_line = pd.DataFrame([powerCut.index, wind_count, line]).T
    #     act_line.columns = [Field_WindSpeed,
    #                         'EffectiveQuantity', Field_ActiverPower]
    #     return act_line

    def buildPowerCurveData(self, group: pd.DataFrame, fieldWindSpeed: str, fieldActivePower: str, bins) -> pd.DataFrame:
        """
        计算设备的功率曲线。
        """
        # 1. 按照固定步长进行分箱统计
        # 注意：这里使用的是固定的 bins (0, 0.5, 1.0 ... 25.0)，即使某区间没数据，也会生成一行索引，只是值为 NaN
        powerCut = group.groupby(pd.cut(group[fieldWindSpeed], bins, labels=np.arange(0, 25.5, 0.5))).agg({
            fieldActivePower: 'median',
            fieldWindSpeed: ['median', 'count']
        })

        # 2. 提取数据
        wind_count = powerCut[fieldWindSpeed]['count'].tolist()
        
        # 获取原始的中位数序列（包含 NaN）
        power_series = powerCut[fieldActivePower]['median']

        # 3. 处理不连续（NaN）的情况
        
        # 步骤 A: 线性插值 (Interpolate)
        # 解决中间的断档。例如：[1000, NaN, 1200] -> [1000, 1100, 1200]
        # limit_direction='forward' 表示只向后插值，防止低风速段无数据时胡乱填充
        power_series = power_series.interpolate(method='linear', limit_direction='forward')

        # 步骤 B: 前向填充 (Forward Fill)
        # 解决高风速段的断档。
        # 场景：筛选后，20m/s 以上全是 NaN。
        # 逻辑：既然是高风速，且之前已经达到了额定功率，那么后面缺失的值应该维持在最后一次观测到的功率（即额定功率）。
        power_series = power_series.ffill()

        # 步骤 C: (可选) 0值填充
        # 如果低风速段（开头）是 NaN，通常是因为没风，补 0
        power_series = power_series.fillna(0)

        line = power_series.round(decimals=2).tolist()

        # 4. 组装结果
        act_line = pd.DataFrame([powerCut.index, wind_count, line]).T
        act_line.columns = [Field_WindSpeed,
                            'EffectiveQuantity', Field_ActiverPower]
        return act_line


    def dataReprocess(self, dataFrameMerge: pd.DataFrame, binsWindSpeed) -> pd.DataFrame:
        # 初始化结果DataFrame
        dataFrames = []

        # 按设备名分组数据
        grouped = dataFrameMerge.groupby(
            [Field_NameOfTurbine, Field_CodeOfTurbine])

        # 计算每个设备的功率曲线
        for name, group in grouped:
            dataFramePowerCurveTurbine = self.buildPowerCurveData(
                group, Field_WindSpeed, Field_ActiverPower, binsWindSpeed)
            dataFramePowerCurveTurbine[Field_NameOfTurbine] = name[0]
            dataFramePowerCurveTurbine[Field_CodeOfTurbine] = name[1]

            dataFrames.append(dataFramePowerCurveTurbine)

        # 绘制全场功率曲线图
        dataFrameReprocess: pd.DataFrame = pd.concat(
            dataFrames, ignore_index=True).reset_index(drop=True)

        return dataFrameReprocess

    def drawOfPowerCurve(self, powerCurveOfTurbines: pd.DataFrame, outputAnalysisDir, conf: Contract, dataFrameGuaranteePowerCurve: pd.DataFrame, turbineModelInfo: pd.Series):
        """
        生成功率曲线并保存为文件。

        参数:
        frames (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        outputAnalysisDir (str): 分析输出目录。
        confData (ConfBusiness): 配置
        """
        # 绘制全场功率曲线图
        # ress =self.dataReprocess(dataFrameMerge,self.binsWindSpeed) # all_res.reset_index(drop=True)

        df1 = self.plot_power_curve(
            powerCurveOfTurbines, outputAnalysisDir, dataFrameGuaranteePowerCurve, Field_NameOfTurbine, conf, turbineModelInfo)

        # 绘制每个设备的功率曲线图
        grouped = powerCurveOfTurbines.groupby(
            [Field_NameOfTurbine, Field_CodeOfTurbine])

        df2 = pd.DataFrame()  # 新建一个空表格，与返回的单图功率曲线合并
        for name, group in grouped:
            df_temp2 = self.plot_single_power_curve(
                powerCurveOfTurbines, group, dataFrameGuaranteePowerCurve, name, outputAnalysisDir, conf)
            df2 = pd.concat([df2, df_temp2], ignore_index=True)

        # 总图与单图的表格合并
        df = pd.concat([df1, df2], ignore_index=True)
        return df

    def plot_power_curve(self, ress, output_path, dataFrameGuaranteePowerCurve: pd.DataFrame, Field_NameOfTurbine, conf: Contract, turbineModelInfo: pd.Series):
        """
        绘制全场功率曲线图。
        """
        # colors = px.colors.sequential.Turbo

        fig = go.Figure()

        for turbine_num in ress[Field_NameOfTurbine].unique():
            turbine_data = ress[ress[Field_NameOfTurbine] == turbine_num]
            # 循环创建风速-功率折线
            fig.add_trace(go.Scatter(
                x=turbine_data[Field_WindSpeed],
                y=turbine_data[Field_ActiverPower],
                mode='lines',
                # line=dict(color=colors[idx % len(colors)]),
                name=f'{turbine_num}'  # 使用风电机组编号作为图例的名称
            )
            )

        if not ress.empty and Field_CutInWS in ress.columns and ress[Field_CutInWS].notna().any():
            cut_in_ws = ress[Field_CutInWS].min() - 1
        else:
            cut_in_ws = 2

        fig.add_trace(go.Scatter(
            x=dataFrameGuaranteePowerCurve[Field_WindSpeed],
            y=dataFrameGuaranteePowerCurve[Field_ActiverPower],
            # mode='lines',
            # line=dict(color='red', dash='dash'),
            mode='lines+markers',
            line=dict(color='red'),
            marker=dict(color='red', size=5),
            name='合同功率曲线',
            showlegend=True
        )
        )
        # 创建布局
        fig.update_layout(
            title={
                "text": f'功率曲线-{turbineModelInfo[Field_MachineTypeCode]}',
                'x': 0.5
            },
            # legend_title='Turbine',
            xaxis=dict(
                title='风速',
                dtick=1,
                tickangle=-45,
                range=[cut_in_ws, 25]
            ),
            yaxis=dict(
                title='有功功率',
                dtick=self.axisStepActivePower,
                range=[self.axisLowerLimitActivePower,
                       self.axisUpperLimitActivePower]
            ),
            legend=dict(
                orientation="h",  # Horizontal orientation
                xanchor="center",  # Anchor the legend to the center
                x=0.5,  # Position legend at the center of the x-axis
                y=-0.2,  # Position legend below the x-axis
                # itemsizing='constant',  # Keep the size of the legend entries constant
                # itemwidth=50
            )
        )

        # 保存HTML
        htmlFileName = '全场-{}-{}-功率曲线.html'.format(self.powerFarmInfo[Field_PowerFarmName].iloc[0],turbineModelInfo[Field_MillTypeCode])
        htmlFilePath = os.path.join(output_path, htmlFileName)
        fig.write_html(htmlFilePath)

        result_rows = []
        result_rows.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: Const_Output_Total,
            Field_Return_FilePath: htmlFilePath,
            Field_Return_IsSaveDatabase: False
        })

        result_df = pd.DataFrame(result_rows)
        return result_df

    def plot_single_power_curve(self, ress, group, dataFrameGuaranteePowerCurve: pd.DataFrame, turbineName, outputAnalysisDir, conf: Contract):

        fig = go.Figure()
        for turbine_num in ress[Field_NameOfTurbine].unique():
            turbine_data = ress[ress[Field_NameOfTurbine] == turbine_num]

            # 循环创建风速-功率折线
            fig.add_trace(go.Scatter(
                x=turbine_data[Field_WindSpeed],
                y=turbine_data[Field_ActiverPower],
                mode='lines',
                line=dict(color='lightgrey'),
                name=f'{turbine_num}',
                showlegend=False
            )
            )

        if not ress.empty and Field_CutInWS in ress.columns and ress[Field_CutInWS].notna().any():
            cut_in_ws = ress[Field_CutInWS].min() - 1
        else:
            cut_in_ws = 2

        fig.add_trace(go.Scatter(
            x=group[Field_WindSpeed],
            y=group[Field_ActiverPower],
            mode='lines',
            line=dict(color='darkblue'),
            name=Field_ActiverPower,
            showlegend=False
        )
        )

        fig.add_trace(go.Scatter(
            x=dataFrameGuaranteePowerCurve[Field_WindSpeed],
            y=dataFrameGuaranteePowerCurve[Field_ActiverPower],
            mode='lines+markers',
            line=dict(color='red'),
            marker=dict(color='red', size=5),
            name='合同功率曲线',
            showlegend=True
        )
        )

        # 创建布局
        fig.update_layout(
            title={
                "text": f'机组: {turbineName[0]}'
            },
            legend=dict(
                orientation="h",  # 或者 "v" 表示垂直
                yanchor="bottom",  # 图例垂直对齐方式
                y=0,  # 图例距离y轴下边界的距离（0到1之间）
                xanchor="right",  # 图例水平对齐方式
                x=1,  # 图例距离x轴右边界的距离（0到1之间）
                bgcolor='rgba(255,255,255,0)'
            ),
            xaxis=dict(
                title='风速',
                dtick=1,
                tickangle=-45,
                range=[cut_in_ws, 25]
            ),
            yaxis=dict(
                title='有功功率',
                dtick=self.axisStepActivePower,
                range=[self.axisLowerLimitActivePower,
                       self.axisUpperLimitActivePower]
            )
        )

        # 保存图像
        # pngFileName = f"{turbineName[0]}.png"
        # pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
        # fig.write_image(pngFilePath, scale=3)

        # # 保存HTML
        # htmlFileName = f"{turbineName[0]}.html"
        # htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        # fig.write_html(htmlFilePath)

        result_rows = []
        # result_rows.append({
        #     Field_Return_TypeAnalyst: self.typeAnalyst(),
        #     Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
        #     Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
        #     Field_CodeOfTurbine: turbineName[1],
        #     Field_Return_FilePath: pngFilePath,
        #     Field_Return_IsSaveDatabase: False
        # })

        # result_rows.append({
        #     Field_Return_TypeAnalyst: self.typeAnalyst(),
        #     Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
        #     Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
        #     Field_CodeOfTurbine: turbineName[1],
        #     Field_Return_FilePath: htmlFilePath,
        #     Field_Return_IsSaveDatabase: False
        # })

        result_df = pd.DataFrame(result_rows)
        return result_df