Do not depend on json template for mocking streaming.

docs/user-guide/getting-started.ipynb

@@ -13,6 +13,27 @@
     "The current setup plots the results of the workflow and saves them to a file, which can we specify using the `--image-path` flag:"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "nbsphinx": "hidden",
+    "tags": [
+     "hide-cell"
+    ]
+   },
+   "outputs": [],
+   "source": [
+    "import sys\n",
+    "import os\n",
+    "\n",
+    "sys.path.append(\"../../tests/helpers\")\n",
+    "\n",
+    "from tests.applications.data import get_path\n",
+    "\n",
+    "os.environ[\"TEST_NEXUS_FILE_PATH\"] = get_path(\"ymir_detectors.nxs\").as_posix()"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -22,8 +43,9 @@
     "!beamlime \\\n",
     "    --workflow dummy \\\n",
     "    --nexus-template-path ../../tests/applications/ymir_detectors.json \\\n",
-    "    --nexus-file-path 'ymir_detectors.nxs' \\\n",
-    "    --image-path-prefix reduction-result"
+    "    --nexus-file-path $TEST_NEXUS_FILE_PATH \\\n",
+    "    --image-path-prefix reduction-result \\\n",
+    "    --fill-dummy-data"
    ]
   },
   {
@@ -103,7 +125,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.10.13"
+   "version": "3.10.15"
   }
  },
  "nbformat": 4,

docs/user-guide/live_data_reduction.ipynb

@@ -31,6 +31,22 @@
     "**so running this application will create a file called ``beamlime_plot.png`` in the current directory.**"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "nbsphinx": "hidden",
+    "tags": [
+     "hide-cell"
+    ]
+   },
+   "outputs": [],
+   "source": [
+    "import sys\n",
+    "\n",
+    "sys.path.append(\"../../tests/helpers\")"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -44,6 +60,17 @@
     "sc.display_logs()"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from tests.applications.data import get_path\n",
+    "\n",
+    "file_path = get_path('ymir_detectors.nxs').as_posix()  # Path to the example file"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -101,7 +128,7 @@
     "        NexusTemplatePath: NexusTemplatePath(\n",
     "            '../../tests/applications/ymir_detectors.json'\n",
     "        ),\n",
-    "        NexusFilePath: NexusFilePath(\"ymir_detectors.nxs\"),\n",
+    "        NexusFilePath: NexusFilePath(file_path),\n",
     "        # TODO: Make a matching file for the template\n",
     "        ResultRegistry: result_registry,\n",
     "    },\n",
@@ -162,7 +189,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.10.13"
+   "version": "3.10.15"
   }
  },
  "nbformat": 4,

src/beamlime/applications/_nexus_helpers.py

@@ -1,5 +1,6 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
+import pathlib
 from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass
 from types import MappingProxyType
@@ -10,11 +11,13 @@
     TypeAlias,
     TypedDict,
     TypeGuard,
+    TypeVar,
     cast,
     get_args,
 )
 
 import numpy as np
+import scippnexus as snx
 
 
 class NexusDataset(TypedDict):
@@ -206,6 +209,85 @@ def _validate_ev44_module_spec(
         )
 
 
+FallBackNameType = TypeVar("FallBackNameType")
+
+
+def _get_instrument_name(
+    nxs_file: snx.File, fail_fall_back: FallBackNameType
+) -> FallBackNameType:
+    try:
+        return nxs_file["entry/instrument/name"][()]
+    except KeyError:
+        return fail_fall_back
+
+
+def _get_nx_class_attribute(group: snx.Group) -> dict[str, str]:
+    if group.nx_class == snx.NXdetector:
+        class_name = "NXdetector"
+    elif group.nx_class == snx.NXmonitor:
+        class_name = "NXmonitor"
+    elif group.nx_class == snx.NXevent_data:
+        class_name = "NXevent_data"
+    else:
+        raise NotImplementedError(f"NX class {group.nx_class} is not supported.")
+
+    return {
+        'name': 'NX_class',
+        'values': class_name,
+    }
+
+
+def _group_to_dict(group: snx.Group) -> NexusGroup:
+    return {
+        "children": [],
+        "name": group.name,
+        "attributes": [_get_nx_class_attribute(group)],
+    }
+
+
+def _retrieve_groups_by_nx_class(
+    nxs_file: snx.File, nx_class: type
+) -> dict[tuple[str, ...], snx.Group]:
+    instrument_path = "entry/instrument"
+    return {
+        tuple("/".join([instrument_path, grp_path]).split('/')): grp
+        for grp_path, grp in nxs_file[instrument_path][nx_class].items()
+    }
+
+
+def _collect_nx_event(
+    groups: dict[tuple[str, ...], snx.Group],
+) -> dict[tuple[str, ...], snx.Group]:
+    children = {}
+    for path, group in groups.items():
+        nx_events = group[snx.NXevent_data]
+        for name, nx_event in nx_events.items():
+            children[(*path, name)] = nx_event
+    return children
+
+
+def collect_streaming_modules_from_nexus_file(
+    nexus_file: str | pathlib.Path,
+) -> dict[StreamModuleKey, StreamModuleValue]:
+    """Collect all stream modules in a nexus file."""
+
+    with snx.File(nexus_file) as f:
+        detectors = _collect_nx_event(_retrieve_groups_by_nx_class(f, snx.NXdetector))
+        monitors = _collect_nx_event(_retrieve_groups_by_nx_class(f, snx.NXmonitor))
+        instrument_name = _get_instrument_name(f, fail_fall_back="unknown")
+        return {
+            StreamModuleKey(
+                module_type='ev44',
+                topic=f"{instrument_name}_detector",
+                source=det.parent.name,
+            ): StreamModuleValue(
+                path=det_path,
+                parent=_group_to_dict(det),
+            )
+            for det_path, det in {**detectors, **monitors}.items()
+        }
+
+
 def collect_streaming_modules(
     structure: Mapping,
 ) -> dict[StreamModuleKey, StreamModuleValue]:
@@ -308,6 +390,7 @@ def _initialize_ev44(module_spec: StreamModuleValue) -> NexusGroup:
             initial_values=np.asarray([], dtype="int32"),
         ),
     ]
+    # TODO This check does not appear to work as expected.
     if not _is_monitor(group):
         # Monitor doesn't have pixel ids.
         group['children'].append(
@@ -370,6 +453,15 @@ def _merge_ev44(group: NexusGroup, data: DeserializedMessage) -> None:
         event_id_dataset["config"]["values"] = np.concatenate(
             (event_id_dataset["config"]["values"], data["pixel_id"])
         )
+    else:
+        # TODO See above, monitor check is not working, remove now that we know there is
+        # no pixel id.
+        try:
+            find_nexus_structure(group, ("event_id",))
+        except KeyError:
+            pass
+        else:
+            del group['children'][-1]
 
 
 def _initialize_f144(module_spec: StreamModuleValue) -> NexusGroup: