Skip to content

sgnts.transforms.bit_vector

BitVector dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.bit_vector.BitVector[BitVector]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.bit_vector.BitVector
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.bit_vector.BitVector href "" "sgnts.transforms.bit_vector.BitVector"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Generate integer-valued output encoding the state of N input streams.

Takes N input streams and produces a single output stream containing single-channel integer values. Each value is calculated by interpreting the buffer/gap state of all inputs as a binary number.

Unlike ANDTransform which outputs gaps where any input has a gap, BitVector always outputs buffers (never gaps), while also preserving information about the gap/buffer status of each input.

Bit assignment can be configured in two ways:

  1. Sequential (default): Pads are assigned to bit positions in order, starting from bit 0 (least significant). With 3 pads: pad0 -> bit 0, pad1 -> bit 1, pad2 -> bit 2.

  2. Explicit via bit_map: A dict mapping bit positions to sink pad names, e.g. {0: "intent", 2: "quality"}. Bits not present in bit_map or on_bits default to 0.

Additionally, on_bits allows setting bit positions to always 1 independent of any input, e.g. [9, 12]. Bit positions not in bit_map, on_bits, or the sequential assignment default to 0.

Output characteristics
  • Sample rate: Configurable via output_rate parameter
  • Data type: uint32
  • Shape: (1, num_samples) - single channel with integer values
  • Always produces buffers, never gaps
For example

3 inputs with different gap patterns at rates [64, 128, 256] Hz Input 0: buffer at t=0.1-0.5s, gap elsewhere Input 1: buffer at t=0.3-0.7s, gap elsewhere Input 2: gap everywhere With output_rate=128: At t=0.05s: 0 (binary 000 = decimal 0) At t=0.2s: 1 (binary 001 = decimal 1) At t=0.4s: 3 (binary 011 = decimal 3) At t=0.6s: 2 (binary 010 = decimal 2)

Parameters:

Name Type Description Default
output_rate Optional[int]

int, the sample rate for the output stream in Hz. Default: None (uses minimum rate among all inputs)

 None
bit_map Optional[dict[int, str]]

Optional dict mapping bit positions (int) to sink pad names (str). When provided, bits are assigned according to this mapping instead of sequentially.

 None
on_bits list[int]

List of bit positions (int) that are always set to 1, regardless of input state.

 list()
Notes

Thread safety: Marked thread_safe = True. Pad layout: N sink pads + 1 source pad (@transform.many_to_one). The N sink pads' pull callbacks CAN run concurrently in the same wave; internal runs alone.

``pull`` (inherited ``TimeSeriesMixin.pull``):
per-pad-keyed dict writes; safe across pads. ``new``
(inherited): read-only ``self.outframes`` lookup.
``process``/``_buf_to_bits``: NumPy reshape/repeat/sum/
comparison kernels (release the GIL for large arrays) on
local data; reads ``self.bit_map`` and ``self.on_bits``
(both post-init read-only).

**Future editors MUST preserve thread safety**: keep
``process``/``_buf_to_bits`` purely functional on their
inputs. Do NOT add element-level state mutated from
``pull`` outside of per-pad-keyed containers.
Source code in src/sgnts/transforms/bit_vector.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
@dataclass
class BitVector(TSTransform):
    """Generate integer-valued output encoding the state of N input streams.

    Takes N input streams and produces a single output stream containing
    single-channel integer values. Each value is calculated by interpreting
    the buffer/gap state of all inputs as a binary number.

    Unlike ANDTransform which outputs gaps where any input has a gap,
    BitVector always outputs buffers (never gaps), while also preserving
    information about the gap/buffer status of each input.

    Bit assignment can be configured in two ways:

    1. Sequential (default): Pads are assigned to bit positions in order,
       starting from bit 0 (least significant). With 3 pads: pad0 -> bit 0,
       pad1 -> bit 1, pad2 -> bit 2.

    2. Explicit via ``bit_map``: A dict mapping bit positions to sink pad
       names, e.g. ``{0: "intent", 2: "quality"}``. Bits not present in
       ``bit_map`` or ``on_bits`` default to 0.

    Additionally, ``on_bits`` allows setting bit positions to always 1
    independent of any input, e.g. ``[9, 12]``. Bit positions not in
    ``bit_map``, ``on_bits``, or the sequential assignment default to 0.

    Output characteristics:
        - Sample rate: Configurable via output_rate parameter
        - Data type: uint32
        - Shape: (1, num_samples) - single channel with integer values
        - Always produces buffers, never gaps

    For example:
        3 inputs with different gap patterns at rates [64, 128, 256] Hz
        Input 0: buffer at t=0.1-0.5s, gap elsewhere
        Input 1: buffer at t=0.3-0.7s, gap elsewhere
        Input 2: gap everywhere
        With output_rate=128:
        At t=0.05s: 0 (binary 000 = decimal 0)
        At t=0.2s: 1 (binary 001 = decimal 1)
        At t=0.4s: 3 (binary 011 = decimal 3)
        At t=0.6s: 2 (binary 010 = decimal 2)

    Args:
        output_rate:
            int, the sample rate for the output stream in Hz.
            Default: None (uses minimum rate among all inputs)
        bit_map:
            Optional dict mapping bit positions (int) to sink pad names
            (str). When provided, bits are assigned according to this
            mapping instead of sequentially.
        on_bits:
            List of bit positions (int) that are always set to 1,
            regardless of input state.

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: N sink pads +
            1 source pad (``@transform.many_to_one``). The N sink
            pads' ``pull`` callbacks CAN run concurrently in the same
            wave; ``internal`` runs alone.

            ``pull`` (inherited ``TimeSeriesMixin.pull``):
            per-pad-keyed dict writes; safe across pads. ``new``
            (inherited): read-only ``self.outframes`` lookup.
            ``process``/``_buf_to_bits``: NumPy reshape/repeat/sum/
            comparison kernels (release the GIL for large arrays) on
            local data; reads ``self.bit_map`` and ``self.on_bits``
            (both post-init read-only).

            **Future editors MUST preserve thread safety**: keep
            ``process``/``_buf_to_bits`` purely functional on their
            inputs. Do NOT add element-level state mutated from
            ``pull`` outside of per-pad-keyed containers.
    """

    thread_safe = True

    # numpy bit-twiddling (asarray / reshape / repeat / uint32).
    backends = frozenset({"numpy"})

    output_rate: Optional[int] = None
    bit_map: Optional[dict[int, str]] = None
    on_bits: list[int] = field(default_factory=list)

    def configure(self) -> None:
        """Initialize transform with buffer alignment enabled."""
        self.adapter_config = AdapterConfig(align_buffers=True)

    def output_prototype(self, pad):
        # Emits a uint32 bit field regardless of input dtype (numpy-only element).
        return self.input_prototype(self.sink_pad_names[0], dtype=np.uint32)

    def _buf_to_bits(
        self, buf: SeriesBuffer, output_rate: int, num_samples: int
    ) -> np.ndarray:
        """Convert a buffer to a per-sample binary array at output_rate.

        Gap buffers produce all zeros. Non-gap buffers are checked per sample
        for truthiness (nonzero = 1). If the buffer's sample rate exceeds
        output_rate, logical downsampling is applied: each chunk is reduced
        via AND so that any zero in the chunk produces a zero output. If
        the buffer's sample rate is below output_rate, values are repeated.
        """
        if buf.is_gap:
            return np.zeros(num_samples, dtype=np.uint32)

        bit_values = (np.asarray(buf.data).flatten() != 0).astype(np.uint32)
        if buf.sample_rate > output_rate:
            chunk_size = buf.sample_rate // output_rate
            bit_values = np.all(bit_values.reshape(-1, chunk_size), axis=1).astype(
                np.uint32
            )
        elif buf.sample_rate < output_rate:
            repeat_factor = output_rate // buf.sample_rate
            bit_values = np.repeat(bit_values, repeat_factor)
        return bit_values

    @transform.many_to_one
    def process(
        self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
    ) -> None:
        """Generate output frame encoding input state as per-sample integers.

        Each input buffer is interpreted per sample: gap samples and zero-valued
        samples contribute 0, nonzero samples contribute 1. Inputs at higher
        sample rates than the output are logically downsampled (conservative
        AND per chunk).

        Algorithm:
            1. Get all aligned frames (aligned to same boundaries via align_buffers)
            2. Determine output sample rate
            3. All frames have same number of buffers after alignment
            4. For each buffer index (time region):
               a. Convert each input to a per-sample binary array at output_rate
               b. Weight each array by its bit position (2^pos)
               c. Sum to produce per-sample integer output
            5. Append all output buffers to output_frame
        """
        if self.output_rate is None:
            output_rate = min(f.sample_rate for f in input_frames.values())
        else:
            output_rate = self.output_rate

        on_bits_value = np.uint32(sum(2**pos for pos in self.on_bits))

        pads = list(input_frames.keys())

        for input_buffers in zip(*input_frames.values()):
            num_samples = Offset.tosamples(input_buffers[0].noffset, output_rate)
            result = np.full(num_samples, on_bits_value, dtype=np.uint32)

            if self.bit_map is not None:
                pad_bufs = dict(zip(pads, input_buffers))
                for bit_pos, pad_name in self.bit_map.items():
                    pad = self.snks[pad_name]
                    bit_values = self._buf_to_bits(
                        pad_bufs[pad], output_rate, num_samples
                    )
                    result += bit_values * np.uint32(2**bit_pos)
            else:
                for i, buf in enumerate(input_buffers):
                    bit_values = self._buf_to_bits(buf, output_rate, num_samples)
                    result += bit_values * np.uint32(2**i)

            buffer = SeriesBuffer(
                offset=input_buffers[0].offset,
                sample_rate=output_rate,
                data=result.reshape(1, -1),
                shape=(1, num_samples),
            )
            output_frame.append(buffer)

