API Reference

Complete API reference for NeuNorm 2.0, generated from the source docstrings.

Data models

Core data models for NeuNorm 2.0.

Pydantic model for event-mode neutron data.

class neunorm.data_models.core.EventData(*, tof, x, y, chip_id=None, pulse_id=None, file_path, total_events, tof_clock=25.0)

Bases: BaseModel

Event-mode neutron data container.

Represents raw list-mode events from TPX3/TPX4 detectors with pixel coordinates and time-of-flight values.

Parameters:

• tof (np.ndarray) – Time-of-flight values in nanoseconds (1D array)

• x (np.ndarray) – X pixel coordinates (1D array, same length as tof)

• y (np.ndarray) – Y pixel coordinates (1D array, same length as tof)

• chip_id (np.ndarray or None, optional) – Chip ID for each event (0-3 for quad detector); 1D integer array, same length as tof if present (default: None)

• pulse_id (np.ndarray or None, optional) – Reconstructed pulse ID for each event; 1D integer array, same length as tof if present (default: None)

• file_path (Path) – Source file path

• total_events (int) – Total number of events

• tof_clock (float) – TOF clock period in nanoseconds (default: 25.0 for TPX3)

Examples

>>> import numpy as np
>>> from pathlib import Path
>>> events = EventData(
...     tof=np.array([1000, 2000, 3000], dtype=np.int64),
...     x=np.array([100, 200, 150], dtype=np.int32),
...     y=np.array([250, 300, 275], dtype=np.int32),
...     file_path=Path('data.h5'),
...     total_events=3,
...     tof_clock=25.0
... )

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_array_1d(v)

Ensure arrays are 1D

classmethod validate_coordinate_dtype(v)

Ensure pixel coordinates are integer type

classmethod validate_tof_dtype(v)

Ensure TOF values are numeric

classmethod validate_optional_array_1d(v)

Ensure optional arrays are 1D if present

classmethod validate_chip_id_dtype(v)

Ensure chip IDs are integer type if present

classmethod validate_pulse_id_dtype(v)

Ensure pulse IDs are integer type if present

classmethod validate_file_exists(v)

Validate file path (can be non-existent for simulated data)

validate_lengths()

Validate all arrays have same length (runs automatically)

Region-of-interest data model for NeuNorm 2.0.

class neunorm.data_models.roi.ROI(*, x0, y0, x1=None, y1=None, width=None, height=None, inclusive=False)

Bases: BaseModel

Rectangular region of interest with named, self-documenting bounds.

Define it either by explicit stop indices or by size — the two forms are equivalent:

ROI(x0=10, y0=20, x1=30, y1=40)          # exclusive stops
ROI(x0=10, y0=20, width=20, height=20)   # the same 20x20 region

Stop indices are exclusive (Python slice semantics), matching apply_roi, apply_air_region_correction and the background_roi flux proxy. An ROI may be passed anywhere those APIs accept an (x0, y0, x1, y1) tuple; the bare-tuple form keeps working unchanged for backward compatibility.

Parameters:

• x0 (int) – Lower (inclusive) pixel bounds in x and y.

• y0 (int) – Lower (inclusive) pixel bounds in x and y.

• x1 (int, optional) – Upper (exclusive) pixel bounds. Provide these or width/height.

• y1 (int, optional) – Upper (exclusive) pixel bounds. Provide these or width/height.

• width (int, optional) – Extent in x and y; x1 = x0 + width and y1 = y0 + height. Provide these or x1/y1.

• height (int, optional) – Extent in x and y; x1 = x0 + width and y1 = y0 + height. Provide these or x1/y1.

• inclusive (bool, optional) – Interpret the upper bounds as inclusive (default False = exclusive Python-slice semantics). When True, the resolved x1/y1 are bumped by one so the region spans (width + 1) x (height + 1) pixels (and an explicit x1/y1 is included). This is the legacy NeuNorm 1.x / iBeatles convention; as_bounds() always returns exclusive stops, so the rest of the library stays exclusive. Bare tuples are always exclusive — use the ROI type to opt into inclusive extents.

Examples

>>> ROI(x0=10, y0=20, x1=30, y1=40).as_bounds()
(10, 20, 30, 40)
>>> ROI(x0=10, y0=20, width=20, height=20).as_bounds()
(10, 20, 30, 40)
>>> ROI(x0=10, y0=20, width=20, height=20, inclusive=True).as_bounds()
(10, 20, 31, 41)

as_bounds()

Return the ROI as an (x0, y0, x1, y1) tuple with exclusive stop indices.

Return type:

tuple[int, int, int, int]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

neunorm.data_models.roi.as_roi_bounds(roi)

Coerce an ROI (or a bare 4-element (x0, y0, x1, y1) sequence) to a bounds tuple.

A bare sequence is returned as a plain tuple so downstream code sees a consistent (x0, y0, x1, y1) form regardless of how the ROI was specified. Element-type/bounds validation is left to the consumer (and to ROI for the named form); only the 4-element length is enforced here so a malformed bare sequence fails fast at one place.

Parameters:

roi (ROI | tuple[int, int, int, int] | list[int])

Return type:

tuple[int, int, int, int]

TOF-specific data models for NeuNorm 2.0.

Includes binning configurations for TOF/Energy/Wavelength spaces.

class neunorm.data_models.tof.BinningConfig(*, bins=5000, bin_space='energy', use_log_bin=True, energy_range=None, wavelength_range=None, tof_range=None)

Bases: BaseModel

Configuration for TOF/Energy/Wavelength binning in neutron imaging.

Supports three binning domains: 1. Energy (eV) - for resonance imaging (nuclear cross-sections) 2. Wavelength (Å) - for Bragg edge imaging (crystallography) 3. TOF (ns) - for raw time-of-flight binning

Each domain can use linear or logarithmic spacing.

Parameters:

• bins (int) – Number of bins (default: 5000)

• bin_space (Literal['tof', 'energy', 'wavelength']) – Binning domain (default: ‘energy’)

• use_log_bin (bool) – Use logarithmic spacing (default: True)

• energy_range (tuple[float, float] or None) – Energy range (E_min, E_max) in eV. Required for bin_space=’energy’

• wavelength_range (tuple[float, float] or None) – Wavelength range (λ_min, λ_max) in Angstrom. Required for bin_space=’wavelength’

• tof_range (tuple[float, float] or None) – TOF range (t_min, t_max) in nanoseconds. Optional for bin_space=’tof’

Examples

>>> # Resonance imaging (energy space, logarithmic)
>>> config = BinningConfig(bins=5000, bin_space='energy', energy_range=(1.0, 100.0))

>>> # Bragg edge imaging (wavelength space, linear)
>>> config = BinningConfig(
...     bins=3000,
...     bin_space='wavelength',
...     wavelength_range=(0.5, 3.0),
...     use_log_bin=False
... )

>>> # Raw TOF binning (full range)
>>> config = BinningConfig(bins=10000, bin_space='tof')

classmethod validate_energy_range_values(v)

Validate energy range has correct format and values

classmethod validate_wavelength_range_values(v)

Validate wavelength range has correct format and values

classmethod validate_tof_range_values(v)

Validate TOF range has correct format and values

validate_range_required()

Ensure required range is provided based on bin_space

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Loaders

Utility module for loading stacks of images in various formats.

neunorm.loaders.stack_loader.load_stack(paths)

Load a stack of images from the given file paths, supporting both TIFF and FITS formats.

Check the extension of the first file in the list and call the appropriate loader function load_tiff_stack or load_fits_stack.

Verify all files have the same extension and raise an error if not.

Parameters:

paths (Sequence[str | Path])

Return type:

DataArray

TIFF loader for NeuNorm.

Loads TIFF stacks as scipp DataArrays.

neunorm.loaders.tiff_loader.load_tiff_stack(paths, tof_edges=None)

Load TIFF stack as scipp DataArray with variance tracking.

Uses Pillow (PIL) to read TIFF images and constructs a scipp DataArray.

Parameters:

• paths (Sequence[str | Path]) – List of paths to TIFF files

• tof_edges (Optional[np.ndarray]) – Time-of-flight values for the first dimension. Accepts either bin edges (N+1) or bin centers (N), where N is the number of images in the loaded stack.

Returns:

DataArray with dimensions (TOF/image, y, x)

• dims: [‘TOF’, ‘y’, ‘x’] if tof_edges provided, else [‘N_image’, ‘y’, ‘x’]

• coords: y, x pixel indices, and optionally TOF. Additionally, TIFF metadata is added as coordinates. Each metadata coordinate may be scalar (when its value is constant across the stack and not float-convertible) or stack-dimensioned (when values are float-convertible or differ across files).

Return type:

sc.DataArray

FITS loader for NeuNorm based on astropy.

Loads FITS files into scipp DataArrays.

neunorm.loaders.fits_loader.load_fits_stack(paths, tof_edges=None)

Load FITS stack as scipp DataArray with metadata and optional TOF coordinates.

Handles:

• List of FITS files (stacked along the first dimension)

• Metadata extraction from FITS headers

Parameters:

• paths (Sequence[str | Path]) – List of paths to FITS files

• tof_edges (Optional[np.ndarray]) – Time-of-flight values for the first dimension. Accepts either bin edges (N+1) or bin centers (N), where N is the number of images in the loaded stack.

Returns:

DataArray with dimensions (TOF/image, y, x)

• dims: [‘TOF’, ‘y’, ‘x’] if tof_edges provided, else [‘N_image’, ‘y’, ‘x’]

• coords: y, x pixel indices, and optionally TOF. Additionally, FITS header keys are added as (unaligned) coordinates. The COMMENT and HISTORY keys are skipped. A key whose value is constant across the stack is stored as a scalar coordinate; a key whose value differs across files is stored as an array coordinate along the stack dimension.

Return type:

sc.DataArray

Event data loader for TPX3/TPX4 HDF5 files.

Loads raw event-mode data from HDF5 files containing tof, x, y arrays.

neunorm.loaders.event_loader.load_event_data(file_path, tof_clock=25.0, max_events=None)

Load event-mode data from HDF5 file.

Expected HDF5 structure: - ‘tof’: Time-of-flight values (int32/int64 array) - ‘x’: X pixel coordinates (int16/int32 array) - ‘y’: Y pixel coordinates (int16/int32 array)

Parameters:

• file_path (str or Path) – Path to HDF5 file containing event data

• tof_clock (float, optional) – TOF clock period in nanoseconds (default: 25.0 for TPX3) Raw TOF ticks are multiplied by this value

• max_events (int, optional) – Maximum number of events to load (for testing/memory limits) If None, loads all events

Returns:

Pydantic model containing event arrays and metadata

Return type:

EventData

Raises:

• FileNotFoundError – If file doesn’t exist

• KeyError – If HDF5 file missing required fields (tof, x, y)

