添加振动数据部分实现

etl/wind_power/wave/WaveTrans.py

@@ -1,6 +1,11 @@
+import datetime
+import json
 import multiprocessing
-import os.path
+from os.path import basename, dirname
 
+import pandas as pd
+
+from service.trans_service import get_wave_conf, save_file_to_db, save_df_to_db
 from utils.file.trans_methods import *
 from utils.systeminfo.sysinfo import get_available_cpu_count_with_percent
 
@@ -11,38 +16,78 @@ class WaveTrans(object):
         self.field_code = field_code
         self.read_path = read_path
         self.save_path = save_path
+        self.begin = datetime.datetime.now()
 
-    def get_data(self, file_path):
-        df = pd.read_csv(file_path, encoding=detect_file_encoding(file_path), header=None)
-        data = [i for i in df[0].values]
-        filename = os.path.basename(file_path)
-        wind_num = filename.split('_')[1]
-        cedian = '齿轮箱' + filename.split('_齿轮箱')[1].split('_Time')[0]
-        cedian_time = filename.split('风机_')[1].split('_齿轮箱')[0].replace('_', ':')
-        name_tmp = 'Time_' + filename.split('Time_')[1].split('_cms')[0]
-        pinlv = name_tmp[0:name_tmp.rfind('_')]
-        zhuansu = name_tmp[name_tmp.rfind('_') + 1:]
-
-        df = pd.DataFrame()
-        df['风机编号'] = [wind_num, wind_num]
-        df['时间'] = [cedian_time, cedian_time]
-        df['频率'] = [pinlv, pinlv]
-        df['测点'] = ['转速', cedian]
-        df['数据'] = [[float(zhuansu)], data]
-
-        return df
-
-    def run(self):
-        all_files = read_files(self.read_path, ['csv'])
+    # def get_data(self, file_path):
+    #     df = pd.read_csv(file_path, encoding=detect_file_encoding(file_path), header=None)
+    #     data = [i for i in df[0].values]
+    #     filename = os.path.basename(file_path)
+    #     wind_num = filename.split('_')[1]
+    #     cedian = '齿轮箱' + filename.split('_齿轮箱')[1].split('_Time')[0]
+    #     cedian_time = filename.split('风机_')[1].split('_齿轮箱')[0].replace('_', ':')
+    #     name_tmp = 'Time_' + filename.split('Time_')[1].split('_cms')[0]
+    #     pinlv = name_tmp[0:name_tmp.rfind('_')]
+    #     zhuansu = name_tmp[name_tmp.rfind('_') + 1:]
+    #
+    #     df = pd.DataFrame()
+    #     df['风机编号'] = [wind_num, wind_num]
+    #     df['时间'] = [cedian_time, cedian_time]
+    #     df['频率'] = [pinlv, pinlv]
+    #     df['测点'] = ['转速', cedian]
+    #     df['数据'] = [[float(zhuansu)], data]
+    #
+    #     return df
 
+    def get_data_exec(self, func_code, arg):
+        exec(func_code)
+        return locals()['get_data'](arg)
+
+    def run(self, map_dict=dict()):
+        all_files = read_files(self.read_path, ['csv'])
+        print(len)
         # 最大取系统cpu的 1/2
         split_count = get_available_cpu_count_with_percent(1 / 2)
 
+        wave_conf = get_wave_conf(self.field_code)
+
+        base_param_exec = wave_conf['base_param_exec']
+        map_dict = {}
+        if base_param_exec:
+            base_param_exec = base_param_exec.replace('\r\n', '\n').replace('\t', '    ')
+            print(base_param_exec)
+            # exec(base_param_exec)
+
+            mesure_poins = [key for key, value in wave_conf.items() if str(key).startswith('conf_') and value]
+
+            for point in mesure_poins:
+                map_dict[wave_conf[point]] = point
+
         with multiprocessing.Pool(split_count) as pool:
-            dfs = pool.starmap(self.get_data, [(i,) for i in all_files])
+            file_datas = pool.starmap(self.get_data_exec, [(base_param_exec, i) for i in all_files])
+
+        # for file_data in file_datas:
+        #     wind_num, data_time, frequency, rotational_speed, measurementp_name, data = file_data[0], file_data[1], \
+        #     file_data[2], file_data[3], file_data[4],
+
+        result_list = list()
+        for file_data in file_datas:
+            wind_turbine_name, time_stamp, sampling_frequency, rotational_speed, mesure_point_name, mesure_data = \
+                file_data[0], file_data[1], file_data[2], file_data[3], file_data[4], file_data[5]
+
+            result_list.append(
+                [wind_turbine_name, time_stamp, sampling_frequency, 'rotational_speed', [float(rotational_speed)]])
+
+            result_list.append(
+                [wind_turbine_name, time_stamp, sampling_frequency, mesure_point_name, mesure_data])
+
+        df = pd.DataFrame(result_list,
+                          columns=['wind_turbine_name', 'time_stamp', 'sampling_frequency', 'mesure_point_name',
+                                   'mesure_data'])
+
+        df['mesure_point_name'] = df['mesure_point_name'].map(map_dict).fillna(df['mesure_point_name'])
 
-        df = pd.concat(dfs, ignore_index=True, copy=False)
+        df['mesure_data'] = df['mesure_data'].apply(lambda x: json.dumps(x))
 
-        df.drop_duplicates(subset=['风机编号', '时间', '频率', '测点'], keep='last')
+        save_df_to_db('SKF001_wave', df, batch_count=1000)
 
-        df.to_csv(os.path.join(self.save_path, self.field_code + '.csv'), index=False, encoding='utf8')
+        print("总耗时:", datetime.datetime.now() - self.begin)

service/trans_service.py

@@ -21,6 +21,15 @@ def get_min_sec_conf(field_code, trans_type) -> dict:
     return res[0]
 
 
+def get_min_sec_conf_test(field_code, trans_type) -> dict:
+    query_sql = "SELECT * FROM trans_conf where wind_name = %s and type = %s and status = 1"
+    res = trans.execute(query_sql, (field_code, trans_type))
+    print(res)
+    if type(res) == tuple:
+        return None
+    return res[0]
+
+
 def get_fault_warn_conf(field_code, trans_type) -> dict:
     types = list()
     if trans_type == 'fault':
@@ -40,6 +49,7 @@ def get_fault_warn_conf(field_code, trans_type) -> dict:
         return None
     return res[0]
 
+
 def get_wave_conf(field_code) -> dict:
     query_sql = "SELECT * FROM wave_conf where wind_code = %s and status = 1"
     res = trans.execute(query_sql, (field_code))
@@ -48,6 +58,7 @@ def get_wave_conf(field_code) -> dict:
         return None
     return res[0]
 
+
 def creat_min_sec_table(table_name, win_names, read_type):
     create_sql = f"""
     CREATE TABLE
@@ -141,6 +152,14 @@ def drop_table(table_name, save_db=True):
             trans_print(traceback.format_exc())
 
 
+def clear_table(table_name, save_db=True):
+    if save_db:
+        rename_sql = f"truncate TABLE `{table_name}`"
+        try:
+            trans.execute(rename_sql)
+        except:
+            trans_print(traceback.format_exc())
+
 def save_file_to_db(table_name: str, file: str, batch_count=100000):
     base_name = os.path.basename(file)
     try:
@@ -155,6 +174,18 @@ def save_file_to_db(table_name: str, file: str, batch_count=100000):
         raise Exception(message)
 
 
+def save_df_to_db(table_name: str, df: pd.DataFrame(), batch_count=100000):
+    split_dfs = [df.iloc[i:i + batch_count] for i in range(0, len(df), batch_count)]
+    try:
+        for i, split_df in enumerate(split_dfs):
+            trans.execute_df_save(split_df, table_name)
+            count = (i + 1) * batch_count
+            trans_print(f"Chunk {count} written to MySQL.")
+    except Exception as e:
+        trans_print(traceback.format_exc())
+        raise Exception(str(e))
+
+
 def batch_statistics(table_name):
     query_sql = f"select count(1) as total_count ,min(t.time_stamp) as min_date ,max(t.time_stamp) as max_date from `{table_name}` t "
     try:

tmp_file/gradio_web.py

@@ -0,0 +1,205 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/6/3
+# @Author  : 魏志亮
+import copy
+
+import gradio as gr
+import yaml
+
+from service.plt_service import get_all_wind_company
+from service.trans_service import get_min_sec_conf, get_min_sec_conf_test
+
+
+# from utils.db.trans_mysql import *
+
+
+def test_click(wind_name, wind_full_name, type, is_vertical_table, merge_columns, vertical_read_cols,
+               vertical_index_cols, vertical_col_key, vertical_col_value, resolve_col_prefix, wind_name_exec,
+               wind_turbine_number, time_stamp, active_power, rotor_speed, generator_speed, wind_velocity,
+               pitch_angle_blade_1, pitch_angle_blade_2, pitch_angle_blade_3, cabin_position, true_wind_direction,
+               yaw_error1, set_value_of_active_power, gearbox_oil_temperature, generatordrive_end_bearing_temperature,
+               generatornon_drive_end_bearing_temperature, wind_turbine_status, wind_turbine_status2, cabin_temperature,
+               twisted_cable_angle, front_back_vibration_of_the_cabin, side_to_side_vibration_of_the_cabin,
+               actual_torque, given_torque, clockwise_yaw_count, counterclockwise_yaw_count, unusable,
+               power_curve_available, required_gearbox_speed, inverter_speed_master_control, outside_cabin_temperature,
+               main_bearing_temperature, gearbox_high_speed_shaft_bearing_temperature,
+               gearboxmedium_speed_shaftbearing_temperature, gearbox_low_speed_shaft_bearing_temperature,
+               generator_winding1_temperature, generator_winding2_temperature, generator_winding3_temperature,
+               turbulence_intensity, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10
+               ):
+    params = copy.deepcopy(vars())
+
+    error_message = ""
+    if wind_name is None or wind_name.strip() == '':
+        error_message += "风机名称必选"
+        gr.Warning(error_message)
+        return error_message
+
+    if wind_full_name is None or wind_full_name.strip() == '':
+        error_message += "风机全称必选"
+        gr.Warning(error_message)
+        return error_message
+
+    # save_to_trans_conf(params)
+    return yaml.dump(vars(), allow_unicode=True, sort_keys=False)
+
+
+def fill_data(wind_name, type):
+    select_cols = ['wind_full_name', 'is_vertical_table', 'merge_columns', 'vertical_read_cols',
+                   'vertical_index_cols', 'vertical_col_key', 'vertical_col_value', 'resolve_col_prefix',
+                   'wind_name_exec',
+                   'wind_turbine_number', 'time_stamp', 'active_power', 'rotor_speed', 'generator_speed',
+                   'wind_velocity', 'pitch_angle_blade_1', 'pitch_angle_blade_2', 'pitch_angle_blade_3',
+                   'cabin_position', 'true_wind_direction', 'yaw_error1', 'set_value_of_active_power',
+                   'gearbox_oil_temperature', 'generatordrive_end_bearing_temperature',
+                   'generatornon_drive_end_bearing_temperature', 'wind_turbine_status', 'wind_turbine_status2',
+                   'cabin_temperature', 'twisted_cable_angle', 'front_back_vibration_of_the_cabin',
+                   'side_to_side_vibration_of_the_cabin', 'actual_torque', 'given_torque', 'clockwise_yaw_count',
+                   'counterclockwise_yaw_count', 'unusable', 'power_curve_available', 'required_gearbox_speed',
+                   'inverter_speed_master_control', 'outside_cabin_temperature', 'main_bearing_temperature',
+                   'gearbox_high_speed_shaft_bearing_temperature', 'gearboxmedium_speed_shaftbearing_temperature',
+                   'gearbox_low_speed_shaft_bearing_temperature', 'generator_winding1_temperature',
+                   'generator_winding2_temperature', 'generator_winding3_temperature', 'turbulence_intensity', 'param1',
+                   'param2', 'param3', 'param4', 'param5', 'param6', 'param7', 'param8', 'param9', 'param10']
+    print(wind_name, type)
+    df = get_min_sec_conf_test(wind_name, type)
+    print(df)
+    if df.keys() == 0:
+        return [''] * len(select_cols)
+    result = []
+    for col in select_cols:
+        result.append(df[col])
+    return result[0], result[1], result[2], result[3], result[4], result[5], result[6], result[7], \
+        result[8], result[9], \
+        result[10], result[11], result[12], result[13], result[14], result[15], result[16], result[17], result[18], \
+        result[19], result[20], result[21], result[22], result[23], result[24], result[25], result[26], result[27], \
+        result[28], result[29], result[30], result[31], result[32], result[33], result[34], result[35], result[36], \
+        result[37], result[38], result[39], result[40], result[41], result[42], result[43], result[44], result[45], \
+        result[46], result[47], result[48], result[49], result[50], result[51], result[52], result[53], result[54], \
+        result[55], result[56], result[57]
+
+
+with gr.Blocks(css=".container.svelte-1sk0pyu.svelte-1sk0pyu {width: 300px}", title='中能智能') as demo:
+    wind_name = gr.Dropdown(label="电场名称", choices=get_all_wind_company())
+
+    types = {
+        '分钟映射': 'minute', '秒映射': 'second'
+    }
+
+    for name in types.keys():
+        with gr.Tab(label=name):
+            type = gr.Text(label="映射类型", value=types[name], visible=False)
+            wind_full_name = gr.Textbox(label="完整的电场名称")
+            merge_columns = gr.Checkbox(label="是否需合并(多个excel列合并成一个才需要选择)", value=False)
+            is_vertical_table = gr.Checkbox(label="是否是竖表", value=False)
+            vertical_read_cols = gr.Textbox(label="竖表--读取的字段", placeholder="逗号分隔")
+            vertical_index_cols = gr.Textbox(label="竖表--分组的字段", placeholder="逗号分隔,一般都是时间,机组")
+            vertical_col_key = gr.Textbox(label="竖表--数据点字段")
+            vertical_col_value = gr.Textbox(label="竖表--数据点数值")
+            resolve_col_prefix = gr.Textbox(label="处理列名",
+                                            placeholder="比如重庆海装 25_#桨距角,只需要 桨距角 可以用 column[column.find('#')+1:]")
+
+            wind_name_exec = gr.Textbox(label="风机编号代码处理",
+                                        placeholder="比如 昌平001号风机,可以配置 wind_name.replace('昌平','').replace('号风机','')")
+
+            wind_turbine_number = gr.Textbox(label="风机编号(wind_turbine_number)")
+            time_stamp = gr.Textbox(label="时间戳(time_stamp)")
+            active_power = gr.Textbox(label="有功功率(active_power)")
+            rotor_speed = gr.Textbox(label="风轮转速(rotor_speed)")
+            generator_speed = gr.Textbox(label="发电机转速(generator_speed)")
+            wind_velocity = gr.Textbox(label="风速(wind_velocity)")
+            pitch_angle_blade_1 = gr.Textbox(label="桨距角1(pitch_angle_blade_1)")
+            pitch_angle_blade_2 = gr.Textbox(label="桨距角2(pitch_angle_blade_2)")
+            pitch_angle_blade_3 = gr.Textbox(label="桨距角3(pitch_angle_blade_3)")
+            cabin_position = gr.Textbox(label="机舱位置(cabin_position)")
+            true_wind_direction = gr.Textbox(label="绝对风向(true_wind_direction)")
+            yaw_error1 = gr.Textbox(label="对风角度(yaw_error1)")
+            set_value_of_active_power = gr.Textbox(label="有功功率设定值(set_value_of_active_power)")
+            gearbox_oil_temperature = gr.Textbox(label="齿轮箱油温(gearbox_oil_temperature)")
+            generatordrive_end_bearing_temperature = gr.Textbox(
+                label="发电机驱动端轴承温度(generatordrive_end_bearing_temperature)")
+            generatornon_drive_end_bearing_temperature = gr.Textbox(
+                label="发电机非驱动端轴承温度(generatornon_drive_end_bearing_temperature)")
+            wind_turbine_status = gr.Textbox(label="风机状态1(wind_turbine_status)")
+            wind_turbine_status2 = gr.Textbox(label="风机状态2(wind_turbine_status2)")
+            cabin_temperature = gr.Textbox(label="机舱内温度(cabin_temperature)")
+            twisted_cable_angle = gr.Textbox(label="扭缆角度(twisted_cable_angle)")
+            front_back_vibration_of_the_cabin = gr.Textbox(label="机舱前后振动(front_back_vibration_of_the_cabin)")
+            side_to_side_vibration_of_the_cabin = gr.Textbox(label="机舱左右振动(side_to_side_vibration_of_the_cabin)")
+            actual_torque = gr.Textbox(label="实际力矩(actual_torque)")
+            given_torque = gr.Textbox(label="给定力矩(given_torque)")
+            clockwise_yaw_count = gr.Textbox(label="顺时针偏航次数(clockwise_yaw_count)")
+            counterclockwise_yaw_count = gr.Textbox(label="逆时针偏航次数(counterclockwise_yaw_count)")
+            unusable = gr.Textbox(label="不可利用(unusable)")
+            power_curve_available = gr.Textbox(label="功率曲线可用(power_curve_available)")
+            required_gearbox_speed = gr.Textbox(label="齿轮箱转速(required_gearbox_speed)")
+            inverter_speed_master_control = gr.Textbox(label="变频器转速(主控)(inverter_speed_master_control)")
+            outside_cabin_temperature = gr.Textbox(label="环境温度(outside_cabin_temperature)")
+            main_bearing_temperature = gr.Textbox(label="主轴承轴承温度(main_bearing_temperature)")
+            gearbox_high_speed_shaft_bearing_temperature = gr.Textbox(
+                label="齿轮箱高速轴轴承温度(gearbox_high_speed_shaft_bearing_temperature)")
+            gearboxmedium_speed_shaftbearing_temperature = gr.Textbox(
+                label="齿轮箱中速轴轴承温度(gearboxmedium_speed_shaftbearing_temperature)")
+            gearbox_low_speed_shaft_bearing_temperature = gr.Textbox(
+                label="齿轮箱低速轴轴承温度(gearbox_low_speed_shaft_bearing_temperature)")
+            generator_winding1_temperature = gr.Textbox(label="发电机绕组1温度(generator_winding1_temperature)")
+            generator_winding2_temperature = gr.Textbox(label="发电机绕组2温度(generator_winding2_temperature)")
+            generator_winding3_temperature = gr.Textbox(label="发电机绕组3温度(generator_winding3_temperature)")
+            turbulence_intensity = gr.Textbox(label="湍流强度(turbulence_intensity)")
+            param1 = gr.Textbox(label="齿轮箱油压(param1)")
+            param2 = gr.Textbox(label="预留字段2(param2)")
+            param3 = gr.Textbox(label="预留字段3(param3)")
+            param4 = gr.Textbox(label="预留字段4(param4)")
+            param5 = gr.Textbox(label="预留字段5(param5)")
+            param6 = gr.Textbox(label="预留字段6(param6)")
+            param7 = gr.Textbox(label="预留字段7(param7)")
+            param8 = gr.Textbox(label="预留字段8(param8)")
+            param9 = gr.Textbox(label="预留字段9(param9)")
+            param10 = gr.Textbox(label="预留字段10(param10)")
+
+            button = gr.Button(value="提交")
+            result = gr.Textbox(label="结果")
+
+            button.click(fn=test_click,
+                         inputs=[wind_name, wind_full_name, type, is_vertical_table, merge_columns, vertical_read_cols,
+                                 vertical_index_cols, vertical_col_key, vertical_col_value, resolve_col_prefix,
+                                 wind_name_exec, wind_turbine_number, time_stamp, active_power, rotor_speed,
+                                 generator_speed, wind_velocity, pitch_angle_blade_1, pitch_angle_blade_2,
+                                 pitch_angle_blade_3, cabin_position, true_wind_direction, yaw_error1,
+                                 set_value_of_active_power, gearbox_oil_temperature,
+                                 generatordrive_end_bearing_temperature, generatornon_drive_end_bearing_temperature,
+                                 wind_turbine_status, wind_turbine_status2, cabin_temperature, twisted_cable_angle,
+                                 front_back_vibration_of_the_cabin, side_to_side_vibration_of_the_cabin, actual_torque,
+                                 given_torque, clockwise_yaw_count, counterclockwise_yaw_count, unusable,
+                                 power_curve_available, required_gearbox_speed, inverter_speed_master_control,
+                                 outside_cabin_temperature, main_bearing_temperature,
+                                 gearbox_high_speed_shaft_bearing_temperature,
+                                 gearboxmedium_speed_shaftbearing_temperature,
+                                 gearbox_low_speed_shaft_bearing_temperature, generator_winding1_temperature,
+                                 generator_winding2_temperature, generator_winding3_temperature, turbulence_intensity,
+                                 param1, param2, param3, param4, param5, param6, param7, param8, param9, param10
+                                 ], outputs=[result])
+            wind_name.change(fill_data, inputs=[wind_name, type],
+                             outputs=[wind_full_name, is_vertical_table, merge_columns, vertical_read_cols,
+                                      vertical_index_cols, vertical_col_key, vertical_col_value, resolve_col_prefix,
+                                      wind_name_exec, wind_turbine_number, time_stamp, active_power, rotor_speed,
+                                      generator_speed, wind_velocity, pitch_angle_blade_1, pitch_angle_blade_2,
+                                      pitch_angle_blade_3, cabin_position, true_wind_direction, yaw_error1,
+                                      set_value_of_active_power, gearbox_oil_temperature,
+                                      generatordrive_end_bearing_temperature,
+                                      generatornon_drive_end_bearing_temperature,
+                                      wind_turbine_status, wind_turbine_status2, cabin_temperature, twisted_cable_angle,
+                                      front_back_vibration_of_the_cabin, side_to_side_vibration_of_the_cabin,
+                                      actual_torque,
+                                      given_torque, clockwise_yaw_count, counterclockwise_yaw_count, unusable,
+                                      power_curve_available, required_gearbox_speed, inverter_speed_master_control,
+                                      outside_cabin_temperature, main_bearing_temperature,
+                                      gearbox_high_speed_shaft_bearing_temperature,
+                                      gearboxmedium_speed_shaftbearing_temperature,
+                                      gearbox_low_speed_shaft_bearing_temperature, generator_winding1_temperature,
+                                      generator_winding2_temperature, generator_winding3_temperature,
+                                      turbulence_intensity,
+                                      param1, param2, param3, param4, param5, param6, param7, param8, param9, param10])
+
+if __name__ == "__main__":
+    demo.launch(server_name='0.0.0.0', server_port=7860, auth=('znzn', "znzn123"))

tmp_file/test_wave.py

@@ -1,10 +1,32 @@
 import os
 import sys
+import pandas as pd
+from os.path import basename, dirname
+
+from service.trans_service import get_wave_conf
 
 sys.path.insert(0, os.path.abspath(__file__).split("tmp_file")[0])
 
 from etl.wind_power.wave.WaveTrans import WaveTrans
 
 if __name__ == '__main__':
-    test = WaveTrans('振动测点', r'/home/wzl/test_data/sdk_data/sdk_data', r'/home/wzl/test_data/sdk_data')
+    # test = WaveTrans('SKF001', r'/home/wzl/test_data/sdk_data/sdk_data', r'/home/wzl/test_data/sdk_data')
+    test = WaveTrans('SKF001', r'D:\data\sdk_data\sdk_data_less', r'/home/wzl/test_data/sdk_data')
+    # D:\data\sdk_data\sdk_data_less
+
+    # wave_conf = get_wave_conf(test.field_code)
+    #
+    # base_param_exec = wave_conf['base_param_exec']
+    # map_dict = {}
+    # if base_param_exec:
+    #     base_param_exec = base_param_exec.replace('\r\n', '\n').replace('\t', '    ')
+    #     print(base_param_exec)
+    #     exec(base_param_exec)
+    #     setattr(test, 'get_data', get_data)
+    #
+    #     mesure_poins = [key for key, value in wave_conf.items() if str(key).startswith('conf_') and value]
+    #
+    #     for point in mesure_poins:
+    #         map_dict[wave_conf[point]] = point
+
     test.run()