Add BIDS reading support and prepare input loading for pydra workflow #7
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| import logging | ||
|
|
||
| import pydra | ||
|
||
|
|
||
| from physutils.io import load_from_bids, load_physio | ||
| from physutils.physio import Physio | ||
|
|
||
| LGR = logging.getLogger(__name__) | ||
| LGR.setLevel(logging.DEBUG) | ||
|
|
||
|
|
||
| @pydra.mark.task | ||
| def transform_to_physio( | ||
| input_file: str, mode="physio", fs=None, bids_parameters=dict(), bids_channel=None | ||
|
||
| ) -> Physio: | ||
| LGR.debug(f"Loading physio object from {input_file}") | ||
| if not fs: | ||
| fs = None | ||
|
||
|
|
||
| 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) | ||
|
|
||
| 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] | ||
| else: | ||
| raise ValueError(f"Invalid transform_to_physio mode: {mode}") | ||
| return physio_obj | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| """Tests for physutils.tasks and their integration.""" | ||
|
|
||
| import os | ||
|
|
||
| import physutils.tasks as tasks | ||
| from physutils import physio | ||
| from physutils.tests.utils import create_random_bids_structure | ||
|
|
||
|
|
||
| def test_transform_to_physio_phys_file(): | ||
| """Test transform_to_physio task.""" | ||
| physio_file = os.path.abspath("physutils/tests/data/ECG.phys") | ||
| task = tasks.transform_to_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 | ||
|
Comment on lines
+14
to
+16
Contributor
There was a problem hiding this comment. pytest is failing..
Contributor
Author
There was a problem hiding this comment.
|
||
|
|
||
| task() | ||
me-pic marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| physio_obj = task.result().output.out | ||
| assert isinstance(physio_obj, physio.Physio) | ||
| assert physio_obj.fs == 1000 | ||
| assert physio_obj.data.shape == (44611,) | ||
|
|
||
|
|
||
| def test_transform_to_physio_bids_file(): | ||
| """Test transform_to_physio task.""" | ||
| 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="bids", | ||
| bids_parameters=bids_parameters, | ||
| bids_channel="cardiac", | ||
|
||
| ) | ||
|
|
||
| assert task.inputs.input_file == bids_dir | ||
| assert task.inputs.mode == "bids" | ||
| 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) | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@maestroque please put yourself before Vicente (in alphabetical order of family name). You have all of the above, minus review, but plus documentation.
@me-pic and @m-miedema please add yourselves as well, mentoring and review.