Skip to content

Base Classes - Core Building Blocks

The sgn.base module provides the fundamental abstractions for building SGN pipelines: Elements and Pads.

Overview

SGN pipelines are built from three types of elements connected by pads:

  • SourceElement - Generates data (e.g., reading from files, sensors, streams)
  • TransformElement - Processes data (e.g., filtering, mapping, aggregating)
  • SinkElement - Consumes data (e.g., writing to files, databases, displays)

Elements communicate through pads:

  • SourcePad - Outputs data from an element
  • SinkPad - Receives data into an element

Quick Start: Creating Custom Elements

Creating a Source Element

A source element generates frames of data. You must implement the new() method:

from sgn.base import SourceElement, Frame

class MySource(SourceElement):
    def __init__(self, **kwargs):
        super().__init__(source_pad_names=["output"], **kwargs)
        self.counter = 0

    def new(self, pad):
        """Generate a new frame with incrementing counter."""
        self.counter += 1
        return Frame(data=self.counter)

# Create instance
source = MySource(name="counter_source")

Creating a Sink Element

A sink element consumes frames. You must implement the pull() method:

from sgn.base import SinkElement, Frame

class MySink(SinkElement):
    def __init__(self, **kwargs):
        super().__init__(sink_pad_names=["input"], **kwargs)
        self.received = []

    def pull(self, pad, frame):
        """Process incoming frame."""
        print(f"Received: {frame.data}")
        self.received.append(frame.data)

# Create instance
sink = MySink(name="printer_sink")

Creating a Transform Element

A transform element both receives and produces frames. Implement both pull() and new():

from sgn.base import TransformElement, Frame

class Multiplier(TransformElement):
    def __init__(self, factor=2, **kwargs):
        super().__init__(
            source_pad_names=["output"],
            sink_pad_names=["input"],
            **kwargs
        )
        self.factor = factor
        self.current_data = None

    def pull(self, pad, frame):
        """Receive and store incoming frame."""
        self.current_data = frame.data

    def new(self, pad):
        """Generate new frame with transformed data."""
        return Frame(data=self.current_data * self.factor)

# Create instance
transform = Multiplier(factor=10, name="multiplier")

Complete Example: Simple Pipeline

Here's a complete example combining all three element types:

from sgn.base import SourceElement, TransformElement, SinkElement, Frame
from sgn.apps import Pipeline

# Define custom elements
class CounterSource(SourceElement):
    def __init__(self, **kwargs):
        super().__init__(source_pad_names=["out"], **kwargs)
        self.count = 0

    def new(self, pad):
        self.count += 1
        if self.count > 5:  # Stop after 5 frames
            return Frame(EOS=True)
        return Frame(data=self.count)

class Doubler(TransformElement):
    def __init__(self, **kwargs):
        super().__init__(
            source_pad_names=["out"],
            sink_pad_names=["in"],
            **kwargs
        )
        self.current_frame = None

    def pull(self, pad, frame):
        self.current_frame = frame

    def new(self, pad):
        # Forward EOS if received
        if self.current_frame.EOS:
            return Frame(EOS=True)
        return Frame(data=self.current_frame.data * 2)

class PrinterSink(SinkElement):
    def __init__(self, **kwargs):
        super().__init__(sink_pad_names=["in"], **kwargs)

    def pull(self, pad, frame):
        if frame.EOS:
            self.mark_eos(pad)
            return
        print(f"Result: {frame.data}")

# Build and run pipeline
pipeline = Pipeline()
source = CounterSource()
transform = Doubler()
sink = PrinterSink()

pipeline.insert(source, transform, sink)
pipeline.link({
    transform.snks["in"]: source.srcs["out"],
    sink.snks["in"]: transform.srcs["out"]
})

pipeline.run()
# Output:
# Result: 2
# Result: 4
# Result: 6
# Result: 8
# Result: 10

Understanding Pads

Source Pads vs Sink Pads

  • SourcePad: Provides frames when called (outputs)
  • Created by SourceElement and TransformElement

  • Access via element.source_pads or element.srcs["name"]

  • SinkPad: Receives frames when called (inputs)

  • Created by TransformElement and SinkElement
  • Access via element.sink_pads or element.snks["name"]

Each pad has two name attributes:

  • pad.pad_name - The short name (e.g., "output")
  • pad.name - The full qualified name (e.g., "my_element:src:output")

Accessing Pads

Elements provide convenient shortcuts for accessing pads:

from sgn.base import SourceElement, Frame

class MySource(SourceElement):
    def __init__(self, source_pad_names=None, **kwargs):
        if source_pad_names is None:
            source_pad_names = ["output"]
        super().__init__(source_pad_names=source_pad_names, **kwargs)
        self.counter = 0

    def new(self, pad):
        self.counter += 1
        return Frame(data=self.counter)

source = MySource(source_pad_names=["out1", "out2"], name="mysrc")

# Multiple ways to access pads:
pad1 = source.source_pads[0]                  # By index
pad2 = source.srcs["out1"]                    # By short name (recommended)
pad3 = source.source_pad_dict["mysrc:src:out1"]  # By full pad name

# Pad naming attributes:
print(pad2.pad_name)  # "out1" - the short name
print(pad2.name)      # "mysrc:src:out1" - the full name

Multiple Pads

Elements can have multiple input/output pads:

from sgn.base import SourceElement, Frame

class MultiOutputSource(SourceElement):
    def __init__(self, **kwargs):
        super().__init__(
            source_pad_names=["numbers", "letters"],
            **kwargs
        )
        self.counter = 0

    def new(self, pad):
        self.counter += 1

        # Different output based on which pad is calling
        if pad == self.srcs["numbers"]:
            return Frame(data=self.counter)
        else:  # letters pad
            return Frame(data=chr(64 + self.counter))  # A, B, C...

source = MultiOutputSource()
# source.srcs["numbers"] outputs: 1, 2, 3, ...
# source.srcs["letters"] outputs: 'A', 'B', 'C', ...

Static Pads (Class-Level Pad Configuration)

For reusable element classes, you can define pads at the class level using static pads. This provides several benefits:

  • Consistency: All instances of your element have the same required pads
  • Simplicity: Users don't need to specify pad names when instantiating
  • Flexibility: Optionally allow users to add extra pads beyond the static ones

Class-Level Attributes

Elements support four class-level attributes for pad configuration:

Attribute Type Default Description
static_sink_pads ClassVar[list[str]] [] Sink pads that are always present
static_source_pads ClassVar[list[str]] [] Source pads that are always present
allow_dynamic_sink_pads ClassVar[bool] True Whether users can add extra sink pads
allow_dynamic_source_pads ClassVar[bool] True Whether users can add extra source pads

Pattern 1: Fixed Pads Only

When you want an element to have a fixed set of pads that users cannot modify:

from dataclasses import dataclass
from typing import ClassVar
from sgn.base import TransformElement, Frame

@dataclass
class AudioMixer(TransformElement):
    """A mixer that always has exactly two inputs and one output."""

    static_sink_pads: ClassVar[list[str]] = ["left", "right"]
    static_source_pads: ClassVar[list[str]] = ["mixed"]
    allow_dynamic_sink_pads: ClassVar[bool] = False
    allow_dynamic_source_pads: ClassVar[bool] = False

    def pull(self, pad, frame):
        # Store frames from left/right channels
        pass

    def new(self, pad):
        # Output mixed audio
        return Frame(data="mixed_audio")

# Create instance - no pad names needed!
mixer = AudioMixer(name="stereo_mixer")

# Pads are automatically created
print(mixer.snks.keys())  # dict_keys(['left', 'right'])
print(mixer.srcs.keys())  # dict_keys(['mixed'])

# Trying to add custom pads raises an error
# mixer = AudioMixer(sink_pad_names=["extra"])  # ValueError!

Pattern 2: Static Pads + User Pads

When you want required pads plus the flexibility for users to add more:

from dataclasses import dataclass
from typing import ClassVar
from sgn.base import SourceElement, Frame

@dataclass
class SensorSource(SourceElement):
    """A sensor that always has a monitor pad, but allows user-defined outputs."""

    static_source_pads: ClassVar[list[str]] = ["monitor"]
    # allow_dynamic_source_pads defaults to True

    def new(self, pad):
        if pad == self.srcs["monitor"]:
            return Frame(data={"status": "ok"})
        return Frame(data="sensor_reading")

# User can add extra pads - they're combined with static pads
sensor = SensorSource(source_pad_names=["temperature", "humidity"])

# All pads are available
print(sensor.srcs.keys())  # dict_keys(['temperature', 'humidity', 'monitor'])

Pattern 3: Dynamic Pads via Property

For advanced use cases, you can compute static pads based on instance attributes using a @property:

from dataclasses import dataclass
from sgn.base import TransformElement, Frame

@dataclass
class MultiBranchRouter(TransformElement):
    """A router with configurable branch outputs."""

    num_branches: int = 3

    @property
    def static_source_pads(self) -> list[str]:
        return [f"branch_{i}" for i in range(self.num_branches)]

    def pull(self, pad, frame):
        self.current_data = frame.data

    def new(self, pad):
        return Frame(data=self.current_data)

# Different instances can have different pads
router3 = MultiBranchRouter(sink_pad_names=["input"], num_branches=3)
router5 = MultiBranchRouter(sink_pad_names=["input"], num_branches=5)

print(router3.srcs.keys())  # dict_keys(['input', 'branch_0', 'branch_1', 'branch_2'])
print(router5.srcs.keys())  # dict_keys(['input', 'branch_0', ..., 'branch_4'])

When to Use Static Pads

  • Fixed pads only: Use when your element's interface is well-defined and shouldn't change (e.g., a stereo audio processor always needs left/right inputs)
  • Static + dynamic: Use when you have required pads (like a monitoring output) but want to allow customization
  • Property-based: Use when pad configuration depends on element parameters

Validation Rules

  • If allow_dynamic_*_pads=False, you must define the corresponding static_*_pads
  • Validation happens at class definition time, not runtime
  • Attempting to set allow_dynamic_source_pads=False without static_source_pads raises TypeError

Element Naming

Every element and pad has a unique name:

from sgn.base import SourceElement, Frame

class MySource(SourceElement):
    def __init__(self, **kwargs):
        super().__init__(source_pad_names=["output"], **kwargs)
        self.counter = 0

    def new(self, pad):
        self.counter += 1
        return Frame(data=self.counter)

source = MySource(name="my_counter")
# Element name: "my_counter"
# Pad full name: "my_counter:src:output"

# If no name provided, a UUID is generated:
source2 = MySource()
# Element name: "a3f4b2c1d5e6..." (UUID)

Naming Best Practices

  • Always provide meaningful names for debugging
  • Pad names are automatically prefixed with element name
  • Use srcs and snks dictionaries for cleaner code

Advanced: Internal Pads and Element Lifecycle

Elements have an internal execution flow:

  1. Sink Pads: Call pull() to receive data
  2. Internal Pad: Call internal() for processing
  3. Source Pads: Call new() to generate output

You can override the internal() method for custom logic:

from sgn.base import TransformElement, Frame

class StatefulTransform(TransformElement):
    def __init__(self, **kwargs):
        super().__init__(
            source_pad_names=["out"],
            sink_pad_names=["in"],
            **kwargs
        )
        self.buffer = []

    def pull(self, pad, frame):
        # Stage 1: Receive data
        self.buffer.append(frame.data)

    def internal(self):
        # Stage 2: Process between pull and new
        self.buffer = sorted(self.buffer)  # Sort accumulated data

    def new(self, pad):
        # Stage 3: Generate output
        return Frame(data=self.buffer.pop(0) if self.buffer else None)

Frame Data Flow

Frames flow through the pipeline via pad connections:

[SourceElement]
      |
  SourcePad.output  ──┐
                      │ (linked)
                      ├─> SinkPad.input
                      │       |
              [TransformElement]
                      |
                 SourcePad.output  ──┐
                                     │ (linked)
                                     ├─> SinkPad.input
                                     │       |
                              [SinkElement]

Important: Pad Linking

  • Sink pads must be linked to source pads before running
  • One sink pad can connect to only one source pad
  • Multiple sink pads can connect to the same source pad (fan-out)
  • Linking is done via SinkPad.link() or Pipeline.link()

API Reference

sgn.base

Base classes for building a graph of elements and pads.

InternalPad dataclass

Bases: UniqueID, PadLike


              flowchart TD
              sgn.base.InternalPad[InternalPad]
              sgn.base.UniqueID[UniqueID]
              sgn.base.PadLike[PadLike]

                              sgn.base.UniqueID --> sgn.base.InternalPad
                
                sgn.base.PadLike --> sgn.base.InternalPad
                


              click sgn.base.InternalPad href "" "sgn.base.InternalPad"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.base.PadLike href "" "sgn.base.PadLike"

A pad that sits inside an element and is called between sink and source pads. Internal pads are connected in the elements internal graph according to the below (data flows top to bottom)

snk1 ... snkN (if exist) \ ... // internal (always exists) // ... \ src1 ... srcM (if exist)

Parameters:

Name Type Description Default
element Element

Element, The Element instance associated with this pad

 required
call Callable

Callable, The function that will be called during graph execution for this pad

 required
name str

str, optional, The unique name for this object

 ''
Source code in sgn/base.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
@dataclass(eq=False, repr=False)
class InternalPad(UniqueID, PadLike):
    """A pad that sits inside an element and is called between sink and source pads.
    Internal pads are connected in the elements internal graph according to the below
    (data flows top to bottom)

    snk1   ...  snkN     (if exist)
      \\   ...   //
         internal      (always exists)
      //   ...   \\
     src1  ...  srcM     (if exist)

    Args:
        element:
            Element, The Element instance associated with this pad
        call:
            Callable, The function that will be called during graph execution for
            this pad
        name:
            str, optional, The unique name for this object
    """

    def __post_init__(self):
        PadLike.__post_init__(self)
        UniqueID.__post_init__(self)

    @property
    def pad_type(self) -> str:
        return "inl"

    async def __call__(self) -> None:
        """When called, an internal pad receives a Frame from the element that the pad
        belongs to."""
        self.call()

__call__() async

When called, an internal pad receives a Frame from the element that the pad belongs to.

Source code in sgn/base.py 
263
264
265
266
async def __call__(self) -> None:
    """When called, an internal pad receives a Frame from the element that the pad
    belongs to."""
    self.call()

SinkElement dataclass

Bases: ABC, ElementLike, Generic[FrameLike]


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

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



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

Sink element represents a terminal node in a pipeline, that typically writes data to disk, etc. Sink_pads must exist but not source_pads.

Parameters:

Name Type Description Default
name str

str, optional, The unique name for this object

 ''
sink_pad_names Sequence[str]

list, optional, Set the list of sink pad names. These need to be unique for an element but not for an application. The resulting full names will be made with ":sink:"

 list()
Source code in sgn/base.py 
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
@dataclass(kw_only=True)
class SinkElement(ABC, ElementLike, Generic[FrameLike]):
    """Sink element represents a terminal node in a pipeline, that typically writes data
    to disk, etc. Sink_pads must exist but not source_pads.

    Args:
        name:
            str, optional, The unique name for this object
        sink_pad_names:
            list, optional, Set the list of sink pad names. These need to be unique for
            an element but not for an application. The resulting full names will be
            made with "<self.name>:sink:<sink_pad_name>"
    """

    sink_pad_names: Sequence[str] = field(default_factory=list)

    def __init_subclass__(cls, **kwargs):
        """Validate pad configuration at class definition time."""
        super().__init_subclass__(**kwargs)

        # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
        has_static_sink = _has_static_pads(cls, "static_sink_pads")
        allow_dynamic_sink = getattr(cls, "allow_dynamic_sink_pads", True)

        if not allow_dynamic_sink and not has_static_sink:
            raise TypeError(
                f"Element '{cls.__name__}' has allow_dynamic_sink_pads=False but "
                f"does not define static_sink_pads. Elements must have at least one "
                f"way to provide pads."
            )

    def __post_init__(self):
        """Establish the sink pads and graph attributes."""
        super().__post_init__()

        # Check if user provided pads when allow_dynamic is False
        if not self.allow_dynamic_sink_pads and self.sink_pad_names:
            raise ValueError(
                f"Element '{self.name}' has allow_dynamic_sink_pads=False. "
                f"Cannot specify sink_pad_names."
            )

        # Add static pads to user-provided pads
        self.sink_pad_names = list(self.sink_pad_names) + list(self.static_sink_pads)

        self.sink_pads = [
            SinkPad(
                name=pad_name,
                element=self,
                call=self.pull,
            )
            for pad_name in self.sink_pad_names
        ]
        # short names for easier recall
        self.snks = {n: p for n, p in zip(self.sink_pad_names, self.sink_pads)}
        self.rsnks = {p: n for n, p in zip(self.sink_pad_names, self.sink_pads)}
        self._at_eos = {p: False for p in self.sink_pads}
        assert self.sink_pads, "SinkElement must specify sink pads"
        assert not self.source_pads, "SinkElement must not specify any source pads"
        self.sink_pad_names_full = [p.name for p in self.sink_pads]

        # Update graph to be (all sinks -> internal)
        self.graph.update({self.internal_pad: set(self.sink_pads)})

    @property
    def at_eos(self) -> bool:
        """If frames on any sink pads are End of Stream (EOS), then mark this whole
        element as EOS.

        Returns:
            bool, True if any sink pad is at EOS, False otherwise
        """
        # TODO generalize this to be able to choose any v. all EOS propagation
        return any(self._at_eos.values())

    def mark_eos(self, pad: SinkPad) -> None:
        """Marks a sink pad as receiving the End of Stream (EOS). The EOS marker signals
        that no more frames will be received on this pad.

        Args:
            pad:
                SinkPad, The sink pad that is receiving the
        """
        self._at_eos[pad] = True

    @abstractmethod
    def pull(self, pad: SinkPad, frame: FrameLike) -> None:
        """Pull for a SinkElement represents the action of associating a frame with a
        particular input source pad a frame. This function must be provided by the
        subclass, and is where any "final" behavior must occur, e.g. writing to disk,
        etc.

        Args:
            pad:
                SinkPad, The sink pad that is receiving the frame
            frame:
                Frame, The frame that is being received
        """
        ...

at_eos property

If frames on any sink pads are End of Stream (EOS), then mark this whole element as EOS.

Returns:

Type Description
bool

bool, True if any sink pad is at EOS, False otherwise

__init_subclass__(**kwargs)

Validate pad configuration at class definition time.

