WTOAAM/dataAnalysisBusiness/algorithm/pitchGeneratorSpeedAnalyst.py

import os
from datetime import datetime

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.analystWithGoodBadPoint import AnalystWithGoodBadPoint


class PitchGeneratorSpeedAnalyst(AnalystWithGoodBadPoint):
    """
    风电机组变桨-发电机转速分析
    """

    def typeAnalyst(self):
        return "pitch_generator_speed"

    # def recalculation(self, turbineModelInfo: pd.Series, dataFrame: pd.DataFrame):
    #     return self.recalculationOfGeneratorSpeed(
    #         dataFrame, Field_RotorSpeed, Field_GeneratorSpeed,  self.turbineModelInfo[Field_RSR].iloc[0])
    def selectColumns(self):
        return [Field_DeviceCode, Field_Time,Field_WindSpeed,Field_ActiverPower,Field_PitchAngel1, Field_GeneratorSpeed]

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf,self.selectColumns())
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes,
                                                   self.turbineInfo)
        dataFrame = self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        return self.plot_speed_pitch_angle(dataFrame, turbineInfos, outputAnalysisDir, conf, )

    def plot_speed_pitch_angle(self, dataFrameMerge, turbineModelInfo: pd.Series, outputAnalysisDir, conf: Contract):
        # 按设备名分组数据
        dataFrameMerge = dataFrameMerge[(dataFrameMerge[Field_ActiverPower] > 0)].sort_values(by=Field_YearMonth)
        grouped = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine])

        # 遍历每个设备并绘制散点图
        result_rows = []
        for name, group in grouped:
            groupNew = group.copy()

            # 处理直驱转速与非直驱的差异
            # if not self.common.isNone(conf.value_gen_speed_multiple):
            #     groupNew[fieldGeneratorSpeed] = group[fieldGeneratorSpeed] * \
            #         conf.value_gen_speed_multiple

            # 创建图形
            fig = go.Figure()

            # 添加散点图
            fig.add_trace(go.Scatter(
                x=groupNew[Field_GeneratorSpeed],
                y=groupNew[Field_PitchAngel1],
                mode='markers',
                # marker=dict(color='blue', size=3.5)
                marker=dict(
                    color=group[Field_UnixYearMonth],
                    colorscale='Rainbow',
                    size=3,
                    opacity=0.7,
                    colorbar=dict(
                        tickvals=np.linspace(
                            group[Field_UnixYearMonth].min(), group[Field_UnixYearMonth].max(), 6),
                        ticktext=[datetime.fromtimestamp(ts).strftime('%Y-%m') for ts in np.linspace(
                            group[Field_UnixYearMonth].min(), group[Field_UnixYearMonth].max(), 6)],
                        thickness=18,
                        len=1,  # 设置颜色条的长度，使其占据整个图的高度
                        outlinecolor='rgba(255,255,255,0)'
                    ),
                    showscale=True
                ),
                showlegend=False
            ))

            # 设置图形布局
            fig.update_layout(
                title=f'机组: {name[0]}',
                xaxis=dict(
                    title='发电机转速',
                    range=[self.axisLowerLimitGeneratorSpeed, self.axisUpperLimitGeneratorSpeed],
                    dtick=self.axisStepGeneratorSpeed,
                    tickangle=-45 # 设置x轴刻度值旋转角度为45度，如果需要
                ),
                yaxis=dict(
                    title='桨距角',
                    range=[self.axisLowerLimitPitchAngle, self.axisUpperLimitPitchAngle],
                    dtick=self.axisStepPitchAngle
                ),
                coloraxis=dict(
                    colorbar=dict(
                        title="时间",
                        ticks="outside",
                        len=1,  # 设置颜色条的长度，使其占据整个图的高度
                        thickness=20,  # 调整颜色条的宽度
                        orientation='v',  # 设置颜色条为垂直方向
                        tickmode='array',  # 确保刻度按顺序排列
                        tickvals=dataFrameMerge[Field_YearMonth].unique(
                        ).tolist(),  # 确保刻度为唯一的年月
                        ticktext=dataFrameMerge[Field_YearMonth].unique(
                        ).tolist()  # 以%Y-%m格式显示标签
                    )
                )
            )
            # 确保从 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]

            # 使用 apply() 对每个元素调用 datetime.fromtimestamp
            group[Field_UnixYearMonth] = group[Field_UnixYearMonth].apply(lambda x: datetime.fromtimestamp(x).strftime('%Y-%m'))

            # 构建最终的JSON对象
            json_output = {
                "analysisTypeCode": "变桨和发电机转速协调性分析",
                "engineCode": engineTypeCode,
                "engineTypeName": engineTypeName,
                "xaixs": "发电机转速(r/min)",
                "yaixs": "桨距角(°)",
                "data": [{
                    "engineName": name[0],
                    "engineCode": name[1],
                    "title": f' 机组: {name[0]}',
                    "xData": groupNew[Field_GeneratorSpeed].tolist(),
                    "yData": groupNew[Field_PitchAngel1].tolist(),
                    "colorbar":group[Field_UnixYearMonth] .tolist(),

                }]
            }
            # Save plot
            # filePathOfImage = os.path.join(outputAnalysisDir, f"{name[0]}.png")
            # fig.write_image(filePathOfImage, scale=3)
            # filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html")
            # fig.write_html(filePathOfHtml)

            # 将JSON对象保存到文件
            output_json_path = os.path.join(outputAnalysisDir, f"pitch_GeneratorSpeed_Analyst{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)

            # 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: filePathOfImage,
            #     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: name[1],
                Field_MillTypeCode: 'total',
                Field_Return_FilePath: output_json_path,
                Field_Return_IsSaveDatabase: True
            })

        result_df = pd.DataFrame(result_rows)

        return result_df