add meta container

Signed-off-by: Atreyee Sinha <asinha@ucm.es>

gammapy/datasets/__init__.py

@@ -3,6 +3,7 @@
 from .flux_points import FluxPointsDataset
 from .io import OGIPDatasetReader, OGIPDatasetWriter
 from .map import MapDataset, MapDatasetOnOff, create_map_dataset_geoms
+from .metadata import MapDatasetMetaData
 from .simulate import MapDatasetEventSampler
 from .spectrum import SpectrumDataset, SpectrumDatasetOnOff
 
@@ -31,4 +32,5 @@
     "OGIPDatasetReader",
     "SpectrumDataset",
     "SpectrumDatasetOnOff",
+    "MapDatasetMetaData",
 ]

gammapy/datasets/map.py

@@ -25,6 +25,7 @@
 from gammapy.utils.table import hstack_columns
 from .core import Dataset
 from .evaluator import MapEvaluator
+from .metadata import MapDatasetMetaData
 from .utils import get_axes
 
 __all__ = ["MapDataset", "MapDatasetOnOff", "create_map_dataset_geoms"]
@@ -216,6 +217,7 @@ def __init__(
         mask_fit=None,
         gti=None,
         meta_table=None,
+        meta=None,
         name=None,
     ):
         self._name = make_name(name)
@@ -247,6 +249,7 @@ def __init__(
         self.gti = gti
         self.models = models
         self.meta_table = meta_table
+        self._meta = meta
 
     # TODO: keep or remove?
     @property
@@ -436,6 +439,13 @@ def energy_range_total(self):
         energy_min_map, energy_max_map = self.energy_range
         return np.nanmin(energy_min_map.quantity), np.nanmax(energy_max_map.quantity)
 
+    @property
+    def meta(self):
+        """Return metadata container."""
+        if self._meta is None:
+            self._meta = MapDatasetMetaData.from_default()
+        return self._meta
+
     def npred(self):
         """Total predicted source and background counts.
 

gammapy/datasets/metadata.py

@@ -0,0 +1,157 @@
+import logging
+from typing import Optional, Union
+import numpy as np
+from astropy.coordinates import SkyCoord
+from pydantic import ValidationError, validator
+from gammapy.utils.metadata import CreatorMetaData, MetaData
+
+__all__ = ["MapDatasetMetaData"]
+
+
+class MapDatasetMetaData(MetaData):
+    """Metadata containing information about the GTI.
+
+    Parameters
+    ----------
+    creation : `~gammapy.utils.CreatorMetaData`
+        the creation metadata
+    instrument : str
+        the instrument used during observation
+    telescope : str
+        The specific telescope subarray
+    observation_mode : str
+        observing mode
+    pointing : ~astropy.coordinates.SkyCoord
+        Telescope pointing direction
+    obs_ids : int
+        Observation ids stacked in the dataset
+    event_types : int
+        Event types used in analysis
+    optional : dict
+        Any other meta information
+    """
+
+    creation: Optional[CreatorMetaData]
+    instrument: Optional[str]
+    telescope: Optional[Union[str, list[str]]]
+    observation_mode: Optional[Union[str, list]]
+    pointing: Optional[Union[SkyCoord, list[SkyCoord]]]
+    obs_ids: Optional[Union[int, list[int]]]
+    event_type: Optional[Union[int, list[int]]]
+    optional: Optional[dict]
+
+    @validator("creation")
+    def validate_creation(cls, v):
+        if v is None:
+            return CreatorMetaData.from_default()
+        elif isinstance(v, CreatorMetaData):
+            return v
+        else:
+            raise ValidationError(
+                f"Incorrect pointing. Expect CreatorMetaData got {type(v)} instead."
+            )
+
+    @validator("instrument")
+    def validate_instrument(cls, v):
+        if isinstance(v, str):
+            return v
+        elif v is None:
+            return v
+        else:
+            raise ValidationError(
+                f"Incorrect instrument. Expect str got {type(v)} instead."
+            )
+
+    @validator("telescope")
+    def validate_telescope(cls, v):
+        if isinstance(v, str):
+            return v
+        elif v is None:
+            return v
+        elif all(isinstance(_, str) for _ in v):
+            return v
+        else:
+            raise ValidationError(
+                f"Incorrect telescope type. Expect str got {type(v)} instead."
+            )
+
+    @validator("pointing")
+    def validate_pointing(cls, v):
+        if v is None:
+            return SkyCoord(np.nan, np.nan, unit="deg", frame="icrs")
+        elif isinstance(v, SkyCoord):
+            return v
+        elif all(isinstance(_, SkyCoord) for _ in v):
+            return v
+        else:
+            raise ValidationError(
+                f"Incorrect pointing. Expect SkyCoord got {type(v)} instead."
+            )
+
+    @validator("obs_ids", "event_type")
+    def validate_obs_ids(cls, v):
+        if v is None:
+            return -999
+        elif isinstance(v, int):
+            return v
+        elif all(isinstance(_, int) for _ in v):
+            return v
+        else:
+            raise ValidationError(
+                f"Incorrect pointing. Expect int got {type(v)} instead."
+            )
+
+    @classmethod
+    def from_default(cls):
+        """Creation metadata containing Gammapy version."""
+        creation = CreatorMetaData.from_default()
+        return cls(creation=creation)
+
+    def stack(self, other):
+        kwargs = {}
+        kwargs["creation"] = self.creation
+        kwargs["instrument"] = self.instrument
+        if self.instrument != other.instrument:
+            logging.warning(
+                f"Stacking data from different instruments {self.instrument} and {other.instrument}"
+            )
+        tel = self.telescope
+        if isinstance(tel, str):
+            tel = [tel]
+        if other.telescope not in tel:
+            tel.append(other.telescope)
+        kwargs["telescope"] = tel
+
+        observation_mode = self.observation_mode
+        if isinstance(observation_mode, str):
+            observation_mode = [observation_mode]
+        observation_mode.append(other.observation_mode)
+        kwargs["observation_mode"] = observation_mode
+
+        pointing = self.pointing
+        if isinstance(pointing, SkyCoord):
+            pointing = [pointing]
+        pointing.append(other.pointing)
+        kwargs["pointing"] = pointing
+
+        obs_ids = self.obs_ids
+        if isinstance(obs_ids, int):
+            obs_ids = [obs_ids]
+        obs_ids.append(other.obs_ids)
+        kwargs["obs_ids"] = obs_ids
+
+        event_type = self.event_type
+        if not isinstance(event_type, list):
+            event_type = [event_type]
+        event_type.append(other.event_type)
+        kwargs["event_type"] = event_type
+
+        if self.optional:
+            optional = self.optional
+            for k in other.optional.keys():
+                if not isinstance(optional[k], list):
+                    optional[k] = [optional[k]]
+                optional[k].append(other.optional[k])
+            kwargs["optional"] = optional
+
+        return self.__class__(**kwargs)

gammapy/datasets/tests/test_metadata.py

@@ -0,0 +1,103 @@
+import pytest
+import numpy as np
+from numpy.testing import assert_allclose
+from astropy.coordinates import SkyCoord
+from pydantic import ValidationError
+from gammapy.datasets import MapDatasetMetaData
+
+
+def test_mapdataset_meta_from_default():
+    meta = MapDatasetMetaData.from_default()
+
+    assert meta.creation.creator.split()[0] == "Gammapy"
+
+
+def test_mapdataset_metadata():
+    input = {
+        "telescope": "cta-north",
+        "instrument": "lst",
+        "observation_mode": "wobble",
+        "pointing": SkyCoord(83.6287, 22.0147, unit="deg", frame="icrs"),
+        "obs_ids": 112,
+        "optional": dict(test=0.5, other=True),
+    }
+    meta = MapDatasetMetaData(**input)
+
+    assert meta.telescope == "cta-north"
+    assert meta.instrument == "lst"
+    assert meta.observation_mode == "wobble"
+    assert_allclose(meta.pointing.dec.value, 22.0147)
+    assert_allclose(meta.pointing.ra.deg, 83.6287)
+    assert meta.obs_ids == 112
+    assert meta.optional["other"] is True
+    assert meta.creation.creator.split()[0] == "Gammapy"
+
+    with pytest.raises(ValidationError):
+        meta.pointing = 2.0
+
+    with pytest.raises(ValidationError):
+        meta.instrument = ["cta", "hess"]
+
+    meta.pointing = None
+    assert isinstance(meta.pointing, SkyCoord)
+    assert np.isnan(meta.pointing.ra.deg)
+
+    input_bad = input.copy()
+    input_bad["obs_ids"] = "bad"
+
+    with pytest.raises(ValueError):
+        MapDatasetMetaData(**input_bad)
+
+
+def test_mapdataset_metadata_lists():
+    input = {
+        "telescope": "cta-north",
+        "instrument": "lst",
+        "observation_mode": "wobble",
+        "pointing": [
+            SkyCoord(83.6287, 22.0147, unit="deg", frame="icrs"),
+            SkyCoord(83.1287, 22.5147, unit="deg", frame="icrs"),
+        ],
+        "obs_ids": [111, 222],
+    }
+    meta = MapDatasetMetaData(**input)
+    assert meta.telescope == "cta-north"
+    assert meta.instrument == "lst"
+    assert meta.observation_mode == "wobble"
+    assert_allclose(meta.pointing[0].dec.value, 22.0147)
+    assert_allclose(meta.pointing[1].ra.deg, 83.1287)
+    assert meta.obs_ids == [111, 222]
+    assert meta.optional is None
+    assert meta.event_type == -999
+
+
+def test_mapdataset_metadata_stack():
+    input1 = {
+        "telescope": "a",
+        "instrument": "H.E.S.S.",
+        "observation_mode": "wobble",
+        "pointing": SkyCoord(83.6287, 22.5147, unit="deg", frame="icrs"),
+        "obs_ids": 111,
+        "optional": dict(test=0.5, other=True),
+    }
+
+    input2 = {
+        "telescope": "b",
+        "instrument": "H.E.S.S.",
+        "observation_mode": "wobble",
+        "pointing": SkyCoord(83.6287, 22.0147, unit="deg", frame="icrs"),
+        "obs_ids": 112,
+        "optional": dict(test=0.1, other=False),
+    }
+
+    meta1 = MapDatasetMetaData(**input1)
+    meta2 = MapDatasetMetaData(**input2)
+
+    meta = meta1.stack(meta2)
+    assert meta.telescope == ["a", "b"]
+    assert meta.instrument == "H.E.S.S."
+    assert meta.observation_mode == ["wobble", "wobble"]
+    assert_allclose(meta.pointing[1].dec.deg, 22.0147)
+    assert meta.obs_ids == [111, 112]
+    assert meta.optional["other"] == [True, False]
+    assert len(meta.event_type) == 2