class flamo.processor.system.Series(*args)

Module for cascading multiple DSP modules in series. Inherits from nn.Sequential. This class serves as a container for a series of DSP modules (preferably constructed from flamo.processor.dsp.DSP), allowing them to be cascaded in a single module. It ensures that all included modules share the same values of nfft and alias_decay_db attributes, hence all parsed modules are expected to have these attributes.

Arguments:

*args: An arbitrary number of DSP modules of the type nn.Module, nn.Sequential, or OrderedDict.

__check_attribute(attr: str) → int | float | None

Checks if all modules have the same value of the requested attribute.

Arguments:

attr (str): The attribute to check.

Returns:

int | float | None: The attribute value.

Raises:

ValueError: If at least one module with incorrect attribute value was found.

__check_io()

Checks if the modules in the Series have compatible input/output shapes.

Returns:

tuple(int,int): The number of input and output channels.

__unpack_modules(modules: tuple, current_keys: list) → OrderedDict

Generate an OrderedDict containing the modules given in the input tuple, and give a key to each module. If any module in the input tuple is a nn.Sequential (or OrderedDict), it is recognized as nested nn.Sequential (or OrderedDict), it is unpacked, and its modules are added to the output OrderedDict. For each encountered module, key generation follows these rules:

  1. if the analyzed module is nn.Sequential or OrderedDict, it is unpacked, thus, if it was nested inside

    another nn.Sequential or OrderedDict, the key given to itself is neglected. This until the analyzed module is a leaf module (i.e. nn.Module and not nn.Sequential or OrderedDict), then rules 2-5 are applied

  2. if the module has a unique custom key (e.g. ‘my_module’), it is used as is

  3. if the module has a custom key, but such key was previously used for another module (hence it is not unique). An error is raised

  4. if the module has no key, a key is generated for it, equal to its position in the series. A warning is raised

  5. if a custom key is found and can be converted to an integer (e.g. ‘3’), such key is considered missing and rule 4 is applied

Arguments:

  • modules (tuple): The input modules.

  • current_keys (list): The current keys of the already unpacked modules.

Returns:

The unpacked modules (OrderedDict).

Raises:

  • ValueError: If modules are not of type nn.Module, nn.Sequential, or OrderedDict.

  • ValueError: If a custom key is not unique.

append(new_module: Module | Sequential | OrderedDict) → Series

Appends a given item to the end of the Series instance.

Arguments:

new_module (nn.Module | nn.Sequential | OrderedDict): item to append.

Returns:

Series: self.

forward(input, ext_param=None)

Forward pass through the Series.

Arguments:

  • input (Tensor): The input tensor.

  • ext_param (torch.Tensor, optional): Parameter values received from external modules (hyper conditioning). Default: None.

Returns:

Tensor: The output tensor.

insert(index: int, new_module: Module | Sequential | OrderedDict) → Series

Inserts a given item at the given index in the Series instance.

Arguments:

  • index (int): index at which to insert the new module.

  • new_module (nn.Module | nn.Sequential | OrderedDict): item to append.

Returns:

Series: self.

prepend(new_module: Module | Sequential | OrderedDict) → Series

Prepends a given item to the beginning of the Series instance.

Arguments:

new_module (nn.Module | nn.Sequential | OrderedDict): item to append.

Returns:

Series: self.

class flamo.processor.system.Recursion(fF: Module | Sequential | OrderedDict | Series, fB: Module | Sequential | OrderedDict | Series)

Recursion module for computing closed-loop transfer function. Inherits from nn.Module. The feedforward and feedback paths if are given as a nn.Module, nn.Sequential, or OrderedDict, they are converted to a Series instance.

Shape:

  • input: \((B, M, N_{in}, ...)\)

  • output: \((B, M, N_{out}, ...)\)

where \(B\) is the batch size, \(M\) is the number of frequency bins, \(N_{in}\) is the number of input channels, and \(N_{out}\) is the number of output channels. Ellipsis \((...)\) represents additional dimensions.

Arguments:

  • fF: The feedforward path with size (M, N_{out}, N_{in}).

  • fB: The feedback path with size (M, N_{in}, N_{out}).

  • alias_decay_db (float, optional): The decaying factor in dB for the time anti-aliasing envelope. The decay refers to the attenuation after nfft samples. Defaults to None.

Attributes:

  • feedforward (nn.Module | Series): The feedforward path.

  • feedback (nn.Module | Series): The feedback path.

  • nfft (int): The number of frequency points.

  • alias_decay_db (float): The decaying factor in dB for the time anti-aliasing envelope. The decay refers to the attenuation after nfft samples.

For details on the closed-loop transfer function see Wikipedia page.

__check_attribute(attr: str) → int | float

Checks if feedforward and feedback paths have the same value of the requested attribute.

Arguments:

attr (str): The attribute to check.

Returns:

int | float: The attribute value.

Raises:

ValueError: The two paths have different values of the requested attribute.

__check_io() → tuple

Checks if the feedforward and feedback paths have compatible input/output shapes.

Returns:

tuple(int,int): The number of input and output channels.

Raises:

  • ValueError: The feedforward or the feedback paths do not possess either the input_channels or the output_channels attributes.

  • AssertionError: The feedforward and the feedback paths’ input and output channels are not compatible.

__generate_identity() → Tensor

Generates the identity matrix necessary for the forward computation.

Returns:

torch.Tensor: The identity matrix.

forward(X)

Applies the closed-loop transfer function to the input tensor X.

Arguments:

X (torch.Tensor): Input tensor of shape \((B, M, N_{in}, ...)\).

Returns:

torch.Tensor: Output tensor of shape \((B, M, N_{out}, ...)\).

class flamo.processor.system.Shell(core: Module | Recursion | Sequential, input_layer: Recursion | Series | Module = Identity(), output_layer: Recursion | Series | Module = Identity())

DSP wrapper class. Interfaces the DSP with dataset and loss function. Inherits from nn.Module.

Shape:

  • input: \((B, M, N_{in}, ...)\)

  • output: \((B, M, N_{out}, ...)\)

where \(B\) is the batch size, \(M\) is the number of frequency bins, \(N_{in}\) is the number of input channels (defined by the core and the input_layer), and \(N_{out}\) is the number of output channels (defined by the core and the output_layer). Ellipsis \((...)\) represents additional dimensions.

Arguments / Attributes:

  • core (nn.Module | nn.Sequential): DSP.

  • input_layer (nn.Module, optional): layer preceeding the DSP and correctly preparing the Dataset input before the DSP processing. Default: Transform(lambda x: x).

  • output_layer (nn.Module, optional): layer following the DSP and preparing its output for the comparison with the Dataset target. Default: Transform(lambda x: x).

Attributes:

  • nfft (int): Number of frequency points.

  • alias_decay_db (float): The decaying factor in dB for the time anti-aliasing envelope. The decay refers to the attenuation after nfft samples.

__check_attribute(attr: str) → int | float

Check if all the modules in core, input layer, and output layer have the same value for the requested attribute.

Argument:

  • attr (str): The attribute to check.

Returns:

int: The attribute value.

Raises:

  • ValueError: The core component does not possess the requested attribute.

  • AssertionError: Core, input layer, and output layer do not have the same value of the requested attribute.

__check_io() → tuple

Checks if the input layer, the core, and the output layer have compatible input/output shapes.

Returns:

tuple(int,int): The number of input and output channels.

Raises:

ValueError: The core, the input layer, and the output layer are not I/O compatible.

forward(x: Tensor, ext_param: dict = None) → Tensor

Forward pass through the input layer, the core, and the output layer. Keeps the three components separated.

Args:

  • x (torch.Tensor): Input tensor of shape \((B, M, N_{in}, ...)\).

Returns:

  • torch.Tensor: Output tensor of shape \((B, M, N_{out}, ...)\).

get_core() → Module | Sequential

Returns the current core DSP.

get_freq_response(fs: int = 48000, identity: bool = False) → Tensor

Generates the frequency response of the DSP.

Arguments:

  • fs (int, optional): Sampling frequency. Defaults to 48000.

  • identity (bool, optional): If False, return the input-to-output frequency responses of the DSP.

    If True, return the input-free frequency responses of the DSP. Defaults to False.

NOTE: Definition of ‘input-to-output’ and ‘input-free’

Let \(A \in \mathbb{R}^{F \times N_{out} \times N_{in}}\) be a frequency filter matrix. If \(x \in \mathbb{R}^{T \times N_{in}}\) is an \(N_{in}\)-dimensional signal having a unit impulse at time \(t=0\) spectrum for each element along \(N_{in}\). Let \(I \in R^{F \times N \times N}\) be an diagonal matrix across second and third dimension, with unit impulse at time \(t=0\) spectra for each element along such diagonal. If * represent the signal-wise matrix product operator, then:

  • \(y = A * x\) is the ‘input-to-output’ frequency response of \(A\).

  • \(A * I\) is the ‘input-free’ frequency response of \(A\).

Returns:

torch.Tensor: Generated DSP frequency response.

get_inputLayer() → Module | Sequential

Returns the current input layer.

get_outputLayer() → Module | Sequential

Returns the current output layer.

get_time_response(fs: int = 48000, identity: bool = False) → Tensor

Generates the impulse response of the DSP.

Arguments:

  • fs (int, optional): Sampling frequency. Defaults to 48000.

  • identity (bool, optional): If False, return the input-to-output impulse responses of the DSP.

    If True, return the input-free impulse responses of the DSP. Defaults to False.

NOTE: Definition of ‘input-to-output’ and ‘input-free’

Let \(A \in \mathbb{R}^{T \times N_{out} \times N_{in}}\) be a time filter matrix. If \(x \in \mathbb{R}^{T \times N_{in}}\) is an \(N_{in}\)-dimensional time signal having a unit impulse at time \(t=0\) for each element along \(N_{in}\). Let \(I \in R^{T \times N \times N}\) be an diagonal matrix across second and third dimension, with unit impulse at time :math:`t=0`for each element along such diagonal. If * represent the signal-wise matrix convolution operator, then:

  • \(y = A * x\) is the ‘input-to-output’ impulse response of \(A\).

  • \(A * I\) is the ‘input-free’ impulse response of \(A\).

Returns:

  • torch.Tensor: Generated DSP impulse response.

set_core(core: Module) → None
Substitutes the current core with a given new one.
Argument:

output_layer (nn.Module, optional): The core DSP system. Defaults to None.

set_inputLayer(input_layer: Module = None) → None
Substitutes the current input layer with a given new one.
Argument:

input_layer (nn.Module | nn.Sequential, optional): The new input layer. Defaults to None.

set_outputLayer(output_layer: Module = None) → None
Substitutes the current output layer with a given new one.
Argument:

output_layer (nn.Module | nn.Sequential, optional): The new input layer. Defaults to None.