Skip to content

sgnts.compose

Composable time-series elements for SGN-TS.

This module provides TS-flavored subclasses of the sgn.compose composed element classes. Composed elements are shells -- they don't process data themselves (no alignment, adapters, or outframes). All processing is delegated to the internal elements, so the TS composed classes bypass TimeSeriesMixin.__post_init__ and delegate pull(), new(), and internal() to the sgn.compose base classes.

Define a dataclass that inherits from one of the three TS composed element base classes, declare parameters as dataclass fields, and override _build() to wire the internal elements using self.insert() and self.connect(). The resulting object IS a TS element -- pass it directly to pipeline.connect().

>>> from dataclasses import dataclass
>>> from sgnts.compose import TSComposedTransformElement
>>> from sgnts.transforms import Amplify
>>>
>>> @dataclass(repr=False, kw_only=True)
... class ScaleChain(TSComposedTransformElement):
...     factor_a: float = 1.0
...     factor_b: float = 1.0
...
...     def _build(self):
...         a = Amplify(
...             name=f"{self.name}_a", factor=self.factor_a,
...             source_pad_names=["data"], sink_pad_names=["data"],
...         )
...         b = Amplify(
...             name=f"{self.name}_b", factor=self.factor_b,
...             source_pad_names=["data"], sink_pad_names=["data"],
...         )
...         self.connect(a, b)
...
>>> composed = ScaleChain(name="sc", factor_a=2.0, factor_b=3.0)
>>> pipeline.connect(upstream, composed)

TSComposedSinkElement dataclass

Bases: TSSink, ComposedSinkElement


              flowchart TD
              sgnts.compose.TSComposedSinkElement[TSComposedSinkElement]
              sgnts.base.base.TSSink[TSSink]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSSink --> sgnts.compose.TSComposedSinkElement
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSSink
                



              click sgnts.compose.TSComposedSinkElement href "" "sgnts.compose.TSComposedSinkElement"
              click sgnts.base.base.TSSink href "" "sgnts.base.base.TSSink"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Composable sink element for SGN-TS.

Bypasses TimeSeriesMixin.__post_init__ and delegates pull() and internal() to ComposedSinkElement, skipping the TS alignment/adapter infrastructure. The internal sink elements handle alignment and EOS propagation themselves.

Source code in src/sgnts/compose.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@dataclass(kw_only=True)
class TSComposedSinkElement(TSSink, ComposedSinkElement):
    """Composable sink element for SGN-TS.

    Bypasses ``TimeSeriesMixin.__post_init__`` and delegates ``pull()``
    and ``internal()`` to ``ComposedSinkElement``, skipping the TS
    alignment/adapter infrastructure. The internal sink elements handle
    alignment and EOS propagation themselves.
    """

    def __post_init__(self):
        # Skip TSSink / TimeSeriesMixin __post_init__; go straight to the
        # composed element, which builds the internal graph and pads.
        ComposedSinkElement.__post_init__(self)

    def pull(self, pad: SinkPad, frame: Frame) -> None:
        ComposedSinkElement.pull(self, pad, frame)

    def internal(self) -> None:
        ComposedSinkElement.internal(self)

TSComposedSourceElement dataclass

Bases: TSSource, ComposedSourceElement


              flowchart TD
              sgnts.compose.TSComposedSourceElement[TSComposedSourceElement]
              sgnts.base.base.TSSource[TSSource]
              sgnts.base.base._TSSource[_TSSource]

                              sgnts.base.base.TSSource --> sgnts.compose.TSComposedSourceElement
                                sgnts.base.base._TSSource --> sgnts.base.base.TSSource
                



              click sgnts.compose.TSComposedSourceElement href "" "sgnts.compose.TSComposedSourceElement"
              click sgnts.base.base.TSSource href "" "sgnts.base.base.TSSource"
              click sgnts.base.base._TSSource href "" "sgnts.base.base._TSSource"

Composable source element for SGN-TS.

Bypasses _TSSource.__post_init__ and delegates new() and internal() to ComposedSourceElement, skipping TSSource's frame-preparation logic (buffer params, stride generators, EOS handling). The internal source elements handle all of that themselves.

Source code in src/sgnts/compose.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass(kw_only=True)
class TSComposedSourceElement(TSSource, ComposedSourceElement):
    """Composable source element for SGN-TS.

    Bypasses ``_TSSource.__post_init__`` and delegates ``new()`` and
    ``internal()`` to ``ComposedSourceElement``, skipping TSSource's
    frame-preparation logic (buffer params, stride generators, EOS
    handling). The internal source elements handle all of that themselves.
    """

    def __post_init__(self):
        # Skip TSSource / _TSSource __post_init__; go straight to the
        # composed element, which builds the internal graph and pads.
        ComposedSourceElement.__post_init__(self)

    def new(self, pad: SourcePad) -> Frame:
        return ComposedSourceElement.new(self, pad)

    def internal(self) -> None:
        ComposedSourceElement.internal(self)

TSComposedTransformElement dataclass

Bases: TSTransform, ComposedTransformElement


              flowchart TD
              sgnts.compose.TSComposedTransformElement[TSComposedTransformElement]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.compose.TSComposedTransformElement
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.compose.TSComposedTransformElement href "" "sgnts.compose.TSComposedTransformElement"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Composable transform element for SGN-TS.

Bypasses TimeSeriesMixin.__post_init__ and delegates pull(), new(), and internal() to ComposedTransformElement so frames flow through the internal element graph instead of the TS alignment system.

Source code in src/sgnts/compose.py 
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
@dataclass(kw_only=True)
class TSComposedTransformElement(TSTransform, ComposedTransformElement):
    """Composable transform element for SGN-TS.

    Bypasses ``TimeSeriesMixin.__post_init__`` and delegates ``pull()``,
    ``new()``, and ``internal()`` to ``ComposedTransformElement`` so
    frames flow through the internal element graph instead of the TS
    alignment system.
    """

    def __post_init__(self):
        # Skip TSTransform / TimeSeriesMixin __post_init__; go straight to
        # the composed element, which builds the internal graph and pads.
        # Composed elements are shells -- alignment, adapters, and outframes
        # are handled by the internal elements, not this wrapper.
        ComposedTransformElement.__post_init__(self)

    def pull(self, pad: SinkPad, frame: TimeSpanFrame) -> None:  # type: ignore
        ComposedTransformElement.pull(self, pad, frame)

    def new(self, pad: SourcePad) -> TimeSpanFrame:  # type: ignore[override]
        return cast(TimeSpanFrame, ComposedTransformElement.new(self, pad))

    def internal(self) -> None:
        ComposedTransformElement.internal(self)