Skip to content

sgnts.base.offset

Offset

A class for bookkeeping of sample points in the SGN-TS package.

MAX_RATE

the maximum sample rate the pipeline will use. Should be a power of 2.

ALLOWED_RATES

will vary from 1 to MAX_RATE by powers of 2.

SAMPLE_STRIDE_AT_MAX_RATE

is the average stride that src pads should acheive per Frame in order to ensure that the pipeline src elements are roughly synchronous. Otherwise queues blow up and the pipelines get behind until they crash.

offset_ref_t0

reference time to count offsets, in nanoseconds

offset

Offsets count the number of samples at the MAX_RATE since the reference time offset_ref_t0, and are used for bookkeeping. Since offsets exactly track samples at the MAX_RATE, any data in ALLOWED_RATES will have sample points that lie exactly on an offset point. We then use offsets to synchronize between data at different sample rates, and to convert number of samples between different sample rate ratios. An offset can also be viewed as a time unit that equals 1/MAX_RATE seconds.

Example:

Suppose a pipeline has data of sample rates 16 Hz, 8 Hz, 4 Hz. If we set MAX_RATE = 16, offsets will track the sample points at 16 Hz. The other two data streams will also have sample points lying exactly on offset points, but with a fixed gap step.

offsets | | | | | | | | | | | | | | | | 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

sample rate 16 x x x x x x x x x x x x x x x x sample rate 8 x x x x x x x x sample rate 4 x x x x

Assumptions

all the sample rates in the buffers are powers of 2.

Using offsets as a clock that is a power of 2 gives better resolution between sample points than using seconds/nanoseconds, because it exactly tracks samples.

Example: If the MAX_RATE = 16384, for a buffer of data at a sample rate of 2048, the time difference between two nearby samples can be represented as integer offsets 16384/2048 = 8 offsets. However, if we want to use integer nanoseconds, the time difference will be 1/2048 * 1e9 = 488281.25 nanoseconds, which cannot be represented by an integer.

The MAX_RATE can be changed to a number that is a power of 2 and larger than the highest sample rate of the pipeline. As long as all sample points lie exactly on offset points, the bookkeeping will still work.

Source code in sgnts/base/offset.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 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
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
class Offset:
    """A class for bookkeeping of sample points in the SGN-TS package.

    MAX_RATE:
      the maximum sample rate the pipeline will use. Should be a power of 2.

    ALLOWED_RATES:
      will vary from 1 to MAX_RATE by powers of 2.

    SAMPLE_STRIDE_AT_MAX_RATE:
      is the average stride that src pads should acheive per Frame in order to ensure
      that the pipeline src elements are roughly synchronous. Otherwise queues blow up
      and the pipelines get behind until they crash.

    offset_ref_t0:
      reference time to count offsets, in nanoseconds

    offset:
      Offsets count the number of samples at the MAX_RATE since the reference time
      offset_ref_t0, and are used for bookkeeping. Since offsets exactly track samples
      at the MAX_RATE, any data in ALLOWED_RATES will have sample points that lie
      exactly on an offset point. We then use offsets to synchronize between data at
      different sample rates, and to convert number of samples between different sample
      rate ratios. An offset can also be viewed as a time unit that equals 1/MAX_RATE
      seconds.

      Example:
      --------
      Suppose a pipeline has data of sample rates 16 Hz, 8 Hz, 4 Hz. If we set
      MAX_RATE = 16, offsets will track the sample points at 16 Hz. The other two data
      streams will also have sample points lying exactly on offset points, but with a
      fixed gap step.


      offsets          |   |   |   |   |   |   |   |   |   |   |   |   |   |   |   |
                       0   1   2   3   4   5   6   7   8   9   10  11  12  13  14  15

      sample rate 16   x   x   x   x   x   x   x   x   x   x   x   x   x   x   x   x
      sample rate 8    x       x       x       x       x       x       x       x
      sample rate 4    x               x               x               x

    Assumptions:
      all the sample rates in the buffers are powers of 2.

    Using offsets as a clock that is a power of 2 gives better resolution between
    sample points than using seconds/nanoseconds, because it exactly tracks samples.

    Example: If the MAX_RATE = 16384, for a buffer of data at a sample rate of 2048,
    the time difference between two nearby samples can be represented as integer offsets
    16384/2048 = 8 offsets. However, if we want to use integer nanoseconds, the time
    difference will be 1/2048 * 1e9 = 488281.25 nanoseconds, which cannot be represented
    by an integer.

    The MAX_RATE can be changed to a number that is a power of 2 and larger than the
    highest sample rate of the pipeline. As long as all sample points lie exactly on
    offset points, the bookkeeping will still work.
    """

    offset_ref_t0 = 0
    MAX_RATE = 16384
    ALLOWED_RATES = set(2**x for x in range(1 + int(numpy.log2(MAX_RATE))))
    SAMPLE_STRIDE_AT_MAX_RATE = 16384

    @classmethod
    def set_max_rate(cls, max_rate):
        cls.MAX_RATE = max_rate
        cls.ALLOWED_RATES = set(2**x for x in range(1 + int(numpy.log2(cls.MAX_RATE))))

    @classmethod
    def convert(
        cls,
        value: Union[int, float],
        from_unit: TimeUnits,
        to_unit: TimeUnits,
        from_sample_rate: Optional[int] = None,
        to_sample_rate: Optional[int] = None,
    ) -> Union[int, float]:
        """Convert a value from one time unit to another.

        Args:
            value: The value to convert.
            from_unit: The unit of the input value.
            to_unit: The unit of the output value.
            from_sample_rate: Required if from_unit is SAMPLES. Optional if
                              from_unit is NANOSECONDS (used for alignment).
            to_sample_rate: Required if to_unit is SAMPLES.

        Returns:
            The converted value in to_unit.
        """
        # 1. Optimize identity conversion
        if from_unit == to_unit:
            # Handle SAMPLES -> SAMPLES resampling
            if from_unit == TimeUnits.SAMPLES:
                if from_sample_rate == to_sample_rate:
                    return value
                # If rates differ, fall through to full conversion logic
            else:
                return value

        # 2. Convert Source -> OFFSETS (Base Unit)
        if from_unit == TimeUnits.SECONDS:
            offset_val = cls.fromsec(value)
        else:
            assert isinstance(value, int), (
                "Value must be an integer when " "converting from non-seconds units"
            )
            if from_unit == TimeUnits.OFFSETS:
                offset_val = value
            elif from_unit == TimeUnits.NANOSECONDS:
                # fromns accepts a sample_rate for alignment purposes
                offset_val = cls.fromns(value, sample_rate=from_sample_rate)
            elif from_unit == TimeUnits.SAMPLES:
                if from_sample_rate is None:
                    raise ValueError(
                        "from_sample_rate required when converting from SAMPLES"
                    )
                offset_val = cls.fromsamples(value, from_sample_rate)
            else:
                raise ValueError(f"Unknown from_unit: {from_unit}")

        # 3. Convert OFFSETS -> Target
        if to_unit == TimeUnits.OFFSETS:
            return offset_val
        elif to_unit == TimeUnits.SECONDS:
            return cls.tosec(offset_val)
        elif to_unit == TimeUnits.NANOSECONDS:
            return cls.tons(offset_val)
        elif to_unit == TimeUnits.SAMPLES:
            if to_sample_rate is None:
                raise ValueError("to_sample_rate required when converting to SAMPLES")
            return cls.tosamples(offset_val, to_sample_rate)
        else:
            raise ValueError(f"Unknown to_unit: {to_unit}")

    @staticmethod
    def sample_stride(rate: int) -> int:
        """Given Offset.SAMPLE_STRIDE_AT_MAX_RATE, derive the sample stride at the
        requested sample rate.

        Args:
            rate:
                int, the sample rate to calculate the sample stride

        Returns:
           int, the number of samples in the stride at the requested sample rate
        """
        return Offset.tosamples(Offset.SAMPLE_STRIDE_AT_MAX_RATE, rate)

    @staticmethod
    def tosec(offset: int) -> float:
        """Convert offsets to seconds.

        Args:
            offset:
                int, the offset to convert to seconds

        Returns:
            float, the time corresponding to the offset, in seconds
        """
        return offset / Offset.MAX_RATE

    @staticmethod
    def tons(offset: int) -> int:
        """Convert offsets to integer nanoseconds.

        Args:
            offset:
                int, the offset to convert to nanoseconds

        Returns:
            int, the time corresponding to the offset, in nanoseconds

        NOTE - for very large offsets, this switches to integer arithmetic to
        preserve precision. A downside is that the result will be truncated
        rather than rounded, leading to a slight bias at the 1 ns scale in
        some case. This is unlikely to cause any actual problems.  As a
        reminder, all serious bookkeeping should be done with offsets not
        timestamps.
        """
        if offset * Offset.MAX_RATE > 1e17:
            return offset * Time.SECONDS // Offset.MAX_RATE
        else:
            return round(offset / Offset.MAX_RATE * Time.SECONDS)

    @staticmethod
    def fromsec(seconds: float) -> int:
        """Convert seconds to offsets.

        Args:
            seconds:
                float, the time to convert to offsets, in seconds

        Returns:
            int, the offset corresponding to the time
        """
        return round(seconds * Offset.MAX_RATE)

    @staticmethod
    def fromns(nanoseconds: int, sample_rate: Optional[int] = None) -> int:
        """Convert nanoseconds to offsets.

        Args:
            nanoseconds:
                int, the time to convert to offsets, in nanoseconds
            sample_rate:
                int, optional sample rate to align the offset to. If provided,
                the offset will be rounded to the nearest sample boundary for
                this rate.

        Returns:
            int, the offset corresponding to the time
        """
        if sample_rate is None:
            # Standard behavior - convert directly
            return round(int(nanoseconds) / int(Time.SECONDS) * Offset.MAX_RATE)
        else:
            # Align to sample boundary
            assert (
                sample_rate in Offset.ALLOWED_RATES
            ), f"Invalid sample rate: {sample_rate}"

            # Convert nanoseconds to samples at the given rate
            samples = round(nanoseconds * sample_rate / Time.SECONDS)

            # Convert samples back to offset - this ensures alignment
            return Offset.fromsamples(int(samples), sample_rate)

    @staticmethod
    def tosamples(offset: int, sample_rate: int) -> int:
        """Convert offsets to number of sample points.

        Args:
            offset:
                int, the offset to convert to number of samples. The offset must map to
                integer number of sample points.
            sample_rate:
                int, the sample rate at which to calculate the number of samples

        Returns:
            int, the number of samples corresponding to the offset at the given sample
            rate
        """
        assert (
            sample_rate in Offset.ALLOWED_RATES
        ), f"Sample rate {sample_rate} not in ALLOWED_RATES: {Offset.ALLOWED_RATES}"
        assert not offset % (Offset.MAX_RATE // sample_rate), (
            "Offset does not map to"
            f" integer sample points. Offset: {offset}, sample rate: {sample_rate}"
        )
        return offset // (Offset.MAX_RATE // sample_rate)

    @staticmethod
    def fromsamples(samples: int, sample_rate: int) -> int:
        """Convert number of sample points to offsets.

        Args:
            samples:
                int, the number of samples to convert to offsets
            sample_rate:
                int, the sample rate at which to calculate the offset

        Returns:
            int, the offset corresponding to the number of sample points at the given
            sample rate
        """
        assert (
            sample_rate in Offset.ALLOWED_RATES
        ), f"Sample rate {sample_rate} not in ALLOWED_RATES: {Offset.ALLOWED_RATES}"
        return samples * Offset.MAX_RATE // sample_rate

