Glows time flags

imap_processing/glows/l1b/glows_l1b_data.py

@@ -653,6 +653,31 @@ def process_direct_events(direct_events: np.ndarray) -> tuple:
         return times, pulse_lengths
 
 
+def get_threshold(thresholds: dict, suffix: str) -> float | None:
+    """
+    Return the threshold value whose key ends with the given suffix.
+
+    Parameters
+    ----------
+    thresholds : dict
+        Dictionary of threshold values.
+    suffix : str
+        The suffix to match against threshold keys.
+
+    Returns
+    -------
+    return_value : float or None
+        The matching threshold value, or None if no match is found.
+    """
+    return_value = None
+    for descriptor, value in thresholds.items():
+        if descriptor.endswith(suffix):
+            return_value = float(value)
+            break
+
+    return return_value
+
+
 @dataclass
 class HistogramL1B:
     """
@@ -870,7 +895,7 @@ def __post_init__(
         # is_inside_excluded_region, is_excluded_by_instr_team,
         # is_suspected_transient] x 3600 bins
         self.histogram_flag_array = self._compute_histogram_flag_array(day_exclusions)
-        self.flags = np.ones((FLAG_LENGTH,), dtype=np.uint8)
+        self.flags = self.compute_flags(pipeline_settings)
 
     def update_spice_parameters(self) -> None:
         """Update SPICE parameters based on the current state."""
@@ -979,6 +1004,70 @@ def deserialize_flags(raw: int) -> np.ndarray[int]:
 
         return flags
 
+    def compute_flags(self, pipeline_settings: PipelineSettings) -> np.ndarray:
+        """
+        Compute the 17 bad-time flags for this histogram.
+
+        Parameters
+        ----------
+        pipeline_settings : PipelineSettings
+            Pipeline settings containing processing thresholds.
+
+        Returns
+        -------
+        flags : numpy.ndarray
+            Array of shape (FLAG_LENGTH,) with dtype uint8. 1 = good, 0 = bad.
+        """
+        thresholds = pipeline_settings.processing_thresholds
+
+        # Section 12.3.1 of the Algorithm Document: onboard generated bad-time flags.
+        # Flags are "stored in a 16-bit integer field.
+        onboard_flags = (
+            1 - self.deserialize_flags(int(self.flags_set_onboard))
+        ).astype(np.uint8)
+
+        # Section 12.3.2 of the Algorithm Document: ground processing flags: flag 1.
+        # Informs if the histogram was generated on-board or on the ground.
+        # Flag 1 = onboard.
+        is_generated_on_ground = np.uint8(1 - int(self.is_generated_on_ground))
+
+        # Section 12.3.2 of the Algorithm Document: ground processing flags: flag 2.
+        # Checks if total count in a given histogram is far from the daily average.
+        # Placeholder until daily histogram is available in glows_l1b.py.
+        # TODO: this equation needs to be clarified.
+        is_beyond_daily_statistical_error = np.uint8(1)
+
+        # Section 12.3.2 of the Algorithm Document: ground processing flags: flag 3-7.
+        # (1=good, 0=bad).
+        temp_threshold = get_threshold(thresholds, "std_dev_threshold__celsius_deg")
+        hv_threshold = get_threshold(thresholds, "std_dev_threshold__volt")
+        spin_std_threshold = get_threshold(thresholds, "std_dev_threshold__sec")
+        pulse_threshold = get_threshold(thresholds, "std_dev_threshold__usec")
+
+        is_temp_ok = np.uint8(self.filter_temperature_std_dev <= temp_threshold)
+        is_hv_ok = np.uint8(self.hv_voltage_std_dev <= hv_threshold)
+        is_spin_std_ok = np.uint8(self.spin_period_std_dev <= spin_std_threshold)
+        is_pulse_ok = np.uint8(self.pulse_length_std_dev <= pulse_threshold)
+
+        # TODO: listed as TBC in Algorithm Document.
+        # Placeholder for now.
+        is_beyond_background_error = np.uint8(1)
+
+        ground_flags = np.array(
+            [
+                is_generated_on_ground,
+                is_beyond_daily_statistical_error,
+                is_temp_ok,
+                is_hv_ok,
+                is_spin_std_ok,
+                is_pulse_ok,
+                is_beyond_background_error,
+            ],
+            dtype=np.uint8,
+        )
+
+        return np.concatenate([onboard_flags, ground_flags])
+
     def flag_uv_and_excluded(self, exclusions: AncillaryExclusions) -> tuple:
         """
         Create boolean mask where True means bin is within radius of UV source.

