From 34fc89d9903d240fddf200ae29d0eac0d7d866c7 Mon Sep 17 00:00:00 2001 From: Stefano Moia Date: Wed, 9 Oct 2024 21:15:29 +0200 Subject: [PATCH] Clean code, remove moved functions, remove pydra mask check, add docstrings, rename transform_to_physio function --- physutils/tasks.py | 94 +++++++++++++++-------------------- physutils/tests/test_tasks.py | 20 ++++---- 2 files changed, 50 insertions(+), 64 deletions(-) diff --git a/physutils/tasks.py b/physutils/tasks.py index 7dfb306..3da6c56 100644 --- a/physutils/tasks.py +++ b/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 diff --git a/physutils/tests/test_tasks.py b/physutils/tests/test_tasks.py index 9ecdc0b..e81ce43 100644 --- a/physutils/tests/test_tasks.py +++ b/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",