Source code in sgn/base.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
def __init_subclass__(cls, **kwargs):
    """Validate pad configuration at class definition time."""
    super().__init_subclass__(**kwargs)

    # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
    has_static_sink = _has_static_pads(cls, "static_sink_pads")
    allow_dynamic_sink = getattr(cls, "allow_dynamic_sink_pads", True)

    if not allow_dynamic_sink and not has_static_sink:
        raise TypeError(
            f"Element '{cls.__name__}' has allow_dynamic_sink_pads=False but "
            f"does not define static_sink_pads. Elements must have at least one "
            f"way to provide pads."
        )

__post_init__()

Establish the sink pads and graph attributes.

Source code in sgn/base.py 
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
def __post_init__(self):
    """Establish the sink pads and graph attributes."""
    super().__post_init__()

    # Check if user provided pads when allow_dynamic is False
    if not self.allow_dynamic_sink_pads and self.sink_pad_names:
        raise ValueError(
            f"Element '{self.name}' has allow_dynamic_sink_pads=False. "
            f"Cannot specify sink_pad_names."
        )

    # Add static pads to user-provided pads
    self.sink_pad_names = list(self.sink_pad_names) + list(self.static_sink_pads)

    self.sink_pads = [
        SinkPad(
            name=pad_name,
            element=self,
            call=self.pull,
        )
        for pad_name in self.sink_pad_names
    ]
    # short names for easier recall
    self.snks = {n: p for n, p in zip(self.sink_pad_names, self.sink_pads)}
    self.rsnks = {p: n for n, p in zip(self.sink_pad_names, self.sink_pads)}
    self._at_eos = {p: False for p in self.sink_pads}
    assert self.sink_pads, "SinkElement must specify sink pads"
    assert not self.source_pads, "SinkElement must not specify any source pads"
    self.sink_pad_names_full = [p.name for p in self.sink_pads]

    # Update graph to be (all sinks -> internal)
    self.graph.update({self.internal_pad: set(self.sink_pads)})

mark_eos(pad)

Marks a sink pad as receiving the End of Stream (EOS). The EOS marker signals that no more frames will be received on this pad.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, The sink pad that is receiving the

 required
Source code in sgn/base.py 
619
620
621
622
623
624
625
626
627
def mark_eos(self, pad: SinkPad) -> None:
    """Marks a sink pad as receiving the End of Stream (EOS). The EOS marker signals
    that no more frames will be received on this pad.

    Args:
        pad:
            SinkPad, The sink pad that is receiving the
    """
    self._at_eos[pad] = True

pull(pad, frame) abstractmethod

Pull for a SinkElement represents the action of associating a frame with a particular input source pad a frame. This function must be provided by the subclass, and is where any "final" behavior must occur, e.g. writing to disk, etc.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, The sink pad that is receiving the frame

 required
frame FrameLike

Frame, The frame that is being received

 required
Source code in sgn/base.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
@abstractmethod
def pull(self, pad: SinkPad, frame: FrameLike) -> None:
    """Pull for a SinkElement represents the action of associating a frame with a
    particular input source pad a frame. This function must be provided by the
    subclass, and is where any "final" behavior must occur, e.g. writing to disk,
    etc.

    Args:
        pad:
            SinkPad, The sink pad that is receiving the frame
        frame:
            Frame, The frame that is being received
    """
    ...

SinkPad dataclass

Bases: UniqueID, PadLike


              flowchart TD
              sgn.base.SinkPad[SinkPad]
              sgn.base.UniqueID[UniqueID]
              sgn.base.PadLike[PadLike]

                              sgn.base.UniqueID --> sgn.base.SinkPad
                
                sgn.base.PadLike --> sgn.base.SinkPad
                


              click sgn.base.SinkPad href "" "sgn.base.SinkPad"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.base.PadLike href "" "sgn.base.PadLike"

A pad that receives a data Frame when called. When linked, it returns a dictionary suitable for building a graph in graphlib.

Parameters:

Name Type Description Default
element Element

Element, The Element instance associated with this pad

 required
call Callable

Callable, The function that will be called during graph execution for this pad, takes two arguments, the pad and the frame

 required
name str

str, optional, The unique name for this object

 ''
other SourcePad | None

SourcePad, optional, This holds the source pad that is linked to this sink pad, default None

 None
input Frame | None

Frame, optional, This holds the Frame provided by the linked source pad. Generally it gets set when this SinkPad is called, default None

 None
data_spec DataSpec | None

DataSpec, optional, This holds a specification for the data stored in frames, and is expected to be consistent across frames passing through this pad. This is set when this sink pad is first called

 None
Source code in sgn/base.py 
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
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@dataclass(eq=False, repr=False)
class SinkPad(UniqueID, PadLike):
    """A pad that receives a data Frame when called.  When linked, it returns a
    dictionary suitable for building a graph in graphlib.

    Args:
        element:
            Element, The Element instance associated with this pad
        call:
            Callable, The function that will be called during graph execution for this
            pad, takes two arguments, the pad and the frame
        name:
            str, optional, The unique name for this object
        other:
            SourcePad, optional, This holds the source pad that is linked to this sink
            pad, default None
        input:
            Frame, optional, This holds the Frame provided by the linked source pad.
            Generally it gets set when this SinkPad is called, default None
        data_spec:
            DataSpec, optional, This holds a specification for the data stored
            in frames, and is expected to be consistent across frames passing
            through this pad. This is set when this sink pad is first called
    """

    other: SourcePad | None = None
    input: Frame | None = None
    data_spec: DataSpec | None = None

    def __post_init__(self):
        PadLike.__post_init__(self)
        UniqueID.__post_init__(self)

    @property
    def pad_type(self) -> str:
        return "snk"

    def link(self, other: SourcePad) -> dict[Pad, set[Pad]]:
        """Returns a dictionary of dependencies suitable for adding to a graphlib graph.

        Args:
            other:
                SourcePad, The source pad to link to this sink pad

        Notes:
            Many-to-one (source, sink) Not Supported:
                Only sink pads can be linked. A sink pad can be linked to only one
                source pad, but multiple sink pads may link to the same source pad.

        Returns:
            dict[SinkPad, set[SourcePad]], a dictionary of dependencies suitable for
            adding to a graphlib graph
        """
        assert isinstance(other, SourcePad), "other is not an instance of SourcePad"
        self.other = other
        self.is_linked = True
        other.is_linked = True
        return {self: {other}}

    async def __call__(self) -> None:
        """When called, a sink pad gets a Frame from the linked source pad and then
        calls the element's provided call function.

        Notes:
            Pad Call Order:
                pads must be called in the correct order such that the upstream sources
                have new information by the time call is invoked. This should be done
                within a directed acyclic graph such as those provided by the
                apps.Pipeline class.
        """
        assert isinstance(self.other, SourcePad), "Sink pad has not been linked"
        self.input = self.other.output
        assert isinstance(self.input, Frame)
        if self.data_spec is None:
            self.data_spec = self.input.spec
        if not self.data_spec == self.input.spec:
            msg = (
                f"frame received by {self.name} is inconsistent with "
                "previously received frames. previous data specification: "
                f"{self.data_spec}, current data specification: {self.input.spec}"
            )
            raise ValueError(msg)
        self.call(self, self.input)
        if self.element is not None:
            self.element.logger.info("\t%s:%s", self, self.input)

__call__() async

When called, a sink pad gets a Frame from the linked source pad and then calls the element's provided call function.

Notes

Pad Call Order: pads must be called in the correct order such that the upstream sources have new information by the time call is invoked. This should be done within a directed acyclic graph such as those provided by the apps.Pipeline class.

Source code in sgn/base.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
async def __call__(self) -> None:
    """When called, a sink pad gets a Frame from the linked source pad and then
    calls the element's provided call function.

    Notes:
        Pad Call Order:
            pads must be called in the correct order such that the upstream sources
            have new information by the time call is invoked. This should be done
            within a directed acyclic graph such as those provided by the
            apps.Pipeline class.
    """
    assert isinstance(self.other, SourcePad), "Sink pad has not been linked"
    self.input = self.other.output
    assert isinstance(self.input, Frame)
    if self.data_spec is None:
        self.data_spec = self.input.spec
    if not self.data_spec == self.input.spec:
        msg = (
            f"frame received by {self.name} is inconsistent with "
            "previously received frames. previous data specification: "
            f"{self.data_spec}, current data specification: {self.input.spec}"
        )
        raise ValueError(msg)
    self.call(self, self.input)
    if self.element is not None:
        self.element.logger.info("\t%s:%s", self, self.input)

Returns a dictionary of dependencies suitable for adding to a graphlib graph.

Parameters:

Name Type Description Default
other SourcePad

SourcePad, The source pad to link to this sink pad

 required
Notes

Many-to-one (source, sink) Not Supported: Only sink pads can be linked. A sink pad can be linked to only one source pad, but multiple sink pads may link to the same source pad.

Returns:

Type Description
dict[Pad, set[Pad]]

dict[SinkPad, set[SourcePad]], a dictionary of dependencies suitable for
dict[Pad, set[Pad]]

adding to a graphlib graph
Source code in sgn/base.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def link(self, other: SourcePad) -> dict[Pad, set[Pad]]:
    """Returns a dictionary of dependencies suitable for adding to a graphlib graph.

    Args:
        other:
            SourcePad, The source pad to link to this sink pad

    Notes:
        Many-to-one (source, sink) Not Supported:
            Only sink pads can be linked. A sink pad can be linked to only one
            source pad, but multiple sink pads may link to the same source pad.

    Returns:
        dict[SinkPad, set[SourcePad]], a dictionary of dependencies suitable for
        adding to a graphlib graph
    """
    assert isinstance(other, SourcePad), "other is not an instance of SourcePad"
    self.other = other
    self.is_linked = True
    other.is_linked = True
    return {self: {other}}

SourceElement dataclass

Bases: ABC, ElementLike


              flowchart TD
              sgn.base.SourceElement[SourceElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]

                              sgn.base.ElementLike --> sgn.base.SourceElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                



              click sgn.base.SourceElement href "" "sgn.base.SourceElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"

Initialize with a list of source pads. Every source pad is added to the graph with no dependencies.

Parameters:

Name Type Description Default
name str

str, optional, The unique name for this object

 ''
source_pad_names Sequence[str]

list, optional, Set the list of source pad names. These need to be unique for an element but not for an application. The resulting full names will be made with ":src:"

 list()
Source code in sgn/base.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
@dataclass(repr=False, kw_only=True)
class SourceElement(ABC, ElementLike):
    """Initialize with a list of source pads. Every source pad is added to the graph
    with no dependencies.

    Args:
        name:
            str, optional, The unique name for this object
        source_pad_names:
            list, optional, Set the list of source pad names. These need to be unique
            for an element but not for an application. The resulting full names will be
            made with "<self.name>:src:<source_pad_name>"
    """

    source_pad_names: Sequence[str] = field(default_factory=list)

    def __init_subclass__(cls, **kwargs):
        """Validate pad configuration at class definition time."""
        super().__init_subclass__(**kwargs)

        # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
        has_static_source = _has_static_pads(cls, "static_source_pads")
        allow_dynamic_source = getattr(cls, "allow_dynamic_source_pads", True)

        if not allow_dynamic_source and not has_static_source:
            raise TypeError(
                f"Element '{cls.__name__}' has allow_dynamic_source_pads=False but "
                f"does not define static_source_pads. Elements must have at least one "
                f"way to provide pads."
            )

    def __post_init__(self):
        """Establish the source pads and graph attributes."""
        super().__post_init__()

        # Check if user provided pads when allow_dynamic is False
        if not self.allow_dynamic_source_pads and self.source_pad_names:
            raise ValueError(
                f"Element '{self.name}' has allow_dynamic_source_pads=False. "
                f"Cannot specify source_pad_names."
            )

        # Add static pads to user-provided pads
        self.source_pad_names = list(self.source_pad_names) + list(
            self.static_source_pads
        )

        self.source_pads = [
            SourcePad(
                name=pad_name,
                element=self,
                call=self.new,
            )
            for pad_name in self.source_pad_names
        ]
        # short names for easier recall
        self.srcs = {n: p for n, p in zip(self.source_pad_names, self.source_pads)}
        self.rsrcs = {p: n for n, p in zip(self.source_pad_names, self.source_pads)}
        assert self.source_pads, "SourceElement must specify source pads"
        assert not self.sink_pads, "SourceElement must not specify sink pads"
        self.graph.update({s: {self.internal_pad} for s in self.source_pads})

    @abstractmethod
    def new(self, pad: SourcePad) -> Frame:
        """New frames are created on "pad". Must be provided by subclass.

        Args:
            pad:
                SourcePad, The source pad through which the frame is passed

        Returns:
            Frame, The new frame to be passed through the source pad
        """
        ...

