ssspy.bss.mnmf

Algorithms

class ssspy.bss.mnmf.FastMNMFBase(n_basis, n_sources=None, partitioning=False, flooring_fn=functools.partial(<function max_flooring>, eps=1e-10), callbacks=None, normalization=True, record_loss=True, reference_id=0, rng=None)

Base class of fast multichannel nonnegative matrix factorization (Fast MNMF).

Parameters:

• n_basis (int) – Number of NMF bases.

• n_sources (int, optional) – Number of sources to be separated. If None is given, n_sources is determined by number of channels in input spectrogram. Default: None.

• partitioning (bool) – Whether to use partioning function. Default: False.

• flooring_fn (callable, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. Default: functools.partial(max_flooring, eps=1e-10).

• normalization (bool or str) – Normalization of diagonalizers and diagonal elements of spatial covariance matrices. Only power-based normalization is supported. Default: True.

• callbacks (callable or list[callable], optional) – Callback functions. Each function is called before separation and at each iteration. Default: None.

• record_loss (bool) – Record the loss at each iteration of the update algorithm if record_loss=True. Default: True.

• reference_id (int) – Reference channel in multichannel Wiener filter. Default: 0.

• rng (numpy.random.Generator, optioinal) – Random number generator. This is mainly used to randomly initialize PSDTF. If None is given, np.random.default_rng() is used. Default: None.

__call__(input, n_iter=100, initial_call=True, **kwargs)

Separate a frequency-domain multichannel signal.

Parameters:

• input (numpy.ndarray) – The mixture signal in frequency-domain. The shape is (n_channels, n_bins, n_frames).

• n_iter (int) – The number of iterations of demixing filter updates. Default: 100.

• initial_call (bool) – If True, perform callbacks (and computation of loss if necessary) before iterations.

Return type:

ndarray

Returns:

numpy.ndarray of the separated signal in frequency-domain. The shape is (n_channels, n_bins, n_frames).

normalize(flooring_fn='self')

Normalize diagonalizers and diagonal elements of spatial covariance matrices.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

normalize_by_power(flooring_fn='self')

Normalize diagonalizers and diagonal elements of spatial covariance matrices by power.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

Diagonalizers are normalized by

\[\boldsymbol{q}_{im} \leftarrow\frac{\boldsymbol{q}_{im}}{\psi_{im}},\]

where

\[\psi_{im} = \sqrt{\frac{1}{IJ}\sum_{i,j}|\boldsymbol{q}_{im}^{\mathsf{H}} \boldsymbol{x}_{ij}|^{2}}.\]

For diagonal elements of spatial covariance matrices,

\[d_{inm} \leftarrow\frac{d_{inm}}{\psi_{im}^{2}}.\]

class ssspy.bss.mnmf.FastGaussMNMF(n_basis, n_sources=None, diagonalizer_algorithm='IP', partitioning=False, flooring_fn=functools.partial(<function max_flooring>, eps=1e-10), pair_selector=None, callbacks=None, normalization=True, record_loss=True, reference_id=0, rng=None)

Fast multichannel nonnegative matrix factorization on Gaussian distribution (Fast Gauss-MNMF).

Parameters:

• n_basis (int) – Number of NMF bases.

• n_sources (int, optional) – Number of sources to be separated. If None is given, n_sources is determined by number of channels in input spectrogram. Default: None.

• diagonalizer_algorithm (str) – Algorithm for diagonalizers. Choose IP, IP1, or IP2. Default: IP.

• partitioning (bool) – Whether to use partioning function. Default: False.

• flooring_fn (callable, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. Default: functools.partial(max_flooring, eps=1e-10).

• callbacks (callable or list[callable], optional) – Callback functions. Each function is called before separation and at each iteration. Default: None.

• record_loss (bool) – Record the loss at each iteration of the update algorithm if record_loss=True. Default: True.

• reference_id (int) – Reference channel in multichannel Wiener filter. Default: 0.

• rng (numpy.random.Generator, optioinal) – Random number generator. This is mainly used to randomly initialize PSDTF. If None is given, np.random.default_rng() is used. Default: None.

__call__(input, n_iter=100, initial_call=True, **kwargs)

Separate a frequency-domain multichannel signal.

Parameters:

• input (numpy.ndarray) – The mixture signal in frequency-domain. The shape is (n_channels, n_bins, n_frames).

• n_iter (int) – The number of iterations of demixing filter updates. Default: 100.

• initial_call (bool) – If True, perform callbacks (and computation of loss if necessary) before iterations.

Return type:

ndarray

Returns:

numpy.ndarray of the separated signal in frequency-domain. The shape is (n_channels, n_bins, n_frames).

compute_logdet(diagonalizer)

Compute log-determinant.

Parameters:

reconstructed – Diagonalizer with shape of (*, n_channels, n_channels).

Return type:

ndarray

Returns:

numpy.ndarray of computed log-determinant values. The shape is (*).

compute_loss()

Compute loss \(\mathcal{L}\).

\(\mathcal{L}\) is defined as follows:

\[\begin{split}\mathcal{L} &:=-\frac{1}{J}\sum_{i,j}\left\{ \mathrm{tr}\left( \boldsymbol{x}_{ij}\boldsymbol{x}_{ij}^{\mathsf{H}}\boldsymbol{R}_{ij}^{-1} \right) - \log\det\boldsymbol{R}_{ij} \right\} \\ &:=\frac{1}{J}\sum_{i,j,m}\left\{ \frac{|\boldsymbol{q}_{im}^{\mathsf{H}}\boldsymbol{x}_{ij}|^{2}} {\sum_{n}\lambda_{ijn}d_{inm}} + \log\sum_{n}\lambda_{ijn}d_{inm}\right\} - 2\sum_{i}\log|\det\boldsymbol{Q}_{i}|.\end{split}\]

Return type:

float

Returns:

Computed loss.

separate(input)

Separate input using multichannel Wiener filter.

Parameters:

input (numpy.ndarray) – The mixture signal in frequency-domain. The shape is (n_channels, n_bins, n_frames).

Return type:

ndarray

Returns:

numpy.ndarray of the separated signal in frequency-domain. The shape is (n_sources, n_bins, n_frames).

update_activation(flooring_fn='self')

Update NMF activations by MM algorithm.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

Update \(v_{kjn}\) as follows:

\[v_{kjn} \leftarrow\left[ \frac{\displaystyle\sum_{i,m}\frac{|\boldsymbol{q}_{im}^{\mathsf{H}}\boldsymbol{x}_{ij}|^{2}d_{inm}t_{ikn}} {\left(\sum_{k',n'}t_{ik'n'}v_{k'jn'}d_{in'm}\right)^{2}}} {\displaystyle\sum_{i,m}\dfrac{d_{inm}t_{ikn}}{\sum_{k',n'}t_{ik'n'}v_{k'jn'}d_{in'm}}} \right]^{\frac{1}{2}}v_{kjn}.\]

update_basis(flooring_fn='self')

Update NMF bases by MM algorithm.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

Update \(t_{ikn}\) as follows:

\[t_{ikn} \leftarrow\left[ \frac{\displaystyle\sum_{j,m}\frac{|\boldsymbol{q}_{im}^{\mathsf{H}}\boldsymbol{x}_{ij}|^{2}d_{inm}v_{kjn}} {\left(\sum_{k',n'}t_{ik'n'}v_{k'jn'}d_{in'm}\right)^{2}}} {\displaystyle\sum_{j,m}\dfrac{d_{inm}v_{kjn}}{\sum_{k',n'}t_{ik'n'}v_{k'jn'}d_{in'm}}} \right]^{\frac{1}{2}}t_{ikn}.\]

update_diagonalizer(flooring_fn='self')

Update diagonalizer.

• If diagonalizer_algorithm is IP or IP1, update_diagonalizer_model_ip1 is called.

• If diagonalizer_algorithm is IP2, update_diagonalizer_model_ip2 is called.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

update_diagonalizer_ip1(flooring_fn='self')

Update diagonalizer once using iterative projection.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

Diagonalizers are updated sequentially for \(m=1,\ldots,M\) as follows:

\[\begin{split}\boldsymbol{q}_{im} &\leftarrow\left(\boldsymbol{Q}_{im}^{\mathsf{H}}\boldsymbol{U}_{im}\right)^{-1} \ \boldsymbol{e}_{m}, \\ \boldsymbol{q}_{im} &\leftarrow\frac{\boldsymbol{q}_{im}} {\sqrt{\boldsymbol{q}_{im}^{\mathsf{H}}\boldsymbol{U}_{im}\boldsymbol{q}_{im}}},\end{split}\]

where

\[\boldsymbol{U}_{im} = \frac{1}{J}\sum_{j} \frac{\boldsymbol{x}_{ij}\boldsymbol{x}_{ij}^{\mathsf{H}}} {\sum_{n}\left(\sum_{k}z_{nk}t_{ik}v_{kj}\right)d_{inm}}\]

if partitioning=True, otherwise

\[\boldsymbol{U}_{im} = \frac{1}{J}\sum_{j} \frac{\boldsymbol{x}_{ij}\boldsymbol{x}_{ij}^{\mathsf{H}}} {\sum_{n}\left(\sum_{k}t_{ikn}v_{kjn}\right)d_{inm}}.\]

update_diagonalizer_ip2(flooring_fn='self')

Update diagonalizer once using pairwise iterative projection.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

For \(m_{1}\) and \(m_{2}\) (\(m_{1}\neq m_{2}\)), compute weighted covariance matrix as follows:

\[\boldsymbol{U}_{im} = \frac{1}{J}\sum_{j} \frac{\boldsymbol{x}_{ij}\boldsymbol{x}_{ij}^{\mathsf{H}}}{\sum_{n}\lambda_{ijn}d_{inm}},\]

\(\lambda_{ijn}\) is computed by

\[\lambda_{ijn}=\sum_{k}z_{nk}t_{ik}v_{kj}\]

if partitioning=True. Otherwise,

\[\lambda_{ijn}=\sum_{k}t_{ikn}v_{kjn}.\]

Using \(\boldsymbol{U}_{im_{1}}\) and \(\boldsymbol{U}_{im_{2}}\), we compute generalized eigenvectors.

\[\left({\boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})}}^{\mathsf{H}}\boldsymbol{U}_{im_{1}} \boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})}\right)\boldsymbol{h}_{i} = \mu_{i} \left({\boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})}}^{\mathsf{H}}\boldsymbol{U}_{im_{2}} \boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})}\right)\boldsymbol{h}_{i},\]

