3325 lo pivot platform ck generation

imap_processing/cli.py

@@ -67,6 +67,7 @@
 from imap_processing.idex.idex_l1b import idex_l1b
 from imap_processing.idex.idex_l2a import idex_l2a
 from imap_processing.idex.idex_l2b import idex_l2b
+from imap_processing.lo import lo_pivot_kernel
 from imap_processing.lo.constants import LoConstants
 from imap_processing.lo.l1a import lo_l1a
 from imap_processing.lo.l1b import lo_l1b
@@ -474,6 +475,33 @@ def _resolve_version(self, descriptor: str) -> Version:
             logger.warning(msg)
         return self.version_map.get(descriptor, self._fallback_version)
 
+    def _resolve_kernel_minor_version(self, descriptor: str) -> int:
+        """
+        Return the minor version to use in the filename of a generated kernel.
+
+        Parameters
+        ----------
+        descriptor : str
+            The descriptor of the kernel job, e.g. "pointing-attitude".
+
+        Returns
+        -------
+        int
+            The minor version for the kernel filename.
+        """
+        resolved_version = self._resolve_version(descriptor)
+        if resolved_version is None:
+            raise ValueError(
+                f"No version provided for {descriptor} processing. "
+                f"Provide a version for the '{descriptor}' descriptor in "
+                "the dependency JSON's version block, or a fallback --version."
+            )
+        return (
+            resolved_version.minor
+            if isinstance(resolved_version, Version)
+            else int(resolved_version.lstrip("v"))
+        )
+
     def upload_products(self, products: list[Path]) -> None:
         """
         Upload data products to the IMAP SDC.
@@ -1431,9 +1459,48 @@ def pre_processing(self) -> ProcessingInputCollection:
 
         return filtered_dependencies
 
+    def _generate_pivot_kernel(
+        self, dependencies: ProcessingInputCollection
+    ) -> list[Path]:
+        """
+        Generate the Lo pivot platform CK for the pointing given by repointing.
+
+        Parameters
+        ----------
+        dependencies : ProcessingInputCollection
+            Object containing dependencies to process.
+
+        Returns
+        -------
+        list[Path]
+            The generated Lo pivot kernel.
+        """
+        if self.repointing is None:
+            raise ValueError(
+                "repointing must be provided for pivot-ckernel processing."
+            )
+        nhk_files = dependencies.get_file_paths(
+            source="lo", data_type="l1b", descriptor="nhk"
+        )
+        if len(nhk_files) != 1:
+            raise ValueError(
+                f"Unexpected dependencies found for IMAP-Lo pivot-ckernel: "
+                f"{nhk_files}. Expected exactly one L1B NHK file."
+            )
+        # The repoint table provides the pointing start and end times.
+        if not dependencies.get_file_paths(data_type=RepointInput.data_type):
+            raise ValueError(
+                "A repoint table dependency is required for IMAP-Lo pivot-ckernel "
+                "processing."
+            )
+        minor_version = self._resolve_kernel_minor_version(self.descriptor)
+        return lo_pivot_kernel.generate_lo_pivot_kernel(
+            nhk_files[0], self.repointing, minor_version
+        )
+
     def do_processing(
         self, dependencies: ProcessingInputCollection
-    ) -> list[xr.Dataset]:
+    ) -> list[xr.Dataset | Path]:
         """
         Perform IMAP-Lo specific processing.
 
