sgnts.transforms.amplify

Amplify dataclass

Bases: TSTransform

              flowchart TD
              sgnts.transforms.amplify.Amplify[Amplify]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.amplify.Amplify
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.amplify.Amplify href "" "sgnts.transforms.amplify.Amplify"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"
            

Amplify data by a factor.

Parameters:

Name	Type	Description	Default
factor	float	float, the factor to multiply the data with. The output dtype follows result_type(input, factor), so a wider-precision factor promotes (e.g. a float64 factor on a float32 input yields float64).	1

Notes

Thread safety: Marked thread_safe = True. Pad layout: 1 sink + 1 source (@transform.one_to_one). No same-element pull/new concurrency. internal runs alone.

``pull`` (inherited): per-pad-keyed dict writes. ``new``
(inherited): read-only lookup. ``process``: NumPy scalar
multiplication (releases the GIL for large arrays) on
local buffers; reads ``self.factor`` (post-init read-only).

**Future editors MUST preserve thread safety**: do not
relax the one-to-one constraint. Keep ``process`` purely
functional on its inputs.

Source code in src/sgnts/transforms/amplify.py

@dataclass
class Amplify(TSTransform):
    """Amplify data by a factor.

    Args:
        factor:
            float, the factor to multiply the data with. The output dtype follows
            ``result_type(input, factor)``, so a wider-precision factor promotes
            (e.g. a float64 factor on a float32 input yields float64).

    Notes:
        Thread safety:
            Marked ``thread_safe = True``. Pad layout: 1 sink + 1
            source (``@transform.one_to_one``). No same-element
            ``pull``/``new`` concurrency. ``internal`` runs alone.

            ``pull`` (inherited): per-pad-keyed dict writes. ``new``
            (inherited): read-only lookup. ``process``: NumPy scalar
            multiplication (releases the GIL for large arrays) on
            local buffers; reads ``self.factor`` (post-init read-only).

            **Future editors MUST preserve thread safety**: do not
            relax the one-to-one constraint. Keep ``process`` purely
            functional on its inputs.
    """

    thread_safe = True

    # data * factor works in any namespace via operator overloading.
    backends = ANY_BACKEND

    factor: float = 1

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

    def output_prototype(self, pad):
        # Output dtype = result_type(input, factor): a wider-precision factor (e.g.
        # float64 on a float32 input) promotes. Namespace/device ride through from
        # the input, so a plain real gain is unchanged.
        return self.input_prototype(self.sink_pads[0].pad_name) * self.factor

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        """Amplify non-gap data by the factor."""
        for buf in input_frame:
            if not buf.is_gap:
                assert buf.data is not None
                data = buf.data * self.factor
                buf = buf.copy(data=data)
            output_frame.append(buf)

process(input_frame, output_frame)

Amplify non-gap data by the factor.

Source code in src/sgnts/transforms/amplify.py

@transform.one_to_one
def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
    """Amplify non-gap data by the factor."""
    for buf in input_frame:
        if not buf.is_gap:
            assert buf.data is not None
            data = buf.data * self.factor
            buf = buf.copy(data=data)
        output_frame.append(buf)