o
    
j`                     @   s   d dl mZ d dlZd dlmZ d dlm  mZ	 d dlmZ d dl
mZ d dlmZmZ d dlmZ d dlmZ G dd	 d	ejZG d
d dejZG dd dejZdS )    )ListN)nn)
checkpoint)	OutputnetOverflowUtils)Prenet)sequence_maskc                       s   e Zd ZdZ	d/dedededededed	ed
edededee de	dedef fddZ
dd Zedd Zdd Zdd Zedd Zedd Zdd  Zed0d"d#Ze d$ejd%ejd&ed'ed(ef
d)d*Ze d+d, Zed-d. Z  ZS )1	NeuralHMMu  Autoregressive left to right HMM model primarily used in "Neural HMMs are all you need (for high-quality attention-free TTS)"

    Paper::
        https://arxiv.org/abs/2108.13320

    Paper abstract::
        Neural sequence-to-sequence TTS has achieved significantly better output quality than statistical speech synthesis using
        HMMs. However, neural TTS is generally not probabilistic and uses non-monotonic attention. Attention failures increase
        training time and can make synthesis babble incoherently. This paper describes how the old and new paradigms can be
        combined to obtain the advantages of both worlds, by replacing attention in neural TTS with an autoregressive left-right
        no-skip hidden Markov model defined by a neural network. Based on this proposal, we modify Tacotron 2 to obtain an
        HMM-based neural TTS model with monotonic alignment, trained to maximise the full sequence likelihood without
        approximation. We also describe how to combine ideas from classical and contemporary TTS for best results. The resulting
        example system is smaller and simpler than Tacotron 2, and learns to speak with fewer iterations and less data, whilst
        achieving comparable naturalness prior to the post-net. Our approach also allows easy control over speaking rate.

    Args:
        frame_channels (int): Output dimension to generate.
        ar_order (int): Autoregressive order of the model. In ablations of Neural HMM it was found that more autoregression while giving more variation hurts naturalness of the synthesised audio.
        deterministic_transition (bool): deterministic duration generation based on duration quantiles as defiend in "S. Ronanki, O. Watts, S. King, and G. E. Henter, “Medianbased generation of synthetic speech durations using a nonparametric approach,” in Proc. SLT, 2016.". Defaults to True.
        encoder_dim (int): Channels of encoder input and character embedding tensors. Defaults to 512.
        prenet_type (str): `original` or `bn`. `original` sets the default Prenet and `bn` uses Batch Normalization version of the Prenet.
        prenet_dim (int): Dimension of the Prenet.
        prenet_n_layers (int): Number of layers in the Prenet.
        prenet_dropout (float): Dropout probability of the Prenet.
        prenet_dropout_at_inference (bool): If True, dropout is applied at inference time.
        memory_rnn_dim (int): Size of the memory RNN to process output of prenet.
        outputnet_size (List[int]): Size of the output network inside the neural HMM.
        flat_start_params (dict): Parameters for the flat start initialization of the neural HMM.
        std_floor (float): Floor value for the standard deviation of the neural HMM. Prevents model cheating by putting point mass and getting infinite likelihood at any datapoint.
        use_grad_checkpointing (bool, optional): Use gradient checkpointing to save memory. Defaults to True.
    Tframe_channelsar_orderdeterministic_transitionencoder_dimprenet_type
prenet_dimprenet_n_layersprenet_dropoutprenet_dropout_at_inferencememory_rnn_dimoutputnet_sizeflat_start_params	std_flooruse_grad_checkpointingc                    s   t    | _| _| _| _|
 _| _t  _	t
  _|dks*J d| | _t|| |||	 fddt|D dd _tj||
d _t||
|||| _ dt|d	 d S )
Nr   z)AR order must be greater than 0 provided c                    s   g | ]} j qS  )r   ).0_selfr   U/home/kuhnn/.local/lib/python3.10/site-packages/TTS/tts/layers/overflow/neural_hmm.py
<listcomp>U   s    z&NeuralHMM.__init__.<locals>.<listcomp>F)in_featuresr   r   dropout_at_inferenceout_featuresbias)
input_sizehidden_size	go_tokens   )super__init__r
   r   r   r   r   r   TransitionModeltransition_modelEmissionModelemission_modelr   rangeprenetr   LSTMCell
memory_rnnr   
output_netregister_buffertorchzeros)r   r
   r   r   r   r   r   r   r   r   r   r   r   r   r   	__class__r   r   r(   0   s0   
zNeuralHMM.__init__c              	   C   s  |j \}}}t|}|ddd}| |}	| ||\}
}}}| |}| || j|\}}t	|D ]}| 
||||\}}| jrQ| jrQt| j||\}}}n	| ||\}}}|dkro|	| |dddf ||| }n | |dd|f |||| |dd|d ddf || }tj|dd|
dd|f< ||
dd|f d |dd|ddf< ||dd|f< ||  q4| ||
|\}
}| ||||}tj|
dd| }||||fS )aV  HMM forward algorithm for training uses logarithmic version of Rabiner (1989) forward algorithm.

        Args:
            inputs (torch.FloatTensor): Encoder outputs
            inputs_len (torch.LongTensor): Encoder output lengths
            mels (torch.FloatTensor): Mel inputs
            mel_lens (torch.LongTensor): Length of mel inputs

        Shapes:
            - inputs: (B, T, D_out_enc)
            - inputs_len: (B)
            - mels: (B, D_mel, T_mel)
            - mel_lens: (B)

        Returns:
            log_prob (torch.FloatTensor): Log probability of the sequence
        r      r&   Ndim)shaper3   maxpermute_initialize_log_state_priors'_initialize_forward_algorithm_variables_add_go_token_init_lstm_statesr   r-   _process_ar_timestepr   trainingr   r1   r,   r*   	logsumexp	unsqueezeappenddetach_mask_lengths#get_absorption_state_scaling_factorsum)r   inputs
inputs_lenmelsmel_lens
batch_sizeNr   T_maxlog_state_priorslog_clog_alpha_scaledtransition_matrixmeans	ar_inputsh_memoryc_memorytmeanstdtransition_vectorlog_alpha_tempsum_final_log_c	log_probsr   r   r   forward^   s6   


",zNeuralHMM.forwardc                 C   s*   t | }|| }|d}|| }||fS )a  
        Mask the lengths of the forward variables so that the variable lenghts
        do not contribute in the loss calculation
        Args:
            mel_inputs (torch.FloatTensor): (batch, T, frame_channels)
            mel_inputs_lengths (torch.IntTensor): (batch)
            log_c (torch.FloatTensor): (batch, T)
        Returns:
            log_c (torch.FloatTensor) : scaled probabilities (batch, T)
            log_alpha_scaled (torch.FloatTensor): forward probabilities (batch, T, N)
        r7   )r   rD   )rM   rR   rS   
mask_log_cmask_log_alpha_scaledr   r   r   rG      s
   
zNeuralHMM._mask_lengthsc                 C   sF   |dd||| j  f d}| |}| |||f\}}||fS )a  
        Process autoregression in timestep
        1. At a specific t timestep
        2. Perform data dropout if applied (we did not use it)
        3. Run the autoregressive frame through the prenet (has dropout)
        4. Run the prenet output through the post prenet rnn

        Args:
            t (int): mel-spec timestep
            ar_inputs (torch.FloatTensor): go-token appended mel-spectrograms
                - shape: (b, D_out, T_out)
            h_post_prenet (torch.FloatTensor): previous timestep rnn hidden state
                - shape: (b, memory_rnn_dim)
            c_post_prenet (torch.FloatTensor): previous timestep rnn cell state
                - shape: (b, memory_rnn_dim)

        Returns:
            h_post_prenet (torch.FloatTensor): rnn hidden state of the current timestep
            c_post_prenet (torch.FloatTensor): rnn cell state of the current timestep
        Nr&   )r   flattenr.   r0   )r   rY   rV   rW   rX   prenet_inputmemory_inputsr   r   r   rA      s    
zNeuralHMM._process_ar_timestepc                 C   sL   |j \}}}| jd|| j| j}tj||fddddd|f }|S )zAppend the go token to create the autoregressive input
        Args:
            mel_inputs (torch.FloatTensor): (batch_size, T, n_mel_channel)
        Returns:
            ar_inputs (torch.FloatTensor): (batch_size, T, n_mel_channel)
        r   r&   r8   N)r:   r%   rD   expandr   r
   r3   cat)r   
mel_inputsrN   Tr   r%   rV   r   r   r   r?      s   "zNeuralHMM._add_go_tokenc           	      C   sH   | j \}}}| |||f}| ||}| |||f}g }||||fS )ao  Initialize placeholders for forward algorithm variables, to use a stable
                version we will use log_alpha_scaled and the scaling constant

        Args:
            mel_inputs (torch.FloatTensor): (b, T_max, frame_channels)
            N (int): number of states
        Returns:
            log_c (torch.FloatTensor): Scaling constant (b, T_max)
        )r:   	new_zeros)	rh   rO   brP   r   rS   rR   rT   rU   r   r   r   r>      s   z1NeuralHMM._initialize_forward_algorithm_variablesc                 C   s   | | || | |fS )a  
        Initialize Hidden and Cell states for LSTM Cell

        Args:
            batch_size (Int): batch size
            hidden_state_dim (Int): dimensions of the h and c
            device_tensor (torch.FloatTensor): useful for the device and type

        Returns:
            (torch.FloatTensor): shape (batch_size, hidden_state_dim)
                can be hidden state for LSTM
            (torch.FloatTensor): shape (batch_size, hidden_state_dim)
                can be the cell state for LSTM
        )rj   )rN   hidden_state_dimdevice_tensorr   r   r   r@      s   

zNeuralHMM._init_lstm_statesc                 C   s   t |}|jd }t||d}|d dd|d}t |d|d}	|	| t	d }	t |d|d}
t 
|
}t|}| ||j}|| t	d }|	| }|jt |jjd}t j|dd}|S )a  Returns the final scaling factor of absorption state

        Args:
            mels_len (torch.IntTensor): Input size of mels to
                    get the last timestep of log_alpha_scaled
            log_alpha_scaled (torch.FloatTEnsor): State probabilities
            text_lengths (torch.IntTensor): length of the states to
                    mask the values of states lengths
                (
                    Useful when the batch has very different lengths,
                    when the length of an observation is less than
                    the number of max states, then the log alpha after
                    the state value is filled with -infs. So we mask
                    those values so that it only consider the states
                    which are needed for that length
                )
            transition_vector (torch.FloatTensor): transtiion vector for each state per timestep

        Shapes:
            - mels_len: (batch_size)
            - log_alpha_scaled: (batch_size, N, T)
            - text_lengths: (batch_size)
            - transition_vector: (batch_size, N, T)

        Returns:
            sum_final_log_c (torch.FloatTensor): (batch_size)

        r7   )max_lenr&   inf)minr8   )r3   r;   r:   r   rD   rf   gathersqueezemasked_fillfloatsigmoidr   log_clampedget_mask_for_last_itemdeviceclampfinfodtyperq   rC   )r   mels_lenrS   rK   r\   rO   max_inputs_lenstate_lengths_masklast_log_alpha_scaled_indexlast_log_alpha_scaledlast_transition_vectorlast_transition_probability log_probability_of_transitioning!last_transition_probability_indexfinal_log_cr^   r   r   r   rH     s$   



z-NeuralHMM.get_absorption_state_scaling_factorNc                 C   sL   t |  }|du rt jd||dnt jd||d}|| dd k}|S )aq  Returns n-1 mask for the last item in the sequence.

        Args:
            lengths (torch.IntTensor): lengths in a batch
            device (str, optional): Defaults to "cpu".
            out_tensor (torch.Tensor, optional): uses the memory of a specific tensor.
                Defaults to None.

        Returns:
            - Shape: :math:`(b, max_len)`
        Nr   )ry   )outr&   )r3   r;   itemarangerD   )lengthsry   
out_tensorrn   idsmaskr   r   r   rx   >  s
   &z NeuralHMM.get_mask_for_last_itemrJ   
input_lenssampling_tempmax_sampling_timeduration_thresholdc                 C   s   |j d }g g g g g d}t|D ]=}| |||d  || |||\}	}
}}|d |	 |d |	j d  |d |
 |d | |d | qtjjj|d d	d
|d< tj	|d |j
|jd|d< |S )a  Inference from autoregressive neural HMM

        Args:
            inputs (torch.FloatTensor): input states
                - shape: :math:`(b, T, d)`
            input_lens (torch.LongTensor): input state lengths
                - shape: :math:`(b)`
            sampling_temp (float): sampling temperature
            max_sampling_temp (int): max sampling temperature
            duration_threshold (float): duration threshold to switch to next state
                - Use this to change the spearking rate of the synthesised audio
        r   )hmm_outputshmm_outputs_len
alignmentsinput_parametersoutput_parametersr&   r   r   r   r   r   T)batch_first)r|   ry   )r:   r-   samplerE   r   utilsrnnpad_sequencer3   tensorr|   ry   )r   rJ   r   r   r   r   rk   outputsineural_hmm_outputsstates_travelledr   r   r   r   r   	inferenceR  s*   

zNeuralHMM.inferencec                 C   s  g g d}}}d}	| |	 | jdd| j| j}
| d| j|
\}}g }g }d}	 | |
	dd}| 
|d||f\}}|dd|	f d}| ||\}}}t|	 }t|	  }| |
|	g | |||g | jj|||d}tj|
|fddddddf }
| |	  t||f}||9 }| js|dd  }n||k }|r|	d7 }	d}| |	 |	|ks|r||d krn|d7 }q-tj|ddt||||fS )aD  Samples an output from the parameter models

        Args:
            inputs (torch.FloatTensor): input states
                - shape: :math:`(1, T, d)`
            input_lens (torch.LongTensor): input state lengths
                - shape: :math:`(1)`
            sampling_temp (float): sampling temperature
            max_sampling_time (int): max sampling time
            duration_threshold (float): duration threshold to switch to next state

        Returns:
            outputs (torch.FloatTensor): Output Observations
                - Shape: :math:`(T, output_dim)`
            states_travelled (list[int]): Hidden states travelled
                - Shape: :math:`(T)`
            input_parameters (list[torch.FloatTensor]): Input parameters
            output_parameters (list[torch.FloatTensor]): Output parameters
        r   r&   TN)r   r8   )rE   r%   rD   rf   r   r
   r@   r   r.   rc   r0   rs   r1   r3   rv   r,   r   rg   r   multinomialr   stackFone_hot
new_tensor)r   rJ   r   r   r   r   r   r   rY   current_staterd   rW   rX   input_parameter_valuesoutput_parameter_valuesquantilememory_inputz_trZ   r[   r\   transition_probabilitystaying_probabilityx_trT   switchr   r   r   r     sL   
"
)zNeuralHMM.samplec                 C   s*   | j d }| |gtd }d|d< |S )zCreates the log pi in forward algorithm.

        Args:
            text_embeddings (torch.FloatTensor): used to create the log pi
                    on current device

        Shapes:
            - text_embeddings: (B, T, D_out_enc)
        r&   rp   g        r   )r:   new_fullru   )text_embeddingsrO   rQ   r   r   r   r=     s   
z&NeuralHMM._initialize_log_state_priors)TN)__name__
__module____qualname____doc__intboolstrru   r   dictr(   r`   staticmethodrG   rA   r?   r>   r@   rH   rx   r3   inference_modeFloatTensor
LongTensorr   r   r=   __classcell__r   r   r5   r   r	      sx    0	
.@
 

8.
Pr	   c                   @   s   e Zd ZdZdd ZdS )r)   zvTransition Model of the HMM, it represents the probability of transitioning
    form current state to all other statesc                 C   s   t |}t | }t|}t|}|| }|| }	|	jddd}	td |	dddf< t|}
tjt j||	fdddd}|	|
 td }|S )aS  
        product of the past state with transitional probabilities in log space

        Args:
            log_alpha_scaled (torch.Tensor): Multiply previous timestep's alphas by
                        transition matrix (in log domain)
                - shape: (batch size, N)
            transition_vector (torch.tensor): transition vector for each state
                - shape: (N)
            inputs_len (int tensor): Lengths of states in a batch
                - shape: (batch)

        Returns:
            out (torch.FloatTensor): log probability of transitioning to each state
        r&   )dimsrp   Nr   r7   r8   )
r3   rv   r   rw   rollru   r   rC   r   rt   )r   rS   r\   rK   transition_p	staying_plog_staying_probabilitylog_transition_probabilitystayingleavinginputs_len_maskr   r   r   r   r`     s   


zTransitionModel.forwardN)r   r   r   r   r`   r   r   r   r   r)     s    r)   c                       s2   e Zd ZdZd
 fddZdd Zdd	 Z  ZS )r+   zrEmission Model of the HMM, it represents the probability of
    emitting an observation based on the current statereturnNc                    s   t    tjj| _d S r   )r'   r(   tdistnormalNormaldistribution_functionr   r5   r   r   r(     s   
