`tprocess`#

Implement blockwise processing in the time domain.

`MaskedTimeOut`	A signal processing block that allows for the selection of specific channels and time samples.
`Trigger`	A signal processing class for detecting and analyzing trigger signals in time-series data.
`AngleTracker`	Compute the rotational angle and RPM per sample from a trigger signal in the time domain.
`ChannelMixer`	A signal processing block that mixes multiple input channels into a single output channel.
`SpatialInterpolator`	Base class for spatial interpolation of microphone data.
`SpatialInterpolatorRotation`	Spatial interpolation class for rotating sound sources.
`SpatialInterpolatorConstantRotation`	Performs spatial linear interpolation for sources undergoing constant rotation.
`Mixer`	Mix signals from multiple sources into a single output.
`TimePower`	Calculate the time-dependent power of a signal by squaring its samples.
`TimeCumAverage`	Calculates the cumulative average of the signal.
`TimeReverse`	Calculates the time-reversed signal of a source.
`Filter`	Abstract base class for IIR filters using SciPy's `lfilter()`.
`FilterBank`	Abstract base class for IIR filter banks based on `scipy.signal.lfilter`.
`FiltFiltOctave`	Octave or third-octave bandpass filter with zero-phase distortion.
`FiltOctave`	Octave or third-octave bandpass filter (causal, with non-zero phase delay).
`TimeExpAverage`	Compute an exponentially weighted moving average of the input signal.
`FiltFreqWeight`	Apply frequency weighting according to IEC 61672-1.
`OctaveFilterBank`	Octave or third-octave filter bank.
`WriteWAV`	Saves time signal from one or more channels as mono, stereo, or multi-channel `.wav` file.
`WriteH5`	Saves time signal data as a `.h5` (HDF5) file.
`TimeConvolve`	Perform frequency domain convolution with the uniformly partitioned overlap-save (UPOLS) method.

class acoular.tprocess.MaskedTimeOut#

Bases: TimeOut

A signal processing block that allows for the selection of specific channels and time samples.

The MaskedTimeOut class is designed to filter data from a given SamplesGenerator (or a derived object) by defining valid time samples and excluding specific channels. It acts as an intermediary between the data source and subsequent processing steps, ensuring that only the selected portion of the data is passed along.

This class is useful for selecting specific portions of data for analysis. The processed data is accessed through the generator method result(), which returns data in block-wise fashion for efficient streaming.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class. This object provides the raw time-domain signals that will be filtered based on the start, stop, and invalid_channels attributes.

start = CInt(0, desc='start of valid samples')#: The index of the first valid sample. Default is 0.

stop = Union(None, CInt, desc='stop of valid samples')#: The index of the last valid sample (exclusive). If set to None, the selection continues until the end of the available data.

invalid_channels = List(int, desc='list of invalid channels')#: List of channel indices to be excluded from processing.

channels = Property(depends_on=['invalid_channels', 'source.num_channels'], desc='channel mask')#: A mask or index array representing valid channels. (automatically updated)

num_channels_total = Delegate('source', 'num_channels')#: Total number of input channels, including invalid channels, as given by source. (read-only).

num_samples_total = Delegate('source', 'num_samples')#: Total number of input channels, including invalid channels. (read-only).