Examples

>>> events = load_event_data('run_12557.h5', tof_clock=25.0)
>>> print(f"Loaded {events.total_events:,} events")
>>> print(f"TOF range: {events.tof.min()} - {events.tof.max()} ns")

neunorm.loaders.event_loader.load_event_nexus(file_path, detector_bank='bank1', detector_shape=(512, 512), event_id_offset=0, max_events=None)

Load event-mode data from SNS NeXus HDF5 file.

Reads from the SNS NeXus event bank structure: /entry/<bank>_events/ with datasets: - event_id: Linearized pixel detector IDs - event_time_offset: Time-of-flight values (in microseconds)

Parameters:

• file_path (str or Path) – Path to NeXus HDF5 file containing event data

• detector_bank (str) – Specific detector bank to load (e.g., ‘bank100’).

• detector_shape (tuple[int, int], optional) – Detector dimensions (x_bins, y_bins) for unrolling event_id to x, y. Default: (512, 512) for SNS VENUS detectors

• event_id_offset (int, optional) – Base offset subtracted from each event_id before unrolling to x, y pixel coordinates (default: 0)

• max_events (int, optional) – Maximum number of events to load (for testing/memory limits) If None, loads all events

Returns:

Pydantic model containing event arrays (tof in ns, x, y) and metadata

Return type:

EventData

Raises:

• FileNotFoundError – If file doesn’t exist

• KeyError – If the NeXus structure, required fields, or the requested detector_bank are not found

Examples

>>> # Use the default detector bank ('bank1')
>>> events = load_event_nexus('VENUS_15159.nxs.h5')

>>> # Specify detector bank
>>> events = load_event_nexus('VENUS_15159.nxs.h5', detector_bank='bank100')

>>> # Custom detector shape
>>> events = load_event_nexus('file.nxs.h5', detector_shape=(256, 256))

Metadata loaders for NeuNorm.

Reads acquisition metadata (proton charge, duration, detector, TOF binning) from VENUS NeXus files, plus optional shutter-count and spectra-TOF sidecar text files.

neunorm.loaders.metadata_loader.load_metadata(file_path, read_shutter_counts=False, read_spectra_tof=False)

Load metadata from NeXus file.

Parameters:

• file_path (str or Path) – Path to NeXus HDF5 file containing metadata

• read_shutter_counts (bool) – Whether to read shutter counts from the image directory specified in the metadata (default: False)

• read_spectra_tof (bool) – Whether to read spectra TOF from the image directory specified in the metadata (default: False)

Returns:

Metadata values for proton charge, duration, image file path, and optionally shutter counts. All values are returned as scipp Variables.

Return type:

dict

neunorm.loaders.metadata_loader.load_shutter_counts(image_path)

Load shutter counts from a text file.

Parameters:

image_path (str or Path) – Path to the directory containing the image files, where we expect to find a shutter count file named *_ShutterCount.txt

Returns:

Variable containing shutter counts loaded from the file, up until the first count of 0 is encountered.

Return type:

sc.Variable

neunorm.loaders.metadata_loader.load_spectra_tof(image_path)

Load TOF values from spectra text file.

Parameters:

image_path (str or Path) – Path to the directory containing the image files, where we expect to find a spectra file named *_Spectra.txt

Returns:

Variable containing TOF values loaded from the file, same number as images in the stack.

Return type:

sc.Variable

Processing

Dark current correction.

neunorm.processing.dark_corrector.subtract_dark(data, dark, clip_negative=True)

Subtract dark current with variance propagation.

data_corr = data - dark

Requirements: - Subtract dark current from sample and OB images - Propagate variance correctly through subtraction using scipp - Handle negative values (clip to zero or flag as invalid) - Support both 2D dark (averaged) and 3D dark (per-frame) inputs

Parameters:

• data (sc.DataArray) – Sample or OB histogram with variance

• dark (sc.DataArray) – Dark current histogram with variance

• clip_negative (bool) – If True, clip negative values to zero after subtraction (default: True) If False, value will be masked

Returns:

Dark-corrected data with propagated variance

Return type:

sc.DataArray

Air region correction.

neunorm.processing.air_region_corrector.apply_air_region_correction(transmission, air_roi)

Scale transmission so air region has mean = 1.0.

Requirements

• Calculate mean transmission in user-specified air region

• Scale entire image so air region = 1.0

• Support per-image correction (radiography) and per-TOF-bin correction (hyperspectral)

• Propagate uncertainty from air region mean

Formula

T_final = T / mean(T[air_ROI])

Uncertainty: σ_T_final = T_final × √[(σ_T/T)² + (σ_air/<T_air>)²]

param transmission:

Normalized transmission (after OB correction)

type transmission:

sc.DataArray

param air_roi:

Air region as an ROI or a bare (x0, y0, x1, y1) tuple, where x1 and y1 are exclusive upper bounds.

type air_roi:

ROI or tuple[int, int, int, int]

Parameters:

• transmission (DataArray)

• air_roi (ROI | tuple[int, int, int, int] | list[int])

Return type:

DataArray

Transmission normalization for neutron imaging.

Implements the core neutron imaging equation: T = Sample / OpenBeam with proper uncertainty propagation and beam corrections.

neunorm.processing.normalizer.as_roi_bounds_list(background_roi)

Normalize a background_roi argument to a list of exclusive (x0, y0, x1, y1) bounds.

Accepts a single ROI (an ROI or a bare 4-int (x0,y0,x1,y1) sequence) — backward compatible — or a sequence of those (pooled). A bare 4-int sequence is treated as ONE ROI; a sequence whose elements are ROIs or sequences is a list of ROIs. NumPy integer bounds are coerced to built-in int so provenance JSON-encodes losslessly.

Parameters:

background_roi (ROI | tuple[int, int, int, int] | list[int] | Sequence[ROI | tuple[int, int, int, int] | list[int]])

Return type:

list[tuple[int, int, int, int]]

neunorm.processing.normalizer.normalize_transmission(sample, ob, proton_charge_sample=None, proton_charge_ob=None, pc_uncertainty=0.005, background_roi=None, background_roi_strict=True)

Normalize sample by open beam to compute transmission.

Formula: T = (Sample / pc_sample) / (OB / pc_ob)

Handles: - Variance propagation (automatic via scipp) - Proton charge corrections (for SNS pulsed beam) - Systematic uncertainties - Mask preservation

Parameters:

• sample (sc.DataArray) – Sample histogram with variance

• ob (sc.DataArray) – Open beam histogram with variance

• proton_charge_sample (float or sc.Variable, optional) – Integrated proton charge during sample acquisition (Coulombs) If provided, normalizes by beam intensity

• proton_charge_ob (float or sc.Variable, optional) – Integrated proton charge during OB acquisition (Coulombs)

• pc_uncertainty (float, optional) – Relative proton charge uncertainty (default: 0.005 = 0.5%) From PLEIADES measurements

• background_roi (ROI/tuple or a sequence of them, optional) – Sample-free background region(s) — an ROI (or a bare (x0, y0, x1, y1) tuple, exclusive stops), or a sequence of those which are pooled (sum(counts over all ROIs) / sum(pixels)). When given, each image is normalized by its pooled background mean — a proton-charge proxy for when proton charge is unavailable (e.g. MARS): T = (S/mean(S[B])) / (O/mean(O[B])). For legacy inclusive extents (a width-w ROI spanning w+1 pixels), use ROI(..., inclusive=True); see apply_background_roi for the open-beam-less form. Mutually exclusive with proton_charge_sample / proton_charge_ob. Uncertainty is propagated first-order (the in-ROI sample/ROI-mean correlation is not corrected). Unless background_roi_strict=False, raises ValueError if the pooled mean is not strictly positive and finite in every image. Indices are resolved against the passed arrays; if a pipeline crops with roi first, give background_roi in the post-crop frame.

• background_roi_strict (bool, optional) – With the default True, a non-positive/non-finite pooled background mean raises ValueError. False skips only that guard and lets zeros propagate through the division (inf/nan output) — the legacy 1.x semantics, for downstreams reproducing 1.x outputs bit for bit. Structural errors (bad ROI bounds, missing dims) always raise.

Returns:

Transmission with dimensions matching input Unit: dimensionless Includes propagated variance and systematic uncertainties

Return type:

sc.DataArray

Examples

>>> # Basic normalization
>>> transmission = normalize_transmission(hist_sample, hist_ob)

>>> # With proton charge correction (SNS)
>>> transmission = normalize_transmission(
...     hist_sample, hist_ob,
...     proton_charge_sample=500.0,
...     proton_charge_ob=505.0
... )

Notes

• Scipp automatically propagates variance through division

• Masks are preserved from both sample and OB (OR combination)

• Zero OB counts produce inf/nan (handle with masks before calling)

neunorm.processing.normalizer.normalize_with_dark(sample, ob, dark, proton_charge_sample=None, proton_charge_ob=None, pc_uncertainty=0.005, background_roi=None, background_roi_strict=True)

Dark-correct and normalize in one step, treating the shared dark frame correctly.

Computes T = (sample - dark) / (ob - dark) (with the optional proton-charge correction) where the same averaged dark is subtracted from both sample and open beam. subtract_dark + normalize_transmission would treat the numerator and denominator as statistically independent and propagate Var(dark) twice; this function removes that spurious shared-dark covariance term.

The transmission values are identical to normalize_transmission(subtract_dark(sample, dark), subtract_dark(ob, dark), ...) — only the propagated variance is corrected (reduced) by 2 * k**2 * (sample-dark) * Var(dark) / (ob-dark)**3, with k = pc_ob / pc_sample (1 when no proton charge). The proton-charge systematic and the sample/open-beam Poisson terms are unchanged.

Parameters:

• sample (sc.DataArray) – Sample, open-beam and (averaged) dark-current frames, each carrying Poisson variance. The same dark is subtracted from both sample and ob.

• ob (sc.DataArray) – Sample, open-beam and (averaged) dark-current frames, each carrying Poisson variance. The same dark is subtracted from both sample and ob.

• dark (sc.DataArray) – Sample, open-beam and (averaged) dark-current frames, each carrying Poisson variance. The same dark is subtracted from both sample and ob.

• proton_charge_sample (float or sc.Variable, optional) – Integrated proton charge for the SNS beam correction (see normalize_transmission).

• proton_charge_ob (float or sc.Variable, optional) – Integrated proton charge for the SNS beam correction (see normalize_transmission).

• pc_uncertainty (float, optional) – Relative proton-charge uncertainty (default 0.005).

