Repository: cta-observatory/ctapipe (python)

Issue:
Support writing high-precision timestamps to hdf5
**Please describe the use case that requires this feature.**

Currently, we write out all timestamps as TAI MJD.
That is one of the possible formats for the CTA "low precision" timestamp.
However, some things might still require the high precision timestamps, so we should have an option on the datawriter whether to use high or low precision timestamps for event times and similar things that are high precision in the input.

**Describe the solution you'd like**

Add option to store timestamps that are high precision keeping this high precision.

Pull request description:
Store event times using CTAO high resolution timestamps
Fixes #2428 

Patch:
diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml
index 9e989cf6..f2d79e8e 100644
--- a/.github/workflows/sonar.yml
+++ b/.github/workflows/sonar.yml
@@ -19,6 +19,26 @@ jobs:
     runs-on: ubuntu-latest
     if: github.event.workflow_run.conclusion == 'success'
     steps:
+      - name: Checkout branch or tag
+        uses: actions/checkout@v4
+        if: github.event.workflow_run.event != 'pull_request'
+        with:
+          repository: ${{ github.event.workflow_run.head_repository.full_name }}
+          ref: ${{ github.event.workflow_run.head_branch }}
+          fetch-depth: 0
+
+      - name: Checkout Pull Request merge result
+        uses: actions/checkout@v4
+        if: github.event.workflow_run.event == 'pull_request'
+        with:
+          ref: refs/pull/${{ github.event.workflow_run.pull_requests[0].number }}/merge
+          fetch-depth: 0
+
+      - name: Git info
+        run: |
+          git status
+          git log -n 5 --oneline
+
       - name: 'Download code coverage'
         uses: actions/github-script@v7
         with:
@@ -41,39 +61,15 @@ jobs:
             fs.writeFileSync(`${process.env.GITHUB_WORKSPACE}/ctapipe-coverage-report.zip`, Buffer.from(download.data));
 
       - name: 'Unzip code coverage'
-        run: unzip ctapipe-coverage-report.zip -d $HOME/coverage
-
-      - name: Set environment
-        run: |
-          cat $HOME/coverage/sonar_env >> "$GITHUB_ENV"
-          cat $HOME/coverage/sonar_env
-
-      - name: Checkout branch or tag
-        uses: actions/checkout@v4
-        if: github.event.workflow_run.event != 'pull_request'
-        with:
-          repository: ${{ github.event.workflow_run.head_repository.full_name }}
-          ref: ${{ github.event.workflow_run.head_branch }}
-          fetch-depth: 0
-
-      - name: Checkout Pull Request merge result
-        uses: actions/checkout@v4
-        if: github.event.workflow_run.event == 'pull_request'
-        with:
-          ref: refs/pull/${{ env.PR_NUMBER }}/merge
-          fetch-depth: 0
+        run: unzip ctapipe-coverage-report.zip -d coverage
 
       - name: Check artifact
-        run: |
-          cp -r $HOME/coverage .
-          ls -l coverage
-          cat coverage/coverage.xml
+        run: ls -l coverage
 
-      - name: Git info
+      - name: Set environment
         run: |
-          git status
-          git log -n 5 --oneline
-          ls -l coverage
+          cat coverage/sonar_env >> "$GITHUB_ENV"
+          cat coverage/sonar_env
 
       - name: Sonarqube Scan
         uses: SonarSource/sonarqube-scan-action@v5
diff --git a/docs/api-reference/time/index.rst b/docs/api-reference/time/index.rst
new file mode 100644
index 00000000..b3842d49
--- /dev/null
+++ b/docs/api-reference/time/index.rst
@@ -0,0 +1,14 @@
+.. _time:
+
+**********************
+time (`~ctapipe.time`)
+**********************
+
+.. currentmodule:: ctapipe.time
+
+
+Reference/API
+=============
+
+.. automodapi:: ctapipe.time
+    :no-inheritance-diagram:
diff --git a/docs/changes/2707.datamodel.rst b/docs/changes/2707.datamodel.rst
new file mode 100644
index 00000000..87c3f81c
--- /dev/null
+++ b/docs/changes/2707.datamodel.rst
@@ -0,0 +1,9 @@
+Change format in which event timestamps are written to HDF5 files.
+Instead of a single float64 MJD value which has ~µs precision,
+the CTAO high precision time format is used now.
+This stores the timestamp as two uint32 values: seconds
+and quarter nanoseconds since ``1970-01-01T00:00:00.0 TAI``.
+
+This only affects the storage format and the precision,
+the in-memory API is unchanged as it relies on ``astropy.time.Time``
+and values are converted when reading/writing from/to HDF5.
diff --git a/docs/changes/2709.feature.rst b/docs/changes/2709.feature.rst
deleted file mode 100644
index 82f4d8e7..00000000
--- a/docs/changes/2709.feature.rst
+++ /dev/null
@@ -1,1 +0,0 @@
-Add a method ``read_scheduling_blocks`` in the ``TableLoader`` class to read scheduling block information.
diff --git a/docs/changes/2715.bugfix.rst b/docs/changes/2715.bugfix.rst
deleted file mode 100644
index 5c5f0217..00000000
--- a/docs/changes/2715.bugfix.rst
+++ /dev/null
@@ -1,1 +0,0 @@
-Fix processing gain-selected dl1 data with the ``PixelStatisticsCalculatorTool``.
diff --git a/src/ctapipe/instrument/optics.py b/src/ctapipe/instrument/optics.py
index 7ddbd744..9bda6ce3 100644
--- a/src/ctapipe/instrument/optics.py
+++ b/src/ctapipe/instrument/optics.py
@@ -378,7 +378,7 @@ class ComaPSFModel(PSFModel):
         help=(
             "Describes the dependency of the polar scale on the "
             "distance to the center of the camera. Used to calculate "
-            "the width :math:`S_\phi` of the polar Laplacian in the PSF "
+            r"the width :math:`S_\phi` of the polar Laplacian in the PSF "
             "as a function of the distance r to the optical axis"
         )
     ).tag(config=True)
