from .parallel_read import parallel_read
from .left_merge import left_merge


__copyright__    = 'Copyright (C) 2023 Tea Kitano'
__version__      = '1.0.0'
__license__      = 'BSD-3-Clause'
__author__       = 'Tea Kitano'
__author_email__ = 'chachamusics@outlook.com'
__url__          = 'https://github.com/TeaKitano/pdhelper'
pdhelper/__init__.py
TeaKitano/pdhelper
import pandas as pd


def left_merge(df: pd.DataFrame, df_sub: pd.DataFrame, left_on: str, right_on: str, target: str | list[str]):
    """
    左側結合専用のmerge関数\n
    input\n
        df: 結合元データ\n
        df_sub: 結合したいデータ\n
        left_on: dfの結合キー\n
        right_on: df_subの結合キー\n
        target: 使いたい列名 複数の場合リストで指定\n
    output\n
        df: マージされたDataFrame
    """
    # df_subの使う列抽出
    if type(target) == list:
        target.append(right_on)
    else:
        target = [target, right_on]

    sub_len = len(df_sub)
    df_sub = df_sub[~df_sub[right_on].duplicated()]  # 被りを消す
    if sub_len != len(df_sub):
        print("warning:mergeするデータに被りがありました。一番上のデータのみを利用しています。")
    df = pd.merge(df, df_sub.loc[:, target],
                  how="left", left_on=left_on, right_on=right_on)
    if right_on != left_on:
        df.drop(right_on, axis=1, inplace=True)
    return df

pdhelper/left_merge.py
TeaKitano/pdhelper
import pandas as pd
from concurrent.futures import ProcessPoolExecutor


def _file_read(file: dict) -> pd.DataFrame:
    # ファイル読み込みのための内部関数
    filename = file["filename"]
    conf = file["conf"]
    header = file["header"]
    index = file["index"]
    if header:
        header = 0
    else:
        header = None
    if index:
        index = 0
    else:
        index = None
    extention = filename.split(".")[-1].lower()  # 拡張子入手
    if extention == "csv":
        if conf == "":
            return pd.read_csv(filename, header=header, index_col=index)
        else:
            return pd.read_csv(filename, encoding=conf, header=header, index_col=index)
    elif extention == "xlsx" or extention == "xls":
        if conf == "":
            return pd.read_excel(filename, header=header, index_col=index)
        else:
            return pd.read_excel(filename, sheet_name=conf, header=header, index_col=index)
    else:
        # 拡張子がcsv,xlsx,xls以外だったらこのエラー
        raise ValueError("FileExtentionError")


def parallel_read(files: dict):
    """
    並列処理でファイルを読み込む関数\n
    input:\n
        files  読み込むファイルのdict\n
            key:出力のdictのkeyにしたい名前
            value:各種設定のdict
                filename:ファイルへのパス
                conf:追加設定 csvの場合文字コードを(空欄ならutf-8)、excelファイルの場合使うシート名を(空欄なら1枚目のシート)書く\n
                header:ヘッダーついてるかどうか　ついてたらTrue ついてなかったらFalse\n
                index:インデックスついてるかどうか　ついてたらTrue ついてなかったらFalse\n
    outout:\n
        読み込んだDataFrameの辞書
        key:inputのkey
        value:読み込んだDataFrame
    """
    df_dict = dict()
    with ProcessPoolExecutor() as executor:
        for i in files:
            df_dict[i] = executor.submit(_file_read, files[i])
    for i in df_dict:
        df_dict[i] = df_dict[i].result()
    return df_dict

pdhelper/parallel_read.py
TeaKitano/pdhelper
from glob import glob
from os.path import basename
from os.path import splitext

from setuptools import setup
from setuptools import find_packages


def _requires_from_file(filename):
    return open(filename).read().splitlines()

setup(
    name='pandashelper',
    version='1.0.0',
    license='BSD 3-Clause License',
    description='Useful pandas helper',
    author='TeaKitano',
    author_email='chachamusics@outlook.com',
    url="https://github.com/TeaKitano/pdhelper",
    packages=["pdhelper"],
    include_package_data=True,
    zip_safe=False,
    install_requires=["pandas","openpyxl"]
)

setup.py
TeaKitano/pdhelper
import unittest
import pdhelper as ph
import pandas as pd
import time
import json

def file_read(file: dict) -> pd.DataFrame:
    filename = file["filename"]
    conf = file["conf"]
    header = file["header"]
    index = file["index"]
    if header:
        header = 0
    else:
        header = None
    if index:
        index = 0
    else:
        index = None
    extention = filename.split(".")[-1].lower()  # 拡張子入手
    if extention == "csv":
        if conf == "":
            return pd.read_csv(filename, header=header, index_col=index)
        else:
            return pd.read_csv(filename, encoding=conf, header=header, index_col=index)
    elif extention == "xlsx" or extention == "xls":
        if conf == "":
            return pd.read_excel(filename, header=header, index_col=index)
        else:
            return pd.read_excel(filename, sheet_name=conf, header=header, index_col=index)
    else:
        # 拡張子がcsv,xlsx,xls以外だったらこのエラー
        raise ValueError("FileExtentionError")

class TestFunc(unittest.TestCase): # テストのためのクラス
    def test_read(self):
        with open("test_data/parallel/test_input.json") as f:
            files = json.load(f)
        start = time.time()
        arr = ph.parallel_read(files)
        end = time.time()
        print("parallel:"+str(end-start))

        l = dict()
        start = time.time()
        for i in files:
            l[i] = file_read(files[i])
        end = time.time()
        print("single:"+str(end-start))
        for i in arr:
            self.assertEqual(l[i].equals(arr[i]),True)

    def test_merge(self):
        left = pd.read_csv("test_data/merge/left1.csv")
        right = pd.read_csv("test_data/merge/right1.csv")
        ans = pd.read_csv("test_data/merge/out.csv")

        df = ph.left_merge(left,right,"name","named",["age"])
        self.assertEqual(df.equals(ans),True)

if __name__ == '__main__':
    unittest.main()
test.py
TeaKitano/pdhelper