Clean code, remove moved functions, remove pydra mask check, add docstrings, rename transform_to_physio function

Author: smoia

physutils/tasks.py

@@ -1,63 +1,49 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""Helper class for holding physiological data and associated metadata information."""
+
 import logging
-from functools import wraps
 
-from bids import BIDSLayout
-from loguru import logger
+from .io import load_from_bids, load_physio
+from .physio import Physio
+from .utils import is_bids_directory
 
-from physutils.io import load_from_bids, load_physio
-from physutils.physio import Physio
+# from loguru import logger
 
-LGR = logging.getLogger(__name__)
-LGR.setLevel(logging.DEBUG)
 
 try:
-    import pydra
-
-    pydra_imported = True
+    from pydra import task
 except ImportError:
-    pydra_imported = False
-
-
-def mark_task(pydra_imported=pydra_imported):
-    def decorator(func):
-        if pydra_imported:
-            # If the decorator exists, apply it
-            @wraps(func)
-            def wrapped_func(*args, **kwargs):
-                logger.debug(f"Creating pydra task for {func.__name__}")
-                return pydra.mark.task(func)(*args, **kwargs)
+    from .utils import task
 
-            return wrapped_func
-        # Otherwise, return the original function
-        return func
 
-    return decorator
+LGR = logging.getLogger(__name__)
+LGR.setLevel(logging.DEBUG)
 
 
-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
+@task
+def generate_physio(
+    input_file: str, mode="auto", fs=None, bids_parameters=dict(), col_physio_type=None
+) -> Physio:
+    """
+    Load a physio object from either a BIDS directory or an exported physio object.
 
+    Parameters
+    ----------
+    input_file : str
+        Path to input file
+    mode : 'auto', 'physio', or 'bids', optional
+        Mode to operate with
+    fs : None, optional
+        Set or force set sapmling frequency (Hz).
+    bids_parameters : dictionary, optional
+        Dictionary containing BIDS parameters
+    col_physio_type : int or None, optional
+        Object to pick up in a BIDS array of physio objects.
 
-@mark_task(pydra_imported=pydra_imported)
-def transform_to_physio(
-    input_file: str, mode="physio", fs=None, bids_parameters=dict(), bids_channel=None
-) -> Physio:
-    if not pydra_imported:
-        LGR.warning(
-            "Pydra is not installed, thus transform_to_physio is not available as a pydra task. Using the function directly"
-        )
-    LGR.debug(f"Loading physio object from {input_file}")
-    if not fs:
-        fs = None
+    """
+    LGR.info(f"Loading physio object from {input_file}")
 
     if mode == "auto":
         if input_file.endswith((".phys", ".physio", ".1D", ".txt", ".tsv", ".csv")):
@@ -66,20 +52,20 @@ def transform_to_physio(
             mode = "bids"
         else:
             raise ValueError(
-                "Could not determine mode automatically, please specify mode"
+                "Could not determine input mode automatically. Please specify it manually."
             )
     if mode == "physio":
-        if fs is not None:
-            physio_obj = load_physio(input_file, fs=fs, allow_pickle=True)
-        else:
-            physio_obj = load_physio(input_file, allow_pickle=True)
+        physio_obj = load_physio(input_file, fs=fs, allow_pickle=True)
 
     elif mode == "bids":
         if bids_parameters is {}:
             raise ValueError("BIDS parameters must be provided when loading from BIDS")
         else:
             physio_array = load_from_bids(input_file, **bids_parameters)
-            physio_obj = physio_array[bids_channel]
+            physio_obj = (
+                physio_array[col_physio_type] if col_physio_type else physio_array
+            )
     else:
-        raise ValueError(f"Invalid transform_to_physio mode: {mode}")
+        raise ValueError(f"Invalid generate_physio mode: {mode}")
+
     return physio_obj

physutils/tests/test_tasks.py

@@ -7,10 +7,10 @@
 from physutils.tests.utils import create_random_bids_structure
 
 
-def test_transform_to_physio_phys_file():
-    """Test transform_to_physio task."""
+def test_generate_physio_phys_file():
+    """Test generate_physio task."""
     physio_file = os.path.abspath("physutils/tests/data/ECG.phys")
-    task = tasks.transform_to_physio(input_file=physio_file, mode="physio")
+    task = tasks.generate_physio(input_file=physio_file, mode="physio")
     assert task.inputs.input_file == physio_file
     assert task.inputs.mode == "physio"
     assert task.inputs.fs is None
@@ -23,8 +23,8 @@ def test_transform_to_physio_phys_file():
     assert physio_obj.data.shape == (44611,)
 
 
-def test_transform_to_physio_bids_file():
-    """Test transform_to_physio task."""
+def test_generate_physio_bids_file():
+    """Test generate_physio task."""
     create_random_bids_structure("physutils/tests/data", recording_id="cardiac")
     bids_parameters = {
         "subject": "01",
@@ -34,7 +34,7 @@ def test_transform_to_physio_bids_file():
         "recording": "cardiac",
     }
     bids_dir = os.path.abspath("physutils/tests/data/bids-dir")
-    task = tasks.transform_to_physio(
+    task = tasks.generate_physio(
         input_file=bids_dir,
         mode="bids",
         bids_parameters=bids_parameters,
@@ -53,7 +53,7 @@ def test_transform_to_physio_bids_file():
     assert isinstance(physio_obj, physio.Physio)
 
 
-def test_transform_to_physio_auto():
+def test_generate_physio_auto():
     create_random_bids_structure("physutils/tests/data", recording_id="cardiac")
     bids_parameters = {
         "subject": "01",
@@ -63,7 +63,7 @@ def test_transform_to_physio_auto():
         "recording": "cardiac",
     }
     bids_dir = os.path.abspath("physutils/tests/data/bids-dir")
-    task = tasks.transform_to_physio(
+    task = tasks.generate_physio(
         input_file=bids_dir,
         mode="auto",
         bids_parameters=bids_parameters,
@@ -82,9 +82,9 @@ def test_transform_to_physio_auto():
     assert isinstance(physio_obj, physio.Physio)
 
 
-def test_transform_to_physio_auto_error(caplog):
+def test_generate_physio_auto_error(caplog):
     bids_dir = os.path.abspath("physutils/tests/data/non-bids-dir")
-    task = tasks.transform_to_physio(
+    task = tasks.generate_physio(
         input_file=bids_dir,
         mode="auto",
         bids_channel="cardiac",