diff --git a/src/ctapipe/io/astropy_helpers.py b/src/ctapipe/io/astropy_helpers.py
index e8077e96..4b86db86 100644
--- a/src/ctapipe/io/astropy_helpers.py
+++ b/src/ctapipe/io/astropy_helpers.py
@@ -120,6 +120,7 @@ def write_table(
     append=False,
     overwrite=False,
     mode="a",
+    time_format="ctao_high_res",
     filters=DEFAULT_FILTERS,
 ):
     """Write a table to an HDF5 file
@@ -145,6 +146,9 @@ def write_table(
     mode: str
         If given a path for ``h5file``, it will be opened in this mode.
         See the docs of ``tables.open_file``.
+    time_format: str
+        Format to use for storing time columns.
+        Either 'ctao_high_res' (the default) or a format supported by `astropy.time.Time`.
     """
     copied = False
     parent, table_name = os.path.split(path)
@@ -173,7 +177,7 @@ def write_table(
                 attrs[f"CTAFIELD_{pos}_DESC"] = column.description
 
             if isinstance(column, Time):
-                transform = TimeColumnTransform(scale="tai", format="mjd")
+                transform = TimeColumnTransform(scale="tai", format=time_format)
                 attrs.update(transform.get_meta(pos))
 
                 if copied is False:
diff --git a/src/ctapipe/io/datawriter.py b/src/ctapipe/io/datawriter.py
index 4b0b4e20..dbd5b36b 100644
--- a/src/ctapipe/io/datawriter.py
+++ b/src/ctapipe/io/datawriter.py
@@ -45,8 +45,10 @@ def _get_tel_index(event, tel_id):
 #   (meaning readers need to update scripts)
 # - increase the minor number if new columns or datasets are added
 # - increase the patch number if there is a small bugfix to the model.
-DATA_MODEL_VERSION = "v6.0.0"
+DATA_MODEL_VERSION = "v7.0.0"
 DATA_MODEL_CHANGE_HISTORY = """
+- v7.0.0: - Use high resolution timestamps for times. CTAO high resolution times
+            are stored as two uint32: seconds and quarter nanoseconds since 1970-01-01T00:00:00 TAI.
 - v6.0.0: - Change R1- and DL0-waveform shape from (n_pixels, n_samples) to be always
             (n_channels, n_pixels, n_samples).
 - v5.1.0: - Remove redundant 'is_valid' column in ``DispContainer``.
@@ -534,10 +536,14 @@ class DataWriter(Component):
         )
 
         for sb in self.event_source.scheduling_blocks.values():
-            self._writer.write("configuration/observation/scheduling_block", sb)
+            self._writer.write(
+                "configuration/observation/scheduling_block", sb, time_format="mjd"
+            )
 
         for ob in self.event_source.observation_blocks.values():
-            self._writer.write("configuration/observation/observation_block", ob)
+            self._writer.write(
+                "configuration/observation/observation_block", ob, time_format="mjd"
+            )
 
     def _write_simulation_configuration(self):
         """
diff --git a/src/ctapipe/io/hdf5eventsource.py b/src/ctapipe/io/hdf5eventsource.py
index 8e8cb3ad..17a03a20 100644
--- a/src/ctapipe/io/hdf5eventsource.py
+++ b/src/ctapipe/io/hdf5eventsource.py
@@ -77,6 +77,7 @@ COMPATIBLE_DATA_MODEL_VERSIONS = [
     "v5.0.0",
     "v5.1.0",
     "v6.0.0",
+    "v7.0.0",
 ]
 
 
@@ -314,15 +315,24 @@ class HDF5EventSource(EventSource):
 
             # we can now read both R1 and DL1
             has_muons = "/dl1/event/telescope/muon" in f.root
+            has_sim = "/simulation/event/telescope" in f.root
+            has_trigger = "/simulation/event/telescope" in f.root
+
             datalevels = set(metadata["CTA PRODUCT DATA LEVELS"].split(","))
-            datalevels = datalevels & {
-                "R1",
-                "DL1_IMAGES",
-                "DL1_PARAMETERS",
-                "DL2",
-                "DL1_MUON",
-            }
-            if not datalevels and not has_muons:
+            datalevels = (
+                len(
+                    datalevels
+                    & {
+                        "R1",
+                        "DL1_IMAGES",
+                        "DL1_PARAMETERS",
+                        "DL2",
+                        "DL1_MUON",
+                    }
+                )
+                > 0
+            )
+            if not any([datalevels, has_sim, has_trigger, has_muons]):
                 return False
 
         return True
diff --git a/src/ctapipe/io/hdf5tableio.py b/src/ctapipe/io/hdf5tableio.py
index 988318a7..df751463 100644
--- a/src/ctapipe/io/hdf5tableio.py
+++ b/src/ctapipe/io/hdf5tableio.py
@@ -257,7 +257,9 @@ class HDF5TableWriter(TableWriter):
     def close(self):
         self.h5file.close()
 
-    def _add_column_to_schema(self, table_name, schema, meta, field, name, value):
+    def _add_column_to_schema(
+        self, table_name, schema, meta, field, name, value, time_format
+    ):
         typename = ""
         shape = 1
 
@@ -302,9 +304,13 @@ class HDF5TableWriter(TableWriter):
             schema.columns[name] = coltype(shape=shape, pos=pos)
 
         elif isinstance(value, Time):
-            # TODO: really should use MET, but need a func for that
-            schema.columns[name] = tables.Float64Col(pos=pos)
-            tr = TimeColumnTransform(scale="tai", format="mjd")
+            tr = TimeColumnTransform(scale="tai", format=time_format)
+            value = tr(value)
+
+            typename = value.dtype.name
+            coltype = PYTABLES_TYPE_MAP[typename]
+            shape = value.shape
+            schema.columns[name] = coltype(shape=shape, pos=pos)
             self.add_column_transform(table_name, name, tr)
 
         elif type(value).__name__ in PYTABLES_TYPE_MAP:
@@ -341,7 +347,7 @@ class HDF5TableWriter(TableWriter):
 
         return True
 
-    def _create_hdf5_table_schema(self, table_name, containers):
+    def _create_hdf5_table_schema(self, table_name, containers, time_format):
         """
         Creates a pytables description class for the given containers
         and registers it in the Writer
@@ -384,6 +390,7 @@ class HDF5TableWriter(TableWriter):
                         field=field,
                         name=col_name,
                         value=value,
+                        time_format=time_format,
                     )
                 except ValueError:
                     self.log.warning(
@@ -396,11 +403,13 @@ class HDF5TableWriter(TableWriter):
         meta["CTAPIPE_VERSION"] = ctapipe.__version__
         return meta
 
-    def _setup_new_table(self, table_name, containers):
+    def _setup_new_table(self, table_name, containers, time_format):
         """set up the table. This is called the first time `write()`
         is called on a new table"""
         self.log.debug("Initializing table '%s' in group '%s'", table_name, self._group)
-        meta = self._create_hdf5_table_schema(table_name, containers)
+        meta = self._create_hdf5_table_schema(
+            table_name, containers, time_format=time_format
+        )
 
         if table_name.startswith("/"):
             raise ValueError("Table name must not start with '/'")
@@ -455,7 +464,7 @@ class HDF5TableWriter(TableWriter):
                     raise
         row.append()
 
-    def write(self, table_name, containers):
+    def write(self, table_name, containers, time_format="ctao_high_res"):
         """
         Write the contents of the given container or containers to a table.
         The first call to write  will create a schema and initialize the table
@@ -469,12 +478,15 @@ class HDF5TableWriter(TableWriter):
             name of table to write to
         containers: `ctapipe.core.Container` or `Iterable[ctapipe.core.Container]`
             container to write
+        time_format: str
+            Format to use for storing time columns.
+            Either 'ctao_high_res' (the default) or a format supported by `astropy.time.Time`.
         """
         if isinstance(containers, Container):
             containers = (containers,)
 
         if table_name not in self._schemas:
-            self._setup_new_table(table_name, containers)
+            self._setup_new_table(table_name, containers, time_format=time_format)
 
         self._append_row(table_name, containers)
 
diff --git a/src/ctapipe/io/tableio.py b/src/ctapipe/io/tableio.py
index 442cdaab..ee059630 100644
--- a/src/ctapipe/io/tableio.py
+++ b/src/ctapipe/io/tableio.py
@@ -13,6 +13,7 @@ from ctapipe.compat import COPY_IF_NEEDED
 
 from ..core import Component
 from ..instrument import SubarrayDescription
+from ..time import ctao_high_res_to_time, time_to_ctao_high_res
 
 __all__ = ["TableReader", "TableWriter"]
 
@@ -305,7 +306,7 @@ class ColumnTransform(metaclass=ABCMeta):
 
 
 class TimeColumnTransform(ColumnTransform):
-    """A Column transformation that converts astropy time objects to MJD TAI"""
+    """A Column transformation that converts astropy time objects to a specific format."""
 
     def __init__(self, scale, format):
         self.scale = scale
@@ -315,9 +316,19 @@ class TimeColumnTransform(ColumnTransform):
         """
         Convert an astropy time object to an mjd value in tai scale
         """
+        # ctao_high_res is not implemented as an astropy format but using
+        # conversion functions defined in ctapipe
+        if self.format == "ctao_high_res":
+            return time_to_ctao_high_res(value)
+        # for all other formats, use astropy mechanism
         return getattr(getattr(value, self.scale), self.format)
 
     def inverse(self, value):
+        # ctao_high_res is not implemented as an astropy format but using
+        # conversion functions defined in ctapipe
+        if self.format == "ctao_high_res":
+            return ctao_high_res_to_time(value[..., 0], value[..., 1])
+        # for all other formats, use astropy mechanism
         return Time(value, scale=self.scale, format=self.format, copy=COPY_IF_NEEDED)
 
     def get_meta(self, colname):
diff --git a/src/ctapipe/io/tableloader.py b/src/ctapipe/io/tableloader.py
index a661b143..041f68f5 100644
--- a/src/ctapipe/io/tableloader.py
+++ b/src/ctapipe/io/tableloader.py
@@ -29,7 +29,6 @@ TRUE_IMPACT_GROUP = "/simulation/event/telescope/impact"
 SIMULATION_CONFIG_TABLE = "/configuration/simulation/run"
 SHOWER_DISTRIBUTION_TABLE = "/simulation/service/shower_distribution"
 OBSERVATION_TABLE = "/configuration/observation/observation_block"
-SCHEDULING_TABLE = "/configuration/observation/scheduling_block"
 FIXED_POINTING_GROUP = "/configuration/telescope/pointing"
 POINTING_GROUP = "/dl0/monitoring/telescope/pointing"
 
@@ -364,12 +363,6 @@ class TableLoader(Component):
         """
         return read_table(self.h5file, OBSERVATION_TABLE)
 
-    def read_scheduling_blocks(self):
-        """
-        Read the scheduling information
-        """
-        return read_table(self.h5file, SCHEDULING_TABLE)
-
     def _join_observation_info(self, table):
         obs_table = self.read_observation_information()
         # in v0.17, obs_id had inconsistent dtypes in different tables
diff --git a/src/ctapipe/monitoring/outlier.py b/src/ctapipe/monitoring/outlier.py
index 1c2ee0d3..98a0e338 100644
--- a/src/ctapipe/monitoring/outlier.py
+++ b/src/ctapipe/monitoring/outlier.py
@@ -65,12 +65,6 @@ class RangeOutlierDetector(OutlierDetector):
     ).tag(config=True)
 
     def __call__(self, column):
