sgnts.transforms.sumindex

SumIndex dataclass

Bases: TSTransform

              flowchart TD
              sgnts.transforms.sumindex.SumIndex[SumIndex]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.sumindex.SumIndex
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.sumindex.SumIndex href "" "sgnts.transforms.sumindex.SumIndex"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"
            

Sum array values over slices in the zero-th dimension.

Parameters:

Name	Type	Description	Default
sl	list[slice]	list[slice], the slices to sum over	required
backend	type[ArrayBackend]	type[ArrayBackend], the wrapper around array operations.	NumpyBackend

Source code in sgnts/transforms/sumindex.py

@dataclass(kw_only=True)
class SumIndex(TSTransform):
    """Sum array values over slices in the zero-th dimension.

    Args:
        sl:
            list[slice], the slices to sum over
        backend:
            type[ArrayBackend], the wrapper around array operations.
    """

    sl: list[slice]
    backend: type[ArrayBackend] = NumpyBackend

    @validator.one_to_one
    def validate(self) -> None:
        for sl in self.sl:
            assert isinstance(sl, slice)

    @transform.one_to_one
    def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        """Sum array values over slices."""
        for buf in input_frame:
            if buf.is_gap:
                data = None
                shape = (len(self.sl),) + buf.shape[-2:]
            else:
                data_all = []
                for sl in self.sl:
                    if sl.stop - sl.start == 1:
                        data_all.append((buf.data[sl.start, :, :]))
                    else:
                        data_all.append(self.backend.sum(buf.data[sl, :, :], axis=0))

                data = self.backend.stack(data_all)
                shape = data.shape

            buf = buf.copy(data=data, shape=shape)
            output_frame.append(buf)

process(input_frame, output_frame)

Sum array values over slices.

Source code in sgnts/transforms/sumindex.py

@transform.one_to_one
def process(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
    """Sum array values over slices."""
    for buf in input_frame:
        if buf.is_gap:
            data = None
            shape = (len(self.sl),) + buf.shape[-2:]
        else:
            data_all = []
            for sl in self.sl:
                if sl.stop - sl.start == 1:
                    data_all.append((buf.data[sl.start, :, :]))
                else:
                    data_all.append(self.backend.sum(buf.data[sl, :, :], axis=0))

            data = self.backend.stack(data_all)
            shape = data.shape

        buf = buf.copy(data=data, shape=shape)
        output_frame.append(buf)