imap_processing/tests/glows/test_glows_l1b.py

@@ -299,19 +299,19 @@ def test_process_histogram(
         0,
         0,
         0,
-        0,
-        0,
+        64,  # flags_set_onboard: bit 6 (is_night) set
+        1,  # is_generated_on_ground
         0,
         3600,
         0,
         encoded_val,
+        np.single(30.0),  # filter_temperature_variance: exceeds 2.03°C threshold
         encoded_val,
+        np.single(3500.0),  # hv_voltage_variance: exceeds 50.0V threshold
         encoded_val,
+        np.single(11000.0),  # spin_period_variance: exceeds 0.033333s threshold
         encoded_val,
-        encoded_val,
-        encoded_val,
-        encoded_val,
-        encoded_val,
+        np.single(2.0),  # pulse_length_variance: exceeds 1.0μs threshold
         time_val,
         time_val,
         time_val,
@@ -329,6 +329,19 @@ def test_process_histogram(
     )
     assert len(output) == len(dataclasses.asdict(test_l1b))
 
+    # flags[0:10]  = onboard flags (1=good, 0=bad), one per bit of flags_set_onboard
+    # flags[10]    = is_generated_on_ground (1=onboard, 0=ground)
+    # flags[11]    = is_beyond_daily_statistical_error (placeholder, always 1)
+    # flags[12:16] = std_dev threshold flags
+    # flags[16]    = is_beyond_background
+    assert test_l1b.flags[6] == 0  # is_night
+    assert test_l1b.flags[10] == 0  # is_generated_on_ground
+    assert test_l1b.flags[12] == 0  # is_temp_ok
+    assert test_l1b.flags[13] == 0  # is_hv_ok
+    assert test_l1b.flags[14] == 0  # is_spin_std_ok
+    assert test_l1b.flags[15] == 0  # is_pulse_ok
+    assert test_l1b.flags[16] == 1  # is_beyond_background
+
 
 @patch.object(
     HistogramL1B,

imap_processing/tests/glows/test_glows_l1b_data.py

@@ -12,6 +12,7 @@
     DirectEventL1B,
     HistogramL1B,
     PipelineSettings,
+    get_threshold,
 )
 from imap_processing.spice.time import met_to_ttj2000ns
 from imap_processing.tests.glows.conftest import mock_update_spice_parameters
@@ -287,3 +288,30 @@ def test_pipeline_settings_from_flattened_json():
 
     assert len(settings.active_bad_angle_flags) == 4
     assert settings.active_bad_angle_flags[3] is False  # is_suspected_transient
+
+
+def test_get_threshold():
+    "Test get_threshold function."
+
+    test_data = {
+        "n_sigma_threshold_lower": 3.0,
+        "n_sigma_threshold_upper": 3.0,
+        "relative_difference_threshold": 7e-05,
+        "std_dev_threshold__celsius_deg": 2.03,
+        "std_dev_threshold__volt": 50.0,
+        "std_dev_threshold__sec": 0.033333,
+        "std_dev_threshold__usec": 1.0,
+    }
+
+    expected = [2.03, 50.0, 0.033333, 1.0, 7e-5]
+    description = [
+        "std_dev_threshold__celsius_deg",
+        "std_dev_threshold__volt",
+        "std_dev_threshold__sec",
+        "std_dev_threshold__usec",
+        "relative_difference_threshold",
+    ]
+
+    for name, exp in zip(description, expected, strict=False):
+        threshold = get_threshold(test_data, name)
+        assert threshold == exp