WTOAAM/dataAnalysisBusiness/algorithm/powerScatterAnalyst.py

import os

import pandas as pd
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodBadPoint import AnalystWithGoodBadPoint


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

    def typeAnalyst(self):
        return "power_scatter"

    def selectColumns(self):
        return [Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower]

    def processDateTime(self, dataFrame: pd.DataFrame, fieldTime:str = None):
        super().processDateTime(dataFrame,Field_YearMonth)

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf, self.selectColumns())
        dataFrame = self.userDataFrame(dictionary, conf.dataContract.configAnalysis, self)
        if len(dataFrame) <= 0:
            print("After screening for blade pitch angle less than the configured value, plot power curve scatter points without data")
            return

        # dataFrameGuaranteePowerCurve=self.powerFarmInfo[Field_PowerContractURL]
        # grouped=self.dataFrameContractOfTurbine.groupby([Field_PowerFarmCode, Field_MillTypeCode])

        # for groupByKey,contractPowerCurveOfMillType in grouped:
        #     break
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)
        return self.drawOfPowerCurveScatter(dataFrame,turbineInfos, outputAnalysisDir, conf)

    """
    def contractGuaranteePowerCurveData(self, csvPowerCurveFilePath):
        dataFrameGuaranteePowerCurve = pd.read_csv(
            csvPowerCurveFilePath, encoding=charset_unify)

        return dataFrameGuaranteePowerCurve
    """
    def drawOfPowerCurveScatter(self, dataFrame: pd.DataFrame,  turbineModelInfo: pd.Series, outputAnalysisDir, conf: Contract):
        """
        绘制风速-功率分布图并保存为文件。

        参数:
        dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        csvPowerCurveFilePath (str): 功率曲线文件路径。
        outputAnalysisDir (str): 分析输出目录。
        conf (ConfBusiness): 配置
        """

        colorsList = [  
                        "#3E409C",
                        "#3586BF",
                        "#52A3AE",
                        "#85D0AE",
                        "#A8DCA2",
                        "#FBFFBE",
                        "#FDF1A9",
                        "#FFE286",
                        "#FCB06C",
                        "#F96F4A",
                        "#E4574C",
                        "#AF254F"]
        cutInWsField = self.turbineModelInfo[Field_CutInWS]
        cut_in_ws = cutInWsField.min() - 1 if cutInWsField.notna().any() else 2
        # if not dataFrame.empty and Field_CutInWS in  dataFrame.columns and  dataFrame[Field_CutInWS].notna().any():
        #     cut_in_ws =  dataFrame[Field_CutInWS].min() - 1
        # else:
        #     cut_in_ws = 2

        grouped = dataFrame.groupby([Field_NameOfTurbine, Field_CodeOfTurbine])

        result_rows = []
        # 遍历每个设备的数据
        for name, group in grouped:
            # 创建颜色映射，将每个年月映射到一个唯一的颜色
            unique_months = group[Field_YearMonth].unique()
            colors = [
                colorsList[i % 12] for i in range(len(unique_months))]
            color_map = dict(zip(unique_months, colors))

            # 使用go.Scatter3d创建3D散点图
            trace = go.Scatter3d(
                x=group[Field_WindSpeed],
                y=group[Field_YearMonth],
                z=group[Field_ActiverPower],
                mode='markers',
                marker=dict(
                    color=[color_map[month]
                           for month in group[Field_YearMonth]],
                    size=1.5,
                    line=dict(
                        color='rgba(0, 0, 0, 0)',  # 设置边框颜色为透明，以去掉白色边框
                        width=0  # 设置边框宽度为0，进一步确保没有边框
                    ),
                    opacity=0.8  # 调整散点的透明度，增加透视效果
                )
            )

            # 创建图形
            fig = go.Figure(data=[trace])

            # 更新图形的布局
            fig.update_layout(
                title={
                    "text": f'当月发电量: {name[0]}',
                    "x": 0.5
                },
                scene=dict(
                    xaxis=dict(
                        title='风速',
                        range=[cut_in_ws, 25],
                    ),
                    yaxis=dict(
                        title='时间',
                        tickmode='array',
                        tickvals=unique_months,
                        ticktext=unique_months,
                        categoryorder='category ascending'
                    ),
                    zaxis=dict(
                        title='功率',
                        range=[self.axisLowerLimitActivePower, self.axisUpperLimitActivePower],
                        dtick=self.axisStepActivePower,
                    )
                ),
                scene_camera=dict(
                    up=dict(x=0, y=0, z=1),  # 保持相机向上方向不变
                    center=dict(x=0, y=0, z=0),  # 保持相机中心位置不变
                    eye=dict(x=-1.8, y=-1.8, z=1.2)  # 调整eye属性以实现水平旋转180°
                ),
                margin=dict(t=50, b=10)  # t为顶部（top）间距，b为底部（bottom）间距
            )

            # 确保从 Series 中提取的是具体的值
            engineTypeCode = turbineModelInfo.get(Field_MillTypeCode, "")
            if isinstance(engineTypeCode, pd.Series):
                engineTypeCode = engineTypeCode.iloc[0]

            engineTypeName = turbineModelInfo.get(Field_MachineTypeCode, "")
            if isinstance(engineTypeName, pd.Series):
                engineTypeName = engineTypeName.iloc[0]
            # 构建最终的JSON对象
            json_output = {
                "analysisTypeCode": "逐月有功功率散点3D分析",
                "engineCode":  engineTypeCode,    
                "engineTypeName": engineTypeName, 
                "xaixs": "风速（m/s）",
                "yaixs": "时间",
                "zaixs": "有功功率（kw)",
                "data": [{
                        "engineName": name[0],
                        "engineCode": name[1],
                        "title":f'当月发电量: {name[0]}',
                        "xData": group[Field_WindSpeed].tolist(),
                        "xrange":[cut_in_ws, 25],
                        "yData":group[Field_YearMonth].tolist(),
                        "zData":group[Field_ActiverPower].tolist(),
                        "zrange":[self.axisLowerLimitActivePower, self.axisUpperLimitActivePower]
                        }]
            }
            # # Save plot
            # filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html")
            # fig.write_html(filePathOfHtml)
            # 将JSON对象保存到文件
            output_json_path = os.path.join(outputAnalysisDir, f"power_scatter{name[0]}.json")
            with open(output_json_path, 'w', encoding='utf-8') as f:
                import json
                json.dump(json_output, f, ensure_ascii=False, indent=4)

            # 如果需要返回DataFrame，可以包含文件路径
            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: name[1],
                Field_Return_FilePath: output_json_path,
                Field_Return_IsSaveDatabase: True
            })
            # result_rows.append({
            #     Field_Return_TypeAnalyst: self.typeAnalyst(),
            #     Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            #     Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            #     Field_CodeOfTurbine: name[1],
            #     Field_Return_FilePath: filePathOfHtml,
            #     Field_Return_IsSaveDatabase: True
            # })

        result_df = pd.DataFrame(result_rows)

        return result_df