Skip to content

sgn.groups

Element grouping and pad selection operations.

This module provides functionality for grouping elements and selecting specific pads for use in pipeline operations. It contains the core abstractions that allow for flexible and intuitive pipeline construction.

ElementGroup dataclass

Bases: PadIteratorMixin


              flowchart TD
              sgn.groups.ElementGroup[ElementGroup]
              sgn.groups.PadIteratorMixin[PadIteratorMixin]

                              sgn.groups.PadIteratorMixin --> sgn.groups.ElementGroup
                


              click sgn.groups.ElementGroup href "" "sgn.groups.ElementGroup"
              click sgn.groups.PadIteratorMixin href "" "sgn.groups.PadIteratorMixin"

A unified group for elements and pad selections.

This class holds a collection of elements and pad selections without determining upfront whether to extract source or sink pads. The actual pad extraction is deferred until the group is used in pipeline.connect(), where the context (source vs sink position) determines which pads to extract.

Source code in src/sgn/groups.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
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
@dataclass
class ElementGroup(PadIteratorMixin):
    """A unified group for elements and pad selections.

    This class holds a collection of elements and pad selections without
    determining upfront whether to extract source or sink pads. The actual
    pad extraction is deferred until the group is used in pipeline.connect(),
    where the context (source vs sink position) determines which pads to extract.
    """

    items: list[Element | PadSelection] = field(default_factory=list)

    def select(self, *pad_names: str) -> ElementGroup:
        """Select pads from items in the group by pad name."""
        selected_items: list[Element | PadSelection] = []
        pad_names_set = set(pad_names)

        for item in self.items:
            if isinstance(item, PadSelection):
                # For existing pad selections, intersect with requested pad names
                element = item.element
                available_pads = item.pad_names
            else:
                # For elements, get all their pad names
                element = item
                available_pads = set()
                if isinstance(element, (SourceElement, TransformElement)):
                    available_pads.update(element.srcs.keys())
                if isinstance(element, (TransformElement, SinkElement)):
                    available_pads.update(element.snks.keys())

            # Find intersection of available pads with requested pad names
            matching_pads = available_pads & pad_names_set
            if matching_pads:
                selected_items.append(
                    PadSelection(element=element, pad_names=matching_pads)
                )

        return ElementGroup(items=selected_items)

    @property
    def elements(self) -> list[Element]:
        """Get all unique elements referenced by this group."""
        seen = set()
        unique_elements = []
        for item in self.items:
            if isinstance(item, PadSelection):
                element = item.element
            else:
                element = item
            element_id = id(element)
            if element_id not in seen:
                seen.add(element_id)
                unique_elements.append(element)
        return unique_elements

    @property
    def srcs(self) -> dict[str, SourcePad]:
        """Extract source pads from all items in the group using names as keys."""
        combined_pads = {}

        for item in self.items:
            if isinstance(item, SinkElement):
                msg = f"Element '{item.name}' is a SinkElement and has no source pads"
                raise ValueError(msg)

            for pad_name, pad in item.srcs.items():
                if pad_name in combined_pads:
                    msg = f"Duplicate pad name '{pad_name}' in group"
                    raise KeyError(msg)
                combined_pads[pad_name] = pad

        return combined_pads

    @property
    def snks(self) -> dict[str, SinkPad]:
        """Extract sink pads from all items in the group using names as keys."""
        combined_pads = {}

        for item in self.items:
            if isinstance(item, SourceElement):
                msg = f"Element '{item.name}' is a SourceElement and has no sink pads"
                raise ValueError(msg)

            for pad_name, pad in item.snks.items():
                if pad_name in combined_pads:
                    msg = f"Duplicate pad name '{pad_name}' in group"
                    raise KeyError(msg)
                combined_pads[pad_name] = pad

        return combined_pads

elements property

Get all unique elements referenced by this group.

snks property

Extract sink pads from all items in the group using names as keys.

srcs property

Extract source pads from all items in the group using names as keys.

select(*pad_names)

Select pads from items in the group by pad name.

