lenML/ChatTTS-Forge

1from typing import List, Optional2 3from fastapi import Body, File, Form, HTTPException, UploadFile4from fastapi.responses import StreamingResponse5from numpy import clip6from pydantic import BaseModel, Field7 8from modules.api import utils as api_utils9from modules.api.Api import APIManager10from modules.api.impl.handler.TTSHandler import TTSHandler11from modules.api.impl.model.audio_model import AdjustConfig, AudioFormat12from modules.api.impl.model.chattts_model import ChatTTSConfig, InferConfig13from modules.api.impl.model.enhancer_model import EnhancerConfig14from modules.data import styles_mgr15from modules.speaker import Speaker, speaker_mgr16 17 18class AudioSpeechRequest(BaseModel):19    input: str  # 需要合成的文本20    model: str = "chattts-4w"21    voice: str = "female2"22    response_format: AudioFormat = "mp3"23    speed: float = Field(1, ge=0.1, le=10, description="Speed of the audio")24    seed: int = 4225 26    temperature: float = 0.327    top_k: int = 2028    top_p: float = 0.729 30    style: str = ""31    batch_size: int = Field(1, ge=1, le=20, description="Batch size")32    spliter_threshold: float = Field(33        100, ge=10, le=1024, description="Threshold for sentence spliter"34    )35    # end of sentence36    eos: str = "[uv_break]"37 38    enhance: bool = False39    denoise: bool = False40 41 42async def openai_speech_api(43    request: AudioSpeechRequest = Body(44        ..., description="JSON body with model, input text, and voice"45    )46):47    model = request.model48    input_text = request.input49    voice = request.voice50    style = request.style51    eos = request.eos52    seed = request.seed53 54    response_format = request.response_format55    if not isinstance(response_format, AudioFormat) and isinstance(56        response_format, str57    ):58        response_format = AudioFormat(response_format)59 60    batch_size = request.batch_size61    spliter_threshold = request.spliter_threshold62    speed = request.speed63    speed = clip(speed, 0.1, 10)64 65    if not input_text:66        raise HTTPException(status_code=400, detail="Input text is required.")67    if speaker_mgr.get_speaker(voice) is None:68        raise HTTPException(status_code=400, detail="Invalid voice.")69    try:70        if style:71            styles_mgr.find_item_by_name(style)72    except:73        raise HTTPException(status_code=400, detail="Invalid style.")74 75    ctx_params = api_utils.calc_spk_style(spk=voice, style=style)76 77    speaker = ctx_params.get("spk")78    if not isinstance(speaker, Speaker):79        raise HTTPException(status_code=400, detail="Invalid voice.")80 81    tts_config = ChatTTSConfig(82        style=style,83        temperature=request.temperature,84        top_k=request.top_k,85        top_p=request.top_p,86    )87    infer_config = InferConfig(88        batch_size=batch_size,89        spliter_threshold=spliter_threshold,90        eos=eos,91        seed=seed,92    )93    adjust_config = AdjustConfig(speaking_rate=speed)94    enhancer_config = EnhancerConfig(95        enabled=request.enhance or request.denoise or False,96        lambd=0.9 if request.denoise else 0.1,97    )98    try:99        handler = TTSHandler(100            text_content=input_text,101            spk=speaker,102            tts_config=tts_config,103            infer_config=infer_config,104            adjust_config=adjust_config,105            enhancer_config=enhancer_config,106        )107 108        buffer = handler.enqueue_to_buffer(response_format)109 110        mime_type = f"audio/{response_format.value}"111        if response_format == AudioFormat.mp3:112            mime_type = "audio/mpeg"113        return StreamingResponse(buffer, media_type=mime_type)114 115    except Exception as e:116        import logging117 118        logging.exception(e)119 120        if isinstance(e, HTTPException):121            raise e122        else:123            raise HTTPException(status_code=500, detail=str(e))124 125 126class TranscribeSegment(BaseModel):127    id: int128    seek: float129    start: float130    end: float131    text: str132    tokens: list[int]133    temperature: float134    avg_logprob: float135    compression_ratio: float136    no_speech_prob: float137 138 139class TranscriptionsVerboseResponse(BaseModel):140    task: str141    language: str142    duration: float143    text: str144    segments: list[TranscribeSegment]145 146 147def setup(app: APIManager):148    app.post(149        "/v1/audio/speech",150        description="""151openai api document: 152[https://platform.openai.com/docs/guides/text-to-speech](https://platform.openai.com/docs/guides/text-to-speech)153 154以下属性为本系统自定义属性，不在openai文档中：155- batch_size: 是否开启batch合成，小于等于1表示不使用batch （不推荐）156- spliter_threshold: 开启batch合成时，句子分割的阈值157- style: 风格158 159> model 可填任意值160        """,161    )(openai_speech_api)162 163    @app.post(164        "/v1/audio/transcriptions",165        response_model=TranscriptionsVerboseResponse,166        description="Transcribes audio into the input language.",167    )168    async def transcribe(169        file: UploadFile = File(...),170        model: str = Form(...),171        language: Optional[str] = Form(None),172        prompt: Optional[str] = Form(None),173        response_format: str = Form("json"),174        temperature: float = Form(0),175        timestamp_granularities: List[str] = Form(["segment"]),176    ):177        # TODO: Implement transcribe178        return api_utils.success_response("not implemented yet")179