rebase, add basic test

docs/heuristics.rst

@@ -108,6 +108,19 @@ or::
         ...
         return seqinfos  # ordered dict containing seqinfo objects: list of DICOMs
 
+---------------------------------------------------------------
+``custom_seqinfo(wrapper, series_files)``
+---------------------------------------------------------------
+If present this function will be called on each group of dicoms with
+a sample nibabel dicom wrapper to extract additional information
+to be used in ``infotodict``.
+
+Importantly the return value of that function needs to be hashable.
+For instance the following non-hashable types can be converted to an alternative
+hashable type:
+- list > tuple
+- dict > frozendict
+- arrays > bytes (tobytes(), or pickle.dumps), str or tuple of tuples.
 
 -------------------------------
 ``POPULATE_INTENDED_FOR_OPTS``

heudiconv/convert.py

@@ -220,6 +220,9 @@ def prep_conversion(
                 dcmfilter=getattr(heuristic, "filter_dicom", None),
                 flatten=True,
                 custom_grouping=getattr(heuristic, "grouping", None),
+                # callable which will be provided dcminfo and returned
+                # structure extend seqinfo
+                custom_seqinfo=getattr(heuristic, 'custom_seqinfo', None),
             )
         elif seqinfo is None:
             raise ValueError("Neither 'dicoms' nor 'seqinfo' is given")

heudiconv/dicoms.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 import sys
 import tarfile
-from typing import TYPE_CHECKING, Any, Dict, List, NamedTuple, Optional, Union, overload
+from typing import TYPE_CHECKING, Any, Dict, Hashable, List, NamedTuple, Optional, Union, Protocol, cast, overload
 from unittest.mock import patch
 import warnings
 
@@ -42,7 +42,17 @@
 compresslevel = 9
 
 
-def create_seqinfo(mw: dw.Wrapper, series_files: list[str], series_id: str) -> SeqInfo:
+class CustomSeqinfoT(Protocol):
+    def __call__(self, wrapper: dw.Wrapper, series_files: list[str]) -> Hashable:
+        ...
+
+
+def create_seqinfo(
+    mw: dw.Wrapper,
+    series_files: list[str],
+    series_id: str,
+    custom_seqinfo: CustomSeqinfoT | None = None,
+) -> SeqInfo:
     """Generate sequence info
 
     Parameters
@@ -80,6 +90,15 @@ def create_seqinfo(mw: dw.Wrapper, series_files: list[str], series_id: str) -> S
     global total_files
     total_files += len(series_files)
 
+    custom_seqinfo_data = custom_seqinfo(wrapper=mw, series_files=series_files) \
+        if custom_seqinfo else None
+    try:
+        hash(custom_seqinfo_data)
+    except TypeError:
+        raise RuntimeError("Data returned by the heuristics custom_seqinfo is not hashable. "
+                           "See https://heudiconv.readthedocs.io/en/latest/heuristics.html#custom_seqinfo for more "
+                           "details.")
+
     return SeqInfo(
         total_files_till_now=total_files,
         example_dcm_file=op.basename(series_files[0]),
@@ -109,6 +128,7 @@ def create_seqinfo(mw: dw.Wrapper, series_files: list[str], series_id: str) -> S
         date=dcminfo.get("AcquisitionDate"),
         series_uid=dcminfo.get("SeriesInstanceUID"),
         time=dcminfo.get("AcquisitionTime"),
+        custom=custom_seqinfo_data,
     )
 
 
@@ -181,6 +201,7 @@ def group_dicoms_into_seqinfos(
         dict[SeqInfo, list[str]],
     ]
     | None = None,
+    custom_seqinfo: CustomSeqinfoT | None = None,
 ) -> dict[Optional[str], dict[SeqInfo, list[str]]]:
     ...
 
@@ -199,6 +220,7 @@ def group_dicoms_into_seqinfos(
         dict[SeqInfo, list[str]],
     ]
     | None = None,
+    custom_seqinfo: CustomSeqinfoT | None = None,
 ) -> dict[SeqInfo, list[str]]:
     ...
 
@@ -215,6 +237,7 @@ def group_dicoms_into_seqinfos(
         dict[SeqInfo, list[str]],
     ]
     | None = None,
+    custom_seqinfo: CustomSeqinfoT | None = None,
 ) -> dict[Optional[str], dict[SeqInfo, list[str]]] | dict[SeqInfo, list[str]]:
     """Process list of dicoms and return seqinfo and file group
     `seqinfo` contains per-sequence extract of fields from DICOMs which
