o ϯi#@sddlmZddlZddlZddlmZddlmmZ ddl m Z ddl m Z GdddejZGdddejZed krXed d Zed d dZeeZeedjdSdS))UnionN) rearrange) NormConv1dcsNeZdZdZdededeffdd ZddZd d Zd d Zd dZ Z S)VectorQuantizea Implementation of VQ similar to Karpathy's repo: https://github.com/karpathy/deep-vector-quantization Additionally uses following tricks from Improved VQGAN (https://arxiv.org/pdf/2110.04627.pdf): 1. Factorized codes: Perform nearest neighbor lookup in low-dimensional space for improved codebook usage 2. l2-normalized codes: Converts euclidean distance to cosine similarity which improves training stability input_dim codebook_size codebook_dimcsHt||_||_t||dd|_t||dd|_t|||_ dS)N) kernel_size) super__init__rrrin_projout_projnn Embeddingcodebook)selfrrr __class__F/home/ubuntu/.local/lib/python3.10/site-packages/dacvae/nn/quantize.pyr s zVectorQuantize.__init__cCs|||}||\}}tj||ddddg}tj||ddddg}|||}||}|||||fS)aQuantized the input tensor using a fixed codebook and returns the corresponding codebook vectors Parameters ---------- z : Tensor[B x D x T] Returns ------- Tensor[B x D x T] Quantized continuous representation of input Tensor[1] Commitment loss to train encoder to predict vectors closer to codebook entries Tensor[1] Codebook loss to update the codebook Tensor[B x T] Codebook indices (quantized discrete representation of input) Tensor[B x D x T] Projected latents (continuous representation of input before quantization) none) reductionr )r decode_latentsFmse_lossdetachmeanr)rzz_ez_qindicescommitment_loss codebook_lossrrrforward#s  zVectorQuantize.forwardcCst||jjSN)r embeddingrweightrembed_idrrr embed_codeIszVectorQuantize.embed_codecCs||ddS)Nr r)r+ transposer)rrr decode_codeLszVectorQuantize.decode_codecCst|d}|jj}t|}t|}|djdddd|||djddd}t| ddd| dd}| |}||fS) Nzb d t -> (b t) drr T)keepdimz (b t) -> b tr)b) rrr(r normalizepowsumtmaxsizer-)rlatents encodingsrdistr"r!rrrrOs     zVectorQuantize.decode_latents) __name__ __module__ __qualname____doc__intr r%r+r-r __classcell__rrrrrs  &rc szeZdZdZ     ddeded ed eeefd ef fd d ZddefddZ de j fddZ de j fddZ ZS)ResidualVectorQuantizezg Introduced in SoundStream: An end2end neural audio codec https://arxiv.org/abs/2107.03312  r n_codebooksrrquantizer_dropoutcshtttrfddt|D||_|_|_t fddt|D|_ ||_ dS)Ncsg|]}qSrr).0_rrr rsz3ResidualVectorQuantize.__init__..csg|] }t|qSr)r)rGirrrrrrJys) r r isinstancer=rangerErrr ModuleList quantizersrF)rrrErrrFrrLrr hs   zResidualVectorQuantize.__init__N n_quantizerscCsrd}|}d}d}g}g}|dur|j}|jrLt|jdf|jd}td|jd|jdf} t|jd|j} | d| |d| <||j }t |j D]R\} } |jdur`| |kr`nD| |\} }}}}tj |jdf| |j d|k}|| |ddddf}|| }||| 7}||| 7}||||qQtj|dd}tj|dd}|||||fS)aQuantized the input tensor using a fixed set of `n` codebooks and returns the corresponding codebook vectors Parameters ---------- z : Tensor[B x D x T] n_quantizers : int, optional No. of quantizers to use (n_quantizers < self.n_codebooks ex: for quantizer dropout) Note: if `self.quantizer_dropout` is True, this argument is ignored when in training mode, and a random number of quantizers is used. Returns ------- dict A dictionary with the following keys: "z" : Tensor[B x D x T] Quantized continuous representation of input "codes" : Tensor[B x N x T] Codebook indices for each codebook (quantized discrete representation of input) "latents" : Tensor[B x N*D x T] Projected latents (continuous representation of input before quantization) "vq/commitment_loss" : Tensor[1] Commitment loss to train encoder to predict vectors closer to codebook entries "vq/codebook_loss" : Tensor[1] Codebook loss to update the codebook rNr F) fill_valuedevicedim)rEtrainingtorchonesshaperandintr=rFtorS enumeraterPfullrappendstackcat)rrrQr!residualr#r$codebook_indicesr6dropout n_dropoutrK quantizerz_q_icommitment_loss_icodebook_loss_i indices_iz_e_imaskcodesrrrr%s>   zResidualVectorQuantize.forwardrlcCsxd}g}|jd}t|D]$}|j||dd|ddf}|||j||}||}q |tj|dd|fS)a?Given the quantized codes, reconstruct the continuous representation Parameters ---------- codes : Tensor[B x N x T] Quantized discrete representation of input Returns ------- Tensor[B x D x T] Quantized continuous representation of input rDr NrT)rYrNrPr-r^rrWr`)rrlr!z_prErKz_p_irfrrr from_codess   "  z!ResidualVectorQuantize.from_codesr6c Csd}g}g}tdgdd|jD}t||jdkdjdddd}t|D]8}||||d}} |j||dd|| ddf\} } || || |j| | } || }q+|t j |ddt j |ddfS) aGiven the unquantized latents, reconstruct the continuous representation after quantization. Parameters ---------- latents : Tensor[B x N x T] Continuous representation of input after projection Returns ------- Tensor[B x D x T] Quantized representation of full-projected space Tensor[B x D x T] Quantized representation of latent space rcSsg|]}|jqSrrI)rGqrrrrJsz7ResidualVectorQuantize.from_latents..r T)axiskeepdimsNrT) npcumsumrPwhererYr4rNrr^rrWr`r_) rr6r!rmrldimsrErKjkrncodes_irfrrr from_latentss  *   z#ResidualVectorQuantize.from_latents)r@rArBrCrDr&)r9r:r;r<r=rlistfloatr r%rWTensorrorzr>rrrrr?bs* Ir?__main__T)rFr@Pr6)typingrnumpyrsrWtorch.nnrtorch.nn.functional functionalreinopsrdacvae.nn.layersrModulerr?r9rvqrandnxyprintrYrrrrs    T"