__init_subclass__(**kwargs)

Validate pad configuration at class definition time.

Source code in sgn/base.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def __init_subclass__(cls, **kwargs):
    """Validate pad configuration at class definition time."""
    super().__init_subclass__(**kwargs)

    # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
    has_static_source = _has_static_pads(cls, "static_source_pads")
    allow_dynamic_source = getattr(cls, "allow_dynamic_source_pads", True)

    if not allow_dynamic_source and not has_static_source:
        raise TypeError(
            f"Element '{cls.__name__}' has allow_dynamic_source_pads=False but "
            f"does not define static_source_pads. Elements must have at least one "
            f"way to provide pads."
        )

__post_init__()

Establish the source pads and graph attributes.

Source code in sgn/base.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
def __post_init__(self):
    """Establish the source pads and graph attributes."""
    super().__post_init__()

    # Check if user provided pads when allow_dynamic is False
    if not self.allow_dynamic_source_pads and self.source_pad_names:
        raise ValueError(
            f"Element '{self.name}' has allow_dynamic_source_pads=False. "
            f"Cannot specify source_pad_names."
        )

    # Add static pads to user-provided pads
    self.source_pad_names = list(self.source_pad_names) + list(
        self.static_source_pads
    )

    self.source_pads = [
        SourcePad(
            name=pad_name,
            element=self,
            call=self.new,
        )
        for pad_name in self.source_pad_names
    ]
    # short names for easier recall
    self.srcs = {n: p for n, p in zip(self.source_pad_names, self.source_pads)}
    self.rsrcs = {p: n for n, p in zip(self.source_pad_names, self.source_pads)}
    assert self.source_pads, "SourceElement must specify source pads"
    assert not self.sink_pads, "SourceElement must not specify sink pads"
    self.graph.update({s: {self.internal_pad} for s in self.source_pads})

new(pad) abstractmethod

New frames are created on "pad". Must be provided by subclass.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, The source pad through which the frame is passed

 required

Returns:

Type Description
Frame

Frame, The new frame to be passed through the source pad
Source code in sgn/base.py 
403
404
405
406
407
408
409
410
411
412
413
414
@abstractmethod
def new(self, pad: SourcePad) -> Frame:
    """New frames are created on "pad". Must be provided by subclass.

    Args:
        pad:
            SourcePad, The source pad through which the frame is passed

    Returns:
        Frame, The new frame to be passed through the source pad
    """
    ...

SourcePad dataclass

Bases: UniqueID, PadLike


              flowchart TD
              sgn.base.SourcePad[SourcePad]
              sgn.base.UniqueID[UniqueID]
              sgn.base.PadLike[PadLike]

                              sgn.base.UniqueID --> sgn.base.SourcePad
                
                sgn.base.PadLike --> sgn.base.SourcePad
                


              click sgn.base.SourcePad href "" "sgn.base.SourcePad"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.base.PadLike href "" "sgn.base.PadLike"

A pad that provides a data Frame when called.

Parameters:

Name Type Description Default
element Element

Element, The Element instance associated with this pad

 required
call Callable

Callable, The function that will be called during graph execution for this pad

 required
name str

str, optional, The unique name for this object

 ''
output Frame | None

Frame, optional, This attribute is set to be the output Frame when the pad is called.

 None
Source code in sgn/base.py 
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
@dataclass(eq=False, repr=False)
class SourcePad(UniqueID, PadLike):
    """A pad that provides a data Frame when called.

    Args:
        element:
            Element, The Element instance associated with this pad
        call:
            Callable, The function that will be called during graph execution for
            this pad
        name:
            str, optional, The unique name for this object
        output:
            Frame, optional, This attribute is set to be the output Frame when the pad
            is called.
    """

    output: Frame | None = None

    def __post_init__(self):
        PadLike.__post_init__(self)
        UniqueID.__post_init__(self)

    @property
    def pad_type(self) -> str:
        return "src"

    async def __call__(self) -> None:
        """When called, a source pad receives a Frame from the element that the pad
        belongs to."""
        self.output = self.call(pad=self)
        assert isinstance(self.output, Frame)
        if self.element is not None:
            self.element.logger.info("\t%s : %s", self, self.output)

__call__() async

When called, a source pad receives a Frame from the element that the pad belongs to.

Source code in sgn/base.py 
137
138
139
140
141
142
143
async def __call__(self) -> None:
    """When called, a source pad receives a Frame from the element that the pad
    belongs to."""
    self.output = self.call(pad=self)
    assert isinstance(self.output, Frame)
    if self.element is not None:
        self.element.logger.info("\t%s : %s", self, self.output)

TransformElement dataclass

Bases: ABC, ElementLike, Generic[FrameLike]


              flowchart TD
              sgn.base.TransformElement[TransformElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]

                              sgn.base.ElementLike --> sgn.base.TransformElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                



              click sgn.base.TransformElement href "" "sgn.base.TransformElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"

Both "source_pads" and "sink_pads" must exist. The execution scheduling flow of the logic within a TransformElement is as follows: 1.) all sink pads, 2.) the internal pad, 3.) all source pads. The execution of all downstream logic will be blocked until logic in all upstream pads within the same TransformElement has exited.

Parameters:

Name Type Description Default
name str

str, optional, The unique name for this object

 ''
source_pad_names Sequence[str]

list, optional, Set the list of source pad names. These need to be unique for an element but not for an application. The resulting full names will be made with ":src:"

 list()
sink_pad_names Sequence[str]

list, optional, Set the list of sink pad names. These need to be unique for an element but not for an application. The resulting full names will be made with ":snk:"

 list()
