添加乾安的处理,添加风场的额外处理

app_run.py

@@ -37,7 +37,7 @@ def run(save_db=True, run_count=1, yaml_config=None, step=0, end=999):
         )
 
     if data["transfer_type"] == "wave":
-        exec_process = WaveTrans(data["id"], data["wind_farm_code"], data["read_dir"])
+        exec_process = WaveTrans(data["id"], data["wind_farm_code"],data["wind_farm_name"], data["read_dir"])
 
     if data["transfer_type"] == "laser":
         exec_process = LaserTrans(data["id"], data["wind_farm_code"], data["read_dir"])

conf/etl_config_qianan.yaml

@@ -0,0 +1,28 @@
+plt:
+  database: energy
+  host: 127.0.0.1
+  user: root
+  password: admin123456
+  port: 4000
+
+trans:
+  database: energy_data_prod
+  host: 127.0.0.1
+  user: root
+  password: admin123456
+  port: 4000
+
+# 如果要放在原始路径,则配置这个 以下面的名称作为切割点,新建清理数据文件夹
+etl_origin_path_contain: 收资数据
+# 如果单独保存,配置这个路径
+save_path:
+
+# 日志保存路径
+log_path_dir: /data/logs/energy_data
+
+# 临时文件存放处,有些甲方公司隔得tmp太小,只好自己配置
+tmp_base_path: /data/tmp
+
+run_batch_count: 2
+
+archive_path: /data/archive

etl/common/CombineAndSaveFormalFile.py

@@ -119,7 +119,7 @@ class CombineAndSaveFormalFile:
         # 使用并行处理
         cpu_count = get_available_cpu_count_with_percent(ParallelProcessing.CPU_USAGE_PERCENT)
         cpu_count = min(cpu_count, ParallelProcessing.MAX_PROCESSES)
-        with multiprocessing.Pool(cpu_count) as pool:
+        with multiprocessing.Pool(cpu_count, maxtasksperchild=5) as pool:
             pool.starmap(self._save_combined_file, process_args)
 
     def run(self) -> List[str]:

etl/common/SaveToDb.py

@@ -36,7 +36,7 @@ class SaveToDb(object):
 
         try:
             # 创建一个进程池处理所有文件
-            with multiprocessing.Pool(max_processes) as pool:
+            with multiprocessing.Pool(max_processes,maxtasksperchild=5) as pool:
                 if self.pathsAndTable.read_type in ['minute', 'second']:
                     # 准备参数
                     params = [(self.pathsAndTable.get_table_name(), file,

etl/common/UnzipAndRemove.py

@@ -71,7 +71,7 @@ class UnzipAndRemove(object):
 
             for index, arr in enumerate(all_arrays):
                 pool_count = min(split_count, len(arr))
-                with multiprocessing.Pool(pool_count) as pool:
+                with multiprocessing.Pool(pool_count,maxtasksperchild=5) as pool:
                     pool.starmap(self.get_and_remove, [(i,) for i in arr])
                 update_trans_transfer_progress(self.pathsAndTable.id,
                                                round(5 + 15 * (index + 1) / len(all_arrays), 2),

etl/common/WaveData.py

@@ -0,0 +1,56 @@
+class WaveData(object):
+    def __init__(self):
+        # 风机编号 varchar(20)
+        self.wind_turbine_number: int | None = None
+
+        # 原始风机编号 varchar(20)
+        self.wind_turbine_name: str | None = None
+
+        # 时间 datetime
+        self.time_stamp: float | str | None = None
+
+        # 频谱单位 varchar(10)
+        self.eu_spectrum: int | None = None
+
+        # 检测类型 0:None,1:Peak,2:Peak-to-peak,3:RMS
+        self.detection_type: int | None = None
+
+        # 窗函数类型 0:Hanning,1:Uniform,2:Flat Top
+        self.window_type: int | None = None
+
+        # 频谱起始频率(Hz) float
+        self.start_frequency: int | None = None
+
+        # 频谱结束频率(Hz) float
+        self.end_frequency: int | None = None
+
+        # 采样点数(时域波形)或谱线数(频谱)
+        self.samples: int | None = None
+
+        # 转速 float
+        self.rotational_speed: float | None = None
+
+        # 采样频率
+        self.sampling_frequency: int | None = None
+
+        # 测点名称 varchar(100)
+        self.mesure_point_name: str | None = None
+
+        # 类型 -1:不存在 0:角度 1:速度 2:加速度 3:位移,默认-1
+        self.type: str | int | None = None
+
+        # 时域数据
+        self.mesure_data_time: str | None = None
+
+        # 频谱数据
+        self.mesure_data_frenquency: str | None = None
+
+        # 包络数据
+        self.mesure_data_env: str | None = None
+
+
+if __name__ == '__main__':
+
+
+    wave = WaveData()
+    print(vars(wave))

etl/wind_power/laser/LaserTrans.py

@@ -59,7 +59,7 @@ class LaserTrans():
         info(self.wind_farm_code, '获取文件总数为:', len(all_files))
         pool_count = 8 if len(all_files) > 8 else len(all_files)
 
-        with multiprocessing.Pool(pool_count) as pool:
+        with multiprocessing.Pool(pool_count,maxtasksperchild=5) as pool:
             dfs = pool.map(self.get_file_data, all_files)
 
         update_trans_transfer_progress(self.id, 80)

etl/wind_power/min_sec/FilterValidData.py

@@ -1,8 +1,11 @@
 import pandas as pd
 
+from utils.log.trans_log import debug
+
 
 class FilterValidData:
-    def __init__(self, df, rated_power):
+    def __init__(self, origin_wind_name, df, rated_power):
+        self.origin_wind_name = origin_wind_name
         self.df = df
         self.rated_power = rated_power
 
@@ -49,5 +52,8 @@ class FilterValidData:
 
             # 组合条件
             final_condition &= (min_cond & max_cond)
+            debug(self.origin_wind_name, col, '过滤规则',
+                  f'{">=" if include_min else ">"}{min_val} and {"<=" if include_max else "<"}{max_val}',
+                  self.df[final_condition].shape)
 
         return self.df[final_condition]

etl/wind_power/min_sec/ReadAndSaveTmp.py

@@ -140,9 +140,14 @@ class ReadAndSaveTmp(object):
                 df.rename(columns=real_cols_trans, inplace=True)
 
                 # 添加使用同一个excel字段的值
+                debug("相同的列:", same_col)
+                debug("数据的列:", df.columns)
+
                 for key in same_col.keys():
-                    for col in same_col[key]:
-                        df[col] = df[key]
+                    if key in df.columns:
+                        for col in same_col[key]:
+                            if col not in df.columns:
+                                df[col] = df[key]
 
                 del_keys = set(df.columns) - set(cols_tran.keys())
 
@@ -224,7 +229,7 @@ class ReadAndSaveTmp(object):
         if self.trans_param.merge_columns:
             for index, arr in enumerate(all_arrays):
                 try:
-                    with multiprocessing.Pool(split_count) as pool:
+                    with multiprocessing.Pool(split_count,maxtasksperchild=5) as pool:
                         pool.starmap(self.save_merge_data, [(ar,) for ar in arr])
 
                 except Exception as e:
@@ -245,7 +250,7 @@ class ReadAndSaveTmp(object):
                 all_arrays = split_array(dirs, split_count)
                 for index, arr in enumerate(all_arrays):
                     try:
-                        with multiprocessing.Pool(split_count) as pool:
+                        with multiprocessing.Pool(split_count,maxtasksperchild=5) as pool:
                             pool.starmap(self.merge_df, [(ar,) for ar in arr])
 
                     except Exception as e:
@@ -260,7 +265,7 @@ class ReadAndSaveTmp(object):
         else:
             for index, arr in enumerate(all_arrays):
                 try:
-                    with multiprocessing.Pool(split_count) as pool:
+                    with multiprocessing.Pool(split_count,maxtasksperchild=5) as pool:
                         dfs = pool.starmap(self.read_excel_to_df, [(ar,) for ar in arr])
                     for df in dfs:
                         self.df_save_to_tmp_file(df)
@@ -434,7 +439,7 @@ class ReadAndSaveTmp(object):
                     cols_dict[col] = 'wind_turbine_number'
 
                 df.rename(columns=cols_dict, inplace=True)
-
+            debug('保存临时文件', df.shape)
             return df
 
     def run(self):

etl/wind_power/min_sec/StatisticsAndSaveTmpFormalFile.py

@@ -10,11 +10,11 @@ from etl.common.PathsAndTable import PathsAndTable
 from etl.wind_power.min_sec import TransParam
 from etl.wind_power.min_sec.ClassIdentifier import ClassIdentifier
 from etl.wind_power.min_sec.FilterValidData import FilterValidData
-from service.trans_conf_service import update_trans_transfer_progress
+from service.trans_conf_service import update_trans_transfer_progress, get_trans_exec_code
 from utils.conf.read_conf import read_conf
 from utils.df_utils.util import estimate_time_interval as get_time_space
 from utils.file.trans_methods import create_file_path, read_excel_files, read_file_to_df
-from utils.log.trans_log import debug, error
+from utils.log.trans_log import debug, error, warning
 from utils.systeminfo.sysinfo import use_files_get_max_cpu_count
 
 exec("import math")
@@ -93,7 +93,7 @@ class StatisticsAndSaveTmpFormalFile(object):
 
         df.drop_duplicates(['wind_turbine_number', 'time_stamp'], keep='first', inplace=True)
 
-        df['time_stamp'] = df['time_stamp'].str.strip()
+        df['time_stamp'] = df['time_stamp'].astype(str).str.strip()
         df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce")
         df.dropna(subset=['time_stamp'], inplace=True)
         df.sort_values(by='time_stamp', inplace=True)
@@ -128,8 +128,8 @@ class StatisticsAndSaveTmpFormalFile(object):
         else:
             debug(origin_wind_name, '过滤数据前数据大小', df.shape)
             debug(origin_wind_name, '额定功率', rated_power_and_cutout_speed_tuple[0])
-            # trans_print(origin_wind_name, '\n', df.head(10))
-            filter_valid_data = FilterValidData(df, rated_power_and_cutout_speed_tuple[0])
+            # debug(origin_wind_name, '\n', df.head(10))
+            filter_valid_data = FilterValidData(origin_wind_name, df, rated_power_and_cutout_speed_tuple[0])
             try:
                 df = filter_valid_data.run()
             except:
@@ -138,54 +138,58 @@ class StatisticsAndSaveTmpFormalFile(object):
             debug(origin_wind_name, '过滤数据后数据大小', df.shape)
 
             # 如果有需要处理的,先进行代码处理,在进行打标签
-            # exec_code = get_trans_exec_code(self.paths_and_table.exec_id, self.paths_and_table.read_type)
-            # if exec_code:
-            #     if 'import ' in exec_code:
-            #         raise Exception("执行代码不支持导入包")
-            #     exec(exec_code)
-
-            if power_df.shape[0] == 0:
-                df.loc[:, 'lab'] = -1
-            else:
-                class_identifier = ClassIdentifier(wind_turbine_number=origin_wind_name, origin_df=df,
-                                                   rated_power=rated_power_and_cutout_speed_tuple[0],
-                                                   cut_out_speed=rated_power_and_cutout_speed_tuple[1])
-                df = class_identifier.run()
+            exec_code = get_trans_exec_code(self.paths_and_table.wind_farm_code, self.paths_and_table.read_type)
+            if exec_code:
+                if 'import ' in exec_code:
+                    raise Exception("执行代码不支持导入包")
+                exec(exec_code)
+
+            if not df.empty:
+                if power_df.shape[0] == 0:
+                    df.loc[:, 'lab'] = -1
+                else:
+                    class_identifier = ClassIdentifier(wind_turbine_number=origin_wind_name, origin_df=df,
+                                                       rated_power=rated_power_and_cutout_speed_tuple[0],
+                                                       cut_out_speed=rated_power_and_cutout_speed_tuple[1])
+                    df = class_identifier.run()
 
-            del power_df
+                del power_df
 
-            df['year'] = df['time_stamp'].dt.year
-            df['month'] = df['time_stamp'].dt.month
-            df['day'] = df['time_stamp'].dt.day
-            df['time_stamp'] = df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
+                df['year'] = df['time_stamp'].dt.year
+                df['month'] = df['time_stamp'].dt.month
+                df['day'] = df['time_stamp'].dt.day
+                df['time_stamp'] = df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
 
-            df['wind_turbine_name'] = str(origin_wind_name)
-            df['year_month'] = df[['year', 'month']].apply(lambda x: str(x['year']) + str(x['month']).zfill(2), axis=1)
-            cols = df.columns
+                df['wind_turbine_name'] = str(origin_wind_name)
+                df['year_month'] = df[['year', 'month']].apply(lambda x: str(x['year']) + str(x['month']).zfill(2),
+                                                               axis=1)
+                cols = df.columns
 
-            if self.paths_and_table.read_type == Types.SECOND:
-                type_col = 'year_month'
-            else:
-                type_col = 'year'
-
-            date_strs = df[type_col].unique().tolist()
-            for date_str in date_strs:
-                save_path = path.join(self.paths_and_table.get_tmp_formal_path(), str(date_str),
-                                      str(origin_wind_name) + '.csv')
-                create_file_path(save_path, is_file_path=True)
-                now_df = df[df[type_col] == date_str][cols]
-                if self.paths_and_table.save_zip:
-                    save_path = save_path + '.gz'
-                    now_df.to_csv(save_path, compression='gzip', index=False, encoding='utf-8')
+                if self.paths_and_table.read_type == Types.SECOND:
+                    type_col = 'year_month'
                 else:
-                    now_df.to_csv(save_path, index=False, encoding='utf-8')
-
-                del now_df
-
-            self.set_statistics_data(df)
-
-            del df
-            debug("保存" + str(wind_col_name) + "成功")
+                    type_col = 'year'
+
+                date_strs = df[type_col].unique().tolist()
+                for date_str in date_strs:
+                    save_path = path.join(self.paths_and_table.get_tmp_formal_path(), str(date_str),
+                                          str(origin_wind_name) + '.csv')
+                    create_file_path(save_path, is_file_path=True)
+                    now_df = df[df[type_col] == date_str][cols]
+                    if self.paths_and_table.save_zip:
+                        save_path = save_path + '.gz'
+                        now_df.to_csv(save_path, compression='gzip', index=False, encoding='utf-8')
+                    else:
+                        now_df.to_csv(save_path, index=False, encoding='utf-8')
+
+                    del now_df
+
+                self.set_statistics_data(df)
+
+                del df
+                debug("保存", origin_wind_name, str(wind_col_name), "成功")
+            else:
+                warning(origin_wind_name, str(wind_col_name), '无数据')
 
     def multiprocessing_to_save_file(self):
         # 开始保存到正式文件
@@ -201,7 +205,7 @@ class StatisticsAndSaveTmpFormalFile(object):
 
         try:
             # 创建一个进程池处理所有文件
-            with multiprocessing.Pool(max_processes) as pool:
+            with multiprocessing.Pool(max_processes,maxtasksperchild=5) as pool:
                 # 分批次处理并更新进度
                 batch_size = max(1, len(all_tmp_files) // ParallelProcessing.MAX_BATCHES)  # 最多10个批次
 

etl/wind_power/wave/WaveTrans.py

@@ -1,168 +1,195 @@
-import json
-import multiprocessing
-import traceback
-from typing import Tuple
-
-from conf.constants import ParallelProcessing, Types
-from service.plt_service import get_all_wind
-from service.trans_conf_service import update_trans_status_running, update_trans_transfer_progress, \
-    update_trans_status_success, update_trans_status_error
-from service.trans_service import get_wave_conf, save_df_to_db, get_or_create_wave_table, \
-    get_wave_data, delete_exist_wave_data
-from utils.file.trans_methods import *
-from utils.log.trans_log import set_trance_id, info, error
-from utils.systeminfo.sysinfo import get_available_cpu_count_with_percent
-
-exec("from os.path import *")
-exec("import re")
-
-
-class WaveTrans(object):
-    """波形数据转换类"""
-
-    def __init__(self, id: int, wind_farm_code: str, read_dir: str):
-        """
-        初始化波形数据转换类
-        
-        Args:
-            id: 任务ID
-            wind_farm_code: 风电场编码
-            read_dir: 读取目录
-        """
-        self.id = id
-        self.wind_farm_code = wind_farm_code
-        self.read_dir = read_dir
-        self.begin = datetime.datetime.now()
-
-        self.engine_count = 0
-        self.min_date = None
-        self.max_date = None
-        self.data_count = 0
-
-    def get_data_exec(self, func_code: str, filepath: str, measupoint_names: List[str]) -> Optional[Tuple]:
-        """
-        执行数据获取函数
-        
-        Args:
-            func_code: 函数代码
-            filepath: 文件路径
-            measupoint_names: 测量点名称列表
-        
-        Returns:
-            数据元组
-        """
-        exec(func_code)
-        return locals()['get_data'](filepath, measupoint_names)
-
-    def del_exists_data(self, df: pd.DataFrame):
-        """
-        删除已存在的数据
-        
-        Args:
-            df: 数据帧
-        """
-        min_date, max_date = df['time_stamp'].min(), df['time_stamp'].max()
-        db_df = get_wave_data(self.wind_farm_code + '_wave', min_date, max_date)
-
-        exists_df = pd.merge(db_df, df,
-                             on=['wind_turbine_name', 'time_stamp', 'sampling_frequency', 'mesure_point_name'],
-                             how='inner')
-        ids = [int(i) for i in exists_df['id'].to_list()]
-        if ids:
-            delete_exist_wave_data(self.wind_farm_code + "_wave", ids)
-
-    def run(self):
-        """运行波形数据转换"""
-        update_trans_status_running(self.id)
-        trance_id = '-'.join([self.wind_farm_code, 'wave'])
-        set_trance_id(trance_id)
-        all_files = read_files(self.read_dir, ['txt', 'csv'])
-        update_trans_transfer_progress(self.id, 5)
-
-        # 最大取系统cpu的 1/2
-        split_count = get_available_cpu_count_with_percent(1 / 2)
-        # 限制最大进程数
-        split_count = min(split_count, ParallelProcessing.MAX_PROCESSES)
-
-        all_wind, _ = get_all_wind(self.wind_farm_code, False)
-
-        get_or_create_wave_table(self.wind_farm_code + '_wave')
-
-        wave_conf = get_wave_conf(self.wind_farm_code)
-
-        base_param_exec = wave_conf.get('base_param_exec', '')
-        map_dict = {}
-        if base_param_exec:
-            base_param_exec = base_param_exec.replace('\r\n', '\n').replace('\t', '    ')
-            info(base_param_exec)
-            if 'import ' in base_param_exec:
-                raise Exception("方法不支持import方法")
-
-        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].strip()] = point.replace('conf_', '')
-
-        wind_turbine_name_set = set()
-
-        # 优化批次大小
-        batch_size = split_count * 10
-        all_array = split_array(all_files, batch_size)
-        total_index = len(all_array)
-
-        for index, now_array in enumerate(all_array):
-            index_begin = datetime.datetime.now()
-            with multiprocessing.Pool(split_count) as pool:
-                try:
-                    file_datas = pool.starmap(self.get_data_exec,
-                                              [(base_param_exec, i, list(map_dict.keys())) for i in now_array])
-                    info(f'总数:{len(now_array)},返回个数{len(file_datas)}')
-                except Exception as e:
-                    message = str(e)
-                    error(traceback.format_exc())
-                    update_trans_status_error(self.id, message[0:len(message) if len(message) < 100 else 100])
-                    raise e
-
-            update_trans_transfer_progress(self.id, 20 + int(index / total_index * 60))
-            info("读取文件耗时:", datetime.datetime.now() - self.begin)
-
-            result_list = list()
-            for file_data in file_datas:
-                if file_data:
-                    wind_turbine_name, time_stamp, sampling_frequency, rotational_speed, mesure_point_name, type, mesure_data = \
-                        file_data[0], file_data[1], file_data[2], file_data[3], file_data[4], file_data[5], file_data[6]
-
-                    if mesure_point_name in map_dict:
-                        wind_turbine_name_set.add(wind_turbine_name)
-                        if self.min_date is None or self.min_date > time_stamp:
-                            self.min_date = time_stamp
-                        if self.max_date is None or self.max_date < time_stamp:
-                            self.max_date = time_stamp
-
-                        result_list.append(
-                            [wind_turbine_name, time_stamp, rotational_speed, sampling_frequency, mesure_point_name,
-                             type,
-                             mesure_data])
-
-            if result_list:
-                self.data_count += len(result_list)
-                df = pd.DataFrame(result_list,
-                                  columns=['wind_turbine_name', 'time_stamp', 'rotational_speed', 'sampling_frequency',
-                                           'mesure_point_name', 'type', 'mesure_data'])
-                df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors='coerce')
-                df['mesure_point_name'] = df['mesure_point_name'].map(map_dict)
-                df.dropna(subset=['mesure_point_name'], inplace=True)
-                df['wind_turbine_number'] = df['wind_turbine_name'].map(all_wind).fillna(df['wind_turbine_name'])
-
-                # 批量处理JSON序列化
-                df['mesure_data'] = df['mesure_data'].apply(lambda x: json.dumps(x))
-
-                df.sort_values(by=['time_stamp', 'mesure_point_name'], inplace=True)
-                # self.del_exists_data(df)
-                save_df_to_db(self.wind_farm_code + '_wave', df, batch_count=400)
-            info(f"总共{total_index}组,当前{index + 1}", "本次写入耗时:", datetime.datetime.now() - index_begin,
-                 "总耗时:", datetime.datetime.now() - self.begin)
-
-        update_trans_status_success(self.id, len(wind_turbine_name_set), Types.WAVE,
-                                    self.min_date, self.max_date, self.data_count)
-
-        info("总耗时:", datetime.datetime.now() - self.begin)
+import json
+import multiprocessing
+import traceback
+
+from conf.constants import ParallelProcessing
+from etl.common.WaveData import WaveData
+from service.plt_service import get_all_wind
+from service.trans_conf_service import update_trans_status_running, update_trans_transfer_progress, \
+    update_trans_status_success, update_trans_status_error
+from service.trans_service import get_wave_conf, save_df_to_db, get_or_create_wave_table, \
+    get_wave_data, delete_exist_wave_data
+from utils.file.trans_methods import *
+from utils.log.trans_log import set_trance_id, info, error
+from utils.systeminfo.sysinfo import get_available_cpu_count_with_percent
+
+# env = "qianan"
+# if len(sys.argv) >= 2:
+#     env = sys.argv[1]
+#
+# if env.endswith(".yaml"):
+#     conf_path = env
+# else:
+#     conf_path = os.path.abspath(f"C:/project/energy-data-trans/conf/etl_config_{env}.yaml")
+#
+# os.environ["ETL_CONF"] = conf_path
+# yaml_config = yaml_conf(conf_path)
+# os.environ["env"] = env
+exec("import datetime")
+exec("from os.path import *")
+exec("import re")
+
+
+class WaveTrans(object):
+    """波形数据转换类"""
+
+    def __init__(self, id: int, wind_farm_code: str, wind_farm_name: str, read_dir: str):
+        """
+        初始化波形数据转换类
+        
+        Args:
+            id: 任务ID
+            wind_farm_code: 风电场编码
+            read_dir: 读取目录
+        """
+        self.id = id
+        self.wind_farm_code = wind_farm_code
+        self.wind_farm_name = wind_farm_name
+        self.read_dir = read_dir
+        self.begin = datetime.datetime.now()
+
+        self.engine_count = 0
+        self.min_date = None
+        self.max_date = None
+        self.data_count = 0
+
+    def get_data_exec(self, func_code: str, filepath: str, measupoint_names: List[str]) -> List[WaveData]:
+        exec(func_code)
+        return locals()['get_data'](filepath, measupoint_names) or []
+        # return self.get_data(filepath, measupoint_names)
+
+    def del_exists_data(self, df: pd.DataFrame):
+        """
+        删除已存在的数据
+        
+        Args:
+            df: 数据帧
+        """
+        min_date, max_date = df['time_stamp'].min(), df['time_stamp'].max()
+
+        if self.min_date is None:
+            self.min_date = min_date
+        if self.max_date is None:
+            self.max_date = max_date
+
+        self.min_date = min(self.min_date, min_date)
+        self.max_date = max(self.max_date, max_date)
+
+        db_df = get_wave_data(self.wind_farm_code + '_wave', min_date, max_date)
+        db_df['type'] = db_df['type'].astype(str)
+        df['type'] = df['type'].astype(str)
+        db_df['time_stamp'] = pd.to_datetime(db_df['time_stamp'], errors='coerce')
+        df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors='coerce')
+
+        exists_df = pd.merge(db_df, df,
+                             on=['detection_type', 'end_frequency', 'eu_spectrum', 'mesure_point_name',
+                                 'sampling_frequency', 'samples', 'start_frequency', 'time_stamp', 'type',
+                                 'wind_turbine_name', 'window_type'],
+                             how='inner')
+        ids = [int(i) for i in exists_df['id'].to_list()]
+        if ids:
+            delete_exist_wave_data(self.wind_farm_code + "_wave", ids)
+
+    def run(self):
+        """运行波形数据转换"""
+        update_trans_status_running(self.id)
+        trance_id = '-'.join([self.wind_farm_code, 'wave'])
+        set_trance_id(trance_id)
+
+        wave_conf = get_wave_conf(self.wind_farm_code)
+
+        filter_types = wave_conf.get("filter_types", "txt,csv")
+        filter_types = filter_types.replace("，", ",")
+
+        all_files = read_files(self.read_dir, [str(i).strip() for i in filter_types.split(",")])
+
+        wind_turbine_name_set = set()
+        if len(all_files) > 0:
+            update_trans_transfer_progress(self.id, 5)
+
+            # 最大取系统cpu的 1/2
+            split_count = get_available_cpu_count_with_percent(1 / 2)
+            # 限制最大进程数
+            split_count = min(split_count, ParallelProcessing.MAX_PROCESSES)
+
+            all_wind, _ = get_all_wind(self.wind_farm_code, False)
+            # all_wind = {}
+
+            get_or_create_wave_table(self.wind_farm_code + '_wave', self.wind_farm_name)
+
+            base_param_exec = wave_conf.get('base_param_exec', '')
+            map_dict = {}
+            if base_param_exec:
+                base_param_exec = base_param_exec.replace('\r\n', '\n').replace('\t', '    ')
+                info(base_param_exec)
+                if 'import ' in base_param_exec:
+                    raise Exception("方法不支持import方法")
+
+            mesure_poins = [key for key, value in wave_conf.items() if str(key).startswith('conf_') and value]
+            for point in mesure_poins:
+                point_names = wave_conf[point].strip().split('|')
+                for name in point_names:
+                    map_dict[name] = point.replace('conf_', '')
+
+            # 优化批次大小
+            batch_size = split_count * 10
+            all_array = split_array(all_files, batch_size)
+            total_index = len(all_array)
+
+            for index, now_array in enumerate(all_array):
+                index_begin = datetime.datetime.now()
+                with multiprocessing.Pool(split_count,maxtasksperchild=5) as pool:
+                    try:
+                        file_datas_result = pool.starmap(self.get_data_exec,
+                                                         [(base_param_exec, i, list(map_dict.keys())) for i in
+                                                          now_array])
+                        file_datas = [x for sub in file_datas_result if sub for x in sub if x]
+                        info(f'总数:{len(now_array)},返回个数{len(file_datas)}')
+                    except Exception as e:
+                        message = str(e)
+                        error(traceback.format_exc())
+                        update_trans_status_error(self.id, message[0:len(message) if len(message) < 100 else 100])
+                        raise e
+
+                update_trans_transfer_progress(self.id, 20 + int(index / total_index * 60))
+                info("读取文件耗时:", datetime.datetime.now() - self.begin)
+
+                result_list = [vars(i) for i in file_datas if i]
+
+                if result_list:
+                    self.data_count += len(result_list)
+                    df = pd.DataFrame(result_list)
+                    df['time_stamp'] = df['time_stamp'].apply(lambda x: x.split('.')[0])
+                    # df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors='coerce')
+                    # df['time_stamp'] = df['time_stamp'].dt.strftime('%Y-%m-%d %H:%M:%S')
+                    df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors='coerce')
+                    df['mesure_point_name'] = df['mesure_point_name'].map(map_dict)
+                    df.dropna(subset=['mesure_point_name'], inplace=True)
+                    df['wind_turbine_number'] = df['wind_turbine_name'].map(all_wind).fillna(df['wind_turbine_name'])
+
+                    # 批量处理JSON序列化
+                    df['mesure_data_time'] = df['mesure_data_time'].apply(lambda x: json.dumps(x))
+                    df['mesure_data_frenquency'] = df['mesure_data_frenquency'].apply(lambda x: json.dumps(x))
+                    df['mesure_data_env'] = df['mesure_data_env'].apply(lambda x: json.dumps(x))
+
+                    df.sort_values(by=['time_stamp', 'mesure_point_name'], inplace=True)
+
+                    for col in df['wind_turbine_name'].unique():
+                        wind_turbine_name_set.add(col)
+
+                    self.del_exists_data(df)
+
+                    save_df_to_db(self.wind_farm_code + '_wave', df, batch_count=40)
+                info(f"总共{total_index}组,当前{index + 1}", "本次写入耗时:", datetime.datetime.now() - index_begin,
+                     "总耗时:", datetime.datetime.now() - self.begin)
+
+        update_trans_status_success(self.id, len(wind_turbine_name_set), None,
+                                    self.min_date, self.max_date, self.data_count)
+
+        info("总耗时:", datetime.datetime.now() - self.begin)
+
+
+# if __name__ == '__main__':
+#     trans = WaveTrans(1, 'WOF043800107', '乾安风电场', r'C:\迅雷云盘\04-01\A28')
+#
+#     trans.run()

