File size: 3,640 Bytes
7934b29 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
# Copyright (c) 2020, NVIDIA CORPORATION. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import glob
import os
import time
import librosa
import sox
from joblib import Parallel, delayed
def resample_file(resampled_dir, filepath, ext, sample_rate):
"""
Resample an audio file to 16kHZ and transform to monochannel
Remove incompatible files.
Args:
resampled_dir: Directory of transformed files.
filepath: Filepath of Audio
ext: File type e.g. "wav", "flac"
Returns:
"""
head, filename = os.path.split(filepath)
_, clsname = os.path.split(head)
filename, _ = os.path.splitext(filename)
new_dir = os.path.join(resampled_dir, clsname)
if not os.path.exists(new_dir):
os.makedirs(new_dir)
new_path = os.path.join(new_dir, filename + f'.{ext}')
# check if the resampled data exists.
if os.path.exists(new_path):
print(f"Resampled file {filepath} exists. Skip it.")
return None
transform = sox.Transformer()
transform.set_output_format(file_type='wav')
transform.convert(samplerate=sample_rate, n_channels=1)
try:
transform.build(filepath, new_path)
print(f"Finished converting file {filepath}.")
return None
except sox.core.SoxError as e:
try:
# Check if the file is readable
librosa.load(path=filepath)
# if it is, force input format and try again
transform.set_input_format(file_type=ext)
transform.build(filepath, new_path)
return None
except Exception:
return filepath
def main():
start = time.time()
parser = argparse.ArgumentParser(description='Freesound data resample')
parser.add_argument("--data_dir", required=True, default=None, type=str)
parser.add_argument('--resampled_dir', required=True, default=None, type=str)
parser.add_argument('--sample_rate', default=16000, type=int)
args = parser.parse_args()
data_dir = args.data_dir
resampled_dir = args.resampled_dir
sample_rate = args.sample_rate
wav_files = sorted(glob.glob(os.path.join(data_dir, '*/*.wav')))
flac_files = sorted(glob.glob(os.path.join(data_dir, '*/*.flac')))
with Parallel(n_jobs=-1, verbose=10) as parallel:
wav_files_failed = parallel(
delayed(resample_file)(resampled_dir, filepath, ext='wav', sample_rate=sample_rate)
for filepath in wav_files
)
flac_files_failed = parallel(
delayed(resample_file)(resampled_dir, filepath, ext='flac', sample_rate=sample_rate)
for filepath in flac_files
)
with open('dataset_conversion_logs.txt', 'w') as f:
for file in wav_files_failed:
if file is not None:
f.write(f"{file}\n")
for file in flac_files_failed:
if file is not None:
f.write(f"{file}\n")
end = time.time()
print(f'Resample data in {data_dir} and save to {resampled_dir} takes {end-start} seconds.')
if __name__ == '__main__':
main()
|