WTOAAM/dataAnalysisBusiness/algorithm/temperatureEnvironmentAnalyst.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.analystWithGoodBadLimitPoint import AnalystWithGoodBadLimitPoint
from geopy.distance import geodesic
from plotly.subplots import make_subplots


class TemperatureEnvironmentAnalyst(AnalystWithGoodBadLimitPoint):
    """
    风电机组环境温度传感器分析
    """

    def typeAnalyst(self):
        return "temperature_environment"

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

        #  环境温度分析
        turbineEnvTempData = dataFrameOfTurbines.groupby(Field_CodeOfTurbine).agg(
            {Field_EnvTemp: 'median'})
        turbineEnvTempData = turbineEnvTempData.reset_index()
        mergeData = self.mergeData(self.turbineInfo, turbineEnvTempData)
        # 分机型
        turbrineInfos = self.common.getTurbineInfos(
            conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)

        returnResult= self.draw(mergeData, outputAnalysisDir, conf,turbrineInfos)

        return returnResult
        # return self.draw(mergeData, outputAnalysisDir, conf)

    def mergeData(self,  turbineInfos: pd.DataFrame, turbineEnvTempData):
        """
        将每台机组的环境温度均值数据与机组信息，按机组合并

        参数:
        turbineInfos (pandas.DataFrame): 机组信息数据
        turbineEnvTempData (pandas.DataFrame): 每台机组的环境温度均值数据

        返回:
        pandas.DataFrame: 每台机组的环境温度均值数据与机组信息合并数据
        """

        """
        合并类型how的选项包括：

        'inner': 内连接，只保留两个DataFrame中都有的键的行。
        'outer': 外连接，保留两个DataFrame中任一或两者都有的键的行。
        'left': 左连接，保留左边DataFrame的所有键，以及右边DataFrame中匹配的键的行。
        'right': 右连接，保留右边DataFrame的所有键，以及左边DataFrame中匹配的键的行。
        """
        # turbineInfos[fieldTurbineName]=turbineInfos[fieldTurbineName].astype(str).apply(confData.add_W_if_starts_with_digit)
        # turbineEnvTempData[Field_NameOfTurbine] = turbineEnvTempData[Field_NameOfTurbine].astype(
        #     str)
        tempDataFrame = pd.merge(turbineInfos, turbineEnvTempData, on=[
                              Field_CodeOfTurbine], how='inner')

        # 保留指定字段，例如 'Key' 和 'Value1'
        mergeDataFrame = tempDataFrame[[Field_CodeOfTurbine, Field_NameOfTurbine, Field_Latitude,Field_Longitude, Field_EnvTemp]]

        return mergeDataFrame

    # 定义查找给定半径内点的函数
    def find_points_within_radius(self, data, center, field_temperature_env, radius):
        points_within_radius = []
        for index, row in data.iterrows():
            distance = geodesic(
                (center[2], center[1]), (row[Field_Latitude], row[Field_Longitude])).meters
            if distance <= radius:
                points_within_radius.append(
                    (row[Field_NameOfTurbine], row[field_temperature_env]))
        return points_within_radius

    fieldTemperatureDiff = "temperature_diff"

    # def draw(self, dataFrame: pd.DataFrame, outputAnalysisDir, conf: Contract, charset=charset_unify):
    def draw(self, dataFrame: pd.DataFrame, outputAnalysisDir, conf: Contract, turbineModelInfo: pd.Series):
        # +1. 去除经纬度或温度为空的数据
        dataFrame = dataFrame.dropna(subset=[Field_NameOfTurbine, Field_Longitude, Field_Latitude, Field_EnvTemp])

        # +2. 筛选合法的经纬度范围 (纬度 [-90, 90], 经度 [-180, 180]); geopy 对纬度非常敏感，超出范围会直接报错
        valid_lat = (dataFrame[Field_Latitude] >= -90) & (dataFrame[Field_Latitude] <= 90)
        valid_lon = (dataFrame[Field_Longitude] >= -180) & (dataFrame[Field_Longitude] <= 180)
        dataFrame = dataFrame[valid_lat & valid_lon]

        # +3. 如果筛选后没有数据，打印日志并返回空 DataFrame，避免后续报错
        if dataFrame.empty:
            print(f"Warning: {self.typeAnalyst()} filtered no data (invalid coordinates or missing values).")
            return pd.DataFrame()

        # 处理数据
        dataFrame['new'] = dataFrame.loc[:, [Field_NameOfTurbine,
                                             Field_Longitude, Field_Latitude, Field_EnvTemp]].apply(tuple, axis=1)
        coordinates = dataFrame['new'].tolist()
        # df = pd.DataFrame(coordinates, columns=[Field_NameOfTurbine, Field_Longitude, Field_Latitude, confData.field_env_temp])

        # 查找半径内的点
        points_within_radius = {coord: self.find_points_within_radius(
            dataFrame, coord, Field_EnvTemp, self.turbineModelInfo[Field_RotorDiameter].iloc[0]*10) for coord in coordinates}
        res = []
        for center, nearby_points in points_within_radius.items():
            current_temp = dataFrame[dataFrame[Field_NameOfTurbine]
                                     == center[0]][Field_EnvTemp].iloc[0]
            target_tuple = (center[0], current_temp)
            if target_tuple in nearby_points:
                nearby_points.remove(target_tuple)
            median_temp = np.median(
                [i[1] for i in nearby_points]) if nearby_points else current_temp
            res.append((center[0], nearby_points, median_temp, current_temp))
        res = pd.DataFrame(
            res, columns=[Field_NameOfTurbine, '周边机组', '周边机组温度', '当前机组温度'])
        res[self.fieldTemperatureDiff] = res['当前机组温度'] - res['周边机组温度']

        # 使用plotly进行数据可视化
        fig1 = make_subplots(rows=1, cols=1)

        # 温度差异条形图
        fig1.add_trace(
            go.Bar(x=res[Field_NameOfTurbine],
                   y=res[self.fieldTemperatureDiff], marker_color='dodgerblue'),
            row=1, col=1
        )
        fig1.update_layout(
            title={'text': f'温度偏差', 'x': 0.5},
            xaxis_title='机组名称',
            yaxis_title='温度偏差',
            shapes=[
                {'type': 'line', 'x0': 0, 'x1': 1, 'xref': 'paper', 'y0': 5,
                    'y1': 5, 'line': {'color': 'red', 'dash': 'dot'}},
                {'type': 'line', 'x0': 0, 'x1': 1, 'xref': 'paper', 'y0': -
                    5, 'y1': -5, 'line': {'color': 'red', 'dash': 'dot'}}
            ],
            xaxis=dict(tickangle=-45)  # 设置x轴刻度旋转角度为45度
        )

        # 确保从 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": "风电机组环境温度传感器分析",
            "engineCode": engineTypeCode,
            "engineTypeName": engineTypeName,
            "xaixs": "机组名称",
            "yaixs": "温度偏差(℃)",
            "data": [{
                "engineName": "",  # Field_NameOfTurbine
                "engineCode": "",  # Field_CodeOfTurbine
                "title": f'温度偏差',
                "xData": res[Field_NameOfTurbine].tolist(),
                "yData": res[self.fieldTemperatureDiff].tolist(),
            }]
        }

        result_rows = []
        # 保存图像
        # pngFileName = '{}环境温差Bias.png'.format(
        #     self.powerFarmInfo[Field_PowerFarmName].iloc[0])
        # pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
        # fig1.write_image(pngFilePath, scale=3)

        # 保存HTML
        # htmlFileName = '{}环境温差Bias.html'.format(
        #    self.powerFarmInfo[Field_PowerFarmName].iloc[0])
        # htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        # fig1.write_html(htmlFilePath)

        # 将JSON对象保存到文件
        output_json_path = os.path.join(outputAnalysisDir, f"total_Bias.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: 'total',
            Field_MillTypeCode: 'total_Bias',
            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: Const_Output_Total,
        #     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: Const_Output_Total,
        #    Field_Return_FilePath: htmlFilePath,
        #    Field_Return_IsSaveDatabase: True
        # })

        # 环境温度中位数条形图
        fig2 = make_subplots(rows=1, cols=1)
        fig2.add_trace(
            go.Bar(x=res[Field_NameOfTurbine],
                   y=res['当前机组温度'], marker_color='dodgerblue'),
            row=1, col=1
        )
        fig2.update_layout(
            title={'text': f'平均温度', 'x': 0.5},
            xaxis_title='机组名称',
            yaxis_title=' 温度',
            xaxis=dict(tickangle=-45)  # 为x轴也设置旋转角度
        )

        # 确保从 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": "风电机组环境温度传感器分析",
            "engineCode": engineTypeCode,
            "engineTypeName": engineTypeName,
            "xaixs": "机组名称",
            "yaixs": "温度(℃)",
            "data": [{
                "engineName": "",  # Field_NameOfTurbine
                "engineCode": "",  # Field_CodeOfTurbine
                "title": f'平均温度',
                "xData": res[Field_NameOfTurbine].tolist(),
                "yData": res['当前机组温度'].tolist(),
            }]
        }


        # 保存图像
        # pngFileName = '{}环境温度中位数.png'.format(
        #     self.powerFarmInfo[Field_PowerFarmName].iloc[0])
        # pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
        # fig2.write_image(pngFilePath, scale=3)

        # 保存HTML
        # htmlFileName = '{}环境温度中位数.html'.format(
        #     self.powerFarmInfo[Field_PowerFarmName].iloc[0])
        # htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        # fig2.write_html(htmlFilePath)

        # 将JSON对象保存到文件
        output_json_path = os.path.join(outputAnalysisDir, f"total_Mid.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: 'total',
            Field_MillTypeCode: 'total_Mid',
            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: Const_Output_Total,
        #     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: Const_Output_Total,
        #    Field_Return_FilePath: htmlFilePath,
        #    Field_Return_IsSaveDatabase: True
        # })

        result_df = pd.DataFrame(result_rows)

        return result_df
        """
        fig, ax = plt.subplots(figsize=(16,8),dpi=96)
       
        # 设置x轴刻度值旋转角度为45度  
        plt.tick_params(axis='x', rotation=45)

        sns.barplot(x=Field_NameOfTurbine,y=self.fieldTemperatureDiff,data=res,ax=ax,color='dodgerblue')
        plt.axhline(y=5,ls=":",c="red")#添加水平直线
        plt.axhline(y=-5,ls=":",c="red")#添加水平直线
        ax.set_ylabel('temperature_difference')
        ax.set_title('temperature Bias')
        plt.savefig(outputAnalysisDir +'//'+ "{}环境温差Bias.png".format(confData.farm_name),bbox_inches='tight',dpi=120)


        fig2, ax2 = plt.subplots(figsize=(16,8),dpi=96)
        # 设置x轴刻度值旋转角度为45度  
        plt.tick_params(axis='x', rotation=45)
        
        sns.barplot(x=Field_NameOfTurbine ,y='当前机组温度',data=res,ax=ax2,color='dodgerblue')
        ax2.set_ylabel('temperature')
        ax2.set_title('temperature median')
        plt.savefig(outputAnalysisDir +'//'+ "{}环境温度均值.png".format(confData.farm_name),bbox_inches='tight',dpi=120)
        """