• background_roi (tuple[int, int, int, int], optional) – Background-ROI flux normalization (see normalize_transmission), used instead of proton charge. The shared-dark correction then uses k = co/cs (ratio of dark-corrected ROI means) in place of the proton-charge ratio, and additionally removes the ROI-mean shared-dark covariance term 2*T^2*Cov(cs,co)/(cs*co) (Cov(cs,co) = Var(mean(D_roi))) — the ROI-mean analog of the pixel-level correction. (The in-ROI pixel/ROI-mean correlation remains uncorrected, as documented on normalize_transmission.)

• background_roi_strict (bool, optional) – See normalize_transmission: False skips the strictly-positive/finite pooled-mean guard and lets zeros propagate (legacy 1.x semantics).

Returns:

Transmission with correctly-propagated variance (no shared-dark double-counting).

Return type:

sc.DataArray

neunorm.processing.normalizer.apply_background_roi(data, background_roi, strict=True)

Flux-flatten a stack by its pooled background-ROI mean (no open beam).

Returns data / pooled_mean(data over background_roi) — the sample-only form of the background-ROI flux proxy, for when there is no open beam to normalize against. background_roi is a single ROI or a sequence of ROIs (pooled as sum(counts) / sum(pixels)); see normalize_transmission(..., background_roi=) for the with-open-beam transmission form.

First-order uncertainty from the pooled ROI mean is propagated (Var += corrected**2 * Var(coeff) / coeff**2); the in-ROI pixel/ROI-mean correlation is not corrected. Reductions are mask-aware. Raises ValueError if the pooled mean is not strictly positive and finite in every image (unless strict=False).

Parameters:

• data (sc.DataArray) – Image stack with x/y dims (e.g. (spectral, x, y)), optionally carrying variance.

• background_roi (ROI/tuple or a sequence of them) – Sample-free background region(s), pooled.

• strict (bool, optional) – With the default True, a non-positive/non-finite pooled mean raises ValueError. False skips only that guard and lets zeros propagate through the division (inf/nan output) — the legacy 1.x semantics, for downstreams reproducing 1.x outputs bit for bit. Structural errors (bad ROI bounds, missing dims) always raise.

Returns:

data scaled so its pooled background-ROI mean is 1 per image.

Return type:

sc.DataArray

Reference image preparation.

neunorm.processing.reference_preparer.median_with_variance(data, dim)

Compute the median and an approximation of the propagation of variance.

Use the approximation of

Var(median) ≈ (π / (2n)) * mean_variance

Parameters:

• data (sc.DataArray) – Input data with associated variances.

• dim (str) – Dimension along which to compute the median.

Returns:

DataArray containing the median values and their estimated variances.

Return type:

sc.DataArray

neunorm.processing.reference_preparer.prepare_reference(stack, method='mean', dim='frame')

Reduce a 3D frame stack to a 2D reference image.

Parameters:

• stack (sc.DataArray) – 3D input with dimensions (frame, y, x).

• method (str) – Reduction method: “mean” or “median”.

• dim (str) – Dimension along which to reduce.

Returns:

2D reference image (y, x) with propagated variances.

Return type:

sc.DataArray

Function for cropping spatial dimensions to a region of interest (ROI).

neunorm.processing.roi_clipper.apply_roi(data, roi)

Crop spatial dimensions to ROI.

Crop to specified ROI: (x0, y0, x1, y1) Work with 2D, 3D, and 4D arrays (preserve other dimensions) Update coordinate arrays if present Validate ROI is within bounds

Parameters:

• data (sc.DataArray) – Input data array to be cropped.

• roi (ROI or tuple[int, int, int, int]) – Region of interest as an ROI (e.g. ROI(x0=10, y0=20, x1=30, y1=40) or ROI(x0=10, y0=20, width=20, height=20)) or a bare (x0, y0, x1, y1) tuple with exclusive stop indices.

Returns:

Cropped data array with updated coordinates.

Return type:

sc.DataArray

Module for combining multiple runs of neutron imaging data into a single dataset with aggregated metadata.

neunorm.processing.run_combiner.combine_runs(runs, metadata_keys_to_sum=('acquisition_time', 'p_charge'), metadata_check_match=(), normalize_by_runs=False)

Combine multiple runs by summing with metadata aggregation.

• Combines a list of sc.DataArrays representing individual runs by summing their data values and variances.

• Masks are combined using logical OR. If a pixel is masked in any run, it will be masked in the combined result. This ensures that all bad pixels are properly accounted for in the final dataset.

• The function also aggregates metadata by summing specified keys across runs, while retaining other metadata from the first run. This allows for accurate representation of total acquisition time, proton charge, and other relevant parameters in the combined dataset.

Parameters:

• runs (list[sc.DataArray]) – List of DataArrays representing individual runs.

• metadata_keys_to_sum (Sequence[str], optional) – Sequence of metadata keys to aggregate across runs, by default (“acquisition_time”, “p_charge”). These coordinates are summed across runs unless normalize_by_runs=True, in which case they are averaged instead.

• metadata_check_match (Sequence[str], optional) – Sequence of metadata keys that must match across all runs for combination, by default ()

• normalize_by_runs (bool, optional) – Whether to normalize the combined data by the number of runs, by default False

Returns:

Combined DataArray with summed data and aggregated metadata.

Return type:

sc.DataArray

Module for rebinning spatial dimensions. Provides functionality to combine adjacent spatial pixels.

neunorm.processing.spatial_rebinner.rebin_spatial(data, factor)

Bin NxN spatial pixels. Trade-off: lose spatial resolution.

Bin NxN spatial pixels by summing counts Support 2D, 3D, and 4D arrays (preserve other dimensions) Propagate variance correctly Handle edge cases (non-divisible dimensions)

Parameters:

• data (sc.DataArray) – Input data with spatial dimensions ‘x’ and ‘y’.

• factor (int or tuple[int, int]) – Number of adjacent pixels to combine in x and y directions. If a single integer is provided, it is used for both dimensions.

Return type:

DataArray

Uncertainty quantification utilities for NeuNorm 2.0.

Provides functions for: - Attaching Poisson variance to count data - Adding systematic uncertainties - Extracting uncertainties from variance

Uses scipp’s automatic variance propagation through arithmetic operations.

neunorm.processing.uncertainty_calculator.attach_poisson_variance(data)

Attach Poisson variance to count data.

For Poisson counting statistics: σ²(N) = N (variance equals counts)

Parameters:

data (sc.DataArray) – Count data with unit=’counts’

Returns:

Copy of data with .variances = values

Return type:

sc.DataArray

Raises:

ValueError – If data unit is not ‘counts’

Notes

Creates a copy - does not modify original data. If data already has variance, it will be overwritten (with warning).

Examples

>>> counts = sc.array(dims=['x'], values=[100, 200, 300], unit='counts')
>>> data = sc.DataArray(data=counts)
>>> data_with_var = attach_poisson_variance(data)
>>> print(data_with_var.variances)  # [100, 200, 300]

neunorm.processing.uncertainty_calculator.add_systematic_variance(data, relative_uncertainty)

Add systematic uncertainty to data variance.

Used for beam monitor corrections (proton charge, shutter counts) and other systematic uncertainties.

Parameters:

• data (sc.DataArray) – Data with or without existing variance

• relative_uncertainty (float) – Relative uncertainty (fractional). Examples: - 0.005 = 0.5% (typical proton charge uncertainty) - 0.01 = 1.0%

Returns:

Copy with systematic variance added to existing variance

Return type:

sc.DataArray

Notes

Systematic variance added in quadrature: var_total = var_existing + (relative_unc * value)²

If no existing variance, only systematic is added.

Examples

>>> data = sc.DataArray(data=sc.array(dims=['x'], values=[100.0, 200.0], unit='counts'))
>>> data.variances = np.array([100.0, 200.0])  # Poisson (float data required for variances)
>>> # Add 0.5% systematic (e.g., proton charge)
>>> data_sys = add_systematic_variance(data, 0.005)
>>> # var_total = 100 + (0.005*100)² = 100.25

neunorm.processing.uncertainty_calculator.get_uncertainty(data)

Get standard deviation (σ) from variance (σ²).

Parameters:

data (sc.DataArray) – Data with .variances attached

Returns:

Standard deviation (sqrt of variance)

Return type:

sc.DataArray

Raises:

ValueError – If data has no variance

Notes

Equivalent to scipp.stddevs() but with better error message.

Examples

>>> data = sc.DataArray(data=sc.array(dims=['x'], values=[100.0], unit='counts'))
>>> data.variances = np.array([100.0])  # float data required for variances
>>> uncertainty = get_uncertainty(data)
>>> print(uncertainty.values)  # [10.0]

neunorm.processing.uncertainty_calculator.get_relative_uncertainty(data)

Get relative uncertainty (σ / value).

Parameters:

data (sc.DataArray) – Data with .variances attached

Returns:

Relative uncertainty (σ / value)

Return type:

sc.DataArray

Raises:

ValueError – If data has no variance

Examples

>>> data = sc.DataArray(data=sc.array(dims=['x'], values=[100.0, 400.0], unit='counts'))
>>> data.variances = np.array([100.0, 400.0])  # Poisson (float data required for variances)
>>> rel_unc = get_relative_uncertainty(data)
>>> # For Poisson: σ/N = √N/N = 1/√N
>>> print(rel_unc.values)  # [0.1, 0.05]

Time-of-flight

Binning utilities for TOF/Energy/Wavelength conversions.

Provides functions for: - Creating TOF bin edges from energy or wavelength specifications - Converting histograms between TOF/energy/wavelength spaces - Physics-based conversions using scipp.constants

All conversions preserve variance (uncertainty) information.

neunorm.tof.binning.tof_to_energy(tof, flight_path)

Convert time-of-flight to neutron energy.

Formula: E = (1/2) * m_n * (L/t)²

Parameters:

• tof (sc.Variable) – Time-of-flight with unit compatible with ‘s’ (seconds)

• flight_path (sc.Variable) – Flight path length with unit compatible with ‘m’ (meters)

Returns:

Neutron energy with unit ‘eV’

Return type:

sc.Variable

Raises:

ValueError – If TOF is zero or negative

Notes

Uses scipp.constants.m_n for neutron mass (no hardcoded values). Higher TOF → lower energy (inverse relationship).

Examples

>>> tof = sc.scalar(1e-3, unit='s')  # 1 ms
>>> L = sc.scalar(25.0, unit='m')
>>> energy = tof_to_energy(tof, L)
>>> print(energy)  # ~3.27 eV

neunorm.tof.binning.tof_to_wavelength(tof, flight_path)

Convert time-of-flight to neutron wavelength.

Formula: λ = h * t / (m_n * L)

Parameters:

• tof (sc.Variable) – Time-of-flight with unit compatible with ‘s’

• flight_path (sc.Variable) – Flight path length with unit compatible with ‘m’

Returns:

Neutron wavelength with unit ‘angstrom’

Return type:

sc.Variable

Notes

Uses scipp.constants.h and scipp.constants.m_n (no hardcoded values). Linear relationship: TOF ∝ wavelength.

