Skip to content

sgn.sinks

Sink elements for the SGN framework.

CollectSink dataclass

Bases: SinkElement


              flowchart TD
              sgn.sinks.CollectSink[CollectSink]
              sgn.base.SinkElement[SinkElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]

                              sgn.base.SinkElement --> sgn.sinks.CollectSink
                                sgn.base.ElementLike --> sgn.base.SinkElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                




              click sgn.sinks.CollectSink href "" "sgn.sinks.CollectSink"
              click sgn.base.SinkElement href "" "sgn.base.SinkElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"

A sink element that has one collection per sink pad.

Each frame that is pulled into the sink is added to the collection for that pad using a ".append" method. If the extract_data flag is set, the data is extracted from the frame and added to the deque, otherwise the frame itself is added to the collection.

Parameters:

Name Type Description Default
collects dict[str, MutableSequence]

dict[str, Collection], a mapping of sink pad names to Collections. The Collection must have an append method.

 dict()
extract_data bool

bool, default True, flag to indicate if the data should be extracted from the frame before adding it to the deque

 True
skip_empty bool

bool, default True, flag to indicate the frame should be skipped and not collected if its data payload is None.

 True
Source code in sgn/sinks.py 
 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
@dataclass
class CollectSink(SinkElement):
    """A sink element that has one collection per sink pad.

    Each frame that is pulled into the sink is added to the collection
    for that pad using a ".append" method. If the extract_data flag is
    set, the data is extracted from the frame and added to the deque,
    otherwise the frame itself is added to the collection.

    Args:
        collects:
            dict[str, Collection], a mapping of sink pad names to
            Collections. The Collection must have an append method.
        extract_data:
            bool, default True, flag to indicate if the data should be
            extracted from the frame before adding it to the deque
        skip_empty:
            bool, default True, flag to indicate the frame should be
            skipped and not collected if its `data` payload is None.

    """

    collects: dict[str, MutableSequence] = field(default_factory=dict)
    extract_data: bool = True
    skip_empty: bool = True
    collection_factory: Callable = list

    def __post_init__(self):
        super().__post_init__()
        if not self.collects:
            self.collects = {
                name: self.collection_factory() for name in self.sink_pad_names
            }
        else:
            self.collects = {
                name: self.collection_factory(iterable)
                for name, iterable in self.collects.items()
            }
        assert set(self.collects) == set(
            self.sink_pad_names
        ), "The `collects` attribute keys should match sink_pad_names"

    def pull(self, pad: SinkPad, frame: Frame) -> None:
        """Pull in frame and add it to pad collection.

        Args:
            pad:
                SinkPad, the pad that the frame is pulled into
            frame:
                Frame, the frame that is pulled into the sink
        """
        if frame.EOS:
            self.mark_eos(pad)
        if self.skip_empty and frame.data is None:
            return
        self.collects[self.rsnks[pad]].append(
            frame.data if self.extract_data else frame
        )

pull(pad, frame)

Pull in frame and add it to pad collection.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, the pad that the frame is pulled into

 required
frame Frame

Frame, the frame that is pulled into the sink

 required
Source code in sgn/sinks.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def pull(self, pad: SinkPad, frame: Frame) -> None:
    """Pull in frame and add it to pad collection.

    Args:
        pad:
            SinkPad, the pad that the frame is pulled into
        frame:
            Frame, the frame that is pulled into the sink
    """
    if frame.EOS:
        self.mark_eos(pad)
    if self.skip_empty and frame.data is None:
        return
    self.collects[self.rsnks[pad]].append(
        frame.data if self.extract_data else frame
    )

DequeSink dataclass

Bases: CollectSink


              flowchart TD
              sgn.sinks.DequeSink[DequeSink]
              sgn.sinks.CollectSink[CollectSink]
              sgn.base.SinkElement[SinkElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]

                              sgn.sinks.CollectSink --> sgn.sinks.DequeSink
                                sgn.base.SinkElement --> sgn.sinks.CollectSink
                                sgn.base.ElementLike --> sgn.base.SinkElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                





              click sgn.sinks.DequeSink href "" "sgn.sinks.DequeSink"
              click sgn.sinks.CollectSink href "" "sgn.sinks.CollectSink"
              click sgn.base.SinkElement href "" "sgn.base.SinkElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"

A sink element that has one double-ended-queue (deque) per sink pad.

Each frame that is pulled into the sink is added to the deque for that pad. If the extract_data flag is set, the data is extracted from the frame and added to the deque , otherwise the frame itself is added to the deque.

Parameters:

Name Type Description Default
collects dict[str, MutableSequence]

dict[str, deque], a mapping of sink pads to deques, where the key is the pad name and the value is the deque

 dict()
extract_data bool

bool, default True, flag to indicate if the data should be extracted from the frame before adding it to the deque

 True
Notes

Ignoring empty frames: If the frame is empty, it is not added to the deque. The motivating principle is that "empty frames preserve the sink deque". An empty deque is equivalent (for our purposes) to a deque filled with "None" values, so we prevent the latter from being possible.

Source code in sgn/sinks.py 
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
@dataclass
class DequeSink(CollectSink):
    """A sink element that has one double-ended-queue (deque) per sink pad.

    Each frame that is pulled into the sink is added to the deque for
    that pad.  If the extract_data flag is set, the data is extracted
    from the frame and added to the deque , otherwise the frame itself
    is added to the deque.

    Args:
        collects:
            dict[str, deque], a mapping of sink pads to deques, where
            the key is the pad name and the value is the deque

        extract_data:
            bool, default True, flag to indicate if the data should be
            extracted from the frame before adding it to the deque

    Notes:
        Ignoring empty frames:
            If the frame is empty, it is not added to the deque. The
            motivating principle is that "empty frames preserve the
            sink deque". An empty deque is equivalent (for our
            purposes) to a deque filled with "None" values, so we
            prevent the latter from being possible.

    """

    collection_factory: Callable = deque

    @property
    def deques(self) -> dict[str, MutableSequence]:
        """Explicit alias for collects.

        Returns:
            dict[str, deque ]: the deques for the sink
        """
        return self.collects

deques property

Explicit alias for collects.

Returns:

Type Description
dict[str, MutableSequence]

dict[str, deque ]: the deques for the sink

NullSink dataclass

Bases: SinkElement


              flowchart TD
              sgn.sinks.NullSink[NullSink]
              sgn.base.SinkElement[SinkElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]

                              sgn.base.SinkElement --> sgn.sinks.NullSink
                                sgn.base.ElementLike --> sgn.base.SinkElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                




              click sgn.sinks.NullSink href "" "sgn.sinks.NullSink"
              click sgn.base.SinkElement href "" "sgn.base.SinkElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"

A sink that does precisely nothing.

It is useful for testing and debugging, or for pipelines that do not need a sink, but require one to be present in the pipeline.

Parameters:

Name Type Description Default
verbose bool

bool, print frames as they pass through the internal pad

 False
Source code in sgn/sinks.py 
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
@dataclass
class NullSink(SinkElement):
    """A sink that does precisely nothing.

    It is useful for testing and debugging, or for pipelines that do
    not need a sink, but require one to be present in the pipeline.

    Args:
        verbose:
            bool, print frames as they pass through the internal pad

    """

    verbose: bool = False

    def pull(self, pad: SinkPad, frame: Frame) -> None:
        """Do nothing on pull.

        Args:
            pad:
                SinkPad, the pad that the frame is pulled into
            frame:
                Frame, the frame that is pulled into the sink
        """
        if frame.EOS:
            self.mark_eos(pad)
        if self.verbose is True:
            print(frame)

pull(pad, frame)

Do nothing on pull.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, the pad that the frame is pulled into

 required
frame Frame

Frame, the frame that is pulled into the sink

 required
Source code in sgn/sinks.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
def pull(self, pad: SinkPad, frame: Frame) -> None:
    """Do nothing on pull.

    Args:
        pad:
            SinkPad, the pad that the frame is pulled into
        frame:
            Frame, the frame that is pulled into the sink
    """
    if frame.EOS:
        self.mark_eos(pad)
    if self.verbose is True:
        print(frame)