diff --git a/imap_processing/ena_maps/ena_maps.py b/imap_processing/ena_maps/ena_maps.py index e2416542f..a31454f0b 100644 --- a/imap_processing/ena_maps/ena_maps.py +++ b/imap_processing/ena_maps/ena_maps.py @@ -1498,11 +1498,25 @@ def build_cdf_dataset( # noqa: PLR0912 # Now set global attributes map_attrs = cdf_attrs.get_global_attributes(f"imap_{instrument}_{level}_enamap") map_attrs["Spacing_degrees"] = str(self.spacing_deg) - for key in ["Data_type", "Logical_source", "Logical_source_description"]: + for key in ["Data_type", "Logical_source"]: map_attrs[key] = map_attrs[key].format( descriptor=descriptor, sensor=sensor, ) + # Use the MapDescriptor to generate the Logical_source_description + # when possible, but preserve previous behavior for non-descriptor + # strings so later validation still raises the intended errors. + try: + md = naming.MapDescriptor.from_string(descriptor) + except ValueError: + map_attrs["Logical_source_description"] = map_attrs[ + "Logical_source_description" + ].format( + descriptor=descriptor, + sensor=sensor, + ) + else: + map_attrs["Logical_source_description"] = md.to_logical_source_description() # Always add the following attributes to the map map_attrs.update( { diff --git a/imap_processing/ena_maps/utils/naming.py b/imap_processing/ena_maps/utils/naming.py index 191239d2c..f883a962e 100644 --- a/imap_processing/ena_maps/utils/naming.py +++ b/imap_processing/ena_maps/utils/naming.py @@ -231,6 +231,225 @@ def build_map_var_catdesc(self, support_var_name: str) -> str | None: else: return None + def _get_instrument_str(self, full: bool = False) -> str: + """ + Get formatted instrument name string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "IMAP-Hi"). + If False, return short format (e.g., "Hi"). Default is False. + + Returns + ------- + str + Formatted instrument name. + """ + instrument_base = self.instrument.name.split("_")[0] + if instrument_base in ("IDEX", "GLOWS"): + return f"IMAP-{instrument_base}" if full else instrument_base + return f"IMAP-{instrument_base.title()}" if full else instrument_base.title() + + def _get_sensor_str(self, full: bool = False) -> str: + """ + Get formatted sensor string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "45 degree sensor"). + If False, return short format (e.g., "45"). Default is False. + + Returns + ------- + str + Formatted sensor string. + """ + if self.sensor == "combined": + return "combined sensor" if full else " Combined" + elif self.sensor in ("45", "90"): + return f"{self.sensor} degree sensor" if full else str(self.sensor) + elif self.sensor: + return f"sensor {self.sensor}" if full else str(self.sensor) + return "" + + def _get_species_str(self, full: bool = False) -> str: + """ + Get formatted species string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "Hydrogen"). + If False, return short format (e.g., "H"). Default is False. + + Returns + ------- + str + Formatted species string. + """ + if full: + species_names = { + "h": "Hydrogen", + "he": "Helium", + "o": "Oxygen", + "uv": "UV", + } + return species_names.get(self.species.lower(), self.species.title()) + return "UV" if self.species == "uv" else self.species.title() + + def _parse_principal_data(self) -> tuple[str, str]: + """ + Parse principal_data and return (data_type, extras) tuple. + + Returns + ------- + tuple[str, str] + A tuple of (data_type, extras) parsed from principal_data. + data_type is one of ["drt", "ena", "int", "isn", "spx"], + extras is any extension on the principal data portion of the descriptor + string. For example, "isnnbkgnd" would return ("isn", "nbkgnd"). + """ + m = re.match( + r"^(drt|ena|int|isn|spx)(?:(?<=spx)\d+)?([^-_\s]*)$", self.principal_data + ) + if not m: + raise ValueError( + "Invalid principal_data format: " + f"{self.principal_data}. Expected one of 'drt', 'ena', 'int', " + "'isn', or 'spx' optionally followed by digits and trailing " + "non-separator text." + ) + return m.group(1), m.group(2) + + def _get_frame_str(self, full: bool = False) -> str: + """ + Get formatted frame string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "heliospheric"). + If False, return short format (e.g., "Helio"). Default is False. + + Returns + ------- + str + Formatted frame string. + """ + if full: + return INERTIAL_FRAME_LONG_NAMES[self.frame_descriptor] + return { + "hf": "Helio", + "hk": "Helio Kin", + "sf": "SC", + }[self.frame_descriptor] + + def _get_survival_str(self, full: bool = False) -> str: + """ + Get formatted survival correction string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "with survival probability correction"). + If False, return short format (e.g., "Surv Corr"). Default is False. + + Returns + ------- + str + Formatted survival correction string. + """ + if full: + if self.survival_corrected == "sp": + return "with survival probability correction" + return "with no survival correction" + return "Surv Corr" if self.survival_corrected == "sp" else "No Surv Corr" + + def _get_spin_phase_str(self, full: bool = False) -> str: + """ + Get formatted spin phase string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "full spin"). + If False, return short format (e.g., "Full Spin"). Default is False. + + Returns + ------- + str + Formatted spin phase string. + """ + if full: + return { + "full": "full spin", + "ram": "ram", + "anti": "anti-ram", + }.get(self.spin_phase.lower(), self.spin_phase) + spin_phase = self.spin_phase.title() + return "Full Spin" if spin_phase == "Full" else spin_phase + + def _get_resolution_str(self, full: bool = False) -> str: + """ + Get formatted resolution string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "rectangular 2 degree"). + If False, return short format (e.g., "2 deg"). Default is False. + + Returns + ------- + str + Formatted resolution string. + """ + m = re.match(r"^(\d+)deg|nside(\d+)", self.resolution_str) + if not m: + raise ValueError( + f"Invalid resolution_str format: {self.resolution_str}. " + "Expected format like '2deg' or 'nside32'." + ) + if full: + if m.group(1): + return f"rectangular {m.group(1)} degree" + return f"HEALPix nside {m.group(2)}" + return f"{m.group(1)} deg" if m.group(1) else f"NSide {m.group(2)}" + + def _get_duration_str(self, full: bool = False) -> str: + """ + Get formatted duration string. + + Parameters + ---------- + full : bool, optional + If True, return full format (e.g., "6 months"). + If False, return short format (e.g., "6 Mon"). Default is False. + + Returns + ------- + str + Formatted duration string. + """ + if isinstance(self.duration, int): + return f"{self.duration} days" if full else f"{self.duration} Day" + + m = re.match(r"^(\d+)(.*)$", self.duration) + num = int(m.group(1)) + unit = m.group(2).lower() + + if full: + if unit == "yr": + return f"{num} year" if num == 1 else f"{num} years" + elif unit == "mo": + return f"{num} month" if num == 1 else f"{num} months" + return f"{num} {unit}" + + duration = f"{num} {m.group(2).title()}" + return duration + "n" if duration.endswith("Mo") else duration + def build_catdesc(self, quantity_text: str) -> str: """ Convert the MapDescriptor instance to a human-readable CATDESC string. @@ -251,31 +470,21 @@ def build_catdesc(self, quantity_text: str) -> str: instrument = instrument.title() sensor = " Combined" if self.sensor == "combined" else self.sensor species = "UV" if self.species == "uv" else self.species.title() - m = re.match( - r"^(drt|ena|int|isn|spx)(?:(?<=spx)\d+)?([^-_\s]*)$", self.principal_data - ) - if m.group(1) == "isn": + + data_type, extras = self._parse_principal_data() + + if data_type == "isn": species = "ISN " + species - extras = m.group(2) + coord = self.coordinate_system.upper() - frame = { - "hf": "Helio", - "hk": "Helio Kin", - "sf": "SC", - }[self.frame_descriptor] + frame = {"hf": "Helio", "hk": "Helio Kin", "sf": "SC"}[self.frame_descriptor] survival = "Surv Corr" if self.survival_corrected == "sp" else "No Surv Corr" spin_phase = self.spin_phase.title() - if spin_phase == "Full": - spin_phase = "Full Spin" - m = re.match(r"^(\d+)deg|nside(\d+)", self.resolution_str) - resolution = f"{m.group(1)} deg" if m.group(1) else f"NSide {m.group(2)}" - if isinstance(self.duration, int): - duration = f"{self.duration} Day" - else: - m = re.match(r"^(\d+)(.*)$", self.duration) - duration = f"{m.group(1)} {m.group(2).title()}" - if duration.endswith("Mo"): - duration += "n" + spin_phase = "Full Spin" if spin_phase == "Full" else spin_phase + + resolution = self._get_resolution_str(full=False) + duration = self._get_duration_str(full=False) + catdesc = ( f"IMAP {instrument}{sensor} {species} {quantity_text}, {coord} " f"{frame} Frame, {survival}, {spin_phase}, {resolution}, {duration}" @@ -290,6 +499,92 @@ def build_catdesc(self, quantity_text: str) -> str: break return catdesc + def to_logical_source_description(self) -> str: + """ + Convert the MapDescriptor instance to a Logical_source_description string. + + Returns + ------- + str + A full description suitable for the Logical_source_description + global CDF attribute. + """ + # Instrument name (e.g., "IMAP-Hi", "IMAP-Ultra", "IMAP-GLOWS") + instrument_base = self.instrument.name.split("_")[0] + instrument = ( + f"{instrument_base}" + if instrument_base in ("IDEX", "GLOWS") + else f"IMAP-{instrument_base.title()}" + ) + + # Sensor (e.g., "45 degree sensor", "combined sensor", "") + if self.sensor == "combined": + sensor = "combined sensor" + elif self.sensor in ("45", "90"): + sensor = f"{self.sensor} degree sensor" + elif self.sensor: + sensor = f"sensor {self.sensor}" + else: + sensor = "" + + # Species (e.g., "Hydrogen", "Helium", "UV") + species_names = {"h": "Hydrogen", "he": "Helium", "o": "Oxygen", "uv": "UV"} + species = species_names.get(self.species.lower(), self.species.title()) + + data_type, _ = self._parse_principal_data() + + # Quantity (e.g., "ENA Intensity", "Rate", "Dust Rate") + quantity = { + "drt": "Dust Rate", + "ena": "ENA Intensity", + "int": "Intensity", + "isn": "Rate", + "spx": "Spectral Index", + }[data_type] + + # Handle special species cases + if data_type == "isn": + species = f"Interstellar Neutral {species}" + elif data_type == "drt": + # Dust rate maps don't have a species + species = "" + + # Frame (e.g., "heliospheric", "spacecraft", "heliocentric kinetic") + frame = INERTIAL_FRAME_LONG_NAMES[self.frame_descriptor] + + # Coordinate system (e.g., "HAE", "GCS") + coord = self.coordinate_system.upper() + + # Survival correction + if self.survival_corrected == "sp": + survival = "with survival probability correction" + else: + survival = "with no survival correction" + + # Spin phase (e.g., "full spin", "ram", "anti-ram") + spin_phase = { + "full": "full spin", + "ram": "ram", + "anti": "anti-ram", + }.get(self.spin_phase.lower(), self.spin_phase) + + duration = self._get_duration_str(full=True) + resolution = self._get_resolution_str(full=True) + + # Build the full description + # Order matches descriptor: instrument-sensor, quantity, species, frame, + # survival, spin_phase, coord, resolution, duration + sensor_part = f" {sensor}" if sensor else "" + species_part = f"{species} " if species else "" + description = ( + f"{instrument} Instrument Level-2{sensor_part} map of {species_part}" + f"{quantity} in the {frame} frame {survival} in the " + f"{spin_phase} direction in {coord} coordinates on {resolution} " + f"tiling over {duration}." + ) + + return description + @property def principal_data_var(self) -> str: """ @@ -394,7 +689,6 @@ def parse_instrument_descriptor( if instrument_descriptor.endswith("c"): sensor = "combined" instrument_short_name = instrument_descriptor[:-2] - instrument = MappableInstrumentShortName(instrument_short_name) else: # Do regex to get the instrument and sensor: # The first 1 or 3 characters are the instrument short name @@ -411,7 +705,6 @@ def parse_instrument_descriptor( # If sensor is 3 digits, convert to int elif sensor_match and len(sensor_match) == 3: sensor = int(sensor_match) - instrument = MappableInstrumentShortName(instrument_short_name) instrument = MappableInstrumentShortName(instrument_short_name) return instrument, sensor diff --git a/imap_processing/tests/ena_maps/test_naming.py b/imap_processing/tests/ena_maps/test_naming.py index cc32debf5..58b0f1f3a 100644 --- a/imap_processing/tests/ena_maps/test_naming.py +++ b/imap_processing/tests/ena_maps/test_naming.py @@ -505,3 +505,62 @@ def test_build_map_var_catdesc_more_vars( def test_principal_data_var(self, descriptor_str, expected_principal_data_var): md = MapDescriptor.from_string(descriptor_str) assert md.principal_data_var == expected_principal_data_var + + @pytest.mark.parametrize( + "descriptor_str, expected_description", + [ + ( + "h45-ena-h-hf-nsp-full-hae-2deg-6mo", + "IMAP-Hi Instrument Level-2 45 degree sensor map of Hydrogen " + "ENA Intensity in the heliospheric frame with no survival " + "correction in the full spin direction in HAE coordinates on " + "rectangular 2 degree tiling over 6 months.", + ), + ( + "hic-ena-h-hf-sp-ram-hae-nside64-1yr", + "IMAP-Hi Instrument Level-2 combined sensor map of Hydrogen " + "ENA Intensity in the heliospheric frame with survival " + "probability correction in the ram direction in HAE coordinates " + "on HEALPix nside 64 tiling over 1 year.", + ), + ( + "u90-ena-h-hf-nsp-full-hae-nside128-6mo", + "IMAP-Ultra Instrument Level-2 90 degree sensor map of Hydrogen " + "ENA Intensity in the heliospheric frame with no survival " + "correction in the full spin direction in HAE coordinates on " + "HEALPix nside 128 tiling over 6 months.", + ), + ( + "ilo-isn-h-sf-nsp-ram-hae-2deg-3mo", + "IMAP-Lo Instrument Level-2 map of Interstellar Neutral Hydrogen " + "Rate in the spacecraft frame with no survival correction " + "in the ram direction in HAE coordinates on rectangular 2 degree " + "tiling over 3 months.", + ), + ( + "glx-int-uv-hf-nsp-full-hae-2deg-6mo", + "GLOWS Instrument Level-2 map of UV Intensity " + "in the heliospheric frame with no survival correction " + "in the full spin direction in HAE coordinates on rectangular " + "2 degree tiling over 6 months.", + ), + ( + "idx-drt-dust-hf-nsp-full-hae-nside32-1yr", + "IDEX Instrument Level-2 map of Dust Rate " + "in the heliospheric frame with no survival correction " + "in the full spin direction in HAE coordinates on HEALPix " + "nside 32 tiling over 1 year.", + ), + ( + "u45-ena-he-hk-sp-anti-hae-4deg-2mo", + "IMAP-Ultra Instrument Level-2 45 degree sensor map of Helium " + "ENA Intensity in the heliocentric kinetic frame with survival " + "probability correction in the anti-ram direction in HAE " + "coordinates on rectangular 4 degree tiling over 2 months.", + ), + ], + ) + def test_to_logical_source_description(self, descriptor_str, expected_description): + md = MapDescriptor.from_string(descriptor_str) + actual_description = md.to_logical_source_description() + assert actual_description == expected_description diff --git a/imap_processing/tests/hi/test_hi_l2.py b/imap_processing/tests/hi/test_hi_l2.py index 42a274511..aa43cfd95 100644 --- a/imap_processing/tests/hi/test_hi_l2.py +++ b/imap_processing/tests/hi/test_hi_l2.py @@ -175,7 +175,9 @@ def test_hi_l2( # Check some global attributes assert l2_dataset.attrs["Data_type"].startswith(f"L2_{descriptor_str}") assert l2_dataset.attrs["Logical_source"] == f"imap_hi_l2_{descriptor_str}" - assert "Hi90" in l2_dataset.attrs["Logical_source_description"] + assert l2_dataset.attrs["Logical_source_description"].startswith( + "IMAP-Hi Instrument Level-2" + ) assert len(l2_dataset.data_vars) == 16 np.testing.assert_array_equal( diff --git a/imap_processing/ultra/l2/ultra_l2.py b/imap_processing/ultra/l2/ultra_l2.py index 2bdce61d8..2080ad6aa 100644 --- a/imap_processing/ultra/l2/ultra_l2.py +++ b/imap_processing/ultra/l2/ultra_l2.py @@ -861,7 +861,7 @@ def ultra_l2( # Get the global attributes, and then fill the sensor, tiling, etc. in the # format-able strings. map_attrs.update(cdf_attrs.get_global_attributes("imap_ultra_l2_enamap")) - for key in ["Data_type", "Logical_source", "Logical_source_description"]: + for key in ["Data_type", "Logical_source"]: map_attrs[key] = map_attrs[key].format( sensor=ultra_sensor_number, tiling=output_map_structure.tiling_type.value.lower(), @@ -875,6 +875,28 @@ def ultra_l2( else f"nside{output_map_structure.nside}" ), inertial_frame_short_name=inertial_frame, + ) + # Use the previously parsed MapDescriptor to generate the + # Logical_source_description + if descriptor is not None: + map_attrs["Logical_source_description"] = ( + map_descriptor.to_logical_source_description() + ) + else: + map_attrs["Logical_source_description"] = map_attrs[ + "Logical_source_description" + ].format( + sensor=ultra_sensor_number, + tiling=output_map_structure.tiling_type.value.lower(), + duration=map_duration, + resolution_string=( + f"{output_map_structure.spacing_deg:.0f}deg" + if ( + output_map_structure.tiling_type + is ena_maps.SkyTilingType.RECTANGULAR + ) + else f"nside{output_map_structure.nside}" + ), inertial_frame_long_name=inertial_frame_long_name, )