Refactored egress tests using NumPy functions

Author: seanlaw

tests/naive.py

@@ -10,8 +10,8 @@ def z_norm(a, axis=0, threshold=1e-7):
     return (a - np.mean(a, axis, keepdims=True)) / std
 
 
-def distance(a, b):
-    return np.linalg.norm(a - b)
+def distance(a, b, axis=0):
+    return np.linalg.norm(a - b, axis=axis)
 
 
 def apply_exclusion_zone(D, trivial_idx, excl_zone):

tests/test_aampi.py

@@ -848,6 +848,7 @@ def test_aampi_identical_subsequence_self_join_egress():
 def test_aampi_profile_index_match():
     T_full = np.random.rand(64)
     m = 3
+    T_full_subseq = core.rolling_window(T_full, m)
     warm_start = 8
 
     T_stream = T_full[:warm_start].copy()
@@ -860,23 +861,17 @@ def test_aampi_profile_index_match():
         t = T_full[i]
         stream.update(t)
 
-        for j in range(stream.I_.shape[0]):
-            I = stream.I_[j]
-            left_I = stream.left_I_[j]
-
-            if I < 0:
-                P[j] = np.inf
-            else:
-                P[j] = naive.distance(
-                    T_full[j + n + 1 : j + n + 1 + m], T_full[I : I + m]
-                )
-
-            if left_I < 0:
-                left_P[j] = np.inf
-            else:
-                left_P[j] = naive.distance(
-                    T_full[j + n + 1 : j + n + 1 + m], T_full[left_I : left_I + m]
-                )
+        P[:] = np.inf
+        idx = np.argwhere(stream.I_ >= 0).flatten()
+        P[idx] = naive.distance(
+            T_full_subseq[idx + n + 1], T_full_subseq[stream.I_[idx]], axis=1
+        )
+
+        left_P[:] = np.inf
+        idx = np.argwhere(stream.left_I_ >= 0).flatten()
+        left_P[idx] = naive.distance(
+            T_full_subseq[idx + n + 1], T_full_subseq[stream.left_I_[idx]], axis=1
+        )
 
         npt.assert_almost_equal(stream.P_, P)
         npt.assert_almost_equal(stream.left_P_, left_P)

tests/test_stumpi.py

@@ -856,6 +856,7 @@ def test_stumpi_identical_subsequence_self_join_egress():
 def test_stumpi_profile_index_match():
     T_full = np.random.rand(64)
     m = 3
+    T_full_subseq = core.rolling_window(T_full, m)
     warm_start = 8
 
     T_stream = T_full[:warm_start].copy()
@@ -868,25 +869,21 @@ def test_stumpi_profile_index_match():
         t = T_full[i]
         stream.update(t)
 
-        for j in range(stream.I_.shape[0]):
-            I = stream.I_[j]
-            left_I = stream.left_I_[j]
-
-            if I < 0:
-                P[j] = np.inf
-            else:
-                P[j] = naive.distance(
-                    naive.z_norm(T_full[j + n + 1 : j + n + 1 + m]),
-                    naive.z_norm(T_full[I : I + m]),
-                )
-
-            if left_I < 0:
-                left_P[j] = np.inf
-            else:
-                left_P[j] = naive.distance(
-                    naive.z_norm(T_full[j + n + 1 : j + n + 1 + m]),
-                    naive.z_norm(T_full[left_I : left_I + m]),
-                )
+        P[:] = np.inf
+        idx = np.argwhere(stream.I_ >= 0).flatten()
+        P[idx] = naive.distance(
+            naive.z_norm(T_full_subseq[idx + n + 1], axis=1),
+            naive.z_norm(T_full_subseq[stream.I_[idx]], axis=1),
+            axis=1,
+        )
+
+        left_P[:] = np.inf
+        idx = np.argwhere(stream.left_I_ >= 0).flatten()
+        left_P[idx] = naive.distance(
+            naive.z_norm(T_full_subseq[idx + n + 1], axis=1),
+            naive.z_norm(T_full_subseq[stream.left_I_[idx]], axis=1),
+            axis=1,
+        )
 
         npt.assert_almost_equal(stream.P_, P)
         npt.assert_almost_equal(stream.left_P_, left_P)