|
import io |
|
import logging |
|
import os |
|
import tempfile |
|
import wave |
|
import uuid |
|
import zipfile |
|
from zipfile import ZipFile |
|
|
|
import numpy as np |
|
from scipy.io import wavfile |
|
|
|
import api.base |
|
from service.tool import audio_normalize, read_wav_file_to_numpy_array |
|
from vextract.vocal_extract import VEX |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
class VocalRemoverHandler(api.base.ApiHandler): |
|
async def post(self): |
|
try: |
|
uploaded_file = self.request.files['srcaudio'][0] |
|
audio_filebody = uploaded_file['body'] |
|
audio_filename = uploaded_file['filename'] |
|
audio_fileext = os.path.splitext(audio_filename)[-1].lower() |
|
|
|
if audio_fileext != ".wav": |
|
logger.debug(f"file format is {audio_fileext}, not wav\n" |
|
f"converting to standard wav data...") |
|
converted_file = await audio_normalize(full_filename=audio_filename, file_data=audio_filebody) |
|
with wave.open(converted_file, 'rb') as wav_file: |
|
num_frames = wav_file.getnframes() |
|
audiofile_body = wav_file.readframes(num_frames) |
|
logger.debug(f"wav conversion completed.") |
|
os.remove(converted_file) |
|
|
|
with tempfile.NamedTemporaryFile(suffix=audio_fileext, delete=False) as temp_wav: |
|
temp_wav.write(audiofile_body) |
|
temp_wav.close() |
|
|
|
converted_file = await audio_normalize(full_filename=audio_filename, file_data=audio_filebody) |
|
|
|
sampling_rate, audio_array = read_wav_file_to_numpy_array(converted_file) |
|
os.remove(converted_file) |
|
|
|
print(f"Input Audio Shape: {audio_array.shape}\n" |
|
f"Input Audio Data Type: {audio_array.dtype}") |
|
|
|
v = VEX() |
|
[(vocal_sampling_rate, vocal_audio), (accompaniment_sampling_rate, accompaniment_audio)] = v.separate((sampling_rate, audio_array)) |
|
|
|
output_dirname = f"{uuid.uuid4()}" |
|
output_dir = f"output/{output_dirname}" |
|
if not os.path.exists(output_dir): |
|
os.mkdir(output_dir) |
|
wavfile.write(f"{output_dir}/vocals.wav", vocal_sampling_rate, vocal_audio) |
|
wavfile.write(f"{output_dir}/accompaniment.wav", accompaniment_sampling_rate, accompaniment_audio) |
|
|
|
zipfilename = f"{output_dir}/output.zip" |
|
with ZipFile(zipfilename, 'w', zipfile.ZIP_DEFLATED) as zip_obj: |
|
zip_obj.write(f"{output_dir}/vocals.wav") |
|
zip_obj.write(f"{output_dir}/accompaniment.wav") |
|
|
|
logger.debug(f"start output data.") |
|
|
|
self.set_header("Content-Type", "application/zip") |
|
self.set_header("Content-Disposition", "attachment; filename=output.zip") |
|
|
|
with open(zipfilename, "rb") as file: |
|
self.write(file.read()) |
|
await self.flush() |
|
logger.debug(f"response completed.") |
|
|
|
os.remove(f"{output_dir}/vocals.wav") |
|
os.remove(f"{output_dir}/accompaniment.wav") |
|
os.remove(f"{output_dir}/output.zip") |
|
os.rmdir(output_dir) |
|
except Exception as e: |
|
logger.exception(e) |
|
self.set_status(500) |
|
self.write({ |
|
"code": 500, |
|
"msg": "system_error", |
|
"data": None |
|
}) |
|
|
