Skip to content

[NO MERGE] Bin event time offset instead of time of flight #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
YooSunYoung wants to merge 6 commits into
base: main
Choose a base branch
from
Open

[NO MERGE] Bin event time offset instead of time of flight #173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4cb7995
Use generic workflow.
YooSunYoung Dec 4, 2025
6bf8310
Update input file pattern retrieval routine.
YooSunYoung Dec 4, 2025
26c8987
Fix origin of detector.
YooSunYoung Dec 5, 2025
488f108
Calculate fast/slow axis.
YooSunYoung Dec 5, 2025
0813b1f
Calculate fast/slow axis.
YooSunYoung Dec 5, 2025
346819a
Temporary fix until we fix the tof simulation. [skip ci]
YooSunYoung Dec 5, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 2 additions & 144 deletions src/ess/nmx/_executable_helper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,11 @@
from types import UnionType
from typing import Literal, TypeGuard, TypeVar, Union, get_args, get_origin

from pydantic import BaseModel, Field
from pydantic import BaseModel
from pydantic.fields import FieldInfo
from pydantic_core import PydanticUndefined

from .types import Compression
from .configurations import InputConfig, OutputConfig, ReductionConfig, WorkflowConfig


def _validate_annotation(annotation) -> TypeGuard[type]:
Expand Down Expand Up @@ -140,148 +140,6 @@ def add_args_from_pydantic_model(
return parser


class InputConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Input Configuration"}
# File IO
input_file: list[str] = Field(
title="Input File",
description="Path to the input file. If multiple file paths are given,"
" the output(histogram) will be merged(summed) "
"and will not save individual outputs per input file. ",
)
swmr: bool = Field(
title="SWMR Mode",
description="Open the input file in SWMR mode",
default=False,
)
# Detector selection
detector_ids: list[int] = Field(
title="Detector IDs",
description="Detector indices to process",
default=[0, 1, 2],
)
# Chunking options
iter_chunk: bool = Field(
title="Iterate in Chunks",
description="Whether to process the input file in chunks "
" based on the hdf5 dataset chunk size. "
"It is ignored if hdf5 dataset is not chunked. "
"If True, it overrides chunk-size-pulse and chunk-size-events options.",
default=False,
)
chunk_size_pulse: int = Field(
title="Chunk Size Pulse",
description="Number of pulses to process in each chunk. "
"If 0 or negative, process all pulses at once.",
default=0,
)
chunk_size_events: int = Field(
title="Chunk Size Events",
description="Number of events to process in each chunk. "
"If 0 or negative, process all events at once."
"If both chunk-size-pulse and chunk-size-events are set, "
"chunk-size-pulse is preferred.",
default=0,
)


class TimeBinUnit(enum.StrEnum):
ms = 'ms'
us = 'us'
ns = 'ns'


class TimeBinCoordinate(enum.StrEnum):
event_time_offset = 'event_time_offset'
time_of_flight = 'time_of_flight'


class WorkflowConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Workflow Configuration"}
time_bin_coordinate: TimeBinCoordinate = Field(
title="Time Bin Coordinate",
description="Coordinate to bin the time data.",
default=TimeBinCoordinate.event_time_offset,
)
nbins: int = Field(
title="Number of Time Bins",
description="Number of Time bins",
default=50,
)
min_time_bin: int | None = Field(
title="Minimum Time Bin",
description="Minimum time edge of [time_bin_coordinate] in [time_bin_unit].",
default=None,
)
max_time_bin: int | None = Field(
title="Maximum Time Bin",
description="Maximum time edge of [time_bin_coordinate] in [time_bin_unit].",
default=None,
)
time_bin_unit: TimeBinUnit = Field(
title="Unit of Time Bins",
description="Unit of time bins.",
default=TimeBinUnit.ms,
)
tof_lookup_table_file_path: str | None = Field(
title="TOF Lookup Table File Path",
description="Path to the TOF lookup table file. "
"If None, the lookup table will be computed on-the-fly.",
default=None,
)
tof_simulation_min_wavelength: float = Field(
title="TOF Simulation Minimum Wavelength",
description="Minimum wavelength for TOF simulation in Angstrom.",
default=1.8,
)
tof_simulation_max_wavelength: float = Field(
title="TOF Simulation Maximum Wavelength",
description="Maximum wavelength for TOF simulation in Angstrom.",
default=3.6,
)
tof_simulation_seed: int = Field(
title="TOF Simulation Seed",
description="Random seed for TOF simulation.",
default=42, # No reason.
)


class OutputConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Output Configuration"}
# Log verbosity
verbose: bool = Field(
title="Verbose Logging",
description="Increase output verbosity.",
default=False,
)
# File output
output_file: str = Field(
title="Output File",
description="Path to the output file.",
default="scipp_output.h5",
)
compression: Compression = Field(
title="Compression",
description="Compress option of reduced output file.",
default=Compression.BITSHUFFLE_LZ4,
)


class ReductionConfig(BaseModel):
"""Container for all reduction configurations."""

inputs: InputConfig
workflow: WorkflowConfig = Field(default_factory=WorkflowConfig)
output: OutputConfig = Field(default_factory=OutputConfig)

@property
def _children(self) -> list[BaseModel]:
return [self.inputs, self.workflow, self.output]


T = TypeVar('T', bound=BaseModel)


Expand Down
149 changes: 149 additions & 0 deletions src/ess/nmx/configurations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
# SPDX-License-Identifier: BSD-3-Clause
# Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
import enum

from pydantic import BaseModel, Field

from .types import Compression


class InputConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Input Configuration"}
# File IO
input_file: list[str] = Field(
title="Input File",
description="Path to the input file. If multiple file paths are given,"
" the output(histogram) will be merged(summed) "
"and will not save individual outputs per input file. ",
)
swmr: bool = Field(
title="SWMR Mode",
description="Open the input file in SWMR mode",
default=False,
)
# Detector selection
detector_ids: list[int] = Field(
title="Detector IDs",
description="Detector indices to process",
default=[0, 1, 2],
)
# Chunking options
iter_chunk: bool = Field(
title="Iterate in Chunks",
description="Whether to process the input file in chunks "
" based on the hdf5 dataset chunk size. "
"It is ignored if hdf5 dataset is not chunked. "
"If True, it overrides chunk-size-pulse and chunk-size-events options.",
default=False,
)
chunk_size_pulse: int = Field(
title="Chunk Size Pulse",
description="Number of pulses to process in each chunk. "
"If 0 or negative, process all pulses at once.",
default=0,
)
chunk_size_events: int = Field(
title="Chunk Size Events",
description="Number of events to process in each chunk. "
"If 0 or negative, process all events at once."
"If both chunk-size-pulse and chunk-size-events are set, "
"chunk-size-pulse is preferred.",
default=0,
)


class TimeBinUnit(enum.StrEnum):
ms = 'ms'
us = 'us'
ns = 'ns'


class TimeBinCoordinate(enum.StrEnum):
event_time_offset = 'event_time_offset'
time_of_flight = 'time_of_flight'


class WorkflowConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Workflow Configuration"}
time_bin_coordinate: TimeBinCoordinate = Field(
title="Time Bin Coordinate",
description="Coordinate to bin the time data.",
default=TimeBinCoordinate.event_time_offset,
)
nbins: int = Field(
title="Number of Time Bins",
description="Number of Time bins",
default=50,
)
min_time_bin: int | None = Field(
title="Minimum Time Bin",
description="Minimum time edge of [time_bin_coordinate] in [time_bin_unit].",
default=None,
)
max_time_bin: int | None = Field(
title="Maximum Time Bin",
description="Maximum time edge of [time_bin_coordinate] in [time_bin_unit].",
default=None,
)
time_bin_unit: TimeBinUnit = Field(
title="Unit of Time Bins",
description="Unit of time bins.",
default=TimeBinUnit.ms,
)
tof_lookup_table_file_path: str | None = Field(
title="TOF Lookup Table File Path",
description="Path to the TOF lookup table file. "
"If None, the lookup table will be computed on-the-fly.",
default=None,
)
tof_simulation_min_wavelength: float = Field(
title="TOF Simulation Minimum Wavelength",
description="Minimum wavelength for TOF simulation in Angstrom.",
default=1.8,
)
tof_simulation_max_wavelength: float = Field(
title="TOF Simulation Maximum Wavelength",
description="Maximum wavelength for TOF simulation in Angstrom.",
default=3.6,
)
tof_simulation_seed: int = Field(
title="TOF Simulation Seed",
description="Random seed for TOF simulation.",
default=42, # No reason.
)


class OutputConfig(BaseModel):
# Add title of the basemodel
model_config = {"title": "Output Configuration"}
# Log verbosity
verbose: bool = Field(
title="Verbose Logging",
description="Increase output verbosity.",
default=False,
)
# File output
output_file: str = Field(
title="Output File",
description="Path to the output file.",
default="scipp_output.h5",
)
compression: Compression = Field(
title="Compression",
description="Compress option of reduced output file.",
default=Compression.BITSHUFFLE_LZ4,
)


class ReductionConfig(BaseModel):
"""Container for all reduction configurations."""

inputs: InputConfig
workflow: WorkflowConfig = Field(default_factory=WorkflowConfig)
output: OutputConfig = Field(default_factory=OutputConfig)

@property
def _children(self) -> list[BaseModel]:
return [self.inputs, self.workflow, self.output]
Loading