(
self, T, m, excl_zone=None, k=1, mp=None, T_subseq_isconstant_func=None
)
| 979 | |
| 980 | class stumpi_egress(object): |
| 981 | def __init__( |
| 982 | self, T, m, excl_zone=None, k=1, mp=None, T_subseq_isconstant_func=None |
| 983 | ): |
| 984 | self._T = np.asarray(T) |
| 985 | self._T = self._T.copy() |
| 986 | self._T_isfinite = np.isfinite(self._T) |
| 987 | self._m = m |
| 988 | self._k = k |
| 989 | if T_subseq_isconstant_func is None: |
| 990 | T_subseq_isconstant_func = core._rolling_isconstant |
| 991 | self._T_subseq_isconstant_func = T_subseq_isconstant_func |
| 992 | self._T_subseq_isconstant = rolling_isconstant( |
| 993 | self._T, self._m, self._T_subseq_isconstant_func |
| 994 | ) |
| 995 | |
| 996 | self._excl_zone = excl_zone |
| 997 | if self._excl_zone is None: |
| 998 | self._excl_zone = int(np.ceil(self._m / config.STUMPY_EXCL_ZONE_DENOM)) |
| 999 | |
| 1000 | self._l = self._T.shape[0] - m + 1 |
| 1001 | |
| 1002 | if mp is None: |
| 1003 | mp = stump( |
| 1004 | self._T, |
| 1005 | self._m, |
| 1006 | exclusion_zone=self._excl_zone, |
| 1007 | k=self._k, |
| 1008 | T_A_subseq_isconstant=self._T_subseq_isconstant, |
| 1009 | ) |
| 1010 | else: |
| 1011 | mp = mp.copy() |
| 1012 | |
| 1013 | self._P = mp[:, :k].astype(np.float64) |
| 1014 | self._I = mp[:, k : 2 * k].astype(np.int64) |
| 1015 | |
| 1016 | self._left_I = mp[:, 2 * k].astype(np.int64) |
| 1017 | self._left_P = np.full_like(self._left_I, np.inf, dtype=np.float64) |
| 1018 | |
| 1019 | for idx, nn_idx in enumerate(self._left_I): |
| 1020 | if nn_idx >= 0: |
| 1021 | if self._T_subseq_isconstant[idx] and self._T_subseq_isconstant[nn_idx]: |
| 1022 | self._left_P[idx] = 0 |
| 1023 | elif ( |
| 1024 | self._T_subseq_isconstant[idx] or self._T_subseq_isconstant[nn_idx] |
| 1025 | ): |
| 1026 | self._left_P[idx] = np.sqrt(self._m) |
| 1027 | else: |
| 1028 | self._left_P[idx] = distance_profile( |
| 1029 | self._T[idx : idx + self._m], |
| 1030 | self._T[nn_idx : nn_idx + self._m], |
| 1031 | m, |
| 1032 | )[0] |
| 1033 | |
| 1034 | self._n_appended = 0 |
| 1035 | |
| 1036 | def update(self, t): |
| 1037 | self._T[:] = np.roll(self._T, -1) |
nothing calls this directly
no test coverage detected