2024. 1. 14. 13:03ㆍDevelopers 공간 [Shorts]/Vision & Audio
<분류>
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? (현상)
이번 글에서는 Audio 및 Text 데이터를 전처리 하는 과정을 정리하고자 합니다.
보통 경로나 메타데이터를 전처리를 통해 정리하고 Manifest로 만들어 Recipe(레시피)에서 활용하는 경우가 많은데, 이런 과정에 대한 예시라고 보시면 될 것 같습니다.
** Recipe(레시피) : 링크(https://github.com/NVIDIA/NeMo/blob/main/examples/nlp/text_classification/conf/text_classification_config.yaml)와 같이 어떻게 학습시킬지를 정의한 파일이나, 학습시 사용하는 프로세스를 정리한 스크립트 등을 "요리 방법"을 의미하는 레시피라고 합니다.
먼저 데이터를 준비해보겠습니다.데이터는 AiHub의 "저음질 전화망 음성인식 데이터"(https://www.aihub.or.kr/aihubdata/data/view.do?currMenu=115&topMenu=100&aihubDataSe=realm&dataSetSn=571)를 활용해 보이고자 합니다.
먼저 데이터를 다운받았을 때, 아래 명령어를 통해 zip파일을 풀고나면 구조는 아래와 같습니다.
unzip '*.zip'
.../007.저음질 전화망 음성인식 데이터/01.데이터
└── 1.Training
├── 라벨링데이터_230316/VL_DXX/DXX/JXX/SXXXXX/XXXX.txt
├── 라벨링데이터_230316/VL_DXX/DXX/JXX/SXXXXX/SXXXXX.json
└── 원천데이터_230316/VS_DXX/DXX/JXX/SXXXXX/XXXX.wav
└── 2.Validation
├── 라벨링데이터_230316/VL_DXX/DXX/JXX/SXXXXX/XXXX.txt
├── 라벨링데이터_230316/VL_DXX/DXX/JXX/SXXXXX/SXXXXX.json
└── 원천데이터_230316/VS_DXX/DXX/JXX/SXXXXX/XXXX.wav
------------------------------------------------------------
<mac의 exfat 파일시스템을 우분투에서 사용하고 싶을 때>
아래와 같은 명령어로 파일시스템 드라이버를 설치할 수 있습니다.
apt-get install exfat-fuse exfat-utils
------------------------------------------------------------
2. Why? (원인)
- X
3. How? (해결책)
먼저 실행하는 부분입니다. 보통 데이터를 처리하는데 시간이 오래걸리므로, TEST_MODE라는 값을 두어 디버깅을 할 수 있게 해 두었습니다.
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--input-path", default="/root/usb/MusicDisk2/007.??뚯쭏 ?꾪솕留??뚯꽦?몄떇 ?곗씠??01.?곗씠??2.Validation")
parser.add_argument("--process", default=6)
args = parser.parse_args()
TEST_MODE=False
if TEST_MODE :
p = get_processing(args)
samples_info=[]
for path in get_audio_path(args):
samples_info.append(p(path))
else:
# From
audio_path_generator = get_audio_path(args)
# To
data_preper = get_processing(args)
from tqdm import tqdm
from multiprocessing import Pool
with Pool(args.process) as p:
samples_info = list(tqdm(p.imap(data_preper, audio_path_generator)))
위 과정을 보면 get_audio_path()라는 함수를 통해 경로를 얻고, get_processing()이라는 함수를 통해 처리하는 구조입니다.
실제 TEST_MODE가 아닌 경우는 multprocessing을 활용해 병렬처리를 해주었으며, tqdm은 프로그램의 진행상황을 시각적으로 보여주기 위한 패키지이며, 데이터를 받아 list로 만들어주었습니다.
먼저 get_audio_path() 함수를 살펴보겠습니다.
import os
import glob
def get_audio_path(args):
input_path = args.input_path
ext = "wav"
for fname in glob.iglob(input_path + "/**/*."+ext, recursive=True):
file_path = os.path.realpath(fname)
yield file_pathglob.glob은파일 경로의 배열을 반환하는데, glob.iglob은iterator를반환하는 특징을 가지고 있습니다. 우리는 multiprocessing의 imap함수를 이용해 iterator를 연결해주기 위해 iglob을 활용했습니다.
또한 **와 recursive 옵션을 활용하면 모든 경로를 재귀적으로 불러올 수 있습니다.
yield와 return의 차이는 아래 더보기를 통해 참조하세요.
------------------------------------------------------------------------
<yield와 return의 차이>
일반함수가 사용하는 return이 일반 데이터구조 값을 반환하는 데 비해, yield는 반복을 필수로 동반하는 생성기(Generator) 객체를 반환합니다. 따라서 yield를 활용하면 메모리를 사용하지 않기 때문에 대용량 처리에 효율적이며, 성능도 좋습니다.
yield로 받은 객체는 일회용이며, for loop, next() 혹은 list()를 통해 사용할 수 있습니다.
1.for loop를 활용
def get_generator(n):
for x in range(n):
if (x%2==0):
yield x
num = get_generator(10)
for i in num:
print(i)2. next()를 활용
def get_generator(n):
for x in range(n):
if (x%2==0):
yield x
num = get_generator(10)
print(next(num))
print(next(num))
print(next(num))
3. list()를 활용
def get_generator(n):
for x in range(n):
if (x%2==0):
yield x
num = get_generator(10)
print(list(num))
------------------------------------------------------------------------
다음으로 이렇게 얻은 경로를 처리하는 get_processing() 함수를 살펴보겠습니다. 위에서 설명드린 폴더구조를 활용해 해당 위치의 데이터들을 연결해준 뒤, 실제 필요한 정보들을 얻어내는 과정입니다.
아래는 하나의 예시로 생각하시고 원하는 정보들을 다 담아보시면 될 것 같습니다.
import librosa
import json
class get_processing:
def __init__(self, args):
pass
def __call__(self, audio_path):
file_path, ext = os.path.splitext(audio_path)
temp_path = file_path.replace('원천데이터', '라벨링데이터').replace('VS','VL')
text_path = temp_path + '.txt'
temp_path2 = temp_path.split('/')[:-1]
json_path = '/'.join(temp_path2+[temp_path2[-1]+'.json'])
if ext != ".wav":
raise Exception("No Audio ext : {}".format(ext))
if not os.path.exists(audio_path):
raise Exception("No Audio Path : {}".format(audio_path))
if not os.path.exists(text_path):
raise Exception("No Text Path : {}".format(text_path))
if not os.path.exists(json_path):
raise Exception("No Json Path : {}".format(json_path))
duration = librosa.core.get_duration(filename=audio_path)
with open(text_path, 'r') as f:
t, is_noise, fp = text_processing(f.readline())
with open(json_path, 'r') as f:
item = json.load(f)
theme = item['dataSet']['typeInfo']['category'].strip()
return audio_path, t, duration, theme, is_noise, fp
위 내용중 text_processing()이라는 함수를 만들어놨는데 text 데이터를 전처리하기 위해 함수를 만들어 놓았습니다.
아래는 실제 text데이터의 예시이며, 이런 데이터를 처리하기 위해서는 관련 데이터 설명서를 참조하는 것이 좋습니다. 저는 해당 데이터의 AIHub에서 제공하는 "어노테이션 포맷 및 데이터 구조>구축활용가이드 다운로드"(https://www.aihub.or.kr/file/down.do?fileSn=10048&cnstcPrcuseFileSn=10048&dataSetSn=571)를 활용해 참조했습니다.
그럼 텍스트 예시들을 한 번 보시겠습니다.
- o/ 학습지원센터입니다. 무엇을 도와드릴까요?
- n/ 아 네 수고하십니다.
- 아 그게 제가 한국지리가 이미 신청이 돼 있다 해가 지고 혹시나 해서 (())) 신청해본 거거든요
- o/ n/ 아 예 주무관님 제가 (1주일)/(일 주일)에 (1번)/(한 번)씩 안부 전화 드리는 어른신이…
- 아/그러시면 고객님.
위와 같이 어떤 포맷들에서 내가 원하는 형태로 텍스트를 따로 처리해서 뽑아야할 것 같습니다. 아래 코드에서 처리한 내용은 아래와 같습니다.
- "n/"을 noise를 표기하기 위해 사용하는데 noise가 있는지를 먼저 체크합니다.
- filled_pause(간투어)를 처리하기 위해 삭제해주었습니다.
** 간투어 : "음, 뭐, 아" 와 같이 발성자가 다음 발성을 준비하기 위해서 소요되는 시간을 벌기 위해서 발성하는 표현 - 숫자와 관련된 경우(1번)/(한 번) 이렇게 선택하게 되어있는데, 이중에 후자를 선택하도록 했습니다.
- 알파벳과 숫자를 모두 제거 해줍니다.
- 일반 단어나 띄어쓰기를 제외하고 다 제거해줍니다.
- 이중 띄어쓰기를 제거해줍니다.
** re 패키지를 활용하는 간단한 예는 아래 더보기를 통해 참조하세요.
import re
def text_processing(text):
is_noise = False
fp = False
filled_pause_list = ["아/", "그/", "저/", "저기/", "에/", "응/", "으/"]
special_token_list = ["n/", "o/", "b/", "l/", "u/"]
# STEP1. noise
if 'n/' in text:
is_noise = True
# STEP2. remove pause
for f in filled_pause_list:
if f in text:
fp = True
text.replace(f,"",-1)
# STEP3. choose text
n=0
t_list=[]
for i in range(len(text.split(")/(")) -1):
idx = text.find('(', n)
idx2 = text.find(')', n)
idx3 = text.find(')', idx2+1)
t = text[idx+1:idx2]
n = idx3+1
t_list.append(t)
for t in t_list:
text = text.replace(t,"")
# STEP4. remove alphabet, number
text = re.sub('[a-zA-Z0-9]+', '', text)
# STEP5. remove except words
text = re.sub(r'[^\w ]', '', text)
# STEP6. remove double space
text = text.replace(" ", " ", -1)
return text.strip(), is_noise, fp--------------------------------------------------------------------------
<re 패키지 활용하기>
re패키지(Regular Expression)는 정규식으로 일치하는 문자열 집합을 지정하기 위해 사용합니다.
그 중 re.sub(pattern, target, text) 메서드를 활용하면 어떤 정규표현식의 문자열을 치환하기 위해 사용합니다. text의 pattern에 해당하는부분을 target형태로 수정합니다.
위 내용중에 활용했던 패턴들을 적어보겠습니다.
| Pattern | 설명 | 예제 |
| [문자들] | 해당 문자들 중에 하나에 속하는 집합 | [0-5][0-9] : 00에서 59까지의 두자리 숫자 [a-z] : 모든 소문자 ASCII문자 [0-9A-Fa-f] : 모든 16진수 |
| [^문자들] | 해당 문자들이 아닌 집합 | [^5] : 5를 제외한 모든 문자 |
| \ | 이스케이프(Escape)라 하며 해당 특수시퀀스를 알리기 위해 사용합니다. | \- : 실제 리터럴 '-' \w : 모든 문자 \w\w\w : xyz, ABC와 같은 문자 3개 |
| + | 1회 이상의 반복을 모두 포함시키기 위해 사용합니다. | ab+ : a다음에 하나이상의 b가 온다면 |
--------------------------------------------------------------------------
위 처리 내용 중에는 포함되어있지 않지만, 오디오 파일을 처리하는 경우 보통 pcm혹은 wav 포맷으로 되어있는데, 어떤 데이터셋의 경우는 직접 wav파일을 잘라서 써야하는 경우가 있습니다. 아래 더보기를 참조하시길 바랍니다.
------------------------------------------------------------------
<wav파일 자르기 정리>
wav를 자르는 예시들을 정리해보고자 합니다. sox와 librosa, 그리고 scipy 세가지를 통해 처리해보겠으며, 아래 예는 7초부터 시작해 5초가량의 오디오를 얻어내고 싶을 때로 코드를 진행했습니다.
1. sox
설치하는 방법은 아래와 같습니다.
apt-get install sox
bash를 활용해 실행하는 경우는 아래와 같습니다.
sox {wav_path} {output} trim 7.0 5.0
python 파일 내에서 import해 처리하려면 아래와 같이 처리합니다.
import sox
before = sox.file_info.duration("wav_file.wav")
tfm=sox.Transformer()
tfm.trim(7.0,12.0)
tfm.build("wav_file.wav", "output.wav")
after = sox.file_info.duration("output.wav")
혹시 system에서 실행하는 것과 같이 실행하고 싶다면 아래 두가지 방법이 더있습니다.
# subprocess 활용
input = 'wav_file.wav'
output = 'output.wav'
cmd = f"sox {input} {output} trim 7.0 5.0".split(" ")
import subprocess
subprocess.run(cmd)
# os.system 활용
input = 'wav_file.wav'
output = 'output.wav'
cmd = f"sox {input} {output} trim 7.0 5.0".split(" ")
import os
os.system(" ".join(cmd))
2. librosa
먼저 설치합니다.
pip3 install librosa아래와 같이 사용할 수 있습니다.참고로 librosa.output.write()함수는 deprecated되어 사라졌으므로, 아래와 같이 soundfile 패키지를 활용해 저장할 수 있습니다.
import librosa
import soundfile as sf
original_sr = librosa.get_samplerate('wav_file.wav')
start = 7.0 # seconds
end = 12.0 # seconds
y, sr = librosa.load('wav_file.wav') # default : 22k(22050) sample rate
# y, sr = librosa.load('test.wav', sr=original_sr)
ny = y[int(start*sr):int(end*sr)]
sf.write("output.wav", ny, sr, format='WAV')
3. scipy
먼저 설치합니다.
pip3 install scipy아래와 같이 활용할 수 있습니다.
from scipy.io import wavfile
start = 7.0 # seconds
end = 12.0 # seconds
y, sr = wavfile.read('wav_file.wav') # default : 16k(16000) sample rate
ny = y[int(start*sr):int(end*sr)]
wavfile.write("output.wav", sr, ny)
------------------------------------------------------------------
이외에 다른 처리하는 방법들은 아래 더보기에 정리해두었습니다.
-----------------------------------------------------
<기타 상황 처리하는 방법>
1. wav 파일이 아닌 경우 wav파일로 바꿔주기
import subprocess
cmd = f"sox -r 16000 -t raw -e signed-integer -b 16 -c 1 {from} {to}".split(" ")
subprocess.run(cmd)
2. ffprobe를 통해 codec 정보를 얻기
먼저 설치하는 방법은 아래와 같습니다.
apt-get install ffprobe
아래와 같이 subprocess로 binary파일을 실시간으로 넘겨줄 수 있습니다.
import subprocess
import json
with open('broadcast.wav', 'rb') as rf:
audio = rf.read()
cmd = f"ffprobe -show_format -show_streams -of json pipe:".split(" ")
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
communicate_kwargs={}
communicate_kwargs['timeout'] = None
communicate_kwargs['input'] = audio
out, err = p.communicate(**communicate_kwargs)
result = json.loads(out.decode('utf-8'))
for stream in result['streams']:
print("codec : {}({}), sr : {}, channels : {}".format(stream['codec_type'], stream['codec_name'], stream['sample_rate'], stream['channels']))
3.duration 정보를 얻기
librosa 패키지를 활용해 쉽게 얻을 수 있습니다.
import librosa
duration = librosa.core.get_duration(wav_path)
-----------------------------------------------------
자 이제 마지막으로 이렇게 얻은 데이터를 처리해 manifest 형태로 output.json을 만들어보겠습니다.
with open('output.json', 'w') as f:
for sample_info in samples_info:
audio_path, t, duration, theme, is_noise, fp = sample_info
metadata = {
"data_path":audio_path,
"label":t,
"duration":duration,
"theme":theme,
"noise":is_noise,
"filled_pause":fp
}
if duration >= 0.1:
json.dump(metadata, f, ensure_ascii=False)
f.write('\n')
https://docs.python.org/ko/3/library/re.html
'Developers 공간 [Shorts] > Vision & Audio' 카테고리의 다른 글
| [Audio] ASR에서 CER, WER 구현 및 측정해보기 (0) | 2024.02.09 |
|---|---|
| [NLP] KenLM으로 Language Model만들기 (1) | 2024.02.09 |
| [NLP] 형태소 분석기 및 토크나이저 활용해보기 (0) | 2024.01.03 |
| [Audio] Python으로 VAD 구현하기 (0) | 2023.12.30 |
| [NLP] 한글을 다양하게 조작해보기 (0) | 2023.11.12 |