-        # Validate that the shape of the statistic values has three dimensions
-        if column.ndim != 3:
-            raise ValueError(
-                f"Invalid shape of the column '{column.name}': '{column.shape}'. "
-                "Expected the statistic values of shape (n_entries, n_channels, n_pixels)."
-            )
         # Remove outliers is statistical values out a given range
         outliers = np.logical_or(
             column < self.validity_range[0],
@@ -99,12 +93,6 @@ class MedianOutlierDetector(OutlierDetector):
     ).tag(config=True)
 
     def __call__(self, column):
-        # Validate that the shape of the statistic values has three dimensions
-        if column.ndim != 3:
-            raise ValueError(
-                f"Invalid shape of the column '{column.name}': '{column.shape}'. "
-                "Expected the statistic values of shape (n_entries, n_channels, n_pixels)."
-            )
         # Camera median
         camera_median = np.ma.median(column, axis=2)
         # Detect outliers based on the deviation of the median distribution
@@ -136,12 +124,6 @@ class StdOutlierDetector(OutlierDetector):
     ).tag(config=True)
 
     def __call__(self, column):
-        # Validate that the shape of the statistic values has three dimensions
-        if column.ndim != 3:
-            raise ValueError(
-                f"Invalid shape of the column '{column.name}': '{column.shape}'. "
-                "Expected the statistic values of shape (n_entries, n_channels, n_pixels)."
-            )
         # Camera median
         camera_median = np.ma.median(column, axis=2)
         # Camera std
