Metadata factory migration

changelog/259.feature.rst

@@ -0,0 +1 @@
+Migrated `e-CALLISTO` and `WAVES` spectrogram creation to the new `from_raw` factory pattern, which extracts parsing logic out of the main `SpectrogramFactory`.

radiospectra/spectrogram/sources/callisto.py

@@ -1,6 +1,9 @@
 import astropy.units as u
 from astropy.coordinates import EarthLocation, SkyCoord
 
+from sunpy.net import attrs as a
+from sunpy.time import parse_time
+
 from radiospectra.spectrogram.meta import SpectrogramMeta
 from radiospectra.spectrogram.spectrogrambase import GenericSpectrogram
 
@@ -60,5 +63,41 @@ def observatory_location(self):
         return self.meta.observer_coordinate
 
     @classmethod
-    def is_datasource_for(cls, data, meta, **kwargs):
-        return meta["instrument"] == "e-CALLISTO" or meta["detector"] == "e-CALLISTO"
+    def is_datasource_for(cls, data_or_header, meta_or_raw, **kwargs):
+        meta = data_or_header if hasattr(data_or_header, "get") else meta_or_raw
+        if not hasattr(meta, "get"):
+            return False
+        return "e-CALLISTO" in meta.get("CONTENT", "") or meta.get("instrument") == "e-CALLISTO"
+
+    @classmethod
+    def from_raw(cls, header, raw_object):
+        hd_pairs = raw_object
+        data = hd_pairs[0].data
+        times = hd_pairs[1].data["TIME"].flatten() * u.s
+        freqs = hd_pairs[1].data["FREQUENCY"].flatten() * u.MHz
+        start_time = parse_time(hd_pairs[0].header["DATE-OBS"] + " " + hd_pairs[0].header["TIME-OBS"])
+        try:
+            end_time = parse_time(hd_pairs[0].header["DATE-END"] + " " + hd_pairs[0].header["TIME-END"])
+        except ValueError:
+            # See https://github.com/sunpy/radiospectra/issues/74
+            time_comps = hd_pairs[0].header["TIME-END"].split(":")
+            time_comps[0] = "00"
+            fixed_time = ":".join(time_comps)
+            date_offset = parse_time(hd_pairs[0].header["DATE-END"] + " " + fixed_time)
+            end_time = date_offset + 1 * u.day
+
+        times = start_time + times
+        meta = CALISTOMeta(
+            {
+                "fits_meta": hd_pairs[0].header,
+                "detector": "e-CALLISTO",
+                "instrument": "e-CALLISTO",
+                "observatory": hd_pairs[0].header["INSTRUME"],
+                "start_time": start_time,
+                "end_time": end_time,
+                "wavelength": a.Wavelength(freqs.min(), freqs.max()),
+                "times": times,
+                "freqs": freqs,
+            }
+        )
+        return cls(data, meta)

radiospectra/spectrogram/sources/eovsa.py

@@ -1,3 +1,9 @@
+import astropy.units as u
+from astropy.time import Time
+
+from sunpy.net import attrs as a
+from sunpy.time import parse_time
+
 from radiospectra.spectrogram.meta import SpectrogramMeta
 from radiospectra.spectrogram.spectrogrambase import GenericSpectrogram
 
@@ -44,8 +50,35 @@ def polarisation(self):
         return self.meta.polarisation
 
     @classmethod
-    def is_datasource_for(cls, data, meta, **kwargs):
-        return meta["instrument"] == "EOVSA" or meta["detector"] == "EOVSA"
+    def is_datasource_for(cls, data_or_header, meta_or_raw, **kwargs):
+        meta = data_or_header if hasattr(data_or_header, "get") else meta_or_raw
+        if not hasattr(meta, "get"):
+            return False
+        return (
+            meta.get("TELESCOP", "") == "EOVSA"
+            or meta.get("instrument", "") == "EOVSA"
+            or meta.get("detector", "") == "EOVSA"
+        )
 