Source code in src/sgn/groups.py 
145
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
def select(self, *pad_names: str) -> ElementGroup:
    """Select pads from items in the group by pad name."""
    selected_items: list[Element | PadSelection] = []
    pad_names_set = set(pad_names)

    for item in self.items:
        if isinstance(item, PadSelection):
            # For existing pad selections, intersect with requested pad names
            element = item.element
            available_pads = item.pad_names
        else:
            # For elements, get all their pad names
            element = item
            available_pads = set()
            if isinstance(element, (SourceElement, TransformElement)):
                available_pads.update(element.srcs.keys())
            if isinstance(element, (TransformElement, SinkElement)):
                available_pads.update(element.snks.keys())

        # Find intersection of available pads with requested pad names
        matching_pads = available_pads & pad_names_set
        if matching_pads:
            selected_items.append(
                PadSelection(element=element, pad_names=matching_pads)
            )

    return ElementGroup(items=selected_items)

PadIteratorMixin

Mixin class that provides iteration methods for pad selections.

Source code in src/sgn/groups.py 
42
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
class PadIteratorMixin:
    """Mixin class that provides iteration methods for pad selections."""

    def select_by_source(self: PadProvider) -> Iterator[tuple[str, PadSelection]]:
        """Iterate over source pads, yielding (pad_name, single_pad_selection) tuples.

        Each yielded PadSelection contains only a single source pad.
        Raises ValueError if there are no source pads.
        """
        srcs = self.srcs
        if not srcs:
            msg = "No source pads available in this selection"
            raise ValueError(msg)

        for pad_name, pad in srcs.items():
            yield pad_name, PadSelection(element=pad.element, pad_names={pad_name})

    def select_by_sink(self: PadProvider) -> Iterator[tuple[str, PadSelection]]:
        """Iterate over sink pads, yielding (pad_name, single_pad_selection) tuples.

        Each yielded PadSelection contains only a single sink pad.
        Raises ValueError if there are no sink pads.
        """
        snks = self.snks
        if not snks:
            msg = "No sink pads available in this selection"
            raise ValueError(msg)

        for pad_name, pad in snks.items():
            yield pad_name, PadSelection(element=pad.element, pad_names={pad_name})

select_by_sink()

Iterate over sink pads, yielding (pad_name, single_pad_selection) tuples.

Each yielded PadSelection contains only a single sink pad. Raises ValueError if there are no sink pads.

Source code in src/sgn/groups.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
def select_by_sink(self: PadProvider) -> Iterator[tuple[str, PadSelection]]:
    """Iterate over sink pads, yielding (pad_name, single_pad_selection) tuples.

    Each yielded PadSelection contains only a single sink pad.
    Raises ValueError if there are no sink pads.
    """
    snks = self.snks
    if not snks:
        msg = "No sink pads available in this selection"
        raise ValueError(msg)

    for pad_name, pad in snks.items():
        yield pad_name, PadSelection(element=pad.element, pad_names={pad_name})

select_by_source()

Iterate over source pads, yielding (pad_name, single_pad_selection) tuples.

Each yielded PadSelection contains only a single source pad. Raises ValueError if there are no source pads.

Source code in src/sgn/groups.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
def select_by_source(self: PadProvider) -> Iterator[tuple[str, PadSelection]]:
    """Iterate over source pads, yielding (pad_name, single_pad_selection) tuples.

    Each yielded PadSelection contains only a single source pad.
    Raises ValueError if there are no source pads.
    """
    srcs = self.srcs
    if not srcs:
        msg = "No source pads available in this selection"
        raise ValueError(msg)

    for pad_name, pad in srcs.items():
        yield pad_name, PadSelection(element=pad.element, pad_names={pad_name})

PadProvider

Bases: Protocol


              flowchart TD
              sgn.groups.PadProvider[PadProvider]

              

              click sgn.groups.PadProvider href "" "sgn.groups.PadProvider"

Protocol defining the interface required by PadIteratorMixin.

Source code in src/sgn/groups.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class PadProvider(Protocol):
    """Protocol defining the interface required by PadIteratorMixin."""

    @property
    def srcs(self) -> dict[str, SourcePad]:
        """Get source pads as a dictionary mapping pad names to SourcePad objects."""
        ...

    @property
    def snks(self) -> dict[str, SinkPad]:
        """Get sink pads as a dictionary mapping pad names to SinkPad objects."""
        ...

    @property
    def elements(self) -> list[Element]:
        """Get all elements referenced by this object."""
        ...

elements property

Get all elements referenced by this object.

snks property

