Acoular 25.01 documentation

SamplesBuffer

«  TimeCache   ::   process   ::   sdinput  »

SamplesBuffer

class acoular.process.SamplesBuffer

Bases: InOut

Handles buffering of samples from a source.

This class is used to buffer samples from a source and provide them in blocks of a specified size. There are several usecases for this class, as demonstrated in the following.

Examples

Let us assume we want to draw blocks of 16 samples from our source, but we want to make sure that we always have twice the number of samples in the buffer. We can achieve this simple behaviour by using the following code:

>>> import acoular as ac
>>> import numpy as np
>>> # create a white noise source with 512 samples
>>> source = ac.TimeSamples(
...     data=ac.WNoiseGenerator(
...         sample_freq=64,
...         num_samples=512,
...     ).signal()[:, np.newaxis],
...     sample_freq=64,
... )
>>> # create a buffer with a size of 32 samples
>>> buffer = ac.process.SamplesBuffer(source=source, length=32)
>>> # get the first block of 16 samples
>>> block = next(buffer.result(num=16))
>>> np.testing.assert_array_equal(block, source.data[:16])

Here, on the first call to the result method, the buffer will fill up by collecting blocks with same size from the source. The buffer will then return the first block of 16 samples. On the next call to the result method, the buffer will be filled again and returns the next block of 16 samples.

In some cases, we might want to draw a different number of samples from the source than we want to return. This can be achieved by setting the source_num trait of the buffer. A special case is the return of a variable number of samples. This is the case, for example, in the class BeamformerTimeTraj, in which a different number of time samples is required from the buffer for further delay-and-sum processing depending on the expected delay, which can be vary for moving sources. At the same time, however, only ‘num’ samples should be written to and removed from the buffer. This behavior can be achieved by setting the shift_index_by trait to ‘num’ and by setting the result_num trait to the number of samples that should be returned by the result function.

>>> buffer = ac.process.SamplesBuffer(source=source, length=32, result_num=20, shift_index_by='num')
>>> block_sizes = []
>>> block_sizes.append(
...     next(buffer.result(num=16)).shape[0]
... )  # this time, the buffer will return 20 samples, but the buffer will only forget the first 16 samples
>>> buffer.result_num = 24
>>> block_sizes.append(
...     next(buffer.result(num=16)).shape[0]
... )  # this time, the buffer will return 24 samples, but the buffer will only forget the first 16 samples
>>> np.testing.assert_array_equal(block_sizes, [20, 24])
length = Int(desc='number of samples that fit in the buffer')

number of samples that fit in the buffer

source_num = Union(

number of samples per block to obtain from the source. If ‘None’, use ‘num’ argument of result method

result_num = Union(

number of samples to return from the buffer. If ‘None’, use ‘num’ argument of result method

shift_index_by = Enum(

index shift value for the buffer. If “result_num”, buffer will return and forget ‘result_num’ samples. If “num”, buffer will return ‘result_num’ samples but will forget ‘num’ samples

level = Property(desc='current filling level of buffer')

current filling level of buffer

dtype = Any(desc='data type of the buffer')

data type of the buffer elements

increase_buffer(num)

Increase the buffer by ‘num’ samples.

Returns:
None
read_from_buffer(num)

Read samples from the buffer.

Parameters:
numint

number of samples to read from the buffer.

Returns:
numpy.ndarray

block of samples from the buffer

fill_buffer(snum)

Fill the buffer with samples from the source.

Parameters:
snumint

number of samples to return from the source.

Yields:
None
result(num)

Return blocks of samples from the buffer.

Parameters:
numint

number of samples to return.

Yields:
numpy.ndarray

block of samples from the buffer

«  TimeCache   ::   process   ::   sdinput  »