Skip to content

sgn.transforms

Transforms elements and related utilities for the SGN framework.

CallableTransform dataclass

Bases: InputPull

A transform element that takes a mapping of {(input, combinations) -> callable}, each of which is mapped to a unique output pad.

Parameters:

Name Type Description Default
callmap dict[str, Callable]

dict[str, Callable], a mapping of output pad names to callables

 dict()
depmap dict[str, tuple[str, ...]]

dict[str, tuple[str, ...]], mapping of output pad names to input combinations

 dict()
Source code in sgn/transforms.py 
 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
 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
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
131
132
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
@dataclass
class CallableTransform(InputPull):
    """A transform element that takes a mapping of {(input, combinations) -> callable},
    each of which is mapped to a unique output pad.

    Args:
        callmap:
            dict[str, Callable], a mapping of output pad names to callables
        depmap:
            dict[str, tuple[str, ...]], mapping of output pad names to input
            combinations
    """

    callmap: dict[str, Callable] = field(default_factory=dict)
    depmap: dict[str, tuple[str, ...]] = field(default_factory=dict)

    def __post_init__(self):
        """Setup callable mappings and name associated source pads."""
        if self.source_pads or self.source_pad_names:
            raise ValueError(
                "CallableTransform does not accept source_pads or "
                "source_pad_names, they are inferred from callmap and namemap"
            )

        # Setup callable maps
        if not self.callmap:
            raise ValueError("CallableTransform requires a callmap")

        # Format callmap keys to ensure name:snk:pad format
        formatted_callmap = {}
        for k, v in self.callmap.items():
            new_key = k
            if not k.startswith(f"{self.name}:src:"):
                new_key = f"{self.name}:src:{k}"
            formatted_callmap[new_key] = v
        self.callmap = formatted_callmap

        # Determine source pad names
        if not self.depmap:
            raise ValueError("CallableTransform requires a depmap")

        # Format namemap keys to ensure name:src:pad format
        formatted_namemap = {}
        for k, v in self.depmap.items():
            new_key = k
            new_val = []
            if not new_key.startswith(f"{self.name}:src:"):
                new_key = f"{self.name}:src:{k}"

            for token in v:
                if token.startswith(f"{self.name}:snk:"):
                    new_val.append(token)
                else:
                    new_val.append(f"{self.name}:snk:{token}")
            new_val = tuple(new_val)
            formatted_namemap[new_key] = new_val

        self.depmap = formatted_namemap

        # Check that callmap and namemap have same set of keys
        if set(self.callmap.keys()) != set(self.depmap.keys()):
            raise ValueError(
                "callmap and namemap must have the same set of keys, "
                f"got {set(self.callmap.keys())} and {set(self.depmap.keys())}"
            )

        # Setup source pad names if needed
        if not self.source_pad_names:
            self.source_pad_names = [
                k.split(":")[-1] for k in sorted(self.depmap.keys())
            ]

        # Setup sink pad names if needed
        if not self.sink_pad_names:
            sink_names = set()
            for v in self.depmap.values():
                sink_names.update(v)
            self.sink_pad_names = [v.split(":")[-1] for v in sorted(sink_names)]

        # Create source pads via parent class
        super().__post_init__()

    def new(self, pad: SourcePad) -> Frame:
        """Apply the callable associated to the pad to the corresponding inputs.

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

        Returns:
            Frame, the output frame
        """
        # Determine input keys
        input_keys = self.depmap[pad.name]

        # Get callable
        func = self.callmap[pad.name]

        # Get inputs
        input_args = tuple(
            self.inputs[k] for k in input_keys
        )  # same order as input_keys

        # Apply callable
        res = func(*input_args)

        return Frame(
            # TODO: generalize this to choose any v. all behavior
            EOS=any(frame.EOS for frame in self.inputs.values()),
            data=res,
        )

    @staticmethod
    def from_combinations(
        name: str,
        combos: Iterable[tuple[tuple[str, ...], Callable, str]],
        sink_pad_names: Optional[Sequence[str]] = None,
        source_pad_names: Optional[Sequence[str]] = None,
    ):
        """Create a CallableTransform from a list of combinations where each combination
        is.

            (input_keys, func, output_name)

        Args:
            name:
                str, the name of the CallableTransform
            combos:
                Sequence[tuple[tuple[str, ...], Callable, str]], a list of
                combinations to create the CallableTransform, where each
                combination is a tuple of the input keys, the callable, and the
                output name
            sink_pad_names:
                Optional[list[str]], the names of the sink pads (input pads). If not
                specified, inferred from the combos
            source_pad_names:
                Optional[list[str]], the names of the source pads (output pads). If
                not specified, inferred from the combos

        Returns:
            CallableTransform, the created CallableTransform
        """
        callmap = {out: func for _, func, out in combos}
        namemap = {out: inp for inp, _, out in combos}
        return CallableTransform(
            name=name,
            callmap=callmap,
            depmap=namemap,
            sink_pad_names=[] if sink_pad_names is None else sink_pad_names,
            source_pad_names=[] if source_pad_names is None else source_pad_names,
        )

    @staticmethod
    def from_callable(
        name: str,
        callable: Callable,
        output_pad_name: str,
        sink_pad_names: list[str],
    ):
        """Create a CallableTransform from a single callable that will be applied to all
        inputs.

        Args:
            name:
                str, the name of the CallableTransform
            callable:
                Callable, the callable to use for the transform
            output_pad_name:
                str, the name of the output pad
            sink_pad_names:
                list[str], the names of the sink pads (input pads)

        Returns:
            CallableTransform, the created CallableTransform
        """
        return CallableTransform(
            name=name,
            sink_pad_names=sink_pad_names,
            callmap={output_pad_name: callable},
            depmap={output_pad_name: tuple(sink_pad_names)},
        )

