Make timebinwidth and nbins mutually exclusive option.

YooSunYoung · YooSunYoung · commit a74a17f9e3bd · 2026-03-25T15:45:33.000+01:00
diff --git a/packages/essnmx/src/ess/nmx/_executable_helper.py b/packages/essnmx/src/ess/nmx/_executable_helper.py
@@ -68,6 +68,53 @@ def _retrieve_field_value(
     return getattr(args, field_name)
 
 
+_EXCLUSIVE_FIELDS_GROUP = {
+    'time_bin_width': 'Bin Edges Configuration',
+    'nbins': 'Bin Edges Configuration',
+}
+
+
+def _add_argument_from_field(
+    group: argparse._ArgumentGroup, field_name: str, field_info: FieldInfo
+):
+    add_argument = partial(group.add_argument, f"--{field_name.replace('_', '-')}")
+
+    if not _validate_annotation(field_info.annotation):
+        raise TypeError(f"Unsupported annotation type: {field_info.annotation}")
+
+    arg_type = _get_no_nonetype_args(field_info.annotation)
+    if _is_appendable_type(arg_type):
+        nargs = '+'
+        arg_type = get_args(field_info.annotation)[0]
+    else:
+        nargs = None
+        arg_type = arg_type
+
+    required = field_info.default is PydanticUndefined
+    default = ... if required else field_info.default
+
+    if arg_type is bool:
+        add_argument = partial(add_argument, action='store_true')
+    elif isinstance(arg_type, type) and issubclass(arg_type, enum.StrEnum):
+        add_argument = partial(
+            add_argument,
+            type=str,
+            choices=[str(e) for e in arg_type],
+        )
+        default = default.name if isinstance(default, enum.StrEnum) else default
+    elif get_origin(arg_type) is Literal:
+        add_argument = partial(
+            add_argument,
+            type=str,
+            choices=[str(lit) for lit in get_args(arg_type)],
+        )
+    else:
+        add_argument = partial(add_argument, type=arg_type, nargs=nargs)
+
+    help_text = ' '.join([field_info.description or '', f"(default: {default})"])
+    add_argument(default=default, required=required, help=help_text)
+
+
 def add_args_from_pydantic_model(
     *, model_cls: type[BaseModel], parser: argparse.ArgumentParser
 ) -> argparse.ArgumentParser:
@@ -99,43 +146,22 @@ def add_args_from_pydantic_model(
     group = parser.add_argument_group(
         model_cls.model_config.get("title", model_cls.__name__)
     )
+    exclusive_groups: dict[str, list[tuple[str, FieldInfo]]] = {}
     for field_name, field_info in model_cls.model_fields.items():
-        add_argument = partial(group.add_argument, f"--{field_name.replace('_', '-')}")
-
-        if not _validate_annotation(field_info.annotation):
-            raise TypeError(f"Unsupported annotation type: {field_info.annotation}")
-
-        arg_type = _get_no_nonetype_args(field_info.annotation)
-        if _is_appendable_type(arg_type):
-            nargs = '+'
-            arg_type = get_args(field_info.annotation)[0]
-        else:
-            nargs = None
-            arg_type = arg_type
-
-        required = field_info.default is PydanticUndefined
-        default = ... if required else field_info.default
-
-        if arg_type is bool:
-            add_argument = partial(add_argument, action='store_true')
-        elif isinstance(arg_type, type) and issubclass(arg_type, enum.StrEnum):
-            add_argument = partial(
-                add_argument,
-                type=str,
-                choices=[str(e) for e in arg_type],
-            )
-            default = default.name if isinstance(default, enum.StrEnum) else default
-        elif get_origin(arg_type) is Literal:
-            add_argument = partial(
-                add_argument,
-                type=str,
-                choices=[str(lit) for lit in get_args(arg_type)],
+        if field_name in _EXCLUSIVE_FIELDS_GROUP:
+            cur_grp = exclusive_groups.setdefault(
+                _EXCLUSIVE_FIELDS_GROUP[field_name], []
             )
+            cur_grp.append((field_name, field_info))
         else:
-            add_argument = partial(add_argument, type=arg_type, nargs=nargs)
+            _add_argument_from_field(
+                group=group, field_name=field_name, field_info=field_info
+            )
 
-        help_text = ' '.join([field_info.description or '', f"(default: {default})"])
-        add_argument(default=default, required=required, help=help_text)
+    for fields in exclusive_groups.values():
+        exclusive_group = group.add_mutually_exclusive_group(required=False)
+        for field_name, field_info in fields:
+            _add_argument_from_field(exclusive_group, field_name, field_info)
 
     return parser
 
diff --git a/packages/essnmx/src/ess/nmx/configurations.py b/packages/essnmx/src/ess/nmx/configurations.py
@@ -2,7 +2,7 @@
 # Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
 import enum
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 from .types import Compression
 
@@ -64,8 +64,37 @@ class TimeBinCoordinate(enum.StrEnum):
     time_of_flight = 'time_of_flight'
 
 
+class _NotSet: ...
+
+
+_notset = _NotSet()
+
+
 class WorkflowConfig(BaseModel):
     # Add title of the basemodel
+    @model_validator(mode='after')
+    def nbins_or_time_bin_width(self):
+        if self.time_bin_width is not None and self.nbins is not None:
+            raise ValueError(
+                "Either `nbins` or `time_bin_width` should be set. "
+                "They cannot be set at the same time. "
+                "It is allowed not setting any of them. "
+                "Then 3 [ms] of `time_bin_width` will be used."
+            )
+        return self
+
+    @model_validator(mode='after')
+    def positive_time_bin_width(self):
+        if self.time_bin_width is not None and self.time_bin_width <= 0:
+            raise ValueError("`time_bin_width` should be a positive number.")
+        return self
+
+    @model_validator(mode='after')
+    def positive_nbins(self):
+        if self.nbins is not None and self.nbins <= 0:
+            raise ValueError("`nbins` should be a positive integer.")
+        return self
+
     model_config = {"title": "Workflow Configuration"}
     time_bin_coordinate: TimeBinCoordinate = Field(
         title="Time Bin Coordinate",
@@ -78,19 +107,17 @@ class WorkflowConfig(BaseModel):
         # Default is time of flight since
         # DIALS should expect the time of flight.
     )
-    time_bin_width: int = Field(
+    time_bin_width: int | None = Field(
         title="Time Bin Width",
         description="Width(Length) of each Time Bin in [time_bin_unit]. "
-        "If `time_bin_width` and `nbins` are both given, "
-        "`time_bin_width` will be preferred. "
-        "Set it to `0` if you want to use `nbins` instead.",
-        default=3,
+        "If none of `time_bin_width` or `nbins` is given, "
+        "3 [ms] of `time_bin_width` will be used.",
+        default=None,
     )
-    nbins: int = Field(
+    nbins: int | None = Field(
         title="Number of Time Bins",
-        description="Number of Time bins. "
-        "If `bin_width` is given, `nbins` will be ignored.",
-        default=50,
+        description="Number of Time bins. ",
+        default=None,
     )
     min_time_bin: int | None = Field(
         title="Minimum Time",
@@ -230,11 +257,3 @@ def to_command_arguments(
         )
     else:
         return arg_list
-
-
-def validate_time_bin_config(config: ReductionConfig) -> None:
-    wfconfig = config.workflow
-    if not (wfconfig.time_bin_width > 0 or (wfconfig.nbins > 0)):
-        raise ValueError(
-            "Either `time-bin-width` or `nbins` should be a positive number."
-        )
diff --git a/packages/essnmx/src/ess/nmx/executables.py b/packages/essnmx/src/ess/nmx/executables.py
@@ -22,7 +22,6 @@
     ReductionConfig,
     TimeBinCoordinate,
     WorkflowConfig,
-    validate_time_bin_config,
 )
 from .nexus import (
     _check_file,
@@ -169,14 +168,20 @@ def _build_time_bin_edges(
         )
 
     # If either min/max were manually selected and bin width is set.
-    if wf_config.time_bin_width > 0:
+    if wf_config.nbins is None:
+        if wf_config.time_bin_width is None:
+            time_bin_width = sc.scalar(3, unit='ms').to(unit=wf_config.time_bin_unit)
+        elif wf_config.time_bin_width is not None:
+            time_bin_width = sc.scalar(
+                wf_config.time_bin_width, unit=wf_config.time_bin_unit
+            )
         # We do not return a scalar bin width since we histogram
         # detector panel individually.
         return sc.arange(
             dim=t_coord_name,
             start=min_t.to(unit=wf_config.time_bin_unit),
             stop=max_t.to(unit=wf_config.time_bin_unit),
-            step=sc.scalar(wf_config.time_bin_width, unit=wf_config.time_bin_unit),
+            step=time_bin_width,
         )
     else:  # Number of bin edges are given but not the bin width.
         n_edges = wf_config.nbins + 1
@@ -222,8 +227,6 @@ def reduction(
     if not config.output.skip_file_output:
         _check_file(config.output.output_file, config.output.overwrite)
 
-    validate_time_bin_config(config=config)
-
     display = _retrieve_display(logger, display)
     input_file_path = _retrieve_input_file(config.inputs.input_file).resolve()
     display(f"Input file: {input_file_path}")
diff --git a/packages/essnmx/tests/executable_test.py b/packages/essnmx/tests/executable_test.py
@@ -69,7 +69,7 @@ def _check_non_default_config(testing_config: ReductionConfig) -> None:
         testing_model = testing_child.model_dump(mode='python')
         default_model = default_child.model_dump(mode='python')
         for key, testing_value in testing_model.items():
-            if key == 'lookup_table_file_path':
+            if key in ['lookup_table_file_path', 'nbins']:
                 # This value may be None or default, so we skip the check.
                 continue
             default_value = default_model[key]
@@ -91,7 +91,6 @@ def test_reduction_config() -> None:
     )
     workflow_options = WorkflowConfig(
         time_bin_width=5,
-        nbins=100,
         min_time_bin=10,
         max_time_bin=100_000,
         time_bin_coordinate=TimeBinCoordinate.event_time_offset,
@@ -194,7 +193,7 @@ def test_executable_runs(small_nmx_nexus_path, tmp_path: pathlib.Path):
     _check_output_file(output_file, bin_width=bin_width)
 
 
-def test_executable_runs_nbins(small_nmx_nexus_path, tmp_path: pathlib.Path):
+def test_executable_runs_exclusive_mutual(small_nmx_nexus_path, tmp_path: pathlib.Path):
     """Test that the executable runs and returns the expected output."""
     output_file = tmp_path / "output.h5"
     assert not output_file.exists()
@@ -206,7 +205,30 @@ def test_executable_runs_nbins(small_nmx_nexus_path, tmp_path: pathlib.Path):
         '--input-file',
         small_nmx_nexus_path,
         '--time-bin-width',
-        '0',
+        '3',
+        '--nbins',
+        str(nbins),
+        '--output-file',
+        output_file.as_posix(),
+    )
+    # Validate that all commands are strings and contain no unsafe characters
+    result = subprocess.run(  # noqa: S603 - We are not accepting arbitrary input here.
+        commands, text=True, capture_output=True, check=False
+    )
+    assert result.returncode == 2  # Should fail with command syntax error.
+
+
+def test_executable_runs_nbins(small_nmx_nexus_path, tmp_path: pathlib.Path):
+    """Test that the executable runs and returns the expected output."""
+    output_file = tmp_path / "output.h5"
+    assert not output_file.exists()
+
+    nbins = 20  # Small number of bins for testing.
+    # The output has 1280x1280 pixels per detector per time bin.
+    commands = (
+        'essnmx-reduce',
+        '--input-file',
+        small_nmx_nexus_path,
         '--nbins',
         str(nbins),
         '--output-file',
@@ -276,7 +298,7 @@ def test_reduction_only_time_bin_width(reduction_config: ReductionConfig) -> Non
 
 
 def test_reduction_only_number_of_time_bins(reduction_config: ReductionConfig) -> None:
-    reduction_config.workflow.time_bin_width = 0
+    reduction_config.workflow.time_bin_width = None
     reduction_config.workflow.nbins = 20
     with known_warnings():
         hist = _retrieve_one_hist(reduction(config=reduction_config))
@@ -300,7 +322,7 @@ def test_histogram_event_time_offset(reduction_config: ReductionConfig) -> None:
 
 
 def test_histogram_event_time_offset_nbins(reduction_config: ReductionConfig) -> None:
-    reduction_config.workflow.time_bin_width = 0
+    reduction_config.workflow.time_bin_width = None
     reduction_config.workflow.nbins = 20
     reduction_config.workflow.time_bin_coordinate = TimeBinCoordinate.event_time_offset
     with known_warnings():
@@ -417,7 +439,8 @@ def test_reduction_with_lut_file(
     # the number of bins changes and the histogram data sizes changes.
     # This test is only for checking if the look up table is used as expected or not
     # therefore using number of bins should be fine.
-    reduction_config.workflow.time_bin_width = 0
+    reduction_config.workflow.time_bin_width = None
+    reduction_config.workflow.nbins = 20
     # Make sure the config uses no lookup table file initially.
     assert reduction_config.workflow.lookup_table_file_path is None
     with known_warnings():
