Implemented Qartod Timing/Gap Test

ioos_qc/timing_gap.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python
+"""Timing/Gap Test based on the IOOS QARTOD manuals."""
+
+import logging
+import warnings
+from collections import namedtuple
+from numbers import Real as N
+from typing import Dict, List, Optional, Sequence, Tuple, Union
+from datetime import datetime, timedelta
+from enum import IntEnum
+import numpy as np
+import pandas as pd
+
+try:
+    from numba.core.errors import NumbaTypeError
+except ImportError:
+    NumbaTypeError = TypeError
+
+from ioos_qc.utils import (
+    add_flag_metadata,
+    great_circle_distance,
+    isfixedlength,
+    isnan,
+    mapdates,
+)
+
+L = logging.getLogger(__name__)
+
+
+class QartodFlags:
+    """Primary flags for QARTOD."""
+
+    GOOD = 1
+    UNKNOWN = 2
+    SUSPECT = 3
+    FAIL = 4
+    MISSING = 9
+
+
+FLAGS = QartodFlags  # Default name for all check modules
+NOTEVAL_VALUE = QartodFlags.UNKNOWN
+
+def timing_gap_test(tim_stmp: float, tim_inc: float) -> np.ndarray:
+    """
+    Timing/Gap Test checks if the data has arrived within the expected time window.
+
+    Parameters
+    ----------
+    tim_stmp : float
+        Timestamp of the most recent data.
+    tim_inc : float
+        Allowed time increment or window for data to arrive (in seconds).
+
+    Returns
+    -------
+    flag_arr : np.ndarray
+        An array with the flag, 1 for Pass, 4 for Fail.
+    """
+
+    # Get the current timestamp
+    now = datetime.now().timestamp()
+
+    # Calculate the time difference between the current time and the data's timestamp
+    time_diff = now - tim_stmp
+
+    # Initialize flag array with a passing value
+    flag_arr = np.ma.ones(1, dtype="uint8")
+
+    # If the time difference exceeds the allowed increment, flag as Fail
+    if time_diff > tim_inc:
+        flag_arr[0] = QartodFlags.FAIL
+
+    return flag_arr

tests/test_timing_gap.py

@@ -0,0 +1,39 @@
+import logging
+import unittest
+import warnings
+
+import numpy as np
+import numpy.testing as npt
+import pandas as pd
+import pytest
+from datetime import datetime, timedelta
+
+from ioos_qc import timing_gap
+
+L = logging.getLogger("ioos_qc")
+L.setLevel(logging.INFO)
+L.handlers = [logging.StreamHandler()]
+
+
+class QartodTimingTest(unittest.TestCase):
+    def test_timing_gap_test(self):
+        # Example 1: Data is within the expected time window
+        tim_stmp = datetime.now().timestamp() - 3000 
+        tim_inc = 3600  # 1 hour window
+
+        flag_arr = timing_gap.timing_gap_test(tim_stmp, tim_inc)
+        self.assertEqual(flag_arr[0], timing_gap.QartodFlags.GOOD, "Timing test should pass when data is within window.")
+
+        # Example case 2: Data is outside the expected time window 
+        tim_stmp = datetime.now().timestamp() - 7200 
+        tim_inc = 3600  # 1 hour window
+
+        flag_arr = timing_gap.timing_gap_test(tim_stmp, tim_inc)
+        self.assertEqual(flag_arr[0], timing_gap.QartodFlags.FAIL, "Timing test should fail when data is outside the window.")
+
+         # Example case 3: (should fail)
+        tim_stmp = datetime(2023, 2, 1, 12, 0, 0).timestamp()
+        tim_inc = 3600 * 24 * 30  # 30-day window
+
+        flag_arr = timing_gap.timing_gap_test(tim_stmp, tim_inc)
+        self.assertEqual(flag_arr[0], timing_gap.QartodFlags.FAIL, "Timing test should fail when data is beyond the 30-day window.")