-    # TODO fix time gaps for plots need to render them as gaps
-    # can prob do when generateing proper pcolormesh grid but then prob doesn't belong here
+    @classmethod
+    def from_raw(cls, header, raw_object):
+        hd_pairs = raw_object
+        times = Time(hd_pairs[2].data["mjd"] + hd_pairs[2].data["time"] / 1000.0 / 86400.0, format="mjd")
+        freqs = hd_pairs[1].data["sfreq"] * u.GHz
+        data = hd_pairs[0].data
+        start_time = parse_time(hd_pairs[0].header["DATE_OBS"])
+        end_time = parse_time(hd_pairs[0].header["DATE_END"])
+        meta = EOVSAMeta(
+            {
+                "fits_meta": hd_pairs[0].header,
+                "detector": "EOVSA",
+                "instrument": "EOVSA",
+                "observatory": "Owens Valley",
+                "start_time": start_time,
+                "end_time": end_time,
+                "wavelength": a.Wavelength(freqs.min(), freqs.max()),
+                "times": times,
+                "freqs": freqs,
+            }
+        )
+        return cls(data, meta)

radiospectra/spectrogram/sources/ilofar357.py

@@ -1,5 +1,14 @@
+import numpy as np
+
+import astropy.units as u
+from astropy.time import Time
+
+from sunpy import log
+from sunpy.net import attrs as a
+
 from radiospectra.spectrogram.meta import SpectrogramMeta
 from radiospectra.spectrogram.spectrogrambase import GenericSpectrogram
+from radiospectra.utils import subband_to_freq
 
 __all__ = [
     "ILOFARMode357Spectrogram",
@@ -40,5 +49,52 @@ def polarisation(self):
         return self.meta.polarisation
 
     @classmethod
-    def is_datasource_for(cls, data, meta, **kwargs):
-        return meta["instrument"] == "ILOFAR"
+    def is_datasource_for(cls, data_or_header, meta_or_raw, **kwargs):
+        meta = data_or_header if hasattr(data_or_header, "get") else meta_or_raw
+        if not hasattr(meta, "get"):
+            return False
+        return meta.get("instrument") == "ILOFAR"
+
+    @classmethod
+    def from_raw(cls, header, raw_object):
+        file = header["file_path"]
+        subbands = (np.arange(54, 454, 2), np.arange(54, 454, 2), np.arange(54, 230, 2))
+        num_subbands = 488
+
+        data = np.fromfile(file)
+        polarisation = file.stem[-1]
+
+        num_times = data.shape[0] / num_subbands
+        if not num_times.is_integer():
+            log.warning("BST file seems incomplete dropping incomplete frequencies")
+            num_times = np.floor(num_times).astype(int)
+            truncate = num_times * num_subbands
+            data = data[:truncate]
+        data = data.reshape(-1, num_subbands).T  # (Freq x Time).T = (Time x Freq)
+        dt = np.arange(num_times) * 1 * u.s
+        start_time = Time.strptime(file.name.split("_bst")[0], "%Y%m%d_%H%M%S")
+        times = start_time + dt
+
+        obs_mode = (3, 5, 7)
+        freqs = [subband_to_freq(sb, mode) for sb, mode in zip(subbands, obs_mode)]
+
+        # 1st 200 sbs mode 3, next 200 sbs mode 5, last 88 sbs mode 7
+        spec = {0: data[:200, :], 1: data[200:400, :], 2: data[400:, :]}
+        results = []
+        for i in range(3):
+            meta = ILOFARMeta(
+                {
+                    "instrument": "ILOFAR",
+                    "observatory": "Birr (IE613)",
+                    "start_time": times[0],
+                    "mode": obs_mode[i],
+                    "wavelength": a.Wavelength(freqs[i][0], freqs[i][-1]),
+                    "freqs": freqs[i],
+                    "times": times,
+                    "end_time": times[-1],
+                    "detector": "ILOFAR",
+                    "polarisation": polarisation,
+                }
+            )
+            results.append(cls(spec[i], meta))
+        return results

radiospectra/spectrogram/sources/nda.py

@@ -45,7 +45,38 @@ def observatory_location(self):
         return None
 
     @classmethod
-    def is_datasource_for(cls, data, meta, **kwargs):
-        telescope = meta.get("fits_meta", {}).get("TELESCOP", "")
+    def is_datasource_for(cls, data_or_header, meta_or_raw, **kwargs):
+        meta = data_or_header if hasattr(data_or_header, "get") else meta_or_raw
+        if not hasattr(meta, "get"):
+            return False
+        telescope = meta.get("fits_meta", {}).get("TELESCOP", "") if "fits_meta" in meta else meta.get("TELESCOP", "")
         instrument = meta.get("instrument", "")
         return telescope == "NDA" or instrument == "NDA"
+
+    @classmethod
+    def from_raw(cls, header, raw_object):
+        from astropy.time import Time
+
+        from sunpy.net import attrs as a
+
+        hd_pairs = raw_object
+        times = Time(hd_pairs[2].data["jd"], format="jd")
+        freqs = hd_pairs[1].data["frq"].flatten() * u.MHz
+        data = hd_pairs[2].data["data"]
+
+        res = []
+        for i, channel in enumerate(["LL", "RR"]):
+            meta = {
+                "fits_meta": hd_pairs[0].header,
+                "detector": hd_pairs[0].header.get("INSTRUME", "newroutine"),
+                "instrument": "NDA",
+                "observatory": "ORN",
+                "start_time": times[0],
+                "end_time": times[-1],
+                "wavelength": a.Wavelength(freqs.min(), freqs.max()),
+                "times": times,
+                "freqs": freqs,
+                "polarisation": channel,
+            }
+            res.append(cls(data[:, :, i].T, meta))
+        return res

radiospectra/spectrogram/sources/psp_rfs.py

@@ -1,3 +1,8 @@
+import astropy.units as u
+from astropy.time import Time
+
+from sunpy.net import attrs as a
+
 from radiospectra.spectrogram.meta import SpectrogramMeta
 from radiospectra.spectrogram.spectrogrambase import GenericSpectrogram
 
@@ -66,7 +71,55 @@ def version(self):
         return self.meta.version
 
     @classmethod
-    def is_datasource_for(cls, data, meta, **kwargs):
+    def is_datasource_for(cls, data_or_header, meta_or_raw, **kwargs):
+        meta = data_or_header if hasattr(data_or_header, "get") else meta_or_raw
+        if not hasattr(meta, "get"):
+            return False
+
+        if (
+            meta.get("observatory") == "PSP"
+            and meta.get("instrument") == "FIELDS/RFS"
+            and meta.get("detector") in ("lfr", "hfr")
+        ):
+            return True
+
+        cdf_globals = meta.get("cdf_globals")
+        if not cdf_globals:
+            return False
         return (
-            meta["observatory"] == "PSP" and meta["instrument"] == "FIELDS/RFS" and meta["detector"] in ("lfr", "hfr")
+            cdf_globals.get("Project", "")[0] == "PSP"
+            and cdf_globals.get("Source_name", [""])[0] == "PSP_FLD>Parker Solar Probe FIELDS"
+            and "Radio Frequency Spectrometer" in cdf_globals.get("Descriptor", [""])[0]
+        )
+
+    @classmethod
+    def from_raw(cls, header, raw_object):
+        cdf = raw_object
+        cdf_globals = header["cdf_globals"]
+        short, _long = cdf_globals["Descriptor"][0].split(">")
+        detector = short[4:].lower()
+        times, data, freqs = (
+            cdf.varget(name)
+            for name in [
+                f"epoch_{detector}_auto_averages_ch0_V1V2",
+                f"psp_fld_l2_rfs_{detector}_auto_averages_ch0_V1V2",
+                f"frequency_{detector}_auto_averages_ch0_V1V2",
+            ]
+        )
+        times = Time("J2000.0", scale="tt") + (times << u.ns)
+        freqs = freqs[0, :] << u.Hz
+        data = data.T << u.Unit("Volt**2/Hz")
+        meta = RFSMeta(
+            {
+                "cdf_globals": cdf_globals,
+                "detector": detector,
+                "instrument": "FIELDS/RFS",
+                "observatory": "PSP",
+                "start_time": times[0],
+                "end_time": times[-1],
+                "wavelength": a.Wavelength(freqs.min(), freqs.max()),
+                "times": times,
+                "freqs": freqs,
+            }
         )
+        return cls(data, meta)