EmptyStreamGenerator

class baseband_tasks.generators.EmptyStreamGenerator(shape, start_time, sample_rate, *, samples_per_frame=1, dtype=<class 'numpy.complex64'>, **kwargs)

Bases: baseband_tasks.base.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_timeTime

Start time of the fake file.

sample_rateQuantity

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.

dtypedtype or anything that initializes one, optional

Type of data produced. Default: complex64.

— **kwargsmeta data for the stream, which usually include

frequencyQuantity, 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)