StreamGenerator

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

Bases: baseband_tasks.base.Base

Generator of data produced by a user-provided function.

The function needs to be aware of stream structure. As an alternative, use EmptyStreamGenerator to generate an empty stream and add a Task that fills data arrays.

Parameters

functioncallable

Function that takes one argument, the Source instance, and returns data with the correct shape, i.e., samples_per_frame samples of sample shape shape[1:]. The function can count on the instance being at the start of the frame (i.e., instance.tell() is correct).

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 can be used for efficiency to reduce the overhead of calling the source function.

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 ones and zeros.

>>> from baseband_tasks.generators import StreamGenerator
>>> import numpy as np
>>> from astropy.time import Time
>>> from astropy import units as u
>>> def alternate(sh):
...     return np.full((1,) + sh.shape[1:], sh.tell() % 2 == 1, sh.dtype)
...
>>> sh = StreamGenerator(alternate, (10, 6), Time('2010-11-12'), 10.*u.Hz)
>>> sh.seek(5)
5
>>> sh.read()  
array([[1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j],
       [0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j],
       [1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j],
       [0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j, 0.+0.j],
       [1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j, 1.+0.j]], dtype=complex64)