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.analyst import Analyst
from utils.directoryUtil import DirectoryUtil as dir
import matplotlib.pyplot as plt
from algorithmContract.confBusiness import *


class WindDirectionFrequencyAnalyst(Analyst):

    def typeAnalyst(self):
        return "wind_direction_frequency"

    def turbinesAnalysis(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness):
        self.windRoseAnalysis(dataFrameMerge, outputAnalysisDir, confData)

    def windRoseAnalysis(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, confData: ConfBusiness):
        # 检查所需列是否存在
        required_columns = {confData.field_wind_dir, confData.field_wind_speed}
        if not required_columns.issubset(dataFrameMerge.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")

        # 风速区间
        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)

        for name, group in grouped:
            speed_bins = pd.cut(
                group[confData.field_wind_speed], bins=bins, labels=speed_labels)
            # 调整风向数据以使东方为0度
            # adjusted_wind_dir = (group[confData.field_wind_dir] - 90) % 360
            # group['风向分组'] = pd.cut(adjusted_wind_dir, bins=wind_directions, labels=wind_directions[:-1])
            group['风向分组'] = pd.cut(
                group[confData.field_wind_dir], 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"Wind Rose {name}",
                    "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="Wind Speed",
                margin=dict(t=50, b=10)  # t为顶部（top）间距，b为底部（bottom）间距
            )

            # # 保存html
            # outputFileHtml = os.path.join(outputAnalysisDir, f"{name}.html")
            # fig.write_html(outputFileHtml)
            # 保存图像
            output_file = os.path.join(outputAnalysisDir, f"{name}.png")
            fig.write_image(output_file, scale=2)