o %ݫi@sdZddlZddlmZddlmmZGdddejj Z Gdddejj Z Gdd d ejj Z Gd d d ejj Z Gd d d ejj ZGdddejj ZGdddejj ZdddZddZddZdddZdS)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 ZddZed ddZZS) 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]) Tct||_dSN)super__init__average)selfr __class__T/home/ubuntu/.local/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 rXXsr r r forward~szCovariance.forwardc Cs |jd}|ddddfd}|ddddfd}t||ddt||dd}t||ddt||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. .rNT)keepdim) shape unsqueezetorchmatmul transpose triu_indicesstackmeanrepeat) rrn_micsXs_reXs_imRxx_reRxx_imidxXXs_reXXs_imr n_time_framesr r r rs$     zCovariance._cov)T) __name__ __module__ __qualname____doc__rr staticmethodr __classcell__r r r r rVs "rcs>eZdZdZfddZ    d ddZed d ZZS) 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) ctdSrrrr r r r rzDelaySum.__init__FNpu@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 doasmicsfsctdoastausn_fft)rAs)rtodevice doas2taus tdoas2taussteeringr. _delaysum) r rlocalization_tensordoa_moder7r8r9r>r=r?Ysr r r rs /   zDelaySum.forwardc Cs|jd}|ddddf|}d|ddddf|}|ddddf}|ddddf}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.rNrrTdimr)rrsumr) rr?rWs_reWs_imr r!Ys_reYs_imrHr r r rE,s zDelaySum._delaysumFNNr3) r(r)r*r+rrr,rEr-r r r r r.s " @r.csBeZdZdZd fdd Z    d dd Zed d d ZZS)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)rreps)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)aThis 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<)rNNsr?)rr@rArBrCrDrR_mvdr) r rrUrFrGr7r8r9r>r=r?rHr r r rs 3     z Mvdr.forwardcCsxtj|ddd\}}t|}|ddd|f}|ddd|f}|ddddfd } d |ddddfd } | d d } d | d d } t|| t|| } t|| t|| }d t| | t| |}t| |d }t||d  }|ddddf}|ddddf}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 Trreturn_inverserK.rN.r.rr?rrJ) runiqueeiginvrrrsqueezerLr)rrUr?rTNNs_valNNs_idxNNs_inv NNs_inv_re NNs_inv_imAsC_reAsC_imAsT_reAsT_imNNs_inv_AsC_reNNs_inv_AsC_imalpharMrNr r!rOrPrHr r r rVs6    z Mvdr._mvdrrSrQ) r(r)r*r+rrr,rVr-r r r r rRUs%  HrRcs4eZdZdZfddZddZeddZZS)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 )rSSsrU)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 rrKTrrW.rr4r[rrJ?N) r@rArrcatr^ranger_pos_defgevdrLrr)rrprUr n_mics_pairsSSs_NNs SSs_NNs_val SSs_NNs_idxVsDsF_reF_imF_normrMrNr r!rOrPrHr r r rqTs8       zGev._gev) r(r)r*r+rrr,rqr-r r r r ros  &rocsReZdZdZdfdd ZddZeddd Zedd d Zed d Z Z S)GccPhataGeneralized 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)rrtdoa_maxrT)r rrTr r r rs  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)rdelays)r _gcc_phatrT_extract_delaysr _interpolate)r rrrr;r r r rszGccPhat.forwardc Cs|jddd}tj|ddd\}}|ddddf}|ddddf}t|d|d|}||}||} t|| fd} | dd } ttjtd krft | d | d } tj j | |d } n tj | d|gd} | d|ddf} | 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^sqrtrrrparse __version__complexfftirfft) rrT n_samplesXXs_valXXs_idxr%r&XXs_abs XXs_re_phat XXs_im_phatXXs_phatrr r r rs   zGccPhat._gcc_phatc Cs|jd}|durtj|ddd}|dd|ddf}|d| dddf}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 r4Nfloor rounding_mode.r)rrdivrtmax) rrr>slice_1slice_2 xxs_sliced_roffsetr$r r r rs 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)aPerform 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)rrfmodrgatherra)rrr>tpy1y2y3 delays_fracr r r r=s $zGccPhat._interpolate)NrSrnr) r(r)r*r+rrr,rrrr-r r r r rs( 0 .rcs@eZdZdZ    d fdd Zdd Zed d d ZZS)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|_||_dSNrcircle)r7r8r9)rrrr6rBr=rT)r r7space sample_rate speed_soundrTr r r rs   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_phatr6rTr rr>r?r6r r r rs zSrpPhat.forwardcCs$||j}||j}|jd}t||}|ddddd|dddff}|ddddd|dddff}|ddddd|dddff}|ddddd|dddff} |||| } || ||} | | jdd} | | jdd} tj|ddd\} } | dddddddddf}| dddddddddf}||jd|jddf}||jd|jddf}t|d|d|}||}||}t|| dd}t|| dd}||}tj |dd \}}||ddfdd| ddf}|S) aPerform 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@rArrrreshaper^rrrr)rr?r6rTrr$As_1_reAs_1_imAs_2_reAs_2_imrMrNrrr%r&r XXs_re_norm XXs_im_normYs_AYs_BrHrdoas_idxr r r rs4    $$$$"""zSrpPhat._srp_phat)rrr3rSrn) r(r)r*r+rrr,rr-r r r r rbsErcsBeZdZdZ     d fdd Zd d Zedd d ZZS)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) rrr3rSrcsHt|dkr t|_|dkr t|j|||d|_||_||_dSr)rrrr6rBr=rTn_sig)r r7rrrrTrr r r rls   zMusic.__init__cCs<|jd}t|j|j|}tj|||j|j|j d}|S)aPerform 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?r6rrT) rrDr=r@rAr_musicr6rrTrr r r rs z Music.forwardc Cs||j}||j}|jd}|jd}|jd}||}tj|ddd\} } t| \} } | ddd|dddd} | dt d|df} | dt d|df}|ddd 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 \} }||ddfdd| ddf}|S)aPerform 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_svdlrrrupermuterrrLrar)rr?r6rrTrn_doasn_bins svd_rangerrUsrUs_reUs_imAs_reAs_im As_mm_Us_re As_mm_Us_im As_mm_Us_abs As_mm_Us_sum As_As_absPsrHrr r r rs:     "$"z Music._music)rrr3rSrrn) r(r)r*r+rrr,rr-r r r r r&sHrr3cCs(||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 rBs$*rBcCsF|jt|jd}tdd|ddd}|dtd|f}|S)aThis 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) rrsr4.r)rlenintru)r;n_pairs n_channelsr=r r r rC(s)rCcCsd}t|dd}d|tjd||jd|}||jd}|t|jdt|j|f}t| |}t | |}t ||ft|j}| t|jdt|jd t|jdt|jd}|S)aThis 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) rrarangerArrrrcossinrr)r=r>pi frame_sizeomegasa_rea_imar r r rDXs+ rDrc Csd}d}d}tjdtjd}tgd|dddf<tgd |d ddf<|td |tdd d |tdddf<|td |tdd d |tdddf<||tdddf<d|td |tdd d |tdd df<d|td |tdd d |tdd df<d||tdd df<tjdtjd}t gd|dddf<t gd|dddf<t gd|dddf<t gd|dddf<t gd|dddf<t gd|d ddf<t gd|dddf<t gd|dddf<t gd|dddf<t gd |d!ddf<t gd"|d#ddf<t gd$|d ddf<t gd%|d&ddf<t gd'|d(ddf<t gd)|d*ddf<t gd+|d,ddf<t gd-|d.ddf<t gd/|d0ddf<t gd1|d2ddf<t gd3|d4ddf<td|D]}|j d}|d}tj|dftjd} |dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|d f<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|d f<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|d f<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|df<|dddf| d|td|d f<tj | ddddgf| ddddgf| dddd gffdd5} tj | dd5\} } t | } | dddf| d| dddf}tj|d6d7\}}tj|j ddf|jd}tj|| dd8d9|dddf<||dddf| d|dddf<tt|d:dd}||dddfddf||dddfddf}|tttj|ddd5d;ddd}q|S)>> import torch >>> from speechbrain.processing.multi_mic import sphere >>> doas = sphere() g%?g%?gPERT! @) r)dtype)rrrrN)rrrI g@rg@rrr4r\)r)rr4r)rrr4)rrr)rrrr)rrrr) rr4) r4r)rrr)rrrr)rrrr)rrrr)rrr)rrrr)r4rr )rrr)rrr)rrr)rrr)rrr)rrr)axisT)rXrr)rrIrs)rzerosfloat FloatTensorrrrurlong LongTensorrrtsortrr^rrrrrepeat_interleaverrL) levels_counthrrptstrs levels_index trs_count subtrs_countsubtrssubtrs_flatten subtrs_sortedr index_max subtrs_scalar unique_scalarunique_indices unique_valuesr r r rs 00$$ ((((((((((((((((((((((((: &  8  r)r3)r)r+r packagingr$speechbrain.processing.decomposition processing decompositionr_nnModulerr.rRrorrrrBrCrDrr r r r s*O o; LE T/0A