Skip to content

sgnts.sources.fake_series

FakeSeriesSource dataclass

Bases: TSSource

A time-series source that generates fake data in fixed-size buffers.

If t0 is not specified the current GPS time will be used as the start time.

Parameters:

Name Type Description Default
rate int

int, the sample rate of the data

 2048
sample_shape tuple[int, ...]

tuple[int, ...], the shape of a sample of data, or the shape of the data in each dimension except the last (time) dimension, i.e., sample_shape = data.shape[:-1]. For example, if the data is a multi-dimensional array and has shape=(2, 4, 16) then sample_shape = (2, 4). Note that if data is one dimensional and has shape (16,), sample_shape would be an empty tuple ().

 ()
signal_type str

str, currently supported types: (1) 'white': white noise data. (2) 'sin' or 'sine': sine wave data. (3) 'impulse': creates an impulse data, where the value is one at one sample point, and everywhere else is zero

 'white'
fsin float

float, the frequency of the sine wave if signal_type = 'sin'

 5
ngap int

int, the frequency to generate gap buffers, will generate a gap buffer every ngap buffers. ngap=0: do not generate gap buffers. ngap=-1: generates gap buffers randomly.

 0
random_seed Optional[int]

int, set the random seed, used for signal_type = 'white' or 'impulse'

 None
impulse_position int

int, the sample point position to place the impulse. If -1, then the impulse will be generated randomly.

 -1
real_time bool

bool, run the source in "real time", such that frames are produced at the rate corresponding to their relative offsets. In real-time mode, t0 will default to the current GPS time if not otherwise specified.

 False
