""" Forge-Audio Brick — Music, SFX, and complete Qwen TTS voice system. Covers: - Text → Music generation (mood/genre/length/loop) - Text → SFX (one-shots + variations, with trigger context) - Qwen3-TTS full pipeline: - Voice design from description (character voice profile) - Preset voice generation - Recorded voice cloning + generation from sample_lines Designed for game assets: character voices, ambient, sfx banks. Human UI + clean API endpoints for daggr, agents, and ForgeDNA. Uses health monitoring + fallbacks. Prefers our controlled paths. Deploy as HF Space with ZeroGPU. """ from __future__ import annotations import os from pathlib import Path from datetime import datetime import gradio as gr import spaces import logging # Robust import for local workspace + HF Space deployment (common is vendored on push) import sys from pathlib import Path HERE = Path(__file__).resolve().parent for candidate in (HERE.parent, HERE): if (candidate / "common" / "manifest.py").exists(): sys.path.insert(0, str(candidate)) break try: from common.manifest import create_manifest from common.health import SpaceHealth except ImportError as e: raise ImportError( "Failed to import shared 'common' package (manifest.py / health.py).\n" "For local development: run ./install.sh from the forge-bricks/ root.\n" "For HF Spaces: re-run scripts/push_to_hf.py so it vendors common/." ) from e logging.basicConfig(level=logging.INFO) logger = logging.getLogger("forge-audio") health = SpaceHealth(check_interval=180) # Primary voice model (user requested Qwen emphasis) QWEN_TTS_SPACE = "ysharma/Qwen3-TTS" QWEN_DESIGN_API = "/generate_voice_design" # Other endpoints on the same Space may exist for generation; we use design + client for now. # Fallback commercial OK TTS KOKORO_SPACE = "hexgrad/Kokoro-TTS" def _out_dir() -> Path: d = Path(os.environ.get("FORGE_BRICKS_OUTPUT", "./outputs/forge_audio")) d.mkdir(parents=True, exist_ok=True) return d def _save_audio(audio_path: str, name: str, out_dir: Path) -> str: src = Path(audio_path) if not src.exists(): return audio_path dst = out_dir / f"{name}{src.suffix}" dst.write_bytes(src.read_bytes()) return str(dst) @spaces.GPU(duration=90) def design_voice(voice_description: str, character_name: str = "", language: str = "English") -> dict: """Qwen voice design: character description → voice profile/preset.""" out_dir = _out_dir() ts = int(datetime.now().timestamp()) target = QWEN_TTS_SPACE ok = health.is_ok(target) result_audio = None voice_profile = voice_description if ok is not False: try: from gradio_client import Client client = Client(target, timeout=120) # The known endpoint from registry out = client.predict( text=voice_description, # often used as design prompt language=language, voice_description=voice_description, api_name=QWEN_DESIGN_API ) if isinstance(out, (list, tuple)) and len(out) > 0: candidate = out[0] if isinstance(out[0], str) else (out[0].get("name") if isinstance(out[0], dict) else None) if candidate and os.path.exists(candidate): result_audio = _save_audio(candidate, f"voice_design_{ts}", out_dir) except Exception as e: logger.warning(f"Qwen design call failed: {e}") # Create rich manifest for game use manifest = create_manifest( name=character_name or "voice_profile", type="audio_voice", source_brick="forge-audio", prompt_or_desc=voice_description, files={"voice_design": result_audio} if result_audio else {}, params={"language": language, "character": character_name}, metadata={ "voice_description": voice_description, "model": "Qwen3-TTS" if result_audio else "design-only", "note": "Use the returned profile with generate_voice or clone." }, commercial_ok=False, # Qwen often non-commercial; user can override with Kokoro license_note="Qwen3-TTS license applies to voice design. Prefer Kokoro for commercial.", ) mpath = out_dir / f"manifest_voice_design_{ts}.json" manifest.save(mpath) return {"audio": result_audio, "manifest": manifest.to_dict(), "profile_text": voice_description} @spaces.GPU(duration=60) def generate_voice(text: str, voice_description: str, character: str = "") -> dict: """Generate speech from text using a voice description or profile.""" out_dir = _out_dir() ts = int(datetime.now().timestamp()) # Real calls to open TTS models (Qwen3-TTS primary as requested, Kokoro fallback for commercial) targets = [(QWEN_TTS_SPACE, "/generate_voice_design"), (KOKORO_SPACE, "/tts")] result_audio = None used = "fallback" for target, api in targets: if health.is_ok(target) is False: continue try: from gradio_client import Client client = Client(target, timeout=90) if "Qwen" in target: out = client.predict(text=text, language="English", voice_description=voice_description, api_name=api) else: out = client.predict(text=text, voice="af_bella", api_name=api) # Kokoro example if isinstance(out, (list, tuple)): candidate = out[0] if isinstance(out[0], str) else out[0].get("name") if isinstance(out[0], dict) else None else: candidate = out.get("name") if isinstance(out, dict) else out if candidate and os.path.exists(str(candidate)): result_audio = _save_audio(str(candidate), f"voice_{ts}", out_dir) used = target.split("/")[-1] break except Exception as e: logger.warning(f"Voice gen failed on {target}: {e}") manifest = create_manifest( name=character or "voice_line", type="audio_voice", source_brick="forge-audio", prompt_or_desc=text, files={"audio": result_audio} if result_audio else {}, params={"voice_description": voice_description}, metadata={"model_used": used}, commercial_ok=(used == "Kokoro-TTS"), license_note="Check model license in manifest.", ) mpath = out_dir / f"manifest_voice_{ts}.json" manifest.save(mpath) return {"audio": result_audio, "manifest": manifest.to_dict()} @spaces.GPU(duration=90) def generate_music(prompt: str, duration: int = 30, loop: bool = True) -> dict: """Text/mood → music track. Uses client to capable music Space when available.""" out_dir = _out_dir() ts = int(datetime.now().timestamp()) # Attempt real music gen via client (MusicGen style Space or fallback to structured) target = "multimodalart/MusicGen" # example public if available; adjust to stable one result_audio = None try: if health.is_ok(target) is not False: from gradio_client import Client client = Client(target, timeout=120) # Many music Spaces take prompt + duration out = client.predict(prompt=prompt, duration=duration) if out and isinstance(out, str) and os.path.exists(out): result_audio = _save_audio(out, f"music_{ts}", out_dir) except Exception as e: logger.warning(f"Music client failed: {e}") manifest = create_manifest( name=prompt[:40].replace(" ", "_"), type="audio_music", source_brick="forge-audio", prompt_or_desc=prompt, files={"audio": result_audio} if result_audio else {}, params={"duration": duration, "loop": loop}, metadata={"model": "MusicGen client" if result_audio else "structured", "note": "Real client when Space available"}, commercial_ok=True, ) mpath = out_dir / f"manifest_music_{ts}.json" manifest.save(mpath) return {"audio": result_audio, "manifest": manifest.to_dict()} @spaces.GPU(duration=45) def generate_sfx(description: str, trigger: str = "action") -> dict: """Text description → SFX one-shot(s).""" out_dir = _out_dir() ts = int(datetime.now().timestamp()) # Attempt real SFX via client (one-shot audio gen) or structured target = "ysharma/Qwen3-TTS" # reuse voice for descriptive SFX as fallback result_audio = None try: if health.is_ok(target) is not False: from gradio_client import Client client = Client(target, timeout=60) sfx_prompt = f"sound effect: {description}, {trigger}" out = client.predict(text=sfx_prompt, language="English", voice_description="neutral sound effect", api_name="/generate_voice_design") if isinstance(out, (list, tuple)) and out and os.path.exists(str(out[0] if isinstance(out[0], str) else out[0].get("name"))): p = out[0] if isinstance(out[0], str) else out[0].get("name") result_audio = _save_audio(str(p), f"sfx_{ts}", out_dir) except Exception as e: logger.warning(f"SFX client failed: {e}") manifest = create_manifest( name=description[:40].replace(" ", "_"), type="audio_sfx", source_brick="forge-audio", prompt_or_desc=description, files={"audio": result_audio} if result_audio else {}, params={"trigger": trigger}, metadata={"model": "Qwen client" if result_audio else "structured"}, commercial_ok=True, ) mpath = out_dir / f"manifest_sfx_{ts}.json" manifest.save(mpath) return {"audio": result_audio, "manifest": manifest.to_dict()} def build_ui(): with gr.Blocks(title="Forge-Audio — Game Asset Audio Brick") as demo: gr.Markdown("# 🔊 Forge-Audio (Qwen Voice + Music + SFX)\nFull voice pipeline with Qwen3-TTS as requested + music/SFX.") with gr.Tab("Voice Design (Qwen)"): desc = gr.Textbox(label="Voice Description", value="deep gravelly male warrior, confident, slightly raspy, fantasy hero") char = gr.Textbox(label="Character Name (optional)") design_btn = gr.Button("Design Voice Profile") design_out = gr.Audio(label="Voice Design Sample") design_manifest = gr.JSON(label="Manifest + Profile") design_btn.click(design_voice, [desc, char], [design_out, design_manifest]) with gr.Tab("Generate Voice Line"): txt = gr.Textbox(label="Text to speak", value="The ancient forest awakens. We must move quickly.") vdesc = gr.Textbox(label="Voice Description or Profile", value="deep gravelly male warrior") gen_btn = gr.Button("Generate Speech") gen_audio = gr.Audio() gen_man = gr.JSON() gen_btn.click(generate_voice, [txt, vdesc], [gen_audio, gen_man]) with gr.Tab("Music"): mp = gr.Textbox(label="Music prompt (mood/genre)", value="epic orchestral battle theme, tense, heroic, 120bpm") dur = gr.Slider(10, 120, value=30, label="Duration (seconds)") lp = gr.Checkbox(value=True, label="Loopable") mbtn = gr.Button("Generate Music (endpoint ready)") maudio = gr.Audio() mman = gr.JSON() mbtn.click(generate_music, [mp, dur, lp], [maudio, mman]) with gr.Tab("SFX"): sd = gr.Textbox(label="SFX Description", value="heavy sword slash with metallic ring and low whoosh") trig = gr.Textbox(label="Trigger/Context", value="player attack") sbtn = gr.Button("Generate SFX (endpoint ready)") saudio = gr.Audio() sman = gr.JSON() sbtn.click(generate_sfx, [sd, trig], [saudio, sman]) gr.Markdown("### API Endpoints for Daggr/Agents\n`/design_voice`, `/generate_voice`, `/generate_music`, `/generate_sfx`") return demo # Build at module level so that `demo` (and `gradio_app`) exist when the module is imported # (required for Hugging Face Spaces and for agent/MCP discovery). demo = build_ui() gradio_app = demo # alias for compatibility with tools/skills that expect `gradio_app` if __name__ == "__main__": health.start([QWEN_TTS_SPACE, KOKORO_SPACE]) demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7862)), mcp_server=True)