提交优化，统一行尾

.gitattributes

@@ -0,0 +1,34 @@
+# 默认所有文本文件统一为 LF 行尾
+* text=auto eol=lf
+
+# 源代码 / 配置文件（明确指定 LF）
+*.py   text eol=lf
+*.sh   text eol=lf
+*.yaml text eol=lf
+*.yml  text eol=lf
+*.txt  text eol=lf
+*.md   text eol=lf
+*.json text eol=lf
+*.toml text eol=lf
+*.ini  text eol=lf
+*.cfg  text eol=lf
+*.xml  text eol=lf
+
+# Windows 批处理脚本必须为 CRLF（Windows 硬性要求）
+*.bat  text eol=crlf
+*.cmd  text eol=crlf
+*.ps1  text eol=crlf
+
+# 二进制文件，不做换行符转换
+*.png  binary
+*.jpg  binary
+*.jpeg binary
+*.gif  binary
+*.pdf  binary
+*.zip  binary
+*.gz   binary
+*.tar  binary
+*.xlsx binary
+*.xls  binary
+*.doc  binary
+*.docx binary

.gitignore

@@ -12,3 +12,5 @@ tmp_file
 /test
 .venv
 /utils/tmp_util/
+.vscode/
+

etl/wind_power/wave/WaveTrans.py

@@ -1,195 +1,195 @@
-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()
+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()

ge_requirement.sh

@@ -1,2 +1,2 @@
-#!/bin/bash
-pip freeze > requirements.txt
+source .venv/bin/activate 
+pipreqs . --force

requirements.txt

@@ -7,5 +7,4 @@ PyYAML
 chardet
 psutil
 openpyxl
-xlrd
-redis
+xlrd

utils/file/trans_methods.py

@@ -161,9 +161,10 @@ def read_file_to_df(file_path: str, read_cols: Optional[List[str]] = None, trans
         if header is None:
             if not_find_header == 'raise':
                 message = '未匹配到开始行，请检查并重新指定:' + str(os.path.basename(file_path))
-                debug(message)
-                error(message + ', 读取列名:' + str(read_cols))
-                raise Exception(message)
+                full_message = '未匹配到开始行，请检查并重新指定:' + str('/'.join(file_path.split('/')[-3:]))
+                debug(full_message)
+                error(full_message + ', 读取列名:' + str(read_cols))
+                raise Exception(full_message)
             elif not_find_header == 'ignore':
                 pass
 
@@ -205,7 +206,7 @@ def read_file_to_df(file_path: str, read_cols: Optional[List[str]] = None, trans
             debug('文件读取成功:', file_path, '数据数量:', df.shape, '耗时:', datetime.datetime.now() - begin)
         except Exception as e:
             error('读取文件出错', file_path, str(e))
-            message = '文件:' + os.path.basename(file_path) + ',' + str(e)
+            message = '文件:' + str('/'.join(file_path.split('/')[-3:])) + ',' + str(e)
             raise ValueError(message)
 
     return df