[ESSIMAGING] Move tiff dumping method to essimaging and make it more generic.

packages/essimaging/src/ess/imaging/io.py

@@ -0,0 +1,81 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Scipp contributors (https://github.com/scipp)
+import io
+from pathlib import Path
+
+import scipp as sc
+import scippnexus as snx
+import scitiff
+
+
+def _add_to_event_time_offset_in_case_of_pulse_skipping(
+    event_time_zero: sc.Variable,
+    pulse_stride: int,
+    pulse_period: sc.Variable,
+    pulse_stride_offset: int = 0,
+) -> sc.Variable:
+    _pulse_period = pulse_period.to(unit=event_time_zero.unit)
+    etz = event_time_zero - sc.datetime(0, unit=event_time_zero.unit)
+    # The offset is used to place some etz value in the center of a binning
+    # where the bins have constant width pulse_period.
+    # That way small deviations in etz will not move the etz to the next
+    # or previous bin and each subsequent pulse will have a different index.
+    offset = _pulse_period / 2 - etz.nanmin() % _pulse_period
+    index = (((etz + offset) // _pulse_period) + pulse_stride_offset) % pulse_stride
+    return index * pulse_period
+
+
+def tiff_from_event_data(
+    nexus_file_name: str | Path | io.BytesIO,
+    output_path: str | Path | io.BytesIO,
+    *,
+    time_bins: int | sc.Variable,
+    pulse_stride: int,
+    pulse_stride_offset: int = 0,
+    detector_group_path: str = "/entry/instrument/event_mode_detectors/timepix3",
+    event_data_field_path: str = "timepix3_events",
+) -> None:
+    '''
+    Write a tiff image file representing the data from the nexus file.
+
+    Parameters
+    ------------
+    nexus_file_name:
+        The file name of the nexus file to write to tiff.
+    output_path:
+        Where to write the tiff file.
+    time_bins:
+        The number of time slices the image should have.
+    pulse_stride:
+        The pulse stride that was used when doing the measurement.
+    pulse_stride_offset:
+        The offset to the pulse stride.
+        For example, if pulse_stride is 2, offset should be
+        0 if the first event_time_zero set is the beginning of the pulse,
+        1 if the first event_time_zero set is the second half of the pulse.
+    detector_group_path:
+        Path to the event data group in the nexus file.
+    event_data_field_path:
+        Path to the event data group in the nexus file.
+    '''
+    with snx.File(nexus_file_name) as f:
+        data = f[detector_group_path][()][event_data_field_path]
+
+    data.bins.coords['event_time_offset'] += (
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            data.bins.coords['event_time_zero'],
+            pulse_stride=pulse_stride,
+            pulse_stride_offset=pulse_stride_offset,
+            pulse_period=sc.scalar(1 / 14, unit="s").to(
+                unit=data.bins.coords['event_time_offset'].unit
+            ),
+        )
+    )
+    # In case dimension names fall back to default ones.
+    ydim_name = 'dim_0' if 'dim_0' in data.dims else 'y_pixel_offset'
+    xdim_name = 'dim_1' if 'dim_1' in data.dims else 'x_pixel_offset'
+    image = data.hist(event_time_offset=time_bins).rename_dims(
+        {'event_time_offset': 't', ydim_name: 'y', xdim_name: 'x'}
+    )
+    image = image.drop_coords([c for c in image.coords if image.coords[c].ndim > 1])
+    scitiff.save_scitiff(image, output_path)

packages/essimaging/src/ess/ymir/io.py

@@ -10,9 +10,9 @@
 
 import scipp as sc
 import scippnexus as snx
-import scitiff
 from tifffile import imwrite
 
+from ess.imaging.io import tiff_from_event_data
 from ess.reduce.nexus.types import FilePath
 
 from .types import (
@@ -362,22 +362,6 @@ def export_image_stacks_as_tiff(
             )
 
 
-def _add_to_event_time_offset_in_case_of_pulse_skipping(
-    event_time_zero: sc.Variable,
-    pulse_stride: int,
-    pulse_period: sc.Variable,
-) -> sc.Variable:
-    _pulse_period = pulse_period.to(unit=event_time_zero.unit)
-    etz = event_time_zero - sc.datetime(0, unit=event_time_zero.unit)
-    # The offset is used to place some etz value in the center of a binning
-    # where the bins have constant width pulse_period.
-    # That way small deviations in etz will not move the etz to the next
-    # or previous bin and each subsequent pulse will have a different index.
-    offset = _pulse_period / 2 - etz.nanmin() % _pulse_period
-    index = ((etz + offset) // _pulse_period) % pulse_stride
-    return index * pulse_period
-
-
 def tiff_from_nexus(
     nexus_file_name: str | Path | io.BytesIO,
     output_path: str | Path | io.BytesIO,
@@ -388,6 +372,8 @@ def tiff_from_nexus(
     '''
     Write a tiff image file representing the data from the nexus file.
 
+    **Use ess.imaging.io.tiff_from_event_data instead.**
+
     Parameters
     ------------
     nexus_file_name:
@@ -399,23 +385,15 @@ def tiff_from_nexus(
     pulse_stride:
         The pulse stride that was used when doing the measurement.
     '''
-    with snx.File(nexus_file_name) as f:
-        data = f['/entry/instrument/event_mode_detectors/timepix3'][()][
-            'timepix3_events'
-        ]
-    data.bins.coords['event_time_offset'] += (
-        _add_to_event_time_offset_in_case_of_pulse_skipping(
-            data.bins.coords['event_time_zero'],
-            pulse_stride=pulse_stride,
-            pulse_period=sc.scalar(1 / 14, unit="s").to(
-                unit=data.bins.coords['event_time_offset'].unit
-            ),
-        )
+    import warnings
+
+    warnings.warn(
+        category=DeprecationWarning,
+        message="Tiff file IO helper moved to ess.imaging.io. "
+        "Use ess.imaging.io.tiff_from_event_data instead.",
+        stacklevel=2,
     )
-    image = (
-        sc.concat([data], dim='c')
-        .hist(event_time_offset=time_bins)
-        .rename_dims(event_time_offset='t', dim_0='y', dim_1='x')
+
+    tiff_from_event_data(
+        nexus_file_name, output_path, time_bins=time_bins, pulse_stride=pulse_stride
     )
-    image = image.drop_coords([c for c in image.coords if image.coords[c].ndim > 1])
-    scitiff.save_scitiff(image, output_path)

packages/essimaging/tests/imaging/tiff_io_test.py

@@ -0,0 +1,76 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Scipp contributors (https://github.com/scipp)
+
+from pathlib import Path
+
+import ess.odin.data
+import scipp as sc
+from scipp.testing import assert_identical
+from scitiff.io import load_scitiff
+
+from ess.imaging.io import (
+    _add_to_event_time_offset_in_case_of_pulse_skipping,
+    tiff_from_event_data,
+)
+
+
+def test_tiff_dumping_helper(tmp_path: Path):
+    small_iron_image = ess.odin.data.iron_simulation_sample_small()
+    output_path = tmp_path / "dump_timepix.tiff"
+
+    tiff_from_event_data(
+        small_iron_image,
+        output_path,
+        time_bins=20,
+        pulse_stride=2,
+    )
+    # Test if the saved tiff file has expected output.
+    loaded = load_scitiff(file_path=output_path, only_image=True)
+    assert loaded.sizes['t'] == 20
+    assert 'x_pixel_offset' in loaded.coords
+    assert 'y_pixel_offset' in loaded.coords
+
+
+def test_correct_event_time_offset() -> None:
+    assert_identical(
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            sc.datetimes(dims='t', values=[0, 1, 2], unit='s'),
+            pulse_stride=2,
+            pulse_period=sc.scalar(1, unit='s'),
+        ),
+        sc.array(dims='t', values=[0, 1.0, 0], unit='s'),
+    )
+    assert_identical(
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            sc.datetimes(dims='t', values=[0, 1, 2], unit='s'),
+            pulse_stride=2,
+            pulse_period=sc.scalar(1, unit='s'),
+            pulse_stride_offset=1,
+        ),
+        sc.array(dims='t', values=[1.0, 0, 1.0], unit='s'),
+    )
+    assert_identical(
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            sc.datetimes(dims='t', values=[10, 999, 2100], unit='ms'),
+            pulse_stride=2,
+            pulse_period=sc.scalar(1, unit='s'),
+        ),
+        sc.array(dims='t', values=[0, 1.0, 0], unit='s'),
+    )
+    assert_identical(
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            sc.datetimes(dims='t', values=[-100, 999, 2100], unit='ms'),
+            pulse_stride=3,
+            pulse_period=sc.scalar(1, unit='s'),
+        ),
+        sc.array(dims='t', values=[2, 0.0, 1], unit='s'),
+    )
+    assert_identical(
+        _add_to_event_time_offset_in_case_of_pulse_skipping(
+            sc.datetimes(dims='t', values=[-100, 999, 2100], unit='ms'),
+            pulse_stride=3,
+            pulse_stride_offset=2,
+            pulse_period=sc.scalar(1, unit='s'),
+        ),
+        sc.array(dims='t', values=[1, 2.0, 0], unit='s'),
+    )

packages/essimaging/tests/ymir/io_test.py

@@ -7,12 +7,10 @@
     FilePath,
     ImageDetectorName,
     RotationMotionSensorName,
-    _add_to_event_time_offset_in_case_of_pulse_skipping,
     load_nexus_histogram_mode_detector,
     load_nexus_rotation_logs,
 )
 from ess.ymir.types import DEFAULT_HISTOGRAM_PATH
-from scipp.testing import assert_identical
 
 
 def test_nexus_histogram_mode_detector_loading_warnings() -> None:
@@ -37,30 +35,3 @@ def test_nexus_rotation_logs_loading() -> None:
         ),
         sc.DataArray,
     )
-
-
-def test_correct_event_time_offset() -> None:
-    assert_identical(
-        _add_to_event_time_offset_in_case_of_pulse_skipping(
-            sc.datetimes(dims='t', values=[0, 1, 2], unit='s'),
-            pulse_stride=2,
-            pulse_period=sc.scalar(1, unit='s'),
-        ),
-        sc.array(dims='t', values=[0, 1.0, 0], unit='s'),
-    )
-    assert_identical(
-        _add_to_event_time_offset_in_case_of_pulse_skipping(
-            sc.datetimes(dims='t', values=[10, 999, 2100], unit='ms'),
-            pulse_stride=2,
-            pulse_period=sc.scalar(1, unit='s'),
-        ),
-        sc.array(dims='t', values=[0, 1.0, 0], unit='s'),
-    )
-    assert_identical(
-        _add_to_event_time_offset_in_case_of_pulse_skipping(
-            sc.datetimes(dims='t', values=[-100, 999, 2100], unit='ms'),
-            pulse_stride=3,
-            pulse_period=sc.scalar(1, unit='s'),
-        ),
-        sc.array(dims='t', values=[2, 0.0, 1], unit='s'),
-    )