o siI;@sRddlmZddlZddlmZddlmZGdddejZGdddeZdS) ) permutationsN)nnlinear_sum_assignmentcseZdZdZdfdd ZdddZed d Zed d Zedd dZ eddZ edddZ ede j fddZZS)PITLossWrappera Permutation invariant loss wrapper. Args: loss_func: function with signature (est_targets, targets, **kwargs). pit_from (str): Determines how PIT is applied. * ``'pw_mtx'`` (pairwise matrix): `loss_func` computes pairwise losses and returns a torch.Tensor of shape :math:`(batch, n\_src, n\_src)`. Each element :math:`(batch, i, j)` corresponds to the loss between :math:`targets[:, i]` and :math:`est\_targets[:, j]` * ``'pw_pt'`` (pairwise point): `loss_func` computes the loss for a batch of single source and single estimates (tensors won't have the source axis). Output shape : :math:`(batch)`. See :meth:`~PITLossWrapper.get_pw_losses`. * ``'perm_avg'`` (permutation average): `loss_func` computes the average loss for a given permutations of the sources and estimates. Output shape : :math:`(batch)`. See :meth:`~PITLossWrapper.best_perm_from_perm_avg_loss`. In terms of efficiency, ``'perm_avg'`` is the least efficicient. perm_reduce (Callable): torch function to reduce permutation losses. Defaults to None (equivalent to mean). Signature of the func (pwl_set, **kwargs) : :math:`(B, n\_src!, n\_src) --> (B, n\_src!)`. `perm_reduce` can receive **kwargs during forward using the `reduce_kwargs` argument (dict). If those argument are static, consider defining a small function or using `functools.partial`. Only used in `'pw_mtx'` and `'pw_pt'` `pit_from` modes. For each of these modes, the best permutation and reordering will be automatically computed. When either ``'pw_mtx'`` or ``'pw_pt'`` is used, and the number of sources is larger than three, the hungarian algorithm is used to find the best permutation. Examples >>> import torch >>> from asteroid.losses import pairwise_neg_sisdr >>> sources = torch.randn(10, 3, 16000) >>> est_sources = torch.randn(10, 3, 16000) >>> # Compute PIT loss based on pairwise losses >>> loss_func = PITLossWrapper(pairwise_neg_sisdr, pit_from='pw_mtx') >>> loss_val = loss_func(est_sources, sources) >>> >>> # Using reduce >>> def reduce(perm_loss, src): >>> weighted = perm_loss * src.norm(dim=-1, keepdim=True) >>> return torch.mean(weighted, dim=-1) >>> >>> loss_func = PITLossWrapper(pairwise_neg_sisdr, pit_from='pw_mtx', >>> perm_reduce=reduce) >>> reduce_kwargs = {'src': sources} >>> loss_val = loss_func(est_sources, sources, >>> reduce_kwargs=reduce_kwargs) pw_mtxNcs2t||_||_||_|jdvrtddS)N)rpw_ptperm_avgzVUnsupported loss function type for now. Expectedone of [`pw_mtx`, `pw_pt`, `perm_avg`])super__init__ loss_funcpit_from perm_reduce ValueError)selfr r r __class__O/home/ubuntu/.local/lib/python3.10/site-packages/asteroid/losses/pit_wrapper.pyr @s  zPITLossWrapper.__init__Fc Ks6|jd}|dksJd||jdkr |j||fi|}n:|jdkr2|j|j||fi|}n(|jdkrX|j|j||fi|\}} t|} |sN| S||| } | | fSdS|jdkscJd |jd |jd ksqJd |durw|nt }|j |fd |j i|\}} t|} |s| S||| } | | fS) aFind the best permutation and return the loss. Args: est_targets: torch.Tensor. Expected shape $(batch, nsrc, ...)$. The batch of target estimates. targets: torch.Tensor. Expected shape $(batch, nsrc, ...)$. The batch of training targets return_est: Boolean. Whether to return the reordered targets estimates (To compute metrics or to save example). reduce_kwargs (dict or None): kwargs that will be passed to the pairwise losses reduce function (`perm_reduce`). **kwargs: additional keyword argument that will be passed to the loss function. Returns: - Best permutation loss for each batch sample, average over the batch. - The reordered targets estimates if ``return_est`` is True. :class:`torch.Tensor` of shape $(batch, nsrc, ...)$.  z(Expected source axis along dim 1, found rrr NzBSomething went wrong with the loss function, please read the docs.rz&PIT loss needs same batch dim as inputr) shaper r get_pw_lossesbest_perm_from_perm_avg_losstorchmeanreorder_sourcendimdictfind_best_permr) r est_targetstargets return_est reduce_kwargskwargsn_src pw_lossesmin_loss batch_indices mean_loss reorderedrrrforwardKsF         zPITLossWrapper.forwardc Ksr|j^}}}||||}t|ddD]!\}} t|ddD]\} } || | fi||dd|| f<q!q|S)aGet pair-wise losses between the training targets and its estimate for a given loss function. Args: loss_func: function with signature (est_targets, targets, **kwargs) The loss function to get pair-wise losses from. est_targets: torch.Tensor. Expected shape $(batch, nsrc, ...)$. The batch of target estimates. targets: torch.Tensor. Expected shape $(batch, nsrc, ...)$. The batch of training targets. **kwargs: additional keyword argument that will be passed to the loss function. Returns: torch.Tensor or size $(batch, nsrc, nsrc)$, losses computed for all permutations of the targets and est_targets. This function can be called on a loss function which returns a tensor of size :math:`(batch)`. There are more efficient ways to compute pair-wise losses using broadcasting. rrN)r new_empty enumerate transpose) r r!r"r% batch_sizer&_pair_wise_lossesest_idxest_src target_idx target_srcrrrrs "zPITLossWrapper.get_pw_lossesc  s~jd}tjttt|tjdtjfddDdd}tj|dd\}}tjfdd|Ddd}||fS)aFind best permutation from loss function with source axis. Args: loss_func: function with signature $(est_targets, targets, **kwargs)$ The loss function batch losses from. est_targets: torch.Tensor. Expected shape $(batch, nsrc, *)$. The batch of target estimates. targets: torch.Tensor. Expected shape $(batch, nsrc, *)$. The batch of training targets. **kwargs: additional keyword argument that will be passed to the loss function. Returns: - :class:`torch.Tensor`: The loss corresponding to the best permutation of size $(batch,)$. - :class:`torch.Tensor`: The indices of the best permutations. rdtypecs*g|]}dd|ffiqSNr).0perm)r!r%r r"rr s*z?PITLossWrapper.best_perm_from_perm_avg_loss..dimcg|]}|qSrrr:mpermsrrr<r) rrtensorlistrrangelongstackmin) r r!r"r%r&loss_setr( min_loss_idxr)r)r!r%r rCr"rrs z+PITLossWrapper.best_perm_from_perm_avg_losscKsR|jd}|dus |dkrtj|fd|i|\}}||fSt|\}}||fS)aFind the best permutation, given the pair-wise losses. Dispatch between factorial method if number of sources is small (<3) and hungarian method for more sources. If ``perm_reduce`` is not None, the factorial method is always used. Args: pair_wise_losses (:class:`torch.Tensor`): Tensor of shape :math:`(batch, n\_src, n\_src)`. Pairwise losses. perm_reduce (Callable): torch function to reduce permutation losses. Defaults to None (equivalent to mean). Signature of the func (pwl_set, **kwargs) : :math:`(B, n\_src!, n\_src) -> (B, n\_src!)` **kwargs: additional keyword argument that will be passed to the permutation reduce function. Returns: - :class:`torch.Tensor`: The loss corresponding to the best permutation of size $(batch,)$. - :class:`torch.Tensor`: The indices of the best permutations. Nrr)rrfind_best_perm_factorialfind_best_perm_hungarian)r2rr%r&r(r)rrrr s  zPITLossWrapper.find_best_permcCstddt||D}|S)ayReorder sources according to the best permutation. Args: source (torch.Tensor): Tensor of shape :math:`(batch, n_src, time)` batch_indices (torch.Tensor): Tensor of shape :math:`(batch, n_src)`. Contains optimal permutation indices for each batch. Returns: :class:`torch.Tensor`: Reordered sources. cSsg|] \}}t|d|qS)r)r index_select)r:sbrrrr<sz1PITLossWrapper.reorder_source..)rrIzip)sourcer)reordered_sourcesrrrrs zPITLossWrapper.reorder_sourcec  s|jd}|dd}|jttt|tjdtd}|durA| g |R d|d}t d||g}||}n|ddt ||df}||fi|}tj|dd\} } tjfd d | Dd d} | | fS) a.Find the best permutation given the pair-wise losses by looping through all the permutations. Args: pair_wise_losses (:class:`torch.Tensor`): Tensor of shape :math:`(batch, n_src, n_src)`. Pairwise losses. perm_reduce (Callable): torch function to reduce permutation losses. Defaults to None (equivalent to mean). Signature of the func (pwl_set, **kwargs) : :math:`(B, n\_src!, n\_src) -> (B, n\_src!)` **kwargs: additional keyword argument that will be passed to the permutation reduce function. Returns: - :class:`torch.Tensor`: The loss corresponding to the best permutation of size $(batch,)$. - :class:`torch.Tensor`: The indices of the best permutations. MIT Copyright (c) 2018 Kaituo XU. See `Original code `__ and `License `__. rMr7Nrz bij,pij->bpr=cr?rrr@rBrrr<&rDz;PITLossWrapper.find_best_perm_factorial..r)rr/ new_tensorrFrrGrrH unsqueeze new_zerossizescatter_einsumarangesqueezerJrI) r2rr%r&pwlidx perms_one_hotrKpwl_setr(rLr)rrBrrNs   " z'PITLossWrapper.find_best_perm_factorialr2cCsX|dd}|}tdd|D|j}t|d|dddg}||fS)aF Find the best permutation given the pair-wise losses, using the Hungarian algorithm. Returns: - :class:`torch.Tensor`: The loss corresponding to the best permutation of size (batch,). - :class:`torch.Tensor`: The indices of the best permutations. rMrVcSsg|]}t|dqS)rr)r:r`rrrr<:sz;PITLossWrapper.find_best_perm_hungarian..rW).N) r/detachcpurrEtodevicegatherr)r2r`pwl_copyr)r(rrrrO)s z'PITLossWrapper.find_best_perm_hungarian)rN)FNr9)__name__ __module__ __qualname____doc__r r, staticmethodrrr rrNrTensorrO __classcell__rrrrrs 8 ;     2rcs"eZdZdZdfdd ZZS) PITReorderzzPermutation invariant reorderer. Only returns the reordered estimates. See `:py:class:asteroid.losses.PITLossWrapper`.Nc s$tjd||d|d|\}}|S)NT)r!r"r#r$r)r r,)rr!r"r$r%r1r+rrrr,Es zPITReorder.forwardr9)rjrkrlrmr,rprrrrrqAsrq) itertoolsrrrscipy.optimizerModulerrqrrrrs   <