2241 ena create function for producing logical source descriptor details from a mapdescriptor object #3090
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -173,6 +173,86 @@ def to_string(self) -> str: | |||||
| ] | ||||||
| ) | ||||||
|
|
||||||
| 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. | ||||||
|
Contributor
There was a problem hiding this comment. Can you add example of return value for someone like me who has not used this function? Similar to what you did in |
||||||
| """ | ||||||
| m = re.match( | ||||||
| r"^(drt|ena|int|isn|spx)(?:(?<=spx)\d+)?([^-_\s]*)$", self.principal_data | ||||||
| ) | ||||||
|
tmplummer marked this conversation as resolved.
|
||||||
| 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_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)}" | ||||||
|
Comment on lines
+212
to
+222
There was a problem hiding this comment.
|
||||||
|
|
||||||
|
tmplummer marked this conversation as resolved.
|
||||||
| 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) | ||||||
|
tmplummer marked this conversation as resolved.
|
||||||
| 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 to_catdesc(self) -> str: | ||||||
| """ | ||||||
| Convert the MapDescriptor instance to a human-readable CATDESC string. | ||||||
|
|
@@ -188,38 +268,30 @@ def to_catdesc(self) -> str: | |||||
| instrument = instrument.title() | ||||||
| sensor = " Combined" if self.sensor == "combined" else self.sensor | ||||||
| species = "UV" if self.species == "uv" else self.species.title() | ||||||
|
|
||||||
| data_type, extras = self._parse_principal_data() | ||||||
|
|
||||||
| # Quantity (e.g., "Inten", "Rate", "Spectral") | ||||||
| quantity = { | ||||||
| "drt": "Rate", | ||||||
| "ena": "Inten", | ||||||
| "int": "Inten", | ||||||
| "isn": "Rate", | ||||||
| "spx": "Spectral", | ||||||
| }[data_type] | ||||||
|
|
||||||
| if data_type == "isn": | ||||||
| species = "ISN " + species | ||||||
|
|
||||||
| coord = self.coordinate_system.upper() | ||||||
| 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() | ||||||
| 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}, {coord} " | ||||||
| f"{frame} Frame, {survival}, {spin_phase}, {resolution}, {duration}" | ||||||
|
|
@@ -234,6 +306,92 @@ def to_catdesc(self) -> 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") | ||||||
|
Contributor
There was a problem hiding this comment. Should we remove
Suggested change
|
||||||
| instrument_base = self.instrument.name.split("_")[0] | ||||||
| instrument = ( | ||||||
| f"IMAP-{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: | ||||||
| """ | ||||||
|
|
||||||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -174,7 +174,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 l2_dataset.attrs["Logical_source_description"].startswith( | ||
| "IMAP-Hi Instrument Level-2" | ||
| ) | ||
|
|
||
| assert len(l2_dataset.data_vars) == 15 | ||
| np.testing.assert_array_equal( | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -749,7 +749,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"]: | ||
| map_attrs[key] = map_attrs[key].format( | ||
| sensor=ultra_sensor_number, | ||
| tiling=output_map_structure.tiling_type.value.lower(), | ||
|
|
@@ -763,6 +763,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, | ||
| ) | ||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Personal inclination here is that the handling of the error case is legitimately different between the Logical_source_description and the CATDESC, so no need to combine.
MapDescriptor.from_stringis cheap.