Add transform_to_physio auto mode

Author: maestroque

physutils/tasks.py

@@ -1,6 +1,7 @@
 import logging
 from functools import wraps
 
+from bids import BIDSLayout
 from loguru import logger
 
 from physutils.io import load_from_bids, load_physio
@@ -33,6 +34,19 @@ def wrapped_func(*args, **kwargs):
     return decorator
 
 
+def is_bids_directory(directory):
+    try:
+        # Attempt to create a BIDSLayout object
+        _ = BIDSLayout(directory)
+        return True
+    except Exception as e:
+        # Catch other exceptions that might indicate the directory isn't BIDS compliant
+        logger.error(
+            f"An error occurred while trying to load {directory} as a BIDS Layout object: {e}"
+        )
+        return False
+
+
 @mark_task(pydra_imported=pydra_imported)
 def transform_to_physio(
     input_file: str, mode="physio", fs=None, bids_parameters=dict(), bids_channel=None
@@ -45,6 +59,15 @@ def transform_to_physio(
     if not fs:
         fs = None
 
+    if mode == "auto":
+        if input_file.endswith((".phys", ".physio", ".1D", ".txt", ".tsv", ".csv")):
+            mode = "physio"
+        elif is_bids_directory(input_file):
+            mode = "bids"
+        else:
+            raise ValueError(
+                "Could not determine mode automatically, please specify mode"
+            )
     if mode == "physio":
         if fs is not None:
             physio_obj = load_physio(input_file, fs=fs, allow_pickle=True)

physutils/tests/data/non-bids-dir/sub-01/ses-01/func/sub-01_ses-01_task-rest_run-01_physio.json

@@ -0,0 +1,12 @@
+{
+    "SamplingFrequency": 10000.0,
+    "StartTime": -3,
+    "Columns": [
+        "time",
+        "respiratory_chest",
+        "trigger",
+        "cardiac",
+        "respiratory_CO2",
+        "respiratory_O2"
+    ]
+}

physutils/tests/data/non-bids-dir/sub-01/ses-01/func/sub-01_ses-01_task-rest_run-01_recording-cardiac_physio.json

@@ -0,0 +1,12 @@
+{
+    "SamplingFrequency": 10000.0,
+    "StartTime": -3,
+    "Columns": [
+        "time",
+        "respiratory_chest",
+        "trigger",
+        "cardiac",
+        "respiratory_CO2",
+        "respiratory_O2"
+    ]
+}

physutils/tests/test_tasks.py

@@ -51,3 +51,57 @@ def test_transform_to_physio_bids_file():
 
     physio_obj = task.result().output.out
     assert isinstance(physio_obj, physio.Physio)
+
+
+def test_transform_to_physio_auto():
+    create_random_bids_structure("physutils/tests/data", recording_id="cardiac")
+    bids_parameters = {
+        "subject": "01",
+        "session": "01",
+        "task": "rest",
+        "run": "01",
+        "recording": "cardiac",
+    }
+    bids_dir = os.path.abspath("physutils/tests/data/bids-dir")
+    task = tasks.transform_to_physio(
+        input_file=bids_dir,
+        mode="auto",
+        bids_parameters=bids_parameters,
+        bids_channel="cardiac",
+    )
+
+    assert task.inputs.input_file == bids_dir
+    assert task.inputs.mode == "auto"
+    assert task.inputs.fs is None
+    assert task.inputs.bids_parameters == bids_parameters
+    assert task.inputs.bids_channel == "cardiac"
+
+    task()
+
+    physio_obj = task.result().output.out
+    assert isinstance(physio_obj, physio.Physio)
+
+
+def test_transform_to_physio_auto_error(caplog):
+    bids_dir = os.path.abspath("physutils/tests/data/non-bids-dir")
+    task = tasks.transform_to_physio(
+        input_file=bids_dir,
+        mode="auto",
+        bids_channel="cardiac",
+    )
+
+    assert task.inputs.input_file == bids_dir
+    assert task.inputs.mode == "auto"
+    assert task.inputs.fs is None
+    assert task.inputs.bids_channel == "cardiac"
+
+    try:
+        task()
+    except Exception:
+        assert caplog.text.count("ERROR") == 1
+        assert (
+            caplog.text.count(
+                "dataset_description.json' is missing from project root. Every valid BIDS dataset must have this file."
+            )
+            == 1
+        )