Source code in sgnts/sources/fake_series.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@dataclass
class FakeSeriesSource(TSSource):
    """A time-series source that generates fake data in fixed-size buffers.

    If `t0` is not specified the current GPS time will be used as the
    start time.

    Args:
        rate:
            int, the sample rate of the data
        sample_shape:
            tuple[int, ...], the shape of a sample of data, or the
            shape of the data in each dimension except the last (time)
            dimension, i.e., sample_shape = data.shape[:-1]. For
            example, if the data is a multi-dimensional array and has
            shape=(2, 4, 16) then sample_shape = (2, 4).  Note that if
            data is one dimensional and has shape (16,), sample_shape
            would be an empty tuple ().
        signal_type:
            str, currently supported types: (1) 'white': white noise data. (2) 'sin' or
            'sine': sine wave data. (3) 'impulse': creates an impulse data, where the
            value is one at one sample point, and everywhere else is zero
        fsin:
            float, the frequency of the sine wave if signal_type = 'sin'
        ngap:
            int, the frequency to generate gap buffers, will generate a gap buffer every
            ngap buffers. ngap=0: do not generate gap buffers. ngap=-1: generates gap
            buffers randomly.
        random_seed:
            int, set the random seed, used for signal_type = 'white' or 'impulse'
        impulse_position:
            int, the sample point position to place the impulse. If -1, then the
            impulse will be generated randomly.
        real_time:
            bool, run the source in "real time", such that frames are
            produced at the rate corresponding to their relative
            offsets.  In real-time mode, t0 will default to the
            current GPS time if not otherwise specified.

    """

    rate: int = 2048
    sample_shape: tuple[int, ...] = ()
    signal_type: str = "white"
    fsin: float = 5
    ngap: int = 0
    random_seed: Optional[int] = None
    impulse_position: int = -1
    real_time: bool = False
    verbose: bool = False

    def __post_init__(self):
        # set the start and end times if not specified
        if self.real_time and self.t0 is None:
            # if t0 not specified set t0 to be the next GPS second
            self.t0 = int(gpsnow()) + 1

        super().__post_init__()

        self.cnt = {p: 0 for p in self.source_pads}

        # setup buffers
        for pad in self.source_pads:
            self.set_pad_buffer_params(
                pad=pad, sample_shape=self.sample_shape, rate=self.rate
            )

        if self.random_seed is not None and (
            self.signal_type == "white" or self.signal_type == "impulse"
        ):
            np.random.seed(self.random_seed)
        if self.signal_type == "impulse":
            assert self.sample_shape == ()
            assert len(self.source_pads) == 1
            if self.impulse_position == -1:
                self.impulse_position = np.random.randint(0, int(self.end * self.rate))
            if self.verbose:
                print("Placing impulse at sample point", self.impulse_position)

        # for real time tracking record to t0 time relative to current
        # real time
        self._start_offset_from_realtime = gpsnow() - self.t0

    def create_impulse_data(self, offset: int, num_samples: int, rate: int) -> Array:
        """Create the impulse data, where data is zero everywhere, and equals one at one
        sample point.

        Args:
            offset:
                int, the offset of the current buffer, used for checking whether the
                the impulse is to be placed in the current buffer
            num_samples:
                int, the number of samples the data should have
            rate:
                int, the sample rate of the data

        Returns:
            Array, the impulse data
        """
        data = np.zeros(num_samples)
        current_samples = Offset.tosamples(offset, rate)
        if (
            current_samples <= self.impulse_position
            and self.impulse_position < current_samples + num_samples
        ):
            if self.verbose:
                print("Creating the impulse")
            data[self.impulse_position - current_samples] = 1
        return data

    def create_data(self, buf: SeriesBuffer, cnt: int) -> Array:
        """Create the fake data, can be (1) white noise (2) sine wave (3) impulse data.

        Args:
            buf:
                SeriesBuffer, the buffer to create the data for
            cnt:
                int, the number of buffers the source pad has generated, used for
                determining whether to generate gap buffers

        Returns:
            Array, the fake data array
        """
        offset = buf.offset
        ngap = self.ngap
        if (ngap == -1 and np.random.rand(1) > 0.5) or (ngap > 0 and cnt % ngap == 0):
            return None
        elif self.signal_type == "white":
            return np.random.randn(*buf.shape)
        elif self.signal_type == "sin" or self.signal_type == "sine":
            t0 = Offset.tosec(offset)
            duration = buf.duration
            return np.sin(
                self.fsin
                * np.tile(
                    np.linspace(t0, t0 + duration, buf.samples, endpoint=False),
                    self.sample_shape + (1,),
                )
            )
        elif self.signal_type == "impulse":
            return self.create_impulse_data(offset, buf.samples, buf.sample_rate)
        else:
            raise ValueError("Unknown signal type")

    def internal(self):
        if self.real_time:
            # get the next time from the new offset.  all pads
            # *should* have the same offset, so just take the offset
            # from the first pad.
            next_time = Offset.tosec(list(self.offset.values())[0])
            sleep = next_time - gpsnow() + self._start_offset_from_realtime
            if sleep < 0:
                LOGGER.warning("Warning: FakeSeriesSource falling behind real time")
            else:
                time.sleep(sleep)

    def new(self, pad: SourcePad) -> TSFrame:
        """New buffers are created on "pad" with an instance specific count and a name
        derived from the pad name. "EOS" is set if we have surpassed the requested
        end time.

        Args:
            Pad:
                SourcePad, the source pad to generate TSFrames

        Returns:
            TSFrame, the TSFrame that carries the buffers with fake data
        """
        self.cnt[pad] += 1

        # setup metadata
        metadata = {"cnt": self.cnt, "name": "'%s'" % pad.name}
        if self.impulse_position is not None:
            metadata["impulse_offset"] = Offset.fromsamples(
                self.impulse_position, self.rate
            )

        frame = self.prepare_frame(pad, data=None, metadata=metadata)
        for buf in frame:
            buf.set_data(self.create_data(buf, self.cnt[pad]))

        return frame

create_data(buf, cnt)

Create the fake data, can be (1) white noise (2) sine wave (3) impulse data.

