o ۷iS=@sddlZddlZddlZddlmZmZmZmZddl m Z ddl Z ddl Z ddlZddlZddlmZddlmZmZmZddlmZmZmZmZeee jeeee jfZGdd d ZdS) N)ListOptionalTupleUnion)urlparse) pad_sequence) AutoConfigAutoFeatureExtractor AutoModel)Qwen3TTSTokenizerV1ConfigQwen3TTSTokenizerV1ModelQwen3TTSTokenizerV2ConfigQwen3TTSTokenizerV2Modelc@seZdZdZddZededdfddZdedefd d Z dedefd d Z d ede fddZ dede dejfddZdedee deejfddZ  d)dedee defddZdeeeje ffddZdefdd Zde fd!d"Zde fd#d$Zde fd%d&Zde fd'd(ZdS)*Qwen3TTSTokenizera. A wrapper for Qwen3 TTS Tokenizer 25Hz/12Hz with HuggingFace-style loading. - from_pretrained(): loads speech tokenizer model via AutoModel and feature_extractor via AutoFeatureExtractor. - encode(): supports wav path(s), base64 audio string(s), numpy array(s). - decode(): accepts either the raw model encode output, or a minimal dict/list-of-dicts. Notes: - For numpy array input, you must pass `sr` so the audio can be resampled to model sample rate. - Returned audio is float32 numpy arrays and the output sample rate. cCsd|_d|_d|_d|_dSN)modelfeature_extractorconfigdeviceselfr\/home/ubuntu/vllm_env/lib/python3.10/site-packages/qwen_tts/inference/qwen3_tts_tokenizer.py__init__9s zQwen3TTSTokenizer.__init__pretrained_model_name_or_pathreturncKs|}tdtttttdttttt||_ tj|fi||_ |j j |_ t |j dd|_ |j durZz t|j j |_ W|StyYt d|_ Y|Sw|S)a Initialize tokenizer with HuggingFace `from_pretrained` style. Args: pretrained_model_name_or_path (str): HuggingFace repo id or local directory. **kwargs (Any): Forwarded to `AutoModel.from_pretrained(...)` directly. Typical examples: device_map="cuda:0", dtype=torch.bfloat16, attn_implementation="eager". Returns: Qwen3TTSTokenizer: Initialized instance with `model`, `feature_extractor`, `config`. qwen3_tts_tokenizer_25hzqwen3_tts_tokenizer_12hzrNcpu)rregisterr r r rrr from_pretrainedrrrgetattrrnext parameters StopIterationtorch)clsrkwargsinstrrrr!?s$        z!Qwen3TTSTokenizer.from_pretrainedscCs2|drdSd|vrd|vrt|dkrdSdS)Nz data:audioT/\F) startswithlen)rr*rrr_is_probably_base64es z%Qwen3TTSTokenizer._is_probably_base64cCs4zt|}|jdvot|jWStyYdSw)N)httphttpsF)rschemeboolnetloc Exception)rr*urrr_is_urlms  zQwen3TTSTokenizer._is_urlb64cCs0d|vr|dr|ddd}t|S)N,zdata:)stripr.splitbase64 b64decode)rr9rrr_decode_base64_to_wav_bytests z-Qwen3TTSTokenizer._decode_base64_to_wav_bytesx target_src Cs"||r?tj| }|}Wdn1swYt|}tj|ddd\}}Wdn1s9wYn4||ri| |}t|}tj|ddd\}}Wdn1scwYn t j |ddd\}}|j dkrt j|dd }||krt j|||d }|t jS) ai Load audio from wav path or base64 string, then resample to target_sr. Args: x (str): A wav file path, or a base64 audio string (raw or data URL). target_sr (int): Target sampling rate. Returns: np.ndarray: 1-D float32 waveform at target_sr. Nfloat32F)dtype always_2dT)srmonor;axisyorig_srrB)r8urllibrequesturlopenreadioBytesIOsfr0r@librosaloadndimnpmeanresampleastyperC) rrArBresp audio_bytesfaudiorF wav_bytesrrr load_audiozs(        zQwen3TTSTokenizer.load_audioaudiosrFcstjjt|ttjfr|g}t|dkrgSt|dtr*fdd|DS|dur2tdg}|D]5}t|tjsBt d|j dkrNtj |dd }t|krbt j |tjt|d }||tjq6|S) ai Normalize all supported input types into a list of 1-D numpy float32 waveforms at `self.feature_extractor.sampling_rate`. Args: audios (AudioInput): - str: wav path OR base64 audio string - np.ndarray: raw waveform (sr must be provided) - list[str] / list[np.ndarray] sr (Optional[int]): Sampling rate for raw numpy input. Required if input is np.ndarray or list[np.ndarray]. Returns: List[np.ndarray]: List of float32 waveforms resampled to model input SR. rcsg|] }j|dqS))rB)ra.0rArrBrr z=Qwen3TTSTokenizer._normalize_audio_inputs..NzIFor numpy waveform input, you must provide `sr` (original sampling rate).zNMixed input types are not supported. Use all paths/base64 or all numpy arrays.r;rHrIrK)intr sampling_rate isinstancestrrXndarrayr/ ValueError TypeErrorrWrYrUrZr[rCappend)rrbrFoutarrer_normalize_audio_inputss&     z)Qwen3TTSTokenizer._normalize_audio_inputsNT return_dictcCs|j||d}|j|t|jjdd}||j|jj}t |jj |d d|d d|d}Wd|S1sAwY|S) a Batch-encode audio into discrete codes (and optional conditioning, depending on 25Hz/12Hz). Args: audios (AudioInput): Supported forms: - np.ndarray: waveform (requires sr) - list[np.ndarray]: waveforms (requires sr) - str: wav path OR base64 audio string - list[str]: wav paths and/or base64 strings sr (Optional[int], default=None): Original sampling rate for numpy waveform input. return_dict (bool, default=True): Forwarded to model.encode(...). If True, returns ModelOutput. Returns: 25Hz: Qwen3TTSTokenizerV1EncoderOutput (if return_dict=True) with fields: - audio_codes: List[torch.LongTensor] each (codes_len,) - xvectors: List[torch.FloatTensor] each (xvector_dim,) - ref_mels: List[torch.FloatTensor] each (mel_len, mel_dim) 12Hz: Qwen3TTSTokenizerV2EncoderOutput (if return_dict=True) with fields: - audio_codes: List[torch.LongTensor] each (codes_len, num_quantizers) If return_dict=False, returns the raw tuple from model.encode. )rFpt) raw_audiorireturn_tensors input_valuesr; padding_maskrsN) rrrrhritorrrDr&inference_modeencodesqueeze)rrbrFrswavsinputsencrrrr|s"!     zQwen3TTSTokenizer.encodec s|j}dddt|dr|j}t|dd}t|dd}nEt|tr5|d}|dd}|dd}n/t|tr`dd|D}d|d vrNd d|Dnd}d|d vr]d d|Dnd}nt d t|t j r|}| d krx| d }n | dkr| d }||j}nfdd|D}t|ddd|j}t |dkr4|dus|durtdt|t j r|}| d kr| d }||j|jj}nfdd|D}t j|d d|j|jj}t|t j r |} | dkr| d } | |j|jj} nfdd|D}t|dd d|j|jj} |jj||| dd} | j} n|dkrE|jj|dd} | j} ntd|Wdn 1sWwYdd| D} | t|jfS)a4 Decode back to waveform. Usage: 1) Pass the raw output of `encode(...)` directly (recommended). - 25Hz: expects fields audio_codes, xvectors, ref_mels - 12Hz: expects field audio_codes 2) Pass a dict or list[dict] (minimal form) for custom pipelines: - 25Hz dict keys: {"audio_codes", "xvectors", "ref_mels"} - 12Hz dict keys: {"audio_codes"} Values can be torch tensors or numpy arrays. Args: encoded (Any): - ModelOutput returned by `encode()`, OR - dict, OR - list[dict] Returns: Tuple[List[np.ndarray], int]: - wavs: list of 1-D float32 numpy arrays - sample_rate: int, model output sampling rate NcSs:t|tjr|St|}t|}|dur||}|Sr)rjr&TensorrXasarray from_numpyrz)rArDtrrr _to_tensor s    z,Qwen3TTSTokenizer.decode.._to_tensor audio_codesxvectorsref_melscSg|]}|dqS)rrrderrrrf5z,Qwen3TTSTokenizer.decode..rcSr)rrrrrrrf6rcSr)rrrrrrrf7rz?`encoded` must be an encode output, a dict, or a list of dicts.r;r cg|] }|tjdqS)rD)r&long)rdcrrrrfHrgTrH) batch_first padding_valuerz/25Hz decode requires `xvectors` and `ref_mels`.crrr&rCrcrrrrfVrg)dimcrrr)rdmrrrrf_rgryrzUnknown model type: cSs$g|]}|tjqSr)rzr&rCdetachrnumpy)rdwrrrrfls$r)rget_model_typehasattrrr"rjdictgetlistrnr&rr unsqueezerzrrr{rmrDstackdecode audio_valuesrhget_output_sample_rate) rencoded model_typeaudio_codes_list xvectors_list ref_mels_listraudio_codes_paddedxvectors_batchref_mels_paddeddec wav_tensorsr~rrrrsh                      !zQwen3TTSTokenizer.decodecCs |jS)z Get the underlying tokenizer model type. Returns: str: Model type string from `self.model.config.model_type` (e.g. "qwen3_tts_tokenizer_25hz" / "qwen3_tts_tokenizer_12hz"). )rrrrrrros z Qwen3TTSTokenizer.get_model_typecCt|jS)z} Get the expected input sample rate for encoding. Returns: int: Input sample rate (Hz). )rhrget_input_sample_raterrrrryz'Qwen3TTSTokenizer.get_input_sample_ratecCr)z Get the output sample rate for decoded waveforms. Returns: int: Output sample rate (Hz). )rhrrrrrrrrz(Qwen3TTSTokenizer.get_output_sample_ratecCr)z Get the encoder downsample rate (waveform samples per code step). Returns: int: Encode downsample rate. )rhrget_encode_downsample_raterrrrrrz,Qwen3TTSTokenizer.get_encode_downsample_ratecCr)z Get the decoder upsample rate (waveform samples per code step). Returns: int: Decode upsample rate. )rhrget_decode_upsample_raterrrrrrz*Qwen3TTSTokenizer.get_decode_upsample_rate)NT)__name__ __module__ __qualname____doc__r classmethodrkr!r4r0r8bytesr@rhrXrlra AudioInputrrrrr|rrrrrrrrrrrr,sN % & 3 3 l    r) r>rRurllib.requestrNtypingrrrr urllib.parserrUrrX soundfilerTr&torch.nn.utils.rnnr transformersrr r corer r rrrkrlrrrrrrs(