ObsConfig data model (#1973)

* implement a ObsConfig data model

* define PointingMode and add UNKNOWNs to some Enums

* use nan instead of None for quantities

* added stub for EventSource.observation_config

* added obs and scheduling block attributes

* write SB and OB info

* read back sb and ob

* added simulation_config to base EventSource

* fix types, test, and docs

* include observation config in merge tool

* made obs_ids return observation_blocks.keys()

* add linking sb_id to ObservationBlock

* better phrasing of PointingMode comment

* add basic check for existence of merged ob/sb

* better OB/SB for ToyEventSource

* fix other dummy event source

* fix spelling

* fixed references

* add missing docstrings

* moved refs to bibliography

Co-authored-by: Maximilian Nöthe <maximilian.noethe@tu-dortmund.de>

Author: kosack

ctapipe/containers.py

@@ -57,20 +57,85 @@
     "StatisticsContainer",
     "IntensityStatisticsContainer",
     "PeakTimeStatisticsContainer",
+    "SchedulingBlockContainer",
+    "ObservationBlockContainer",
+    "ObservingMode",
+    "ObservationBlockState",
 ]
 
 
 # see https://github.com/astropy/astropy/issues/6509
 NAN_TIME = Time(0, format="mjd", scale="tai")
 
+#: Used for unsigned integer obs_id or sb_id default values:
+UNKNOWN_ID = np.uint64(np.iinfo(np.uint64).max)
 