etl/wind_power/wave/WaveTrans_1.py

@@ -0,0 +1,180 @@
+import json
+import multiprocessing
+import traceback
+
+from conf.constants import ParallelProcessing
+from etl.common.WaveData import WaveData
+from service.plt_service import get_all_wind
+from service.trans_conf_service import update_trans_status_running, update_trans_transfer_progress, \
+    update_trans_status_success, update_trans_status_error
+from service.trans_service import get_wave_conf, save_df_to_db, get_or_create_wave_table, \
+    get_wave_data, delete_exist_wave_data
+from utils.file.trans_methods import *
+from utils.log.trans_log import set_trance_id, info, error
+from utils.systeminfo.sysinfo import get_available_cpu_count_with_percent
+
+exec("from os.path import *")
+exec("import re")
+
+
+class WaveTrans(object):
+    """波形数据转换类"""
+
+    def __init__(self, id: int, wind_farm_code: str, wind_farm_name: str, read_dir: str):
+        """
+        初始化波形数据转换类
+        
+        Args:
+            id: 任务ID
+            wind_farm_code: 风电场编码
+            read_dir: 读取目录
+        """
+        self.id = id
+        self.wind_farm_code = wind_farm_code
+        self.wind_farm_name = wind_farm_name
+        self.read_dir = read_dir
+        self.begin = datetime.datetime.now()
+
+        self.engine_count = 0
+        self.min_date = None
+        self.max_date = None
+        self.data_count = 0
+
+    def get_data_exec(self, func_code: str, filepath: str, measupoint_names: List[str]) -> Optional[WaveData]:
+        """
+        执行数据获取函数
+
+        Args:
+            func_code: 函数代码
+            filepath: 文件路径
+            measupoint_names: 测量点名称列表
+
+        Returns:
+            WaveData: 波形数据对象 / None
+        """
+        exec(func_code)
+        return locals()['get_data'](filepath, measupoint_names)
+
+    def del_exists_data(self, df: pd.DataFrame):
+        """
+        删除已存在的数据
+        
+        Args:
+            df: 数据帧
+        """
+        min_date, max_date = df['time_stamp'].min(), df['time_stamp'].max()
+        db_df = get_wave_data(self.wind_farm_code + '_wave', min_date, max_date)
+
+        exists_df = pd.merge(db_df, df,
+                             on=['wind_turbine_name', 'time_stamp', 'sampling_frequency', 'mesure_point_name'],
+                             how='inner')
+        ids = [int(i) for i in exists_df['id'].to_list()]
+        if ids:
+            delete_exist_wave_data(self.wind_farm_code + "_wave", ids)
+
+    def run(self):
+        """运行波形数据转换"""
+        update_trans_status_running(self.id)
+        trance_id = '-'.join([self.wind_farm_code, 'wave'])
+        set_trance_id(trance_id)
+
+        wave_conf = get_wave_conf(self.wind_farm_code)
+
+        filter_types = wave_conf.get("filter_types", "txt,csv")
+        filter_types = filter_types.replace("，", ",")
+
+        all_files = read_files(self.read_dir, [str(i).strip() for i in filter_types.split(",")])
+
+        wind_turbine_name_set = set()
+        if len(all_files) > 0:
+            update_trans_transfer_progress(self.id, 5)
+
+            # 最大取系统cpu的 1/2
+            split_count = get_available_cpu_count_with_percent(1 / 2)
+            # 限制最大进程数
+            split_count = min(split_count, ParallelProcessing.MAX_PROCESSES)
+
+            all_wind, _ = get_all_wind(self.wind_farm_code, False)
+
+            get_or_create_wave_table(self.wind_farm_code + '_wave', self.wind_farm_name)
+
+            base_param_exec = wave_conf.get('base_param_exec', '')
+            map_dict = {}
+            if base_param_exec:
+                base_param_exec = base_param_exec.replace('\r\n', '\n').replace('\t', '    ')
+                info(base_param_exec)
+                if 'import ' in base_param_exec:
+                    raise Exception("方法不支持import方法")
+
+            mesure_poins = [key for key, value in wave_conf.items() if str(key).startswith('conf_') and value]
+            for point in mesure_poins:
+                point_names = wave_conf[point].strip().split('|')
+                for name in point_names:
+                    map_dict[name] = point.replace('conf_', '')
+
+            # 优化批次大小
+            batch_size = split_count * 10
+            all_array = split_array(all_files, batch_size)
+            total_index = len(all_array)
+
+            for index, now_array in enumerate(all_array):
+                index_begin = datetime.datetime.now()
+                with multiprocessing.Pool(split_count,maxtasksperchild=5) as pool:
+                    try:
+                        file_datas = pool.starmap(self.get_data_exec,
+                                                  [(base_param_exec, i, list(map_dict.keys())) for i in now_array])
+                        info(f'总数:{len(now_array)},返回个数{len(file_datas)}')
+                    except Exception as e:
+                        message = str(e)
+                        error(traceback.format_exc())
+                        update_trans_status_error(self.id, message[0:len(message) if len(message) < 100 else 100])
+                        raise e
+
+                update_trans_transfer_progress(self.id, 20 + int(index / total_index * 60))
+                info("读取文件耗时:", datetime.datetime.now() - self.begin)
+
+                result_list = list()
+                for file_data in file_datas:
+                    if file_data:
+                        wind_turbine_name, time_stamp, sampling_frequency, rotational_speed, mesure_point_name, type, mesure_data = \
+                            file_data[0], file_data[1], file_data[2], file_data[3], file_data[4], file_data[5], \
+                                file_data[6]
+
+                        if mesure_point_name in map_dict:
+                            wind_turbine_name_set.add(wind_turbine_name)
+                            if self.min_date is None or self.min_date > time_stamp:
+                                self.min_date = time_stamp
+                            if self.max_date is None or self.max_date < time_stamp:
+                                self.max_date = time_stamp
+
+                            result_list.append(
+                                [wind_turbine_name, time_stamp, rotational_speed, sampling_frequency, mesure_point_name,
+                                 type,
+                                 mesure_data])
+
+                if result_list:
+                    self.data_count += len(result_list)
+                    df = pd.DataFrame(result_list,
+                                      columns=['wind_turbine_name', 'time_stamp', 'rotational_speed',
+                                               'sampling_frequency',
+                                               'mesure_point_name', 'type', 'mesure_data'])
+                    df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors='coerce')
+                    df['mesure_point_name'] = df['mesure_point_name'].map(map_dict)
+                    df.dropna(subset=['mesure_point_name'], inplace=True)
+                    df['wind_turbine_number'] = df['wind_turbine_name'].map(all_wind).fillna(df['wind_turbine_name'])
+
+                    # 批量处理JSON序列化
+                    df['mesure_data'] = df['mesure_data'].apply(lambda x: json.dumps(x))
+
+                    df.sort_values(by=['time_stamp', 'mesure_point_name'], inplace=True)
+
+                    self.del_exists_data(df)
+
+                    save_df_to_db(self.wind_farm_code + '_wave', df, batch_count=400)
+                info(f"总共{total_index}组,当前{index + 1}", "本次写入耗时:", datetime.datetime.now() - index_begin,
+                     "总耗时:", datetime.datetime.now() - self.begin)
+
+        update_trans_status_success(self.id, len(wind_turbine_name_set), None,
+                                    self.min_date, self.max_date, self.data_count)
+
+        info("总耗时:", datetime.datetime.now() - self.begin)