Parameters:

Name Type Description Default
buf SeriesBuffer

SeriesBuffer, the buffer to create the data for

 required
cnt int

int, the number of buffers the source pad has generated, used for determining whether to generate gap buffers

 required

Returns:

Type Description
Array

Array, the fake data array
Source code in sgnts/sources/fake_series.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def create_data(self, buf: SeriesBuffer, cnt: int) -> Array:
    """Create the fake data, can be (1) white noise (2) sine wave (3) impulse data.

    Args:
        buf:
            SeriesBuffer, the buffer to create the data for
        cnt:
            int, the number of buffers the source pad has generated, used for
            determining whether to generate gap buffers

    Returns:
        Array, the fake data array
    """
    offset = buf.offset
    ngap = self.ngap
    if (ngap == -1 and np.random.rand(1) > 0.5) or (ngap > 0 and cnt % ngap == 0):
        return None
    elif self.signal_type == "white":
        return np.random.randn(*buf.shape)
    elif self.signal_type == "sin" or self.signal_type == "sine":
        t0 = Offset.tosec(offset)
        duration = buf.duration
        return np.sin(
            self.fsin
            * np.tile(
                np.linspace(t0, t0 + duration, buf.samples, endpoint=False),
                self.sample_shape + (1,),
            )
        )
    elif self.signal_type == "impulse":
        return self.create_impulse_data(offset, buf.samples, buf.sample_rate)
    else:
        raise ValueError("Unknown signal type")

create_impulse_data(offset, num_samples, rate)

Create the impulse data, where data is zero everywhere, and equals one at one sample point.

Parameters:

Name Type Description Default
offset int

int, the offset of the current buffer, used for checking whether the the impulse is to be placed in the current buffer

 required
num_samples int

int, the number of samples the data should have

 required
rate int

int, the sample rate of the data

 required

Returns:

Type Description
Array

Array, the impulse data
Source code in sgnts/sources/fake_series.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create_impulse_data(self, offset: int, num_samples: int, rate: int) -> Array:
    """Create the impulse data, where data is zero everywhere, and equals one at one
    sample point.

    Args:
        offset:
            int, the offset of the current buffer, used for checking whether the
            the impulse is to be placed in the current buffer
        num_samples:
            int, the number of samples the data should have
        rate:
            int, the sample rate of the data

    Returns:
        Array, the impulse data
    """
    data = np.zeros(num_samples)
    current_samples = Offset.tosamples(offset, rate)
    if (
        current_samples <= self.impulse_position
        and self.impulse_position < current_samples + num_samples
    ):
        if self.verbose:
            print("Creating the impulse")
        data[self.impulse_position - current_samples] = 1
    return data

new(pad)

New buffers are created on "pad" with an instance specific count and a name derived from the pad name. "EOS" is set if we have surpassed the requested end time.

Parameters:

Name Type Description Default
Pad

SourcePad, the source pad to generate TSFrames

 required

Returns:

Type Description
TSFrame

TSFrame, the TSFrame that carries the buffers with fake data
Source code in sgnts/sources/fake_series.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def new(self, pad: SourcePad) -> TSFrame:
    """New buffers are created on "pad" with an instance specific count and a name
    derived from the pad name. "EOS" is set if we have surpassed the requested
    end time.

    Args:
        Pad:
            SourcePad, the source pad to generate TSFrames

    Returns:
        TSFrame, the TSFrame that carries the buffers with fake data
    """
    self.cnt[pad] += 1

    # setup metadata
    metadata = {"cnt": self.cnt, "name": "'%s'" % pad.name}
    if self.impulse_position is not None:
        metadata["impulse_offset"] = Offset.fromsamples(
            self.impulse_position, self.rate
        )

    frame = self.prepare_frame(pad, data=None, metadata=metadata)
    for buf in frame:
        buf.set_data(self.create_data(buf, self.cnt[pad]))

    return frame