-class EventType(enum.Enum):
-    """Enum of EventTypes as defined in the CTA Data Model
 
-    These numbers come from  the document *CTA R1/Event Data Model Specification*
-    version 1 revision C.  They may be updated in future revisions
+class SchedulingBlockType(enum.Enum):
+    """
+    Types of Scheduling Block
+    """
+
+    UNKNOWN = -1
+    OBSERVATION = 0
+    CALIBRATION = 1
+    ENGINEERING = 2
+
+
+class ObservationBlockState(enum.Enum):
+    """Observation Block States. Part of the Observation Configuration data
+    model.
+    """
+
+    UNKNOWN = -1
+    FAILED = 0
+    COMPLETED_SUCCEDED = 1
+    COMPLETED_CANCELED = 2
+    COMPLETED_TRUNCATED = 3
+    ARCHIVED = 4
+
+
+class ObservingMode(enum.Enum):
+    """How a scheduling block is observed. Part of the Observation Configuration
+    data model.
+
+    """
+
+    UNKNOWN = -1
+    WOBBLE = 0
+    ON_OFF = 1
+    GRID = 2
+    CUSTOM = 3
+
+
+class PointingMode(enum.Enum):
+    """Describes how the telescopes move. Part of the Observation Configuration
+    data model.
+
+    """
+
+    UNKNOWN = -1
+    #: drives track a point that moves with the sky
+    TRACK = 0
+    #: drives stay fixed at an alt/az point while the sky drifts by
+    DRIFT = 1
+
+
+class CoordinateFrameType(enum.Enum):
+    """types of coordinate frames used in ObservationBlockContainers. Part of
+    the Observation Configuration data model.
+
     """
 
+    UNKNOWN = -1
+    ALTAZ = 0
+    ICRS = 1
+    GALACTIC = 2
+
+
+class EventType(enum.Enum):
+    """Enum of EventTypes as defined in the CTA Data Model [cta_r1event]_"""
+
     # calibrations are 0-15
     FLATFIELD = 0
     SINGLE_PE = 1
@@ -79,14 +144,14 @@ class EventType(enum.Enum):
     ELECTRONIC_PEDESTAL = 4
     OTHER_CALIBRATION = 15
 
-    # For mono-telescope triggers (not used in MC)
+    #: For mono-telescope triggers (not used in MC)
     MUON = 16
     HARDWARE_STEREO = 17
 
-    # ACADA (DAQ) software trigger
+    #: ACADA (DAQ) software trigger
     DAQ = 24
 
-    # Standard Physics  stereo trigger
+    #: Standard Physics  stereo trigger
     SUBARRAY = 32
 
     UNKNOWN = 255
@@ -1143,3 +1208,78 @@ class ArrayEventContainer(Container):
         default_factory=MonitoringContainer,
         description="container for event-wise monitoring data (MON)",
     )
+
+
+class SchedulingBlockContainer(Container):
+    """Stores information about the scheduling block. This is a simplified
+    version of the SB model, only storing what is necessary for analysis. From
+    [cta_sb_ob]_
+
+    """
+
+    default_prefix = ""
+    sb_id = Field(UNKNOWN_ID, "Scheduling block ID", type=np.uint64)
+    sb_type = Field(
+        SchedulingBlockType.UNKNOWN,
+        description="Type of scheduling block",
+        type=SchedulingBlockType,
+    )
+    producer_id = Field(
+        "unknown",
+        "Origin of the sb_id, i.e. name of the telescope site or 'simulation'",
+        type=str,
+    )
+    observing_mode = Field(
+        ObservingMode.UNKNOWN,
+        "Defines how observations within the Scheduling Block are distributed in space",
+        type=ObservingMode,
+    )
+    pointing_mode = Field(
+        PointingMode.UNKNOWN, "Defines how the telescope drives move", type=PointingMode
+    )
+
+
+class ObservationBlockContainer(Container):
+    """Stores information about the observation"""
+
+    default_prefix = ""
+    obs_id = Field(UNKNOWN_ID, "Observation Block ID", type=np.uint64)
+    sb_id = Field(UNKNOWN_ID, "ID of the parent SchedulingBlock", type=np.uint64)
+    producer_id = Field(
+        "unknown",
+        "Origin of the obs_id, i.e. name of the telescope site or 'simulation'",
+        type=str,
+    )
+
+    state = Field(
+        ObservationBlockState.UNKNOWN, "State of this OB", type=ObservationBlockState
+    )
+
+    subarray_pointing_lat = Field(
+        nan * u.deg,
+        "latitude of the nominal center coordinate of this observation",
+        unit=u.deg,
+    )
+
+    subarray_pointing_lon = Field(
+        nan * u.deg,
+        "longitude of the nominal center coordinate of this observation",
+        unit=u.deg,
+    )
+
+    subarray_pointing_frame = Field(
+        CoordinateFrameType.UNKNOWN,
+        (
+            "Frame in which the subarray_target is non-moving. If the frame is ALTAZ, "
+            "the meaning of (lon,lat) is (azimuth, altitude) while for ICRS it is "
+            "(right-ascension, declination)"
+        ),
+        type=CoordinateFrameType,
+    )
+
+    scheduled_duration = Field(
+        nan * u.min, "expected duration from scheduler", unit=u.min
+    )
+    scheduled_start_time = Field(NAN_TIME, "expected start time from scheduler")
+    actual_start_time = Field(NAN_TIME, "true start time")
+    actual_duration = Field(nan * u.min, "true duration", unit=u.min)

ctapipe/io/datawriter.py

@@ -40,7 +40,7 @@ def _get_tel_index(event, tel_id):
     )
 
 
-# define the version of the DL1 data model written here. This should be updated
+# define the version of the data model written here. This should be updated
 # when necessary:
 # - increase the major number if there is a breaking change to the model
 #   (meaning readers need to update scripts)
@@ -61,6 +61,7 @@ def _get_tel_index(event, tel_id):
           - The reference_location (EarthLocation origin of the telescope coordinates)
             is now included in SubarrayDescription
           - Only unique optics are stored in the optics table
+          - include observation configuration
 - v3.0.0: reconstructed core uncertainties splitted in their X-Y components
 - v2.2.0: added R0 and R1 outputs
 - v2.1.0: hillas and timing parameters are per default saved in telescope frame (degree) as opposed to camera frame (m)
@@ -137,7 +138,7 @@ def write_reference_metadata_headers(
 
 class DataWriter(Component):
     """
-    Serialize a sequence of events into a HDF5 DL1 file, in the correct format
+    Serialize a sequence of events into a HDF5 file, in the correct format
 
     Examples
     --------
