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 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 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
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))))

    @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
        """
        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) -> int:
        """Convert nanoseconds to offsets.

        Args:
            nanoseconds:
                int, the time to convert to offsets, in nanoseconds

        Returns:
            int, the offset corresponding to the time
        """
        return round(nanoseconds / Time.SECONDS * Offset.MAX_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
        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
        return samples * Offset.MAX_RATE // sample_rate

fromns(nanoseconds) staticmethod

Convert nanoseconds to offsets.

Parameters:

Name Type Description Default
nanoseconds int

int, the time to convert to offsets, in nanoseconds

 required

Returns:

Type Description
int

int, the offset corresponding to the time
Source code in sgnts/base/offset.py 
127
128
129
130
131
132
133
134
135
136
137
138
@staticmethod
def fromns(nanoseconds: int) -> int:
    """Convert nanoseconds to offsets.

    Args:
        nanoseconds:
            int, the time to convert to offsets, in nanoseconds

    Returns:
        int, the offset corresponding to the time
    """
    return round(nanoseconds / Time.SECONDS * Offset.MAX_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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@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
    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 
114
115
116
117
118
119
120
121
122
123
124
125
@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 
74
75
76
77
78
79
80
81
82
83
84
85
86
@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
Source code in sgnts/base/offset.py 
101
102
103
104
105
106
107
108
109
110
111
112
@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
    """
    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 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@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
    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 
88
89
90
91
92
93
94
95
96
97
98
99
@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