@@ -1444,12 +1511,15 @@ def do_processing(
 
         Returns
         -------
-        dataset : xr.Dataset
-            Xr.Dataset of output files.
+        datasets : list[xarray.Dataset | Path]
+            The list of processed products.
         """
         print(f"Processing IMAP-Lo {self.data_level}")
-        datasets: list[xr.Dataset] = []
-        if self.data_level == "l1a":
+        datasets: list[xr.Dataset | Path] = []
+        if self.data_level == "l1b" and self.descriptor == "pivot-ckernel":
+            datasets.extend(self._generate_pivot_kernel(dependencies))
+
+        elif self.data_level == "l1a":
             # L1A packet / products are 1 to 1. Should only have
             # one dependency file
             science_files = dependencies.get_file_paths(source="lo", data_type="l0")
@@ -1860,18 +1930,7 @@ def do_processing(
                 data_type=SPICESource.SPICE.value
             )
             ah_paths = [path for path in spice_inputs if ".ah" in path.suffixes]
-            resolved_version = self._resolve_version(self.descriptor)
-            if resolved_version is None:
-                raise ValueError(
-                    "No version provided for pointing-attitude processing. "
-                    "Provide a version for the 'pointing-attitude' descriptor in "
-                    "the dependency JSON's version block, or a fallback --version."
-                )
-            minor_version = (
-                resolved_version.minor
-                if isinstance(resolved_version, Version)
-                else int(resolved_version.lstrip("v"))
-            )
+            minor_version = self._resolve_kernel_minor_version(self.descriptor)
             pointing_kernel_paths = pointing_frame.generate_pointing_attitude_kernel(
                 ah_paths, self.start_date, minor_version
             )

imap_processing/lo/l1b/lo_l1b.py

@@ -1862,6 +1862,44 @@ def get_pivot_angle_from_nhk(ds_nhk: xr.Dataset) -> float:
     return ds_nhk["pcc_cumulative_cnt_pri"].isel(epoch=nitems // 2).item()
 
 
+def get_median_pivot_angle(ds_nhk: xr.Dataset) -> float:
+    """
+    Get the median pivot angle from the NHK dataset.
+
+    The median of ``pcc_coarse_pot_pri`` is taken over the samples between
+    ``PIVOT_HK_HOUR_RANGE`` hours after the first NHK sample, which avoids the
+    pivot platform motion at the start of a pointing.
+
+    Parameters
+    ----------
+    ds_nhk : xr.Dataset
+        The NHK dataset containing pivot angle information.
+
+    Returns
+    -------
+    pivot_angle : float
+        The median pivot angle [degrees], or NaN if the dataset has no records
+        or there are no valid samples within the time range.
+    """
+    if ds_nhk.sizes.get("epoch", 0) == 0:
+        return np.nan
+
+    hk_epoch_ets = ttj2000ns_to_et(ds_nhk["epoch"])
+    start_et_hk = (
+        hk_epoch_ets[0] + timedelta(hours=c.PIVOT_HK_HOUR_RANGE[0]).total_seconds()
+    )
+    end_et_hk = (
+        hk_epoch_ets[0] + timedelta(hours=c.PIVOT_HK_HOUR_RANGE[1]).total_seconds()
+    )
+
+    coarse_pot_pri = ds_nhk["pcc_coarse_pot_pri"].values
+    return float(
+        np.nanmedian(
+            coarse_pot_pri[(hk_epoch_ets >= start_et_hk) & (hk_epoch_ets <= end_et_hk)]
+        )
+    )
+
+
 def _get_esa_level_indices(epochs: np.ndarray, anc_dependencies: list) -> np.ndarray:
     """
     Get the ESA level indices (reswept indices) for the given epochs.
@@ -2485,18 +2523,7 @@ def l1b_bgrates_and_goodtimes(  # noqa: PLR0912
     pivot: float = 90.0
     cdf_hk = sci_dependencies.get("imap_lo_l1b_nhk")
     if cdf_hk is not None and "pcc_coarse_pot_pri" in cdf_hk:
-        hk_epoch_ets = ttj2000ns_to_et(cdf_hk["epoch"])
-        start_et_hk = (
-            hk_epoch_ets[0] + timedelta(hours=c.PIVOT_HK_HOUR_RANGE[0]).total_seconds()
-        )
-        end_et_hk = (
-            hk_epoch_ets[0] + timedelta(hours=c.PIVOT_HK_HOUR_RANGE[1]).total_seconds()
-        )
-
-        coarse_pot_pri = cdf_hk["pcc_coarse_pot_pri"].values
-        pivot = np.nanmedian(  # type: ignore
-            coarse_pot_pri[(hk_epoch_ets >= start_et_hk) & (hk_epoch_ets <= end_et_hk)]
-        )
+        pivot = get_median_pivot_angle(cdf_hk)
         if np.isnan(pivot):
             pivot = 90.0