diff --git a/src/ctapipe/time.py b/src/ctapipe/time.py
new file mode 100644
index 00000000..ec87ea3a
--- /dev/null
+++ b/src/ctapipe/time.py
@@ -0,0 +1,61 @@
+"""Functions handling time, mainly conversion between CTAO timestamps and astropy."""
+import numpy as np
+from astropy.time import Time
+
+EPOCH = Time(0, format="unix_tai", scale="tai")
+DAY_TO_S = 86400
+S_TO_QNS = 4e9
+QNS_TO_S = 0.25e-9
+MAX_UINT32 = np.iinfo(np.uint32).max
+
+
+def ctao_high_res_to_time(seconds, quarter_nanoseconds):
+    """Convert CTAO high resolution timestamp to astropy Time."""
+    # unix_tai accepts two floats for maximum precision
+    # we can just pass integral and fractional part
+    fractional_seconds = quarter_nanoseconds * QNS_TO_S
+    return Time(
+        seconds,
+        fractional_seconds,
+        format="unix_tai",
+        # this is only for displaying iso timestamp, not any actual precision
+        precision=9,
+    )
+
+
+def to_seconds(days):
+    """Return fractional and whole seconds from a number in days."""
+    seconds = days * DAY_TO_S
+    return np.modf(seconds)
+
+
+def time_to_ctao_high_res(time: Time):
+    """
+    Convert astropy Time to CTAO high precision timestamp.
+
+    Out-of-range values are clipped to 0 (before 1970-01-01T00:00:00 TAI) and
+    MAX_UINT32 (after 2106-02-07T06:28:15 TAI) respectively.
+    """
+    # make sure we are in TAI
+    time_tai = time.tai
+
+    # internally, astropy always uses jd values
+    # jd1 is integral and jd2 is in [-0.5, 0.5]
+    # we get the integral and fractional seconds from both jd values
+    # relative to the epoch
+    fractional_seconds_jd1, seconds_jd1 = to_seconds(time_tai.jd1 - EPOCH.jd1)
+    fractional_seconds_jd2, seconds_jd2 = to_seconds(time_tai.jd2 - EPOCH.jd2)
+
+    # add up the integral number of seconds
+    seconds = seconds_jd1 + seconds_jd2
+
+    # convert fractional seconds to quarter nanoseconds
+    fractional_seconds = fractional_seconds_jd1 + fractional_seconds_jd2
+    quarter_nanoseconds = np.round(fractional_seconds * S_TO_QNS)
+
+    result = np.empty(time_tai.shape + (2,), np.uint32)
+    result[..., 0] = np.asanyarray(np.clip(seconds, 0, MAX_UINT32), dtype=np.uint32)
+    result[..., 1] = np.asanyarray(
+        np.clip(0, quarter_nanoseconds, None), dtype=np.uint32
+    )
+    return result
diff --git a/src/ctapipe/tools/calculate_pixel_stats.py b/src/ctapipe/tools/calculate_pixel_stats.py
index 56b9c1aa..3e672d06 100644
--- a/src/ctapipe/tools/calculate_pixel_stats.py
+++ b/src/ctapipe/tools/calculate_pixel_stats.py
@@ -82,8 +82,6 @@ class PixelStatisticsCalculatorTool(Tool):
         TableLoader,
     ] + classes_with_traits(PixelStatisticsCalculator)
 
