Me gustaría saber si hay una mejor manera de convertir archivos de audio en formato base64 en archivos .wav sin almacenarlos en el directorio actual.
El caso es que obtengo archivos de audio en formato base64 del usuario que carga con una solicitud POST con FastAPI, luego los decodifico y los convierto en archivos .wav porque necesito pasar los archivos sobre algunas funciones que creé para preprocesar y transcriba los archivos y esas funciones usan el módulo wave para archivos .wav. Debido a que creé archivos .wav solo para transcribirlos, no necesito almacenarlos y finalmente los elimino con la función os.unlink.
import fastapi_server.preprocessing_f as pr app = FastAPI() class AudioBase64(BaseModel): audio_name: str = Field(..., min_length=1, example="my-audio") data_base64: str = Field(..., min_length=1) @app.post( path="/upload-base64-audios/", status_code=status.HTTP_200_OK ) async def upload_base64_audios(audios: list[AudioBase64] = Body(...)): model: str = "~/models" dir_name = os.path.expanduser(model) output_graph, scorer = pr.resolve_models(dir_name) model_retval: List[str] = pr.load_model(output_graph, scorer) all_names: list[str] = [] all_datas: list[str] = [] all_decode: list[str] = [] aggresive = 1 transcriptions: list[str] = [] new_data: list[str] = [] final_data: list[str] = [] header: list[str] = ["audio_name", "transcriptions"] for i in range(len(audios)): name = audios[i].audio_name data = audios[i].data_base64 decode = base64.b64decode(data) all_names.append(name) all_datas.append(data) all_decode.append(decode) filename = "%s.wav" % name with open(filename, "wb") as f: f.write(decode) cwd = os.getcwd() files = glob.glob(cwd + "/" + name + ".wav") segments, sample_rate, audio_length = pr.vad_segment_generator( files[0], aggresive ) for k, segment in enumerate(segments): audio = np.frombuffer(segment, dtype=np.int16) output = pr.stt(model_retval[0], audio) transcript = output[0] transcriptions.append(transcript) new_data = [all_names[i], transcriptions[i]] final_data.append(new_data) dir_files = glob.glob(cwd + "/*.wav") for file in dir_files: os.unlink(file) new_df = pd.DataFrame(final_data, columns=header) stream = io.StringIO() new_df.to_csv(stream, index=False) response: Response = StreamingResponse( iter([stream.getvalue()]), media_type="text/csv" ) response.headers["Content-Disposition"] = "attachment; filename=my-file.csv" return responseComo sugirió @martineau , intente escribir de forma intermedia en un io.BytesIO , pero después de escribir, llame a .seek(0) para devolver la posición de vapor al inicio, en lugar de llamar a getbuffer()
(después de escribir, la posición de la transmisión estará al final, lista para recibir más datos)
with io.BytesIO() as buffer: buffer.write(decode) buffer.seek(0) # rewind stream ... segments, sample_rate, audio_length = pr.vad_segment_generator( buffer, aggresive)