Examples

>>> tof = sc.scalar(1e-3, unit='s')  # 1 ms
>>> L = sc.scalar(25.0, unit='m')
>>> wavelength = tof_to_wavelength(tof, L)
>>> print(wavelength)  # ~0.158 Å

neunorm.tof.binning.wavelength_to_energy(wavelength)

Convert neutron wavelength to energy via de Broglie relation.

Formula: E = h² / (2 * m_n * λ²)

Parameters:

wavelength (sc.Variable) – Wavelength with unit compatible with ‘angstrom’

Returns:

Energy with unit ‘eV’

Return type:

sc.Variable

Notes

Uses scipp.constants (no hardcoded values). Inverse square relationship: E ∝ 1/λ².

Examples

>>> wl = sc.scalar(1.8, unit='angstrom')  # Thermal neutron
>>> energy = wavelength_to_energy(wl)
>>> print(energy)  # ~0.025 eV

neunorm.tof.binning.energy_to_wavelength(energy)

Convert neutron energy to wavelength via de Broglie relation.

Formula: λ = h / sqrt(2 * m_n * E)

Parameters:

energy (sc.Variable) – Energy with unit compatible with ‘eV’

Returns:

Wavelength with unit ‘angstrom’

Return type:

sc.Variable

Raises:

ValueError – If energy is zero or negative

Notes

Uses scipp.constants (no hardcoded values).

Examples

>>> energy = sc.scalar(0.025, unit='eV')  # Thermal
>>> wl = energy_to_wavelength(energy)
>>> print(wl)  # ~1.8 Å

neunorm.tof.binning.create_tof_bins(config, flight_path=None, offset=<scipp.Variable> ()      int64            [µs]  0)

Create TOF bin edges from binning configuration.

Supports three binning strategies: 1. bin_space=’energy’: Create energy bins, convert to TOF (reversed) 2. bin_space=’wavelength’: Create wavelength bins, convert to TOF (ascending) 3. bin_space=’tof’: Create TOF bins directly

Parameters:

• config (BinningConfig) – Binning configuration specifying domain and range

• flight_path (sc.Variable, optional) – Flight path in meters. Required for energy/wavelength modes.

• offset (sc.Variable, optional) – Detector time offset (e.g. TIDelay) applied so energy/wavelength bin edges live in raw detector-TOF space — matching the raw event TOF that is histogrammed into them and the later coordinate labeling. Default: 0 us. Ignored for ‘tof’ mode.

Returns:

TOF bin edges with dimension ‘tof’ and unit ‘ns’

Return type:

sc.Variable

Examples

>>> from neunorm.data_models.tof import BinningConfig
>>> config = BinningConfig(bins=1000, bin_space='energy', energy_range=(1, 100))
>>> L = sc.scalar(25.0, unit='m')
>>> tof_bins = create_tof_bins(config, L)

