MAG L1C: support cross-day inherited timeline (T017/T018) for IMAP-Science-Operations-Center#2925

When a MAG L1C processing window has no usable normal-mode data, build the L1C
timeline by inheriting cadence and phase from the previous day's MAG normal mode -
a real NM L1B (preferred) or an interpolated L1C - supplied as a neighbor dependency.
This covers the "time inheritance" cases T017/T018 from the SDC Data Validation
Document section 4.3.3 (issue IMAP-Science-Operations-Center#2925). Every path reachable without neighbor files is
unchanged.

- mag_l1c.py: add derive_inherited_timeline() and a phase-aligned timeline branch in
  process_mag_l1c(); mag_l1c() gains a keyword-only neighbor_datasets argument. The
  day-window computation is shared via _day_window_ns().
- cli.py: _collect_mag_l1c_inputs() partitions MAG L1C dependency files into current-day
  inputs and previous-day neighbor context, by sensor and date.
- Synthetic stand-in validation under tests/mag/validation/L1c/T017-T019, generated by
  generate_synthetic_inherited_validation.py, plus unit tests in test_mag_l1c.py and a
  validation test in test_mag_validation.py. MAG-approved validation data for these
  cases has not been delivered and is still owed; the fabricated data is clearly marked.

- T019 (no NM at all) keeps the existing day-aligned burst-only behavior. The
  validation-document rule (first BM timestamp + half the NM sampling period) conflicts
  with the behavior validated by T024 / fix IMAP-Science-Operations-Center#2274, so it is captured as an xfail test
  pending MAG clarification rather than changed here.

Companion sds-data-manager change adds date_range: ["-1d", "0d"] to the MAG L1C
norm-mode L1B dependency so the previous day is delivered.

Author: sapols

imap_processing/cli.py

@@ -1273,6 +1273,46 @@ def do_processing(
         return datasets
 
 
+def _collect_mag_l1c_inputs(
+    dependencies: ProcessingInputCollection, sensor: str, start_date: str
+) -> tuple[list[xr.Dataset], list[xr.Dataset]]:
+    """
+    Partition MAG L1C dependency files into current-day inputs and neighbor context.
+
+    Current-day L1B norm/burst files for the given sensor are the primary L1C inputs.
+    L1B and L1C files from other dates are returned as neighbor context for inheriting
+    a timeline. A same-day L1C dependency is anomalous and is ignored.
+
+    Parameters
+    ----------
+    dependencies : ProcessingInputCollection
+        Object containing dependencies to process.
+    sensor : str
+        The MAG sensor, "mago" or "magi".
+    start_date : str
+        The processing day in YYYYMMDD format.
+
+    Returns
+    -------
+    tuple[list[xr.Dataset], list[xr.Dataset]]
+        The current-day input datasets and the neighboring-day context datasets.
+    """
+    current_day_inputs: list[xr.Dataset] = []
+    neighbor_datasets: list[xr.Dataset] = []
+    for data_type in ("l1b", "l1c"):
+        for path in dependencies.get_file_paths(source="mag", data_type=data_type):
+            science_file = imap_data_access.ScienceFilePath(path.name)
+            if sensor not in science_file.descriptor:
+                continue
+            if science_file.start_date != start_date:
+                neighbor_datasets.append(load_cdf(path))
+            elif data_type == "l1b":
+                current_day_inputs.append(load_cdf(path))
+            else:
+                logger.info("Ignoring anomalous same-day MAG L1C dependency %s", path)
+    return current_day_inputs, neighbor_datasets
+
+
 class Mag(ProcessInstrument):
     """Process MAG."""
 
@@ -1345,18 +1385,37 @@ def do_processing(  # noqa: PLR0912
             ]
 
         if self.data_level == "l1c":
-            science_files = dependencies.get_file_paths(source="mag", data_type="l1b")
-            input_data = [load_cdf(dep) for dep in science_files]
-            # Input datasets can be in any order, and are validated within mag_l1c
-            if len(input_data) == 1:
-                datasets = [mag_l1c(input_data[0], current_day)]
-            elif len(input_data) == 2:
-                datasets = [mag_l1c(input_data[0], current_day, input_data[1])]
+            # MAG L1C is per-sensor; the descriptor is "norm-mago" or "norm-magi".
+            if self.descriptor.endswith("mago"):
+                sensor = "mago"
+            elif self.descriptor.endswith("magi"):
+                sensor = "magi"
             else:
                 raise ValueError(
-                    f"Invalid dependencies found for MAG L1C:"
-                    f"{dependencies}. Expected one or two dependencies."
+                    f"Unexpected MAG L1C descriptor '{self.descriptor}'; "
+                    f"expected norm-mago or norm-magi."
+                )
+            # Today's L1B norm/burst files are the primary inputs. The previous day's
+            # norm L1B/L1C files - supplied as L1C dependencies via the date_range entry
+            # in imap_mag_dependencies.yaml - provide timeline context (T017/T018) for
+            # days with no usable normal mode data.
+            current_day_inputs, neighbor_datasets = _collect_mag_l1c_inputs(
+                dependencies, sensor, self.start_date
+            )
+            if not 1 <= len(current_day_inputs) <= 2:
+                raise ValueError(
+                    f"Invalid current-day L1B dependencies found for MAG L1C: "
+                    f"{dependency_list}. Expected one or two."
                 )
+            # Datasets can be in any order, and are validated within mag_l1c.
+            datasets = [
+                mag_l1c(
+                    current_day_inputs[0],
+                    current_day,
+                    current_day_inputs[1] if len(current_day_inputs) == 2 else None,
+                    neighbor_datasets=neighbor_datasets or None,
+                )
+            ]
         if self.data_level == "l1d":
             science_files = dependencies.get_file_paths(source="mag", data_type="l1c")
             science_files.extend(

imap_processing/mag/l1c/mag_l1c.py

@@ -1,6 +1,7 @@
 """MAG L1C processing module."""
 
 import logging
+from typing import NamedTuple
 
 import numpy as np
 import xarray as xr
@@ -22,6 +23,8 @@ def mag_l1c(
     first_input_dataset: xr.Dataset,
     day_to_process: np.datetime64,
     second_input_dataset: xr.Dataset = None,
+    *,
+    neighbor_datasets: list[xr.Dataset] | None = None,
 ) -> xr.Dataset:
     """
     Will process MAG L1C data from L1A data.
@@ -40,6 +43,12 @@ def mag_l1c(
         The second input dataset to process. This should be burst if first_input_dataset
         was norm, or norm if first_input_dataset was burst. It should match the
         instrument - both inputs should be mago or magi.
+    neighbor_datasets : list[xr.Dataset], optional
+        Previous-day MAG normal mode L1B/L1C datasets, supplied as L1C dependencies via
+        the ``date_range`` entry in ``imap_mag_dependencies.yaml``. When the current day
+        has no usable normal mode data in the processing window, the L1C timeline
+        inherits its cadence and phase from a neighbor (the T017/T018 "time inheritance"
+        cases in the SDC Data Validation Document). See ``derive_inherited_timeline``.
 
     Returns
     -------
@@ -49,8 +58,8 @@ def mag_l1c(
     # TODO:
     # find missing sequences and output them
     # Missing burst file - just pass through norm file
-    # Missing norm file - go back to previous L1C file to find timestamps, then
-    # interpolate the entire day from burst
+    # (Missing norm file across a day boundary is handled below via neighbor_datasets:
+    # the timeline is inherited from the previous day's normal mode - T017/T018.)
 
     input_logical_source_1 = first_input_dataset.attrs["Logical_source"]
     if isinstance(first_input_dataset.attrs["Logical_source"], list):
@@ -63,9 +72,31 @@ def mag_l1c(
         first_input_dataset, second_input_dataset
     )
 
+    # When no usable normal mode data falls in the processing window, the L1C timeline
+    # is inherited from the previous day's MAG normal mode (T017/T018), supplied as a
+    # neighbor dependency. With no neighbor available, the burst-only fallback applies.
+    inherited_timeline = None
+    if neighbor_datasets:
+        day_start_ns, day_end_ns = _day_window_ns(day_to_process)
+        if normal_mode_dataset is None or not has_usable_norm_in_window(
+            normal_mode_dataset, day_start_ns, day_end_ns
+        ):
+            inherited_timeline = derive_inherited_timeline(neighbor_datasets)
+
     interp_function = InterpolationFunction[configuration.L1C_INTERPOLATION_METHOD]
-    if burst_mode_dataset is not None:
+    if inherited_timeline is not None and burst_mode_dataset is not None:
+        logger.info(
+            "MAG L1C inheriting timeline from %s neighbor", inherited_timeline.source
+        )
         full_interpolated_timeline: np.ndarray = process_mag_l1c(
+            None,
+            burst_mode_dataset,
+            interp_function,
+            day_to_process,
+            inherited_timeline=inherited_timeline,
+        )
+    elif burst_mode_dataset is not None:
+        full_interpolated_timeline = process_mag_l1c(
             normal_mode_dataset, burst_mode_dataset, interp_function, day_to_process
         )
     elif normal_mode_dataset is not None:
@@ -272,11 +303,142 @@ def select_datasets(
     return normal_mode_dataset, burst_mode_dataset
 
 
+class InheritedTimeline(NamedTuple):
+    """Cadence and phase inherited from a neighboring MAG normal-mode product."""
+
+    rate: int
+    anchor_ns: int
+    source: str
+
+
+def _day_window_ns(day_to_process: np.datetime64) -> tuple[int, int]:
+    """
+    Return the L1C processing window in TTJ2000 nanoseconds.
+
+    The window is the processing day extended by 30 minutes on each side.
+
+    Parameters
+    ----------
+    day_to_process : numpy.datetime64
+        The day to process, in np.datetime64[D] format.
+
+    Returns
+    -------
+    tuple[int, int]
+        The (start, end) of the processing window in TTJ2000 nanoseconds.
+    """
+    day_start = day_to_process.astype("datetime64[s]") - np.timedelta64(30, "m")
+    day_end = (
+        day_to_process.astype("datetime64[s]")
+        + np.timedelta64(1, "D")
+        + np.timedelta64(30, "m")
+    )
+    return (
+        int(et_to_ttj2000ns(str_to_et(str(day_start)))),
+        int(et_to_ttj2000ns(str_to_et(str(day_end)))),
+    )
+
+
+def has_usable_norm_in_window(
+    normal_mode_dataset: xr.Dataset, day_start_ns: int, day_end_ns: int
+) -> bool:
+    """
+    Check whether the normal mode dataset has samples inside the processing window.
+
+    Parameters
+    ----------
+    normal_mode_dataset : xarray.Dataset
+        The normal mode dataset.
+    day_start_ns : int
+        Start of the processing window, in TTJ2000 nanoseconds.
+    day_end_ns : int
+        End of the processing window, in TTJ2000 nanoseconds.
+
+    Returns
+    -------
+    bool
+        True if at least one normal mode epoch falls within the window.
+    """
+    epoch = normal_mode_dataset["epoch"].data
+    return bool(np.any((epoch >= day_start_ns) & (epoch <= day_end_ns)))
+
+
+def derive_inherited_timeline(
+    neighbor_datasets: list[xr.Dataset],
+) -> InheritedTimeline | None:
+    """
+    Derive a timeline to inherit from a neighboring day's MAG product.
+
+    When the current day has no usable normal mode data, L1C inherits the normal-mode
+    cadence and phase from a neighboring day. A real normal mode L1B product is
+    preferred over an interpolated L1C product. L1B carries its own
+    ``vectors_per_second`` attribute; L1C does not, so its cadence is derived from epoch
+    spacing.
+
+    Parameters
+    ----------
+    neighbor_datasets : list[xr.Dataset]
+        Neighboring-day MAG datasets supplied as dependencies. Normal mode L1B and L1C
+        datasets are usable; anything else is ignored.
+
+    Returns
+    -------
+    InheritedTimeline or None
+        The inherited cadence and phase, or None if no usable neighbor is found.
+    """
+    l1c: xr.Dataset | None = None
+    for dataset in neighbor_datasets:
+        epoch = dataset["epoch"].data
+        if epoch.size == 0:
+            continue
+        logical_source = dataset.attrs.get("Logical_source", "")
+        if isinstance(logical_source, list):
+            logical_source = logical_source[0]
+
+        if "l1b" in logical_source and "norm" in logical_source:
+            if "vectors_per_second" not in dataset.attrs:
+                continue
+            vecsec_dict = vectors_per_second_from_string(
+                dataset.attrs["vectors_per_second"]
+            )
+            if not vecsec_dict:
+                continue
+            rate = vecsec_dict[max(vecsec_dict)]
+            try:
+                VecSec(rate)
+            except ValueError:
+                continue
+            return InheritedTimeline(
+                rate=rate, anchor_ns=int(epoch[-1]), source="real_l1b_norm"
+            )
+        if l1c is None and "l1c" in logical_source and epoch.size >= 2:
+            l1c = dataset
+
+    if l1c is None:
+        return None
+
+    epoch = l1c["epoch"].data
+    median_spacing = float(np.median(np.diff(epoch)))
+    if median_spacing <= 0:
+        return None
+    observed_rate = 1e9 / median_spacing
+    rate = min(
+        (vec_sec.value for vec_sec in VecSec),
+        key=lambda candidate: abs(candidate - observed_rate),
+    )
+    if abs(rate - observed_rate) > rate * L1C_TIMESTAMP_GAP_TOLERANCE:
+        return None
+
+    return InheritedTimeline(rate=rate, anchor_ns=int(epoch[-1]), source="l1c")
+
+
 def process_mag_l1c(
     normal_mode_dataset: xr.Dataset | None,
     burst_mode_dataset: xr.Dataset,
     interpolation_function: InterpolationFunction,
     day_to_process: np.datetime64 | None = None,
+    *,
+    inherited_timeline: InheritedTimeline | None = None,
 ) -> np.ndarray:
     """
     Create MAG L1C data from L1B datasets.
@@ -307,6 +469,10 @@ def process_mag_l1c(
         The day to process, in np.datetime64[D] format. This is used to fill
         gaps at the beginning or end of the day if needed. If not included, these
         gaps will not be filled.
+    inherited_timeline : InheritedTimeline, optional
+        Cadence and phase inherited from a neighboring day. When provided (and there is
+        no normal mode dataset), the whole-window timeline is built at the inherited
+        cadence and phase instead of the default 2 vectors-per-second day grid.
 
     Returns
     -------
@@ -315,20 +481,19 @@ def process_mag_l1c(
     """
     day_start_ns = None
     day_end_ns = None
-
     if day_to_process is not None:
-        day_start = day_to_process.astype("datetime64[s]") - np.timedelta64(30, "m")
-
-        # get the end of the day plus 30 minutes
-        day_end = (
-            day_to_process.astype("datetime64[s]")
-            + np.timedelta64(1, "D")
-            + np.timedelta64(30, "m")
+        # Processing window: the day extended by 30 minutes on each side (see
+        # ``_day_window_ns``, also used by ``mag_l1c`` to gate timeline inheritance).
+        day_start_ns, day_end_ns = _day_window_ns(day_to_process)
+
+    if inherited_timeline is not None and day_start_ns is None:
+        # The inherited timeline is built against the day window, so day_to_process is
+        # required here. Fail clearly instead of erroring later on a None window bound.
+        raise ValueError(
+            "process_mag_l1c: inherited_timeline requires day_to_process to bound the "
+            "synthetic timeline window."
         )
 
-        day_start_ns = et_to_ttj2000ns(str_to_et(str(day_start)))
-        day_end_ns = et_to_ttj2000ns(str_to_et(str(day_end)))
-
     if normal_mode_dataset:
         norm_epoch = normal_mode_dataset["epoch"].data
         if "vectors_per_second" in normal_mode_dataset.attrs:
@@ -339,6 +504,22 @@ def process_mag_l1c(
             normal_vecsec_dict = None
 
         gaps = find_all_gaps(norm_epoch, normal_vecsec_dict, day_start_ns, day_end_ns)
+    elif inherited_timeline is not None:
+        # No usable normal mode data: build a synthetic timeline at the neighbor's
+        # cadence, phase-aligned to its anchor. interpolate_gaps() filters gap
+        # interiors strictly, so the gap must start one cadence before the first
+        # aligned timestamp or that first sample would never be interpolated.
+        period_ns = int(1e9 / inherited_timeline.rate)
+        window_start = int(np.rint(day_start_ns))
+        window_end = int(np.rint(day_end_ns))
+        first_aligned_ns = window_start + (
+            (inherited_timeline.anchor_ns - window_start) % period_ns
+        )
+        gap_start_ns = first_aligned_ns - period_ns
+        norm_epoch = [gap_start_ns, window_end]
+        gaps = np.array(
+            [[gap_start_ns, window_end, inherited_timeline.rate]], dtype=np.int64
+        )
     else:
         norm_epoch = [day_start_ns, day_end_ns]
         gaps = np.array(