Skip to content

sgn.frames

Frame classes for the SGN framework.

DataSpec dataclass

A specification for the type of data stored in frames.

All properties in this specification are expected to match what is stored in the frame, and what is being transferred between source and sink pads. Matching is the responsibility of the connecting pads (see base.py); this class does not enforce it.

Subclasses must also be declared @dataclass(frozen=True) so that :func:dataclasses.replace works consistently and instances remain hashable.

Source code in src/sgn/frames.py
@dataclass(frozen=True, eq=True)
class DataSpec:
    """A specification for the type of data stored in frames.

    All properties in this specification are expected to match what is stored in
    the frame, and what is being transferred between source and sink pads.
    Matching is the responsibility of the connecting pads (see ``base.py``);
    this class does not enforce it.

    Subclasses must also be declared ``@dataclass(frozen=True)`` so that
    :func:`dataclasses.replace` works consistently and instances remain
    hashable.

    """

    def update(self, **kwargs) -> Self:
        """Return a new spec of the same concrete type with fields replaced."""
        return replace(self, **kwargs)

update(**kwargs)

Return a new spec of the same concrete type with fields replaced.

Source code in src/sgn/frames.py
def update(self, **kwargs) -> Self:
    """Return a new spec of the same concrete type with fields replaced."""
    return replace(self, **kwargs)

Frame dataclass

Generic class to hold the basic unit of data that flows through a graph.

Subclasses may override :meth:__post_init__ and should call super().__post_init__() to remain forward-compatible if Frame ever grows its own initialization logic.

Parameters:

Name Type Description Default
EOS bool

bool, default False, Whether this frame indicates end of stream (EOS)

False
is_gap bool

bool, default False, Whether this frame is marked as a gap

False
spec DataSpec

DataSpec, optional, a specification for the data captured in this frame

DataSpec()
data Any

Any, the data to store in the frame

None
metadata dict[str, Any]

dict, optional, Metadata associated with this frame.

dict()
Source code in src/sgn/frames.py
@dataclass
class Frame:
    """Generic class to hold the basic unit of data that flows through a graph.

    Subclasses may override :meth:`__post_init__` and should call
    ``super().__post_init__()`` to remain forward-compatible if Frame ever
    grows its own initialization logic.

    Args:
        EOS:
            bool, default False, Whether this frame indicates end of stream (EOS)
        is_gap:
            bool, default False, Whether this frame is marked as a gap
        spec:
            DataSpec, optional, a specification for the data captured in this frame
        data:
            Any, the data to store in the frame
        metadata:
            dict, optional, Metadata associated with this frame.
    """

    EOS: bool = False
    is_gap: bool = False
    spec: DataSpec = field(default_factory=DataSpec)
    data: Any = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self):
        # Intentional no-op: extension point so downstream subclasses can
        # safely call ``super().__post_init__()`` and chain through Frame.
        pass

IterFrame dataclass

Bases: Frame


              flowchart TD
              sgn.frames.IterFrame[IterFrame]
              sgn.frames.Frame[Frame]

                              sgn.frames.Frame --> sgn.frames.IterFrame
                


              click sgn.frames.IterFrame href "" "sgn.frames.IterFrame"
              click sgn.frames.Frame href "" "sgn.frames.Frame"
            

A frame whose data attribute is an iterable.

Parameters:

Name Type Description Default
data Iterable[Any]

Iterable, the data to store in the frame

list()
Source code in src/sgn/frames.py
@dataclass
class IterFrame(Frame):
    """A frame whose data attribute is an iterable.

    Args:
        data:
            Iterable, the data to store in the frame
    """

    data: Iterable[Any] = field(default_factory=list)