Pythonでよく使う自作関数

よく使う(使っていきたい)自作関数まとめ

githubに上げるか迷った(多分そのうち上げる)

Load modules

import sys, os
import gc
import time
import pandas as pd
import numpy as np
import codecs
from glob import glob
from pathlib import Path
from googletrans import Translator

Functions

お品書き

• pro_round()
• translate_to_ja()
• list_segments()
• pro_read_csv()
• check_var_info()
• check_data_list()

pro_round(num, ndigits=0)

def pro_round(num, ndigits=0):
    """
    数字を四捨五入で丸める。

    Args:
        num: int or float
            丸めたい数字。
        
        ndigits: int, optional(default=0)
            丸めた後の小数部分の桁数。
    
    Returns:
        rounded: int or float
            丸めた後の数字。
    """
    num *= 10 ** ndigits
    rounded = ( 2* num + 1 ) // 2
    rounded /= 10 ** ndigits
    
    if ndigits == 0:
        rounded = int(rounded)
    
    return rounded

translate_to_ja(text)

def translate_to_ja(text):
    """
    Google翻訳で日本語に翻訳する。
    
    Args:
        text: str
            翻訳したいテキスト。
    
    Returns:
        ja_text: str
            日本語訳されたテキスト。
    """
    text = str(text)
    translator = Translator()
    # srcはデフォルトでautoになっており自動で元の言語を判定してくれる
    ja_text = translator.translate(text, dest='ja').text
    ja_text = str(ja_text)
    return ja_text

list_segments(dir_path='./', rescursive=False, extension=None)

def list_segments(dir_path='./', rescursive=False, extension=None):
    """
    dir_path以下にあるファイルの相対パスのリストを返す
    
    Args:
        dir_path: str, optional(default='./')
            検索したいディレクトリのパス。
        
        rescursive: bool, optional(default=False)
            再帰的に検索するかどうか。
        
        extension: str, list of str or tuple of str, optional(default=None)
            拡張子。csvとか[csv, xlsx]とかみたいな形で指定。
    
    Returns:
        path_list: list of str
            ファイルの相対パスのリスト。
    """
    # ディレクトリ
    dir_p = Path(dir_path)

    # 再帰的に検索するかどうか
    resc_path = './*'
    if rescursive:
        resc_path = '**/*'
    
    # 拡張子
    if extension is None:
        ext_list = ['']
    elif (type(extension) == tuple) or (type(extension) == list):
        extension = list(extension)
        ext_list = ['.' + str(ii) for ii in extension]
    else:
        ext_list = ['.' + str(extension)]
    
    # それぞれの拡張子について検索
    path_list = []
    for ext in ext_list:
        path_list += list(dir_p.glob(resc_path + ext))
    
    # strに直す
    path_list = [str(ii) for ii in path_list]
    # 重複の削除
    path_list = sorted(set(path_list), key=path_list.index) # 3.6以降ではl=list(dict.fromkeys(l))でも

    return path_list

pro_read_csv(path, encoding='utf-8', usecols=None)

def pro_read_csv(path, encoding='utf-8', usecols=None):
    """
    pd.read_csv()で読めないcsvファイルを読み込む。
    
    Args:
        path: str
            読み込むファイルのパス。

        encoding: str, optional(default='utf-8)
            エンコード。

        usecols: list of str, optional(default=None)
            指定したカラムのみを読み込む場合に使用。
    
    Returns:
        df: DataFrame
            読み込んだDataFrame。
    """
    if usecols is None:
        print('usecols: all columns')
        with codecs.open(path, 'r', encoding, 'ignore') as file:
            df = pd.read_table(file, delimiter=',')
    else:
        # 指定したカラムのみ読み込む場合
        print('usecols:', usecols)
        with codecs.open(path, 'r', encoding, 'ignore') as file:
            df = pd.read_table(file, delimiter=',', usecols=usecols)

    return df

check_var_info(df, file_name)