num_channels = Property( …#: Number of valid input channels after excluding invalid_channels. (read-only)

num_samples = Property( …#: Number of valid time-domain samples, based on start and stop indices. (read-only)

basename = Property(depends_on=['source.digest'], desc='basename for cache file')#: The name of the cache file (without extension). It serves as an internal reference for data caching and tracking processed files. (automatically generated)

digest = Property(depends_on=['source.digest', 'start', 'stop', 'invalid_channels'])#: A unique identifier for the object, based on its properties. (read-only)

result(num)#

Generate blocks of processed data, selecting only valid samples and channels.

This method fetches data from the source object, applies the defined start and stop constraints on time samples, and filters out invalid_channels. The data is then yielded in block-wise fashion to facilitate efficient streaming.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array of shape (num, MaskedTimeOut.num_channels), contatining blocks of a filtered time-domain signal. The last block may contain fewer samples if the total number of samples is not a multiple of num. MaskedTimeOut.num_channels is not inherited directly and may be smaller than the source’s number of channels.

Raises:

OSError: If no valid samples are available within the defined start and stop range. This can occur if start is greater than or equal to stop or if the source is not containing any valid samples in the given range.

class acoular.tprocess.ChannelMixer#

Bases: TimeOut

A signal processing block that mixes multiple input channels into a single output channel.

The ChannelMixer class takes a multi-channel signal from a SamplesGenerator (or a derived object) and applies an optional set of amplitude weights to each channel. The resulting weighted sum is then output as a single-channel signal.

This class is particularly useful for cases where a combined signal representation is needed, such as beamforming, array signal processing, or for reducing the dimensionality of multi-channel time signal data.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class. It provides the multi-channel time-domain signals that will be mixed.

weights = CArray(desc='channel weights')#: An array of amplitude weight factors applied to each input channel before summation. If not explicitly set, all channels are weighted equally (delault is 1). The shape of weights must match the number of input channels. If an incompatible shape is provided, a ValueError will be raised.

num_channels = Constant(1)#: The number of output channels, which is always 1 for this class since it produces a single mixed output. (read-only)

digest = Property(depends_on=['source.digest', 'weights'])#: A unique identifier for the object, based on its properties. (read-only)

result(num)#

Generate the mixed output signal in blocks.

This method retrieves data from the source object, applies the specified amplitude weights to each channel, and sums them to produce a single-channel output. The data is processed and yielded in block-wise fashion for efficient memory handling.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array of shape (num, 1) containing blocks a of single-channel mixed signal. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Raises:

ValueError: If the weights array is provided but its shape does not match the expected shape (num_channels,) or (1,), a ValueError is raised indicating that the weights cannot be broadcasted properly.

class acoular.tprocess.Trigger#

Bases: TimeOut

A signal processing class for detecting and analyzing trigger signals in time-series data.

The Trigger class identifies trigger events in a single-channel signal provided by a SamplesGenerator source. The detection process involves:

Identifying peaks that exceed a specified positive or negative threshold.
Estimating the approximate duration of one revolution based on the largest sample distance between consecutive peaks.
Dividing the estimated revolution duration into segments called “hunks,” allowing only one peak per hunk.
Selecting the most appropriate peak per hunk based on a chosen criterion (e.g., first occurrence or extremum value).
Validating the consistency of the detected peaks by ensuring the revolutions have a stable duration with minimal variation.

This class is typically used for rotational speed analysis, where trigger events correspond to periodic markers in a signal (e.g., TDC signals in engine diagnostics).

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class. The signal must be single-channel.

threshold = Union(None, Float)#

The threshold value for detecting trigger peaks. The meaning of this threshold depends on the trigger type (:attr;`trigger_type`). The sign is relevant:

A positive threshold detects peaks above this value.
A negative threshold detects peaks below this value.

If None, an estimated threshold is used, calculated as 75% of the extreme deviation from the mean signal value. Default is None.

E.g: If the mean value is \(0\) and there are positive extrema at \(400\) and negative extrema at \(-800\). Then the estimated threshold would be \(0.75 \cdot (-800) = -600\).

max_variation_of_duration = Float(0.02)#: The maximum allowable variation in duration between two trigger instances. If any revolution exceeds this variation threshold, a warning is issued. Default is 0.02.

hunk_length = Float(0.1)#: Defines the length of “hunks” as a fraction of the estimated duration between two trigger instances. If multiple peaks occur within a hunk, only one is retained based on multiple_peaks_in_hunk. Default is 0.1.

trigger_type = Enum('dirac', 'rect')#

Specifies the type of trigger detection:

'dirac': A single impulse is considered a trigger. The sign of threshold determines whether positive or negative peaks are detected.
'rect': A repeating rectangular waveform is assumed. Only every second edge is considered a trigger. The sign of threshold determines whether rising (+) or falling (-) edges are used.

Default is 'dirac'.

multiple_peaks_in_hunk = Enum('extremum', 'first')#

Defines the criterion for selecting a peak when multiple occur within a hunk (see hunk_length):

'extremum': Selects the most extreme peak.
'first': Selects the first peak encountered.

Default is 'extremum'.

trigger_data = Property( …#

A tuple containing:

A numpy.ndarray of sample indices corresponding to detected trigger events.
The maximum number of samples between consecutive trigger peaks.
The minimum number of samples between consecutive trigger peaks.

digest = Property( …#: A unique identifier for the trigger, based on its properties. (read-only)

result(num)#

Generate signal data from the source without modification.

This method acts as a pass-through, providing data blocks directly from the source generator. It is included for interface consistency but does not apply trigger-based transformations to the data.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing num samples from the source signal. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Warning

This method is not implemented for trigger-based transformations. A warning is issued, indicating that data is passed unprocessed.

class acoular.tprocess.AngleTracker#

Bases: MaskedTimeOut

Compute the rotational angle and RPM per sample from a trigger signal in the time domain.

This class retrieves samples from the specified trigger signal and interpolates angular position and rotational speed. The results are stored in the properties angle and rpm.

The algorithm assumes a periodic trigger signal marking rotational events (e.g., a tachometer pulse or an encoder signal) and interpolates the angle and RPM using cubic splines. It is capable of handling different rotational directions and numbers of triggers per revolution.

trigger = Instance(Trigger)#: Trigger data source, expected to be an instance of Trigger.

digest = Property( …#: A unique identifier for the tracker, based on its properties. (read-only)

trigger_per_revo = Int(1, desc='trigger signals per revolution')#: Number of trigger signals per revolution. This allows tracking scenarios where multiple trigger pulses occur per rotation. Default is 1, meaning a single trigger per revolution.

rot_direction = Int(-1, desc='mathematical direction of rotation')#

Rotation direction flag:

1: counter-clockwise rotation.
-1: clockwise rotation.

Default is -1.

interp_points = Int(4, desc='Points of interpolation used for spline')#: Number of points used for spline interpolation. Default is 4.

start_angle = Float(0, desc='rotation angle for trigger position')#: Initial rotation angle (in radians) corresponding to the first trigger event. This allows defining a custom starting reference angle. Default is 0.

rpm = Property(depends_on=['digest'], desc='revolutions per minute for each sample')#: Revolutions per minute (RPM) computed for each sample. It is based on the trigger data. (read-only)

average_rpm = Property(depends_on=['digest'], desc='average revolutions per minute')#: Average revolutions per minute over the entire dataset. It is computed based on the trigger intervals. (read-only)

angle = Property(depends_on=['digest'], desc='rotation angle for each sample')#: Computed rotation angle (in radians) for each sample. It is interpolated from the trigger data. (read-only)

class acoular.tprocess.SpatialInterpolator#

Bases: TimeOut

Base class for spatial interpolation of microphone data.

This class retrieves samples from a specified source and performs spatial interpolation to generate output at virtual microphone positions. The interpolation is executed using various methods such as linear, spline, radial basis function (RBF), and inverse distance weighting (IDW).

See also

SpatialInterpolatorRotation: Spatial interpolation class for rotating sound sources.
SpatialInterpolatorConstantRotation: Performs spatial linear interpolation for sources undergoing constant rotation.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class. It provides the time-domain pressure samples from microphones.

mics = Instance(MicGeom(), desc='microphone geometry')#: The physical microphone geometry. An instance of MicGeom that defines the positions of the real microphones used for measurement.

mics_virtual = Property(desc='microphone geometry')#: The virtual microphone geometry. This property defines the positions of virtual microphones where interpolated pressure values are computed. Default is the physical microphone geometry (mics).

method = Enum( …#

Interpolation method used for spatial data estimation.

Options:

'linear': Uses NumPy linear interpolation.
'spline': Uses SciPy’s CubicSpline interpolator
'rbf-multiquadric': Radial basis function (RBF) interpolation with a multiquadric kernel.
'rbf-cubic': RBF interpolation with a cubic kernel.
'IDW': Inverse distance weighting interpolation.
'custom': Allows user-defined interpolation methods.
'sinc': Uses sinc-based interpolation for signal reconstruction.

array_dimension = Enum('1D', '2D', 'ring', '3D', 'custom', desc='spatial dimensionality of the array geometry')#

Defines the spatial dimensionality of the microphone array.

Possible values:

'1D': Linear microphone arrays.
'2D': Planar microphone arrays.
'ring': Circular arrays where rotation needs to be considered.
'3D': Three-dimensional microphone distributions.
'custom': User-defined microphone arrangements.

sample_freq = Delegate('source', 'sample_freq')#: Sampling frequency of the output signal, inherited from the source. This defines the rate at which microphone pressure samples are acquired and processed.

num_channels = Property()#: Number of channels in the output data. This corresponds to the number of virtual microphone positions where interpolated pressure values are computed. The value is ´determined based on the mics_virtual geometry.

num_samples = Delegate('source', 'num_samples')#: Number of time-domain samples in the output signal, inherited from the source.

interp_at_zero = Bool(False)#: Whether to interpolate a virtual microphone at the origin. If set to True, an additional virtual microphone position at the coordinate origin \((0,0,0)\) will be interpolated.

Q = CArray(dtype=np.float64, shape=(3, 3), value=np.identity(3))#

Transformation matrix for coordinate system alignment.

This 3x3 orthogonal matrix is used to align the microphone coordinates such that rotations occur around the z-axis. If the original coordinates do not conform to the expected alignment (where the x-axis transitions into the y-axis upon rotation), applying this matrix modifies the coordinates accordingly. The transformation is defined as

\[\begin{split}\begin{bmatrix}x'\\y'\\z'\end{bmatrix} = Q \cdot \begin{bmatrix}x\\y\\z\end{bmatrix}\end{split}\]

where \(Q\) is the transformation matrix and \((x', y', z')\) are the modified coordinates. If no transformation is needed, \(Q\) defaults to the identity matrix.

num_IDW = Int(3, desc='number of neighboring microphones, DEFAULT=3') # noqa: N815#: Number of neighboring microphones used in IDW interpolation. This parameter determines how many physical microphones contribute to the weighted sum in inverse distance weighting (IDW) interpolation.

p_weight = Float( …#: Weighting exponent for IDW interpolation. This parameter controls the influence of distance in inverse distance weighting (IDW). A higher value gives more weight to closer microphones.

digest = Property( …#: Unique identifier for the current configuration of the interpolator. (read-only)

sinc_mic(r)#

Compute a modified sinc function for use in Radial Basis Function (RBF) approximation.

This function is used as a kernel in sinc-based interpolation methods, where the sinc function serves as a basis function for reconstructing signals based on spatially distributed microphone data. The function is scaled according to the number of virtual microphone positions, ensuring accurate signal approximation.

Parameters:

rfloat or list of floats: The radial distance(s) at which to evaluate the sinc function, typically representing the spatial separation between real and virtual microphone positions.

Returns:

numpy.ndarray: Evaluated sinc function values at the given radial distances.

result(num)#

Generate interpolated microphone data over time.

This method retrieves pressure samples from the physical microphones and applies spatial interpolation to estimate the pressure at virtual microphone locations. The interpolation method is determined by method.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array of shape (num, n), where n is the number of virtual microphones, containing interpolated pressure values for the virtual microphones at each time step. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.SpatialInterpolatorRotation#

Bases: SpatialInterpolator

Spatial interpolation class for rotating sound sources.

This class extends SpatialInterpolator to handle sources that undergo rotational movement. It retrieves samples from the source attribute and angle data from the AngleTracker instance (angle_source). Using these inputs, it computes interpolated outputs through the result() generator method.

See also

SpatialInterpolator: Base class for spatial interpolation of microphone data.

angle_source = Instance(AngleTracker)#: Provides real-time tracking of the source’s rotation angles, instance of AngleTracker.

digest = Property( …#: Unique identifier for the current configuration of the interpolator. (read-only)

result(num=128)#

Generate interpolated output samples in block-wise fashion.

This method acts as a generator, yielding time-domain time signal samples that have been spatially interpolated based on rotational movement.

Parameters:

numint, optional: Number of samples per block. Default is 128.

Yields:

numpy.ndarray: Interpolated time signal samples in blocks of shape (num, num_channels), where num_channels is inherited from the SpatialInterpolator base class. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.SpatialInterpolatorConstantRotation#

Bases: SpatialInterpolator

Performs spatial linear interpolation for sources undergoing constant rotation.

This class interpolates signals from a rotating sound source based on a constant rotational speed. It retrieves samples from the source and applies interpolation before generating output through the result() generator.

See also

SpatialInterpolator: Base class for spatial interpolation of microphone data.
SpatialInterpolatorRotation: Spatial interpolation class for rotating sound sources.

rotational_speed = Float(0.0)#: Rotational speed of the source in revolutions per second (rps). A positive value indicates counterclockwise rotation around the positive z-axis, meaning motion from the x-axis toward the y-axis.

digest = Property( …#: Unique identifier for the current configuration of the interpolator. (read-only)

result(num=1)#

Generate interpolated time signal data in blocks of size num.

This generator method continuously processes incoming time signal data while applying rotational interpolation. The phase delay is computed based on the rotational speed and applied to the signal.

Parameters:

numint, optional: Number of samples per block. Default is 1.

Yields:

numpy.ndarray: An array containing the interpolated time signal samples in blocks of shape (num, num_channels), where num_channels is inherited from the SpatialInterpolator base class. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.Mixer#

Bases: TimeOut

Mix signals from multiple sources into a single output.

This class takes a primary time signal source and a list of additional sources with the same sampling rates and channel counts across all primary time signal source, and outputs a mixed signal. The mixing process is performed block-wise using a generator.

If one of the additional sources holds a shorter signal than the other sources the result() method will stop yielding mixed time signal at that point.

source = Instance(SamplesGenerator)#: The primary time signal source. It must be an instance of a SamplesGenerator-derived class.

sources = List(Instance(SamplesGenerator, ()))#: A list of additional time signal sources to be mixed with the primary source, each must be an instance of SamplesGenerator.

sample_freq = Delegate('source')#: The sampling frequency of the primary time signal, delegated from source.

num_channels = Delegate('source')#: The number of channels in the output, delegated from source.

num_samples = Delegate('source')#: The number of samples in the output, delegated from source.

sdigest = Str()#: Internal identifier that tracks changes in the sources list.

digest = Property(depends_on=['source.digest', 'sdigest'])#: A unique identifier for the Mixer instance, based on the primary source and the list of additional sources.

validate_sources()#

Validate whether the additional sources are compatible with the primary source.

This method checks if all sources have the same sampling frequency and the same number of channels. If a mismatch is detected, a ValueError is raised.

Raises:

ValueError: If any source in sources has a different sampling frequency or number of channels than source.

result(num)#

Generate mixed time signal data in blocks of num samples.

This generator method retrieves time signal data from all sources and sums them together to produce a combined output. The data from each source is processed in blocks of the same size, ensuring synchronized mixing.

Note

Yielding stops when one of the additionally provied signals ends; i.e. if one of the additional sources holds a signal of shorter length than that of the primary source that (shorter) signal forms the lower bound of the length of the mixed time signal yielded.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the mixed time samples in blocks of shape (num, num_channels), where num_channels is inhereted from the TimeOut base class. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.TimePower#

Bases: TimeOut

Calculate the time-dependent power of a signal by squaring its samples.

This class computes the power of the input signal by squaring the value of each sample. It processes the signal in blocks, making it suitable for large datasets or real-time signal processing. The power is calculated on a per-block basis, and each block of the output is yielded as a NumPy array.

Attributes:

sourceSamplesGenerator: The input data source, which provides the time signal or signal samples to be processed. It must be an instance of SamplesGenerator or any derived class that provides a result() method.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

result(num)#

Generate the time-dependent power of the input signal in blocks.

This method iterates through the signal samples provided by the source and calculates the power by squaring each sample. The output is yielded block-wise to facilitate processing large signals in chunks.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the squared samples from the source. Each block will have the shape (num, num_channels), where num_channels is inhereted from the TimeOut base class. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.TimeCumAverage#

Bases: TimeOut

Calculates the cumulative average of the signal.

This class computes the cumulative average of the input signal over time, which is useful for metrics like the Equivalent Continuous Sound Level (Leq). It processes the signal in blocks, maintaining a running average of the samples. The result is yielded in blocks, allowing for memory-efficient processing of large datasets.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

result(num)#

Generate the cumulative average of the input signal in blocks.

This method iterates through the signal samples provided by the source, and for each block, it computes the cumulative average of the samples up to that point. The result is yielded in blocks, with each block containing the cumulative average of the signal up to that sample.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the cumulative average of the samples. Each block will have the shape (num, num_channels), where num_channels is inhereted from the source. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Notes

The cumulative average is updated iteratively by considering the previously accumulated sum and the current block of samples. For each new sample, the cumulative average is recalculated by summing the previous cumulative value and the new samples, then dividing by the total number of samples up to that point.

class acoular.tprocess.TimeReverse#

Bases: TimeOut

Calculates the time-reversed signal of a source.

This class takes the input signal from a source and computes the time-reversed version of the signal. It processes the signal in blocks, yielding the time-reversed signal block by block. This can be useful for various signal processing tasks, such as creating echoes or reversing the playback of time signal signals.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

result(num)#

Generate the time-reversed version of the input signal block-wise.

This method processes the signal provided by the source in blocks, and for each block, it produces the time-reversed version of the signal. The result is yielded in blocks, with each block containing the time-reversed version of the signal for that segment. The signal is reversed in time by flipping the order of samples within each block.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the time-reversed version of the signal for the current block. Each block will have the shape (num, num_channels), where num_channels is inherited from the source. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Notes

The time-reversal is achieved by reversing the order of samples in each block of the signal. The result() method first collects all the blocks from the source, then processes them in reverse order, yielding the time-reversed signal in blocks. The first block yielded corresponds to the last block of the source signal, and so on, until the entire signal has been processed in reverse.

class acoular.tprocess.Filter#

Bases: TimeOut

Abstract base class for IIR filters using SciPy’s lfilter().

This class implements a digital Infinite Impulse Response (IIR) filter that applies filtering to a given signal in a block-wise manner. The filter coefficients can be dynamically changed during processing.

See also

scipy.signal.lfilter(): Filter data along one-dimension with an IIR or FIR (finite impulse response) filter.
scipy.signal.sosfilt(): Filter data along one dimension using cascaded second-order sections.
FiltOctave: Octave or third-octave bandpass filter (causal, with non-zero phase delay).
FiltFiltOctave: Octave or third-octave bandpass filter with zero-phase distortion.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

sos = Property()#: Second-order sections representation of the filter coefficients. This property is dynamically updated and can change during signal processing.

result(num)#

Apply the IIR filter to the input signal and yields filtered data block-wise.

This method processes the signal provided by source, applying the defined filter coefficients (sos) using the scipy.signal.sosfilt() function. The filtering is performed in a streaming fashion, yielding blocks of filtered signal data.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the bandpass-filtered signal for the current block. Each block has the shape (num, num_channels), where num_channels is inherited from the source. The last block may contain fewer samples if the total number of samples is not a multiple of num.

class acoular.tprocess.FiltOctave#

Bases: Filter

Octave or third-octave bandpass filter (causal, with non-zero phase delay).

This class implements a bandpass filter that conforms to octave or third-octave frequency band standards. The filter is designed using a second-order section (SOS) Infinite Impulse Response (IIR) approach.

The filtering process introduces a non-zero phase delay due to its causal nature. The center frequency and the octave fraction determine the frequency band characteristics.

See also

Filter: The base class implementing a general IIR filter.
FiltFiltOctave: Octave or third-octave bandpass filter with zero-phase distortion.

band = Float(1000.0, desc='band center frequency')#: The center frequency of the octave or third-octave band. Default is 1000.

fraction = Map({'Octave': 1, 'Third octave': 3}, default_value='Octave', desc='fraction of octave')#

Defines whether the filter is an octave-band or third-octave-band filter.

'Octave': Full octave band filter.
'Third octave': Third-octave band filter.

Default is 'Octave'.

order = Int(3, desc='IIR filter order')#: The order of the IIR filter, which affects the steepness of the filter’s roll-off. Default is 3.

sos = Property(depends_on=['band', 'fraction', 'source.digest', 'order'])#: Second-order sections representation of the filter coefficients. This property depends on band, fraction, order, and the source’s digest.

digest = Property(depends_on=['source.digest', 'band', 'fraction', 'order'])#: A unique identifier for the filter, based on its properties. (read-only)

class acoular.tprocess.FiltFiltOctave#

Bases: FiltOctave

Octave or third-octave bandpass filter with zero-phase distortion.

This filter applies an IIR bandpass filter in both forward and reverse directions, effectively eliminating phase distortion. It provides zero-phase filtering but requires significantly more memory compared to causal filtering.

See also

Filter: The base class implementing a general IIR filter.
FiltOctave: The standard octave or third-octave filter with causal filtering.

Notes

Due to the double-pass filtering, additional bandwidth correction is applied to maintain accurate frequency response.
This approach requires storing the entire signal in memory before processing, making it unsuitable for real-time applications with large datasets.

order = Int(2, desc='IIR filter half order')#: The half-order of the IIR filter, applied twice (once forward and once backward). This results in a final filter order twice as large as the specified value. Default is 2.

digest = Property(depends_on=['source.digest', 'band', 'fraction', 'order'])#: A unique identifier for the filter, based on its properties. (read-only)

result(num)#

Apply the filter to the input signal and yields filtered data block-wise.

The input signal is first stored in memory, then filtered in both forward and reverse directions to achieve zero-phase distortion. The processed signal is yielded in blocks.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array containing the filtered signal for the current block. Each block has shape (num, num_channels), where num_channels is inherited from the source. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Notes

This method requires the entire signal to be stored in memory, making it unsuitable for streaming or real-time applications.
Filtering is performed separately for each channel to optimize memory usage.

class acoular.tprocess.TimeExpAverage#

Bases: Filter

Compute an exponentially weighted moving average of the input signal.

This filter implements exponential averaging as defined in IEC 61672-1, which is commonly used for sound level measurements. The time weighting determines how quickly past values decay in significance.

See also

Filter: Base class for implementing IIR filters.

Notes

The Impulse ('I') weighting is not part of IEC 61672-1 but is included for additional flexibility.

weight = Map({'F': 0.125, 'S': 1.0, 'I': 0.035}, default_value='F', desc='time weighting')#

Time weighting constant, determining the exponential decay rate.

'F' (Fast) → 0.125
'S' (Slow) → 1.0
'I' (Impulse) → 0.035 (non-standard)

Default is 'F'.

sos = Property(depends_on=['weight', 'source.digest'])#: Filter coefficients in second-order section (SOS) format.

digest = Property(depends_on=['source.digest', 'weight'])#: A unique identifier for the filter, based on its properties. (read-only)

class acoular.tprocess.FiltFreqWeight#

Bases: Filter

Apply frequency weighting according to IEC 61672-1.

This filter implements frequency weighting curves commonly used in sound level meters for noise measurement. It provides A-weighting, C-weighting, and Z-weighting options.

See also

Filter: Base class for implementing IIR filters.

Notes

The filter is designed following IEC 61672-1:2002, the standard for sound level meters.
The weighting curves are implemented using bilinear transformation of analog filter coefficients to the discrete domain.

weight = Enum('A', 'C', 'Z', desc='frequency weighting')#

Defines the frequency weighting curve:

'A': Mimics human hearing sensitivity at low sound levels.
'C': Used for high-level sound measurements with less attenuation at low frequencies.
'Z': A flat response with no frequency weighting.

Default is 'A'.

sos = Property(depends_on=['weight', 'source.digest'])#: Second-order sections (SOS) representation of the filter coefficients. This property is dynamically computed based on weight and the Filter.source’s digest.

digest = Property(depends_on=['source.digest', 'weight'])#: A unique identifier for the filter, based on its properties. (read-only)

class acoular.tprocess.FilterBank#

Bases: TimeOut

Abstract base class for IIR filter banks based on scipy.signal.lfilter.

Implements a bank of parallel filters. This class should not be instantiated by itself.

Inherits from TimeOut, and defines the structure for working with filter banks for processing multi-channel time series data, such as time signal signals.

See also

TimeOut: ABC for signal processing blocks that interact with data from a source.
SamplesGenerator: Interface for any generating multi-channel time domain signal processing block.
scipy.signal: SciPy module for signal processing.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

sos = Property()#: The list containing second order section (SOS) coefficients for the filters in the filter bank.

bands = Property()#: A list of labels describing the different frequency bands of the filter bank.

num_bands = Property()#: The total number of bands in the filter bank.

num_channels = Property()#: The total number of output channels resulting from the filter bank operation.

result(num)#

Yield the bandpass filtered output of the source in blocks of samples.

This method uses the second order section coefficients (sos) to filter the input samples provided by the source in blocks. The result is returned as a generator.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: An array of shape (num, num_channels), delivering the filtered samples for each band. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Notes

The returned samples are bandpass filtered according to the coefficients in sos. Each block corresponds to the filtered samples for each frequency band.

class acoular.tprocess.OctaveFilterBank#

Bases: FilterBank

Octave or third-octave filter bank.

Inherits from FilterBank and implements an octave or third-octave filter bank. This class is used for filtering multi-channel time series data, such as time signal signals, using bandpass filters with center frequencies at octave or third-octave intervals.

See also

FilterBank: The base class for implementing IIR filter banks.
SamplesGenerator: Interface for generating multi-channel time domain signal processing blocks.
scipy.signal: SciPy module for signal processing.

lband = Int(21, desc='lowest band center frequency index')#: The lowest band center frequency index. Default is 21. This index refers to the position in the scale of octave or third-octave bands.

hband = Int(40, desc='highest band center frequency index + 1')#: The highest band center frequency index + 1. Default is 40. This is the position in the scale of octave or third-octave bands.

fraction = Map({'Octave': 1, 'Third octave': 3}, default_value='Octave', desc='fraction of octave')#: The fraction of an octave, either 'Octave' or 'Third octave'. Default is 'Octave'. Determines the width of the frequency bands. ‘Octave’ refers to full octaves, and 'Third octave' refers to third-octave bands.

ba = Property(depends_on=['lband', 'hband', 'fraction', 'source.digest'])#: The list of filter coefficients for all filters in the filter bank. The coefficients are computed based on the lband, hband, and fraction attributes.

bands = Property(depends_on=['lband', 'hband', 'fraction'])#: The list of labels describing the frequency bands in the filter bank.

num_bands = Property(depends_on=['lband', 'hband', 'fraction'])#: The total number of bands in the filter bank.

digest = Property(depends_on=['source.digest', 'lband', 'hband', 'fraction', 'order'])#: A unique identifier for the filter, based on its properties. (read-only)

class acoular.tprocess.WriteWAV#

Bases: TimeOut

Saves time signal from one or more channels as mono, stereo, or multi-channel .wav file.

Inherits from TimeOut and allows for exporting time-series data from one or more channels to a WAV file. Supports saving mono, stereo, or multi-channel signals to disk with automatic or user-defined file naming.

See also

TimeOut: ABC for signal processing blocks that interact with data from a source.
SamplesGenerator: Interface for generating multi-channel time domain signal processing blocks.
wave: Python module for handling WAV files.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

file = File(filter=['*.wav'], desc='name of wave file')#: The name of the file to be saved. If none is given, the name will be automatically generated from the source.

basename = Property(depends_on=['digest'])#: The name of the cache file (without extension). It serves as an internal reference for data caching and tracking processed files. (automatically generated)

channels = List(int, desc='channels to save')#: The list of channels to save. Can only contain one or two channels.

digest = Property(depends_on=['source.digest', 'channels'])#: A unique identifier for the filter, based on its properties. (read-only)

result(num)#

Generate and save time signal data as a WAV file in blocks.

This generator method retrieves time signal data from the source and writes it to a WAV file in blocks of size num. The data is scaled and encoded according to the selected bit depth and channel configuration. If no file name is specified, a name is generated automatically. The method yields each block of data after it is written to the file, allowing for streaming or real-time processing.

Parameters:

numint: Number of samples per block to write and yield.

Yields:

numpy.ndarray: The block of time signal data that was written to the WAV file, with shape (num, number of channels).

Raises:

ValueError: If no channels are specified for output.
Warning: If more than two channels are specified, or if the sample frequency is not an integer. Also warns if clipping occurs due to data range limitations.

See also

save(): Save the entire source output to a WAV file in one call.

save()#

Save the entire source output to a WAV file.

This method writes all available time signal data from the source to the specified WAV file in blocks. It calls the result() method internally and discards the yielded data. The file is written according to the current channels, encoding, and scaling settings. If no file name is specified, a name is generated automatically.

See also

result(): Generator for writing and yielding data block-wise.

class acoular.tprocess.WriteH5#

Bases: TimeOut

Saves time signal data as a .h5 (HDF5) file.

Inherits from TimeOut and provides functionality for saving multi-channel time-domain signal data to an HDF5 file. The file can be written in blocks and supports metadata storage, precision control, and dynamic file generation based on timestamps.

See also

TimeOut: ABC for signal processing blocks interacting with data from a source.
SamplesGenerator: Interface for generating multi-channel time-domain signal processing blocks.
h5py: Python library for reading and writing HDF5 files.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

file = File(filter=['*.h5'], desc='name of data file')#: The name of the file to be saved. If none is given, the name is automatically generated based on the current timestamp.

num_samples_write = Int(-1)#: The number of samples to write to file per call to result method. Default is -1, meaning all available data from the source will be written.

write_flag = Bool(True)#: A flag that can be set to stop file writing. Default is True.

digest = Property(depends_on=['source.digest'])#: A unique identifier for the object, based on its properties. (read-only)

precision = Enum('float32', 'float64', desc='precision of H5 File')#: Precision of the entries in the HDF5 file, represented as numpy data types. Default is 'float32'.

metadata = Dict(desc='metadata to be stored in .h5 file')#: Metadata to be stored in the HDF5 file.

create_filename()#

Generate a filename for the HDF5 file if needed.

Generate a filename for the HDF5 file based on the current timestamp if no filename is provided. If a filename is provided, it is used as the file name.

get_initialized_file()#

Initialize the HDF5 file and prepare the necessary datasets and metadata.

This method creates the file (if it doesn’t exist), sets up the main data array, and appends metadata to the file.

Returns:

h5py.File: The initialized HDF5 file object ready for data insertion.

save()#

Save the source output to a HDF5 file.

This method writes the processed time-domain signal data from the source to the specified HDF5 file. Data is written in blocks and appended to the extendable 'time_data' array.

Notes

If no file is specified, a file name is automatically generated.
Metadata defined in the metadata attribute is stored in the file.

add_metadata(f5h)#

Add metadata to the HDF5 file.

Metadata is stored in a separate ‘metadata’ group within the HDF5 file. The metadata is stored as arrays with each key-value pair corresponding to a separate array.

Parameters:

f5hh5py.File: The HDF5 file object to which metadata will be added.

result(num)#

Python generator that saves source output to an HDF5 file.

This method processes data from the source in blocks and writes the data to the HDF5 file. It yields the processed blocks while the data is being written.

Parameters:

numint: Number of samples per block.

Yields:

numpy.ndarray: A numpy array of shape (num, num_channels), where num_channels is inhereted from the source, delivering the processed time-domain signal data. The last block may contain fewer samples if the total number of samples is not a multiple of num.

Notes

If num_samples_write is set to a value other than -1, only that number of samples will be written to the file.
The data is echoed as it is yielded, after being written to the file.

class acoular.tprocess.TimeConvolve#

Bases: TimeOut

Perform frequency domain convolution with the uniformly partitioned overlap-save (UPOLS) method.

This class convolves a source signal with a kernel in the frequency domain. It uses the UPOLS method, which efficiently computes convolutions by processing signal blocks and kernel blocks separately in the frequency domain. For detailed theoretical background, refer to [24].

Inherits from TimeOut, which allows the class to process signals generated by a source object. The kernel used for convolution can be one-dimensional or two-dimensional, and it can be applied across one or more channels of the source signal.

See also

TimeOut: The parent class for signal processing blocks.
SamplesGenerator: The interface for generating multi-channel time-domain signals.

source = Instance(SamplesGenerator)#: The input data source. It must be an instance of a SamplesGenerator-derived class.

kernel = CArray(dtype=float, desc='Convolution kernel.')#: Convolution kernel in the time domain. The second dimension of the kernel array has to be either 1 or match the source’s num_channels attribute. If only a single kernel is supplied, it is applied to all channels.

digest = Property(depends_on=['source.digest', 'kernel'])#: A unique identifier for the object, based on its properties. (read-only)

result(num=128)#

Convolve the source signal with the kernel and yield the result in blocks.

The method generates the convolution of the source signal with the kernel by processing the signal in small blocks, performing the convolution in the frequency domain, and yielding the results block by block.

Parameters:

numint, optional: Number of samples per block. Default is 128.

Yields:

numpy.ndarray: A array of shape (num, num_channels), where num_channels is inhereted from the source, representing the convolution result in blocks.

Notes

The kernel is first validated and reshaped if necessary.
The convolution is computed efficiently using the FFT in the frequency domain.

tprocess#

`tprocess`#