convert(value, from_unit, to_unit, from_sample_rate=None, to_sample_rate=None) classmethod

Convert a value from one time unit to another.

Parameters:

Name Type Description Default
value Union[int, float]

The value to convert.

 required
from_unit TimeUnits

The unit of the input value.

 required
to_unit TimeUnits

The unit of the output value.

 required
from_sample_rate Optional[int]

Required if from_unit is SAMPLES. Optional if from_unit is NANOSECONDS (used for alignment).

 None
to_sample_rate Optional[int]

Required if to_unit is SAMPLES.

 None

Returns:

Type Description
Union[int, float]

The converted value in to_unit.
Source code in sgnts/base/offset.py 
 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
@classmethod
def convert(
    cls,
    value: Union[int, float],
    from_unit: TimeUnits,
    to_unit: TimeUnits,
    from_sample_rate: Optional[int] = None,
    to_sample_rate: Optional[int] = None,
) -> Union[int, float]:
    """Convert a value from one time unit to another.

    Args:
        value: The value to convert.
        from_unit: The unit of the input value.
        to_unit: The unit of the output value.
        from_sample_rate: Required if from_unit is SAMPLES. Optional if
                          from_unit is NANOSECONDS (used for alignment).
        to_sample_rate: Required if to_unit is SAMPLES.

    Returns:
        The converted value in to_unit.
    """
    # 1. Optimize identity conversion
    if from_unit == to_unit:
        # Handle SAMPLES -> SAMPLES resampling
        if from_unit == TimeUnits.SAMPLES:
            if from_sample_rate == to_sample_rate:
                return value
            # If rates differ, fall through to full conversion logic
        else:
            return value

    # 2. Convert Source -> OFFSETS (Base Unit)
    if from_unit == TimeUnits.SECONDS:
        offset_val = cls.fromsec(value)
    else:
        assert isinstance(value, int), (
            "Value must be an integer when " "converting from non-seconds units"
        )
        if from_unit == TimeUnits.OFFSETS:
            offset_val = value
        elif from_unit == TimeUnits.NANOSECONDS:
            # fromns accepts a sample_rate for alignment purposes
            offset_val = cls.fromns(value, sample_rate=from_sample_rate)
        elif from_unit == TimeUnits.SAMPLES:
            if from_sample_rate is None:
                raise ValueError(
                    "from_sample_rate required when converting from SAMPLES"
                )
            offset_val = cls.fromsamples(value, from_sample_rate)
        else:
            raise ValueError(f"Unknown from_unit: {from_unit}")

    # 3. Convert OFFSETS -> Target
    if to_unit == TimeUnits.OFFSETS:
        return offset_val
    elif to_unit == TimeUnits.SECONDS:
        return cls.tosec(offset_val)
    elif to_unit == TimeUnits.NANOSECONDS:
        return cls.tons(offset_val)
    elif to_unit == TimeUnits.SAMPLES:
        if to_sample_rate is None:
            raise ValueError("to_sample_rate required when converting to SAMPLES")
        return cls.tosamples(offset_val, to_sample_rate)
    else:
        raise ValueError(f"Unknown to_unit: {to_unit}")

