RF: Add sanitized_id field to FieldmapEstimation

sdcflows/fieldmaps.py

@@ -300,6 +300,9 @@ class FieldmapEstimation:
     bids_id = attr.ib(default=None, kw_only=True, type=str, on_setattr=_id_setter)
     """The unique ``B0FieldIdentifier`` field of this fieldmap."""
 
+    sanitized_id = attr.ib(init=False, repr=False)
+    """Sanitized version of the bids_id with special characters replaced by underscores."""
+
     _wf = attr.ib(init=False, default=None, repr=False)
     """Internal pointer to a workflow."""
 
@@ -436,6 +439,10 @@ def __attrs_post_init__(self):
         for intent_file in intents_meta:
             _intents[intent_file].add(self.bids_id)
 
+        # Provide a sanitized identifier that can be used in cases where
+        # special characters are not allowed.
+        self.sanitized_id = re.sub(r'[^a-zA-Z0-9]', '_', self.bids_id)
+
     def paths(self):
         """Return a tuple of paths that are sorted."""
         return tuple(sorted(str(f.path) for f in self.sources))
@@ -446,7 +453,7 @@ def get_workflow(self, set_inputs=True, **kwargs):
             return self._wf
 
         # Override workflow name
-        kwargs["name"] = f"wf_{self.bids_id}"
+        kwargs["name"] = f"wf_{self.sanitized_id}"
 
         if self.method in (EstimatorType.MAPPED, EstimatorType.PHASEDIFF):
             from .workflows.fit.fieldmap import init_fmap_wf

sdcflows/tests/test_fieldmaps.py

@@ -111,6 +111,7 @@ def test_FieldmapEstimation(dsA_dir, inputfiles, method, nsources, raises):
     assert fe.method == method
     assert len(fe.sources) == nsources
     assert fe.bids_id is not None and fe.bids_id.startswith("auto_")
+    assert fe.bids_id == fe.sanitized_id  # Auto-generated IDs are sanitized
 
     # Attempt to change bids_id
     with pytest.raises(ValueError):
@@ -243,6 +244,33 @@ def test_FieldmapEstimationIdentifier(monkeypatch, dsA_dir):
 
     fm.clear_registry()
 
+    fe = fm.FieldmapEstimation(
+        [
+            fm.FieldmapFile(
+                dsA_dir / "sub-01" / "fmap/sub-01_fieldmap.nii.gz",
+                metadata={
+                    "Units": "Hz",
+                    "B0FieldIdentifier": "fmap-with^special#chars",
+                    "IntendedFor": ["file1.nii.gz", "file2.nii.gz"],
+                },
+            ),
+            fm.FieldmapFile(
+                dsA_dir / "sub-01" / "fmap/sub-01_magnitude.nii.gz",
+                metadata={"Units": "Hz", "B0FieldIdentifier": "fmap-with^special#chars"},
+            ),
+        ]
+    )
+    assert fe.bids_id == "fmap-with^special#chars"
+    assert fe.sanitized_id == "fmap_with_special_chars"
+    # The unsanitized ID is used for lookups
+    assert fm.get_identifier("file1.nii.gz") == ("fmap-with^special#chars",)
+    assert fm.get_identifier("file2.nii.gz") == ("fmap-with^special#chars",)
+
+    wf = fe.get_workflow()
+    assert wf.name == "wf_fmap_with_special_chars"
+
+    fm.clear_registry()
+
 
 def test_type_setter():
     """Cover the _type_setter routine."""

sdcflows/workflows/base.py

@@ -124,7 +124,7 @@ def init_fmap_preproc_wf(
             output_dir=str(output_dir),
             write_coeff=True,
             bids_fmap_id=estimator.bids_id,
-            name=f"fmap_derivatives_wf_{estimator.bids_id}",
+            name=f"fmap_derivatives_wf_{estimator.sanitized_id}",
         )
         fmap_derivatives_wf.inputs.inputnode.source_files = source_files
         fmap_derivatives_wf.inputs.inputnode.fmap_meta = [
@@ -135,15 +135,15 @@ def init_fmap_preproc_wf(
             output_dir=str(output_dir),
             fmap_type=str(estimator.method).rpartition(".")[-1].lower(),
             bids_fmap_id=estimator.bids_id,
-            name=f"fmap_reports_wf_{estimator.bids_id}",
+            name=f"fmap_reports_wf_{estimator.sanitized_id}",
         )
         fmap_reports_wf.inputs.inputnode.source_files = source_files
 
         if estimator.method not in (EstimatorType.MAPPED, EstimatorType.PHASEDIFF):
             fields = INPUT_FIELDS[estimator.method]
             inputnode = pe.Node(
                 niu.IdentityInterface(fields=fields),
-                name=f"in_{estimator.bids_id}",
+                name=f"in_{estimator.sanitized_id}",
             )
             # fmt:off
             workflow.connect([

sdcflows/workflows/fit/base.py

@@ -61,7 +61,7 @@ def init_sdcflows_wf():
                 output_dir=config.execution.output_dir,
                 bids_fmap_id=estim.bids_id,
                 write_coeff=True,
-                name=f"fmap_derivatives_{estim.bids_id}",
+                name=f"fmap_derivatives_{estim.sanitized_id}",
             )
 
             source_paths = [
@@ -76,7 +76,7 @@ def init_sdcflows_wf():
                 fmap_type=estim.method,
                 output_dir=config.execution.output_dir,
                 bids_fmap_id=estim.bids_id,
-                name=f"fmap_reports_{estim.bids_id}",
+                name=f"fmap_reports_{estim.sanitized_id}",
             )
             reportlets_wf.inputs.inputnode.source_files = source_paths
 

sdcflows/workflows/outputs.py

@@ -21,6 +21,8 @@
 #     https://www.nipreps.org/community/licensing/
 #
 """Writing out outputs."""
+import re
+
 from nipype.pipeline import engine as pe
 from nipype.interfaces import utility as niu
 from niworkflows.interfaces.bids import DerivativesDataSink as _DDS
@@ -77,7 +79,7 @@ def init_fmap_reports_wf(
 
     custom_entities = custom_entities or {}
     if bids_fmap_id:
-        custom_entities["fmapid"] = bids_fmap_id.replace("_", "")
+        custom_entities["fmapid"] = re.sub(r'[^a-zA-Z0-9]', '', bids_fmap_id)
 
     workflow = pe.Workflow(name=name)
     inputnode = pe.Node(
@@ -156,7 +158,7 @@ def init_fmap_derivatives_wf(
     """
     custom_entities = custom_entities or {}
     if bids_fmap_id:
-        custom_entities["fmapid"] = bids_fmap_id.replace("_", "")
+        custom_entities["fmapid"] = re.sub(r'[^a-zA-Z0-9]', '', bids_fmap_id)
 
     workflow = pe.Workflow(name=name)
     inputnode = pe.Node(