scipp
diff --git a/‎docs/user-guide/workflow_chunk.ipynb‎
Lines changed: 45 additions & 53 deletions b/‎docs/user-guide/workflow_chunk.ipynb‎
Lines changed: 45 additions & 53 deletions
diff --git a/‎pyproject.toml‎
Lines changed: 4 additions & 0 deletions b/‎pyproject.toml‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎src/ess/nmx/mcstas/__init__.py‎
Lines changed: 5 additions & 1 deletion b/‎src/ess/nmx/mcstas/__init__.py‎
Lines changed: 5 additions & 1 deletion
@@ -134,25 +134,38 @@
     "    raw_data_metadatas[detector_i] = results[NMXRawDataMetadata]\n",
     "\n",
     "# We take the min/maximum values of the scale factor\n",
-    "min_toa = min(dg['min_toa'] for dg in raw_data_metadatas.values())\n",
-    "max_toa = max(dg['max_toa'] for dg in raw_data_metadatas.values())\n",
-    "max_probability = max(dg['max_probability'] for dg in raw_data_metadatas.values())\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",
+    "Therefore we export the metadata first and append detector data later."
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
    "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."
    ]
   },
   {
@@ -181,68 +194,47 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from functools import partial\n",
-    "from ess.reduce.streaming import StreamProcessor, EternalAccumulator\n",
+    "from ess.nmx.nexus import NXLauetofWriter\n",
     "\n",
-    "# Stream processor building helper\n",
-    "final_stream_processor = partial(\n",
-    "    StreamProcessor,\n",
-    "    dynamic_keys=(RawEventProbability,),\n",
-    "    target_keys=(NMXReducedDataGroup,),\n",
-    "    accumulators={NMXReducedCounts: EternalAccumulator},\n",
-    ")"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "from ess.nmx.types import DetectorName\n",
-    "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",
     "\n",
-    "CHUNK_SIZE = 10  # Number of event rows to process at once\n",
-    "# Increase this number to speed up the processing\n",
-    "NUM_DETECTORS = 3\n",
-    "final_wf[TimeBinSteps] = sc.linspace(\n",
-    "    't', 0.1, 0.15, 51, unit='s'\n",
-    ")  # Time bin edges can be calculated from the data\n",
-    "# But streaming processor only sees the first chunk\n",
-    "# So we need to set it manually before processing the first chunk\n",
-    "# It is a bit cumbersome since we have to know the range of the time bins in advance\n",
-    "\n",
-    "# Loop over the detectors\n",
-    "file_path = final_wf.compute(FilePath)\n",
-    "for detector_i in range(0, NUM_DETECTORS):\n",
-    "    temp_wf = final_wf.copy()\n",
-    "    temp_wf[DetectorIndex] = detector_i\n",
-    "    # First compute static information\n",
-    "    detector_name = temp_wf.compute(DetectorName)\n",
-    "    temp_wf[PixelIds] = temp_wf.compute(PixelIds)\n",
+    "def temp_generator(file_path, detector_name):\n",
     "    max_chunk_id = calculate_number_of_chunks(\n",
     "        file_path, detector_name=detector_name, chunk_size=CHUNK_SIZE\n",
     "    )\n",
     "    cur_detector_progress_bar = IntProgress(\n",
     "        min=0, max=max_chunk_id, description=f\"Detector {detector_i}\"\n",
     "    )\n",
     "    display(cur_detector_progress_bar)\n",
-    "\n",
-    "    # Build the stream processor\n",
-    "    processor = final_stream_processor(temp_wf)\n",
     "    for da in raw_event_data_chunk_generator(\n",
     "        file_path=file_path, detector_name=detector_name, chunk_size=CHUNK_SIZE\n",
     "    ):\n",
-    "        if any(da.sizes.values()) == 0:\n",
-    "            continue\n",
-    "        else:\n",
-    "            results = processor.add_chunk({RawEventProbability: da})\n",
+    "        yield da\n",
     "        cur_detector_progress_bar.value += 1\n",
     "\n",
-    "    result = results[NMXReducedDataGroup]\n",
-    "    display(result)\n"
+    "\n",
+    "# When a panel is added to the writer,\n",
+    "# the writer will start processing the data from the generator\n",
+    "# and store the results in memory\n",
+    "# The writer will then write the data to the file\n",
+    "# when ``save`` is called\n",
+    "writer = NXLauetofWriter(\n",
+    "    chunk_generator=temp_generator,\n",
+    "    chunk_insert_key=RawEventProbability,\n",
+    "    workflow=final_wf,\n",
+    "    output_filename=\"test.h5\",\n",
+    "    overwrite=True,\n",
+    "    extra_meta={\"McStasWeight2CountScaleFactor\": scale_factor},\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "for detector_i in range(3):\n",
+    "    display(writer.add_panel(detector_id=detector_i))"
    ]
   }
  ],
 
@@ -42,10 +42,14 @@ dependencies = [
   "pandas",
   "gemmi",
   "defusedxml",
+  "bitshuffle",
 ]
 
 dynamic = ["version"]
 
+[project.scripts]
+essnmx_reduce_mcstas = "ess.nmx.mcstas.executables:main"
+
 [project.optional-dependencies]
 test = [
     "pytest",
 
@@ -1,5 +1,8 @@
 # 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():
@@ -27,5 +30,6 @@ def McStasWorkflow():
             reduce_raw_event_probability,
             raw_event_probability_to_counts,
             format_nmx_reduced_data,
-        )
+        ),
+        params=default_parameters,
     )