fromns(nanoseconds, sample_rate=None) staticmethod

Convert nanoseconds to offsets.

Parameters:

Name Type Description Default
nanoseconds int

int, the time to convert to offsets, in nanoseconds

 required
sample_rate Optional[int]

int, optional sample rate to align the offset to. If provided, the offset will be rounded to the nearest sample boundary for this rate.

 None

Returns:

Type Description
int

int, the offset corresponding to the time
Source code in sgnts/base/offset.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@staticmethod
def fromns(nanoseconds: int, sample_rate: Optional[int] = None) -> int:
    """Convert nanoseconds to offsets.

    Args:
        nanoseconds:
            int, the time to convert to offsets, in nanoseconds
        sample_rate:
            int, optional sample rate to align the offset to. If provided,
            the offset will be rounded to the nearest sample boundary for
            this rate.

    Returns:
        int, the offset corresponding to the time
    """
    if sample_rate is None:
        # Standard behavior - convert directly
        return round(int(nanoseconds) / int(Time.SECONDS) * Offset.MAX_RATE)
    else:
        # Align to sample boundary
        assert (
            sample_rate in Offset.ALLOWED_RATES
        ), f"Invalid sample rate: {sample_rate}"

        # Convert nanoseconds to samples at the given rate
        samples = round(nanoseconds * sample_rate / Time.SECONDS)

        # Convert samples back to offset - this ensures alignment
        return Offset.fromsamples(int(samples), sample_rate)

fromsamples(samples, sample_rate) staticmethod

Convert number of sample points to offsets.

Parameters:

Name Type Description Default
samples int

int, the number of samples to convert to offsets

 required
sample_rate int

int, the sample rate at which to calculate the offset

 required

Returns:

Type Description
int

int, the offset corresponding to the number of sample points at the given
int

sample rate
Source code in sgnts/base/offset.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
@staticmethod
def fromsamples(samples: int, sample_rate: int) -> int:
    """Convert number of sample points to offsets.

    Args:
        samples:
            int, the number of samples to convert to offsets
        sample_rate:
            int, the sample rate at which to calculate the offset

    Returns:
        int, the offset corresponding to the number of sample points at the given
        sample rate
    """
    assert (
        sample_rate in Offset.ALLOWED_RATES
    ), f"Sample rate {sample_rate} not in ALLOWED_RATES: {Offset.ALLOWED_RATES}"
    return samples * Offset.MAX_RATE // sample_rate

fromsec(seconds) staticmethod

Convert seconds to offsets.

Parameters:

Name Type Description Default
seconds float

float, the time to convert to offsets, in seconds

 required

Returns:

Type Description
int

int, the offset corresponding to the time
Source code in sgnts/base/offset.py 
205
206
207
208
209
210
211
212
213
214
215
216
@staticmethod
def fromsec(seconds: float) -> int:
    """Convert seconds to offsets.

    Args:
        seconds:
            float, the time to convert to offsets, in seconds

    Returns:
        int, the offset corresponding to the time
    """
    return round(seconds * Offset.MAX_RATE)

sample_stride(rate) staticmethod

