acoupipe.pipeline#

All classes in this module can be used to calculate and provide data.

Purpose of the Pipeline Module#

Classes defined in the pipeline.py module have the ability to iteratively perform tasks on the related computational pipeline to build up a dataset. The results of these tasks are the features (and labels) associated with a specific sample of the dataset. Feature creation tasks can be specified by passing callable functions that are evoked at each iteration of the BasePipeline’s get_data() generator method. It is worth noting that such a data generator can also be used directly to feed a machine learning model without saving the data to file, as common machine learning frameworks, such as Tensorflow, offer the possibility to consume data from Python generators. Control of the state of the sampling process is maintained via the sampler attribute holding a list of BaseSampler derived instances.

import acoular as ac
from acoupipe.sampler import NumericAttributeSampler
from acoupipe.pipeline import BasePipeline, DistributedPipeline
from scipy.stats import norm

random_var = norm(loc=1.0, scale=0.5)

n1 = ac.WNoiseGenerator(sample_freq=24000, numsamples=24000 * 5, rms=1.0, seed=1)

rms_sampler = NumericAttributeSampler(target=[n1], attribute='rms', random_var=random_var)


def calculate_squared_rms(sampler):
    n1 = sampler[0].target[0]
    return {'rms_sq': n1.rms**2}


pipeline = BasePipeline(
    numsamples=5,
    # random_seeds = [range(5)],
    sampler=[rms_sampler],
    features=calculate_squared_rms,
)

for data in pipeline.get_data():
    print(data)

The example returns the following output:

100%|██████████| 5/5 [00:00<00:00, 2250.65it/s]
{'idx': 0, 'seeds': array([[0, 0]]), 'rms_sq': 1.1296822432174416}
{'idx': 1, 'seeds': array([[0, 1]]), 'rms_sq': 1.3754413005160537}
{'idx': 2, 'seeds': array([[0, 2]]), 'rms_sq': 1.197988677085426}
{'idx': 3, 'seeds': array([[0, 3]]), 'rms_sq': 4.082256836394099}
{'idx': 4, 'seeds': array([[0, 4]]), 'rms_sq': 0.454416774044029}

Module Contents#

class acoupipe.pipeline.BaseSampler#

Bases: traits.api.ABCHasStrictTraits

Base class that represents a random process.

This class has no functionality and should not be used in practice. It provides a common interface for all classes that manipulate attributes of an instance or a list of instances according to a specified random distribution.

rvs(size=1)#

Random variable sampling (for internal use).

Parameters:

sizeint, optional: The size of the output array. Defaults to 1.

Returns:

array-like: Random values drawn from the random distribution.

abstractmethod sample()#

Utilizes rvs() function to draw random values from the random distribution.

This method needs to be implemented by derived classes to perform the actual sampling and assign values to the target instances.

class acoupipe.pipeline.DataGenerator#

Bases: traits.api.ABCHasStrictTraits

Abstract base class that serves as a data generator.

This class should not be used directly. It provides a common interface for all classes that generate data via the result() method in a block-wise or sample-wise manner.

abstractmethod get_data()#

Python generator that iteratively yields data set samples as a dictionary.

This method needs to be implemented by derived classes.

Returns:

dict: Dictionary containing a sample of the data set {feature_name[key] : feature[values]}.

class acoupipe.pipeline.BasePipeline#

Bases: acoupipe.base.DataGenerator

Control the random process and iteratively extract and pass a specified amount of data.

This class can be used to calculate data (extract features) by assigning a name and a callable function to features. Furthermore this class automatically controles the sampling of instances of type BaseSampler specified to the sampler list. Re-seeding is performed at each iteration if random_seeds are given.

validate_random_seeds()#: Validate specified random seeds.

get_data(progress_bar=True, start_idx=1)#

Provide the extracted features, sampler seeds and indices.

Parameters:

progress_barbool, optional: if True, a progress bar is displayed, by default True
start_idxint, optional: the index of the first data sample to be calculated, by default 1

Yields:

dict: a sample of the dataset containing the extracted feature data, seeds, and index

class acoupipe.pipeline.DistributedPipeline#

Bases: BasePipeline

Class to calculate data (extract features) in parallel to build large datasets.

This class can be used to calculate data (extract various features) by assigning a name and a callable function to features. Furthermore this class automatically controles the sampling of instances of type BaseSampler specified to the sampler list. Re-seeding is performed at each iteration if random_seeds are given.

get_data(progress_bar=True, start_idx=1)#

Provide the extracted features, sampler seeds and indices.

The calculation of all data samples is performed in parallel and asynchronously. In case of specifying more than one worker in the numworker attribute, the output of this generator yields non-ordered features/data samples. However, the exact order can be recovered via the “idx” item (or “seeds” item) provided in the output dictionary.

Parameters:

progress_barbool, optional: if True, a progress bar is displayed, by default True
start_idxint, optional: the index of the first data sample to be calculated, by default 1

Yields:

dict: A sample of the dataset containing the extracted feature data, seeds, and index