o i~@sUdZddlmZddlZddlZddlZddlZddlZddlZddl m Z m Z ddl m Z mZmZmZmZmZeeZdmdnd d ZdodpddZe GdddZGdddeZGdddZe GdddZdaded<dqddZdrdd Z dsd"d#Z!dtd%d&Z"dud'd(Z#dvd,d-Z$    .  / 0 1       dwdxdDdEZ%dydFdGZ&dzdHdIZ' J K L M N  Od{d|dZd[Z(d}d^d_Z) J K L M N  ` Od~ddbdcZ* J K L M N  d  ddddidjZ+ddkdlZ,dS)a TTS runtime wrapper (container-scoped singleton). In Modal, each container should load the model once and then handle many requests. This wrapper owns: - SparkTTSModel (vLLM engine) - IndicPromptBuilder - BiCodecDecoder - SparkTTSPipeline + Veena3SlidingWindowPipeline - optional SuperResolutionService Imports framework-agnostic inference + processing code from `veena3modal/core`, `veena3modal/processing`, and `veena3modal/audio`. ) annotationsN) dataclassfield)AnyAsyncGeneratorDictListOptionalTuplenamestrdefaultintminimumreturncCDtj|d}|s |Szt|}Wn ty|YSwt||SN)osenvirongetr ValueErrormaxr r rrawvaluer;/home/ubuntu/veenaModal/veena3modal/services/tts_runtime.py_env_int   rfloatcCrr)rrrr!rrrrrr _env_float)rr"c@sZeZdZUdZdZded<dZded<dZd ed <dZd ed <dZ d ed <dZ ded <dS)StreamingAdmissionLeasez2Admission ticket for one active streaming request.r r!wait_msFboolqueuedrrinflight_on_grantwaiters_on_grantqueue_depth_on_entryreleasedN) __name__ __module__ __qualname____doc__r$__annotations__r&r'r(r)r*rrrrr#4s      r#cs*eZdZdZddddfd d ZZS)StreamingAdmissionErrorz@Raised when streaming admission control cannot accept a request.r N)retry_after_mssnapshotmessager reasonr1r!r2Optional[Dict[str, float]]cs0t|||_ttd||_|pi|_dS)Nr )super__init__r4r!rr1r2)selfr3r4r1r2 __class__rrr7Cs z StreamingAdmissionError.__init__)r3r r4r r1r!r2r5)r+r,r-r.r7 __classcell__rrr9rr0@s r0c@sPeZdZdZddd Zedd d ZdddZdddZdddZ d ddZ dS)!StreamingAdmissionControllerz Lightweight admission controller for streaming requests. Keeps a hard cap on active streams (`max_inflight`) and an optional bounded wait queue (`max_queue`) with per-request wait timeout (`max_wait_ms`). max_inflightr max_queue max_wait_msr!poll_mscCsttd||_ttd||_ttd||_ttd||_|jdk|_t |_ d|_ d|_ d|_ d|_d|_d|_d|_dS)Nrr ?)rrr=r>r!r?r@enabled threadingLock_lock _inflight_waiters_total_admitted_total_queue_entries_total_queue_wait_ms_total_rejected_queue_full_total_rejected_timeout)r8r=r>r?r@rrrr7Ys   z%StreamingAdmissionController.__init__r'StreamingAdmissionController'c Cs8|tddddtddddtddddtddd dd S) N$VEENA3_STREAM_ADMISSION_MAX_INFLIGHTr)r!VEENA3_STREAM_ADMISSION_MAX_QUEUE#VEENA3_STREAM_ADMISSION_MAX_WAIT_MSr VEENA3_STREAM_ADMISSION_POLL_MSg@rA)r=r>r?r@)rr")clsrrrfrom_envos     z%StreamingAdmissionController.from_envDict[str, float]c Cs|jdkr |j|jnd}|jrdndt|jt|jt|jt|jt|jt|j t|j t|jt|t|j t|j d S)Nrr ?) rBr=r>r?r@ inflight_now waiters_nowadmitted_totalqueue_entries_totalqueue_wait_ms_avgrejected_queue_full_totalrejected_timeout_total) rIrJrBr!r=r>r?r@rFrGrHrKrL)r8 queue_avgrrr_snapshot_lockedxs"   z-StreamingAdmissionController._snapshot_lockedcCs2|j |WdS1swYdSN)rEr^)r8rrrr2s$z%StreamingAdmissionController.snapshotr#cst}d}d}|jdkr||jdnd} t}|j|jr)|j|jkrn||d}|jd7_|rKtd|jd|_|j td|7_ |j d7_ t t |t |t|jt|jt|dWdS|s|j|jkr|jd7_|}tdd |j|d d}|jd7_|j}|jd7_Wdn1swY|dur||kr|j|rtd|jd|_|jd7_|}Wdn1swYtd d |j|d t|jdIdHq) NFrr @@T)r$r&r'r(r)zstreaming admission queue full queue_full)r4r1r2z streaming admission wait timeout wait_timeout)time perf_counterr?rErBrFr=rrGrJrHr#r!r%rr>rKr^r0r@rIrLasynciosleep)r8startr&r)deadlinenowr$r2rrracquiresr    z$StreamingAdmissionController.acquirelease!Optional[StreamingAdmissionLease]NonecCsV|dus|jr dSd|_|jtd|jd|_WddS1s$wYdS)NTrra)r*rErrF)r8rlrrrreleases "z$StreamingAdmissionController.releaseN)r=rr>rr?r!r@r!)rrMrrTrr#rlrmrrn) r+r,r-r.r7 classmethodrSr^r2rkrorrrrr<Qs     :r<c@seZdZUdZdZded<dZded<dZded<dZded<dZ ded<dZ ded <dZ d ed <d Z d ed<dZ ded<dZded<dZd ed<dZd ed<dZded<dZd ed<eedZded<dZded <dS)! TTSRuntimea, Holds long-lived, per-container inference objects. Thread Safety: This class is designed for async concurrency within a single container. The vLLM engine handles internal batching and scheduling. Do NOT share across processes without proper synchronization. Nrmodelpipelinestreaming_pipelinelong_text_processorprompt_builderbicodec_decoderz Optional[Any] sr_service not_loadedr model_versionFr% is_loadedr r! load_time_msr model_path bicodec_path Optional[str]sr_checkpoint_dircudadevice)default_factoryzDict[str, List[int]]speaker_global_cachez&Optional[StreamingAdmissionController]stream_admission)r+r,r-r.rur/rvrwrxryrzr{r}r~rrrrrrdictrrrrrrrts$               rtOptional[TTSRuntime]_runtimecCstS)z&Get the current TTS runtime singleton.)rrrrr get_runtimesrr%cCstduotjS)z.Check if the runtime is initialized and ready.N)rr~rrrris_initializedsrruntimecCs0|durtd|j}|durt}||_|S)zCReturn initialized streaming admission controller for this runtime.NTTS runtime not initialized) RuntimeErrorrr<rSr controllerrrr _get_stream_admission_controllersrrTcCst}|dur iSt|S)z Return current admission state + cumulative counters. Safe to call anytime; when runtime is not initialized returns an empty dict. N)rrr2)rrrr get_streaming_admission_snapshots rcs,tstdt}t|}|IdHS)zFAcquire one streaming admission slot or raise StreamingAdmissionError.rN)rrrrrkrrrracquire_streaming_slots rrlrmrncCs(t}|dur dSt|}||dS)z:Release a previously acquired streaming slot (idempotent).N)rrro)rlrrrrrrelease_streaming_slot#s rr?Frarrrrrhf_tokengpu_memory_utilization enable_sr num_enginesmax_num_batched_tokens Optional[int] max_num_seqsenable_chunked_prefillOptional[bool]enable_prefix_cachingdisable_log_stats enforce_eagerprecompute_speaker_globalsc*Cst}|ptjdtjdd}|ptjd|}|p"tjd}|p0tjdp0tjd}|durDtjd d }|d v}td td |td|td|d|zddlm }ddl m }ddl m }ddlm}ddlm}i}t|tr|dkr||d<t| tr| dkr| |d<| durt| |d<| durt| |d<| durt| |d<| durt| |d<|dkrddlm}||}td|d |d!d"|dL||||d#|}ntd$|dL|||d%|}td&||j|d'}td(tjd)d*}|d v}||||d+}td,||||d-}d}zdd.lm} td/| |||d0}Wntyd}!ztd1|!WYd}!~!nd}!~!ww|||d2}"d}#|r|rz*dd3lm}$td4|d5|$|d6}#|#j|d7rtd8ntd9d}#Wnty}!ztd:|!WYd}!~!nd}!~!wwtj !|"d;}%|%sd<}%t|d=}&t#$}'t%||||"|||#|%d>|&|||||'d?a&td@|&dAdBtdC|%|'j'r tdD|'j(|'j)|'j*|'j+ntdE|r!|dkr!tdFt,t&n|dkr,tdGntdHzddIl-m.}(|(|%Wt&WStyKYt&WSwtyp}!zt/dJ|!ddl0})|)1t2dK|!|!d}!~!ww)Ma Initialize the TTS runtime with all components. This should be called once per container (e.g., in Modal's @modal.enter). Args: model_path: Path to Spark TTS model (env: SPARK_TTS_MODEL_PATH) bicodec_path: Path to BiCodec model (env: BICODEC_MODEL_PATH, defaults to model_path) sr_checkpoint_dir: Path to super-resolution checkpoints (env: AP_BWE_CHECKPOINT_DIR) device: Device for inference (cuda/cpu) hf_token: HuggingFace token for private models (env: HF_TOKEN) gpu_memory_utilization: vLLM GPU memory fraction (default: 0.25) enable_sr: Enable super-resolution service num_engines: Number of vLLM engine instances (default: 1, set to 2-3 for Tier 3 optimization) max_num_batched_tokens: Optional vLLM scheduler cap override. max_num_seqs: Optional vLLM concurrent sequence cap override. enable_chunked_prefill: Optional vLLM chunked prefill toggle. enable_prefix_caching: Optional vLLM prefix caching toggle. disable_log_stats: Optional vLLM internal stats log toggle. enforce_eager: Optional vLLM eager-mode toggle (disables CUDA graphs when true). precompute_speaker_globals: Whether to warm speaker global tokens at startup. If None, reads PRECOMPUTE_SPEAKER_GLOBALS env var (default: false). Returns: Initialized TTSRuntime instance Raises: RuntimeError: If initialization fails SPARK_TTS_MODEL_PATH MODEL_PATHz/models/spark_tts_4speakerBICODEC_MODEL_PATHAP_BWE_CHECKPOINT_DIRHF_TOKENHUGGING_FACE_HUB_TOKENNPRECOMPUTE_SPEAKER_GLOBALSfalse>1onyestruezInitializing TTS runtime...z Model path: z BiCodec path: z SR enabled: z, path: r) SparkTTSModel)SparkTTSPipeline)BiCodecDecoder)IndicPromptBuilderLongTextProcessorrrrrrrra)create_multi_engine_modelzLoading z vLLM engines (z.2fz GPU mem each)...)rrrgpu_memory_per_enginez#Loading SparkTTS model with vLLM...)rrrzInitializing prompt builder...) tokenizerruzInitializing BiCodec decoder...VEENA3_BICODEC_BATCHINGr)rrenable_batchingzInitializing TTS pipeline...)ruryrz)Veena3SlidingWindowPipelinez"Initializing streaming pipeline...)rury snac_decoderz"Streaming pipeline not available: rvrw)SuperResolutionServicez#Initializing super-resolution from z...)checkpoint_dir)ru.✅ Super-resolution model loaded successfullyz%Super-resolution model failed to loadz Super-resolution not available: /z spark-ttsT)rurvrwrxryrzr{r}r~rrrrrru✅ TTS runtime initialized in .0fmsz Model version: zG Streaming admission: inflight<=%d queue<=%d wait<=%.1fms poll=%.1fmsz Streaming admission: disabledz*Speaker globals startup precompute enabledzDSpeaker globals cache startup precompute skipped (multi-engine mode)z;Speaker globals cache startup precompute skipped (disabled))set_model_versionu&❌ Failed to initialize TTS runtime: z#TTS runtime initialization failed: r)3rdrrrstriplowerloggerinfoveena3modal.core.model_loaderrveena3modal.core.pipeliner veena3modal.core.bicodec_decoderr%veena3modal.processing.prompt_builderr*veena3modal.processing.long_text_processorr isinstancerr%veena3modal.core.multi_enginerr#veena3modal.core.streaming_pipeliner ImportErrorwarning!veena3modal.core.super_resolutionr load_model Exceptionpathbasenamerstripr<rSrtrrBr=r>r?r@_precompute_speaker_globalsveena3modal.api.fastapi_apprerror traceback print_excr)*rrrrrrrrrrrrrrr start_timeraw_flagrrrrr engine_kwargsrper_engine_memrury batch_rawenable_bicodec_batchingrzrvrwrerxr{rr} load_timerrrrrrinitialize_runtime,sD0                                  rc sXddlddlddlm}ddlmm}m}m}m }t dt }d||||d|dd dfdd fddz6 }|rrddl}|j} | fddWdn1slwYn|WntyYnwt |d} tj} t} t d| d| d| dddS)as Pre-compute and cache global tokens for all 12 speakers at startup. OPTIMIZATION: BiCodec streaming requires 32 "global tokens" before any audio can be emitted. These encode speaker identity via FSQ quantization. Since we have only 12 fixed speakers, we generate one short utterance per speaker at startup, capture the 32 global tokens, and cache them. During streaming, we inject cached globals via build_prefix_with_globals(), skipping the ~110ms global token pre-roll phase entirely. This runs synchronously at startup (adds ~5-10s to cold start, saves ~110ms per streaming TTFB). rN)SamplingParams) SPEAKER_MAPTRAINING_STOP_TOKEN_IDSDEFAULT_TEMPERATURE DEFAULT_TOP_K DEFAULT_TOP_Pz/Pre-computing global tokens for all speakers...zHello, this is a voice test.F) temperaturetop_ktop_p max_tokensstopskip_special_tokens speaker_namer rOptional[List[int]]csj|}d|dttd}d}jjj||d2z3dHW}|}q"6|dur2dS|jdj} d|}dd |D}t |d krQ|dd SdS) zcSsg|]}t|qSr)r).0trrr TszU_precompute_speaker_globals.._capture_globals_for_speaker.. ) ry build_prefixrrdruenginegenerateoutputstextfindalllen)rrr final_outputrequest_outputgenerated_textglobal_matches global_ids)rerr test_textrr_capture_globals_for_speakerBs&    zA_precompute_speaker_globals.._capture_globals_for_speakerc sD]F}z'|IdH}|r%t|dkr%|j|<td|ntd|WqtyK}ztd|d|WYd}~qd}~wwdS)Nrz Cached globals for z Failed to capture globals for z Error caching globals for z: )keysrrrrrr)rr r)rr rrr_run_allZs  "z-_precompute_speaker_globals.._run_allcs Sr_)runr)rrfrrms z-_precompute_speaker_globals..rzSpeaker globals cached: rz speakers in rr)rr rr)rfr vllmrveena3modal.core.constantsrrrrrrrrdget_event_loop is_runningconcurrent.futuresfuturesThreadPoolExecutorsubmitresultrun_until_completerrrr) rrrrrrt_startloop concurrentpoolelapsedcachedtotalr)rr rrfr rrr rr sD         $rcCs.|jdurddlm}||j|jd|_|jS)z>Return cached LongTextProcessor, creating it lazily if needed.Nrrr)rxrrrvrw)rrrrr_get_long_text_processorzs  r#皙?2rU?16khzrspeakerrrrrrepetition_penaltyseedoutput_sample_rate&Tuple[Optional[bytes], Dict[str, Any]]c  ststdt} ddddddd} t} z| jj||||||||dIdH\} } | r3| | t| }t|d | d <| d | d <| rd}|d ksUt t j rgt d || j du| j rd| j jnd|d kr| j r| j jrz:t}tdt| dt| | j } t|d }t|| d<d| d<d}td|ddt| dWn"ty}ztd|ddl}|WYd}~nd}~ww|| d<t| d|d}td|| d<| | fWSty}z td|d}~ww)a Generate speech audio (non-streaming). Args: text: Text to synthesize (already normalized) speaker: Internal speaker name (resolved) temperature: Sampling temperature top_k: Top-k sampling top_p: Nucleus sampling max_tokens: Maximum tokens to generate repetition_penalty: Repetition penalty seed: Random seed for reproducibility output_sample_rate: "16khz" or "48khz" (triggers super-resolution) Returns: Tuple of (audio_bytes, metrics_dict) audio_bytes: WAV audio bytes (16kHz or 48kHz, 16-bit PCM) or None if failed metrics_dict: Dictionary with timing metrics Raises: RuntimeError: If runtime not initialized rrr F>)ttfb_ms generation_mstokens_generatedaudio_duration_seconds sr_appliedr,r)rrrrrr*r+Nrr0r/48khzzz<4sI4s4sIHHIIHH4sIsRIFF$sWAVEsfmt sdata)numpystructtorchrr frombufferint16astypefloat32 from_numpy process_chunksqueezecpuclippacktobytes)rGr{nprUrVpcm_data audio_float audio_tensorupsampled_tensor upsampledupsampled_int16rJ num_channelsbits_per_sample byte_rate block_align data_size wav_headerrrrrEsD       rEr.rJc  ststdt} dddddddd} t} zt| } | |r9d| d<| j|||||||||d Id H}n| jj||||||||d Id H\}}|rS| |t| }t |d | d <| d | d<|rd}| dkr| j r| j j rz:t}t dt|dt|| j }t|d }t || d<d| d<d}t d|ddt|dWn"ty}zt d|dd l}|WYd }~nd }~ww|| d<t|d|d}td|| d<|| fWSty}z t d|d }~ww)a Generate speech with automatic text chunking for long inputs. Uses LongTextProcessor to split text and stitch audio. Args: text: Text to synthesize speaker: Internal speaker name ... (same as generate_speech) sample_rate: Internal generation sample rate (always 16000) output_sample_rate: "16khz" or "48khz" (triggers super-resolution) Returns: Tuple of (audio_bytes, metrics_dict) rrr Fr.)r/r0chunks_processedr2 text_chunkedr3r,Trp) rr)rrrrr*r+rJNr4rr0r/r5z,Applying super-resolution to chunked audio (z bytes)...r6r3r7r8r9r:r;r<r,r=r>r2z"Chunked speech generation failed: )rrrrdr# should_chunkgenerate_with_chunkingrvr?r@rr{r~rrrrErrrrrr)rr)rrrrr*r+rJr,rrFrlong_processorrGrHrIfinal_sample_raterKrLrrrMrrrgenerate_speech_chunked2s        " ruTenable_chunkingadmission_leaserelease_admission_lease2AsyncGenerator[Tuple[bytes, Dict[str, Any]], None]c Csjtstdt} | jdurtd| } | dur tIdH} t| } tt| ddp,d}ddlm }dddddt|t t| d dt t| d dpKdt t| d dpTdt t| d dp]dt | j t | j t| jd }t|d}d}d}d}z/t| }|o||}|rd|d<t| |||||||||| 2zV3dHW\}}|durt}t ||d|d<||dd}||}|t||ddkrdnd7}|dd7<|d|d<||d|d<|r||||fVq6n| j|}|r| jj|||||||||dd }n| jj||||||||dd }|2z~3dHW}i}|}t|trGt|dkrGt|dtrG|d}|d}|rO|||sTq |durqt}t ||d|d<||dd}||}|t||ddkr~dnd7}|dd7<|d|d<||d|d<||fVq 6W| rt| dSdS| rt| ww)a Generate speech audio with true streaming (yields chunks as they're generated). This is the core streaming implementation for M4. First yield includes WAV header + first PCM chunk. Subsequent yields are raw PCM chunks. Args: text: Text to synthesize (already normalized) speaker: Internal speaker name (resolved) temperature: Sampling temperature top_k: Top-k sampling top_p: Nucleus sampling max_tokens: Maximum tokens to generate repetition_penalty: Repetition penalty seed: Random seed for reproducibility enable_chunking: Enable text chunking for long inputs with voice consistency admission_lease: Optional pre-acquired admission lease. release_admission_lease: Whether to release lease before generator returns. Yields: Tuple of (audio_bytes, metrics_dict) - First yield: WAV header (44 bytes) prepended to first PCM chunk - Subsequent yields: Raw PCM chunks (int16, 16kHz) - Final yield: metrics_dict has final timing info Raises: RuntimeError: If runtime not initialized or streaming pipeline unavailable rNz Streaming pipeline not availabler$r r)create_wav_headerFr&r'r(r)) r/ chunks_sent total_bytesr2rpadmission_wait_msadmission_queuedadmission_inflight_on_grantadmission_waiters_on_grantadmission_queue_depth_on_entryadmission_max_inflightadmission_max_queueadmission_max_wait_msr`r.Trprr/)rJrmr{r=rar|r>r2) r)rr rrrrr*r+ emit_progress) r)rrrrrr*r+r)rrrrwrrr!getattrveena3modal.audio.utilsrzr%rr=r>r?rdr#rq_stream_chunked_textrr@rr)generate_speech_stream_indic_continuationgenerate_speech_stream_indicrtuplerr)rr)rrrrr*r+rvrwrxrrlrr}rzrFrfirst_chunk_timetotal_pcm_bytesrJrsneeds_chunking audio_chunk chunk_metricsrncached_globals stream_gen stream_itemstream_metricsrrrgenerate_speech_streamings*              "   rc  Csddlm} ||} | sdSd} d}ddi}t| D]\}}|d|d<|dkr\|jj|||||||| d2z!3dHW\}}| durH|rH|} | ||| dd\}}|rY||fVq86q| durtd |dd |jj|||||||| d2z3dHW}| ||| dd\}}|r||fVqy6q|jj ||| |||||| d 2z3dHW}| ||| dd\}}|r||fVq6q|r||fVdSdS) z Internal helper: stream chunked text with voice consistency. Uses global token caching from first chunk to maintain voice across chunks. r)crossfade_bytes_int16Nrorar4r%)sample_rate_hz crossfade_mszNo captured globals for chunk z, using regular streaming) r)rr rrrrr*r+) veena3modal.audio.crossfader chunk_text enumeraterw(generate_speech_stream_indic_first_chunkrrrr)rrsrr)rrrrr*r+rJrchunkscaptured_globalsprevious_chunk_tailrirrGr to_emitrrrrUs            r)r)r r r rrrrr)r )r r r r!rr!rr!)rr)rr%)rrrr<rprqrr)NNNrNrFraNNNNNNN) rrrrrrrr rrrr!rr%rrrrrrrrrrrrrrrrrrt)rrtrrn)rrt)r$r%rUr&r'Nr()rr r)r rr!rrrr!rrr*r!r+rr,r rr-)rGrOrrO)r$r%rUr&r'Nr.r()rr r)r rr!rrrr!rrr*r!r+rrJrr,r rr-) r$r%rUr&r'NTNT)rr r)r rr!rrrr!rrr*r!r+rrvr%rwrmrxr%rry)rrtrr r)r rr!rrrr!rrr*r!r+rrJrrry)-r. __future__rrsysrdrBrCrf dataclassesrrtypingrrrrr r getLoggerr+rrr"r#rr0r<rtrr/rrrrrrrrr#rNrErurrrrrrs        "       u Z jE o 8