EmptyStreamGenerator¶
- class baseband_tasks.generators.EmptyStreamGenerator(shape, start_time, sample_rate, *, samples_per_frame=1, dtype=<class 'numpy.complex64'>, **kwargs)[source] [edit on github]¶
Bases:
Base
Generator of an empty data stream.
The stream is meant to be filled with a
Task
.- Parameters:
- shapetuple
First element is the total number of samples of the fake file, the others are the sample shape.
- start_time
Time
Start time of the fake file.
- sample_rate
Quantity
Sample rate, in units of frequency.
- samples_per_frameint
Blocking factor. This is mostly useful to make the function task that uses the stream more efficient.
- dtype
dtype
or anything that initializes one, optional Type of data produced. Default:
complex64
.- — **kwargsmeta data for the stream, which usually include
- frequency
Quantity
, optional Frequencies for each channel. Should be broadcastable to the sample shape. Default: unknown.
- sidebandarray, optional
Whether frequencies are upper (+1) or lower (-1) sideband. Should be broadcastable to the sample shape. Default: unknown.
- polarizationarray or (nested) list of char, optional
Polarization labels. Should broadcast to the sample shape, i.e., the labels are in the correct axis. For instance,
['X', 'Y']
, or[['L'], ['R']]
. Default: unknown.
Examples
Produce alternating +/-1 in single-channel data with decent-sized blocks.
>>> from baseband_tasks.generators import EmptyStreamGenerator >>> from baseband_tasks.base import Task >>> import numpy as np >>> from astropy import time as t, units as u >>> def alternate(data): ... value = 2 * (np.arange(data.shape[0]) % 2) - 1 ... data[...] = value ... return data ... >>> eh = EmptyStreamGenerator((1000,), t.Time('2010-11-12'), ... 1.*u.kHz, samples_per_frame=100, ... dtype='f4') >>> sh = Task(eh, alternate) >>> sh.seek(995) 995 >>> sh.read() array([ 1., -1., 1., -1., 1.], dtype=float32)