configure()

Initialize transform with buffer alignment enabled.

Source code in src/sgnts/transforms/bit_vector.py 
105
106
107
def configure(self) -> None:
    """Initialize transform with buffer alignment enabled."""
    self.adapter_config = AdapterConfig(align_buffers=True)

process(input_frames, output_frame)

Generate output frame encoding input state as per-sample integers.

Each input buffer is interpreted per sample: gap samples and zero-valued samples contribute 0, nonzero samples contribute 1. Inputs at higher sample rates than the output are logically downsampled (conservative AND per chunk).

Algorithm
  1. Get all aligned frames (aligned to same boundaries via align_buffers)
  2. Determine output sample rate
  3. All frames have same number of buffers after alignment
  4. For each buffer index (time region): a. Convert each input to a per-sample binary array at output_rate b. Weight each array by its bit position (2^pos) c. Sum to produce per-sample integer output
  5. Append all output buffers to output_frame
Source code in src/sgnts/transforms/bit_vector.py 
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
@transform.many_to_one
def process(
    self, input_frames: dict[SinkPad, TSFrame], output_frame: TSCollectFrame
) -> None:
    """Generate output frame encoding input state as per-sample integers.

    Each input buffer is interpreted per sample: gap samples and zero-valued
    samples contribute 0, nonzero samples contribute 1. Inputs at higher
    sample rates than the output are logically downsampled (conservative
    AND per chunk).

    Algorithm:
        1. Get all aligned frames (aligned to same boundaries via align_buffers)
        2. Determine output sample rate
        3. All frames have same number of buffers after alignment
        4. For each buffer index (time region):
           a. Convert each input to a per-sample binary array at output_rate
           b. Weight each array by its bit position (2^pos)
           c. Sum to produce per-sample integer output
        5. Append all output buffers to output_frame
    """
    if self.output_rate is None:
        output_rate = min(f.sample_rate for f in input_frames.values())
    else:
        output_rate = self.output_rate

    on_bits_value = np.uint32(sum(2**pos for pos in self.on_bits))

    pads = list(input_frames.keys())

    for input_buffers in zip(*input_frames.values()):
        num_samples = Offset.tosamples(input_buffers[0].noffset, output_rate)
        result = np.full(num_samples, on_bits_value, dtype=np.uint32)

        if self.bit_map is not None:
            pad_bufs = dict(zip(pads, input_buffers))
            for bit_pos, pad_name in self.bit_map.items():
                pad = self.snks[pad_name]
                bit_values = self._buf_to_bits(
                    pad_bufs[pad], output_rate, num_samples
                )
                result += bit_values * np.uint32(2**bit_pos)
        else:
            for i, buf in enumerate(input_buffers):
                bit_values = self._buf_to_bits(buf, output_rate, num_samples)
                result += bit_values * np.uint32(2**i)

        buffer = SeriesBuffer(
            offset=input_buffers[0].offset,
            sample_rate=output_rate,
            data=result.reshape(1, -1),
            shape=(1, num_samples),
        )
        output_frame.append(buffer)