Skip to content

sgnts.transforms.align

Align dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.align.Align[Align]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.align.Align
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.align.Align href "" "sgnts.transforms.align.Align"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Align frames from multiple sink pads.

Notes

Thread safety: Marked thread_safe = True. Pad layout: matched sink and source pads (@validator.pad_names_match). Multiple sink pads' pull callbacks CAN run concurrently in the same wave, and so can multiple source pads' new callbacks; internal runs alone.

``pull`` (inherited ``TimeSeriesMixin.pull``):
per-pad-keyed dict writes; safe across pads. ``new``
(inherited): read-only ``self.outframes`` lookup, safe.
``process``: pure pass-through — reads ``self.pad_map``
(set in ``configure()``, read-only) and copies frames
from sink to source dicts; no element-level mutation.

Direct speedup is none (pass-through, no GIL-releasing
compute), but marking thread_safe lets the element run
alongside other thread_safe elements without a
serialization point.

**Future editors MUST preserve thread safety**: keep
``process`` stateless. Do NOT add element-level state
mutated from ``pull`` outside of per-pad-keyed containers.
Source code in src/sgnts/transforms/align.py 
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
@dataclass
class Align(TSTransform):
    """Align frames from multiple sink pads.

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: matched sink
            and source pads (``@validator.pad_names_match``).
            Multiple sink pads' ``pull`` callbacks CAN run concurrently
            in the same wave, and so can multiple source pads' ``new``
            callbacks; ``internal`` runs alone.

            ``pull`` (inherited ``TimeSeriesMixin.pull``):
            per-pad-keyed dict writes; safe across pads. ``new``
            (inherited): read-only ``self.outframes`` lookup, safe.
            ``process``: pure pass-through — reads ``self.pad_map``
            (set in ``configure()``, read-only) and copies frames
            from sink to source dicts; no element-level mutation.

            Direct speedup is none (pass-through, no GIL-releasing
            compute), but marking thread_safe lets the element run
            alongside other thread_safe elements without a
            serialization point.

            **Future editors MUST preserve thread safety**: keep
            ``process`` stateless. Do NOT add element-level state
            mutated from ``pull`` outside of per-pad-keyed containers.
    """

    thread_safe = True

    def configure(self) -> None:
        self.pad_map = {
            src_pad: self.snks[src_pad.pad_name] for src_pad in self.source_pads
        }

    @validator.pad_names_match
    def validate(self) -> None:
        pass

    def output_prototype(self, pad: SourcePad) -> Array:
        # Pass-through: each output pad inherits its matched input pad's spec,
        # which stays well-defined even when the pads carry different dtypes.
        return self.input_prototype(self.pad_map[pad].pad_name)

    def process(
        self,
        input_frames: dict[SinkPad, TSFrame],
        output_frames: dict[SourcePad, TSCollectFrame],
    ) -> None:
        """Pass through frames from sink to source."""
        # just pass through frames from sink to source
        for src_pad, sink_pad in self.pad_map.items():
            output_frames[src_pad].extend(input_frames[sink_pad])

process(input_frames, output_frames)

Pass through frames from sink to source.

Source code in src/sgnts/transforms/align.py 
55
56
57
58
59
60
61
62
63
def process(
    self,
    input_frames: dict[SinkPad, TSFrame],
    output_frames: dict[SourcePad, TSCollectFrame],
) -> None:
    """Pass through frames from sink to source."""
    # just pass through frames from sink to source
    for src_pad, sink_pad in self.pad_map.items():
        output_frames[src_pad].extend(input_frames[sink_pad])