ACQ_ScanPipeJobSettings + fix issues found after slicepackage inference changed

brukerapi/config/properties_rawdata_core.json

@@ -28,7 +28,7 @@
     {
         "cmd": "np.dtype('f8').newbyteorder('<' if #BYTORDA=='little' else '>')",
         "conditions": [
-          "#ACQ_ScanPipeJobSettings.nested[#ACQ_jobs.nested.index(@job_desc)][1]=='STORE_64bit_float'"
+          "@rawdata_job_settings is not None and @rawdata_job_settings[1]=='STORE_64bit_float'"
         ],
         "comment": "spec 13.1: storageDataType is STORE_32bit_signed (default) or STORE_64bit_float"
     },
@@ -109,7 +109,7 @@
   ],
   "channels": [
     {
-      "cmd": "#PVM_EncNReceivers",
+      "cmd": "@rawdata_channels",
       "conditions": [
       ]
     }
@@ -122,9 +122,9 @@
   ],
   "shape_storage": [
     {
-      "cmd": "(@job_desc[0],) + (#PVM_EncNReceivers,) + (@job_desc[6] if len(@job_desc)==9 else @job_desc[-1],)",
+      "cmd": "(@job_desc[0], @rawdata_channels, @rawdata_stored_scans)",
       "conditions": [],
-      "comment": "spec 3.3: do not read the scan count from [3], which is nTotalScans -- the number acquired, NAE times the number written. nStoredScans is the last element of the 8-field form and [6] of the 9-field form"
+      "comment": "spec 3.3/13.1: settings nStoredScans defines the raw file size; ACQ_jobs is cross-checked as a fallback"
     }
   ]
 }

brukerapi/dataset.py

@@ -596,6 +596,81 @@ def _parameter_value(self, key, default=None):
         except KeyError:
             return default
 
