Support new versioning scheme

imap_l3_data_processor.py

@@ -1,11 +1,13 @@
 import argparse
+import json
 import logging
 import re
 from datetime import datetime
 from tempfile import TemporaryDirectory
 
 import imap_data_access
 import spiceypy
+from imap_data_access.file_validation import Version
 from imap_data_access.processing_input import ProcessingInputCollection
 
 from imap_l3_processing.codice.l3.hi.codice_hi_processor import CodiceHiProcessor
@@ -18,7 +20,7 @@
 from imap_l3_processing.lo.l3.lo_sp_initializer import LoSPInitializer, LO_SP_MAP_DESCRIPTORS
 from imap_l3_processing.lo.lo_processor import LoProcessor
 from imap_l3_processing.maps.map_descriptors import parse_map_descriptor
-from imap_l3_processing.models import InputMetadata
+from imap_l3_processing.models import InputMetadata, VersionMap
 from imap_l3_processing.swapi.swapi_processor import SwapiProcessor
 from imap_l3_processing.swe.swe_processor import SweProcessor
 from imap_l3_processing.ultra.ultra_combined_nsp_initializer import UltraCombinedNSPInitializer, \
@@ -38,7 +40,7 @@ def _parse_cli_arguments():
     parser.add_argument("--start-date")
     parser.add_argument("--end-date", required=False)
     parser.add_argument("--repointing", required=False)
