**Configuration(구성/설정)**는 시스템이나 소프트웨어의 특정 작동 방식을 정의하거나 설정하는 과정이나 상태를 의미
파일 처리
def file_process(config) -> None:
print('Start file processing')
preprocessor = PrepareDataset()
if config.target_file:
if not (
config.csv or
config.pkl or
config.split_whole_data or
config.split_train_test
):
print(f'If --target-file (-t) is feed, \
one of --csv, --pkl, --split-train-test (-s) or \
--split-whole-data (-w) must be set.')
return
if config.csv:
preprocessor.save_trn_to_csv(config.target_file)
if config.pkl:
preprocessor.save_trn_to_pkl(config.target_file)
if config.split_whole_data:
preprocessor.split_whole_data(config.target_file)
if config.split_train_test:
preprocessor.split_train_test(
target_file=config.target_file,
train_size=config.ratio
)
if config.convert_all_to_utf:
if not config.target_dir:
print('If --convert-all-to-utf8 (-c) flagged, you must feed --target-dir')
preprocessor.convert_all_files_to_utf8(config.target_dir)
if config.remove_all_text_files:
if not config.target_dir:
print(f'If --remove-all-text-files (-R) flagged,\
you must feed --target-dir')
return
preprocessor.remove_all_text_files(
target_dir=config.target_dir,
ext=config.remove_file_type
)
오디오 와 파일을 다루기 위한 parser / subparser 정의
: argparse는 Python에서 명령줄 인자를 처리하기 위한 표준 라이브러리로, 다양한 옵션과 인자를 간편하게 정의하고 처리할 수 있습니다. 특히 딥러닝 프로젝트에서 오디오 파일을 처리할 때 argparse를 활용해 명령줄에서 다양한 동작을 관리할 수 있습니다. -> 터미널에서 바로 쓸 수 있음
1. Parser와 Subparser 정의
Parser: 최상위 명령어를 처리하는 객체입니다. 사용자가 실행하는 메인 프로그램의 명령줄 인자를 처리합니다. 예를 들어, 오디오 파일의 경로, 배치 크기, 학습률 등을 최상위 parser로 정의할 수 있습니다.
Subparser: 각 기능별로 세분화된 명령어입니다. 이를 통해 한 프로그램 내에서 여러 작업을 처리할 수 있습니다. 예를 들어, 오디오 파일을 전처리하거나 학습하거나 평가하는 명령을 각각 하부 명령어로 등록할 수 있습니다.
def get_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog='Korean speech dataset pre-processor',
description='Process Korean speech dataset'
)
sub_parser = parser.add_subparsers(title='sub-command')
### Parser for sub-command 'audio' ###
parser_audio = sub_parser.add_parser(
'audio',
help='sub-command for audio processing'
)
parser_audio.add_argument(
'--target-dir', '-t',
required=True,
help='directory of audio files'
)
parser_audio.add_argument(
'--remove-original-audio', '-r',
action='store_true',
help='Remove original audio files'
)
parser_audio.set_defaults(func=audio_process)
### Parser for sub-command 'file' ###
parser_file = sub_parser.add_parser(
'file',
help='handling txt encoding, generate pkl/csv file,\
or split file (train/test)'
)
parser_file.add_argument(
'--target-file', '-t',
# required=True,
help='Target file name for processing'
)
parser_file.add_argument(
'--convert-all-to-utf', '-c',
action='store_true',
help='Convert all text files to utf-8 under target_dir'
)
parser_file.add_argument(
'--target-dir', '-d',
# required=True,
help='Target directory for converting file encoding to utf-8\
Use by combining --convert-all-to-utf (-c) flag'
)
parser_file.add_argument(
'--split-whole-data', '-w',
action='store_true',
help='Split whole data file int group'
)
parser_file.add_argument(
'--csv',
action='store_true',
help='Generate csv file'
)
parser_file.add_argument(
'--pkl',
action='store_true',
help='Generate pickle file'
)
parser_file.add_argument(
'--split-train-test', '-s',
action='store_true',
help='Flag split train/test set, \
default: 0.8 (train:test -> 80:20)'
)
parser_file.add_argument(
'--ratio',
type=float,
default=0.8,
help='Split file into .train & .test files'
)
parser_file.add_argument(
'--remove-all-text-files', '-R',
action='store_true',
help='Remove all specific type files under target dir'
)
parser_file.add_argument(
'--remove-file-type',
default='txt',
help='Set remove file type'
)
parser_file.set_defaults(func=file_process)
config = parser.parse_args()
return config
- Move or copy files in our working directory : unzip -q KsponSpeeh_03.zip
- convert all text files into utf-8 encoding :
- Process '.pcm' files into '.wav' format : python <파일이름.py> file --opts
utils.py를 만든다
PrepareDataset 클래스를 만든다
def pcm2audio(
self,
audio_path: str,
ext: str = 'wav',
save_file: bool = True,
remove: bool = False, # Keep original audio file
) -> object:
'''참고 블로그: https://noggame.tistory.com/15'''
buf = None
with open(audio_path, 'rb') as tf:
buf = tf.read()
# zero (0) padding
# 경우에 따라서 PCM 파일의 길이가 8bit(1byte)로
# 나누어 떨어지지 않는 경우가 있어 0으로 패딩을 더해준다.
# 패딩하지 않으면 numpy나 librosa 사용 시 오류가 날 수 있다.
buf = buf+b'0' if len(buf)%2 else buf
pcm_data = np.frombuffer(buf, dtype='int16')
wav_data = lr.util.buf_to_float(x=pcm_data, n_bytes=2)
# 음성 파일을 변환하여 저장: .pcm -> .wav
if save_file:
save_file_name = audio_path.replace('.pcm', f'.{ext}')
sf.write(
file=save_file_name,
data=wav_data,
samplerate=16000,
format='WAV',
endian='LITTLE',
subtype='PCM_16'
)
# 파일 삭제 옵션일 경우에는 .pcm 파일 삭제
if remove:
if os.path.isfile(audio_path):
os.remove(audio_path)
return wav_data
.pcm 파일을 .wav 파일로 변환한 후 현재 디렉토리에 저장
def process_audio(
self,
source_dir: str,
remove_original_audio: bool = True
) -> None:
'''.pcm 파일을 .wav 파일로 변환한 후 현재 디렉토리에 저장'''
print(f'source_dir: {source_dir}')
sub_directories = sorted(os.listdir(source_dir))
print(f'Processing audios: {len(sub_directories)} directories')
for directory in tqdm(sub_directories, desc=f'Processing directory: {source_dir}'):
if os.path.isdir(directory):
files = os.listdir(os.path.join(source_dir, directory))
for file_name in files:
if file_name.endswith('.pcm'):
self.pcm2audio(
audio_path=os.path.join(source_dir, directory, file_name),
ext='wav',
remove=remove_original_audio,
)
else:
file_name = directory
if file_name.endswith('.pcm'):
self.pcm2audio(
audio_path=os.path.join(source_dir, file_name),
ext='wav',
remove=remove_original_audio,
)
모든 텍스트 파일을 utf-8 인코딩으로 변경
def convert_text_utf(self, file_path: str) -> None:
'''파일 인코딩 변경: cp494 -> utf-8로 변환하여 저장'''
try:
with open(file_path, 'rt', encoding='cp949') as f:
lines = f.readlines()
except:
with open(file_path, 'rt', encoding='utf-8') as f:
lines = f.readlines()
with open(file_path, 'wt', encoding='utf-8') as f:
for line in lines:
f.write(line)
def convert_all_files_to_utf8(self, target_dir: str) -> None:
'''디렉토리 내부의 모든 텍스트 파일을 utf-8 인코딩으로 변경'''
print(f'Target directory: {target_dir}')
sub_directories = sorted(os.listdir(target_dir))
num_files = 0
for directory in tqdm(sub_directories, desc='converting cp949 -> utf8'):
files = sorted(os.listdir(os.path.join(target_dir, directory)))
for file_name in files:
if file_name.endswith('.txt'):
self.convert_text_utf(
os.path.join(target_dir, directory, file_name)
)
num_files += 1
print(f'{num_files} txt files are converted.')
print('Done!')
전체 데이터 파일 (train.trn)을 그룹별로 구분
def split_whole_data(self, target_file:str) -> None:
'''전체 데이터 파일 (train.trn)을 그룹별로 구분
For example, in train.trn file
KsponSpeech_01/KsponSpeech_0001/KsponSpeech_000001.pcm :: 'some text'
-> this file will be stored in train_KsponSpeech_01.trn
KsponSpeech_02/KsponSpeech_0001/KsponSpeech_000002.pcm :: 'some text'
-> this file will be stored in train_KsponSpeech_02.trn
'''
with open(target_file, 'rt') as f:
lines = f.readlines()
data_group = set()
for line in lines:
data_group.add(line.split('/')[0])
data_group = sorted(list(data_group))
data_dic = { group: [] for group in data_group} # dict comprehension
for line in lines:
data_dic[line.split('/')[0]].append(line)
# Save file seperately
# target_file: data/info/train.trn -> ['data', 'info', 'train.trn']
save_dir = target_file.split('/')[:-1]
save_dir = '/'.join(save_dir)
for group, line_list in data_dic.items():
file_path = os.path.join(save_dir, f'train_{group}.trn')
with open(file_path, 'wt', encoding='utf-8') as f:
for text in line_list:
f.write(text)
print(f'File created -> {file_path}')
print('Done!')
dict 형태로 가공하여 리턴 -> .csv로 통합하여 저장
def get_dataset_dict(self, file_name: str, ext: str = 'wav') -> dict:
'''path_dir에 있는 파일을 dict 형태로 가공하여 리턴
return data_dic = {
'audio': ['file_path1', 'file_path2', ...],
'text': ['text1', 'text2', ...]
}'''
data_dic = {'path': [], 'sentence': []}
print(f'file_name: {file_name}')
with open(file_name, 'rt', encoding='utf-8') as f:
lines = f.readlines()
for line in lines:
audio, text = line.split('::')
audio = audio.strip()
audio = os.path.join(
os.getcwd(), # '/home/kafa46/finetune-tutorial'
self.VOICE_DIR.replace('./', ''), # './data/audio' -> 'data/audio'
audio
)
if audio.endswith('.pcm'):
audio = audio.replace('.pcm', f'.{ext}')
text = text.strip()
data_dic['path'].append(audio)
data_dic['sentence'].append(text)
return data_dic
def save_trn_to_pkl(self, file_name: str) -> None:
'''.trn 파일을 dict로 만든 후 .pkl 바이너리로 그냥 저장(dump)'''
data_dict = self.get_dataset_dict(file_name)
# pickle file dump
file_name_pickle = file_name + '.dic.pkl'
with open(file_name_pickle, 'wb') as f:
pickle.dump(data_dict, f)
print(f'Dataset is saved via dictionary pickle')
print(f'Dataset path: {file_name_pickle}')
def save_trn_to_csv(self, file_name: str) -> None:
'''.trn 파일을 .csv로 저장'''
data_dic = self.get_dataset_dict(file_name)
file_name_csv = file_name.split('.')[:-1]
file_name_csv = ''.join(file_name_csv) + '.csv'
if file_name.startswith('.'):
file_name_csv = '.' + file_name_csv
data_df = pd.DataFrame(data_dic)
# Modified by Giseop Noh since Whisper dataset needs header ㅠㅠ
# modified: header=False -> header=True
data_df.to_csv(file_name_csv, index=False, header=True)
print(f'Dataset is saved via csv')
print(f'Dataset path: {file_name_csv}')
- Split dataset into 'train' & 'test'
def split_train_test(self,
target_file: str,
train_size: float = 0.8
) -> None:
'''입력 파일(.trn)을 train/test 분류하여 저장
if train_size is 0.8,
train:test = 80%:20%
'''
with open(target_file, 'rt') as f:
data = f.readlines()
train_num = int(len(data) * train_size)
# If you set header (header=True) in csv file, you need following codes
# - Modified by Giseop Noh since Whisper dataset needs header ㅠㅠ
header = None
if target_file.endswith('.csv'):
header = data[0]
data = data[1:]
train_num = int(len(data)*train_size)
shuffle(data)
data_train = sorted(data[0:train_num])
data_test = sorted(data[train_num:])
# train_set 파일 저장
train_file = target_file.split('.')[:-1]
train_file = ''.join(train_file) + '_train.csv'
if target_file.startswith('.'):
train_file = '.' + train_file
with open(train_file, 'wt', encoding='utf-8') as f:
if header:
f.write(header)
for line in data_train:
f.write(line)
print(f'Train_dataset saved -> {train_file} ({train_size*100:.1f}%)')
# test_set 파일 저장
test_file = target_file.split('.')[:-1]
test_file = ''.join(test_file) + '_test.csv'
if target_file.startswith('.'):
test_file = '.' + test_file
with open(test_file, 'wt', encoding='utf-8') as f:
if header:
f.write(header)
for line in data_test:
f.write(line)
print(f'Test_dataset saved -> {test_file} ({(1.0-train_size)*100:.1f}%)')
에릭슨(Ericsson),인텔(Intel),미디어텍(MediaTek),노키아 벨 연구소(Nokia Bell Labs),퀄컴 테크놀로지(Qualcomm Technologies),보다폰(Vodafone)등 세계 주요 통신 및 반도체 기업들이NextG프로그램의 창립 멤버로 참여했다
앤시스(Ansys), Arm, 스위스 취리히연방공과대학(ETH Zurich), 후지쯔(Fujitsu), 키사이트(Keysight), 노키아(Nokia), 노스이스턴 대학교(Northeastern University), 로데슈바르츠(Rohde & Schwarz), 삼성, 소프트뱅크(Softbank), 비아비(Viavi) 등이 이 솔루션을 가장 먼저 도입한 에코시스템 파트너이다. http://www.tnews.kr/news/articleView.html?idxno=112737 엔비디아, AI로 무선 통신 발전시킬 6G 리서치 클라우드 플랫폼 공개 출처 : 티뉴스(http://www.tnews.kr)