+    def _rawdata_job_index(self):
+        """Index of this PV360 rawdata job in ``ACQ_jobs`` (spec 13.1)."""
+        if self.type != "rawdata":
+            return None
+        try:
+            jobs = self["ACQ_jobs"].nested
+        except KeyError:
+            jobs = []
+        if not jobs:
+            return None
+        if len(jobs[0]) == 9:
+            title = f"<{self.subtype}>"
+            return next((index for index, job in enumerate(jobs) if job[-1] == title), None)
+        try:
+            return int(self.subtype.removeprefix("job"))
+        except ValueError:
+            return None
+
+    @property
+    def rawdata_job_settings(self):
+        """The §13.1 settings record paired with this rawdata job, if present."""
+        index = self._rawdata_job_index()
+        try:
+            settings = self["ACQ_ScanPipeJobSettings"].nested
+        except KeyError:
+            settings = []
+        return settings[index] if index is not None and index < len(settings) else None
+
+    @property
+    def rawdata_job_discarded(self):
+        """Whether §13.1 says this job's data is intentionally not written."""
+        settings = self.rawdata_job_settings
+        return bool(settings and str(settings[0]).strip("<>") == "STORE_discard")
+
+    @property
+    def rawdata_stored_scans(self):
+        """Number of scans physically written for this rawdata job (spec 3.3)."""
+        index = self._rawdata_job_index()
+        try:
+            jobs = self["ACQ_jobs"].nested
+        except KeyError:
+            jobs = []
+        if index is None or index >= len(jobs):
+            return None
+        job = jobs[index]
+        job_stored = int(job[6] if len(job) == 9 else job[-1])
+        settings = self.rawdata_job_settings
+        if settings is None or len(settings) <= 9:
+            return job_stored
+        settings_stored = int(settings[9])
+        if settings_stored != job_stored and not getattr(self, "_warned_rawdata_stored_scans", False):
+            warnings.warn(
+                f"ACQ_ScanPipeJobSettings nStoredScans={settings_stored} disagrees with "
+                f"ACQ_jobs nStoredScans={job_stored} for {self.path}; using the settings value",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+            self._warned_rawdata_stored_scans = True
+        return settings_stored
+
+    @property
+    def rawdata_channels(self):
+        """Number of receivers recorded for this job (spec 3.3)."""
+        selected = self._parameter_value("ACQ_ReceiverSelectPerChan")
+        fallback = int(self._parameter_value("PVM_EncNReceivers", 1))
+        if selected is not None:
+            values = np.atleast_1d(selected)
+            channels = sum(str(value).strip("<>").casefold() in {"yes", "on", "1"} for value in values)
+            # Some systems expose more physical receiver paths than the job
+            # writes. Only promote the per-channel declaration when it agrees
+            # with the method's logical receiver count.
+            if channels == fallback:
+                return channels
+        return fallback
+
     def _infer_scheme_id(self):
         # Source of truth for format inference:
         # https://github.com/gdevenyi/brkraw-legacy/blob/main/FILE_FORMAT.md
@@ -1116,6 +1191,12 @@ def slice_packages_index(self):
         if packs is not None:
             return [(int(first), int(count)) for first, count in np.atleast_2d(np.asarray(packs, dtype=int))]
 
+        if self._parameter_value("VisuCorePosition") is None or self._parameter_value("VisuCoreOrientation") is None:
+            # Without geometry there is no evidence for a package boundary.
+            # Keep the dataset loadable (and let affine access report the
+            # missing geometry) rather than treating it as a malformed split.
+            return [(0, int(self._parameter_value("VisuCoreFrameCount", 1)))]
+
         position, orientation = self._frame_geometry()
         count = position.shape[0]
         if orientation.shape[0] < count:

brukerapi/folders.py

@@ -5,7 +5,7 @@
 from copy import deepcopy
 from pathlib import Path
 
-from .dataset import Dataset
+from .dataset import LOAD_STAGES, Dataset
 from .exceptions import (
     FilterEvalFalse,
     IncompleteDataset,
@@ -229,7 +229,16 @@ def make_tree(self, *, recursive: bool = True) -> list:
 
             if Dataset.is_supported_path(path, self._dataset_index):
                 try:
-                    children.append(Dataset(path, **self._dataset_state))
+                    if path.stem == "rawdata" and self._dataset_state["load"] >= LOAD_STAGES["parameters"]:
+                        discovery_state = {**self._dataset_state, "load": LOAD_STAGES["parameters"]}
+                        dataset = Dataset(path, **discovery_state)
+                        if dataset.rawdata_job_discarded:
+                            continue
+                        if self._dataset_state["load"] > LOAD_STAGES["parameters"]:
+                            dataset = Dataset(path, **self._dataset_state)
+                    else:
+                        dataset = Dataset(path, **self._dataset_state)
+                    children.append(dataset)
                 except InvalidDataset as error:
                     warnings.warn(f"Skipping invalid dataset {path}: {error}", RuntimeWarning, stacklevel=2)
                     continue

test/config/properties_0.2H2.json

@@ -1184,7 +1184,7 @@
         "id": "2DSEQ_1_1_LEGO_PHANTOM_2",
         "imaging_frequency": 400.322522538113,
         "is_single_slice": false,
-        "num_slice_packages": 1,
+        "num_slice_packages": 3,
         "numpy_dtype": "int16",
         "offset": [
             0,

test/test_latent_conformance.py

@@ -10,6 +10,7 @@
 import pytest
 
 from brukerapi.dataset import LOAD_STAGES, Dataset
+from brukerapi.folders import Experiment
 from brukerapi.jcampdx import JCAMPDX
 from brukerapi.utils import transposed_size
 from test.synthetic import Verbatim, write_2dseq, write_binary, write_jcampdx
@@ -161,6 +162,59 @@ def test_a_64_bit_float_job_is_read_as_float64(tmp_path):
     assert dataset.numpy_dtype == np.dtype("<f8")
 
 
+def test_rawdata_settings_control_stored_scans_receivers_and_discarded_jobs(tmp_path):
+    """Specs 3.3/13.1: settings describe the physical rawdata file."""
+    experiment = tmp_path / "27"
+    settings = (
+        "STORE_processed, STORE_32bit_signed, DISPLAY_none, LOG_none, ACCUM_average, "
+        "0, 0, 0, 0, 3, 3, NORMALIZE_none, PIPELINE_processed, 0, STREAMING_none, "
+        "DISPLAY_CoilsSideBySide, 1"
+    )
+    write_jcampdx(
+        experiment / "acqp",
+        {
+            "ACQ_word_size": "_32_BIT",
+            "ACQ_sw_version": "<PV-360.3.7>",
+            "BYTORDA": "little",
+            "ACQ_jobs": Verbatim("( 1 )\n(8, 20, 5, 2, 101, 1, 2, 1, <job0>)"),
+            "ACQ_ScanPipeJobSettings": Verbatim(f"( 1 )\n({settings})"),
+            "ACQ_ReceiverSelectPerChan": ["Yes", "Yes"],
+        },
+    )
+    write_jcampdx(experiment / "method", {"Method": "<Bruker:FLASH>", "PVM_EncNReceivers": 2})
+    write_binary(experiment / "rawdata.job0", np.arange(8 * 2 * 3, dtype="<i4"), np.dtype("<i4"))
+
+    with pytest.warns(RuntimeWarning, match="nStoredScans=3 disagrees"):
+        dataset = Dataset(experiment / "rawdata.job0", load=LOAD_STAGES["properties"])
+
+    assert dataset.channels == 2
+    assert dataset.shape_storage == (8, 2, 3)
+
+    discard = tmp_path / "28"
+    discard_settings = (
+        "STORE_discard, STORE_32bit_signed, DISPLAY_none, LOG_none, ACCUM_average, "
+        "0, 0, 0, 0, 1, 1, NORMALIZE_none, PIPELINE_processed, 0, STREAMING_none, "
+        "DISPLAY_CoilsSideBySide, 1"
+    )
+    write_jcampdx(
+        discard / "acqp",
+        {
+            "ACQ_word_size": "_32_BIT",
+            "ACQ_sw_version": "<PV-360.3.7>",
+            "BYTORDA": "little",
+            "ACQ_jobs": Verbatim("( 1 )\n(8, 20, 5, 1, 101, 1, 1, 1, <job0>)"),
+            "ACQ_ScanPipeJobSettings": Verbatim(f"( 1 )\n({discard_settings})"),
+        },
+    )
+    write_jcampdx(discard / "method", {"Method": "<Bruker:FLASH>", "PVM_EncNReceivers": 1})
+    write_binary(discard / "rawdata.job0", np.array([], dtype="<i4"), np.dtype("<i4"))
+
+    with pytest.warns(RuntimeWarning, match="nStoredScans=3 disagrees"):
+        discovered = Experiment(experiment).get_dataset_list()
+    assert [job.path.name for job in discovered] == ["rawdata.job0"]
+    assert Experiment(discard).get_dataset_list() == []
+
+
 def test_a_dollar_comment_inside_a_string_is_data(tmp_path):
     """Spec 2.2: the text inside `<...>` is free-form, `$$` included."""
     path = tmp_path / "method"

test/test_property_configs.py

@@ -80,8 +80,8 @@ def test_rawdata_job_layout_is_selected_by_record_arity():
             "conditions": ["len(#ACQ_jobs.nested[0])==8"],
         },
     ]
-    # spec 3.3: [3] is nTotalScans, which is NAE times the number written
-    assert config["shape_storage"][0]["cmd"] == "(@job_desc[0],) + (#PVM_EncNReceivers,) + (@job_desc[6] if len(@job_desc)==9 else @job_desc[-1],)"
+    # Specs 3.3/13.1: settings nStoredScans defines the physical file size.
+    assert config["shape_storage"][0]["cmd"] == "(@job_desc[0], @rawdata_channels, @rawdata_stored_scans)"
 
 
 def test_fid_pv360_word_size_branches_match_rawdata():