Source code for tsaug.cross_sum

"""
Cross sum
"""
from typing import Tuple, Optional

import numpy as np
from .dimensionalize import dimensionalize
from .augmentor import _Augmentor


[docs]@dimensionalize def cross_sum( X: np.ndarray, Y: Optional[np.ndarray] = None, inds: Optional[np.ndarray] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """Sum cross given time series. Time series will be summed with others based on the given indices. Time points at which at least one time series of summation is anomalous will be marked as anomalous. Parameters ---------- X : numpy.ndarray Time series to be augmented. Matrix with shape (n,), (N, n) or (N, n, c), where n is the length of each series, N is the number of series, and c is the number of channels. Y : numpy.ndarray, optional Binary labels of time series, where 0 represents a normal point and 1 represents an anomalous points. Matrix with shape (n,), (N, n) or (N, n, cl), where n is the length of each series, N is the number of series, and cl is the number of classes (i.e. types of anomaly). Default: None. inds : numpy.array, optional Indices of time series to sum with. Matrix with shape (N, m), where N is the number of series, and m is the maximal number of series to sum with each series. The i-th output series is the sum of the i-th input series and the ind[i][j]-th time series for all j. Values of ind[i][j] can be NaN for series to be summed with less than m series. Returns ------- tuple (numpy.ndarray, numpy.ndarray) Augmented time series and augmented labels (if argument `Y` exists). """ N = ( 0 ) # type: int # NOTE: this is a horrible hack to type hint for Python 3.5 n = 0 # type: int c = 0 # type: int N, n, c = X.shape if inds is None: inds = np.zeros((N, 0)) if inds.ndim == 1: inds = inds.reshape((N, 1)) if inds.shape[0] != N: raise ValueError("Wrong shape of inds") m = inds.shape[1] # type: int X_aug = X.copy() # type: np.ndarray if Y is None: Y_aug = None # type: Optional[np.ndarray] else: Y_aug = Y.copy() for k in range(m): X_aug[np.isnan(inds[:, k]) >= 0] = ( X_aug[np.isnan(inds[:, k]) >= 0] + X[inds[np.isnan(inds[:, k]) >= 0, k]] ) if (Y_aug is not None) and (Y is not None): Y_aug[np.isnan(inds[:, k]) >= 0] = ( Y_aug[np.isnan(inds[:, k]) >= 0] + Y[inds[np.isnan(inds[:, k]) >= 0, k]] ) if Y_aug is not None: Y_aug = (Y_aug >= 1).astype(int) return X_aug, Y_aug
[docs]@dimensionalize def random_cross_sum( X: np.ndarray, Y: Optional[np.ndarray] = None, max_sum_series: Optional[int] = 5, random_seed: Optional[int] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """Sum cross given time series randomly. Time series will be summed with others randomly. Time points at which at least one time series of summation is anomalous will be marked as anomalous. Parameters ---------- X : numpy.ndarray Time series to be augmented. Matrix with shape (n,), (N, n) or (N, n, c), where n is the length of each series, N is the number of series, and c is the number of channels. Y : numpy.ndarray, optional Binary labels of time series, where 0 represents a normal point and 1 represents an anomalous points. Matrix with shape (n,), (N, n) or (N, n, cl), where n is the length of each series, N is the number of series, and cl is the number of classes (i.e. types of anomaly). Default: None. max_sum_series : int, optional Maximal number of time series to cross sum. Default: 5. random_seed : int, optional Random seed used to initialize the pseudo-random number generator. Default: None. Returns ------- tuple (numpy.ndarray, numpy.ndarray) Augmented time series and augmented labels (if argument `Y` exists). """ N = ( 0 ) # type: int # NOTE: this is a horrible hack to type hint for Python 3.5 n = 0 # type: int c = 0 # type: int N, n, c = X.shape rand = np.random.RandomState(random_seed) inds = rand.choice(range(-1, N), size=(N, max_sum_series)) return cross_sum(X, Y, inds)
[docs]class CrossSum(_Augmentor): """Augmentor that sum cross given time series. Time series will be summed with others based on the given indices. Time points at which at least one time series of summation is anomalous will be marked as anomalous. Parameters ---------- inds : numpy.array, optional Indices of time series to sum with. Matrix with shape (N, m), where N is the number of series, and m is the maximal number of series to sum with each series. The i-th output series is the sum of the i-th input series and the ind[i][j]-th time series for all j. Values of ind[i][j] can be NaN for series to be summed with less than m series. """ def __init__(self, inds: np.ndarray = None) -> None: super().__init__(augmentor_func=cross_sum, is_random=False, inds=inds) @property def inds(self) -> np.ndarray: return self._params["inds"] @inds.setter def inds(self, inds: Optional[np.ndarray]) -> None: self._params["inds"] = inds
[docs]class RandomCrossSum(_Augmentor): """Augmentor that sums cross given time series randomly. Time series will be summed with others randomly. Time points at which at least one time series of summation is anomalous will be marked as anomalous. Parameters ---------- max_sum_series : int, optional Maximal number of time series to cross sum. Default: 5. random_seed : int, optional Random seed used to initialize the pseudo-random number generator. Default: None. """ def __init__( self, max_sum_series: Optional[int] = 5, random_seed: Optional[int] = None, ) -> None: super().__init__( augmentor_func=random_cross_sum, is_random=True, max_sum_series=max_sum_series, random_seed=random_seed, ) @property def max_sum_series(self) -> int: return self._params["max_sum_series"] @max_sum_series.setter def max_sum_series(self, max_sum_series: Optional[int]) -> None: self._params["max_sum_series"] = max_sum_series @property def random_seed(self) -> Optional[int]: return self._params["random_seed"] @random_seed.setter def random_seed(self, random_seed: Optional[int]) -> None: self._params["random_seed"] = random_seed