Merge pull request #171 from DanRyanIrish/spice

SPICE metadata object

changelog/171.feature.rst

@@ -0,0 +1 @@
+Create new Metadata classes for defining mapping of metadata from instrument-specific files to a general metedata API. Includes a specific mapping for SolO/SPICE.

sunraster/instr/spice.py

@@ -0,0 +1,121 @@
+import textwrap
+
+import numpy as np
+from astropy.io import fits
+import astropy.units as u
+from astropy.coordinates import SkyCoord
+from astropy.time import Time
+from sunpy.coordinates import HeliographicStonyhurst
+
+from sunraster.meta import Meta, SlitSpectrographMetaABC
+
+
+__all__ = ["SPICEMeta"]
+
+
+class SPICEMeta(Meta, metaclass=SlitSpectrographMetaABC):
+    def _get_unit(self, key):
+        try:
+            return [s.split("]") for s in self.get_comment(key).split("[")[1:]][0][:-1][0]
+        except IndexError:
+            return None
+
+    def _construct_quantity(self, key):
+        val = self.get(key, None)
+        if val:
+            val *= u.Unit(self._get_unit(key))
+        return val
+
+    def _construct_time(self, key):
+        val = self.get(key, None)
+        scale = self._get_unit(key).lower()
+        if val:
+            val = Time(val, format="fits", scale=scale)
+        return val
+
+    @property
+    def spectral_window(self):
+        spectral_window = self.get("EXTNAME")
+        redundant_txt = "WINDOW"
+        if redundant_txt in spectral_window:
+            spectral_window = np.asanyarray(spectral_window.split("_"))
+            idx = np.array([redundant_txt not in window_chunk for window_chunk in spectral_window])
+            spectral_window = "_".join(spectral_window[idx])
+        return spectral_window
+
+    @property
+    def detector(self):
+        return self.get("DETECTOR", None)
+
+    @property
+    def instrument(self):
+        return self.get("INSTRUME", None)
+
+    @property
+    def observatory(self):
+        return self.get("OBSRVTRY", None)
+
+    @property
+    def processing_level(self):
+        return self.get("LEVEL", None)
+
+    @property
+    def rsun_meters(self):
+        return self._construct_quantity("RSUN_REF")
+
+    @property
+    def rsun_angular(self):
+        return self._construct_quantity("RSUN_ARC")
+
+    @property
+    def observing_mode_id(self):
+        return self.get("OBS_ID", None)
+
+    @property
+    def observatory_radial_velocity(self):
+        return self.get("OBS_VR", None)
+
+    @property
+    def distance_to_sun(self):
+        return self._construct_quantity("DSUN_OBS")
+
+    @property
+    def date_reference(self):
+        return self._construct_time("DATE-OBS")
+
+    @property
+    def date_start(self):
+        return self._construct_time("DATE-BEG")
+
+    @property
+    def date_end(self):
+        return self._construct_time("DATE-END")
+
+    @property
+    def observer_coordinate(self):
+        lon_unit = u.deg
+        lat_unit = u.deg
+        radius_unit = u.m
+        lon_key = "HGLN_OBS"
+        lat_key = "HGLT_OBS"
+        kwargs = {'lon': u.Quantity(self.get(lon_key), unit=self._get_unit(lon_key)).to_value(lon_unit),
+                  'lat': u.Quantity(self.get(lat_key), unit=self._get_unit(lat_key)).to_value(lat_unit),
+                  'radius': self.distance_to_sun.to_value(radius_unit),
+                  'unit': (lon_unit, lat_unit, radius_unit),
+                  'frame': HeliographicStonyhurst}
+        return SkyCoord(obstime=self.date_reference, **kwargs)
+
+    def __str__(self):
+        return textwrap.dedent(f"""\
+                SPICEMeta
+                ---------
+                Observatory:\t\t{self.observatory}
+                Instrument:\t\t{self.instrument}
+                Detector:\t\t{self.detector}
+                Spectral Window:\t{self.spectral_window}
+                Date:\t\t\t{self.date}
+                OBS ID:\t\t\t{self.obsid}
+                """)
+
+    def __repr__(self):
+        return f"{object.__repr__(self)}\n{str(self)}"

sunraster/meta.py

@@ -0,0 +1,124 @@
+import abc
+import copy
+
+__all__ = ["Meta", "RemoteSensorMetaABC", "SlitSpectrographMetaABC"]
+
+
+class RemoteSensorMetaABC(abc.ABCMeta):
+    @abc.abstractmethod
+    def detector(self):
+        pass
+
+    @abc.abstractmethod
+    def instrument(self):
+        pass
+
+    @abc.abstractmethod
+    def observatory(self):
+        pass
+
+    @abc.abstractmethod
+    def processing_level(self):
+        """The level to which the data has been processed."""
+        pass
+
+    @abc.abstractmethod
+    def rsun_meters(self):
+        """Solar radius in units of length."""
+        pass
+
+    @abc.abstractmethod
+    def rsun_angular(self):
+        """Solar radius in angular units as seen from observatory."""
+        pass
+
+    @abc.abstractmethod
+    def distance_to_sun(self):
+        """Distance to Sun center from observatory."""
+        pass
+
+    @abc.abstractmethod
+    def observer_coordinate(self):
+        """Coordinate of observatory location based on header info."""
+        pass
+
+    @abc.abstractmethod
+    def date_reference(self):
+        """The base time from which time axis values are measured.
+
+        Often the same or very similar to date_start.
+        """
+        pass
+
+    @abc.abstractmethod
+    def date_start(self):
+        pass
+
+    @abc.abstractmethod
+    def date_end(self):
+        pass
+
+
+class SlitSpectrographMetaABC(RemoteSensorMetaABC):
+    @abc.abstractmethod
+    def spectral_window(self):
+        pass
+
+    @abc.abstractmethod
+    def observing_mode_id(self):
+        """Unique identifier for the observing mode. Often referred to as OBS ID."""
+        pass
+
+    @abc.abstractmethod
+    def observatory_radial_velocity(self):
+        """Velocity of observatory in direction of source."""
+        pass
+
+
+class Meta():
+    def __init__(self, header):
+        self.raw_meta = header
+        self.original_header = copy.deepcopy(header)
+
+    def get(self, key, default=None):
+        return self._get_fits(key, default)
+
+    def _get_fits(self, key, default=None):
+        return self.raw_meta.get(key, default)
+
+    def get_comment(self, key):
+        return self._get_comment_fits(key)
+
+    def _get_comment_fits(self, key):
+        return self.raw_meta.comments[key]
+
+    def update(self, key_value_pairs):
+        """Update or add an entry in the metadata.
+
+        Parameters
+        ----------
+        key_value_pairs: iterable
+            An iterable of (key, value) pair tuples giving the metadata key/name
+            and the value with which to update it.
+            If key doesn't exist, a new entry is added.
+        """
+        self._update_fits(key_value_pairs)
+
+    def _update_fits(self, key_value_pairs):
+        for key, value in key_value_pairs:
+            self.raw_meta[key] = value
+
+    def update_comments(self, key_comment_pairs):
+        """Update comment associated with existing metadata entry.
+
+        Parameters
+        ----------
+        key_comment_pairs: iterable
+            An iterable of (key, comment) pair tuples giving the metadata key/name
+            and the new comment to assign to it.
+        """
+        self._update_comments_fits(key_comment_pairs)
+
+    def _update_comments_fits(self, key_comment_pairs):
+        for key, comment in key_comment_pairs:
+            self.raw_meta.comments[key] = comment

sunraster/tests/test_meta.py

@@ -0,0 +1,56 @@
+
+import pytest
+from astropy.io import fits
+
+from sunraster.meta import Meta
+
+
+KEY = "KEY"
+VALUE = 0
+COMMENT = "This is a comment."
+MISSING_KEY = "NO KEY"
+ADDITIVE_KEY = "HISTORY"
+ADDITIVE_ENTRY = "1st entry."
+
+@pytest.fixture
+def fits_header():
+    hdr = fits.Header()
+    hdr[KEY] = (VALUE, COMMENT)
+    hdr[ADDITIVE_KEY] = ADDITIVE_ENTRY
+    return hdr
+
+
+@pytest.fixture
+def meta_fits(fits_header):
+    return Meta(fits_header)
+
+
+def test_meta_get_fits(meta_fits):
+    assert meta_fits.get(KEY) == VALUE
+
+
+def test_meta_get_fits_default(meta_fits):
+    assert meta_fits.get(MISSING_KEY, "None") == "None"
+
+
+def test_meta_get_fits_comment(meta_fits):
+    assert meta_fits.get_comment(KEY) == COMMENT
+
+
+@pytest.mark.parametrize("key,new_value", [(KEY, 1),
+                                           (MISSING_KEY, 1)])
+def test_meta_update_fits(meta_fits, key, new_value):
+    meta_fits.update([(key, new_value)])
+    assert meta_fits.get(key) == new_value
+
+
+def test_meta_update_fits_history(meta_fits):
+    second_entry = "2nd entry."
+    meta_fits.update([(ADDITIVE_KEY, second_entry)])
+    assert list(meta_fits.get(ADDITIVE_KEY)) == [ADDITIVE_ENTRY, second_entry]
+
+
+def test_meta_update_comments(meta_fits):
+    new_comment = "This is a new comment."
+    meta_fits.update_comments([(KEY, new_comment)])
+    assert meta_fits.get_comment(KEY) == new_comment