def apply()

in augment/effects.py [0:0]

• 39 lines of code
• 17 McCabe index (conditional complexity)

    def apply(self,
              input_tensor: torch.Tensor,
              src_info: Dict[str, Union[int, float]],
              target_info: Optional[Dict[str, Union[int, float]]] = None) -> torch.Tensor:
        """
        input_tensor (torch.Tensor): the input wave to be transformed;
            expected shape is (n_channels, length). If it has only
            one dimension, it is automatically expanded to have 1 channel.
        src_info (Dict): description of the input signal
        target_info (Dict): description of the output signal

        Fields that src_info and target_info can contain:
            * rate (mandatory for input),
            * length,
            * precision,
            * bits_per_sample.
        Those fields are only used by sox-based effects.

        Minimally, src_info must contain rate field (e.g. `{rate: 16_000}`).
        Both src_info and target_info can set `length`. If src_info specifies
        `length`, only first `length` samples are used; if target_info has `length`,
        further output is trimmed.

        It is might happen that sox will return wave shorter than `length` - in this
        case the output will be padded with zeroes.

        returns: torch.Tensor with transformed sound.
        """
        target_info = dict() if target_info is None else target_info

        if not torch.is_tensor(input_tensor) or input_tensor.is_cuda:
            raise RuntimeError('Expected a CPU tensor')

        if not self.in_place:
            input_tensor = input_tensor.clone()
        if 'rate' not in src_info:
            raise RuntimeError("'rate' must be specified for the input")
        if len(input_tensor.size()) == 1:
            input_tensor = input_tensor.unsqueeze(0)
        if 'length' in src_info and src_info['length'] > input_tensor.size(1):
            raise RuntimeError("'length' is beyond the tensor length")

        src_info = dict(src_info)  # can be mutated in process
        sr = src_info['rate']

        if not self._chain:
            out = input_tensor
            return out

        sox_effects: List[SoxEffect] = []
        x = input_tensor

        # To minimize the number of sox calls, we stack consequent sox effects together in a single
        # sox-side chain. In contrast, we apply python effects immediately.

        for effect in self._chain:
            if callable(effect):
                if sox_effects:
                    x, sr = EffectChain._apply_sox_effects(
                        sox_effects, x, src_info, target_info)
                    src_info = dict(target_info)
                    assert src_info['rate'] == sr

                    sox_effects = []
                # effect should not mutate src_info or target_info, but
                # return new ones if changed
                x, src_info, target_info = effect(x, src_info, target_info)
            elif isinstance(effect, SoxEffect):
                sox_effects.append(effect)
            else:
                assert False

        if sox_effects:
            x, _ = EffectChain._apply_sox_effects(
                sox_effects, x, src_info, target_info)
        return x