energy-data-trans/etl/wind_power/laser/LaserTrans.py

import datetime
import multiprocessing
import os.path

import pandas as pd

from service.plt_service import get_all_wind
from service.trans_service import save_df_to_db
from utils.file.trans_methods import read_files, read_file_to_df
from utils.log.trans_log import set_trance_id


class LaserTrans():
    """
    激光测距仪转化
    """

    def __init__(self, field_code, read_path, save_path: str):
        self.field_code = field_code
        self.read_path = read_path
        self.save_path = save_path
        self.begin = datetime.datetime.now()
        self.wind_col_trans, _ = get_all_wind(self.field_code, need_rated_param=False)

    def get_file_data(self, file_path):
        file_name = os.path.basename(file_path)
        wind_farm, wind_turbine_number, acquisition_time, sampling_frequency = file_name.split("_")
        result_df = pd.DataFrame()
        result_df['wind_turbine_number'] = wind_turbine_number
        result_df['acquisition_time'] = pd.to_datetime(acquisition_time, format='%Y%m%d%H%M%S')
        result_df['sampling_frequency'] = sampling_frequency
        result_df['wind_turbine_number'] = result_df['wind_turbine_number'].map(self.wind_col_trans).fillna(
            result_df['wind_turbine_number'])
        # 获取数据
        df = read_file_to_df(file_path)
        result_df['pk_no'] = df['PkNo'].values[0]
        result_df['echo_type'] = df['EchoType'].values[0]
        result_df['echo1_dist'] = df['Echo1Dist'].values
        result_df['echo1_grey'] = df['Echo1Grey'].values
        result_df['echo2_dist'] = df['Echo2Dist'].values
        result_df['echo2_grey'] = df['Echo2Grey'].values
        result_df['echo3_dist'] = df['Echo3Dist'].values
        result_df['echo3_grey'] = df['Echo3Grey'].values

        save_df_to_db(self.field_code + "_laser", result_df)

    def run(self):
        trance_id = '-'.join([self.field_code, 'laser'])
        set_trance_id(trance_id)
        all_files = read_files(self.read_path, ['csv'])
        pool_count = 8 if len(all_files) > 8 else len(all_files)

        with multiprocessing.Pool(pool_count) as pool:
            pool.map(self.get_file_data, all_files)