Data reduction wrapper interface

docs/user-guide/workflow_chunk.ipynb

@@ -8,6 +8,45 @@
     "In this example, we will process McStas events chunk by chunk, panel by panel."
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# TL;DR\n",
+    "\n",
+    "There is a reduction script that process the dataset chunk by chunk.\n",
+    "\n",
+    "> It may have some hard-coded parameters due to unfinished work on the reduction workflow."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "nbsphinx": "hidden"
+   },
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "from ess.nmx.data import small_mcstas_3_sample\n",
+    "\n",
+    "os.environ['INTPUT_FILE_PATH'] = small_mcstas_3_sample()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!essnmx_reduce_mcstas \\\n",
+    "    --input_file ${INTPUT_FILE_PATH} \\\n",
+    "    --output_file script-test.h5 \\\n",
+    "    --verbose \\\n",
+    "    --chunk_size 10_000_000 \\\n",
+    "    --detector_ids 0 1 2\n"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
@@ -37,17 +76,17 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## Compute Maximum Probabiliity\n",
+    "## Compute Raw Data Metadata\n",
     "\n",
-    "`McStasWeight2CountScaleFactor` should not be different from chunk to chunk.\n",
+    "`time-of-flight` coordinate and `McStasWeight2CountScaleFactor` should not be different from chunk to chunk.\n",
     "\n",
-    "Therefore we need to compute `McStasWeight2CoutScaleFactor` before we compute `NMXReducedDataGroup`.\n",
+    "Therefore we need to compute `TimeBinStep` and `McStasWeight2CoutScaleFactor` before we compute `NMXReducedData`.\n",
     "\n",
     "It can be done by `ess.reduce.streaming.StreamProcessor`.\n",
     "\n",
-    "In this example, `MaximumProbability` will be renewed every time a chunk is added to the streaming processor.\n",
+    "In this example, `MinimumTimeOfArrival`, `MaximumTimeOfArrival` and `MaximumProbability` will be renewed every time a chunk is added to the streaming processor.\n",
     "\n",
-    "`MaxAccumulator` remembers the previous maximum value and compute new maximum value with the new chunk.\n",
+    "`(Min/Max)Accumulator` remembers the previous minimum/maximum value and compute new minimum/maximum value with the new chunk.\n",
     "\n",
     "``raw_event_data_chunk_generator`` yields a chunk of raw event probability from mcstas h5 file.\n",
     "\n",
@@ -62,16 +101,20 @@
    "source": [
     "from functools import partial\n",
     "from ess.reduce.streaming import StreamProcessor\n",
-    "from ess.nmx.streaming import MaxAccumulator\n",
+    "from ess.nmx.streaming import MaxAccumulator, MinAccumulator\n",
     "\n",
     "# Stream processor building helper\n",
     "scalefactor_stream_processor = partial(\n",
     "    StreamProcessor,\n",
     "    dynamic_keys=(RawEventProbability,),\n",
-    "    target_keys=(McStasWeight2CountScaleFactor,),\n",
-    "    accumulators={MaximumProbability: MaxAccumulator},\n",
+    "    target_keys=(NMXRawDataMetadata,),\n",
+    "    accumulators={\n",
+    "        MaximumProbability: MaxAccumulator,\n",
+    "        MaximumTimeOfArrival: MaxAccumulator,\n",
+    "        MinimumTimeOfArrival: MinAccumulator,\n",
+    "    },\n",
     ")\n",
-    "scalefactor_wf = wf.copy()"
+    "metadata_wf = wf.copy()"
    ]
   },
   {
@@ -80,7 +123,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "scalefactor_wf.visualize(McStasWeight2CountScaleFactor, graph_attr={\"rankdir\": \"TD\"}, compact=True)"
+    "metadata_wf.visualize(NMXRawDataMetadata, graph_attr={\"rankdir\": \"TD\"}, compact=True)"
    ]
   },
   {
@@ -90,7 +133,10 @@
    "outputs": [],
    "source": [
     "from ess.nmx.types import DetectorName\n",
-    "from ess.nmx.mcstas.load import raw_event_data_chunk_generator\n",
+    "from ess.nmx.mcstas.load import (\n",
+    "    raw_event_data_chunk_generator,\n",
+    "    mcstas_weight_to_probability_scalefactor,\n",
+    ")\n",
     "from ess.nmx.streaming import calculate_number_of_chunks\n",
     "from ipywidgets import IntProgress\n",
     "\n",
@@ -99,10 +145,11 @@
     "NUM_DETECTORS = 3\n",
     "\n",
     "# Loop over the detectors\n",
-    "file_path = scalefactor_wf.compute(FilePath)\n",
-    "scale_factors = {}\n",
+    "file_path = metadata_wf.compute(FilePath)\n",
+    "raw_data_metadatas = {}\n",
+    "\n",
     "for detector_i in range(0, NUM_DETECTORS):\n",
-    "    temp_wf = scalefactor_wf.copy()\n",
+    "    temp_wf = metadata_wf.copy()\n",
     "    temp_wf[DetectorIndex] = detector_i\n",
     "    detector_name = temp_wf.compute(DetectorName)\n",
     "    max_chunk_id = calculate_number_of_chunks(\n",
@@ -123,11 +170,54 @@
     "        else:\n",
     "            results = processor.add_chunk({RawEventProbability: da})\n",
     "        cur_detector_progress_bar.value += 1\n",
-    "    scale_factors[detector_i] = results[McStasWeight2CountScaleFactor]\n",
+    "    display(results[NMXRawDataMetadata])\n",
+    "    raw_data_metadatas[detector_i] = results[NMXRawDataMetadata]\n",
+    "\n",
+    "# We take the min/maximum values of the scale factor\n",
+    "min_toa = min(meta.min_toa for meta in raw_data_metadatas.values())\n",
+    "max_toa = max(meta.max_toa for meta in raw_data_metadatas.values())\n",
+    "max_probability = max(meta.max_probability for meta in raw_data_metadatas.values())\n",
+    "\n",
+    "toa_bin_edges = sc.linspace(dim='t', start=min_toa, stop=max_toa, num=51)\n",
+    "scale_factor = mcstas_weight_to_probability_scalefactor(\n",
+    "    max_counts=wf.compute(MaximumCounts), max_probability=max_probability\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Compute Metadata\n",
+    "\n",
+    "Other metadata does not require any chunk-based computation.\n",
     "\n",
-    "# We take the minimum scale factor for the entire dataset\n",
-    "scale_factor = min(scale_factors.values())\n",
-    "scale_factor\n"
+    "Therefore we export the metadata first and append detector data later."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from ess.nmx.nexus import export_metadata_as_nxlauetof\n",
+    "\n",
+    "output_file = \"test.h5\"\n",
+    "experiment_metadata = wf.compute(NMXExperimentMetadata)\n",
+    "detector_metas = []\n",
+    "for detector_i in range(3):\n",
+    "    temp_wf = wf.copy()\n",
+    "    temp_wf[DetectorIndex] = detector_i\n",
+    "    detector_metas.append(temp_wf.compute(NMXDetectorMetadata))\n",
+    "\n",
+    "export_metadata_as_nxlauetof(\n",
+    "    *detector_metas,\n",
+    "    experiment_metadata=experiment_metadata,\n",
+    "    output_file=output_file,\n",
+    "    # Arbitrary metadata falls into ``entry`` group as a variable.\n",
+    "    mcstas_weight2count_scale_factor=scale_factor,\n",
+    ")"
    ]
   },
   {
@@ -136,9 +226,11 @@
    "source": [
     "## Compute Final Output\n",
     "\n",
-    "Now with the `scale_factor: McStasWeight2CountScaleFactor`, we can compute the final output chunk by chunk.\n",
+    "Now with all the metadata, we can compute the final output chunk by chunk.\n",
+    "\n",
+    "We will also compute static parameters in advance so that stream processor does not compute them every time another chunk is added.\n",
     "\n",
-    "We will also compute static parameters in advance so that stream processor does not compute them every time another chunk is added."
+    "We will as well export the reduced data detector by detector."
    ]
   },
   {
@@ -150,15 +242,14 @@
     "from ess.nmx.mcstas.xml import McStasInstrument\n",
     "\n",
     "final_wf = wf.copy()\n",
-    "# Add the scale factor to the workflow\n",
+    "# Set the scale factor and time bin edges\n",
     "final_wf[McStasWeight2CountScaleFactor] = scale_factor\n",
+    "final_wf[TimeBinSteps] = toa_bin_edges\n",
     "\n",
-    "# Compute the static information in advance\n",
-    "# static_info = wf.compute([CrystalRotation, McStasInstrument])\n",
-    "static_info = wf.compute([McStasInstrument])\n",
-    "# final_wf[CrystalRotation] = static_info[CrystalRotation]\n",
-    "final_wf[CrystalRotation] = sc.vector([0, 0, 0.,], unit='deg')\n",
-    "final_wf[McStasInstrument] = static_info[McStasInstrument]\n",
+    "# Set the crystal rotation manually for now ...\n",
+    "final_wf[CrystalRotation] = sc.vector([0, 0, 0.0], unit='deg')\n",
+    "# Set static info\n",
+    "final_wf[McStasInstrument] = wf.compute(McStasInstrument)\n",
     "final_wf.visualize(NMXReducedDataGroup, compact=True)"
    ]
   },
@@ -190,6 +281,7 @@
     "from ess.nmx.mcstas.load import raw_event_data_chunk_generator\n",
     "from ess.nmx.streaming import calculate_number_of_chunks\n",
     "from ipywidgets import IntProgress\n",
+    "from ess.nmx.nexus import export_reduced_data_as_nxlauetof\n",
     "\n",
     "CHUNK_SIZE = 10  # Number of event rows to process at once\n",
     "# Increase this number to speed up the processing\n",
@@ -229,7 +321,8 @@
     "        cur_detector_progress_bar.value += 1\n",
     "\n",
     "    result = results[NMXReducedDataGroup]\n",
-    "    display(result)\n"
+    "    display(result)\n",
+    "    export_reduced_data_as_nxlauetof(result, output_file=output_file)\n"
    ]
   }
  ],

pyproject.toml

@@ -51,6 +51,9 @@ test = [
     "pytest",
 ]
 
+[project.scripts]
+essnmx_reduce_mcstas = "ess.nmx.mcstas.executables:main"
+
 [project.urls]
 "Bug Tracker" = "https://github.com/scipp/essnmx/issues"
 "Documentation" = "https://scipp.github.io/essnmx"

requirements/base.txt

@@ -29,7 +29,7 @@ dnspython==2.7.0
     # via email-validator
 email-validator==2.2.0
     # via scippneutron
-essreduce==25.2.4
+essreduce==25.2.5
     # via -r base.in
 fonttools==4.56.0
     # via matplotlib
@@ -57,7 +57,7 @@ lazy-loader==0.4
     #   scippneutron
 locket==1.0.0
     # via partd
-matplotlib==3.10.0
+matplotlib==3.10.1
     # via
     #   mpltoolbox
     #   plopp

requirements/docs.txt

@@ -54,7 +54,7 @@ ipydatawidgets==4.3.5
     # via pythreejs
 ipykernel==6.29.5
     # via -r docs.in
-ipython==8.32.0
+ipython==8.33.0
     # via
     #   -r docs.in
     #   ipykernel

requirements/nightly.txt

@@ -32,7 +32,7 @@ dnspython==2.7.0
     # via email-validator
 email-validator==2.2.0
     # via scippneutron
-essreduce==25.2.4
+essreduce==25.2.5
     # via -r nightly.in
 exceptiongroup==1.2.2
     # via pytest
@@ -64,7 +64,7 @@ lazy-loader==0.4
     #   scippneutron
 locket==1.0.0
     # via partd
-matplotlib==3.10.0
+matplotlib==3.10.1
     # via
     #   mpltoolbox
     #   plopp

src/ess/nmx/__init__.py

@@ -15,13 +15,10 @@
 from .reduction import NMXReducedDataGroup
 from .types import MaximumCounts, NMXRawEventCountsDataGroup
 
-default_parameters = {MaximumCounts: 10000}
-
 del MaximumCounts
 
 __all__ = [
     "NMXRawEventCountsDataGroup",
     "NMXReducedDataGroup",
-    "default_parameters",
     "small_mcstas_3_sample",
 ]

src/ess/nmx/mcstas/__init__.py

@@ -1,11 +1,16 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
+from ..types import MaximumCounts
+
+default_parameters = {MaximumCounts: 10000}
 
 
 def McStasWorkflow():
     import sciline as sl
 
     from ess.nmx.reduction import (
+        calculate_maximum_toa,
+        calculate_minimum_toa,
         format_nmx_reduced_data,
         proton_charge_from_event_counts,
         raw_event_probability_to_counts,
@@ -18,10 +23,13 @@ def McStasWorkflow():
     return sl.Pipeline(
         (
             *loader_providers,
+            calculate_maximum_toa,
+            calculate_minimum_toa,
             read_mcstas_geometry_xml,
             proton_charge_from_event_counts,
             reduce_raw_event_probability,
             raw_event_probability_to_counts,
             format_nmx_reduced_data,
-        )
+        ),
+        params=default_parameters,
     )