pyr0ball
a6f60c9e07
Full FastAPI backend for the AI music continuation editor:
Services
- chain.py: chain + node CRUD, commit/discard, recursive CTE spine query
- musicgen.py: MusicGenClient with cf-orch allocation + mock mode (CF_MUSICGEN_MOCK=1)
- stems.py: Demucs 4-stem separation subprocess wrapper + mock mode
- export.py: ffmpeg concat demuxer to stitch committed spine into WAV/MP3
API endpoints
- chains: CRUD, multipart audio upload (WAV/MP3/FLAC/OGG/M4A/AIFF)
- nodes: branch creation (202 + BackgroundTasks), commit, discard, audio stream
- gpu: cf-orch capacity status; session allocation stubbed pending cf-orch#43
- stems: Paid-tier stem separation (Demucs, gated via tiers.py)
- export: POST /{chain_id}/export → FileResponse download
- events: SSE stream (node-status events) per chain via asyncio Queue pub/sub
Infrastructure
- lifespan: reads SPARROW_DB_PATH/DATA_DIR at startup (not import time)
- events_store: subscribe/unsubscribe/broadcast pattern for SSE
- CORS: open in dev, SPARROW_CORS_ORIGINS in production
- Background generation opens its own DB connection (WAL-safe)
Tests: 30/30 passing across service units and API integration
36 lines
1,017 B
Python
36 lines
1,017 B
Python
# tests/test_musicgen_service.py — unit tests for MusicGenClient mock mode
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import pytest
|
|
|
|
# Ensure mock mode
|
|
os.environ["CF_MUSICGEN_MOCK"] = "1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mock_generate_copies_file(tmp_path):
|
|
from app.services.musicgen import MusicGenClient
|
|
|
|
source = tmp_path / "source.wav"
|
|
source.write_bytes(b"RIFF" + b"\x00" * 40)
|
|
output = str(tmp_path / "output.wav")
|
|
|
|
client = MusicGenClient()
|
|
duration = await client.generate(
|
|
source_audio_path=str(source),
|
|
output_path=output,
|
|
prompt="test",
|
|
duration_s=10.0,
|
|
cfg_coef=3.0,
|
|
prompt_duration_s=5.0,
|
|
)
|
|
|
|
assert os.path.exists(output)
|
|
assert duration >= 0.0 # torchaudio may not parse our fake WAV; 0.0 is fine
|
|
|
|
|
|
def test_make_output_path():
|
|
from app.services.musicgen import make_output_path
|
|
path = make_output_path("data", "chain-abc", "node-xyz")
|
|
assert path == "data/chains/chain-abc/nodes/node-xyz.wav"
|