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.

Source code in sgnts/transforms/align.py

@dataclass
class Align(TSTransform):
    """Align frames from multiple sink pads."""

    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 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 sgnts/transforms/align.py

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])