Add support for a custom seqinfo to extract from DICOMs any additional metadata desired for a heuristic

README.rst

@@ -71,7 +71,7 @@ You can run your conversion automatically (which will produce a ``.heudiconv`` d
 .. image:: figs/workflow.png
 
 
-``heudiconv`` comes with `existing heuristics <https://github.com/nipy/heudiconv/tree/master/heudiconv/heuristics>`_ which can be used as is, or as examples. 
+``heudiconv`` comes with `existing heuristics <https://github.com/nipy/heudiconv/tree/master/heudiconv/heuristics>`_ which can be used as is, or as examples.
 For instance, the Heuristic `convertall <https://github.com/nipy/heudiconv/blob/master/heudiconv/heuristics/convertall.py>`_ extracts standard metadata from all matching DICOMs.
 ``heudiconv`` creates mapping files, ``<something>.edit.text`` which lets researchers simply establish their own conversion mapping.
 

docs/heuristics.rst

@@ -119,6 +119,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

@@ -221,6 +221,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,18 @@
 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,
+    Protocol,
+    Union,
+    overload,
+)
 from unittest.mock import patch
 import warnings
 
@@ -42,7 +53,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 +101,20 @@ 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 +144,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 +217,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 +236,7 @@ def group_dicoms_into_seqinfos(
         dict[SeqInfo, list[str]],
     ]
     | None = None,
+    custom_seqinfo: CustomSeqinfoT | None = None,
 ) -> dict[SeqInfo, list[str]]:
     ...
 
@@ -215,6 +253,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 +275,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 +399,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,12 @@
 from __future__ import annotations
 
+import logging
 from typing import Optional
 
 from heudiconv.utils import SeqInfo
 
+lgr = logging.getLogger("heudiconv")
+
 
 def create_key(
     template: Optional[str],

heudiconv/heuristics/convertall_custom.py

@@ -0,0 +1,32 @@
+"""A demo convertall heuristic with custom_seqinfo extracting affine and sample DICOM path
+
+This heuristic also demonstrates on how to create a "derived" heuristic which would augment
+behavior of an already existing heuristic without complete rewrite.  Such approach could be
+useful for heuristic like  reproin  to overload mapping etc.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import nibabel.nicom.dicomwrappers as dw
+
+from .convertall import *  # noqa: F403
+
+
+def custom_seqinfo(
+    series_files: list[str], wrapper: dw.Wrapper, **kw: Any  # noqa: U100
+) -> tuple[str | None, str]:
+    """Demo for extracting custom header fields into custom_seqinfo field
+
+    Operates on already loaded DICOM data.
+    Origin: https://github.com/nipy/heudiconv/pull/333
+    """
+
+    from nibabel.nicom.dicomwrappers import WrapperError
+
+    try:
+        affine = str(wrapper.affine)
+    except WrapperError:
+        lgr.exception("Errored out while obtaining/converting affine")  # noqa: F405
+        affine = None
+    return affine, series_files[0]

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,25 @@ def test_group_dicoms_into_seqinfos() -> None:
     ]
 
 
+def test_custom_seqinfo() -> None:
+    """Tests for custom seqinfo extraction"""
+
+    from heudiconv.heuristics.convertall_custom 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
+    )  # type: ignore
+
+    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):