Spaces:
Running
Running
| from fastapi import FastAPI, File, UploadFile, Form | |
| from fastapi.responses import JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| import torch | |
| import os | |
| from pipeline import build_audiosep, separate_audio | |
| # Ensure directories exist | |
| os.makedirs('uploaded', exist_ok=True) | |
| os.makedirs('processed', exist_ok=True) | |
| app = FastAPI() | |
| device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') | |
| model = build_audiosep(config_yaml='config/audiosep_base.yaml',checkpoint_path='checkpoint/audiosep_base_4M_steps.ckpt',device=device) | |
| # Mount the static files directory | |
| app.mount("/processed", StaticFiles(directory="processed"), name="processed") | |
| def process_audio(audio_file_path, text): | |
| output_file = os.path.join('processed', 'SeparatedAudio.wav') | |
| separate_audio(model, audio_file_path, text, output_file, device) | |
| return output_file | |
| async def upload_file(file: UploadFile = File(...), text: str = Form(...)): | |
| contents = await file.read() | |
| # Save the uploaded audio file | |
| input_file_path = os.path.join('uploaded', file.filename) | |
| with open(input_file_path, "wb") as f: | |
| f.write(contents) | |
| # Process the audio file | |
| processed_audio_file_path = process_audio(input_file_path, text) | |
| # Response with processed audio file URL | |
| processed_file_url = f"/processed/{os.path.basename(processed_audio_file_path)}" | |
| return JSONResponse(status_code=200, content={"message": "Audio file received and processed.", "processed_audio_file": processed_file_url}) | |