-    DL1_COLUMN_NAMES = ["image", "peak_time"]
-
     def setup(self):
         # Read the input data with the 'TableLoader'
         self.input_data = self.enter_context(
@@ -139,10 +137,6 @@ class PixelStatisticsCalculatorTool(Tool):
                     f"Column '{self.input_column_name}' not found "
                     f"in the input data for telescope 'tel_id={tel_id}'."
                 )
-            # Check if the dl1 data is gain selected and add an extra dimension for n_channels
-            for col_name in self.DL1_COLUMN_NAMES:
-                if col_name in dl1_table.colnames and dl1_table[col_name].ndim == 2:
-                    dl1_table[col_name] = dl1_table[col_name][:, np.newaxis]
             # Perform the first pass of the statistics calculation
             aggregated_stats = self.stats_calculator.first_pass(
                 table=dl1_table,


Test patch:
diff --git a/src/ctapipe/io/tests/test_datawriter.py b/src/ctapipe/io/tests/test_datawriter.py
index f4924b4c..d5dec4c5 100644
--- a/src/ctapipe/io/tests/test_datawriter.py
+++ b/src/ctapipe/io/tests/test_datawriter.py
@@ -71,10 +71,7 @@ def test_write(tmpdir: Path):
     """
 
     output_path = Path(tmpdir / "events.dl1.h5")
-    source = EventSource(
-        get_dataset_path("gamma_prod5.simtel.zst"),
-        focal_length_choice="EQUIVALENT",
-    )
+    source = EventSource(get_dataset_path("gamma_prod5.simtel.zst"))
     calibrate = CameraCalibrator(subarray=source.subarray)
 
     with DataWriter(
@@ -149,10 +146,7 @@ def test_roundtrip(tmpdir: Path):
     """
 
     output_path = Path(tmpdir / "events.DL1DL2.h5")
-    source = EventSource(
-        get_dataset_path("gamma_prod5.simtel.zst"),
-        focal_length_choice="EQUIVALENT",
-    )
+    source = EventSource(get_dataset_path("gamma_prod5.simtel.zst"))
     calibrate = CameraCalibrator(subarray=source.subarray)
 
     events = []
diff --git a/src/ctapipe/io/tests/test_hdf5.py b/src/ctapipe/io/tests/test_hdf5.py
index 65b9e908..03d35566 100644
--- a/src/ctapipe/io/tests/test_hdf5.py
+++ b/src/ctapipe/io/tests/test_hdf5.py
@@ -20,6 +20,7 @@ from ctapipe.core.container import Container, Field
 from ctapipe.io import read_table
 from ctapipe.io.datalevels import DataLevel
 from ctapipe.io.hdf5tableio import HDF5TableReader, HDF5TableWriter
+from ctapipe.time import time_to_ctao_high_res
 
 
 @pytest.fixture(scope="session")
@@ -658,9 +659,8 @@ def test_column_transforms(tmp_path):
             "mytable", "image", FixedPointColumnTransform(100, 0, np.float64, np.int32)
         )
         # add user generated transform for the "value" column
