| from fastapi import FastAPI, UploadFile, File, BackgroundTasks, HTTPException |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.staticfiles import StaticFiles |
| from fastapi.responses import HTMLResponse |
| import shutil |
| import os |
| import uuid |
| from services.audio_separator import AudioSeparator |
| from services.chord_analyzer import ChordAnalyzer |
| from services.midi_converter import MidiConverter |
| from services.beat_tracker import BeatTracker |
|
|
| app = FastAPI(title="Tunebase AI Engine", description="High-performance audio separation for Math Rock", version="1.0.0") |
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def read_root(): |
| return """ |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <title>Tunebase</title> |
| <style> |
| body, html { margin: 0; padding: 0; height: 100%; overflow: hidden; } |
| iframe { width: 100%; height: 100%; border: none; } |
| </style> |
| </head> |
| <body> |
| <iframe src="https://tunebase.vercel.app" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe> |
| </body> |
| </html> |
| """ |
|
|
| |
| origins = ["*"] |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=origins, |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| |
| UPLOAD_DIR = "uploads" |
| PROCESSED_DIR = "processed" |
| os.makedirs(UPLOAD_DIR, exist_ok=True) |
| os.makedirs(PROCESSED_DIR, exist_ok=True) |
|
|
| |
| app.mount("/processed", StaticFiles(directory=PROCESSED_DIR), name="processed") |
| |
|
|
| |
| |
| try: |
| separator = AudioSeparator() |
| analyzer = ChordAnalyzer() |
| midi_converter = MidiConverter() |
| beat_tracker = BeatTracker() |
| except Exception as e: |
| print(f"Warning: Failed to load models on startup. {e}") |
| import traceback |
| traceback.print_exc() |
| separator = None |
| analyzer = None |
| midi_converter = None |
| beat_tracker = None |
|
|
| @app.get("/") |
| def read_root(): |
| return {"message": "Tunebase AI Engine Ready 🎸"} |
|
|
| @app.post("/upload") |
| async def upload_audio(background_tasks: BackgroundTasks, file: UploadFile = File(...)): |
| """ |
| Upload file audio dan mulai proses separasi di background. |
| """ |
| file_id = str(uuid.uuid4()) |
| file_ext = file.filename.split(".")[-1] |
| file_path = os.path.join(UPLOAD_DIR, f"{file_id}.{file_ext}") |
| |
| with open(file_path, "wb") as buffer: |
| shutil.copyfileobj(file.file, buffer) |
| |
| return {"id": file_id, "filename": file.filename, "status": "uploaded"} |
|
|
| |
| import time |
| import shutil |
|
|
| def cleanup_old_files(): |
| print("Running cleanup task...") |
| now = time.time() |
| cutoff = now - 3600 |
| |
| for folder in os.listdir(PROCESSED_DIR): |
| folder_path = os.path.join(PROCESSED_DIR, folder) |
| if os.path.isdir(folder_path): |
| try: |
| |
| mtime = os.path.getmtime(folder_path) |
| if mtime < cutoff: |
| print(f"Deleting old session: {folder}") |
| shutil.rmtree(folder_path) |
| except Exception as e: |
| print(f"Error cleaning {folder}: {e}") |
|
|
| |
| processing_status = {} |
|
|
| @app.get("/status/{file_id}") |
| async def get_status(file_id: str): |
| return processing_status.get(file_id, {"status": "unknown", "progress": 0, "step": "Waiting"}) |
|
|
| def update_progress(file_id, step, progress, status="processing", data=None): |
| state = { |
| "status": status, |
| "progress": progress, |
| "step": step |
| } |
| if data: |
| state.update(data) |
| processing_status[file_id] = state |
|
|
| @app.post("/process/{file_id}") |
| def process_audio(file_id: str, background_tasks: BackgroundTasks, mode: str = "4stem"): |
| """ |
| Trigger separasi dan analisis. |
| Modes: 2stem (vocals+instruments), 4stem (default), 6stem (full) |
| """ |
| if not separator: |
| raise HTTPException(status_code=503, detail="AI Engine not initialized") |
|
|
| |
| input_path = None |
| |
| for f in os.listdir(UPLOAD_DIR): |
| if f.startswith(file_id): |
| input_path = os.path.join(UPLOAD_DIR, f) |
| break |
| |
| if not input_path: |
| raise HTTPException(status_code=404, detail="File not found") |
| |
| output_dir = os.path.join(PROCESSED_DIR, file_id) |
| |
| |
| space_host = os.environ.get("SPACE_HOST") |
| if space_host: |
| base_url = f"https://{space_host}/processed" |
| else: |
| base_url = "http://localhost:8000/processed" |
| |
| try: |
| update_progress(file_id, "Separating Audio Stems...", 10) |
| |
| |
| def progress_cb(step, prog): |
| update_progress(file_id, step, prog) |
| |
| stems, duration = separator.separate(input_path, output_dir, callback=progress_cb, mode=mode) |
| |
| update_progress(file_id, "Analyzing Rhythm...", 50) |
| |
| |
| update_progress(file_id, "Converting to MIDI...", 70) |
| midi_files = {} |
| if midi_converter: |
| |
| |
| target_stems = ['piano', 'bass', 'guitar_rhythm', 'guitar_lead', 'vocals'] |
| for stem_name in target_stems: |
| if stem_name in stems: |
| stem_path = stems[stem_name] |
| midi_out = os.path.join(output_dir, f"{stem_name}.mid") |
| if midi_converter.convert(stem_path, midi_out): |
| midi_files[stem_name] = f"{base_url}/{file_id}/{stem_name}.mid" |
| |
| update_progress(file_id, "Analyzing Rhythm...", 85) |
| bpm = 0 |
| beats = [] |
| if beat_tracker: |
| |
| beat_source = stems.get('drums') or stems.get('other') or input_path |
| rhythm_data = beat_tracker.track(beat_source) |
| bpm = rhythm_data['bpm'] |
| beats = rhythm_data['beats'] |
| |
| update_progress(file_id, "Finalizing...", 95) |
| |
| update_progress(file_id, "Finalizing...", 95) |
| |
| |
| stems_url = {k: f"{base_url}/{file_id}/{os.path.basename(v)}" for k, v in stems.items()} |
| |
| final_data = { |
| "stems": stems_url, |
| "midi": midi_files, |
| "bpm": bpm, |
| "beats": beats, |
| "duration": duration |
| } |
| print(f"Final Data for {file_id}: {final_data}") |
| update_progress(file_id, "Completed", 100, status="completed", data=final_data) |
| |
| return { |
| "status": "completed", |
| **final_data |
| } |
| |
| except Exception as e: |
| import traceback |
| traceback.print_exc() |
| processing_status[file_id] = {"status": "error", "error": str(e)} |
| print(f"Error processing: {e}") |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