__post_init__()

Setup callable mappings and name associated source pads.

Source code in sgn/transforms.py 
 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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __post_init__(self):
    """Setup callable mappings and name associated source pads."""
    if self.source_pads or self.source_pad_names:
        raise ValueError(
            "CallableTransform does not accept source_pads or "
            "source_pad_names, they are inferred from callmap and namemap"
        )

    # Setup callable maps
    if not self.callmap:
        raise ValueError("CallableTransform requires a callmap")

    # Format callmap keys to ensure name:snk:pad format
    formatted_callmap = {}
    for k, v in self.callmap.items():
        new_key = k
        if not k.startswith(f"{self.name}:src:"):
            new_key = f"{self.name}:src:{k}"
        formatted_callmap[new_key] = v
    self.callmap = formatted_callmap

    # Determine source pad names
    if not self.depmap:
        raise ValueError("CallableTransform requires a depmap")

    # Format namemap keys to ensure name:src:pad format
    formatted_namemap = {}
    for k, v in self.depmap.items():
        new_key = k
        new_val = []
        if not new_key.startswith(f"{self.name}:src:"):
            new_key = f"{self.name}:src:{k}"

        for token in v:
            if token.startswith(f"{self.name}:snk:"):
                new_val.append(token)
            else:
                new_val.append(f"{self.name}:snk:{token}")
        new_val = tuple(new_val)
        formatted_namemap[new_key] = new_val

    self.depmap = formatted_namemap

    # Check that callmap and namemap have same set of keys
    if set(self.callmap.keys()) != set(self.depmap.keys()):
        raise ValueError(
            "callmap and namemap must have the same set of keys, "
            f"got {set(self.callmap.keys())} and {set(self.depmap.keys())}"
        )

    # Setup source pad names if needed
    if not self.source_pad_names:
        self.source_pad_names = [
            k.split(":")[-1] for k in sorted(self.depmap.keys())
        ]

    # Setup sink pad names if needed
    if not self.sink_pad_names:
        sink_names = set()
        for v in self.depmap.values():
            sink_names.update(v)
        self.sink_pad_names = [v.split(":")[-1] for v in sorted(sink_names)]

    # Create source pads via parent class
    super().__post_init__()

from_callable(name, callable, output_pad_name, sink_pad_names) staticmethod

Create a CallableTransform from a single callable that will be applied to all inputs.

Parameters:

Name Type Description Default
name str

str, the name of the CallableTransform

 required
callable Callable

Callable, the callable to use for the transform

 required
output_pad_name str

str, the name of the output pad

 required
sink_pad_names list[str]

list[str], the names of the sink pads (input pads)

 required

Returns:

Type Description

CallableTransform, the created CallableTransform
Source code in sgn/transforms.py 
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
@staticmethod
def from_callable(
    name: str,
    callable: Callable,
    output_pad_name: str,
    sink_pad_names: list[str],
):
    """Create a CallableTransform from a single callable that will be applied to all
    inputs.

    Args:
        name:
            str, the name of the CallableTransform
        callable:
            Callable, the callable to use for the transform
        output_pad_name:
            str, the name of the output pad
        sink_pad_names:
            list[str], the names of the sink pads (input pads)

    Returns:
        CallableTransform, the created CallableTransform
    """
    return CallableTransform(
        name=name,
        sink_pad_names=sink_pad_names,
        callmap={output_pad_name: callable},
        depmap={output_pad_name: tuple(sink_pad_names)},
    )

