Merge pull request #176 from scipp/restructure-workflow-for-streaming

Restructure workflow for streaming

Author: SimonHeybrock

docs/user-guide/common/vanadium_processing.ipynb

@@ -72,7 +72,9 @@
     "\n",
     "workflow[UncertaintyBroadcastMode] = UncertaintyBroadcastMode.drop\n",
     "\n",
-    "workflow[TofMask] = lambda x: (x < sc.scalar(0.0, unit=\"us\")) | (x > sc.scalar(16666.67, unit=\"us\"))\n",
+    "workflow[TofMask] = lambda x: (x < sc.scalar(0.0, unit=\"us\")) | (\n",
+    "    x > sc.scalar(16666.67, unit=\"us\")\n",
+    ")\n",
     "workflow[TwoThetaMask] = None\n",
     "workflow[WavelengthMask] = None\n",
     "\n",
@@ -114,7 +116,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "peaked_data.hist().plot()"
+    "peaked_data.plot()"
    ]
   },
   {
@@ -150,42 +152,24 @@
    "cell_type": "markdown",
    "id": "10",
    "metadata": {},
-   "source": [
-    "We need to histogram the data to perform fits:"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "11",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "peak_histogram = peaked_data.hist()"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "12",
-   "metadata": {},
    "source": [
     "The fits require a bin-center coordinate, so convert from bin-edges:"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "13",
+   "id": "11",
    "metadata": {},
    "outputs": [],
    "source": [
-    "to_fit = peak_histogram.copy(deep=False)\n",
+    "to_fit = peaked_data.copy(deep=False)\n",
     "to_fit.coords['dspacing'] = sc.midpoints(to_fit.coords['dspacing'])"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "14",
+   "id": "12",
    "metadata": {},
    "source": [
     "Perform the fits:"
@@ -194,7 +178,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "15",
+   "id": "13",
    "metadata": {},
    "outputs": [],
    "source": [
@@ -203,7 +187,7 @@
   },
   {
    "cell_type": "markdown",
-   "id": "16",
+   "id": "14",
    "metadata": {},
    "source": [
     "Remove the fitted peaks to obtain the incoherent scattering.\n",
@@ -218,18 +202,18 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "17",
+   "id": "15",
    "metadata": {},
    "outputs": [],
    "source": [
     "incoherent = scn.peaks.remove_peaks(sc.values(to_fit), fit_results)\n",
-    "incoherent.coords['dspacing'] = peak_histogram.coords['dspacing']\n",
+    "incoherent.coords['dspacing'] = peaked_data.coords['dspacing']\n",
     "incoherent.plot()"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "18",
+   "id": "16",
    "metadata": {},
    "source": [
     "We can further inspect the results.\n",
@@ -262,7 +246,7 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "19",
+   "id": "17",
    "metadata": {},
    "outputs": [],
    "source": [
@@ -272,29 +256,36 @@
     "    fit_results: list[scn.peaks.FitResult],\n",
     "    peak_estimates: sc.Variable,\n",
     "    *,\n",
-    "    xlim: tuple[sc.Variable, sc.Variable] | None=None,\n",
+    "    xlim: tuple[sc.Variable, sc.Variable] | None = None,\n",
     "):\n",
     "    if xlim is not None:\n",
+    "\n",
     "        def in_range(x: sc.Variable) -> bool:\n",
     "            return sc.isfinite(x) and (xlim[0] <= x) and (x < xlim[1])\n",
-    "        data = data[data.dim, xlim[0]:xlim[1]]\n",
-    "        removed = removed[removed.dim, xlim[0]:xlim[1]]\n",
-    "        fit_results, peak_estimates = zip(*(\n",
-    "            (r, e)\n",
-    "            for r, e in zip(fit_results, peak_estimates, strict=True)\n",
-    "            if in_range(r.window[0]) and in_range(r.window[1])\n",
-    "        ), strict=True)\n",
+    "\n",
+    "        data = data[data.dim, xlim[0] : xlim[1]]\n",
+    "        removed = removed[removed.dim, xlim[0] : xlim[1]]\n",
+    "        fit_results, peak_estimates = zip(\n",
+    "            *(\n",
+    "                (r, e)\n",
+    "                for r, e in zip(fit_results, peak_estimates, strict=True)\n",
+    "                if in_range(r.window[0]) and in_range(r.window[1])\n",
+    "            ),\n",
+    "            strict=True,\n",
+    "        )\n",
     "\n",
     "    # The actual data\n",
     "    plot_data = {'data': data, 'removed': removed}\n",
     "    linestyles = {}\n",
     "    markers = {}\n",
-    "    colors = {'data': 'C0','removed': 'C2'}\n",
+    "    colors = {'data': 'C0', 'removed': 'C2'}\n",
     "\n",
     "    # Overlay with fit models evaluated at optimized parameters\n",
     "    for i, result in enumerate(fit_results):\n",
     "        if all(not sc.isnan(param).value for param in result.popt.values()):\n",
-    "            best_fit = data[data.dim, result.window[0] : result.window[1]].copy(deep=False)\n",
+    "            best_fit = data[data.dim, result.window[0] : result.window[1]].copy(\n",
+    "                deep=False\n",
+    "            )\n",
     "            best_fit.coords[best_fit.dim] = sc.midpoints(best_fit.coords[best_fit.dim])\n",
     "            best_fit.data = result.eval_model(best_fit.coords[best_fit.dim])\n",
     "\n",
@@ -340,12 +331,12 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "20",
+   "id": "18",
    "metadata": {},
    "outputs": [],
    "source": [
     "peak_removal_diagnostic(\n",
-    "    peak_histogram,\n",
+    "    peaked_data,\n",
     "    incoherent,\n",
     "    fit_results,\n",
     "    peak_estimates,\n",
@@ -355,12 +346,12 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "21",
+   "id": "19",
    "metadata": {},
    "outputs": [],
    "source": [
     "peak_removal_diagnostic(\n",
-    "    peak_histogram,\n",
+    "    peaked_data,\n",
     "    incoherent,\n",
     "    fit_results,\n",
     "    peak_estimates,\n",
@@ -370,7 +361,7 @@
   },
   {
    "cell_type": "markdown",
-   "id": "22",
+   "id": "20",
    "metadata": {},
    "source": [
     "The resulting data array `incoherent` can be saved and used in the main workflow [POWGEN_data_reduction](../sns-instruments/POWGEN_data_reduction.rst) to replace `FocussedDataDspacing[VanadiumRun]`."

docs/user-guide/dream/dream-advanced-powder-reduction.ipynb

@@ -140,8 +140,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "grouped_dspacing.hist().plot(title=grouped_dspacing.coords['detector'].value.capitalize(),\n",
-    "                             norm=\"log\")"
+    "grouped_dspacing.hist().plot(\n",
+    "    title=grouped_dspacing.coords['detector'].value.capitalize(), norm=\"log\"\n",
+    ")"
    ]
   },
   {
@@ -170,7 +171,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "workflow = dream.DreamGeant4Workflow(run_norm=powder.RunNormalization.monitor_integrated)\n",
+    "workflow = dream.DreamGeant4Workflow(\n",
+    "    run_norm=powder.RunNormalization.monitor_integrated\n",
+    ")\n",
     "\n",
     "workflow[Filename[SampleRun]] = dream.data.simulated_diamond_sample()\n",
     "workflow[Filename[VanadiumRun]] = dream.data.simulated_vanadium_sample()\n",
@@ -326,14 +329,8 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "intermediates = workflow.compute(\n",
-    "    (\n",
-    "        DataWithScatteringCoordinates[SampleRun],\n",
-    "        MaskedData[SampleRun],\n",
-    "    )\n",
-    ")\n",
-    "\n",
-    "intermediates[DataWithScatteringCoordinates[SampleRun]]"
+    "intermediates = workflow.compute((CountsWavelength[SampleRun], MaskedData[SampleRun]))\n",
+    "intermediates[CountsWavelength[SampleRun]]"
    ]
   },
   {
@@ -344,9 +341,9 @@
    "outputs": [],
    "source": [
     "two_theta = sc.linspace(\"two_theta\", 0.8, 2.4, 301, unit=\"rad\")\n",
-    "intermediates[MaskedData[SampleRun]].hist(\n",
-    "    two_theta=two_theta, wavelength=300\n",
-    ").plot(norm=\"log\")"
+    "intermediates[MaskedData[SampleRun]].hist(two_theta=two_theta, wavelength=300).plot(\n",
+    "    norm=\"log\"\n",
+    ")"
    ]
   },
   {
@@ -419,12 +416,13 @@
    "source": [
     "detector_names = [\"mantle\", \"endcap_forward\", \"endcap_backward\", \"high_resolution\"]\n",
     "parameter_table = pd.DataFrame(\n",
-    "    {NeXusDetectorName: detector_names},\n",
-    "    index=detector_names\n",
+    "    {NeXusDetectorName: detector_names}, index=detector_names\n",
     ").rename_axis(index='detector')\n",
     "\n",
     "all_detector_workflow = workflow.copy()\n",
-    "mapped = all_detector_workflow[EmptyCanSubtractedIofDspacing[SampleRun]].map(parameter_table)\n",
+    "mapped = all_detector_workflow[EmptyCanSubtractedIofDspacing[SampleRun]].map(\n",
+    "    parameter_table\n",
+    ")\n",
     "all_detector_workflow[EmptyCanSubtractedIofDspacing[SampleRun]] = mapped.reduce(\n",
     "    func=powder.grouping.collect_detectors\n",
     ")"
@@ -446,9 +444,7 @@
    "outputs": [],
    "source": [
     "all_detector_workflow.visualize(\n",
-    "    EmptyCanSubtractedIofDspacing[SampleRun],\n",
-    "    graph_attr={\"rankdir\": \"LR\"},\n",
-    "    compact=True\n",
+    "    EmptyCanSubtractedIofDspacing[SampleRun], graph_attr={\"rankdir\": \"LR\"}, compact=True\n",
     ")"
    ]
   },
@@ -533,14 +529,17 @@
     "    sc.linspace(dim=\"two_theta\", unit=\"rad\", start=2.91, stop=3.11, num=51),\n",
     "]\n",
     "parameter_table = pd.DataFrame(\n",
-    "    {NeXusDetectorName: detector_names,\n",
-    "     TwoThetaBins: two_theta_bins,\n",
-    "     },\n",
-    "    index=detector_names\n",
+    "    {\n",
+    "        NeXusDetectorName: detector_names,\n",
+    "        TwoThetaBins: two_theta_bins,\n",
+    "    },\n",
+    "    index=detector_names,\n",
     ").rename_axis(index='detector')\n",
     "\n",
     "all_detector_workflow = workflow.copy()\n",
-    "mapped = all_detector_workflow[EmptyCanSubtractedIofDspacingTwoTheta[SampleRun]].map(parameter_table)\n",
+    "mapped = all_detector_workflow[EmptyCanSubtractedIofDspacingTwoTheta[SampleRun]].map(\n",
+    "    parameter_table\n",
+    ")\n",
     "all_detector_workflow[EmptyCanSubtractedIofDspacingTwoTheta[SampleRun]] = mapped.reduce(\n",
     "    func=powder.grouping.collect_detectors\n",
     ")"
@@ -556,7 +555,7 @@
     "all_detector_workflow.visualize(\n",
     "    EmptyCanSubtractedIofDspacingTwoTheta[SampleRun],\n",
     "    graph_attr={\"rankdir\": \"LR\"},\n",
-    "    compact=True\n",
+    "    compact=True,\n",
     ")"
    ]
   },

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
   "plopp>=25.03.0",
   "pythreejs",
   "sciline>=25.04.1",
-  "scipp>=24.09.1",
+  "scipp>=25.05.1",
   "scippneutron>=25.02.0",
   "scippnexus>=23.12.0",
   "tof>=25.01.2",

requirements/base.in

@@ -9,7 +9,7 @@ numpy
 plopp>=25.03.0
 pythreejs
 sciline>=25.04.1
-scipp>=24.09.1
+scipp>=25.05.1
 scippneutron>=25.02.0
 scippnexus>=23.12.0
 tof>=25.01.2

requirements/base.txt

@@ -1,9 +1,9 @@
-# SHA1:b1c1e81e8ad714d967dc65bce97a390ae3880cc6
+# SHA1:81114073dd5e4b765405b54b3976f92d6a96439b
 #
-# This file was generated by pip-compile-multi.
+# This file is autogenerated by pip-compile-multi
 # To update, run:
 #
-#    requirements upgrade
+#    pip-compile-multi
 #
 annotated-types==0.7.0
     # via pydantic
@@ -35,7 +35,7 @@ exceptiongroup==1.3.0
     # via ipython
 executing==2.2.0
     # via stack-data
-fonttools==4.58.0
+fonttools==4.58.1
     # via matplotlib
 fsspec==2025.5.1
     # via dask
@@ -51,7 +51,7 @@ importlib-metadata==8.7.0
     # via dask
 ipydatawidgets==4.3.5
     # via pythreejs
-ipython==8.36.0
+ipython==8.37.0
     # via ipywidgets
 ipywidgets==8.1.7
     # via
@@ -76,7 +76,7 @@ matplotlib==3.10.3
     #   plopp
 matplotlib-inline==0.1.7
     # via ipython
-mpltoolbox==25.4.0
+mpltoolbox==25.5.0
     # via scippneutron
 networkx==3.4.2
     # via cyclebane
@@ -136,7 +136,7 @@ sciline==25.5.2
     # via
     #   -r base.in
     #   essreduce
-scipp==25.5.0
+scipp==25.5.1
     # via
     #   -r base.in
     #   essreduce
@@ -191,5 +191,5 @@ wcwidth==0.2.13
     # via prompt-toolkit
 widgetsnbextension==4.0.14
     # via ipywidgets
-zipp==3.21.0
+zipp==3.22.0
     # via importlib-metadata

requirements/basetest.txt

@@ -1,9 +1,9 @@
 # SHA1:c4f3e9aaa3abd10fcdf497bea14415857f62cc89
 #
-# This file was generated by pip-compile-multi.
+# This file is autogenerated by pip-compile-multi
 # To update, run:
 #
-#    requirements upgrade
+#    pip-compile-multi
 #
 certifi==2025.4.26
     # via requests