Given Offset.SAMPLE_STRIDE_AT_MAX_RATE, derive the sample stride at the requested sample rate.

Parameters:

Name Type Description Default
rate int

int, the sample rate to calculate the sample stride

 required

Returns:

Type Description
int

int, the number of samples in the stride at the requested sample rate
Source code in sgnts/base/offset.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
@staticmethod
def sample_stride(rate: int) -> int:
    """Given Offset.SAMPLE_STRIDE_AT_MAX_RATE, derive the sample stride at the
    requested sample rate.

    Args:
        rate:
            int, the sample rate to calculate the sample stride

    Returns:
       int, the number of samples in the stride at the requested sample rate
    """
    return Offset.tosamples(Offset.SAMPLE_STRIDE_AT_MAX_RATE, rate)

tons(offset) staticmethod

Convert offsets to integer nanoseconds.

Parameters:

Name Type Description Default
offset int

int, the offset to convert to nanoseconds

 required

Returns:

Type Description
int

int, the time corresponding to the offset, in nanoseconds

NOTE - for very large offsets, this switches to integer arithmetic to preserve precision. A downside is that the result will be truncated rather than rounded, leading to a slight bias at the 1 ns scale in some case. This is unlikely to cause any actual problems. As a reminder, all serious bookkeeping should be done with offsets not timestamps.

Source code in sgnts/base/offset.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@staticmethod
def tons(offset: int) -> int:
    """Convert offsets to integer nanoseconds.

    Args:
        offset:
            int, the offset to convert to nanoseconds

    Returns:
        int, the time corresponding to the offset, in nanoseconds

    NOTE - for very large offsets, this switches to integer arithmetic to
    preserve precision. A downside is that the result will be truncated
    rather than rounded, leading to a slight bias at the 1 ns scale in
    some case. This is unlikely to cause any actual problems.  As a
    reminder, all serious bookkeeping should be done with offsets not
    timestamps.
    """
    if offset * Offset.MAX_RATE > 1e17:
        return offset * Time.SECONDS // Offset.MAX_RATE
    else:
        return round(offset / Offset.MAX_RATE * Time.SECONDS)

tosamples(offset, sample_rate) staticmethod

Convert offsets to number of sample points.

Parameters:

Name Type Description Default
offset int

int, the offset to convert to number of samples. The offset must map to integer number of sample points.

 required
sample_rate int

int, the sample rate at which to calculate the number of samples

 required

Returns:

Type Description
int

int, the number of samples corresponding to the offset at the given sample
int

rate
Source code in sgnts/base/offset.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
@staticmethod
def tosamples(offset: int, sample_rate: int) -> int:
    """Convert offsets to number of sample points.

    Args:
        offset:
            int, the offset to convert to number of samples. The offset must map to
            integer number of sample points.
        sample_rate:
            int, the sample rate at which to calculate the number of samples

    Returns:
        int, the number of samples corresponding to the offset at the given sample
        rate
    """
    assert (
        sample_rate in Offset.ALLOWED_RATES
    ), f"Sample rate {sample_rate} not in ALLOWED_RATES: {Offset.ALLOWED_RATES}"
    assert not offset % (Offset.MAX_RATE // sample_rate), (
        "Offset does not map to"
        f" integer sample points. Offset: {offset}, sample rate: {sample_rate}"
    )
    return offset // (Offset.MAX_RATE // sample_rate)

tosec(offset) staticmethod

Convert offsets to seconds.

Parameters:

Name Type Description Default
offset int

int, the offset to convert to seconds

 required

Returns:

Type Description
float

float, the time corresponding to the offset, in seconds
Source code in sgnts/base/offset.py 
169
170
171
172
173
174
175
176
177
178
179
180
@staticmethod
def tosec(offset: int) -> float:
    """Convert offsets to seconds.

    Args:
        offset:
            int, the offset to convert to seconds

    Returns:
        float, the time corresponding to the offset, in seconds
    """
    return offset / Offset.MAX_RATE

TimeUnits

Bases: str, Enum


              flowchart TD
              sgnts.base.offset.TimeUnits[TimeUnits]

              

              click sgnts.base.offset.TimeUnits href "" "sgnts.base.offset.TimeUnits"

Enumeration of available time units for TSSlices.

Source code in sgnts/base/offset.py 
11
12
13
14
15
16
17
class TimeUnits(str, Enum):
    """Enumeration of available time units for TSSlices."""

    OFFSETS = "offsets"
    SECONDS = "seconds"
    NANOSECONDS = "nanoseconds"
    SAMPLES = "samples"