from_combinations(name, combos, sink_pad_names=None, source_pad_names=None) staticmethod

Create a CallableTransform from a list of combinations where each combination is.

(input_keys, func, output_name)

Parameters:

Name Type Description Default
name str

str, the name of the CallableTransform

 required
combos Iterable[tuple[tuple[str, ...], Callable, str]]

Sequence[tuple[tuple[str, ...], Callable, str]], a list of combinations to create the CallableTransform, where each combination is a tuple of the input keys, the callable, and the output name

 required
sink_pad_names Optional[Sequence[str]]

Optional[list[str]], the names of the sink pads (input pads). If not specified, inferred from the combos

 None
source_pad_names Optional[Sequence[str]]

Optional[list[str]], the names of the source pads (output pads). If not specified, inferred from the combos

 None

Returns:

Type Description

CallableTransform, the created CallableTransform
Source code in sgn/transforms.py 
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
@staticmethod
def from_combinations(
    name: str,
    combos: Iterable[tuple[tuple[str, ...], Callable, str]],
    sink_pad_names: Optional[Sequence[str]] = None,
    source_pad_names: Optional[Sequence[str]] = None,
):
    """Create a CallableTransform from a list of combinations where each combination
    is.

        (input_keys, func, output_name)

    Args:
        name:
            str, the name of the CallableTransform
        combos:
            Sequence[tuple[tuple[str, ...], Callable, str]], a list of
            combinations to create the CallableTransform, where each
            combination is a tuple of the input keys, the callable, and the
            output name
        sink_pad_names:
            Optional[list[str]], the names of the sink pads (input pads). If not
            specified, inferred from the combos
        source_pad_names:
            Optional[list[str]], the names of the source pads (output pads). If
            not specified, inferred from the combos

    Returns:
        CallableTransform, the created CallableTransform
    """
    callmap = {out: func for _, func, out in combos}
    namemap = {out: inp for inp, _, out in combos}
    return CallableTransform(
        name=name,
        callmap=callmap,
        depmap=namemap,
        sink_pad_names=[] if sink_pad_names is None else sink_pad_names,
        source_pad_names=[] if source_pad_names is None else source_pad_names,
    )

new(pad)

Apply the callable associated to the pad to the corresponding inputs.

Parameters:

Name Type Description Default
pad SourcePad

SourcePad, The source pad through which the frame is passed

 required

Returns:

Type Description
Frame

Frame, the output frame
Source code in sgn/transforms.py 
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
144
145
def new(self, pad: SourcePad) -> Frame:
    """Apply the callable associated to the pad to the corresponding inputs.

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

    Returns:
        Frame, the output frame
    """
    # Determine input keys
    input_keys = self.depmap[pad.name]

    # Get callable
    func = self.callmap[pad.name]

    # Get inputs
    input_args = tuple(
        self.inputs[k] for k in input_keys
    )  # same order as input_keys

    # Apply callable
    res = func(*input_args)

    return Frame(
        # TODO: generalize this to choose any v. all behavior
        EOS=any(frame.EOS for frame in self.inputs.values()),
        data=res,
    )

InputPull dataclass

Bases: TransformElement

Input Pull mixin class for Transforms creates a dictionary of inputs and a pull method to populate the dictionary.

The new method remains abstract and must be implemented in the subclass.

Source code in sgn/transforms.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@dataclass
class InputPull(TransformElement):
    """Input Pull mixin class for Transforms creates a dictionary of inputs and a pull
    method to populate the dictionary.

    The new method remains abstract and must be implemented in the subclass.
    """

    def __post_init__(self):
        self.inputs = {}
        super().__post_init__()

    def pull(self, pad: SinkPad, frame: Frame) -> None:
        """Pull a frame into the transform element.

        Args:
            pad:
                SinkPad, The sink pad that is receiving the frame
            frame:
                Frame, the frame to pull into the pad.
        """
        self.inputs[pad.name] = frame

pull(pad, frame)

Pull a frame into the transform element.

Parameters:

Name Type Description Default
pad SinkPad

SinkPad, The sink pad that is receiving the frame

 required
frame Frame

Frame, the frame to pull into the pad.

 required
Source code in sgn/transforms.py 
23
24
25
26
27
28
29
30
31
32
def pull(self, pad: SinkPad, frame: Frame) -> None:
    """Pull a frame into the transform element.

    Args:
        pad:
            SinkPad, The sink pad that is receiving the frame
        frame:
            Frame, the frame to pull into the pad.
    """
    self.inputs[pad.name] = frame