def check_var_info(df, file_name):
    """
    DataFrameの各カラムの示す変数の概要をまとめる。

    Args:
        df: DataFrame
            概要を調べるDataFrame。
        
        file_name: str
            DataFrameが保存されていたファイルの名前。一応パスではない想定。
        
    Returns:
        var_info_df: DataFrame
            各カラムの示す変数の概要をまとめたDataFrame。
    """
    # 各変数についてまとめたい事柄(今後増やしていきたい)
    var_info_cols = [
        'file_name',        # ファイル名
        'var_name',         # 変数名
        'var_name_ja',      # 変数名の日本語訳
        'dtype',            # データ型
        'missing_rate',     # 欠損率
        'n_exist',          # 非欠損数
        'n_missing',        # 欠損数
        'n_rows'            # 行数。レコード数。
    ]

    key = 'var_name'
    var_info_df = pd.DataFrame(columns=['file_name', key])
    var_info_df[key] = df.columns.tolist()
    
    # ファイル名
    var_info_df['file_name'] = file_name

    # 日本語訳
    var_info_df['var_name_ja'] = var_info_df[key].apply(translate_to_ja)
    time.sleep(0.5)

    # データ型
    dtype_df = pd.DataFrame(df.dtypes).reset_index()
    dtype_df.columns = [key, 'dtype']
    var_info_df = pd.merge(var_info_df, dtype_df, on=key, how='inner')

    # 非欠損数
    n_exist_df = pd.DataFrame(df.notnull().sum()).reset_index()
    n_exist_df.columns = [key, 'n_exist']
    var_info_df = pd.merge(var_info_df, n_exist_df, on=key, how='inner')

    # 欠損数
    n_missing_df = pd.DataFrame(df.isnull().sum()).reset_index()
    n_missing_df.columns = [key, 'n_missing']
    var_info_df = pd.merge(var_info_df, n_missing_df, on=key, how='inner')

    # 行数
    var_info_df['n_rows'] = len(df)

    # 欠損率
    e = 'n_exist'
    m = 'n_missing'
    var_info_df['missing_rate'] = var_info_df.apply(lambda row: row[m]/(row[e]+row[m]) if ((row[e]+row[m])!=0) else 0, axis=1)

    # 整理
    var_info_df = var_info_df.reset_index(drop=True)[var_info_cols]

    return var_info_df

check_data_list(path_list, encoding='utf-8')

def check_data_list(path_list, encoding='utf-8'):
    """
    各ファイルの概要をまとめたDataFrameを返す。
    csvとxlsxに対応。

    Args:
        path_list: list of str
            概要を知りたいファイルパスのリスト。
        
        encoding: str, optional(default='utf-8)
            各ファイルを読み込むときのエンコード。
    
    Returns:
        data_info_df: DataFrame
            データの概要をまとめたDataFrame。
        
        var_info_df: DataFrame
            カラムの概要をまとめたDataFrame。
    """
    n_files = len(path_list)    # file数
    print('num of all files:', n_files)

    # dataの概要をまとめたDataFrame
    info_cols = ['path', 'file_name', 'n_rows', 'n_cols', 'columns']
    data_info_df = pd.DataFrame(columns=info_cols)

    # 各カラムの概要をまとめたDataFrame
    var_info_cols = [
        'file_name',
        'var_name',
        'var_name_ja',
        'dtype',
        'missing_rate',
        'n_exist',
        'n_missing',
        'n_rows'
    ]
    var_info_df = pd.DataFrame(columns=var_info_cols)

    # 各ファイルについてまとめる
    for path in path_list:
        path = str(path)
        file_name = path.split('/')[-1].split('\\')[-1]     # ファイル名
        extension = file_name.split('.')[-1]     # 拡張子
        
        if extension == 'csv':
            df = pro_read_csv(path, encoding=encoding)
        elif extension == 'xlsx':
            df = pd.read_excel(path, index=False)
        else:
            # csvでもxlsxでも無ければ一旦無視する。
            continue
        
        shape = df.shape
        n_rows = int(shape[0])   # 行数。レコード数。
        n_cols = int(shape[1])   # 列数。カラム数。
        columns = df.columns.tolist()   # カラム一覧。

        tmp = pd.DataFrame([[path, file_name, n_rows, n_cols, columns]], columns=info_cols)
        data_info_df = data_info_df.append(tmp)

        var_tmp = check_var_info(df, file_name)
        var_info_df = var_info_df.append(var_tmp)

        # 掃除
        del df, tmp, var_tmp
        gc.collect()
    
    data_info_df = data_info_df.reset_index(drop=True)
    var_info_df = var_info_df.reset_index(drop=True)
    print('num of files of data:', len(data_info_df))

    # 戻り値
    ret = data_info_df, var_info_df
    return ret