-        writer.write("mytable", cont)
+        writer.write("mytable", cont, time_format="mjd")
 
-    # check that we get a length-3 array when reading back
     with HDF5TableReader(tmp_file, mode="r") as reader:
         data = next(reader.read("/data/mytable", SomeContainer))
         assert data.current.value == 1e6
@@ -741,7 +741,14 @@ def test_column_transforms_regexps(tmp_path):
         assert data.hillas_y == 10
 
 
-def test_time(tmp_path):
+@pytest.mark.parametrize(
+    ("time_format", "tolerance"),
+    [
+        ("mjd", 0.1 * u.us),
+        ("ctao_high_res", 0.01 * u.ns),
+    ],
+)
+def test_time(tmp_path, time_format, tolerance):
     tmp_file = tmp_path / "test_time.hdf5"
 
     class TimeContainer(Container):
@@ -751,15 +758,16 @@ def test_time(tmp_path):
     container = TimeContainer(time=time)
 
     with HDF5TableWriter(tmp_file, group_name="data") as writer:
-        # add user generated transform for the "value" column
-        writer.write("table", container)
+        writer.write("table", container, time_format=time_format)
 
     with HDF5TableReader(tmp_file, mode="r") as reader:
         for data in reader.read("/data/table", TimeContainer):
             assert isinstance(data.time, Time)
             assert data.time.scale == "tai"
-            assert data.time.format == "mjd"
-            assert (data.time - time).to(u.s).value < 1e-7
+            assert data.time.format == (
+                "unix_tai" if time_format == "ctao_high_res" else time_format
+            )
+            assert np.abs((data.time - time).to(u.s)) < tolerance
 
 
 def test_filters(tmp_path):
@@ -1019,3 +1027,46 @@ def test_hdf5_datalevels(input_type, dl2_shower_geometry_file):
         DataLevel.DL1_PARAMETERS,
         DataLevel.DL2,
     }
+
+
+def test_column_transform_high_res_timestamp(tmp_path):
+    """Test high-res timestamp column transformation."""
+    from ctapipe.containers import NAN_TIME
+    from ctapipe.io.tableio import TimeColumnTransform
+
+    tmp_file = tmp_path / "test_high_res_time.hdf5"
+
+    class SomeContainer(Container):
+        default_prefix = ""
+        time = Field(NAN_TIME)
+
+    # low-resolution timestamp only has ~us precision, so test we can actually
+    # store much smaller differences
+    dt = [0, 0.25, 0.5, 0.75, 1.0, 1.25] * u.ns
+    times = Time("2025-01-01T00:00:12.123456789") + dt
+
+    with HDF5TableWriter(tmp_file) as writer:
+        writer.add_column_transform(
+            "events", "time", TimeColumnTransform("tai", "ctao_high_res")
+        )
+
+        for t in times:
+            writer.write("events", SomeContainer(time=t))
+
+    with HDF5TableReader(tmp_file, mode="r") as reader:
+        times_read = [c.time for c in reader.read("/events", SomeContainer)]
+        assert isinstance(times_read[0], Time)
+        assert times_read[0].scale == "tai"
+        assert times_read[0].format == "unix_tai"
+
+        # convert list of scalar time to 1d Time object
+        times_read = Time(times_read)
+        # check absolute error due to floating point imprecision
+        roundtrip_error = (times - times_read).to_value(u.ns)
+        np.testing.assert_array_less(np.abs(roundtrip_error), 0.02)
+
+        # check that times are exactly equal in ctao representation
+        np.testing.assert_array_equal(
+            time_to_ctao_high_res(times),
+            time_to_ctao_high_res(times_read),
+        )
diff --git a/src/ctapipe/io/tests/test_hdf5eventsource.py b/src/ctapipe/io/tests/test_hdf5eventsource.py
index fc8fe6bf..f15a8d84 100644
--- a/src/ctapipe/io/tests/test_hdf5eventsource.py
+++ b/src/ctapipe/io/tests/test_hdf5eventsource.py
@@ -30,7 +30,7 @@ def test_is_compatible(compatible_file, request):
 def test_metadata(dl1_file):
     with HDF5EventSource(input_url=dl1_file) as source:
         assert source.is_simulation
