o p’×i´Õã@sÞdZddlZddlmZddlmmZGdd„dejj ƒZ Gdd„dejj ƒZ Gdd „d ejj ƒZ Gd d „d ejj ƒZ Gd d „d ejj ƒZGdd„dejj ƒZGdd„dejj ƒZddd„Zdd„Zdd„Zddd„ZdS)ak Multi-microphone components. This library contains functions for multi-microphone signal processing. Example ------- >>> import torch >>> >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT, ISTFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, SrpPhat, Music >>> from speechbrain.processing.multi_mic import DelaySum, Mvdr, Gev >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channels] >>> xs_noise_diff = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs_noise_diff = xs_noise_diff.unsqueeze(0) >>> xs_noise_loc = read_audio('tests/samples/multi-mic/noise_0.70225_-0.70225_0.11704.flac') >>> xs_noise_loc = xs_noise_loc.unsqueeze(0) >>> fs = 16000 # sampling rate >>> ss = xs_speech >>> nn_diff = 0.05 * xs_noise_diff >>> nn_loc = 0.05 * xs_noise_loc >>> xs_diffused_noise = ss + nn_diff >>> xs_localized_noise = ss + nn_loc >>> # Delay-and-Sum Beamforming with GCC-PHAT localization >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> delaysum = DelaySum() >>> istft = ISTFT(sample_rate=fs) >>> Xs = stft(xs_diffused_noise) >>> Ns = stft(nn_diff) >>> XXs = cov(Xs) >>> NNs = cov(Ns) >>> tdoas = gccphat(XXs) >>> Ys_ds = delaysum(Xs, tdoas) >>> ys_ds = istft(Ys_ds) >>> # Mvdr Beamforming with SRP-PHAT localization >>> mvdr = Mvdr() >>> mics = torch.zeros((4,3), dtype=torch.float) >>> mics[0,:] = torch.FloatTensor([-0.05, -0.05, +0.00]) >>> mics[1,:] = torch.FloatTensor([-0.05, +0.05, +0.00]) >>> mics[2,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> mics[3,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> srpphat = SrpPhat(mics=mics) >>> doas = srpphat(XXs) >>> Ys_mvdr = mvdr(Xs, NNs, doas, doa_mode=True, mics=mics, fs=fs) >>> ys_mvdr = istft(Ys_mvdr) >>> # Mvdr Beamforming with MUSIC localization >>> music = Music(mics=mics) >>> doas = music(XXs) >>> Ys_mvdr2 = mvdr(Xs, NNs, doas, doa_mode=True, mics=mics, fs=fs) >>> ys_mvdr2 = istft(Ys_mvdr2) >>> # GeV Beamforming >>> gev = Gev() >>> Xs = stft(xs_localized_noise) >>> Ss = stft(ss) >>> Ns = stft(nn_loc) >>> SSs = cov(Ss) >>> NNs = cov(Ns) >>> Ys_gev = gev(Xs, SSs, NNs) >>> ys_gev = istft(Ys_gev) Authors: * William Aris * Francois Grondin éN)Úversioncs8eZdZdZd ‡fdd„ Zdd„Zed dd„ƒZ‡ZS) Ú CovarianceaComputes the covariance matrices of the signals. Arguments --------- average : bool Informs the module if it should return an average (computed on the time dimension) of the covariance matrices. The Default value is True. Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT >>> from speechbrain.processing.multi_mic import Covariance >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channels] >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs_noise = xs_noise.unsqueeze(0) >>> xs = xs_speech + 0.05 * xs_noise >>> fs = 16000 >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> XXs.shape torch.Size([1, 1001, 201, 2, 10]) Tcótƒ ¡||_dS©N)ÚsuperÚ__init__Úaverage)Úselfr©Ú __class__©ú^/home/ubuntu/SoloSpeech/.venv/lib/python3.10/site-packages/speechbrain/processing/multi_mic.pyrys  zCovariance.__init__cCstj||jd}|S)a.This method uses the utility function _cov to compute covariance matrices. Therefore, the result has the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics + n_pairs). The order on the last dimension corresponds to the triu_indices for a square matrix. For instance, if we have 4 channels, we get the following order: (0, 0), (0, 1), (0, 2), (0, 3), (1, 1), (1, 2), (1, 3), (2, 2), (2, 3) and (3, 3). Therefore, XXs[..., 0] corresponds to channels (0, 0) and XXs[..., 1] corresponds to channels (0, 1). Arguments: ---------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics) )ÚXsr)rÚ_covr)r rÚXXsr r r Úforward~szCovariance.forwardc Cs |jd}|dddd…f d¡}|dddd…f d¡}t || dd¡¡t || dd¡¡}t || dd¡¡t || dd¡¡}t ||¡}|d|d|df}|d|d|df} t || fd¡} |durƒ| jd} tj| ddd} |  d| ddd¡} | S) a\Computes the covariance matrices (XXs) of the signals. The result will have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics + n_pairs). Arguments: ---------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics) average : boolean Informs the function if it should return an average (computed on the time dimension) of the covariance matrices. Default value is True. é.rNééT)Úkeepdim) ÚshapeÚ unsqueezeÚtorchÚmatmulÚ transposeÚ triu_indicesÚstackÚmeanÚrepeat) rrÚn_micsÚXs_reÚXs_imÚRxx_reÚRxx_imÚidxÚXXs_reÚXXs_imrÚ n_time_framesr r r r“s$  ÿ ÿ  zCovariance._cov)T) Ú__name__Ú __module__Ú __qualname__Ú__doc__rrÚ staticmethodrÚ __classcell__r r r r rVs "rcs>eZdZdZ‡fdd„Z    d dd„Zed d „ƒZ‡ZS) ÚDelaySumaQPerforms delay and sum beamforming by using the TDOAs and the first channel as a reference. Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT, ISTFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, DelaySum >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech. unsqueeze(0) # [batch, time, channel] >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs_noise = xs_noise.unsqueeze(0) #[batch, time, channels] >>> fs = 16000 >>> xs = xs_speech + 0.05 * xs_noise >>> >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> delaysum = DelaySum() >>> istft = ISTFT(sample_rate=fs) >>> >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> tdoas = gccphat(XXs) >>> Ys = delaysum(Xs, tdoas) >>> ys = istft(Ys) cótƒ ¡dSr©rr©r r r r rèózDelaySum.__init__FNçpu@c CsT|jd}| |j¡}|rt||||d}nt|d}t||d} tj|| d} | S)a6This method computes a steering vector by using the TDOAs/DOAs and then calls the utility function _delaysum to perform beamforming. The result has the following format: (batch, time_step, n_fft, 2, 1). Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics) localization_tensor : torch.Tensor A tensor containing either time differences of arrival (TDOAs) (in samples) for each timestamp or directions of arrival (DOAs) (xyz coordinates in meters). If localization_tensor represents TDOAs, then its format is (batch, time_steps, n_mics + n_pairs). If localization_tensor represents DOAs, then its format is (batch, time_steps, 3) doa_mode : bool The user needs to set this parameter to True if localization_tensor represents DOAs instead of TDOAs. Its default value is set to False. mics : torch.Tensor The cartesian position (xyz coordinates in meters) of each microphone. The tensor must have the following format (n_mics, 3). This parameter is only mandatory when localization_tensor represents DOAs. fs : int The sample rate in Hertz of the signals. This parameter is only mandatory when localization_tensor represents DOAs. c : float The speed of sound in the medium. The speed is expressed in meters per second and the default value of this parameter is 343 m/s. This parameter is only used when localization_tensor represents DOAs. Returns ------- Ys : torch.Tensor é©ÚdoasÚmicsÚfsÚc©Útdoas©ÚtausÚn_fft)rÚAs)rÚtoÚdeviceÚ doas2tausÚ tdoas2tausÚsteeringr.Ú _delaysum) r rÚlocalization_tensorÚdoa_moder7r8r9r>r=r?ÚYsr r r rìs /   zDelaySum.forwardc Csª|jd}|dddd…f|}d|dddd…f|}|dddd…f}|dddd…f}tj||||ddd }tj||||ddd }t ||fd¡} | S) alPerform delay and sum beamforming. The result has the following format: (batch, time_step, n_fft, 2, 1). Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics) As : torch.Tensor The steering vector to point in the direction of the target source. The tensor must have the format (batch, time_step, n_fft/2 + 1, 2, n_mics) Returns ------- Ys : torch.Tensor r.rNéÿÿÿÿrrT©Údimr)rrÚsumr) rr?rÚWs_reÚWs_imr r!ÚYs_reÚYs_imrHr r r rE,s zDelaySum._delaysum©FNNr3) r(r)r*r+rrr,rEr-r r r r r.Ås " ù@r.csBeZdZdZd ‡fdd„ Z    d dd „Zed d d „ƒZ‡ZS)ÚMvdraSPerform minimum variance distortionless response (MVDR) beamforming by using an input signal in the frequency domain, its covariance matrices and tdoas (to compute a steering vector). Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT, ISTFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, DelaySum >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channel] >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs_noise = xs_noise.unsqueeze(0) #[batch, time, channels] >>> fs = 16000 >>> xs = xs_speech + 0.05 * xs_noise >>> >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> mvdr = Mvdr() >>> istft = ISTFT(sample_rate=fs) >>> >>> Xs = stft(xs) >>> Ns = stft(xs_noise) >>> XXs = cov(Xs) >>> NNs = cov(Ns) >>> tdoas = gccphat(XXs) >>> Ys = mvdr(Xs, NNs, tdoas) >>> ys = istft(Ys) ç#B’ ¡œÇ;crr)rrÚeps)r rTr r r r{s  z Mvdr.__init__FNr3c Csv|jd}| |j¡}| |j¡}|dur| |j¡}|r&t||||d} nt|d} t| |d} tj||| d} | S)aÎThis method computes a steering vector before using the utility function _mvdr to perform beamforming. The result has the following format: (batch, time_step, n_fft, 2, 1). Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics) NNs : torch.Tensor The covariance matrices of the noise signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs) localization_tensor : torch.Tensor A tensor containing either time differences of arrival (TDOAs) (in samples) for each timestamp or directions of arrival (DOAs) (xyz coordinates in meters). If localization_tensor represents TDOAs, then its format is (batch, time_steps, n_mics + n_pairs). If localization_tensor represents DOAs, then its format is (batch, time_steps, 3) doa_mode : bool The user needs to set this parameter to True if localization_tensor represents DOAs instead of TDOAs. Its default value is set to False. mics : torch.Tensor The cartesian position (xyz coordinates in meters) of each microphone. The tensor must have the following format (n_mics, 3). This parameter is only mandatory when localization_tensor represents DOAs. fs : int The sample rate in Hertz of the signals. This parameter is only mandatory when localization_tensor represents DOAs. c : float The speed of sound in the medium. The speed is expressed in meters per second and the default value of this parameter is 343 m/s. This parameter is only used when localization_tensor represents DOAs. Returns ------- Ys : torch.Tensor r4Nr5r:r<)rÚNNsr?)rr@rArBrCrDrRÚ_mvdr) r rrUrFrGr7r8r9r>r=r?rHr r r rs 3     z Mvdr.forwardcCsxtj|ddd\}}t |¡}|ddd…|f}|ddd…|f}|dddd…f d ¡} d |dddd…f d ¡} |  d d ¡} d |  d d ¡} t || ¡t || ¡} t || ¡t || ¡}d t | | ¡t | |¡}t | |¡ d ¡}t ||¡ d ¡ }|dddd…f}|dddd…f}tj||||d dd }tj||||d dd }t  ||fd¡}|S)a;Perform minimum variance distortionless response beamforming. Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics). NNs : torch.Tensor The covariance matrices of the noise signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). As : torch.Tensor The steering vector to point in the direction of the target source. The tensor must have the format (batch, time_step, n_fft/2 + 1, 2, n_mics). eps : float A small value to avoid division by zero. Returns ------- Ys : torch.Tensor Tr©Úreturn_inverserK©.rN©.r.rrçð?rçð¿rJéþÿÿÿ) rÚuniqueÚeigÚinvrrrÚsqueezerLr)rrUr?rTÚNNs_valÚNNs_idxÚNNs_invÚ NNs_inv_reÚ NNs_inv_imÚAsC_reÚAsC_imÚAsT_reÚAsT_imÚNNs_inv_AsC_reÚNNs_inv_AsC_imÚalpharMrNr r!rOrPrHr r r rVÉs6  ÿÿ  ÿÿz Mvdr._mvdr©rSrQ) r(r)r*r+rrr,rVr-r r r r rRUs%  øHrRcs4eZdZdZ‡fdd„Zdd„Zedd„ƒZ‡ZS)ÚGevaGGeneralized EigenValue decomposition (GEV) Beamforming. Example ------- >>> from speechbrain.dataio.dataio import read_audio >>> import torch >>> >>> from speechbrain.processing.features import STFT, ISTFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import Gev >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channels] >>> xs_noise = read_audio('tests/samples/multi-mic/noise_0.70225_-0.70225_0.11704.flac') >>> xs_noise = xs_noise.unsqueeze(0) >>> fs = 16000 >>> ss = xs_speech >>> nn = 0.05 * xs_noise >>> xs = ss + nn >>> >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gev = Gev() >>> istft = ISTFT(sample_rate=fs) >>> >>> Ss = stft(ss) >>> Nn = stft(nn) >>> Xs = stft(xs) >>> >>> SSs = cov(Ss) >>> NNs = cov(Nn) >>> >>> Ys = gev(Xs, SSs, NNs) >>> ys = istft(Ys) cr/rr0r1r r r r6r2z Gev.__init__cCstj|||d}|S)agThis method uses the utility function _gev to perform generalized eigenvalue decomposition beamforming. Therefore, the result has the following format: (batch, time_step, n_fft, 2, 1). Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics). SSs : torch.Tensor The covariance matrices of the target signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). NNs : torch.Tensor The covariance matrices of the noise signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). Returns ------- Ys : torch.Tensor )rÚSSsrU)roÚ_gev)r rrprUrHr r r r:sz Gev.forwardcCsˆ| |j¡}| |j¡}|jd}|jd}tj||fdd}tj|ddd\}}|dtd|ƒf}|dt|d|ƒf}t |¡}t  ||¡\}} |d|ddf} |d|ddf} d tj | d| dd dd d   ddd|¡} | | 9} | | 9} | d d …|f} | d d …|f}|ddd d …f}|ddd d …f}tj | |||d dd }tj | |||d dd }t  ||fd ¡}|S)a&Perform generalized eigenvalue decomposition beamforming. The result has the following format: (batch, time_step, n_fft, 2, 1). Arguments --------- Xs : torch.Tensor A batch of audio signals in the frequency domain. The tensor must have the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics). SSs : torch.Tensor The covariance matrices of the target signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). NNs : torch.Tensor The covariance matrices of the noise signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). Returns ------- Ys : torch.Tensor r©rKTrrW.rr4r[rrJçà?N) r@rArrÚcatr^Úranger_Úpos_defÚgevdrLrr)rrprUrÚ n_mics_pairsÚSSs_NNsÚ SSs_NNs_valÚ SSs_NNs_idxÚVsÚDsÚF_reÚF_imÚF_normrMrNr r!rOrPrHr r r rqTs8     ÿ  þzGev._gev) r(r)r*r+rrr,rqr-r r r r ros  &rocsReZdZdZd‡fdd„ Zdd„Zeddd „ƒZedd d „ƒZed d „ƒZ ‡Z S)ÚGccPhata¶Generalized Cross-Correlation with Phase Transform localization. Arguments --------- tdoa_max : int Specifies a range to search for delays. For example, if tdoa_max = 10, the method will restrict its search for delays between -10 and 10 samples. This parameter is optional and its default value is None. When tdoa_max is None, the method will search for delays between -n_fft/2 and n_fft/2 (full range). eps : float A small value to avoid divisions by 0 with the phase transformation. The default value is 1e-20. Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT, ISTFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, DelaySum >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channel] >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs_noise = xs_noise.unsqueeze(0) #[batch, time, channels] >>> fs = 16000 >>> xs = xs_speech + 0.05 * xs_noise >>> >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> tdoas = gccphat(XXs) NrScstƒ ¡||_||_dSr)rrÚtdoa_maxrT)r r‚rTr r r rÀs  zGccPhat.__init__cCs2tj||jd}tj||jd}tj||d}|S)aPerform generalized cross-correlation with phase transform localization by using the utility function _gcc_phat and by extracting the delays (in samples) before performing a quadratic interpolation to improve the accuracy. The result has the format: (batch, time_steps, n_mics + n_pairs). The order on the last dimension corresponds to the triu_indices for a square matrix. For instance, if we have 4 channels, we get the following order: (0, 0), (0, 1), (0, 2), (0, 3), (1, 1), (1, 2), (1, 3), (2, 2), (2, 3) and (3, 3). Therefore, delays[..., 0] corresponds to channels (0, 0) and delays[..., 1] corresponds to channels (0, 1). Arguments: ---------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). )rrT)Úxxsr‚)rƒÚdelays)rÚ _gcc_phatrTÚ_extract_delaysr‚Ú _interpolate)r rrƒr„r;r r r rÆszGccPhat.forwardc Cs|jddd}tj|ddd\}}|dddd…f}|dddd…f}t |d|d¡|}||}||} t || fd¡} |  dd ¡} t tj¡t d ¡krft  | d | d ¡} tj j | |d } n tj | d|gd} | d|dd…f} |  dd ¡} | S)aIEvaluate GCC-PHAT for each timestamp. It returns the result in the time domain. The result has the format: (batch, time_steps, n_fft, n_mics + n_pairs). Arguments --------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). eps : float A small value to avoid divisions by 0 with the phase transform. The default value is 1e-20. Returns ------- xxs : torch.Tensor r4rTrrW.rNrz1.8.0rYrZ)Ún)Ú signal_ndimÚ signal_sizes) rrr^ÚsqrtrrrÚparseÚ __version__ÚcomplexÚfftÚirfft) rrTÚ n_samplesÚXXs_valÚXXs_idxr%r&ÚXXs_absÚ XXs_re_phatÚ XXs_im_phatÚXXs_phatrƒr r r r…Ýs   zGccPhat._gcc_phatc Cs°|jd}|durtj|ddd}|dd|…dd…f}|d| d…dd…f}t ||fd¡}t |d¡\}}||jd}||jdk} || |7<|| |8<|S)aBExtract the rounded delays from the cross-correlation for each timestamp. The result has the format: (batch, time_steps, n_mics + n_pairs). Arguments --------- xxs : torch.Tensor The correlation signals obtained after a gcc-phat operation. The tensor must have the format (batch, time_steps, n_fft, n_mics + n_pairs). tdoa_max : int Specifies a range to search for delays. For example, if tdoa_max = 10, the method will restrict its search for delays between -10 and 10 samples. This parameter is optional and its default value is None. When tdoa_max is None, the method will search for delays between -n_fft/2 and +n_fft/2 (full range). Returns ------- delays : torch.Tensor r4NÚfloor©Ú rounding_mode.r)rrÚdivrtÚmax) rƒr‚r>Úslice_1Úslice_2Ú xxs_slicedÚ_r„Úoffsetr$r r r r†s zGccPhat._extract_delayscCs¸|jd}t |d||¡ d¡}t |d|¡ d¡}t |||¡ d¡}t |d|¡ d¡}t |d||¡ d¡}t |d|¡ d¡}|||d|d|d|}|S)a¯Perform quadratic interpolation on the cross-correlation to improve the tdoa accuracy. The result has the format: (batch, time_steps, n_mics + n_pairs) Arguments --------- xxs : torch.Tensor The correlation signals obtained after a gcc-phat operation. The tensor must have the format (batch, time_steps, n_fft, n_mics + n_pairs). delays : torch.Tensor The rounded tdoas obtained by selecting the sample with the highest amplitude. The tensor must have the format (batch, time_steps, n_mics + n_pairs). Returns ------- delays_frac : torch.Tensor r4rr)rrÚfmodrÚgatherra)rƒr„r>ÚtpÚy1Úy2Úy3Ú delays_fracr r r r‡=s $zGccPhat._interpolate)NrSrnr) r(r)r*r+rrr,r…r†r‡r-r r r r r—s( 0 .rcs@eZdZdZ    d ‡fdd„ Zdd „Zed d d „ƒZ‡ZS)ÚSrpPhataV Steered-Response Power with Phase Transform Localization. Arguments --------- mics : torch.Tensor The cartesian coordinates (xyz) in meters of each microphone. The tensor must have the following format (n_mics, 3). space : string If this parameter is set to 'sphere', the localization will be done in 3D by searching in a sphere of possible doas. If it set to 'circle', the search will be done in 2D by searching in a circle. By default, this parameter is set to 'sphere'. Note: The 'circle' option isn't implemented yet. sample_rate : int The sample rate in Hertz of the signals to perform SRP-PHAT on. By default, this parameter is set to 16000 Hz. speed_sound : float The speed of sound in the medium. The speed is expressed in meters per second and the default value of this parameter is 343 m/s. eps : float A small value to avoid errors like division by 0. The default value of this parameter is 1e-20. Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import SrpPhat >>> xs_speech = read_audio('tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac') >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> fs = 16000 >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channels] >>> xs_noise = xs_noise.unsqueeze(0) >>> ss1 = xs_speech >>> ns1 = 0.05 * xs_noise >>> xs1 = ss1 + ns1 >>> ss2 = xs_speech >>> ns2 = 0.20 * xs_noise >>> xs2 = ss2 + ns2 >>> ss = torch.cat((ss1,ss2), dim=0) >>> ns = torch.cat((ns1,ns2), dim=0) >>> xs = torch.cat((xs1,xs2), dim=0) >>> mics = torch.zeros((4,3), dtype=torch.float) >>> mics[0,:] = torch.FloatTensor([-0.05, -0.05, +0.00]) >>> mics[1,:] = torch.FloatTensor([-0.05, +0.05, +0.00]) >>> mics[2,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> mics[3,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> srpphat = SrpPhat(mics=mics) >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> doas = srpphat(XXs) Úsphereé€>r3rScsBtƒ ¡|dkr tƒ|_|dkr t|j|||d|_||_dS©NrªÚcircle)r7r8r9)rrrªr6rBr=rT)r r7ÚspaceÚ sample_rateÚ speed_soundrTr r r r¥s  ÿ zSrpPhat.__init__cCs8|jd}t|j |j¡|ƒ}tj|||j|jd}|S)aGPerform SRP-PHAT localization on a signal by computing a steering vector and then by using the utility function _srp_phat to extract the doas. The result is a tensor containing the directions of arrival (xyz coordinates (in meters) in the direction of the sound source). The output tensor has the format (batch, time_steps, 3). This localization method uses Global Coherence Field (GCF): https://www.researchgate.net/publication/221491705_Speaker_localization_based_on_oriented_global_coherence_field Arguments --------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). Returns ------- doas : torch.Tensor r4)rr?r6rT) rrDr=r@rAr©Ú _srp_phatr6rT©r rr>r?r6r r r r¿s zSrpPhat.forwardcCs$| |j¡}| |j¡}|jd}t ||¡}|dd…dd…d|ddd…ff}|dd…dd…d|ddd…ff}|dd…dd…d|ddd…ff}|dd…dd…d|ddd…ff} |||| } || ||} |  | jdd¡} |  | jdd¡} tj|ddd\} } | dd…dd…dd…ddd…f}| dd…dd…dd…ddd…f}| |jd|jddf¡}| |jd|jddf¡}t |d|d¡|}||}||}t ||   dd¡¡}t ||   dd¡¡}||}tj |dd \}}||dd…fdd…| dd…f}|S) aôPerform srp-phat to find the direction of arrival of the sound source. The result is a tensor containing the directions of arrival (xyz coordinates (in meters) in the direction of the sound source). The output tensor has the format: (batch, time_steps, 3). Arguments --------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). As : torch.Tensor The steering vector that cover the all the potential directions of arrival. The tensor must have the format (n_doas, n_fft/2 + 1, 2, n_mics). doas : torch.Tensor All the possible directions of arrival that will be scanned. The tensor must have the format (n_doas, 3). eps : float A very small value used to avoid division by 0. Returns ------- doas : torch.Tensor rNrrrITrWr4rr) r@rArrrÚreshaper^r‹rrrœ)rr?r6rTrr$ÚAs_1_reÚAs_1_imÚAs_2_reÚAs_2_imrMrNr’r“r%r&r”Ú XXs_re_normÚ XXs_im_normÚYs_AÚYs_BrHr Údoas_idxr r r r±Þs4    $$$$"""zSrpPhat._srp_phat)rªr«r3rSrn) r(r)r*r+rrr,r±r-r r r r r©bsEúr©csBeZdZdZ     d ‡fdd„ Zd d „Zedd d „ƒZ‡ZS)ÚMusicaÁ Multiple Signal Classification (MUSIC) localization. Arguments --------- mics : torch.Tensor The cartesian coordinates (xyz) in meters of each microphone. The tensor must have the following format (n_mics, 3). space : string If this parameter is set to 'sphere', the localization will be done in 3D by searching in a sphere of possible doas. If it set to 'circle', the search will be done in 2D by searching in a circle. By default, this parameter is set to 'sphere'. Note: The 'circle' option isn't implemented yet. sample_rate : int The sample rate in Hertz of the signals to perform SRP-PHAT on. By default, this parameter is set to 16000 Hz. speed_sound : float The speed of sound in the medium. The speed is expressed in meters per second and the default value of this parameter is 343 m/s. eps : float A small value to avoid errors like division by 0. The default value of this parameter is 1e-20. n_sig : int An estimation of the number of sound sources. The default value is set to one source. Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import SrpPhat >>> xs_speech = read_audio('tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac') >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> fs = 16000 >>> xs_speech = xs_speech.unsqueeze(0) # [batch, time, channels] >>> xs_noise = xs_noise.unsqueeze(0) >>> ss1 = xs_speech >>> ns1 = 0.05 * xs_noise >>> xs1 = ss1 + ns1 >>> ss2 = xs_speech >>> ns2 = 0.20 * xs_noise >>> xs2 = ss2 + ns2 >>> ss = torch.cat((ss1,ss2), dim=0) >>> ns = torch.cat((ns1,ns2), dim=0) >>> xs = torch.cat((xs1,xs2), dim=0) >>> mics = torch.zeros((4,3), dtype=torch.float) >>> mics[0,:] = torch.FloatTensor([-0.05, -0.05, +0.00]) >>> mics[1,:] = torch.FloatTensor([-0.05, +0.05, +0.00]) >>> mics[2,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> mics[3,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> music = Music(mics=mics) >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> doas = music(XXs) rªr«r3rSrcsHtƒ ¡|dkr tƒ|_|dkr t|j|||d|_||_||_dSr¬)rrrªr6rBr=rTÚn_sig)r r7r®r¯r°rTr¾r r r rls  ÿ zMusic.__init__cCs<|jd}t|j |j¡|ƒ}tj|||j|j|j d}|S)aƒPerform MUSIC localization on a signal by computing a steering vector and then by using the utility function _music to extract the doas. The result is a tensor containing the directions of arrival (xyz coordinates (in meters) in the direction of the sound source). The output tensor has the format (batch, time_steps, 3). Arguments --------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). Returns ------- doas : torch.Tensor r4)rr?r6r¾rT) rrDr=r@rAr½Ú_musicr6r¾rTr²r r r rŠs ÿz Music.forwardc CsÒ| |j¡}| |j¡}|jd}|jd}|jd}||}tj|ddd\} } t | ¡\} } |  d¡ dd|dddd¡} | dt d|ƒdf} | dt d|ƒdf}| d¡ d¡ d¡  dddddd d ¡}| | jd| jdddddd¡}|d }|d }t  || ¡t  ||¡}t  ||¡t  || ¡}t  |d|d¡}tj |d d }tj |dd d tj |dd d }||| d ¡}tj |dd |}tj|dd \} }||dd…fdd…| dd…f}|S)aÞPerform multiple signal classification to find the direction of arrival of the sound source. The result has the format: (batch, time_steps, 3). Arguments --------- XXs : torch.Tensor The covariance matrices of the input signal. The tensor must have the format (batch, time_steps, n_fft/2 + 1, 2, n_mics + n_pairs). As : torch.Tensor The steering vector that covers the all the potential directions of arrival. The tensor must have the format. (n_doas, n_fft/2 + 1, 2, n_mics). doas : torch.Tensor All the possible directions of arrival that will be scanned. The tensor must have the format (n_doas, 3). n_sig : int The number of signals in the signal + noise subspace (default is 1). eps : float A small number to avoid div by zero errors. Returns ------- doas : torch.Tensor rrr4TrrW.éérrYrZrrN)r@rArrr^r_ÚsvdlrrruÚpermuterr‹rLrarœ)rr?r6r¾rTrÚn_doasÚn_binsÚ svd_ranger’r“ÚUsr ÚUs_reÚUs_imÚAs_reÚAs_imÚ As_mm_Us_reÚ As_mm_Us_imÚ As_mm_Us_absÚ As_mm_Us_sumÚ As_As_absÚPsrHr¼r r r r¿¨s:     ü"$"z Music._music)rªr«r3rSrrn) r(r)r*r+rrr,r¿r-r r r r r½&sHùr½r3cCs(||t | |j¡| dd¡¡}|S)aThis function converts directions of arrival (xyz coordinates expressed in meters) in time differences of arrival (expressed in samples). The result has the following format: (batch, time_steps, n_mics). Arguments --------- doas : torch.Tensor The directions of arrival expressed with cartesian coordinates (xyz) in meters. The tensor must have the following format: (batch, time_steps, 3). mics : torch.Tensor The cartesian position (xyz) in meters of each microphone. The tensor must have the following format (n_mics, 3). fs : int The sample rate in Hertz of the signals. c : float The speed of sound in the medium. The speed is expressed in meters per second and the default value of this parameter is 343 m/s. Returns ------- taus : torch.Tensor Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.multi_mic import sphere, doas2taus >>> xs = read_audio('tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac') >>> xs = xs.unsqueeze(0) # [batch, time, channels] >>> fs = 16000 >>> mics = torch.zeros((4,3), dtype=torch.float) >>> mics[0,:] = torch.FloatTensor([-0.05, -0.05, +0.00]) >>> mics[1,:] = torch.FloatTensor([-0.05, +0.05, +0.00]) >>> mics[2,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> mics[3,:] = torch.FloatTensor([+0.05, +0.05, +0.00]) >>> doas = sphere() >>> taus = doas2taus(doas, mics, fs) rr)rrr@rAr)r6r7r8r9r=r r r rBùs$*rBcCsF|jt|jƒd}tdd|dddƒ}|dtd|ƒf}|S)a½This function selects the tdoas of each channel and put them in a tensor. The result has the following format: (batch, time_steps, n_mics). Arguments --------- tdoas : torch.Tensor The time difference of arrival (TDOA) (in samples) for each timestamp. The tensor has the format (batch, time_steps, n_mics + n_pairs). Returns ------- taus : torch.Tensor Example ------- >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, tdoas2taus >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs = xs_speech + 0.05 * xs_noise >>> xs = xs.unsqueeze(0) >>> fs = 16000 >>> >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> >>> Xs = stft(xs) >>> XXs = cov(Xs) >>> tdoas = gccphat(XXs) >>> taus = tdoas2taus(tdoas) rérsr4.r)rÚlenÚintru)r;Ún_pairsÚ n_channelsr=r r r rC(s)rCcCsÞd}t|ddƒ}d|tjd||jd|}| |jd¡}| t|jƒ¡ dt|jƒ|f¡}t | |¡}t  | |¡}t  ||ft|jƒ¡}|  t|jƒdt|jƒd¡  t|jƒdt|jƒd¡}|S)aÂThis function computes a steering vector by using the time differences of arrival for each channel (in samples) and the number of bins (n_fft). The result has the following format: (batch, time_step, n_fft/2 + 1, 2, n_mics). Arguments: ---------- taus : torch.Tensor The time differences of arrival for each channel. The tensor must have the following format: (batch, time_steps, n_mics). n_fft : int The number of bins resulting of the STFT. It is assumed that the argument "onesided" was set to True for the STFT. Example: --------f >>> import torch >>> from speechbrain.dataio.dataio import read_audio >>> from speechbrain.processing.features import STFT >>> from speechbrain.processing.multi_mic import Covariance >>> from speechbrain.processing.multi_mic import GccPhat, tdoas2taus, steering >>> >>> xs_speech = read_audio( ... 'tests/samples/multi-mic/speech_-0.82918_0.55279_-0.082918.flac' ... ) >>> xs_noise = read_audio('tests/samples/multi-mic/noise_diffuse.flac') >>> xs = xs_speech + 0.05 * xs_noise >>> xs = xs.unsqueeze(0) # [batch, time, channels] >>> fs = 16000 >>> stft = STFT(sample_rate=fs) >>> cov = Covariance() >>> gccphat = GccPhat() >>> >>> Xs = stft(xs) >>> n_fft = Xs.shape[2] >>> XXs = cov(Xs) >>> tdoas = gccphat(XXs) >>> taus = tdoas2taus(tdoas) >>> As = steering(taus, n_fft) g-DTû! @rr4r)rA)rr) rÔrÚarangerArrrrÓÚcosÚsinrr)r=r>ÚpiÚ frame_sizeÚomegasÚa_reÚa_imÚar r r rDXs+ÿ ÿrDrc CsÌd}d}d}tjdtjd}t gd¢¡|ddd…f<t gd ¢¡|d dd…f<|t d |t dd ¡d ¡|tddƒdf<|t d |t dd ¡d ¡|tddƒdf<||tddƒdf<d|t d |t dd ¡d ¡|tdd ƒdf<d|t d |t dd ¡d ¡|tdd ƒdf<d||tdd ƒdf<tjdtjd}t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd¢¡|d dd…f<t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd¢¡|ddd…f<t  gd ¢¡|d!dd…f<t  gd"¢¡|d#dd…f<t  gd$¢¡|d dd…f<t  gd%¢¡|d&dd…f<t  gd'¢¡|d(dd…f<t  gd)¢¡|d*dd…f<t  gd+¢¡|d,dd…f<t  gd-¢¡|d.dd…f<t  gd/¢¡|d0dd…f<t  gd1¢¡|d2dd…f<t  gd3¢¡|d4dd…f<td|ƒD]±}|j d}|d}tj|dftjd} |dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡d f<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡d f<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡d f<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡df<|dd…df| d|t d|¡d f<tj | dd…ddgf| dd…ddgf| dd…dd gffdd5} tj | dd5\} } t  | ¡} | dd…df| d| dd…df}tj|d6d7\}}tj|j ddf|jd}tj|| dd8d9|dd…df<||dd…df| d|dd…df<t t |d:¡dd¡}||dd…dfdd…f||dd…dfdd…f}|t t tj|ddd5d;d¡dd¡}q±|S)>> import torch >>> from speechbrain.processing.multi_mic import sphere >>> doas = sphere() gÚí¿Å%ŸÜ?gÚí¿Å%Ÿì?gPERTû! @)é r)Údtype)rrrrN)rrrIé g@rÁg@rrÀr4r\)ér)rr4r)rrr4)rrr)rrÁrr)rrrÁr)é rr4)é r4r)rÀrré)rærrÁrÒ)rÒrÁrrä)rrærÀrå)rÁrÒræ)rrärÒrà)r4råräé )rrÀråé)rârÀræé)rârærÒé)rârÒräé)râräråé)rârårÀé)ÚaxisT)rXr˜r™)rrIrs)rÚzerosÚfloatÚ FloatTensorrÙr×rurØÚlongÚ LongTensorrrtÚsortrœr^rár›rr³Úrepeat_interleaverrL)Ú levels_countÚhÚrrÚÚptsÚtrsÚ levels_indexÚ trs_countÚ subtrs_countÚsubtrsÚsubtrs_flattenÚ subtrs_sortedr Ú index_maxÚ subtrs_scalarÚ unique_scalarÚunique_indicesÚ unique_valuesr r r rª™s° 00$ÿ$ÿ ((((((((((((((((((((((((:ÿ &ÿ ÿÿ ÿÿ8  ÿrª)r3)r)r+rÚ packagingrÚ$speechbrain.processing.decompositionÚ processingÚ decompositionr_ÚnnÚModulerr.rRrorr©r½rBrCrDrªr r r r Ús*O o; LE T/0A