qianan_wave.py

@@ -0,0 +1,122 @@
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import pymysql
+import redis
+
+# -------------------------- 日志配置 --------------------------
+# 创建日志目录（如果不存在）
+log_dir = Path("/data/logs/wave_data")
+log_dir.mkdir(exist_ok=True)
+
+# 配置日志格式和输出
+logging.basicConfig(
+    level=logging.INFO,  # 日志级别：DEBUG/INFO/WARNING/ERROR/CRITICAL
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",  # 日志格式
+    handlers=[
+        # 输出到文件（按日期命名，编码UTF-8避免中文乱码）
+        logging.FileHandler(log_dir / f"{datetime.now().strftime('%Y%m%d')}.log", encoding="utf-8"),
+        # 同时输出到控制台
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger("WaveData")  # 日志器名称，便于区分不同模块
+
+# -------------------------- 配置信息 --------------------------
+# Redis 配置
+REDIS_CONFIG = {
+    'host': 'localhost',
+    'port': 6379,
+    'db': 10,
+    'password': '123456',
+    'decode_responses': True  # 自动将返回值解码为字符串，避免字节串处理
+}
+
+# MySQL 配置
+MYSQL_CONFIG = {
+    'host': '127.0.0.1',
+    'port': 4000,
+    'user': 'root',
+    'password': 'admin123456',
+    'database': 'energy_data_prod',
+    'charset': 'utf8mb4'
+}
+
+now = datetime.now()
+last_hour = now - timedelta(hours=1)
+FILE_PATH = f"/data/skf-data/National Grid/乾安风电场/{last_hour.year}/{str(last_hour.month).zfill(2)}{str(last_hour.day).zfill(2)}/{str(last_hour.hour).zfill(2)}"
+
+
+# -------------------------- 核心逻辑 --------------------------
+def main():
+    target_key = f"energy:skf:shake:{last_hour.strftime('%Y%m%d%H')}"
+    logger.info(f"开始处理任务，目标 Redis Key：{target_key}")
+
+    redis_conn = None
+    mysql_conn = None
+    mysql_cursor = None
+
+    try:
+        # 2. 连接 Redis 并检查 Key 是否存在
+        redis_conn = redis.Redis(**REDIS_CONFIG)
+        # 先判断 Key 是否存在
+        if not redis_conn.exists(target_key):
+            logger.warning(f"Redis Key {target_key} 不存在，跳过后续操作")
+            return  # 直接退出，不执行后续逻辑
+
+        # 3. 获取 Key 的值并判断是否等于 '1'
+        value = redis_conn.get(target_key)
+        logger.info(f"Redis Key {target_key} 的值为：{value}")
+
+        if value == '1':
+            # 3.1 删除 Redis 中的目标 Key
+            redis_conn.delete(target_key)
+            logger.info(f"已删除 Redis Key：{target_key}")
+
+            # 3.2 连接 MySQL 并插入数据
+            mysql_conn = pymysql.connect(**MYSQL_CONFIG)
+            mysql_cursor = mysql_conn.cursor()
+
+            # 定义插入语句（参数化查询，避免 SQL 注入）
+            insert_sql = """
+            INSERT INTO `data_transfer` (
+                `task_name`, `wind_farm_code`, `wind_farm_name`, 
+                `transfer_type`, `read_dir`, `data_collector`, 
+                `transfer_status`, `trans_sys_status`
+            ) VALUES ( %s, 'WOF043800107', '乾安一期', 'wave', %s, '系统',  -1, -1);
+            """
+            # 构造插入参数
+            task_name = f"自动-{last_hour.strftime('%Y-%m-%d %H')}"
+            read_dir = FILE_PATH
+            # 执行插入
+            mysql_cursor.execute(insert_sql, (task_name, read_dir))
+            mysql_conn.commit()
+            logger.info(f"成功向 MySQL 插入记录：任务名称={task_name}，读取目录={read_dir}")
+        else:
+            logger.warning(f"Redis Key {target_key} 的值不是 '1'（当前值：{value}），跳过插入操作")
+
+    except redis.RedisError as e:
+        logger.error(f"Redis 操作出错：{str(e)}", exc_info=True)  # exc_info=True 打印详细异常栈
+    except pymysql.MySQLError as e:
+        logger.error(f"MySQL 操作出错：{str(e)}", exc_info=True)
+        # 插入失败时回滚 MySQL 事务
+        if mysql_conn:
+            mysql_conn.rollback()
+    except Exception as e:
+        logger.error(f"未知错误：{str(e)}", exc_info=True)
+    finally:
+        # 确保所有连接关闭
+        # 关闭 MySQL 游标和连接
+        if mysql_cursor:
+            mysql_cursor.close()
+        if mysql_conn:
+            mysql_conn.close()
+        # 关闭 Redis 连接
+        if redis_conn:
+            redis_conn.close()
+        logger.info("任务处理完成，所有连接已关闭")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -1,10 +1,11 @@
-pandas~=2.2.2
-numpy~=2.0.0
-PyMySQL~=1.1.0
-SQLAlchemy~=2.0.30
-rarfile~=4.2
-PyYAML~=6.0.1
-chardet~=3.0.4
-psutil~=6.0.0
-openpyxl ~= 3.1.4
-xlrd ~=2.0.1
+pandas
+numpy
+PyMySQL
+SQLAlchemy
+rarfile
+PyYAML
+chardet
+psutil
+openpyxl
+xlrd
+redis

service/trans_conf_service.py

@@ -113,23 +113,68 @@ def get_batch_exec_data() -> dict:
     return data[0]
 
 
-def create_wave_table(table_name, save_db=True):
+def create_wave_table(table_name, wind_farm_name: str = '', save_db=True):
+    # if save_db:
+    #     exec_sql = f"""
+    #     CREATE TABLE `{table_name}` (
+    #       `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
+    #       `wind_turbine_number` varchar(20) DEFAULT NULL COMMENT '风机编号',
+    #       `wind_turbine_name` varchar(20) DEFAULT NULL COMMENT '原始风机编号',
+    #       `time_stamp` datetime DEFAULT NULL COMMENT '时间',
+    #       `rotational_speed` float DEFAULT NULL COMMENT '转速',
+    #       `sampling_frequency` int DEFAULT NULL COMMENT '采样频率',
+    #       `mesure_point_name` varchar(100) DEFAULT NULL COMMENT '测点名称',
+    #       `origin_type`  int(11) DEFAULT '-1' COMMENT '厂商原始的类型,默认 -1',
+    #       `type` int(11) DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移 4:包络,默认 -1',
+    #       `mesure_data` longtext COMMENT '测点数据',
+    #       PRIMARY KEY (`id`),
+    #       KEY `wind_turbine_number` (`wind_turbine_number`),
+    #       KEY `time_stamp` (`time_stamp`),
+    #       KEY `mesure_point_name` (`mesure_point_name`)
+    #     ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4  COMMENT='{wind_farm_name}'
+    #     """
+    #     trans.execute(exec_sql)
+
     if save_db:
         exec_sql = f"""
-        CREATE TABLE `{table_name}` (
-          `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
-          `wind_turbine_number` varchar(20) DEFAULT NULL COMMENT '风机编号',
-          `wind_turbine_name` varchar(20) DEFAULT NULL COMMENT '原始风机编号',
-          `time_stamp` datetime DEFAULT NULL COMMENT '时间',
-          `rotational_speed` float DEFAULT NULL COMMENT '转速',
-          `sampling_frequency` varchar(50) DEFAULT NULL COMMENT '采样频率',
-          `mesure_point_name` varchar(100) DEFAULT NULL COMMENT '测点名称',
-          `type` int(11) DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移,默认 -1',
-          `mesure_data` longtext COMMENT '测点数据',
-          PRIMARY KEY (`id`),
-          KEY `wind_turbine_number` (`wind_turbine_number`),
-          KEY `time_stamp` (`time_stamp`),
-          KEY `mesure_point_name` (`mesure_point_name`)
-        ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4
-        """
+            CREATE TABLE `{table_name}` (
+              `id` int NOT NULL AUTO_INCREMENT COMMENT '主键',
+              `wind_turbine_number` varchar(20) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '风机编号',
+              `wind_turbine_name` varchar(20) CHARACTER SET utf8mb4  DEFAULT NULL COMMENT '原始风机编号',
+              `time_stamp` datetime DEFAULT NULL COMMENT '时间',
+              `eu_spectrum` varchar(10) CHARACTER SET utf8mb4  DEFAULT NULL COMMENT '频谱单位',
+              `detection_type` int DEFAULT NULL COMMENT '频谱的检测类型 0:None,1:Peak,2:Peak-to-peak,3:RMS',
+              `window_type` int DEFAULT NULL COMMENT '窗函数类型:0:Hanning,1:Uniform,2:Flat Top',
+              `start_frequency` int DEFAULT NULL COMMENT '频谱的起始频率(Hz)',
+              `end_frequency` int DEFAULT NULL COMMENT '频谱的结束频率(Hz)',
+              `samples` int DEFAULT NULL COMMENT '采样点数(时域波形)或谱线数(频谱)',
+              `rotational_speed` float DEFAULT NULL COMMENT '转速',
+              `sampling_frequency` int DEFAULT NULL COMMENT '采样频率',
+              `mesure_point_name` varchar(100) CHARACTER SET utf8mb4  DEFAULT NULL COMMENT '测点名称',
+              `type` int DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移 4:包络,默认 -1',
+              `mesure_data_time` longtext CHARACTER SET utf8mb4  COMMENT '时域数据',
+              `mesure_data_frenquency` longtext CHARACTER SET utf8mb4  COMMENT '频谱数据',
+              `mesure_data_env` longtext CHARACTER SET utf8mb4  COMMENT '包络数据',
+              `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+              `status_1` int DEFAULT NULL COMMENT '轴承诊断 0:正常 1:预警 2:危险',
+              `status_2` int DEFAULT NULL COMMENT '齿轮诊断 0:正常 1:预警 2:危险',
+              `status_3` int DEFAULT NULL COMMENT '不对中诊断 0:正常 1:预警 2:危险',
+              `status_4` int DEFAULT NULL COMMENT '不平衡诊断 0:正常 1:预警 2:危险',
+              `status_5` int DEFAULT NULL COMMENT '松动诊断 0:正常 1:预警 2:危险',
+              `diag_flag` int DEFAULT NULL COMMENT '诊断标识 0：未诊断 1：已诊断',
+              `retry_times` int(11) DEFAULT '0' COMMENT '尝试计算次数',
+              PRIMARY KEY (`id`) USING BTREE,
+              KEY `wind_turbine_number` (`wind_turbine_number`) USING BTREE,
+              KEY `time_stamp` (`time_stamp`) USING BTREE,
+              KEY `mesure_point_name` (`mesure_point_name`) USING BTREE
+            ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4  COMMENT='{wind_farm_name}'
+            """
         trans.execute(exec_sql)
+
+
+def get_trans_exec_code(wind_code, trans_type):
+    exec_sql = f"select exec_code from  batch_exec_code where wind_code = '{wind_code}' and type = '{trans_type}' and status = 1"
+    data = trans.execute(exec_sql)
+    if type(data) == tuple:
+        return None
+    return data[0]['exec_code']

service/trans_service.py

@@ -309,7 +309,7 @@ def drop_table(table_name):
         pass
 
 
-def get_or_create_wave_table(table_name):
+def get_or_create_wave_table(table_name, wind_farm_name: str = ''):
     create_table = False
     query_sql = f"select 1 from `{table_name}` limit 1"
     try:
@@ -318,12 +318,17 @@ def get_or_create_wave_table(table_name):
         create_table = True
 
     if create_table:
-        create_wave_table(table_name)
+        create_wave_table(table_name,wind_farm_name)
 
 
 def get_wave_data(table_name, min_data, max_data):
     query_sql = f"""
-    select  id,wind_turbine_number,wind_turbine_name,time_stamp,sampling_frequency,mesure_point_name from `{table_name}` where time_stamp >= '{min_data}' and time_stamp <= '{max_data}'
+    select  
+    detection_type,end_frequency,eu_spectrum,id,
+    mesure_point_name,rotational_speed,sampling_frequency,
+    samples,start_frequency,time_stamp,type,wind_turbine_name,
+    wind_turbine_number,window_type
+    from `{table_name}` where time_stamp >= '{min_data}' and time_stamp <= '{max_data}'
     """
     return trans.read_sql_to_df(query_sql)
 

utils/file/trans_methods.py

@@ -82,17 +82,16 @@ def split_array(array: List, num: int) -> List[List]:
 def find_read_header(file_path: str, trans_cols: List[str], resolve_col_prefix: Optional[str] = None) -> Optional[int]:
     """
     查找文件的表头行
-    
+
     Args:
         file_path: 文件路径
         trans_cols: 要匹配的列名列表
         resolve_col_prefix: 列名前缀解析表达式
-    
+
     Returns:
         表头行索引
     """
     df = read_file_to_df(file_path, nrows=20)
-    df.reset_index(inplace=True)
     count = 0
     header = None
     df_cols = df.columns
@@ -103,22 +102,33 @@ def find_read_header(file_path: str, trans_cols: List[str], resolve_col_prefix:
     for col in trans_cols:
         if col in df_cols:
             count += 1
+            debug("第一行,找到相同列:", col, count)
             if count >= 2:
                 header = 0
-                break
+                debug("第一行,返回Header", header)
+                return header
 
     count = 0
-
+    index_value = 0
     for index, row in df.iterrows():
+        index_value = index_value + 1
+        all_cols = row.to_list()
+        name_part = list(row.name) if isinstance(row.name, str) else [str(row.name)]
+        debug(os.path.basename(file_path), "索引列:", name_part)
+        all_cols.extend(name_part)
+
         if resolve_col_prefix:
-            values = [eval(resolve_col_prefix) for column in row.values]
+            values = [eval(resolve_col_prefix) for column in all_cols]
         else:
-            values = row.values
+            values = all_cols
+        debug(os.path.basename(file_path), "查找到列名:", values)
         for col in trans_cols:
             if col in values:
                 count += 1
-                if count > 2:
-                    header = index + 1
+                debug("非第一行,找到相同列:", col, count)
+                if count >= 2:
+                    header = index_value
+                    debug("非第一行,返回Header", header)
                     return header
 
     return header
@@ -150,8 +160,9 @@ def read_file_to_df(file_path: str, read_cols: Optional[List[str]] = None, trans
         debug(os.path.basename(file_path), "读取第", header, "行")
         if header is None:
             if not_find_header == 'raise':
-                message = '未匹配到开始行，请检查并重新指定'
+                message = '未匹配到开始行，请检查并重新指定:' + str(os.path.basename(file_path))
                 debug(message)
+                error(message + ', 读取列名:' + str(read_cols))
                 raise Exception(message)
             elif not_find_header == 'ignore':
                 pass
@@ -167,10 +178,10 @@ def read_file_to_df(file_path: str, read_cols: Optional[List[str]] = None, trans
                     if end_with_gz:
                         df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, compression='gzip',
                                          header=header,
-                                         nrows=nrows)
+                                         nrows=nrows, index_col=False)
                     else:
                         df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, header=header,
-                                         on_bad_lines='warn', nrows=nrows)
+                                         on_bad_lines='warn', nrows=nrows, index_col=False)
                 else:
                     if end_with_gz:
                         df = pd.read_csv(file_path, encoding=encoding, compression='gzip', header=header, nrows=nrows)
@@ -184,9 +195,9 @@ def read_file_to_df(file_path: str, read_cols: Optional[List[str]] = None, trans
                 for sheet_name in sheet_names:
                     if read_cols:
                         now_df = pd.read_excel(xls, sheet_name=sheet_name, header=header, usecols=read_cols,
-                                               nrows=nrows)
+                                               nrows=nrows, index_col=False)
                     else:
-                        now_df = pd.read_excel(xls, sheet_name=sheet_name, header=header, nrows=nrows)
+                        now_df = pd.read_excel(xls, sheet_name=sheet_name, header=header, nrows=nrows, index_col=False)
 
                     now_df['sheet_name'] = sheet_name
                     df = pd.concat([df, now_df])

utils/log/trans_log.py

@@ -46,7 +46,8 @@ def init_logger():
     # 根据环境设置日志级别
     env = environ.get('env', 'dev')
 
-    stout_handle.setLevel(logging.INFO)
+    # stout_handle.setLevel(logging.INFO)
+    stout_handle.setLevel(logging.DEBUG)
 
     stout_handle.addFilter(ContextFilter())
     logger.addHandler(stout_handle)
@@ -66,10 +67,11 @@ def init_logger():
         if not path.exists(file_path):
             makedirs(file_path, exist_ok=True)
         # 普通日志文件（INFO及以上）
-        file_name = file_path + sep + str(datetime.date.today()) + '.log'
+        file_name = file_path + sep + str(datetime.date.today()) + '.info.log'
         file_handler = logging.FileHandler(file_name, encoding='utf-8')
         file_handler.setFormatter(formatter)
-        file_handler.setLevel(logging.INFO)
+        # file_handler.setLevel(logging.INFO)
+        file_handler.setLevel(logging.DEBUG)
         file_handler.addFilter(ContextFilter())
         logger.addHandler(file_handler)