neunorm.tof.binning.get_energy_histogram(hist_tof, flight_path, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert TOF histogram to energy histogram.

Converts TOF bin edges to energy and reverses data order (high TOF → low energy).

Parameters:

• hist_tof (sc.DataArray) – Histogram with ‘tof’ dimension

• flight_path (sc.Variable) – Flight path in meters

• offset (sc.Variable) – Detector time offset (e.g. TIDelay) applied during TOF→energy labeling so it matches offset-aware bin edges. Default: 0 us. Pass the same offset used to build the histogram’s energy bins.

Returns:

Histogram with ‘energy’ dimension (reversed)

Return type:

sc.DataArray

Notes

Preserves variance if present. Both data and variance are reversed.

neunorm.tof.binning.get_wavelength_histogram(hist_tof, flight_path, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert TOF histogram to wavelength histogram.

Converts TOF bin edges to wavelength. NO reversal needed (low TOF → low wavelength, both ascending).

Parameters:

• hist_tof (sc.DataArray) – Histogram with ‘tof’ dimension

• flight_path (sc.Variable) – Flight path in meters

• offset (sc.Variable) – Detector time offset (e.g. TIDelay) applied during TOF→wavelength labeling so it matches offset-aware bin edges. Default: 0 us. Pass the same offset used to build the histogram’s wavelength bins.

Returns:

Histogram with ‘wavelength’ dimension

Return type:

sc.DataArray

Notes

Preserves variance if present. No data reversal (unlike energy conversion).

Convert between TOF and wavelength or energy

neunorm.tof.coordinate_converter.convert_tof_to_wavelength(tof, distance, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert time-of-flight (TOF) data to wavelength.

TOF → Wavelength: λ = (h × TOF) / (m_n × L)

Parameters:

• tof (sc.Variable) – Variable containing TOF values with appropriate units (e.g., microseconds).

• distance (sc.Variable) – Variable representing the distance from the source to the detector, with appropriate units (e.g., meters).

• offset (sc.Variable) – Variable representing the time offset, with appropriate units (e.g., microseconds).

Returns:

Variable containing wavelength values corresponding to the input TOF data in units of Angstroms.

Return type:

sc.Variable

neunorm.tof.coordinate_converter.convert_wavelength_to_tof(wavelength, distance, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert wavelength data to time-of-flight (TOF).

Wavelength → TOF: TOF = (λ × m_n × L) / h

Parameters:

• wavelength (sc.Variable) – Variable containing wavelength values with appropriate units (e.g., Angstroms).

• distance (sc.Variable) – Variable representing the distance from the source to the detector, with appropriate units (e.g., meters).

• offset (sc.Variable) – Variable representing the time offset, with appropriate units (e.g., microseconds).

Returns:

Variable containing TOF values corresponding to the input wavelength data in units matching the offset unit.

Return type:

sc.Variable

neunorm.tof.coordinate_converter.convert_tof_to_energy(tof, distance, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert time-of-flight (TOF) data to energy.

TOF → Energy: E = (1/2) × m_n × (L / TOF)²

Parameters:

• tof (sc.Variable) – Variable containing TOF values with appropriate units (e.g., microseconds).

• distance (sc.Variable) – Variable representing the distance from the source to the detector, with appropriate units (e.g., meters).

• offset (sc.Variable) – Variable representing the time offset, with appropriate units (e.g., microseconds).

Returns:

Variable containing energy values corresponding to the input TOF data in units of meV.

Return type:

sc.Variable

neunorm.tof.coordinate_converter.convert_energy_to_tof(energy, distance, offset=<scipp.Variable> ()      int64            [µs]  0)

Convert energy data to time-of-flight (TOF).

Energy → TOF (the exact inverse of convert_tof_to_energy()): TOF = L / sqrt(2E / m_n) − offset

Parameters:

• energy (sc.Variable) – Variable containing energy values with appropriate units (e.g., meV or eV).

• distance (sc.Variable) – Variable representing the distance from the source to the detector, with appropriate units (e.g., meters).

• offset (sc.Variable) – Variable representing the time offset, with appropriate units (e.g., microseconds).

Returns:

Variable containing TOF values corresponding to the input energy data, in units matching the offset unit.

Return type:

sc.Variable

Event-to-histogram converter for TOF event data.

Converts event-mode data to 3D histograms using chunked processing for memory efficiency. Ported from venus_tof with enhancements.

neunorm.tof.event_converter.convert_events_to_histogram(events, binning, flight_path, x_bins=514, y_bins=514, chunk_size=500000000, compute_variance=True, detector_time_offset=<scipp.Variable> ()      int64            [µs]  0)

Convert event-mode data to 3D TOF histogram.

Uses chunked processing for memory efficiency (can handle billions of events). Optionally attaches Poisson variance for uncertainty quantification.

Parameters:

• events (EventData) – Event data (tof, x, y arrays)

• binning (BinningConfig) – TOF/energy/wavelength binning configuration

• flight_path (sc.Variable) – Flight path in meters (required for energy/wavelength binning)

• x_bins (int, optional) – Number of spatial bins in x (default: 514, native TPX3 resolution)

• y_bins (int, optional) – Number of spatial bins in y (default: 514)

• chunk_size (int, optional) – Events per chunk for processing (default: 500M) Larger = faster but more memory

• compute_variance (bool, optional) – Attach Poisson variance (var = counts). Default: True

• detector_time_offset (sc.Variable, optional) – Detector time offset (e.g. TIDelay) applied when building energy/wavelength bin edges so they live in raw detector-TOF space, matching the raw event TOF histogrammed into them. Default: 0 us. Has no effect for bin_space='tof'.

Returns:

3D histogram (tof, x, y) with optional variance

Return type:

sc.DataArray

Notes

Chunked processing allows handling datasets larger than RAM. Based on venus_tof implementation with performance optimizations.

Examples

>>> events = load_event_data('data.h5')
>>> binning = BinningConfig(bins=5000, bin_space='energy', energy_range=(1, 100))
>>> hist = convert_events_to_histogram(events, binning, flight_path=sc.scalar(25.0, unit='m'))
>>> print(hist.shape)  # (5000, 514, 514)

neunorm.tof.event_converter.convert_events_to_2d_histogram(events, detector_shape, chunk_size=500000000)

Convert events to 2D spatial histogram (no TOF).

Parameters:

• events (EventData) – Event data (tof, x, y arrays)

• detector_shape (tuple[int, int]) – (x_bins, y_bins) defining the spatial dimensions of the histogram

• chunk_size (int, optional) – Events per chunk for processing (default: 500M) Larger = faster but more memory

Returns:

2D histogram (x, y) with counts

Return type:

sc.DataArray

Module for rebinning TOF histograms. Provides functionality to combine adjacent TOF bins and update edges accordingly.

neunorm.tof.histogram_rebinner.rebin_with_snapped_boundaries(old_edges, requested_tof_edges)

For requested TOF edges that don’t align with original TOF edges, snap to the nearest original edge on the left. This ensures that we only combine adjacent bins and don’t create arbitrary bin schemes, which is a requirement for histogram-mode data.

Parameters:

• old_edges (sc.Variable) – Original TOF bin edges.

• requested_tof_edges (sc.Variable) – Desired TOF bin edges that may not align with original edges.

Returns:

New TOF edges snapped to the nearest original edges on the left.

Return type:

sc.Variable

neunorm.tof.histogram_rebinner.rebin_tof(data, width, unit='bins', logarithmic=False, tof_dim='tof', l_source_to_detector=25.0, detector_time_offset=5000.0)

Combine N adjacent TOF bins. Returns rebinned data.

Requirements - Combine N adjacent TOF bins by summing counts - Update TOF bin edges accordingly - Propagate variance correctly through summation

Constraints

For histogram-mode data (TPX1, TPX3 histogram mode): - Can ONLY combine adjacent bins - Cannot create arbitrary bin schemes (would require raw events) - Cannot split bins

Parameters:

• data (sc.DataArray) – Input data with TOF dimension.

• width (Union[int, float, sc.Variable]) – Width of the new TOF bins in terms of the specified unit. Must be positive. If a sc.Variable is provided, it is interpreted as the desired edges of the new TOF bins, can be in units of time, wavelength or dimensionless (interpreted as bin indices), and will be convertible to the unit of the TOF coordinates.

• unit (str) – Unit by which the new bin width is specified. Must be one of time, wavelength, bins, or manual. Default is bins. If bins, width is interpreted as the number of adjacent bins to combine. If time, width is interpreted as the desired width of the new TOF bins in the same unit as the coordinates. If wavelength, width is interpreted as the desired width of the new TOF bins in Angstrom units, and converted to time using the provided source-to-detector distance and detector time offset. If manual, width is required to be a 1-D sc.Variable (at least two values) representing the desired edges of the new TOF bins. Its interpretation depends on the variable’s unit: if dimensionless, the values are treated as integer bin indices into the existing TOF coordinate (must be of integer dtype and within bounds) and the corresponding original edges are selected; otherwise the values are interpreted as explicit TOF edges in, or convertible to, the unit of the TOF coordinates (a wavelength unit is also accepted and converted using l_source_to_detector and detector_time_offset), and are then snapped to the nearest original edges on the left.

• logarithmic (bool) – Whether to use logarithmic binning. Default is False.

• tof_dim (str) – Name of the TOF dimension in the DataArray. Default is “tof”.

• l_source_to_detector (float) – Distance from the source to the detector in meters. Required for wavelength binning. Default is 25.0.

• detector_time_offset (float) – Time offset of the detector in same unit as TOF. Required for wavelength binning. Default is 5000.0.

Returns:

Rebinned DataArray with updated TOF bins and propagated variance.

Return type:

sc.DataArray

Bad pixel detection for TOF neutron imaging.

Provides tools for detecting dead and hot pixels in event-mode and histogram neutron imaging data. Uses MAD (Median Absolute Deviation) for robust outlier detection.

Ported from venus_tof.masking with generalizations for tof/energy/wavelength dimensions.

neunorm.tof.pixel_detector.detect_dead_pixels(hist)

Detect dead pixels (zero counts across all spectral bins).

Dead pixels can be: - Permanently damaged (physical failure) - Temporarily disabled (radiation damage, recoverable via power cycle)

Parameters:

hist (sc.DataArray) – 3D histogram with dimensions (tof/energy/wavelength, x, y) Spectral dimension can be any of: ‘tof’, ‘energy’, ‘wavelength’

Returns:

Boolean mask with dimensions (x, y) where True = dead pixel

Return type:

sc.Variable

Examples

>>> dead_mask = detect_dead_pixels(hist_ob)
>>> print(f"Found {dead_mask.values.sum()} dead pixels")

neunorm.tof.pixel_detector.detect_hot_pixels(hist, sigma=5.0)

Detect hot pixels using MAD (Median Absolute Deviation) threshold.

Hot pixels are caused by radiation damage and generate fake events uniformly across all spectral bins, resulting in abnormally high spatial sum values.

MAD is more robust than standard deviation for outlier detection because it is not affected by the outliers themselves.

Parameters:

• hist (sc.DataArray) – 3D histogram with dimensions (tof/energy/wavelength, x, y)

• sigma (float, optional) – Threshold in units of MAD (default: 5.0) Common values: - 3.0: Aggressive (catches more pixels, may have false positives) - 5.0: Balanced (recommended for most cases) - 10.0: Conservative (only catches extreme outliers)

Returns:

Boolean mask with dimensions (x, y) where True = hot pixel

Return type:

sc.Variable

Notes

The MAD threshold is converted to approximate standard deviations using the scale factor 1.4826, which makes MAD equivalent to sigma for normally distributed data.

Formula: threshold = median + sigma × MAD × 1.4826

Examples

>>> hot_mask = detect_hot_pixels(hist_ta, sigma=5.0)
>>> print(f"Found {hot_mask.values.sum()} hot pixels")

>>> # Try different thresholds
>>> hot_conservative = detect_hot_pixels(hist_ta, sigma=10.0)
>>> hot_aggressive = detect_hot_pixels(hist_ta, sigma=3.0)

neunorm.tof.pixel_detector.detect_bad_pixels_for_transmission(sample, ob, sigma=5.0)

Detect bad pixels from both sample and open beam for transmission imaging.

For transmission imaging (T = Sample/OB), a pixel is invalid if it’s problematic in EITHER dataset: - Hot pixel in OB → denominator wrong → T wrong - Hot pixel in sample → numerator wrong → T wrong - Dead pixel in OB → division by zero → T undefined - Dead pixel in sample → zero numerator → T=0 (looks like perfect attenuation)

This function applies 4 separate masks to each histogram: - dead_pixels_sample - hot_pixels_sample - dead_pixels_ob - hot_pixels_ob

Scipp automatically combines all masks with OR during operations, so any pixel flagged by any mask will be excluded from calculations.

Parameters:

• sample (sc.DataArray) – Sample histogram (tof/energy/wavelength, x, y)

• ob (sc.DataArray) – Open beam reference histogram (tof/energy/wavelength, x, y)

• sigma (float, optional) – MAD threshold for hot pixel detection (default: 5.0)

Returns:

Dictionary containing individual masks for diagnostics: - ‘dead_sample’: Dead pixels in sample - ‘hot_sample’: Hot pixels in sample - ‘dead_ob’: Dead pixels in open beam - ‘hot_ob’: Hot pixels in open beam

Return type:

dict

Examples

>>> masks = detect_bad_pixels_for_transmission(hist_ta, hist_ob, sigma=5.0)
>>>
>>> # Both histograms now have all 4 masks applied
>>> print(list(hist_ta.masks.keys()))
>>>
>>> # Calculate transmission (masks automatically applied by scipp)
>>> transmission = hist_ta / hist_ob

Notes

The function modifies the input histograms in-place by adding masks to their .masks dictionaries. The original data values are not changed.

Pulse ID reconstruction from TOF data with rollover correction.

Implements a three-pass algorithm to reconstruct pulse assignments from time-of-flight data that exhibits rollover behavior and short-range temporal disorder from TPX3 readout FIFOs.

The algorithm handles: - TOF rollovers when neutron pulse period resets - Clustered false positive detections from temporal disorder - Fine-grained boundary refinement near rollover regions - Multi-chip processing (4 independent TPX3 chips) - Parallel processing for multi-chip detectors (P4 optimization)

See also

examples, tests

neunorm.tof.pulse_reconstruction.assign_chip_ids(x, y, detector_shape=(514, 514))

Assign a chip id (0-3) to each event from its pixel quadrant, for a 2x2 quad detector.

The loaders do not record which physical chip an event came from, but multi-chip pulse reconstruction (reconstruct_pulse_ids() with chip_id) needs the events partitioned by chip. For a standard 2x2 quad Timepix3 detector each chip tiles one spatial quadrant, so the chip can be recovered from the pixel (x, y) position.

Note

This assumes the standard 2x2 quad layout — four equal chips splitting the detector at x_bins // 2 and y_bins // 2 — with chip numbering chip = (x >= W/2) + 2*(y >= H/2) (row-major: 0=lower-left, 1=lower-right, 2=upper-left, 3=upper-right). The numbering itself is arbitrary for reconstruction (which only needs the four chips separated), but if the VENUS detector uses a different physical tiling this mapping must be adjusted. Single-chip detectors do not need this helper.

Parameters:

• x (np.ndarray) – 1D integer pixel-coordinate arrays (same length), e.g. events.x / events.y.

• y (np.ndarray) – 1D integer pixel-coordinate arrays (same length), e.g. events.x / events.y.

• detector_shape (tuple[int, int], optional) – (x_bins, y_bins) of the full detector; the chip boundary is at the midpoint of each axis. Default (514, 514) to match the VENUS TPX3 pipeline / event_converter.

Returns:

1D int array of chip ids in {0, 1, 2, 3}, same length as x/y.

Return type:

np.ndarray

neunorm.tof.pulse_reconstruction.reconstruct_pulse_ids(tof, chip_id=None, threshold=-10.0, window=20, late_margin=14.0, n_jobs=None)

Reconstruct pulse IDs from TOF data using three-pass algorithm.

Handles TOF rollovers when neutron pulse period resets and corrects for short-range temporal disorder from TPX3 readout FIFOs. For multi-chip detectors, processes each chip independently (pulse IDs naturally sync because all chips measure the same physical pulses).

Parameters:

• tof (np.ndarray) – Time-of-flight values (1D array, milliseconds)

• chip_id (np.ndarray, optional) – Chip ID for each event (0-3 for quad detector). If None, assumes single chip. If provided, processes each chip independently.

• threshold (float, optional) – Negative threshold for rollover detection (default: -10.0 ms) TOF drop below this value indicates pulse boundary

• window (int, optional) – Number of events to examine on each side of rollover (default: 20) for boundary refinement

• late_margin (float, optional) – TOF value above which events are considered late hits from previous pulse (default: 14.0 ms, appropriate for 16.67ms pulse period)

• n_jobs (int, optional) – Number of parallel workers for multi-chip processing. - None or 1: Sequential processing (default, safe) - -1: Use all available CPU cores - N > 1: Use N parallel workers Only affects multi-chip processing; single chip always runs sequentially.

Returns:

Pulse ID array (int32) with same length as tof Values: 0, 1, 2, … for sequential pulses

Return type:

np.ndarray

Raises:

ValueError – If n_jobs is 0 or < -1 If chip_id length doesn’t match tof length

Notes

The three-pass algorithm: 1. Pass 1: Vectorized rollover detection using np.diff 2. Pass 2: Clean clustered rollovers (false positives from disorder) 3. Pass 2b: Coarse pulse assignment based on cleaned rollover positions 4. Pass 3: Refine boundaries by optimizing within local window

For multi-chip detectors (VENUS quad TPX3), each chip is processed independently. Pulse IDs automatically synchronize because: - All chips measure the same physical neutron pulses - TDC triggers are synchronized to pulse generation - Rollover detection finds same pulse boundaries per chip

Parallel processing (n_jobs > 1 or n_jobs=-1) uses ProcessPoolExecutor to process each chip in a separate process. This provides ~3-4x speedup for 4-chip detectors on multi-core systems.

Test accuracy: 99.67% on synthetic data with extreme temporal disorder (window size 8-11 events). Real TPX3 data likely has better performance.

Examples

Single chip detector:

>>> from neunorm.loaders.event_loader import load_event_data
>>> from neunorm.tof.pulse_reconstruction import reconstruct_pulse_ids
>>> events = load_event_data('run_12557.h5')
>>> # load_event_data stores TOF in nanoseconds; this API expects milliseconds
>>> tof_ms = events.tof / 1e6
>>> pulse_ids = reconstruct_pulse_ids(tof_ms)
>>> print(f"Found {pulse_ids.max() + 1} pulses")

Multi-chip detector (VENUS quad) with parallel processing:

>>> import numpy as np
>>> from neunorm.tof.pulse_reconstruction import assign_chip_ids
>>> events = load_event_data('run_14749.h5')
>>> tof_ms = events.tof / 1e6  # nanoseconds -> milliseconds
>>> # the loaders don't record the chip; derive it from the pixel quadrant
>>> chip_id = assign_chip_ids(events.x, events.y, detector_shape=(514, 514))
>>> pulse_ids = reconstruct_pulse_ids(
...     tof_ms,
...     chip_id=chip_id,
...     threshold=-10.0,
...     late_margin=14.0,
...     n_jobs=4,  # Process 4 chips in parallel
... )
>>> # Pulse IDs synchronized across all 4 chips
>>> for chip in np.unique(chip_id):  # robust: only chips actually present
...     mask = chip_id == chip
...     print(f"Chip {chip}: {pulse_ids[mask].max() + 1} pulses")

Filter events by pulse:

>>> # Skip first 5 pulses (warmup)
>>> kept = events[pulse_ids >= 5]  # EventData is indexable; filters all per-event arrays

Resonance detection for neutron transmission imaging.

Provides automatic detection of resonance dips in energy-space transmission spectra using tiered filtering (background subtraction, SNR, peak shape).

Ported from venus_tof.resonance with minimal modifications.

class neunorm.tof.resonance.ResonanceDetectionConfig(*, background_sigma_fraction=0.05, initial_prominence=0.01, initial_width=3, min_snr=50.0, snr_window_fraction=0.15, min_peak_width=3, max_peak_width=60, min_prom_width_ratio=0.001)

Bases: BaseModel

Configuration for automatic resonance detection in transmission spectra.

This class encapsulates parameters for tiered filtering: - Background subtraction (Gaussian filter) - Initial peak detection (find_peaks) - SNR filtering (Poisson statistics) - Peak shape filtering (width and prominence/width ratio)

Parameters:

• background_sigma_fraction (float) – Gaussian filter width as fraction of spectrum length (default: 0.05 = 5%)

• initial_prominence (float) – Minimum prominence for initial peak detection (default: 0.01)

• initial_width (int) – Minimum peak width in bins for initial detection (default: 3)

• min_snr (float) – Minimum signal-to-noise ratio using Poisson statistics (default: 50.0)

• snr_window_fraction (float) – Relative energy window for SNR calculation (default: 0.15 = ±15% of E)

• min_peak_width (int) – Minimum allowed peak width in bins (default: 3)

• max_peak_width (int) – Maximum allowed peak width in bins (default: 60)

• min_prom_width_ratio (float) – Minimum prominence/width ratio (default: 0.001)

Examples

>>> config = ResonanceDetectionConfig(min_snr=100.0, max_peak_width=40)
>>> result = detect_resonances(hist_ta, hist_ob, config=config)

classmethod validate_max_greater_than_min(v, info)

Validate that max_peak_width exceeds min_peak_width.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

neunorm.tof.resonance.detect_resonances(hist_ta, hist_ob, config=None, known_resonances=None, validation_tolerance=0.05)

Auto-detect resonance dips in neutron transmission data.

Uses tiered filtering approach: 1. Background subtraction (Gaussian filter) 2. Initial peak detection (scipy.signal.find_peaks) 3. SNR filtering (Poisson statistics with relative energy windows) 4. Peak shape filtering (width and prominence/width ratio)

Parameters:

• hist_ta (sc.DataArray) – Sample histogram with dimensions (energy, x, y)

• hist_ob (sc.DataArray) – Open beam histogram with dimensions (energy, x, y)

• config (ResonanceDetectionConfig, optional) – Detection parameters. If None, uses defaults.

• known_resonances (np.ndarray, optional) – Known resonance energies (eV) for validation

• validation_tolerance (float) – Relative tolerance for matching known resonances (default: 0.05 = ±5%)

Returns:

Detection results containing: - ‘resonance_energies’: np.ndarray of detected energies (eV) - ‘resonance_indices’: np.ndarray of bin indices - ‘snr_values’: np.ndarray of SNR for each resonance - ‘n_initial’: int, peaks after initial detection - ‘n_snr_filtered’: int, peaks after SNR filter - ‘n_shape_filtered’: int, peaks after shape filter - ‘validation’: dict, only when known_resonances is given and at least one peak passes the SNR filter

Return type:

dict

Examples

>>> result = detect_resonances(hist_ta, hist_ob)
>>> print(f"Detected {len(result['resonance_energies'])} resonances")

neunorm.tof.resonance.aggregate_resonance_image(hist_ta, hist_ob, resonance_indices)

Create aggregated 2D transmission image from resonance bins.

Sums raw counts over detected resonance bins, THEN computes transmission. This is mathematically correct: (Σa)/(Σb) ≠ Σ(a/b)

Parameters:

• hist_ta (sc.DataArray) – Sample histogram with dimensions (energy, x, y)

• hist_ob (sc.DataArray) – Open beam histogram with dimensions (energy, x, y)

• resonance_indices (np.ndarray) – Energy bin indices corresponding to detected resonances

Returns:

Aggregated transmission image with dimensions (x, y)

Return type:

sc.DataArray

Examples

>>> result = detect_resonances(hist_ta, hist_ob)
>>> trans_image = aggregate_resonance_image(hist_ta, hist_ob, result['resonance_indices'])

neunorm.tof.resonance.print_detection_summary(result)

Print human-readable summary of detection results.

Parameters:

result (dict) – Output from detect_resonances()

Return type:

None

Per-TOF-bin counting-statistics analysis and rebinning recommendations.

class neunorm.tof.statistics_analyzer.StatisticsReport(counts_per_bin, snr_per_bin, low_statistics_bins, recommended_rebinning, preserve_regions)

Bases: object

Summary of per-TOF-bin counting statistics.

Parameters:

• counts_per_bin (ndarray)

• snr_per_bin (ndarray)

• low_statistics_bins (ndarray)

• recommended_rebinning (int)

• preserve_regions (list[tuple[int, int]])

counts_per_bin

Total counts in each TOF bin.

Type:

np.ndarray

snr_per_bin

Signal-to-noise ratio per bin (sqrt(N) for Poisson statistics).

Type:

np.ndarray

low_statistics_bins

Indices of bins whose SNR is below the requested minimum.

Type:

np.ndarray

recommended_rebinning

Suggested rebinning factor to reach the target SNR.

Type:

int

preserve_regions

(start, end) index ranges flagged to preserve (e.g. Bragg edges).

Type:

list[tuple[int, int]]

neunorm.tof.statistics_analyzer.analyze_statistics(data, min_snr=3.0, tof_dim='tof')

Analyze per-TOF-bin statistics and recommend rebinning.

Requirements - Calculate total counts per TOF bin - Calculate SNR per TOF bin: SNR = √(N) - Identify bins with inadequate statistics (below threshold) - Generate rebinning recommendation - Flag features to preserve (Bragg edges, resonances) # TODO

Parameters:

• data (sc.DataArray) – Input data with TOF dimension and counts as values. Should have Poisson statistics (variance = counts).

• min_snr (float) – Minimum acceptable signal-to-noise ratio (SNR) per bin. Default is 3.0.

• tof_dim (str) – Name of the TOF dimension in the DataArray. Default is “tof”.

Return type:

StatisticsReport

Pipelines

VENUS CCD/CMOS normalization pipeline.

neunorm.pipelines.venus_ccd.run_venus_ccd_pipeline(sample_paths, ob_paths, dark_paths=None, output_path=None, roi=None, gamma_filter=True, air_roi=None, background_roi=None)

Execute VENUS CCD/CMOS normalization pipeline.

Pipeline Steps (12 total) - Load TIFF/FITS (sample, OB, dark [optional]) - Load p_charge metadata - Run combine (critical for VENUS) - ROI clip (optional) - Average dark (optional) / OB - Dead pixel detection - Gamma filtering (optional, less critical than MARS) - Dark correction (optional) - p_charge beam correction - Normalization - Air region correction (optional) - Error propagation (includes p_charge uncertainty) - Output

Parameters:

• sample_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to sample TIFF or FITS files. Each inner list represents a run that should be combined before processing.

• ob_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to open beam TIFF or FITS files. Each inner list represents a run that should be combined before processing.

• dark_paths (Optional[Sequence[Sequence[str | Path]]]) – List of lists of paths to dark current TIFF or FITS files. Each inner list represents a run that should be combined before processing. Optional (default: None). If omitted (None or an empty list), dark correction is skipped and the dark-frame variance does not contribute to the propagated uncertainty.

• output_path (Optional[Path]) – Path to save the output file (HDF5 or TIFF). Required; a value of None raises ValueError (the default exists only so dark_paths can keep its positional slot).

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• gamma_filter (bool) – Whether to apply gamma filtering to the sample data (default: True)

• air_roi (Optional[tuple]) – Region of interest for air correction — an ROI or a bare (x0, y0, x1, y1) tuple. If None, air correction is not applied.

• background_roi (ROI/tuple or a sequence of them) – Sample-free background region(s) — one ROI or a pooled sequence of ROIs (see normalize_transmission) — for flux-proxy normalization when proton charge is unavailable. Mutually exclusive with proton-charge correction. If roi is also given the detector is cropped first, so background_roi indices are resolved in the post-crop frame.

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_paths (Sequence[Sequence[str | Path]])

• ob_paths (Sequence[Sequence[str | Path]])

• dark_paths (Sequence[Sequence[str | Path]] | None)

• output_path (Path | None)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• gamma_filter (bool)

• air_roi (ROI | tuple[int, int, int, int] | list[int] | None)

• background_roi (ROI | tuple[int, int, int, int] | list[int] | Sequence[ROI | tuple[int, int, int, int] | list[int]] | None)

VENUS TPX1 pipeline.

neunorm.pipelines.venus_tpx1.run_venus_tpx1_pipeline(sample_hdf5_paths, ob_hdf5_paths, sample_tiff_paths, ob_tiff_paths, output_path, roi=None, air_roi=None, rebin_by_tof=False, rebin_by_spatial=None, flight_path=<scipp.Variable> ()    float64              [m]  25)

Execute VENUS TPX1 normalization pipeline.

Pipeline Steps (11 total) - Load TIFF stack (pre-binned histograms from auto-reduction) - Load TOF bin edges - Load metadata (including proton charge and detector time offset) - Run combine - ROI clip (optional) - Dead pixel detection - Statistics analysis + rebinning recommendation (only when rebin_by_tof=True) - Rebinning (TOF and/or spatial, optional) - Beam correction (proton charge) - Normalization (TOF-resolved) - Air region correction (optional) - Error propagation - Output

Parameters:

• sample_hdf5_paths (Sequence[str | Path]) – List of paths to sample HDF5 files containing metadata.

• ob_hdf5_paths (Sequence[str | Path]) – List of paths to open beam HDF5 files containing metadata.

• sample_tiff_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to sample TIFF files. Each inner list represents a run that should be combined before processing.

• ob_tiff_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to open beam TIFF files. Each inner list represents a run that should be combined before processing.

• output_path (Path) – Path to save the output file (HDF5 or TIFF)

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• air_roi (Optional[tuple]) – Region of interest for air correction — an ROI or a bare (x0, y0, x1, y1) tuple. If None, air correction is not applied.

• rebin_by_tof (Optional[Union[bool,int]]) – Whether to apply TOF rebinning based on statistics analysis. If an integer is provided, it will be used as the rebinning factor instead of the recommended one.

• rebin_by_spatial (Optional[int | tuple[int, int]]) – Whether to apply spatial rebinning. If an integer is provided, it is used as the rebinning factor for both spatial axes. A (x, y) tuple selects per-axis rebinning factors (x and y). If None, no spatial rebinning is applied.

• flight_path (sc.Variable) – Source-to-detector flight path used for TOF→energy/wavelength coordinate labeling. Defaults to VENUS_FLIGHT_PATH_M (25 m); set it per detector/sample position.

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_hdf5_paths (Sequence[str | Path])

• ob_hdf5_paths (Sequence[str | Path])

• sample_tiff_paths (Sequence[Sequence[str | Path]])

• ob_tiff_paths (Sequence[Sequence[str | Path]])

• output_path (Path)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• air_roi (ROI | tuple[int, int, int, int] | list[int] | None)

• rebin_by_tof (bool | int | None)

• rebin_by_spatial (int | tuple[int, int] | None)

• flight_path (Variable)

VENUS TPX3 histogram pipeline.

neunorm.pipelines.venus_tpx3_histogram.run_venus_tpx3_histogram_pipeline(sample_hdf5_paths, ob_hdf5_paths, sample_tiff_paths, ob_tiff_paths, output_path, roi=None, air_roi=None, rebin_by_tof=False, rebin_by_spatial=None, flight_path=<scipp.Variable> ()    float64              [m]  25)

Execute VENUS TPX3 histogram normalization pipeline.

Pipeline Steps - Load TIFF stack (pre-binned by DAQ pipeline) - Load TOF bin edges - Load metadata (including proton charge and detector time offset) - Run combine - ROI clip (optional) - Dead pixel detection - Hot pixel detection (TPX3-specific, even in histogram mode) - Statistics analysis + rebinning recommendation (only when rebin_by_tof=True) - Rebinning (TOF and/or spatial, optional) - Beam correction (proton charge) - Normalization (TOF-resolved) - Air region correction (optional) - Error propagation - Output

Parameters:

• sample_hdf5_paths (Sequence[str | Path]) – List of paths to sample HDF5 files containing metadata.

• ob_hdf5_paths (Sequence[str | Path]) – List of paths to open beam HDF5 files containing metadata.

• sample_tiff_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to sample TIFF files. Each inner list represents a run that should be combined before processing.

• ob_tiff_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to open beam TIFF files. Each inner list represents a run that should be combined before processing.

• output_path (Path) – Path to save the output file (HDF5 or TIFF)

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• air_roi (Optional[tuple]) – Region of interest for air correction — an ROI or a bare (x0, y0, x1, y1) tuple. If None, air correction is not applied.

• rebin_by_tof (Optional[Union[bool,int]]) – Whether to apply TOF rebinning based on statistics analysis. If an integer is provided, it will be used as the rebinning factor instead of the recommended one.

• rebin_by_spatial (Optional[int | tuple[int, int]]) – Whether to apply spatial rebinning. If a single integer is provided, it is used as the rebinning factor for both spatial axes. A (x, y) tuple selects per-axis rebinning factors (x and y). If None, no spatial rebinning is applied.

• flight_path (sc.Variable) – Source-to-detector flight path used for TOF→energy/wavelength coordinate labeling. Defaults to VENUS_FLIGHT_PATH_M (25 m); set it per detector/sample position.

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_hdf5_paths (Sequence[str | Path])

• ob_hdf5_paths (Sequence[str | Path])

• sample_tiff_paths (Sequence[Sequence[str | Path]])

• ob_tiff_paths (Sequence[Sequence[str | Path]])

• output_path (Path)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• air_roi (ROI | tuple[int, int, int, int] | list[int] | None)

• rebin_by_tof (bool | int | None)

• rebin_by_spatial (int | tuple[int, int] | None)

• flight_path (Variable)

VENUS TPX3 event pipeline.

neunorm.pipelines.venus_tpx3_event.run_venus_tpx3_event_pipeline(sample_paths, ob_paths, binning, output_path, roi=None, air_roi=None, rebin_by_tof=False, rebin_by_spatial=None, detector_shape=(514, 514), event_id_offset=1000000, bank_name='bank100', flight_path=<scipp.Variable> ()    float64              [m]  25)

Execute VENUS TPX3 event normalization pipeline.

Pipeline Steps - Load event data - Run combine (optional) - ROI clip (optional) - Dead pixel detection - Hot pixel detection - Statistics analysis (only when rebin_by_tof=True) - Coarsening strategy (spatial/TOF) - Event → histogram conversion (flexible binning) - Beam correction (p_charge) - Normalization (TOF-resolved) - Air region correction (optional) - Error propagation - Output

Parameters:

• sample_paths (Sequence[str | Path]) – List of paths to sample HDF5 files.

• ob_paths (Sequence[str | Path]) – List of paths to open beam HDF5 files

• binning (BinningConfig) – Configuration for TOF/energy/wavelength binning. Required for event → histogram conversion.

• output_path (Path) – Path to save the output file (HDF5 or TIFF)

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• air_roi (Optional[tuple]) – Region of interest for air correction — an ROI or a bare (x0, y0, x1, y1) tuple.

• rebin_by_tof (Optional[bool,int]) – Whether to apply TOF rebinning based on statistics analysis. If an integer is provided, it will be used as the rebinning factor instead of the recommended one.

• rebin_by_spatial (Optional[int | tuple[int, int]]) – Whether to apply spatial rebinning. If an integer is provided, it is used as the rebinning factor for both spatial axes. A (x, y) tuple selects per-axis rebinning factors. If None, no spatial rebinning is applied.

• detector_shape (tuple[int, int]) – Shape of the TPX3 detector (default: (514, 514))

• event_id_offset (int) – Offset to apply when unrolling event_id to x, y coordinates. This accounts for any non-zero starting point in the event_ids.

• bank_name (str) – Name of the detector bank in the NeXus file to load (default: “bank100”)

• flight_path (sc.Variable) – Source-to-detector flight path used for both energy/wavelength binning and the TOF→energy/wavelength coordinate labeling. Defaults to VENUS_FLIGHT_PATH_M (25 m); set it per detector/sample position (the VENUS L2 varies ~24.5–25.5 m).

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_paths (Sequence[str | Path])

• ob_paths (Sequence[str | Path])

• binning (BinningConfig)

• output_path (Path)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• air_roi (ROI | tuple[int, int, int, int] | list[int] | None)

• rebin_by_tof (bool | int | None)

• rebin_by_spatial (int | tuple[int, int] | None)

• detector_shape (tuple[int, int])

• event_id_offset (int)

• bank_name (str)

• flight_path (Variable)

MARS CCD/CMOS normalization pipeline.

neunorm.pipelines.mars_ccd.run_mars_ccd_pipeline(sample_paths, ob_paths, dark_paths=None, output_path=None, roi=None, gamma_filter=True, background_roi=None)

Execute MARS CCD/CMOS normalization pipeline.

Pipeline Steps (10 total) - Load TIFF/FITS (sample, OB, dark [optional]) - Run combine (optional) - ROI clip (optional) - Average dark (optional) / OB - Dead pixel detection (existing tof/pixel_detector.py) - Gamma filtering (filters/gamma_filter.py) - Dark correction (optional, processing/dark_corrector.py) - Normalization (existing processing/normalizer.py) - Output (exporters/hdf5_writer.py, exporters/tiff_writer.py)

Parameters:

• sample_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to sample TIFF or FITS files. Each inner list represents a run that should be combined before processing.

• ob_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to open beam TIFF or FITS files. Each inner list represents a run that should be combined before processing.

• dark_paths (Optional[Sequence[Sequence[str | Path]]]) – List of lists of paths to dark current TIFF or FITS files. Each inner list represents a run that should be combined before processing. Optional (default: None). If omitted (None or an empty list), dark correction is skipped and the dark-frame variance does not contribute to the propagated uncertainty.

• output_path (Optional[Path]) – Path to save the output file (HDF5 or TIFF). Required; a value of None raises ValueError (the default exists only so dark_paths can keep its positional slot).

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• gamma_filter (bool) – Whether to apply gamma filtering to the sample data (default: True)

• background_roi (ROI/tuple or a sequence of them) – Sample-free background region(s) — one ROI or a pooled sequence of ROIs (see normalize_transmission) — for flux-proxy normalization when proton charge is unavailable. Mutually exclusive with proton-charge correction. If roi is also given the detector is cropped first, so background_roi indices are resolved in the post-crop frame.

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_paths (Sequence[Sequence[str | Path]])

• ob_paths (Sequence[Sequence[str | Path]])

• dark_paths (Sequence[Sequence[str | Path]] | None)

• output_path (Path | None)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• gamma_filter (bool)

• background_roi (ROI | tuple[int, int, int, int] | list[int] | Sequence[ROI | tuple[int, int, int, int] | list[int]] | None)

MARS TPX3 normalization pipeline.

neunorm.pipelines.mars_tpx3.run_mars_tpx3_pipeline(sample_paths, ob_paths, output_path, roi=None, gamma_filter=True, detector_shape=(514, 514), background_roi=None)

Execute MARS TPX3 normalization pipeline.

Pipeline Step 1. Load event data 2. Convert events to 2D histogram 3. Run combine (optional) 4. ROI clip (optional) 5. Dead pixel detection 6. Hot pixel detection 7. Gamma filtering 8. Normalization 9. Output

Parameters:

• sample_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to sample HDF5 files. Each inner list corresponds to one run and will be combined before processing.

• ob_paths (Sequence[Sequence[str | Path]]) – List of lists of paths to open beam HDF5 files Each inner list corresponds to one run and will be combined before processing.

• output_path (Path) – Path to save the output file (HDF5 or TIFF)

• roi (Optional[tuple]) – Region of interest to crop to — an ROI or a bare (x0, y0, x1, y1) tuple.

• gamma_filter (bool) – Whether to apply gamma filtering to the sample data (default: True)

• detector_shape (tuple[int, int]) – Shape of the TPX3 detector (default: (514, 514))

• background_roi (ROI/tuple or a sequence of them) – Sample-free background region(s) — one ROI or a pooled sequence of ROIs (see normalize_transmission) — for flux-proxy normalization when proton charge is unavailable. Mutually exclusive with proton-charge correction. If roi is also given the detector is cropped first, so background_roi indices are resolved in the post-crop frame.

Return type:

DataArray

Notes

This function writes the normalized transmission data to disk in either HDF5 or TIFF format, depending on the file extension of output_path. Metadata and detector masks are included in the output.

Returns:

Final normalized transmission DataArray with metadata and masks

Return type:

sc.DataArray

Parameters:

• sample_paths (Sequence[Sequence[str | Path]])

• ob_paths (Sequence[Sequence[str | Path]])

• output_path (Path)

• roi (ROI | tuple[int, int, int, int] | list[int] | None)

• gamma_filter (bool)

• detector_shape (tuple[int, int])

• background_roi (ROI | tuple[int, int, int, int] | list[int] | Sequence[ROI | tuple[int, int, int, int] | list[int]] | None)

Exporters

HDF5 writer for Neunorm outputs, including provenance metadata.

neunorm.exporters.hdf5_writer.write_hdf5(output_path, transmission, dead_pixel_mask='dead', hot_pixel_mask='hot', metadata=None)

Write processed data to HDF5 with full provenance.

Write transmission array (3D or 4D depending on workflow) Write uncertainty array (same shape as transmission) Write pixel masks (dead, hot) Write TOF bin edges (for hyperspectral data) Store full processing metadata/provenance

Output Structure:

/transmission     # (θ, [TOF,] y, x) float32
/uncertainty      # (θ, [TOF,] y, x) float32 (only if variances are present)
/masks/dead       # (y, x) bool (only if the named dead-pixel mask exists)
/masks/hot        # (y, x) bool (only if the named hot-pixel mask exists)
/tof              # (N+1,) float64 (only if the data carries a "tof" coordinate)
/metadata/        # processing provenance (only when metadata is supplied)

Metadata values are stored as native datasets (strings, scalars, flat arrays). Nested values such as per-run file-path lists (sample_paths=[[...], [...]]) are stored as a JSON string with a encoding="json" dataset attribute; read them back with json.loads(dataset.asstr()[()]). The round-trip is lossless for string-valued provenance (what the pipelines emit); only non-JSON-serializable leaves are coerced via str() (json.dumps(..., default=str)), so JSON-native numbers/bools/null keep their types.

Metadata contents:

• Input file paths

• Processing timestamp

• Gamma filter parameters used

• ROI applied (if any)

• Number of runs combined (if any)

• Software version

Parameters:

• output_path (Union[Path, str]) – Path to output HDF5 file

• transmission (sc.DataArray) – Processed transmission data to write. Variances, coordinates, and masks should be included in this DataArray.

• dead_pixel_mask (str) – Name of the dead pixel mask in transmission.masks (default: “dead”)

• hot_pixel_mask (str) – Name of the hot pixel mask in transmission.masks (default: “hot”)

• metadata (Optional[dict]) – Dictionary of metadata to store in the file (default: None)

Return type:

None

TIFF writing utilities using scitiff to preserve scipp metadata and coordinates.

neunorm.exporters.tiff_writer.convert_metadata_to_scitiff_coords(metadata)

Convert metadata dictionary to scitiff-compatible DataGroup.

This function takes a metadata dictionary and converts it into a scitiff DataGroup format, which can be embedded in TIFF tags when saving with scitiff. It handles various data types and ensures that the resulting DataGroup is compatible with scitiff’s requirements.

Sequence values (e.g. lists of source paths, an ROI tuple) are stored as JSON strings rather than object-dtype scipp scalars: scitiff’s metadata schema only accepts scalar and typed 1-D/2-D variables, so an object-dtype (PyObject) scalar fails serialization. JSON encoding mirrors the HDF5 writer’s provenance convention; read a value back with json.loads(extra[key]).

Parameters:

metadata (dict) – Dictionary containing metadata key-value pairs to convert.

Returns:

A DataGroup containing the converted metadata, ready for embedding in scitiff.

Return type:

sc.DataGroup

neunorm.exporters.tiff_writer.write_tiff_stack(output_path, transmission, metadata=None, daqmetadata=None)

Write transmission stack as TIFF files using scitiff.

Uses scitiff to preserve scipp DataArray metadata and coordinates.

Requirements

Write transmission as TIFF stack (one file per image or multi-page TIFF) Write uncertainty (stdevs) and a mask packed into the same stack via the channel dimension (concat_stdevs_and_mask=True), not as a separate file Support 32-bit float output Embed metadata in scitiff format (JSON in TIFF tags) Preserve scipp coordinate information through scitiff

Scitiff Metadata Schema https://scipp.github.io/scitiff/index.html#scitiff-metadata-schema

Parameters:

• output_path (Union[Path, str]) – Path to save TIFF files

• transmission (sc.DataArray) – DataArray containing transmission data, variances, coordinates

• metadata (Optional[dict]) – Additional metadata to embed in the TIFF files (default: None). Can include any key-value pairs.

• daqmetadata (Optional[dict]) – Additional DAQ metadata to embed in the TIFF files (default: None). Created as scitiff.DAQMetadata with keys: ‘facility’, ‘instrument’, ‘detector_type’, ‘source_type’, ‘source’, ‘simulated’

Return type:

None

Filters

Gamma filter to remove outliers from neutron imaging data.

neunorm.filters.gamma_filter.apply_gamma_filter(data, threshold_sigma=5.0, kernel_size=3, preserve_variance=True)

Remove gamma contamination by replacing outliers with local median.

Formula:

FOR each pixel (i, j):
    IF value > median(neighborhood) + k * σ(neighborhood):
        Replace with median(neighborhood)
        Update variance estimate

The filter will:

• Detect gamma spikes (statistical outliers above threshold)

• Replace detected spikes with local median (3x3 or 5x5 neighborhood)

• Support both single images and stacks

• Configurable threshold (default: median + k×σ)

Parameters:

• data (sc.DataArray) – Input neutron imaging data.

• threshold_sigma (float) – Number of standard deviations to use as the threshold for identifying outliers.

• kernel_size (int) – Size of the local neighborhood for computing the median.

• preserve_variance (bool) – If True, keep the original variance for all pixels. If False, update the variance of outliers to the local median variance.

Returns:

Gamma-filtered data with propagated variance if requested.

Return type:

sc.DataArray

Utilities

Physical constants for neutron TOF calculations.

All fundamental physical constants sourced from scipy.constants (CODATA 2018). NO hardcoded values for standard physics quantities - always use scipy.

Detector-specific and facility-specific constants are documented with sources.

neunorm.utils.constants.M_N = 1.67492750056e-27

scipy.constants.m_n (CODATA 2018)

Type:

Neutron mass in kg. Source

neunorm.utils.constants.H = 6.62607015e-34

scipy.constants.h (exact, 2019 SI definition)

Type:

Planck constant in J·s. Source

neunorm.utils.constants.E_CHARGE = 1.602176634e-19

scipy.constants.e (exact, 2019 SI definition)

Type:

Elementary charge in Coulombs. Source

neunorm.utils.constants.H_OVER_MN = 3.956034006119441e-07

h / m_n ratio for wavelength conversion.

Used in: λ = (h/m_n) * (t/L) Value: ~3.956034e-7 m·s Computed from scipy.constants for full precision (no rounding).

neunorm.utils.constants.DE_BROGLIE_EV_ANGSQ = 0.08180421023520228

de Broglie constant for wavelength-energy conversion.

Used in: E(eV) = DE_BROGLIE_EV_ANGSQ / λ²(Å) Value: ~0.0818 eV·Å² Computed from scipy.constants with unit conversion (m² → Ų).

Derivation: E = h² / (2·m_n·λ²) [SI units: J = (J·s)² / (kg·m²)] E(eV) = E(J) / e [Convert J to eV] With λ in Å: multiply by 1e20 (since 1 Å = 1e-10 m, so 1 Ų = 1e-20 m²)

neunorm.utils.constants.TPX3_CLOCK_NS = 25.0

Timepix3 detector time resolution.

Source: - Timepix3 Manual, Section 3.2, Medipix Collaboration (2014) - URL: https://medipix.web.cern.ch/ - Technical specification: 1.5625 ns × 16 (clock divider) = 25 ns

Last verified: 2025-11-17 Precision: Exact (clock specification from manufacturer)

neunorm.utils.constants.TPX4_CLOCK_NS = 0.195

Timepix4 detector time resolution (improved from TPX3).

Source: - Timepix4 Specifications, Medipix Collaboration (2023) - URL: https://medipix.web.cern.ch/

Last verified: 2025-11-17 Note: TPX4 not yet deployed at VENUS as of 2025-11-17

neunorm.utils.constants.VENUS_FLIGHT_PATH_M = 25.0

Default VENUS beamline L2 distance (moderator to detector).

Source: - VENUS beamline design documents, Spallation Neutron Source, ORNL - URL: https://neutrons.ornl.gov/venus - Contact: Instrument scientists for precise measurements

Last verified: 2025-11-17

Important Notes: - Actual flight path varies by detector position and sample position - Always verify with beamline staff for high-precision measurements - Typical range: 24.5 - 25.5 m depending on detector mount

neunorm.utils.constants.MARS_FLIGHT_PATH_M = None

MARS beamline uses continuous neutron beam (HFIR reactor).

No pulsed source → no time-of-flight capability → flight path concept not applicable. For MARS normalization, use traditional flat-field correction only.

neunorm.utils.constants.ev_to_joule(energy_ev)

Convert electron volts to Joules.

Uses scipy.constants.e (elementary charge).

Parameters:

energy_ev (float) – Energy in electron volts

Returns:

Energy in Joules

Return type:

float

neunorm.utils.constants.joule_to_ev(energy_j)

Convert Joules to electron volts.

Uses scipy.constants.e (elementary charge).

Parameters:

energy_j (float) – Energy in Joules

Returns:

Energy in electron volts

Return type:

float

neunorm.utils.constants.angstrom_to_meter(wavelength_a)

Convert Angstrom to meters.

Parameters:

wavelength_a (float) – Wavelength in Angstrom

Returns:

Wavelength in meters

Return type:

float

Notes

Conversion factor: 1 Å = 1e-10 m (exact, SI definition)

neunorm.utils.constants.meter_to_angstrom(wavelength_m)

Convert meters to Angstrom.

Parameters:

wavelength_m (float) – Wavelength in meters

Returns:

Wavelength in Angstrom

Return type:

float

Notes

Conversion factor: 1 m = 1e10 Å (exact, SI definition)