Merge pull request #2791 from cta-observatory/generalise_table_preprocessing

Generalise table preprocessing

Author: kosack

docs/api-reference/io/index.rst

@@ -281,6 +281,9 @@ Reference/API
 .. automodapi:: ctapipe.io.metadata
     :no-inheritance-diagram:
 
+.. automodapi:: ctapipe.io.dl2_tables_preprocessing
+    :no-inheritance-diagram:
+
 .. automodapi:: ctapipe.io.eventsource
     :no-inheritance-diagram:
 

docs/api-reference/irf/index.rst

@@ -33,7 +33,6 @@ Submodules
     irfs
     benchmarks
     binning
-    preprocessing
     spectra
 
 

docs/api-reference/irf/preprocessing.rst

@@ -0,0 +1,2 @@
+Generalise the DL2 table processing implemented in IRF tools in order to be used by other usecases that ingest, filter and merge DL2 tables.
+The module was moved from IRF to IO.

docs/changes/2791.feature.rst

@@ -905,11 +905,11 @@ def irf_event_loader_test_config():
 
     return Config(
         {
-            "EventPreprocessor": {
+            "DL2EventPreprocessor": {
                 "energy_reconstructor": "ExtraTreesRegressor",
                 "geometry_reconstructor": "HillasReconstructor",
                 "gammaness_classifier": "ExtraTreesClassifier",
-                "EventQualityQuery": {
+                "DL2EventQualityQuery": {
                     "quality_criteria": [
                         (
                             "multiplicity 4",
@@ -936,15 +936,19 @@ def event_loader_config_path(irf_event_loader_test_config, irf_tmp_path):
 
 @pytest.fixture(scope="session")
 def irf_events_table():
-    from ctapipe.irf import EventPreprocessor
+    from ctapipe.io import DL2EventPreprocessor
 
     N1 = 1000
     N2 = 100
     N = N1 + N2
-    epp = EventPreprocessor()
+    epp = DL2EventPreprocessor()
     tab = epp.make_empty_table()
 
-    ids, bulk, unitless = tab.colnames[:2], tab.colnames[2:-2], tab.colnames[-2:]
+    ids = ["obs_id", "event_id"]
+    unitless = set(
+        [colname for colname in tab.colnames if tab[colname].unit is None]
+    ) - set(ids)
+    bulk = set(tab.colnames) - set(ids) - set(unitless)
 
     id_tab = QTable(
         data=np.zeros((N, len(ids)), dtype=np.uint64),
@@ -956,16 +960,21 @@ def irf_events_table():
         names=bulk,
         units={c: tab[c].unit for c in bulk},
     )
-
     # Setting values following pyirf test in pyirf/irf/tests/test_background.py
-    bulk_tab["reco_energy"] = np.append(np.full(N1, 1), np.full(N2, 2)) * u.TeV
-    bulk_tab["true_energy"] = np.append(np.full(N1, 0.9), np.full(N2, 2.1)) * u.TeV
-    bulk_tab["reco_source_fov_offset"] = (
-        np.append(np.full(N1, 0.1), np.full(N2, 0.05)) * u.deg
+    bulk_tab.replace_column(
+        "reco_energy", np.append(np.full(N1, 1), np.full(N2, 2)) * u.TeV
+    )
+    bulk_tab.replace_column(
+        "true_energy", np.append(np.full(N1, 0.9), np.full(N2, 2.1)) * u.TeV
     )
-    bulk_tab["true_source_fov_offset"] = (
-        np.append(np.full(N1, 0.11), np.full(N2, 0.04)) * u.deg
+    bulk_tab.replace_column(
+        "reco_source_fov_offset", np.append(np.full(N1, 0.1), np.full(N2, 0.05)) * u.deg
     )
+    bulk_tab.replace_column(
+        "true_source_fov_offset",
+        np.append(np.full(N1, 0.11), np.full(N2, 0.04)) * u.deg,
+    )
+
     for name in unitless:
         bulk_tab.add_column(
             Column(name=name, unit=tab[name].unit, data=np.zeros(N) * np.nan)
@@ -975,3 +984,40 @@ def irf_events_table():
 
     ev = vstack([e_tab, tab], join_type="exact", metadata_conflicts="silent")
     return ev
+
+
+@pytest.fixture(scope="function")
+def test_config():
+    return {
+        "DL2EventLoader": {"event_reader_function": "read_telescope_events_chunked"},
+        "DL2EventPreprocessor": {
+            "energy_reconstructor": "ExtraTreesRegressor",
+            "gammaness_classifier": "ExtraTreesClassifier",
+            "columns_to_rename": {},
+            "output_table_schema": [
+                Column(
+                    name="obs_id", dtype=np.uint64, description="Observation Block ID"
+                ),
+                Column(name="event_id", dtype=np.uint64, description="Array event ID"),
+                Column(name="tel_id", dtype=np.uint64, description="Telescope ID"),
+                Column(
+                    name="ExtraTreesRegressor_tel_energy",
+                    unit=u.TeV,
+                    description="Reconstructed energy",
+                ),
+                Column(
+                    name="ExtraTreesRegressor_tel_energy_uncert",
+                    unit=u.TeV,
+                    description="Reconstructed energy uncertainty",
+                ),
+            ],
+            "apply_derived_columns": False,
+            # "disable_column_renaming": True,
+            "allow_unsupported_pointing_frames": True,
+        },
+        "DL2EventQualityQuery": {
+            "quality_criteria": [
+                ("valid reco", "ExtraTreesRegressor_tel_is_valid"),
+            ]
+        },
+    }

src/ctapipe/conftest.py

@@ -6,6 +6,7 @@
 """
 from .astropy_helpers import read_table, write_table  # noqa: I001
 from .datalevels import DataLevel
+from .dl2_tables_preprocessing import DL2EventPreprocessor, DL2EventLoader
 from .eventsource import EventSource
 from .eventseeker import EventSeeker
 from .tableio import TableReader, TableWriter
@@ -41,5 +42,7 @@
     "DataWriter",
     "DATA_MODEL_VERSION",
     "get_hdf5_datalevels",
+    "DL2EventPreprocessor",
+    "DL2EventLoader",
     "get_hdf5_monitoring_types",
 ]