-    parser.add_argument("--version")
+    parser.add_argument("--version", required=False)
     parser.add_argument("--dependency")
     parser.add_argument(
         "--upload-to-sdc",
@@ -67,7 +69,14 @@ def imap_l3_processor():
             args.dependency = f.read()
 
     processing_input_collection = ProcessingInputCollection()
-    processing_input_collection.deserialize(args.dependency)
+
+    parsed_dependency = json.loads(args.dependency)
+    if isinstance(parsed_dependency, list):
+        processing_input_collection.deserialize(args.dependency)
+        version_map = VersionMap({}, Version(None, args.version))
+    else:
+        processing_input_collection.deserialize(json.dumps(parsed_dependency["dependency"]))
+        version_map = VersionMap({k: Version(v["major_version"], v["minor_version"]) for k, v in parsed_dependency["version"].items()})
 
     repointing_number = None
     if args.repointing is not None:
@@ -81,7 +90,7 @@ def imap_l3_processor():
                                      args.data_level,
                                      _convert_to_datetime(args.start_date),
                                      _convert_to_datetime(args.end_date or args.start_date),
-                                     args.version, descriptor=args.descriptor, repointing=repointing_number)
+                                     version_map, descriptor=args.descriptor, repointing=repointing_number)
     if args.instrument in ["hi", "lo", "ultra"] and args.data_level == 'l3' and not parse_map_descriptor(
             args.descriptor):
         initializer_class, processor_class, descriptors = {
@@ -98,7 +107,8 @@ def imap_l3_processor():
         paths = []
         maps_to_produce = []
         for map_descriptor in descriptors:
-            maps_to_produce.extend(initializer.get_maps_that_should_be_produced(map_descriptor))
+            major_version_from_dependency = version_map.lookup(map_descriptor).major
+            maps_to_produce.extend(initializer.get_maps_that_should_be_produced(map_descriptor, major_version_from_dependency))
 
         logger.info(f"maps to produce {[m.input_metadata.descriptor for m in maps_to_produce]}")
         if len(maps_to_produce) == 0:

imap_l3_processing/glows/descriptors.py

@@ -26,3 +26,22 @@
 GLOWS_L3E_LO_DESCRIPTOR = 'survival-probability-lo'
 GLOWS_L3E_ULTRA_SF_DESCRIPTOR = 'survival-probability-ul-sf'
 GLOWS_L3E_ULTRA_HF_DESCRIPTOR = 'survival-probability-ul-hf'
+
+GLOWS_L3BCDE_DESCRIPTORS = [
+    GLOWS_L3B_DESCRIPTOR,
+    GLOWS_L3C_DESCRIPTOR,
+    GLOWS_L3D_DESCRIPTOR,
+    GLOWS_L3E_HI_45_DESCRIPTOR,
+    GLOWS_L3E_HI_90_DESCRIPTOR,
+    GLOWS_L3E_LO_DESCRIPTOR,
+    GLOWS_L3E_ULTRA_SF_DESCRIPTOR,
+    GLOWS_L3E_ULTRA_HF_DESCRIPTOR
+]
+
+GLOWS_L3E_DESCRIPTORS = [
+    GLOWS_L3E_HI_45_DESCRIPTOR,
+    GLOWS_L3E_HI_90_DESCRIPTOR,
+    GLOWS_L3E_LO_DESCRIPTOR,
+    GLOWS_L3E_ULTRA_SF_DESCRIPTOR,
+    GLOWS_L3E_ULTRA_HF_DESCRIPTOR,
+]

imap_l3_processing/glows/l3bc/glows_l3bc_dependencies.py

@@ -3,14 +3,15 @@
 from pathlib import Path
 
 import imap_data_access
+from imap_data_access.file_validation import Version
 
 from imap_l3_processing.glows.l3a.utils import create_glows_l3a_dictionary_from_cdf
 from imap_l3_processing.glows.l3bc.models import CRToProcess, ExternalDependencies
 
 
 @dataclass
 class GlowsL3BCDependencies:
-    version: int
+    version: Version
     carrington_rotation_number: int
     start_date: datetime
     end_date: datetime
@@ -24,7 +25,7 @@ def l3a_file_names(self):
         return [l3a['filename'] for l3a in self.l3a_data]
 
     @classmethod
-    def download_from_cr_to_process(cls, cr_to_process: CRToProcess, version: int,
+    def download_from_cr_to_process(cls, cr_to_process: CRToProcess, version: Version,
                                     external_dependencies: ExternalDependencies, repointing_file_path: Path):
         external_files = {
             'f107_raw_data': external_dependencies.f107_index_file_path,

imap_l3_processing/glows/l3bc/glows_l3bc_initializer.py

@@ -6,6 +6,7 @@
 
 import imap_data_access
 from imap_data_access import ScienceFilePath, ProcessingInputCollection, RepointInput
+from imap_data_access.file_validation import Version
 from imap_processing.spice.repoint import set_global_repoint_table_paths
 from spacepy.pycdf import CDF
 
@@ -37,7 +38,7 @@ class GlowsL3BCInitializerData:
 
 class GlowsL3BCInitializer:
     @staticmethod
-    def get_crs_to_process(dependencies: ProcessingInputCollection) -> GlowsL3BCInitializerData:
+    def get_crs_to_process(dependencies: ProcessingInputCollection, major_version_to_process: int) -> GlowsL3BCInitializerData:
         [repoint_file] = dependencies.get_file_paths(data_type=RepointInput.data_type)
         repoint_downloaded_path = imap_data_access.download(repoint_file)
         set_global_repoint_table_paths([repoint_downloaded_path])
@@ -108,7 +109,7 @@ def get_crs_to_process(dependencies: ProcessingInputCollection) -> GlowsL3BCInit
                 )
 
                 if version := GlowsL3BCInitializer.should_process_cr_candidate(cr_candidate, l3bs_by_cr,
-                                                                               external_dependencies):
+                                                                               external_dependencies, major_version_to_process):
                     l3bc_dependencies = GlowsL3BCDependencies.download_from_cr_to_process(cr_candidate, version,
                                                                                           external_dependencies,
                                                                                           repoint_downloaded_path)
@@ -128,7 +129,7 @@ def get_crs_to_process(dependencies: ProcessingInputCollection) -> GlowsL3BCInit
 
     @staticmethod
     def should_process_cr_candidate(cr_candidate: CRToProcess, l3bs_by_cr: dict[int, str],
-                                    external_dependencies: ExternalDependencies) -> Optional[int]:
+                                    external_dependencies: ExternalDependencies, major_version: int|None) -> Optional[Version]:
         if not cr_candidate.buffer_time_has_elapsed_since_cr():
             logger.warning(f"Not enough time has elapsed for cr {cr_candidate.cr_rotation_number}")
             return None
