Feature/pattern change detection

README.md

@@ -348,6 +348,7 @@ Distributed under the BSD 3-Clause License. See [LICENCE](./LICENSE) for more in
 Authors:
 
 - Ahmad Tarraf
+- Amine Aherbil
 
 This work is a result of cooperation between the Technical University of Darmstadt and INRIA in the scope of
 the [EuroHPC ADMIRE project](https://admire-eurohpc.eu/).

ftio/freq/_dft.py

@@ -79,6 +79,8 @@ def dft_fast(b: np.ndarray) -> np.ndarray:
     - np.ndarray, DFT of the input signal.
     """
     N = len(b)
+    if N == 0:
+        return np.array([])
     X = np.repeat(complex(0, 0), N)  # np.zeros(N)
     for k in range(0, N):
         for n in range(0, N):
@@ -98,6 +100,8 @@ def numpy_dft(b: np.ndarray) -> np.ndarray:
     Returns:
     - np.ndarray, DFT of the input signal.
     """
+    if len(b) == 0:
+        return np.array([])
     return np.fft.fft(b)
 
 

ftio/freq/_dft_workflow.py

@@ -46,6 +46,10 @@ def ftio_dft(
             - analysis_figures (AnalysisFigures): Data and plot figures.
             - share (SharedSignalData): Contains shared information, including sampled bandwidth and total bytes.
     """
+    # Suppress numpy warnings for empty array operations
+    import warnings
+    warnings.filterwarnings('ignore', category=RuntimeWarning, module='numpy')
+
     #! Default values for variables
     share = SharedSignalData()
     prediction = Prediction(args.transformation)
@@ -75,42 +79,66 @@ def ftio_dft(
     n = len(b_sampled)
     frequencies = args.freq * np.arange(0, n) / n
     X = dft(b_sampled)
-    X = X * np.exp(
-        -2j * np.pi * frequencies * time_stamps[0]
-    )  # Correct phase offset due to start time t0
+
+    # Safety check for empty time_stamps array
+    if len(time_stamps) > 0:
+        X = X * np.exp(
+            -2j * np.pi * frequencies * time_stamps[0]
+        )  # Correct phase offset due to start time t0
+    # If time_stamps is empty, skip phase correction
+
     amp = abs(X)
     phi = np.arctan2(X.imag, X.real)
     conf = np.zeros(len(amp))
     # welch(bandwidth,freq)
 
     #!  Find the dominant frequency
-    (dominant_index, conf[1 : int(n / 2) + 1], outlier_text) = outlier_detection(
-        amp, frequencies, args
-    )
+    # Safety check for empty arrays
+    if n > 0:
+        (dominant_index, conf[1 : int(n / 2) + 1], outlier_text) = outlier_detection(
+            amp, frequencies, args
+        )
 
-    #  Ignore DC offset
-    conf[0] = np.inf
-    if n % 2 == 0:
-        conf[int(n / 2) + 1 :] = np.flip(conf[1 : int(n / 2)])
+        #  Ignore DC offset
+        conf[0] = np.inf
+        if n % 2 == 0:
+            conf[int(n / 2) + 1 :] = np.flip(conf[1 : int(n / 2)])
+        else:
+            conf[int(n / 2) + 1 :] = np.flip(conf[1 : int(n / 2) + 1])
     else:
-        conf[int(n / 2) + 1 :] = np.flip(conf[1 : int(n / 2) + 1])
+        # Handle empty data case
+        dominant_index = np.array([])
+        outlier_text = "No data available for outlier detection"
 
     #! Assign data
-    prediction.dominant_freq = frequencies[dominant_index]
-    prediction.conf = conf[dominant_index]
-    if args.periodicity_detection is not None:
-        prediction.periodicity = conf[dominant_index]
-    prediction.amp = amp[dominant_index]
-    prediction.phi = phi[dominant_index]
-    prediction.t_start = time_stamps[0]
-    prediction.t_end = time_stamps[-1]
+    if n > 0 and len(dominant_index) > 0:
+        prediction.dominant_freq = frequencies[dominant_index]
+        prediction.conf = conf[dominant_index]
+        if args.periodicity_detection is not None:
+            prediction.periodicity = conf[dominant_index]
+        prediction.amp = amp[dominant_index]
+        prediction.phi = phi[dominant_index]
+    else:
+        # Handle empty data case
+        prediction.dominant_freq = np.array([])
+        prediction.conf = np.array([])
+        prediction.amp = np.array([])
+        prediction.phi = np.array([])
+
+    # Safety check for empty time_stamps
+    if len(time_stamps) > 0:
+        prediction.t_start = time_stamps[0]
+        prediction.t_end = time_stamps[-1]
+    else:
+        prediction.t_start = 0.0
+        prediction.t_end = 0.0
     prediction.freq = args.freq
     prediction.ranks = ranks
     prediction.total_bytes = total_bytes
     prediction.n_samples = n
 
     #! Save up to n_freq from the top candidates
-    if args.n_freq > 0:
+    if args.n_freq > 0 and n > 0:
         arr = amp[0 : int(np.ceil(n / 2))]
         top_candidates = np.argsort(-arr)  # from max to min
         n_freq = int(min(len(arr), args.n_freq))
@@ -124,7 +152,11 @@ def ftio_dft(
 
     periodicity_score = new_periodicity_scores(amp, b_sampled, prediction, args)
 
-    t_sampled = time_stamps[0] + np.arange(0, n) * 1 / args.freq
+    # Safety check for empty time_stamps
+    if len(time_stamps) > 0 and args.freq > 0:
+        t_sampled = time_stamps[0] + np.arange(0, n) * 1 / args.freq
+    else:
+        t_sampled = np.arange(0, n) * (1 / args.freq if args.freq > 0 else 1.0)
     #! Fourier fit if set
     if args.fourier_fit:
         fourier_fit(args, prediction, analysis_figures, b_sampled, t_sampled)

ftio/freq/discretize.py

@@ -38,6 +38,9 @@ def sample_data(
     Raises:
         RuntimeError: If no data is found in the sampled bandwidth.
     """
+    if len(t) == 0:
+        return np.empty(0), 0, " "
+
     if args is not None:
         freq = args.freq
         memory_limit = args.memory_limit * 1000**3  # args.memory_limit GB
@@ -53,9 +56,6 @@ def sample_data(
         f"Frequency step: {1/ duration if duration > 0 else 0:.3e} Hz\n"
     )
 
-    if len(t) == 0:
-        return np.empty(0), 0, " "
-
     # Calculate recommended frequency:
     if freq == -1:
         # Auto-detect frequency based on smallest time delta

ftio/freq/time_window.py

@@ -33,12 +33,20 @@ def data_in_time_window(
         indices = np.where(time_b >= args.ts)
         time_b = time_b[indices]
         bandwidth = bandwidth[indices]
-        total_bytes = int(
-            np.sum(bandwidth * (np.concatenate([time_b[1:], time_b[-1:]]) - time_b))
-        )
-        text += f"[green]Start time set to {args.ts:.2f}[/] s\n"
+
+        if len(time_b) > 0:
+            total_bytes = int(
+                np.sum(bandwidth * (np.concatenate([time_b[1:], time_b[-1:]]) - time_b))
+            )
+            text += f"[green]Start time set to {args.ts:.2f}[/] s\n"
+        else:
+            total_bytes = 0
+            text += f"[red]Warning: No data after start time {args.ts:.2f}[/] s\n"
     else:
-        text += f"Start time: [cyan]{time_b[0]:.2f}[/] s \n"
+        if len(time_b) > 0:
+            text += f"Start time: [cyan]{time_b[0]:.2f}[/] s \n"
+        else:
+            text += f"[red]Warning: No data available[/]\n"
 
     # shorten data according to end time
     if args.te:
@@ -50,7 +58,10 @@ def data_in_time_window(
         )
         text += f"[green]End time set to {args.te:.2f}[/] s\n"
     else:
-        text += f"End time: [cyan]{time_b[-1]:.2f}[/] s\n"
+        if len(time_b) > 0:
+            text += f"End time: [cyan]{time_b[-1]:.2f}[/] s\n"
+        else:
+            text += f"[red]Warning: No data in time window[/]\n"
 
     # ignored bytes
     ignored_bytes = ignored_bytes - total_bytes

ftio/parse/args.py

@@ -257,6 +257,14 @@ def parse_args(argv: list, name="") -> argparse.Namespace:
             help="specifies the number of hits needed to adapt the time window. A hit occurs once a dominant frequency is found",
         )
         parser.set_defaults(hits=3)
+        parser.add_argument(
+            "--algorithm",
+            dest="algorithm",
+            type=str,
+            choices=["adwin", "cusum", "ph"],
+            help="change point detection algorithm to use. 'adwin' (default) uses Adaptive Windowing with automatic window sizing and mathematical guarantees. 'cusum' uses Cumulative Sum detection for rapid change detection. 'ph' uses Page-Hinkley test for sequential change point detection.",
+        )
+        parser.set_defaults(algorithm="adwin")
         parser.add_argument(
             "-v",
             "--verbose",