Merge handles_phoenix_file into dcm_physio

heudiconv/dicoms.py

@@ -266,8 +266,11 @@ def group_dicoms_into_seqinfos(files, grouping, file_filter=None,
         series_id = '-'.join(map(str, series_id))
         if mw.image_shape is None:
             # this whole thing has no image data (maybe just PSg DICOMs)
-            # If this is a Siemens PhysioLog, keep it:
-            if mw.dcm_data.SeriesDescription.endswith('_PhysioLog'):
+            # If this is a Siemens PhoenixZipReport or PhysioLog, keep it:
+            if (
+                mw.dcm_data.SeriesDescription == 'PhoenixZIPReport'
+                or mw.dcm_data.SeriesDescription.endswith('_PhysioLog')
+            ):
                 # just give it a dummy shape, so that we can continue:
                 mw.image_shape = (0, 0, 0)
             else:

heudiconv/heuristics/bids_PhoenixReport.py

@@ -0,0 +1,38 @@
+"""Heuristic demonstrating conversion of the PhoenixZIPReport from Siemens.
+
+It only cares about converting a series with have PhoenixZIPReport in their
+series_description and outputs **only to sourcedata**.
+"""
+
+
+def create_key(template, outtype=('nii.gz',), annotation_classes=None):
+    if template is None or not template:
+        raise ValueError('Template must be a valid format string')
+    return template, outtype, annotation_classes
+
+
+def infotodict(seqinfo):
+    """Heuristic evaluator for determining which runs belong where
+
+    allowed template fields - follow python string module:
+
+    item: index within category
+    subject: participant id
+    seqitem: run number during scanning
+    subindex: sub index within group
+    """
+    sbref = create_key('sub-{subject}/func/sub-{subject}_task-QA_sbref', outtype=('nii.gz', 'dicom',))
+    scout = create_key('sub-{subject}/anat/sub-{subject}_T1w', outtype=('nii.gz', 'dicom',))
+    phoenix_doc = create_key('sub-{subject}/misc/sub-{subject}_phoenix', outtype=('dicom',))
+
+    info = {sbref: [], scout: [], phoenix_doc: []}
+    for s in seqinfo:
+        if (
+            'PhoenixZIPReport' in s.series_description
+            and s.image_type[3] == 'CSA REPORT'
+        ):
+            info[phoenix_doc].append({'item': s.series_id})
+        if 'scout' in s.series_description.lower():
+            info[scout].append({'item': s.series_id})
+
+    return info

heudiconv/tests/test_dicoms.py

@@ -1,12 +1,18 @@
 import os.path as op
 import json
+from glob import glob
 
 import pytest
 
 from heudiconv.external.pydicom import dcm
 from heudiconv.cli.run import main as runner
 from heudiconv.convert import nipype_convert
-from heudiconv.dicoms import parse_private_csa_header, embed_dicom_and_nifti_metadata
+from heudiconv.dicoms import (
+    OrderedDict,
+    embed_dicom_and_nifti_metadata,
+    group_dicoms_into_seqinfos,
+    parse_private_csa_header,
+)
 from .utils import (
     assert_cwd_unchanged,
     TESTS_DATA_PATH,
@@ -64,3 +70,18 @@ def test_embed_dicom_and_nifti_metadata(tmpdir):
 
     assert out3.pop("existing") == "data"
     assert out3 == out2
+
+
+def test_group_dicoms_into_seqinfos(tmpdir):
+    """Tests for group_dicoms_into_seqinfos"""
+
+    # 1) Check that it works for PhoenixDocuments:
+    # set up testing files
+    dcmfolder = op.join(TESTS_DATA_PATH, 'Phoenix')
+    dcmfiles = glob(op.join(dcmfolder, '*', '*.dcm'))
+
+    seqinfo = group_dicoms_into_seqinfos(dcmfiles, 'studyUID', flatten=True)
+
+    assert type(seqinfo) is OrderedDict
+    assert len(seqinfo) == len(dcmfiles)
+    assert [s.series_description for s in seqinfo] == ['AAHead_Scout_32ch-head-coil', 'PhoenixZIPReport']

heudiconv/tests/test_heuristics.py

@@ -176,3 +176,20 @@ def test_notop(tmpdir, bidsoptions):
             assert not op.exists(pjoin(tmppath, 'Halchenko/Yarik/950_bids_test4', fname))
         else:
             assert op.exists(pjoin(tmppath, 'Halchenko/Yarik/950_bids_test4', fname))
+
+
+def test_phoenix_doc_conversion(tmpdir):
+    tmppath = tmpdir.strpath
+    subID = 'Phoenix'
+    args = (
+        "-c dcm2niix -o %s -b -f bids_PhoenixReport --files %s -s %s"
+        % (tmpdir, pjoin(TESTS_DATA_PATH, 'Phoenix'), subID)
+    ).split(' ')
+    runner(args)
+
+    # check that the Phoenix document has been extracted (as gzipped dicom) in
+    # the sourcedata/misc folder:
+    assert op.exists(pjoin(tmppath, 'sourcedata', 'sub-%s', 'misc', 'sub-%s_phoenix.dicom.tgz') % (subID, subID))
+    # check that no "sub-<subID>/misc" folder has been created in the BIDS
+    # structure:
+    assert not op.exists(pjoin(tmppath, 'sub-%s', 'misc') % subID)