@@ -268,6 +269,7 @@ def __init__(self, event_source: EventSource, config=None, parent=None, **kwargs
 
     def _setup_outputfile(self):
         self._subarray.to_hdf(self._writer.h5file)
+        self._write_scheduling_and_observation_blocks()
         if self._is_simulation:
             self._write_simulation_configuration()
 
@@ -321,7 +323,7 @@ def __call__(self, event: ArrayEventContainer):
 
     def finish(self):
         """called after all events are done"""
-        self.log.info("Finishing DL1 output")
+        self.log.info("Finishing output")
         if not self._at_least_one_event:
             self.log.warning("No events have been written to the output file")
         if self._writer:
@@ -496,6 +498,21 @@ def _write_subarray_pointing(self, event: ArrayEventContainer, writer: TableWrit
             writer.write("dl1/monitoring/subarray/pointing", [event.trigger, pnt])
             self._last_pointing = current_pointing
 
+    def _write_scheduling_and_observation_blocks(self):
+        """write out SB and OB info"""
+
+        self.log.debug(
+            "writing %d sbs and %d obs",
+            len(self.event_source.scheduling_blocks.values()),
+            len(self.event_source.observation_blocks.values()),
+        )
+
+        for sb in self.event_source.scheduling_blocks.values():
+            self._writer.write("configuration/observation/scheduling_block", sb)
+
+        for ob in self.event_source.observation_blocks.values():
+            self._writer.write("configuration/observation/observation_block", ob)
+
     def _write_simulation_configuration(self):
         """
         Write the simulation headers to a single row of a table. Later

ctapipe/io/eventsource.py

@@ -3,11 +3,16 @@
 """
 import warnings
 from abc import abstractmethod
-from typing import Generator, List, Tuple
+from typing import Dict, Generator, List, Tuple
 
 from traitlets.config.loader import LazyConfigValue
 
-from ..containers import ArrayEventContainer
+from ..containers import (
+    ArrayEventContainer,
+    ObservationBlockContainer,
+    SchedulingBlockContainer,
+    SimulationConfigContainer,
+)
 from ..core import Provenance, ToolConfigurationError
 from ..core.component import Component, find_config_in_hierarchy, non_abstract_children
 from ..core.traits import CInt, Int, Path, Set, TraitError, Undefined
@@ -205,11 +210,38 @@ def subarray(self) -> SubarrayDescription:
 
         """
 
+    @property
+    def simulation_config(self) -> Dict[int, SimulationConfigContainer]:
+        """The simulation configurations of all observations provided by the
+        EventSource, or None if the source does not provide simulated data
+
+        Returns
+        -------
+        Dict[int,ctapipe.containers.SimulationConfigContainer] | None
+        """
+        return None
+
+    @property
+    @abstractmethod
+    def observation_blocks(self) -> Dict[int, ObservationBlockContainer]:
+        """
+        Obtain the ObservationConfigurations from the EventSource, indexed by obs_id
+        """
+        pass
+
+    @property
+    @abstractmethod
+    def scheduling_blocks(self) -> Dict[int, SchedulingBlockContainer]:
+        """
+        Obtain the ObservationConfigurations from the EventSource, indexed by obs_id
+        """
+        pass
+
     @property
     @abstractmethod
     def is_simulation(self) -> bool:
         """
-        Weither the currently opened file is simulated
+        Whether the currently opened file is simulated
 
         Returns
         -------
@@ -240,7 +272,6 @@ def has_any_datalevel(self, datalevels) -> bool:
         return any(dl in self.datalevels for dl in datalevels)
 
     @property
-    @abstractmethod
     def obs_ids(self) -> List[int]:
         """
         The observation ids of the runs located in the file
@@ -250,6 +281,7 @@ def obs_ids(self) -> List[int]:
         -------
         list[int]
         """
+        return list(self.observation_blocks.keys())
 
     @abstractmethod
     def _generator(self) -> Generator[ArrayEventContainer, None, None]: