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