""" NovaAi – TTS-Engine-Hub main.py Version: v0.1.0 Description: Refactored main.py: uses utils modules for chunking, concat, and cache key generation. All endpoints, features, and logic as before, but cleaner and more modular. Author: Abby (ChatGPT) Date: 2025-07-23 Canvas: main.py """ import asyncio from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse, FileResponse from pydantic import BaseModel import os import base64 import shutil import uvicorn from app.config import settings from app.engines.piper import PiperEngine from app.engines.styletts import StyleTTSEngine from app.engines.chattts import ChatTTSEngine from app.engines.f5_tts import F5TTSEngine from app.utils.text import chunk_text from app.utils.audio import concat_audio from app.utils.cache import build_cache_key # --- Master list of all possible engine classes. --- ALL_ENGINES = { "piper": PiperEngine, "styletts": StyleTTSEngine, "chattts": ChatTTSEngine, "f5-tts": F5TTSEngine, } def create_app(): app = FastAPI( title="NovaAi – TTS-Engine-Hub", version="0.3.0", description="Local-first, modular multi-engine TTS server for your homelab and automation." ) # Dynamically build the registry of active engines based on settings. # This registry is local to the app instance created by this function. app.ENGINE_REGISTRY = {} for engine_name in settings.ACTIVE_ENGINES: if engine_name in ALL_ENGINES: print(f"Activating engine: {engine_name}") app.ENGINE_REGISTRY[engine_name] = ALL_ENGINES[engine_name]() else: print(f"Warning: Engine '{engine_name}' requested in config but not found in ALL_ENGINES.") # Ensure the audio asset/cache directory exists. os.makedirs(settings.AUDIO_CACHE_DIR, exist_ok=True) class TTSRequest(BaseModel): text: str engine: str model: str = None speaker: str = None format: str = "ogg" chunking: bool = False @app.post("/tts") async def tts_endpoint(req: TTSRequest, as_base64: bool = Query(False, alias="as")): # --- 1. Check for engine and handle health --- engine = app.ENGINE_REGISTRY.get(req.engine.lower()) if not engine: raise HTTPException(status_code=404, detail=f"Engine '{req.engine}' not found.") health = engine.healthcheck() if health.get("status") != "ok": raise HTTPException(status_code=503, detail=f"Engine '{req.engine}' is not available. Status: {health.get('status')}") # --- 2. Input validation --- available_models = engine.list_models() if req.model and available_models and req.model not in available_models: raise HTTPException(status_code=400, detail=f"Model '{req.model}' not found for engine '{req.engine}'. Available models: {available_models}") available_voices = engine.list_voices(req.model) if req.speaker and available_voices and req.speaker not in available_voices: raise HTTPException(status_code=400, detail=f"Speaker '{req.speaker}' not found for model '{req.model}'. Available speakers: {available_voices}") # --- 3. Check cache --- cache_key = build_cache_key(req) ext = f'.{req.format.lower()}' output_filename = f"tts_{cache_key}{ext}" output_filepath = os.path.join(settings.AUDIO_CACHE_DIR, output_filename) is_cached = await asyncio.to_thread(os.path.isfile, output_filepath) if is_cached: if as_base64: audio_bytes = await asyncio.to_thread(lambda: open(output_filepath, "rb").read()) audio_b64 = base64.b64encode(audio_bytes).decode("utf-8") return JSONResponse({ "engine": req.engine, "model": req.model, "speaker": req.speaker, "format": req.format, "audio_base64": audio_b64, "chunking": req.chunking, "message": "Audio from cache, base64 included" }) return JSONResponse({ "engine": req.engine, "model": req.model, "speaker": req.speaker, "format": req.format, "audio_url": f"/audio/{output_filename}", "cached": True, "chunking": req.chunking, "message": "Audio served from cache. Download from audio_url" }) # --- 4. Synthesize audio --- try: if req.chunking and len(req.text) > 250: chunks = chunk_text(req.text, maxlen=250) synthesis_tasks = [engine.synthesize(c, speaker=req.speaker, model=req.model, fmt=req.format) for c in chunks] chunk_files = await asyncio.gather(*synthesis_tasks) synthesized_path = await asyncio.to_thread(concat_audio, chunk_files, req.format.lower()) else: synthesized_path = await engine.synthesize(req.text, speaker=req.speaker, model=req.model, fmt=req.format) except (RuntimeError, ValueError, FileNotFoundError) as e: raise HTTPException(status_code=500, detail=f"Error during synthesis: {e}") except Exception as e: raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {e}") # --- 5. Cache and return result --- await asyncio.to_thread(shutil.copy, synthesized_path, output_filepath) # If the synth created a temp file in a different directory, clean it up if settings.AUDIO_CACHE_DIR not in os.path.abspath(synthesized_path): await asyncio.to_thread(os.remove, synthesized_path) if as_base64: audio_bytes = await asyncio.to_thread(lambda: open(output_filepath, "rb").read()) audio_b64 = base64.b64encode(audio_bytes).decode("utf-8") return JSONResponse({ "engine": req.engine, "model": req.model, "speaker": req.speaker, "format": req.format, "audio_base64": audio_b64, "chunking": req.chunking, "message": "Audio from synth, base64 included" }) return JSONResponse({ "engine": req.engine, "model": req.model, "speaker": req.speaker, "format": req.format, "audio_url": f"/audio/{output_filename}", "cached": False, "chunking": req.chunking, "message": "Synthesized new audio. Download from audio_url" }) @app.get("/audio/{filename}") async def audio_file(filename: str): # Made async fpath = os.path.join(settings.AUDIO_CACHE_DIR, filename) if not await asyncio.to_thread(os.path.isfile, fpath) or not await asyncio.to_thread(lambda: fpath.startswith(os.path.abspath(settings.AUDIO_CACHE_DIR))): raise HTTPException(status_code=404, detail="Audio file not found") media_type = "audio/wav" if filename.endswith(".wav") else ( "audio/ogg" if filename.endswith(".ogg") else "audio/mpeg" ) return FileResponse(fpath, media_type=media_type, filename=filename) @app.get("/engines") def engines_endpoint(): engines = {} for name, engine in app.ENGINE_REGISTRY.items(): engines[name] = engine.healthcheck() return engines @app.get("/models") def models_endpoint(): result = {} for name, engine in app.ENGINE_REGISTRY.items(): try: result[name] = engine.list_models() except Exception as e: result[name] = {"error": str(e)} return result @app.get("/speakers") def speakers_endpoint(engine: str, model: str = None): e = app.ENGINE_REGISTRY.get(engine.lower()) if not e: raise HTTPException(status_code=404, detail=f"Engine '{engine}' not found.") try: speakers = e.list_voices(model) except Exception as err: speakers = [] return {"engine": engine, "model": model, "speakers": speakers} @app.get("/version") def version(): return {"version": app.version} @app.get("/health") def health(): status = {name: engine.healthcheck()["status"] for name, engine in app.ENGINE_REGISTRY.items()} return {"status": status, "detail": "API and engines loaded"} return app # If main.py is executed directly, create the app and run uvicorn if __name__ == "__main__": app_instance = create_app() uvicorn.run( app_instance, # Pass the app instance host=settings.HOST, port=settings.PORT, reload=True ) app = create_app()