o mi)@sldZddlmZmZddlZddlZddlZddlmZddl m Z eGdddZ Gd d d ej Z dS) z)Residual vector quantizer implementation.) dataclassfieldN)nn)ResidualVectorQuantizationc@sReZdZUejed<ejed<ejed<dZejejed<e e dZ e ed<dS)QuantizedResult quantizedcodes bandwidthNpenalty)default_factorymetrics) __name__ __module__ __qualname__torchTensor__annotations__r tpOptionalrdictr rr?/home/ubuntu/SpeechTokenizer/speechtokenizer/quantization/vq.pyrs   rcseZdZdZ       dd ed ed ed ed ededeffdd Zd dej d e j ede j e de fddZd dej d e j ede j edej fddZd!dej dedej fddZZS)"ResidualVectorQuantizeraResidual Vector Quantizer. Args: dimension (int): Dimension of the codebooks. n_q (int): Number of residual vector quantizers used. bins (int): Codebook size. decay (float): Decay for exponential moving average over the codebooks. kmeans_init (bool): Whether to use kmeans to initialize the codebooks. kmeans_iters (int): Number of iterations used for kmeans initialization. threshold_ema_dead_code (int): Threshold for dead code expiration. Replace any codes that have an exponential moving average cluster size less than the specified threshold with randomly selected vector from the current batch. Gz?T2 dimensionn_qbinsdecay kmeans_init kmeans_itersthreshold_ema_dead_codec s^t||_||_||_||_||_||_||_t |j|j|j|j|j|j|jd|_ dS)N)dim codebook_sizenum_quantizersr#r$r%r&) super__init__r!r r"r#r$r%r&rvq)selfr r!r"r#r$r%r& __class__rrr+)s"  z ResidualVectorQuantizer.__init__NxlayersreturncCsd|r|n|j}|rt||krtdt|d|jd|j|||d\}}}}||t||fS)aResidual vector quantization on the given input tensor. Args: x (torch.Tensor): Input tensor. n_q (int): Number of quantizer used to quantize. Default: All quantizers. layers (list): Layer that need to return quantized. Defalt: None. Returns: QuantizedResult: The quantized (or approximately quantized) representation with the associated numbert quantizers and layer quantized required to return. zLast layer index in layers: A z!. Number of quantizers in RVQ: B z. A must less than B.)r!r1)r!max ValueErrorr,rmean)r-r0r!r1rr commit_lossquantized_listrrrforwardEs  zResidualVectorQuantizer.forwardstcCs,|r|n|j}|p d}|jj|||d}|S)aEncode a given input tensor with the specified sample rate at the given bandwidth. The RVQ encode method sets the appropriate number of quantizer to use and returns indices for each quantizer. Args: x (torch.Tensor): Input tensor. n_q (int): Number of quantizer used to quantize. Default: All quantizers. st (int): Start to encode input from which layers. Default: 0. r)r!r9)r!r,encode)r-r0r!r9r rrrr:Ws zResidualVectorQuantizer.encoderr cCs|jj||d}|S)zDecode the given codes to the quantized representation. Args: codes (torch.Tensor): Input indices for each quantizer. st (int): Start to decode input codes from which layers. Default: 0. )r9)r,decode)r-r r9rrrrr;eszResidualVectorQuantizer.decode)rrrrTrr)NN)r)rrr__doc__intfloatboolr+rrrrlistrr8r:r; __classcell__rrr.rrs6*,$r)r< dataclassesrrmathtypingrrrcore_vqrrModulerrrrrs