Spaces:
Sleeping
Sleeping
File size: 2,484 Bytes
01e655b d5b3cd8 01e655b d5b3cd8 01e655b d5b3cd8 01e655b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
from fastapi import HTTPException, Body
from fastapi.responses import StreamingResponse
import io
from pydantic import BaseModel
from fastapi.responses import FileResponse
from modules.normalization import text_normalize
from modules.ssml import parse_ssml
from modules.SynthesizeSegments import (
SynthesizeSegments,
synthesize_segment,
combine_audio_segments,
)
from modules.api import utils as api_utils
from modules.api.Api import APIManager
class SSMLRequest(BaseModel):
ssml: str
format: str = "mp3"
batch_size: int = 4
async def synthesize_ssml(
request: SSMLRequest = Body(
..., description="JSON body with SSML string and format"
)
):
try:
ssml = request.ssml
format = request.format
batch_size = request.batch_size
if batch_size < 1:
raise HTTPException(
status_code=400, detail="Batch size must be greater than 0."
)
if not ssml:
raise HTTPException(status_code=400, detail="SSML content is required.")
segments = parse_ssml(ssml)
for seg in segments:
seg["text"] = text_normalize(seg["text"], is_end=True)
if batch_size != 1:
synthesize = SynthesizeSegments(batch_size)
audio_segments = synthesize.synthesize_segments(segments)
combined_audio = combine_audio_segments(audio_segments)
buffer = io.BytesIO()
combined_audio.export(buffer, format="wav")
buffer.seek(0)
if format == "mp3":
buffer = api_utils.wav_to_mp3(buffer)
return StreamingResponse(buffer, media_type=f"audio/{format}")
else:
def audio_streamer():
for segment in segments:
audio_segment = synthesize_segment(segment=segment)
buffer = io.BytesIO()
audio_segment.export(buffer, format="wav")
buffer.seek(0)
if format == "mp3":
buffer = api_utils.wav_to_mp3(buffer)
yield buffer.read()
return StreamingResponse(audio_streamer(), media_type=f"audio/{format}")
except Exception as e:
import logging
logging.exception(e)
raise HTTPException(status_code=500, detail=str(e))
def setup(api_manager: APIManager):
api_manager.post("/v1/ssml", response_class=FileResponse)(synthesize_ssml)
|