[Python] RPA 기초 템플릿

<분류>
A. 수단
- OS/Platform/Tool : Linux, Kubernetes(k8s), Docker, AWS
- Package Manager : node.js, yarn, brew, 
- Compiler/Transpillar : React, Nvcc, gcc/g++, Babel, Flutter
- Module Bundler  : React, Webpack, Parcel

B. 언어
- C/C++, python, Javacsript, Typescript, Go-Lang, CUDA, Dart, HTML/CSS

C. 라이브러리 및 프레임워크 및 SDK
- OpenCV, OpenCL, FastAPI, PyTorch, Tensorflow, Nsight

 

---

1. What? (현상)

업무를 하다보면 간단히 RPA(Robotic Process Automation)가 필요한 경우가 있습니다.
** RPA(Robotic Process Automation) : 사람이 반복적으로 처리해야하는 단순 반복 업무를 하는 프로그램을 의미합니다.

 

이럴 때 필자는 주로 python을 활용하는데, 기초 템플릿을 정리해두고자 합니다.

---

2. Why? (원인)

• X

---

3. How? (해결책)

• 대량의 데이터셋이 있을 때, 해당 데이터를 활용하기 위해 train 리스트와 test 리스트를 7:3으로 나누어 만들어내는 간단한 코드를 기록해두고자 합니다. 아래의 특징이 있습니다.
  • INPUT_DATA_DIR 이라는 실제 데이터가 존재하는 위치를 확인하고자 합니다.
    아래의 예에서 데이터의 구조 : /mnt/dataspace/FROM/PATH/.../data.txt
  • OUTPUT_LIST_DIR에 train.list와 test.list파일을 만들어 데이터의 절대경로들을 기록하고자합니다.
  • OUTPUT_DATA_DIR이라는 새로운 절대경로로 데이터가 존재하는 위치를 기록하고자 합니다.
    아래의 예에서 데이터의 구조 : /mnt/dataspace/TO/data.txt

import  glob
import  os
import  random

RANDOM_ON=True

INPUT_DATA_DIR="/mnt/dataspace/FROM"
OUTPUT_DATA_DIR="/mnt/dataspace/TO"
OUTPUT_LIST_DIR="./"

f_list_train =  open (OUTPUT_LIST_DIR+ '/train.list' , 'w' )
f_list_test =  open (OUTPUT_LIST_DIR+ '/test.list' , 'w' )

#Early Exit ?⑦꽩
for depth1 in os.scandir(os.path.abspath(INPUT_DATA_DIR)):
    if not os.path.isdir(depth1.path):
        continue;
    if depth1.name != "PATH":
        continue
    for depth2 in glob.glob("{}/**/*.txt".format(depth1.path), recursive=True):
        if not os.path.isfile(depth2):
            continue;
        depth2_root = depth2.split('/')[:-1]
        depth2_name = depth2.split('/')[-1]
        if len(depth2_root)<9 or depth2_root[8].split('_')[0] != 'VL':
            continue;
        temp = '/'.join([OUTPUT_DATA_DIR,depth2_name])
        if RANDOM_ON and random.randrange(1,11) <= 7:
            f_list_train.write(temp+'\n')
        elif RANDOM_ON:
            f_list_test.write(temp+'\n')
        else:
            SysystemError('No Random Working')
f_list_train.close()
f_list_test.close()

 

위 예에서는 파일을 Search 하기 위해 glob와 os.scandir을 활용했는데 아래와 같은 특징이 있습니다.

• glob : 어떤 format을 가진 경로에서 디렉토리를 search할 수 있으며, 경로가 나옵니다.
• os.listdir : 지정된 경로에서 디렉토리를 search할 수 있으며, 파일이름이 나옵니다.
• os.scandir : 지정된 경로에서 디렉토리와 파일을 search할 수 있으며, 오브젝트가 나옵니다.

데이터가 많은 경우 느리게 느껴집니다..

 

그럼 병렬로 처리하면 좋겠습니다. RPA작업의 경우 시스템을 활용하는 경우가 많으니 GPU를 활용하기는 어렵고, multi-processing을 구현해보겠습니다.

 

multi-processing으로 구현시에는 critical section이 생기는 경우를 주의해서 구현해야합니다. 위의 경우는 하나의 파일에 기록을 해야하기 때문에, critical section이 발생하고 해당 부분에 대해 lock을 활용해 exclusive하게 동작하도록 만들 수도 있으나, 성능에 문제가 생길 수도 있으니 따로 구분을 해서 실행하는 것이 나을 것 같습니다.

 

먼저 아래와 같이 위 내용을 path를 얻는 get_path 함수와, 처리를 위한 modify_path 함수 두개로 나누었습니다.

import  glob
import  os
import  random

RANDOM_ON=True

INPUT_DATA_DIR="/mnt/dataspace/FROM"
OUTPUT_DATA_DIR="/mnt/dataspace/TO"
OUTPUT_LIST_DIR="./"

f_list_train =  open (OUTPUT_LIST_DIR+ '/train.list' , 'w' )
f_list_test =  open (OUTPUT_LIST_DIR+ '/test.list' , 'w' )

def get_path(input_data_dir):
    for depth1 in os.scandir(os.path.abspath(input_data_dir)):
        if not os.path.isdir(depth1.path):
            continue;
        if depth1.name != "PATH":
            continue
        for depth2 in glob.glob("{}/**/*.txt".format(depth1.path), recursive=True):
            if not os.path.isfile(depth2):
                continue;
            yield depth2

def modify_path(depth2):
    depth2_root = depth2.split('/')[:-1]
    depth2_name = depth2.split('/')[-1]
    if len(depth2_root)<9 or depth2_root[8].split('_')[0] != 'VL':
        return None
    temp = '/'.join([OUTPUT_DATA_DIR,depth2_name])
    return temp

 

자 이제 process pool을 활용해 병렬 처리를 해 보겠습니다.

from multiprocessing import Pool
from tqdm import tqdm
with Pool(8) as p:
    path_list = list(tqdm(p.imap(modify_path,get_path(INPUT_DATA_DIR))))

for temp in path_list:
    if temp==None:
        continue
    if RANDOM_ON and random.randrange(1,11) <= 7:
        f_list_train.write(temp+'\n')
    elif RANDOM_ON:
        f_list_test.write(temp+'\n')
    else:
        SysystemError('No Random Working')
f_list_train.close()
f_list_test.close()

 

두 개의 함수를 연결해주는 함수로 imap을 활용했으나, 다른 함수들도 있습니다.

• p.map : iterator를 받아 처리하며, 단일 인자 input, return은 list
• p.starmap : iterator를 받아 처리하며, 두개 이상 인자 input, return은 list
• p.imap : iterator를 받아 처리하며, 단일 인자 input, return은 iterator

---