Source code in sgn/base.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
@dataclass(repr=False, kw_only=True)
class TransformElement(ABC, ElementLike, Generic[FrameLike]):
    """Both "source_pads" and "sink_pads" must exist. The execution scheduling
    flow of the logic within a TransformElement is as follows: 1.) all sink
    pads, 2.) the internal pad, 3.) all source pads. The execution of all
    downstream logic will be blocked until logic in all upstream pads within
    the same TransformElement has exited.

    Args:
        name:
            str, optional, The unique name for this object
        source_pad_names:
            list, optional, Set the list of source pad names. These need to be unique
            for an element but not for an application. The resulting full names will
            be made with "<self.name>:src:<source_pad_name>"
        sink_pad_names:
            list, optional, Set the list of sink pad names. These need to be unique
            for an element but not for an application. The resulting full names will
            be made with "<self.name>:snk:<sink_pad_name>"
    """

    source_pad_names: Sequence[str] = field(default_factory=list)
    sink_pad_names: Sequence[str] = field(default_factory=list)

    def __init_subclass__(cls, **kwargs):
        """Validate pad configuration at class definition time."""
        super().__init_subclass__(**kwargs)

        # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
        has_static_sink = _has_static_pads(cls, "static_sink_pads")
        allow_dynamic_sink = getattr(cls, "allow_dynamic_sink_pads", True)
        has_static_source = _has_static_pads(cls, "static_source_pads")
        allow_dynamic_source = getattr(cls, "allow_dynamic_source_pads", True)

        if not allow_dynamic_sink and not has_static_sink:
            raise TypeError(
                f"Element '{cls.__name__}' has allow_dynamic_sink_pads=False but "
                f"does not define static_sink_pads. Elements must have at least one "
                f"way to provide pads."
            )
        if not allow_dynamic_source and not has_static_source:
            raise TypeError(
                f"Element '{cls.__name__}' has allow_dynamic_source_pads=False but "
                f"does not define static_source_pads. Elements must have at least one "
                f"way to provide pads."
            )

    def __post_init__(self):
        """Establish the source pads and sink pads and graph attributes."""
        super().__post_init__()

        # Check if user provided pads when allow_dynamic is False
        if not self.allow_dynamic_sink_pads and self.sink_pad_names:
            raise ValueError(
                f"Element '{self.name}' has allow_dynamic_sink_pads=False. "
                f"Cannot specify sink_pad_names."
            )
        if not self.allow_dynamic_source_pads and self.source_pad_names:
            raise ValueError(
                f"Element '{self.name}' has allow_dynamic_source_pads=False. "
                f"Cannot specify source_pad_names."
            )

        # Add static pads to user-provided pads
        self.sink_pad_names = list(self.sink_pad_names) + list(self.static_sink_pads)
        self.source_pad_names = list(self.source_pad_names) + list(
            self.static_source_pads
        )

        self.source_pads = [
            SourcePad(
                name=pad_name,
                element=self,
                call=self.new,
            )
            for pad_name in self.source_pad_names
        ]
        self.sink_pads = [
            SinkPad(
                name=pad_name,
                element=self,
                call=self.pull,
            )
            for pad_name in self.sink_pad_names
        ]
        # short names for easier recall
        self.srcs = {n: p for n, p in zip(self.source_pad_names, self.source_pads)}
        self.snks = {n: p for n, p in zip(self.sink_pad_names, self.sink_pads)}
        self.rsrcs = {p: n for n, p in zip(self.source_pad_names, self.source_pads)}
        self.rsnks = {p: n for n, p in zip(self.sink_pad_names, self.sink_pads)}
        assert (
            self.source_pads and self.sink_pads
        ), "TransformElement must specify both sink and source pads"

        # Make maximal bipartite graph in two pieces
        # First, (all sinks -> internal)
        self.graph.update({self.internal_pad: set(self.sink_pads)})
        # Second, (internal -> all sources)
        self.graph.update({s: {self.internal_pad} for s in self.source_pads})

    @abstractmethod
    def pull(self, pad: SinkPad, frame: FrameLike) -> None:
        """Pull data from the input pads (source pads of upstream elements), must be
        implemented by subclasses.

        Args:
            pad:
                SinkPad, The sink pad that is receiving the frame
            frame:
                Frame, The frame that is pulled from the source pad
        """
        ...

    @abstractmethod
    def new(self, pad: SourcePad) -> FrameLike:
        """New frames are created on "pad". Must be provided by subclass.

        Args:
            pad:
                SourcePad, The source pad through which the frame is passed

        Returns:
            Frame, The new frame to be passed through the source pad
        """
        ...

__init_subclass__(**kwargs)

Validate pad configuration at class definition time.

Source code in sgn/base.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
def __init_subclass__(cls, **kwargs):
    """Validate pad configuration at class definition time."""
    super().__init_subclass__(**kwargs)

    # Check: allow_dynamic_*_pads=False requires static_*_pads to be set
    has_static_sink = _has_static_pads(cls, "static_sink_pads")
    allow_dynamic_sink = getattr(cls, "allow_dynamic_sink_pads", True)
    has_static_source = _has_static_pads(cls, "static_source_pads")
    allow_dynamic_source = getattr(cls, "allow_dynamic_source_pads", True)

    if not allow_dynamic_sink and not has_static_sink:
        raise TypeError(
            f"Element '{cls.__name__}' has allow_dynamic_sink_pads=False but "
            f"does not define static_sink_pads. Elements must have at least one "
            f"way to provide pads."
        )
    if not allow_dynamic_source and not has_static_source:
        raise TypeError(
            f"Element '{cls.__name__}' has allow_dynamic_source_pads=False but "
            f"does not define static_source_pads. Elements must have at least one "
            f"way to provide pads."
        )

__post_init__()

Establish the source pads and sink pads and graph attributes.

Source code in sgn/base.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def __post_init__(self):
    """Establish the source pads and sink pads and graph attributes."""
    super().__post_init__()

    # Check if user provided pads when allow_dynamic is False
    if not self.allow_dynamic_sink_pads and self.sink_pad_names:
        raise ValueError(
            f"Element '{self.name}' has allow_dynamic_sink_pads=False. "
            f"Cannot specify sink_pad_names."
        )
    if not self.allow_dynamic_source_pads and self.source_pad_names:
        raise ValueError(
            f"Element '{self.name}' has allow_dynamic_source_pads=False. "
            f"Cannot specify source_pad_names."
        )

    # Add static pads to user-provided pads
    self.sink_pad_names = list(self.sink_pad_names) + list(self.static_sink_pads)
    self.source_pad_names = list(self.source_pad_names) + list(
        self.static_source_pads
    )

    self.source_pads = [
        SourcePad(
            name=pad_name,
            element=self,
            call=self.new,
        )
        for pad_name in self.source_pad_names
    ]
    self.sink_pads = [
        SinkPad(
            name=pad_name,
            element=self,
            call=self.pull,
        )
        for pad_name in self.sink_pad_names
    ]
    # short names for easier recall
    self.srcs = {n: p for n, p in zip(self.source_pad_names, self.source_pads)}
    self.snks = {n: p for n, p in zip(self.sink_pad_names, self.sink_pads)}
    self.rsrcs = {p: n for n, p in zip(self.source_pad_names, self.source_pads)}
    self.rsnks = {p: n for n, p in zip(self.sink_pad_names, self.sink_pads)}
    assert (
        self.source_pads and self.sink_pads
    ), "TransformElement must specify both sink and source pads"

    # Make maximal bipartite graph in two pieces
    # First, (all sinks -> internal)
    self.graph.update({self.internal_pad: set(self.sink_pads)})
    # Second, (internal -> all sources)
    self.graph.update({s: {self.internal_pad} for s in self.source_pads})

