WTOAAM/dataAnalysisBusiness/algorithm/windDirectionFrequencyAnalyst.py

import os
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from behavior.analystNotFilter import AnalystNotFilter
from utils.directoryUtil import DirectoryUtil as dir
import matplotlib.pyplot as plt
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract


class WindDirectionFrequencyAnalyst(AnalystNotFilter):

    def typeAnalyst(self):
        return "wind_direction_frequency"

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary=self.processTurbineData(turbineCodes,conf,[Field_DeviceCode,Field_Time,Field_WindDirection,Field_WindSpeed,Field_ActiverPower])
        dataFrameMerge=self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        return self.windRoseAnalysis(dataFrameMerge, outputAnalysisDir, conf)

    def windRoseAnalysis(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, conf: Contract):
        # 检查所需列是否存在
        required_columns = {Field_WindDirection, Field_WindSpeed}
        if not required_columns.issubset(dataFrameMerge.columns) or dataFrameMerge[Field_WindDirection].isna().all() or dataFrameMerge[Field_WindSpeed].isna().all():
            msg=f"DataFrame缺少必要的列及值。涉及的列有: {required_columns}"
            # raise ValueError(msg)
            self.logger.warning(msg)
            return pd.DataFrame()

        # 风速区间
        bins = [0, 3, 6, 9, np.inf]
        speed_labels = ['[0,3)', '[3,6)', '[6,9)', '>=9']
        wind_directions = np.arange(0, 360, 22.5)

        colorscale = {
            '[0,3)': 'rgba(247.0, 251.0, 255.0, 1.0)',
            '[3,6)': 'rgba(171.33333333333334, 207.66666666666666, 229.66666666666669, 1.0)',
            '[6,9)': 'rgba(55.0, 135.0, 192.33333333333334, 1.0)',
            '>=9': 'rgba(8.0, 48.0, 107.0, 1.0)'
        }

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

        result_rows = []
        for name, group in grouped:
            speed_bins = pd.cut(
                group[Field_WindSpeed], bins=bins, labels=speed_labels)
            # 调整风向数据以使东方为0度
            # adjusted_wind_dir = (group[Field_WindDirection] - 90) % 360
            # group['风向分组'] = pd.cut(adjusted_wind_dir, bins=wind_directions, labels=wind_directions[:-1])
            group['风向分组'] = pd.cut(
                group[Field_WindDirection], bins=wind_directions, labels=wind_directions[:-1])

            # 初始化子图，设定为极坐标
            fig = make_subplots(rows=1, cols=1, specs=[[{'type': 'polar'}]])

            for label in speed_labels:
                subset = group[speed_bins == label]
                counts = subset['风向分组'].value_counts().reindex(
                    wind_directions[:-1], fill_value=0)
                # 转换为百分比
                percentage = (counts / counts.sum()) * 100

                # 创建 Barpolar 跟踪，并应用单色渐变
                trace = go.Barpolar(
                    r=percentage.values,
                    theta=counts.index,  # 这里的角度已经适配上北下南左西右东的布局
                    name=label,
                    marker_color=colorscale[label],  # 应用颜色尺度
                    marker_showscale=False,  # 不显示颜色条
                    marker_line_color='white',  # 设置线条颜色，增加扇区之间的分隔
                    marker_line_width=1  # 设置线条宽度
                )

                fig.add_trace(trace)

            # 设置图表的一些基本属性
            fig.update_layout(
                title={
                    "text": f"机组: {name[0]}",
                    #"x": 0.5
                },
                polar=dict(
                    radialaxis=dict(visible=True),
                    angularaxis=dict(
                        tickmode="array",
                        tickvals=wind_directions,
                        # 明确标注北、东、南、西等方向，以适应以北为0度的布局
                        ticktext=['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE',
                                  'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW']

                        # 更新角度标签，以适应以东为0度的布局
                        # ticktext=['E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW', 'N', 'NNE', 'NE', 'ENE']
                    )
                ),
                legend_title="风速",
                margin=dict(t=50, b=10)  # t为顶部（top）间距，b为底部（bottom）间距
            )

            # 保存图像
            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)

            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_Return_FilePath: filePathOfHtml,
                Field_Return_IsSaveDatabase: True
            })

        result_df = pd.DataFrame(result_rows)

        return result_df