Get sink pads as a dictionary mapping pad names to SinkPad objects.

srcs property

Get source pads as a dictionary mapping pad names to SourcePad objects.

PadSelection dataclass

Bases: PadIteratorMixin


              flowchart TD
              sgn.groups.PadSelection[PadSelection]
              sgn.groups.PadIteratorMixin[PadIteratorMixin]

                              sgn.groups.PadIteratorMixin --> sgn.groups.PadSelection
                


              click sgn.groups.PadSelection href "" "sgn.groups.PadSelection"
              click sgn.groups.PadIteratorMixin href "" "sgn.groups.PadIteratorMixin"

Represents a selection of specific pads from an element.

This allows users to specify exactly which pads from an element should be used in grouping operations.

Source code in src/sgn/groups.py 
 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
101
102
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
@dataclass
class PadSelection(PadIteratorMixin):
    """Represents a selection of specific pads from an element.

    This allows users to specify exactly which pads from an element
    should be used in grouping operations.
    """

    element: Element
    pad_names: set[str]

    def __post_init__(self) -> None:
        """Validate that the selected pad names exist on the element."""
        all_pad_names = set()

        # Add source pad names if element has them
        if isinstance(self.element, (SourceElement, TransformElement)):
            all_pad_names.update(self.element.srcs.keys())

        # Add sink pad names if element has them
        if isinstance(self.element, (TransformElement, SinkElement)):
            all_pad_names.update(self.element.snks.keys())

        invalid_names = self.pad_names - all_pad_names
        if invalid_names:
            msg = (
                f"Pad names {invalid_names} not found on element '{self.element.name}'"
                f" Pad names available: {all_pad_names}"
            )
            raise ValueError(msg)

    @property
    def srcs(self) -> dict[str, SourcePad]:
        """Extract selected source pads from the element using names as keys."""
        if isinstance(self.element, (SourceElement, TransformElement)):
            return {
                name: pad
                for name, pad in self.element.srcs.items()
                if name in self.pad_names
            }
        return {}

    @property
    def snks(self) -> dict[str, SinkPad]:
        """Extract selected sink pads from the element using names as keys."""
        if isinstance(self.element, (TransformElement, SinkElement)):
            return {
                name: pad
                for name, pad in self.element.snks.items()
                if name in self.pad_names
            }
        return {}

    @property
    def elements(self) -> list[Element]:
        """Get the element referenced by this selection."""
        return [self.element]

elements property

Get the element referenced by this selection.

snks property

Extract selected sink pads from the element using names as keys.

srcs property

Extract selected source pads from the element using names as keys.

__post_init__()

Validate that the selected pad names exist on the element.

Source code in src/sgn/groups.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def __post_init__(self) -> None:
    """Validate that the selected pad names exist on the element."""
    all_pad_names = set()

    # Add source pad names if element has them
    if isinstance(self.element, (SourceElement, TransformElement)):
        all_pad_names.update(self.element.srcs.keys())

    # Add sink pad names if element has them
    if isinstance(self.element, (TransformElement, SinkElement)):
        all_pad_names.update(self.element.snks.keys())

    invalid_names = self.pad_names - all_pad_names
    if invalid_names:
        msg = (
            f"Pad names {invalid_names} not found on element '{self.element.name}'"
            f" Pad names available: {all_pad_names}"
        )
        raise ValueError(msg)

group(*items)

Create a unified group from elements, pad selections, and existing groups.

This function always returns an ElementGroup that holds elements and/or pad selections. The decision of which pads to extract is deferred until the group is used in pipeline.connect().

Parameters:

Name Type Description Default
*items Element | PadSelection | ElementGroup

Elements, pad selections, and/or existing ElementGroups to combine.

 ()

Returns:

Type Description
ElementGroup

A group containing all the provided items.

Examples:

>>> src1 = IterSource(name="src1", source_pad_names=["H1"])
>>> src2 = IterSource(name="src2", source_pad_names=["L1", "V1"])
>>> snk = Sink(name="snk", sink_pad_names=["H1", "L1"])
>>>
>>> # Group elements
>>> sources = group(src1, src2)
>>>
>>> # Group with pad selection
>>> l1_only = select(src2, "L1")
>>> sources_with_selection = group(src1, l1_only)
>>>
>>> # Connect groups
>>> pipeline.connect(sources_with_selection, snk)
Source code in src/sgn/groups.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def group(*items: Element | PadSelection | ElementGroup) -> ElementGroup:
    """Create a unified group from elements, pad selections, and existing groups.

    This function always returns an ElementGroup that holds elements and/or
    pad selections. The decision of which pads to extract is deferred until
    the group is used in pipeline.connect().

    Args:
        *items: Elements, pad selections, and/or existing ElementGroups to combine.

    Returns:
        A group containing all the provided items.

    Examples:
        >>> src1 = IterSource(name="src1", source_pad_names=["H1"])
        >>> src2 = IterSource(name="src2", source_pad_names=["L1", "V1"])
        >>> snk = Sink(name="snk", sink_pad_names=["H1", "L1"])
        >>>
        >>> # Group elements
        >>> sources = group(src1, src2)
        >>>
        >>> # Group with pad selection
        >>> l1_only = select(src2, "L1")
        >>> sources_with_selection = group(src1, l1_only)
        >>>
        >>> # Connect groups
        >>> pipeline.connect(sources_with_selection, snk)
    """
    all_items: list[Element | PadSelection] = []

    for item in items:
        if isinstance(item, PadSelection):
            all_items.append(item)
        elif isinstance(item, (SourceElement, TransformElement, SinkElement)):
            all_items.append(item)
        elif isinstance(item, ElementGroup):
            all_items.extend(item.items)
        else:
            msg = f"Expected Element, PadSelection, or ElementGroup, got {type(item)}"
            raise TypeError(msg)

    return ElementGroup(items=all_items)

select(target, *pad_names) 

select(target: Element, *pad_names: str) -> PadSelection

select(target: PadSelection, *pad_names: str) -> PadSelection

select(target: ElementGroup, *pad_names: str) -> ElementGroup

Create or refine pad selections from elements, existing selections, or groups.

Parameters:

Name Type Description Default
target Element | PadSelection | ElementGroup

The element, pad selection, or element group to select pads from.

 required
*pad_names str

Names of the pads to select.

 ()

Returns:

Type Description
PadSelection | ElementGroup

A PadSelection if target is Element or PadSelection,
PadSelection | ElementGroup

or ElementGroup if target is ElementGroup.

Examples:

>>> src = IterSource(name="src", source_pad_names=["H1", "L1", "V1"])
>>> h1_only = select(src, "H1")
>>> h1_l1_only = select(src, "H1", "L1")
>>>
>>> # Narrow an existing selection
>>> h1_from_selection = select(h1_l1_only, "H1")
>>>
>>> # Select from a group
>>> sources = group(src1, src2)
>>> h1_from_group = select(sources, "H1")
Source code in src/sgn/groups.py 
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
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def select(
    target: Element | PadSelection | ElementGroup, *pad_names: str
) -> PadSelection | ElementGroup:
    """Create or refine pad selections from elements, existing selections, or groups.

    Args:
        target: The element, pad selection, or element group to select pads from.
        *pad_names: Names of the pads to select.

    Returns:
        A PadSelection if target is Element or PadSelection,
        or ElementGroup if target is ElementGroup.

    Examples:
        >>> src = IterSource(name="src", source_pad_names=["H1", "L1", "V1"])
        >>> h1_only = select(src, "H1")
        >>> h1_l1_only = select(src, "H1", "L1")
        >>>
        >>> # Narrow an existing selection
        >>> h1_from_selection = select(h1_l1_only, "H1")
        >>>
        >>> # Select from a group
        >>> sources = group(src1, src2)
        >>> h1_from_group = select(sources, "H1")
    """
    match target:
        case SourceElement() | TransformElement() | SinkElement():
            return PadSelection(element=target, pad_names=set(pad_names))
        case PadSelection():
            # Narrow the existing selection by intersecting pad names
            new_pad_names = target.pad_names & set(pad_names)
            if not new_pad_names:
                msg = (
                    f"No matching pads found. Requested: {set(pad_names)}, "
                    f"Available: {target.pad_names}"
                )
                raise ValueError(msg)
            return PadSelection(element=target.element, pad_names=new_pad_names)
        case ElementGroup():
            return target.select(*pad_names)
        case _:
            msg = f"Expected Element, PadSelection, or ElementGroup, got {type(target)}"
            raise TypeError(msg)