where

\[\begin{split}\boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})} &= (\boldsymbol{Q}_{i}\boldsymbol{U}_{im_{1}})^{-1} ( \begin{array}{cc} \boldsymbol{e}_{m_{1}} & \boldsymbol{e}_{m_{2}} \end{array} ), \\ \boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})} &= (\boldsymbol{Q}_{i}\boldsymbol{U}_{im_{2}})^{-1} ( \begin{array}{cc} \boldsymbol{e}_{m_{1}} & \boldsymbol{e}_{m_{2}} \end{array} ).\end{split}\]

After that, we standardize two eigenvectors \(\boldsymbol{h}_{im_{1}}\) and \(\boldsymbol{h}_{im_{2}}\).

\[\begin{split}\boldsymbol{h}_{im_{1}} &\leftarrow\frac{\boldsymbol{h}_{im_{1}}} {\sqrt{\boldsymbol{h}_{im_{1}}^{\mathsf{H}} \left({\boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})}}^{\mathsf{H}}\boldsymbol{U}_{im_{1}} \boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})}\right) \boldsymbol{h}_{im_{1}}}}, \\ \boldsymbol{h}_{im_{2}} &\leftarrow\frac{\boldsymbol{h}_{im_{2}}} {\sqrt{\boldsymbol{h}_{im_{2}}^{\mathsf{H}} \left({\boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})}}^{\mathsf{H}}\boldsymbol{U}_{im_{2}} \boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})}\right) \boldsymbol{h}_{im_{2}}}}.\end{split}\]