@@ -236,9 +259,11 @@ def group_dicoms_into_seqinfos(
       Creates a flattened `seqinfo` with corresponding DICOM files. True when
       invoked with `dicom_dir_template`.
     custom_grouping: str or callable, optional
-     grouping key defined within heuristic. Can be a string of a
-     DICOM attribute, or a method that handles more complex groupings.
-
+      grouping key defined within heuristic. Can be a string of a
+      DICOM attribute, or a method that handles more complex groupings.
+    custom_seqinfo: callable, optional
+      A callable which will be provided MosaicWrapper giving possibility to
+      extract any custom DICOM metadata of interest.
 
     Returns
     -------
@@ -358,7 +383,7 @@ def group_dicoms_into_seqinfos(
             else:
                 # nothing to see here, just move on
                 continue
-        seqinfo = create_seqinfo(mw, series_files, series_id_str)
+        seqinfo = create_seqinfo(mw, series_files, series_id_str, custom_seqinfo)
 
         key: Optional[str]
         if per_studyUID:

heudiconv/heuristics/convertall.py

@@ -1,9 +1,13 @@
 from __future__ import annotations
 
+import logging
 from typing import Optional
 
+from heudiconv.dicoms import dw
 from heudiconv.utils import SeqInfo
 
+lgr = logging.getLogger('heudiconv')
+
 
 def create_key(
     template: Optional[str],
@@ -15,6 +19,20 @@ def create_key(
     return (template, outtype, annotation_classes)
 
 
+def custom_seqinfo(wrapper: dw.Wrapper, series_files: list[str]) -> tuple[str, str]:
+    # Just a dummy demo for what custom_seqinfo could get/do
+    # for already loaded DICOM data, and including storing/returning
+    # the sample series file as was requested
+    # in https://github.com/nipy/heudiconv/pull/333
+    from nibabel.nicom.dicomwrappers import WrapperError
+    try:
+        affine = wrapper.affine.tobytes()
+    except WrapperError:
+        lgr.exception("Errored out while obtaining/converting affine")
+        affine = None
+    return affine, series_files[0]
+
+
 def infotodict(
     seqinfo: list[SeqInfo],
 ) -> dict[tuple[str, tuple[str, ...], None], list[str]]:

heudiconv/parser.py

@@ -224,6 +224,7 @@ def get_study_sessions(
             file_filter=getattr(heuristic, "filter_files", None),
             dcmfilter=getattr(heuristic, "filter_dicom", None),
             custom_grouping=getattr(heuristic, "grouping", None),
+            custom_seqinfo=getattr(heuristic, 'custom_seqinfo', None),
         )
 
         if sids:

heudiconv/tests/test_dicoms.py

@@ -99,6 +99,27 @@ def test_group_dicoms_into_seqinfos() -> None:
     ]
 
 
+def test_custom_seqinfo() -> None:
+    """Tests for custom seqinfo extraction"""
+
+    from heudiconv.heuristics.convertall import custom_seqinfo
+
+    dcmfiles = glob(op.join(TESTS_DATA_PATH, "phantom.dcm"))
+
+    seqinfos = group_dicoms_into_seqinfos(
+        dcmfiles,
+        "studyUID",
+        flatten=True,
+        custom_seqinfo=custom_seqinfo)
+
+    seqinfo = list(seqinfos.keys())[0]
+
+    assert hasattr(seqinfo, 'custom')
+    assert isinstance(seqinfo.custom, tuple)
+    assert len(seqinfo.custom) == 2
+    assert seqinfo.custom[1] == dcmfiles[0]
+
+
 def test_get_datetime_from_dcm_from_acq_date_time() -> None:
     typical_dcm = dcm.dcmread(
         op.join(TESTS_DATA_PATH, "phantom.dcm"), stop_before_pixels=True

heudiconv/utils.py

@@ -24,6 +24,7 @@
 from typing import (
     Any,
     AnyStr,
+    Hashable,
     Mapping,
     NamedTuple,
     Optional,
@@ -69,6 +70,7 @@ class SeqInfo(NamedTuple):
     date: Optional[str]  # 24
     series_uid: Optional[str]  # 25
     time: Optional[str]  # 26
+    custom: Optional[Hashable]  # 27
 
 
 class StudySessionInfo(NamedTuple):