-        assert source.datamodel_version == (6, 0, 0)
+        assert source.datamodel_version == (7, 0, 0)
         assert set(source.datalevels) == {
             DataLevel.DL1_IMAGES,
             DataLevel.DL1_PARAMETERS,
diff --git a/src/ctapipe/io/tests/test_table_loader.py b/src/ctapipe/io/tests/test_table_loader.py
index 3e0aa171..398b58d4 100644
--- a/src/ctapipe/io/tests/test_table_loader.py
+++ b/src/ctapipe/io/tests/test_table_loader.py
@@ -369,21 +369,6 @@ def test_read_shower_distributions(dl2_merged_file):
         assert np.all(histograms["histogram"].sum(axis=(1, 2)) == [2000, 1000])
 
 
-def test_read_scheduling_blocks(dl2_merged_file):
-    from ctapipe.io import TableLoader
-
-    with TableLoader(dl2_merged_file) as table_loader:
-        scheduling = table_loader.read_scheduling_blocks()
-        assert len(scheduling) == 2
-        np.testing.assert_array_equal(scheduling["sb_id"], [4, 1])
-        np.testing.assert_array_equal(scheduling["sb_type"], [0, 0])
-        np.testing.assert_array_equal(
-            scheduling["producer_id"], ["simulation", "simulation"]
-        )
-        np.testing.assert_array_equal(scheduling["observing_mode"], [-1, -1])
-        np.testing.assert_array_equal(scheduling["pointing_mode"], [1, 1])
-
-
 def test_read_unavailable_telescope(dl2_shower_geometry_file):
     """Reading a telescope that is not part of the subarray of the file should fail."""
     from ctapipe.io import TableLoader
diff --git a/src/ctapipe/io/tests/test_write_table.py b/src/ctapipe/io/tests/test_write_table.py
index e546681e..8ca27f68 100644
--- a/src/ctapipe/io/tests/test_write_table.py
+++ b/src/ctapipe/io/tests/test_write_table.py
@@ -6,7 +6,14 @@ from astropy.table import Table
 from astropy.time import Time
 
 
-def test_write_table(tmp_path):
+@pytest.mark.parametrize(
+    ("time_format", "tolerance"),
+    [
+        ("mjd", 1 * u.us),
+        ("ctao_high_res", 0.01 * u.ns),
+    ],
+)
+def test_write_table(tmp_path, time_format, tolerance):
     """Test write_table"""
     from ctapipe.io.astropy_helpers import read_table, write_table
 
@@ -26,7 +33,7 @@ def test_write_table(tmp_path):
     output_path = tmp_path = tmp_path / "table.h5"
     table_path = "/foo/bar"
 
-    write_table(table, output_path, table_path)
+    write_table(table, output_path, table_path, time_format=time_format)
     read = read_table(output_path, table_path)
 
     for name, column in table.columns.items():
@@ -37,7 +44,9 @@ def test_write_table(tmp_path):
 
         # time conversion is not lossless
         if name == "time":
-            assert np.allclose(column.tai.mjd, read[name].tai.mjd)
+            np.testing.assert_array_less(
+                np.abs((column - read[name]).to(u.ns)), tolerance
+            )
         else:
             assert np.all(column == read[name]), f"Column {name} differs after reading"
 
@@ -50,7 +59,7 @@ def test_write_table(tmp_path):
         write_table(table, output_path, table_path)
 
     # test we can append
-    write_table(table, output_path, table_path, append=True)
+    write_table(table, output_path, table_path, append=True, time_format=time_format)
     read = read_table(output_path, table_path)
     assert len(read) == 2 * len(table)
 
diff --git a/src/ctapipe/monitoring/tests/test_outlier.py b/src/ctapipe/monitoring/tests/test_outlier.py
index 93ea8ef1..da7d7619 100644
--- a/src/ctapipe/monitoring/tests/test_outlier.py
+++ b/src/ctapipe/monitoring/tests/test_outlier.py
@@ -3,12 +3,10 @@ Tests for OutlierDetector and related functions
 """
 
 import numpy as np
-import pytest
 from astropy.table import Table
 
 from ctapipe.monitoring.outlier import (
     MedianOutlierDetector,
-    OutlierDetector,
     RangeOutlierDetector,
     StdOutlierDetector,
 )
@@ -104,28 +102,3 @@ def test_std_detection(example_subarray):
     # Check if outliers where detected correctly
     np.testing.assert_array_equal(ff_outliers, ff_expected_outliers)
     np.testing.assert_array_equal(ped_outliers, ped_expected_outliers)
-
-
-def test_check_for_column_shape(example_subarray):
-    """test the check for the shape of the column"""
-
-    # Create dummy data for testing
-    rng = np.random.default_rng(0)
-    # Distribution mimics the median values of gain-selected charge images of flat-field events
-    median = rng.normal(77.0, 0.6, size=(50, 1855))
-    # Create astropy table
-    table = Table([median], names=("median",))
-    # Initialize the outlier detector based on the deviation from the camera median
-    outlier_detectors = [
-        "RangeOutlierDetector",
-        "MedianOutlierDetector",
-        "StdOutlierDetector",
-    ]
-    for outlier_detector_name in outlier_detectors:
-        outlier_detector = OutlierDetector.from_name(
-            outlier_detector_name,
-            subarray=example_subarray,
-        )
-        # Check if ValueError is raised when the provided column shape is invalid
-        with pytest.raises(ValueError, match="Invalid shape of the column"):
-            outlier_detector(table["median"])
diff --git a/src/ctapipe/tests/test_time.py b/src/ctapipe/tests/test_time.py
new file mode 100644
index 00000000..0e531ba9
--- /dev/null
+++ b/src/ctapipe/tests/test_time.py
@@ -0,0 +1,44 @@
+import astropy.units as u
+import numpy as np
+import pytest
+from astropy.time import Time
+
+from ctapipe.containers import NAN_TIME
+
+roundtrip_times = [
+    "2020-01-01T00:00:00.12345678925",
+    "2020-01-01T00:00:00.1234567895",
+    "2020-01-01T00:00:00.12345678975",
+    "2020-01-01T00:00:00.123456790",
+]
+
+
+@pytest.mark.parametrize("timestamp", roundtrip_times)
+def test_ctao_high_round_trip(timestamp):
+    from ctapipe.time import ctao_high_res_to_time, time_to_ctao_high_res
+
+    # note: precision=9 only affects text conversion, not actual precision
+    time = Time(timestamp, scale="tai", precision=9)
+    time_s, time_qns = time_to_ctao_high_res(time)
+    time_back = ctao_high_res_to_time(time_s, time_qns)
+
+    roundtrip_error = (time - time_back).to_value(u.ns)
+    np.testing.assert_almost_equal(roundtrip_error, 0.0)
+
+
+test_data = [
+    (Time(0, 12.25e-9, format="unix_tai"), 0, 49),
+    (Time(12345, 12.25e-9, format="unix_tai"), 12345, 49),
+    (Time(65123, 123456.25e-9, format="unix_tai"), 65123, 493825),
+    (Time("2200-01-02T00:00:00"), np.iinfo(np.uint32).max, 0),
+    (NAN_TIME, 0, 0),
+]
+
+
+@pytest.mark.parametrize(("time", "expected_s", "expected_qns"), test_data)
+def test_time_to_ctao_high_res(time, expected_s, expected_qns):
+    from ctapipe.time import time_to_ctao_high_res
+
+    time_s, time_qns = time_to_ctao_high_res(time)
+    assert time_s == expected_s
+    assert time_qns == expected_qns
diff --git a/src/ctapipe/tools/tests/test_calculate_pixel_stats.py b/src/ctapipe/tools/tests/test_calculate_pixel_stats.py
index 42c842ed..9e2e9674 100644
--- a/src/ctapipe/tools/tests/test_calculate_pixel_stats.py
+++ b/src/ctapipe/tools/tests/test_calculate_pixel_stats.py
@@ -29,13 +29,6 @@ def test_calculate_pixel_stats_tool(tmp_path, dl1_image_file):
                 "stats_aggregator_type": [
                     ("id", tel_id, "PlainAggregator"),
                 ],
-                "outlier_detector_list": [
-                    {
-                        "apply_to": "mean",
-                        "name": "MedianOutlierDetector",
-                        "config": {"median_range_factors": [-2.0, 2.0]},
-                    }
-                ],
             },
             "PlainAggregator": {
                 "chunk_size": 1,
@@ -57,13 +50,13 @@ def test_calculate_pixel_stats_tool(tmp_path, dl1_image_file):
     )
     # Check that the output file has been created
     assert monitoring_file.exists()
-    # Check if the shape of the aggregated statistic values has three dimension
+    # Check that the output file is not empty
     assert (
         read_table(
             monitoring_file,
             path=f"/dl1/monitoring/telescope/statistics/tel_{tel_id:03d}",
-        )["mean"].ndim
-        == 3
+        )["mean"]
+        is not None
     )
     # Read subarray description from the created monitoring file
     subarray = SubarrayDescription.from_hdf(monitoring_file)
diff --git a/src/ctapipe/tools/tests/test_process.py b/src/ctapipe/tools/tests/test_process.py
index 424e10d1..4fe6bfea 100644
--- a/src/ctapipe/tools/tests/test_process.py
+++ b/src/ctapipe/tools/tests/test_process.py
@@ -509,6 +509,17 @@ def test_only_trigger_and_simulation(tmp_path):
         assert "tels_with_trigger" in events.colnames
         assert "true_energy" in events.colnames
 
+    # test that telescope trigger times are different from each other
+    # this needs ns resolution, which is enabled by storing them as
+    # high resolution timestamps
+    with EventSource(output) as source:
+        for e in source:
+            it = iter(e.trigger.tel.values())
+            tel1 = next(it)
+            tel2 = next(it)
+            diff = (tel1.time - tel2.time).to_value(u.ns)
+            assert diff != 0
+
 
 @pytest.mark.parametrize(
     ("input_url", "args"),