Then, update \(\boldsymbol{q}_{im_{1}}\) and \(\boldsymbol{q}_{im_{2}}\) simultaneously.

\[\begin{split}\boldsymbol{q}_{im_{1}} &\leftarrow \boldsymbol{P}_{im_{1}}^{(m_{1},m_{2})}\boldsymbol{h}_{im_{1}} \\ \boldsymbol{q}_{im_{2}} &\leftarrow \boldsymbol{P}_{im_{2}}^{(m_{1},m_{2})}\boldsymbol{h}_{im_{2}}\end{split}\]

At each iteration, we update pairs of \(m_{1}\) and \(m_{2}\) for \(m_{1}\neq m_{2}\).

update_once(flooring_fn='self')

Update MNMF parameters, diagonalizers, and diagonal elements of spatial covariance matrices once.

Parameters:

flooring_fn (callable or str, optional) – A flooring function for numerical stability. This function is expected to return the same shape tensor as the input. If you explicitly set flooring_fn=None, the identity function (lambda x: x) is used. If self is given as str, self.flooring_fn is used. Default: self.

Return type:

None

update_spatial()

Update diagonal elements of spatial covariance matrix by MM algorithm.

Update \(d_{inm}\) as follows: :rtype: None

\[d_{inm}\leftarrow\left[ \dfrac{\displaystyle\sum_{j}\frac{\lambda_{ijn}|\boldsymbol{q}_{im}^{\mathsf{H}}\boldsymbol{x}_{ij}|^{2}} {\left(\sum_{n'}\lambda_{ijn'}d_{in'm}\right)^{2}}} {\displaystyle\sum_{j}\frac{\lambda_{ijn}}{\sum_{n'}\lambda_{ijn'}d_{in'm}}} \right]^{\frac{1}{2}}d_{inm}.\]