new(pad) abstractmethod

New frames are created on "pad". Must be provided by subclass.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, The source pad through which the frame is passed

 required

Returns:

Type Description
FrameLike

Frame, The new frame to be passed through the source pad
Source code in sgn/base.py 
530
531
532
533
534
535
536
537
538
539
540
541
@abstractmethod
def new(self, pad: SourcePad) -> FrameLike:
    """New frames are created on "pad". Must be provided by subclass.

    Args:
        pad:
            SourcePad, The source pad through which the frame is passed

    Returns:
        Frame, The new frame to be passed through the source pad
    """
    ...

pull(pad, frame) abstractmethod

Pull data from the input pads (source pads of upstream elements), must be implemented by subclasses.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, The sink pad that is receiving the frame

 required
frame FrameLike

Frame, The frame that is pulled from the source pad

 required
Source code in sgn/base.py 
517
518
519
520
521
522
523
524
525
526
527
528
@abstractmethod
def pull(self, pad: SinkPad, frame: FrameLike) -> None:
    """Pull data from the input pads (source pads of upstream elements), must be
    implemented by subclasses.

    Args:
        pad:
            SinkPad, The sink pad that is receiving the frame
        frame:
            Frame, The frame that is pulled from the source pad
    """
    ...

UniqueID dataclass

Generic class from which all classes that participate in an execution graph should be derived. Enforces a unique name and hashes based on that name.

Parameters:

Name Type Description Default
name str

str, optional, The unique name for this object, defaults to the objects unique uuid4 hex string if not specified

 ''
Source code in sgn/base.py 
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
@dataclass
class UniqueID:
    """Generic class from which all classes that participate in an execution graph
    should be derived. Enforces a unique name and hashes based on that name.

    Args:
        name:
            str, optional, The unique name for this object, defaults to the objects
            unique uuid4 hex string if not specified
    """

    name: str = ""
    _id: str = field(init=False)

    def __post_init__(self):
        """Handle setup of the UniqueID class, including the `._id` attribute."""
        # give every element a truly unique identifier
        self._id = uuid.uuid4().hex
        if not self.name:
            self.name = self._id

    def __hash__(self) -> int:
        """Compute the hash of the object based on the unique id.

        Notes:
            Motivation:
                we need the Base class to be hashable, so that it can be
                used as a key in a dictionary, but mutable dataclasses are not
                hashable by default, so we have to define our own hash function
                here.
            Stability:
                As currently implemented, the hash of a UniqueID object will not be
                stable across python sessions, and should therefore not be used for
                checksum purposes.

        Returns:
            int, hash of the object
        """
        return hash(self._id)

    def __eq__(self, other) -> bool:
        """Check if two objects are equal based on their unique id and types."""
        return hash(self) == hash(other)

__eq__(other)

Check if two objects are equal based on their unique id and types.

Source code in sgn/base.py 
59
60
61
def __eq__(self, other) -> bool:
    """Check if two objects are equal based on their unique id and types."""
    return hash(self) == hash(other)

__hash__()

Compute the hash of the object based on the unique id.

Notes

Motivation: we need the Base class to be hashable, so that it can be used as a key in a dictionary, but mutable dataclasses are not hashable by default, so we have to define our own hash function here. Stability: As currently implemented, the hash of a UniqueID object will not be stable across python sessions, and should therefore not be used for checksum purposes.

Returns:

Type Description
int

int, hash of the object
Source code in sgn/base.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __hash__(self) -> int:
    """Compute the hash of the object based on the unique id.

    Notes:
        Motivation:
            we need the Base class to be hashable, so that it can be
            used as a key in a dictionary, but mutable dataclasses are not
            hashable by default, so we have to define our own hash function
            here.
        Stability:
            As currently implemented, the hash of a UniqueID object will not be
            stable across python sessions, and should therefore not be used for
            checksum purposes.

    Returns:
        int, hash of the object
    """
    return hash(self._id)

__post_init__()

Handle setup of the UniqueID class, including the ._id attribute.

Source code in sgn/base.py 
33
34
35
36
37
38
def __post_init__(self):
    """Handle setup of the UniqueID class, including the `._id` attribute."""
    # give every element a truly unique identifier
    self._id = uuid.uuid4().hex
    if not self.name:
        self.name = self._id

_has_static_pads(cls, attr_name)

Check if a class defines static pads.

Static pads are considered defined if: 1. The attribute is overridden as a property (assume non-empty), OR 2. The attribute is overridden as a non-empty class variable

Parameters:

Name Type Description Default
cls type

The class to check

 required
attr_name str

The attribute name to check (e.g., 'static_sink_pads')

 required

Returns:

Type Description
bool

True if static pads are defined, False otherwise
Source code in sgn/base.py 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
def _has_static_pads(cls: type, attr_name: str) -> bool:
    """Check if a class defines static pads.

    Static pads are considered defined if:
    1. The attribute is overridden as a property (assume non-empty), OR
    2. The attribute is overridden as a non-empty class variable

    Args:
        cls: The class to check
        attr_name: The attribute name to check (e.g., 'static_sink_pads')

    Returns:
        True if static pads are defined, False otherwise
    """
    if attr_name not in cls.__dict__:
        # Not overridden in this class, using base class default (empty)
        return False

    value = cls.__dict__[attr_name]

    if isinstance(value, property):
        # It's a property - assume it will return non-empty
        return True
    else:
        # It's a class variable - check if non-empty
        return bool(value)