@@ -139,11 +140,14 @@ def should_process_cr_candidate(cr_candidate: CRToProcess, l3bs_by_cr: dict[int,
 
         match l3bs_by_cr.get(cr_candidate.cr_rotation_number):
             case None:
-                return 1
+                return Version(major_version, 1)
             case l3b_file_name:
                 l3b_parents = read_cdf_parents(l3b_file_name)
-                if not cr_candidate.pipeline_dependency_file_names().issubset(l3b_parents):
-                    return int(ScienceFilePath(l3b_file_name).version[1:]) + 1
+                existing_version = Version.from_version(ScienceFilePath(l3b_file_name).version)
+                major_version_updated = major_version != existing_version.major
+                inputs_changed = not cr_candidate.pipeline_dependency_file_names().issubset(l3b_parents)
+                if major_version_updated or inputs_changed:
+                    return Version(major_version, existing_version.minor+1)
         return None
 
     @staticmethod

imap_l3_processing/glows/l3d/glows_l3d_initializer.py

@@ -4,6 +4,7 @@
 
 import imap_data_access
 from imap_data_access import ProcessingInputCollection, ScienceInput, AncillaryInput, ScienceFilePath
+from imap_data_access.file_validation import Version
 
 from imap_l3_processing.glows.descriptors import PLASMA_SPEED_DESCRIPTOR, PROTON_DENSITY_DESCRIPTOR, \
     UV_ANISOTROPY_DESCRIPTOR, PHOTOION_DESCRIPTOR, LYA_DESCRIPTOR, ELECTRON_DENSITY_DESCRIPTOR, \
@@ -19,8 +20,8 @@
 class GlowsL3DInitializer:
 
     @staticmethod
-    def should_process_l3d(external_deps: ExternalDependencies, l3bs: list[str], l3cs: list[str]) -> Optional[
-        tuple[int, GlowsL3DDependencies, Optional[Path]]]:
+    def should_process_l3d(external_deps: ExternalDependencies, l3bs: list[str], l3cs: list[str], major_version: int|None) -> Optional[
+        tuple[Version, GlowsL3DDependencies, Optional[str]]]:
         if len(l3bs) == 0 and len(l3cs) == 0:
             logger.info("Found no L3b and L3c files!")
             return None
@@ -106,12 +107,18 @@ def should_process_l3d(external_deps: ExternalDependencies, l3bs: list[str], l3c
             old_l3d = Path(most_recent_l3d["file_path"]).name
 
             logger.info(f"Old L3d parents: {l3d_parents}, new L3d deps: {updated_input_files}")
-            if updated_input_files.issubset(l3d_parents):
+            if 'major_version' in most_recent_l3d and 'minor_version' in most_recent_l3d:
+                most_recent_l3d_version = Version(most_recent_l3d['major_version'], most_recent_l3d['major_version'])
+            else:
+                most_recent_l3d_version = Version.from_version(most_recent_l3d['version'])
+            same_major_version = most_recent_l3d_version.major == major_version
+            if same_major_version and updated_input_files.issubset(l3d_parents):
                 return None
-            version_to_generate = int(most_recent_l3d['version'][1:]) + 1
+            minor_version_to_generate = most_recent_l3d_version.minor + 1
         else:
             old_l3d = None
-            version_to_generate = 1
+            minor_version_to_generate = 1
+        version_to_generate = Version(major_version, minor_version_to_generate)
 
         return (version_to_generate,
                 GlowsL3DDependencies.fetch_dependencies(processing_input_collection, external_deps),

imap_l3_processing/glows/l3d/utils.py

@@ -6,6 +6,7 @@
 
 import imap_data_access
 import numpy as np
+from imap_data_access.file_validation import Version
 from spacepy.pycdf import CDF
 
 import imap_l3_processing
@@ -142,5 +143,11 @@ def rename_l3d_text_outputs(paths: list[Path], version: str) -> list[Path]:
 
 def query_for_most_recent_l3d(descriptor: str) -> Optional[dict]:
     query_result = imap_data_access.query(instrument="glows", data_level="l3d", descriptor=descriptor)
-    sorted_query_result = sorted(query_result, key=lambda qr: (qr["cr"], qr["version"]), reverse=True)
+    sorted_query_result = sorted(
+        query_result,
+        key=lambda qr: (
+            qr["cr"],
+            (qr["minor_version"] if "minor_version" in qr else Version.from_version(qr["version"]).minor)
+        ),
+        reverse=True)
     return next(iter(sorted_query_result), None)

imap_l3_processing/glows/l3e/glows_l3e_initializer.py

@@ -6,32 +6,33 @@
 import imap_data_access
 from imap_data_access import ProcessingInputCollection, AncillaryInput, ScienceInput, ScienceFilePath, SPICEInput, \
     RepointInput
+from imap_data_access.file_validation import Version
 
 from imap_l3_processing.glows.l3bc.utils import get_pointing_date_range
 from imap_l3_processing.glows.l3d.models import GlowsL3DProcessorOutput
 from imap_l3_processing.glows.l3d.utils import get_most_recently_uploaded_ancillary
 from imap_l3_processing.glows.l3e.glows_l3e_dependencies import GlowsL3EDependencies
-from imap_l3_processing.glows.l3e.glows_l3e_utils import find_first_updated_cr, determine_l3e_files_to_produce, \
-    GlowsL3eRepointings
+from imap_l3_processing.glows.l3e.glows_l3e_utils import find_first_updated_cr, identify_versions_for_l3e_output_files, \
+    GlowsL3eVersionsForRepointings
+from imap_l3_processing.models import VersionMap
 
 logger = logging.getLogger(__name__)
 
 @dataclass
 class GlowsL3EInitializerOutput:
     dependencies: GlowsL3EDependencies
-    repointings: GlowsL3eRepointings
+    repointings: GlowsL3eVersionsForRepointings
     l3d_cdf_path: Path
 
 
 class GlowsL3EInitializer:
     @staticmethod
-    def get_repointings_to_process(l3d_output: GlowsL3DProcessorOutput, previous_l3d: Optional[str], repointing_file_path: Path) -> Optional[GlowsL3EInitializerOutput]:
-        latest_l3d_cr = None
-        if previous_l3d is not None:
-            latest_l3d_cr = find_first_updated_cr(l3d_output.l3d_cdf_file_path, previous_l3d)
-            if not latest_l3d_cr:
-                return None
-
+    def get_repointings_to_process(
+            l3d_output: GlowsL3DProcessorOutput,
+            previous_l3d: Optional[str],
+            repointing_file_path: Path,
+            version_map: VersionMap,
+    ) -> Optional[GlowsL3EInitializerOutput]:
         pipeline_settings_l3bcde = get_most_recently_uploaded_ancillary(imap_data_access.query(table='ancillary', instrument='glows', descriptor='pipeline-settings-l3bcde'))
         energy_grid_lo = get_most_recently_uploaded_ancillary(imap_data_access.query(table='ancillary', instrument='glows', descriptor='energy-grid-lo'))
         tess_xyz_8 = get_most_recently_uploaded_ancillary(imap_data_access.query(table='ancillary', instrument='glows', descriptor='tess-xyz-8'))
@@ -55,11 +56,15 @@ def get_repointings_to_process(l3d_output: GlowsL3DProcessorOutput, previous_l3d
         l3e_deps.copy_dependencies()
 
         first_cr = l3e_deps.pipeline_settings["start_cr"]
-        if latest_l3d_cr is not None:
-            first_cr = max(first_cr, latest_l3d_cr - 1)
 
-        last_cr = ScienceFilePath(l3d_output.l3d_cdf_file_path).cr
-        glows_repointings = determine_l3e_files_to_produce(first_cr, last_cr, repointing_file_path)
+        first_updated_cr = first_cr
+        if previous_l3d is not None:
+            first_updated_cr = find_first_updated_cr(l3d_output.l3d_cdf_file_path, previous_l3d)
+            if first_updated_cr is not None:
+                first_updated_cr -= 1
+
+        last_cr = l3d_output.last_processed_cr
+        glows_repointings = identify_versions_for_l3e_output_files(first_cr, last_cr, first_updated_cr, repointing_file_path, version_map)
 
         if len(glows_repointings.repointing_numbers) > 0:
             earliest_repointing_start, _ = get_pointing_date_range(min(glows_repointings.repointing_numbers))