zEmissionModel.__init__c                 C   s    |dkr|  |||  S |S )Nr   )r   r   )r   rU   stdsr   r   r   r   r     s    zEmissionModel.samplec                 C   s@   |  ||}||d}t|d}tj|| dd}|S )a  Calculates the log probability of the the given data (x_t)
            being observed from states with given means and stds
        Args:
            x_t (float tensor) : observation at current time step
                - shape: (batch, feature_dim)
            means (float tensor): means of the distributions of hidden states
                - shape: (batch, hidden_state, feature_dim)
            stds (float tensor): standard deviations of the distributions of the hidden states
                - shape: (batch, hidden_state, feature_dim)
            state_lengths (int tensor): Lengths of states in a batch
                - shape: (batch)

        Returns:
            out (float tensor): observation log likelihoods,
                                    expressing the probability of an observation
                being generated from a state i
                shape: (batch, hidden_state)
        r&   r7   r8   )r   log_probrD   r   r3   rI   )r   r   rU   r   state_lengthsemission_distsr   r   r   r   r   r`     s
   zEmissionModel.forward)r   N)r   r   r   r   r(   r   r`   r   r   r   r5   r   r+     s
    r+   )typingr   r3   torch.distributionsdistributionsr   torch.nn.functionalr   
functionalr   torch.utils.checkpointr   %TTS.tts.layers.overflow.common_layersr   r   %TTS.tts.layers.tacotron.common_layersr   TTS.tts.utils.helpersr   Moduler	   r)   r+   r   r   r   r   <module>   s       X$