Skip to content

sgnts.transforms.converter

Converter dataclass

Bases: TSTransform


              flowchart TD
              sgnts.transforms.converter.Converter[Converter]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.converter.Converter
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.converter.Converter href "" "sgnts.transforms.converter.Converter"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"

Change the data type or the device of the data.

Parameters:

Name Type Description Default
backend str

str, the backend to convert the data to: 'numpy' or 'torch'.

 'numpy'
dtype str

the data type to convert to -- a name like 'float32' / 'float16' / 'float64' or a backend dtype object (e.g. torch.float16). Any numeric dtype the target backend provides is accepted.

 'float32'
device str

str, the device to convert the data to. For backend='numpy' only 'cpu'; for backend='torch', 'cpu' / 'cuda' / 'cuda:<GPU number>' where <GPU number> is the GPU device number.

 'cpu'
Notes

Thread safety: Marked thread_safe = True. Pad layout: matched sink and source pads (@validator.pad_names_match). Multiple sink pads' pull callbacks CAN run concurrently in the same wave, and so can multiple source pads' new callbacks; internal runs alone.

``pull`` (inherited ``TimeSeriesMixin.pull``):
per-pad-keyed dict writes; safe across pads. ``new``
(inherited): read-only lookup. ``process``: ``xp.asarray``
conversion (releases the GIL for large transfers/casts) on
local buffers; reads ``self.pad_map``, ``self.xp``,
``self.target_dtype``, ``self.device``, ``self.backend`` (all
post-init read-only).

Useful speedup from threading is real for GPU transfers
and large dtype conversions on parallel branches.

**Future editors MUST preserve thread safety**: keep
``process`` purely functional on its inputs. Do NOT add
element-level state mutated from ``pull`` outside of
per-pad-keyed containers.
Source code in src/sgnts/transforms/converter.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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
@dataclass
class Converter(TSTransform):
    """Change the data type or the device of the data.

    Args:
        backend:
            str, the backend to convert the data to: ``'numpy'`` or ``'torch'``.
        dtype:
            the data type to convert to -- a name like ``'float32'`` / ``'float16'``
            / ``'float64'`` or a backend dtype object (e.g. ``torch.float16``). Any
            numeric dtype the target backend provides is accepted.
        device:
            str, the device to convert the data to. For ``backend='numpy'`` only
            ``'cpu'``; for ``backend='torch'``, ``'cpu'`` / ``'cuda'`` /
            ``'cuda:<GPU number>'`` where ``<GPU number>`` is the GPU device number.

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: matched sink
            and source pads (``@validator.pad_names_match``).
            Multiple sink pads' ``pull`` callbacks CAN run concurrently
            in the same wave, and so can multiple source pads' ``new``
            callbacks; ``internal`` runs alone.

            ``pull`` (inherited ``TimeSeriesMixin.pull``):
            per-pad-keyed dict writes; safe across pads. ``new``
            (inherited): read-only lookup. ``process``: ``xp.asarray``
            conversion (releases the GIL for large transfers/casts) on
            local buffers; reads ``self.pad_map``, ``self.xp``,
            ``self.target_dtype``, ``self.device``, ``self.backend`` (all
            post-init read-only).

            Useful speedup from threading is real for GPU transfers
            and large dtype conversions on parallel branches.

            **Future editors MUST preserve thread safety**: keep
            ``process`` purely functional on its inputs. Do NOT add
            element-level state mutated from ``pull`` outside of
            per-pad-keyed containers.
    """

    thread_safe = True

    # The edge element: accepts any source backend and converts to the target.
    backends = ANY_BACKEND
    # The one element that *changes* the stream's backend (namespace/device), so the
    # output_prototype namespace/device guardrail does not apply to it.
    changes_backend = True

    backend: str = "numpy"
    dtype: str = "float32"
    device: str = "cpu"

    def configure(self) -> None:
        self.xp = _target_namespace(self.backend)
        if self.backend == "numpy" and self.device != "cpu":
            raise ValueError("Converting to numpy only supports device as cpu")
        self.target_dtype = normalize_dtype(self.xp, self.dtype)
        self.pad_map = {
            src_pad: self.snks[src_pad.pad_name] for src_pad in self.source_pads
        }

    @validator.pad_names_match
    def validate(self) -> None:
        pass

    def output_prototype(self, pad):
        """This element *changes* backend, so its output is the configured target --
        always, independent of the input (``changes_backend`` lifts the guardrail).
        A zero-length example in the target namespace / dtype / device.
        """
        return self.xp.zeros(0, dtype=self.target_dtype, device=self.device)

    def process(
        self,
        input_frames: dict[SinkPad, TSFrame],
        output_frames: dict[SourcePad, TSCollectFrame],
    ) -> None:
        """Convert data type and device via the target Array API namespace."""
        for pad in self.source_pads:
            frame = input_frames[self.pad_map[pad]]
            out: None | np.ndarray | torch.Tensor
            for buf in frame:
                if buf.is_gap:
                    out = None
                else:
                    data = buf.data
                    # The source backend is read from the data itself; only the
                    # *target* (this edge element's job) is configured.
                    src = backend_name(data)
                    if src is None:
                        raise ValueError("Unsupported data type")
                    if self.backend == "numpy" and src == "torch":
                        # NumPy cannot read GPU memory; move to host first.
                        out = self.xp.asarray(
                            data.detach().cpu(),
                            dtype=self.target_dtype,
                            device="cpu",
                        )
                    else:
                        out = self.xp.asarray(
                            data, dtype=self.target_dtype, device=self.device
                        )

                buf = buf.copy(data=out)
                output_frames[pad].append(buf)

output_prototype(pad)

This element changes backend, so its output is the configured target -- always, independent of the input (changes_backend lifts the guardrail). A zero-length example in the target namespace / dtype / device.

Source code in src/sgnts/transforms/converter.py 
109
110
111
112
113
114
def output_prototype(self, pad):
    """This element *changes* backend, so its output is the configured target --
    always, independent of the input (``changes_backend`` lifts the guardrail).
    A zero-length example in the target namespace / dtype / device.
    """
    return self.xp.zeros(0, dtype=self.target_dtype, device=self.device)

process(input_frames, output_frames)

Convert data type and device via the target Array API namespace.

Source code in src/sgnts/transforms/converter.py 
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
def process(
    self,
    input_frames: dict[SinkPad, TSFrame],
    output_frames: dict[SourcePad, TSCollectFrame],
) -> None:
    """Convert data type and device via the target Array API namespace."""
    for pad in self.source_pads:
        frame = input_frames[self.pad_map[pad]]
        out: None | np.ndarray | torch.Tensor
        for buf in frame:
            if buf.is_gap:
                out = None
            else:
                data = buf.data
                # The source backend is read from the data itself; only the
                # *target* (this edge element's job) is configured.
                src = backend_name(data)
                if src is None:
                    raise ValueError("Unsupported data type")
                if self.backend == "numpy" and src == "torch":
                    # NumPy cannot read GPU memory; move to host first.
                    out = self.xp.asarray(
                        data.detach().cpu(),
                        dtype=self.target_dtype,
                        device="cpu",
                    )
                else:
                    out = self.xp.asarray(
                        data, dtype=self.target_dtype, device=self.device
                    )

            buf = buf.copy(data=out)
            output_frames[pad].append(buf)