Adds ParallelGroupAFQ

[teresamg]

Creates a ParallelGroupAFQ class inheriting from GroupAFQ to allow for easier parallelization through pydra. Does not currently work; in progress.

[pep8speaks]

Hello @teresamg! Thanks for updating this PR. We checked the lines you've touched for PEP 8 issues, and found:

• In the file AFQ/api/group.py:

Line 352:80: E501 line too long (86 > 79 characters)
Line 1012:1: W293 blank line contains whitespace

Comment last updated at 2024-05-15 23:45:53 UTC

[arokem]

This is a really neat implementation now! My main suggestion is to change ParticipantAFQ so that the names of kwargs and attributes are well-matched, so that you can make this even neater.

[arokem]

The code looks quite neat now!

@36000 : could you take a quick look when you get a chance?

Any ideas about how we can test this? I guess that we'd have to set this up as a nightly test, possibly running this parallelized with concurrent futures across two HBN subjects?

[36000]

my thoughts, I will implement these:

1. add a similar looking export function to wrap export like the export_all implemented here
2. add pydra to setup.cfg
3. add a nighly test on 2 hbn subjects using concurrent futures for pydra

[36000]

I made some minor changes here. I made pydra a required library for pyAFQ (its pip installable with few dependencies, this will just be easier for most users). We now try to catch that error Ariel mentioned (let me know if this doesnt work). And I added an export function based on the export_all function. Next I will add the nightly test then I think we can merge this

[36000]

I wrote this test:

def test_AFQ_pydra():
    _, bids_path = afd.fetch_hbn_preproc(["NDARAA948VFH", "NDARAV554TP2"])
    pga = ParallelGroupAFQ(bids_path, preproc_pipeline="qsiprep")
    pga.export_all()

and i am getting this error:

FAILED AFQ/tests/test_api.py::test_AFQ_pydra - ValueError: Split is missing values for the following fields ['pAFQ_kwargs']

Either of you two know what causes this?

[arokem]

Could you please push that test to this PR? I'm not sure what's up and would like to try to debug locally.

[36000]

Sure, I figured out this was due to me using a different pydra version (the latest, 0.23). But now I am getting picking problems. I will push some of the changes I have made, but I think maybe we will have to talk about this in person at some point. I am running into a few issues

[36000]

@teresamg mind trying this to see if it works?
I factored out bidslayout

[teresamg]

Looks like it ran successfully, both locally and on Hyak!

[arokem]

Did it generate the combined tract profiles file? I also ran a test on hyak:

import os.path as op

from AFQ.api.group import ParallelGroupAFQ
from AFQ.definitions.image import RoiImage
import AFQ.api.bundle_dict as abd
import AFQ.data.fetch as afd


_, bids_path = afd.fetch_hbn_preproc(
        ["NDARZT957CWG",
         "NDARZU279XR3",
         "NDARZU401RCU",
         "NDARZU822WN3",
         "NDARZV421TCZ",
         "NDARZW262ZLV",
         "NDARZX163EWC",
         ],
        path="/gscratch/scrubbed/arokem/data/")
    
my_afq = ParallelGroupAFQ(
    bids_path=bids_path,
    preproc_pipeline="qsiprep",
    parallel_params={
        "submitter_params": {
            "plugin": "slurm",
            "sbatch_args": "-J test \
                            -p ckpt \
                            --nodes=1 \
                            --cpus-per-task=8 \
                            --gpus=1 \
                            -A escience \
                            --mem=64G \
                            --time=2:00:00 \
                            -o /gscratch/scrubbed/arokem/logs/test.out \
                            -e /gscratch/scrubbed/arokem/logs/test.err \
                            --mail-user=arokem@uw.edu \
                            --mail-type=ALL"
        },
        "cache_dir": "/gscratch/scrubbed/arokem/tmp"
    }
)

my_afq.export_all()

which worked great at individual subject level, but I can't find the final combined tract profiles file, which I was expecting to find.

[teresamg]

It did for me... does each subject have the *_desc-profiles_dwi.csv?

[36000]

I am a little confused at what these lines here are trying to do

[arokem]

The idea behind this is that we want to generate a group-level csv file that contains the merged tract profiles. But we only want to do this once, so each task checks whether all of the other tasks have completed. If they have not, it returns and nothing happens. But if it is the last task running, the condition is always fulfilled and the code continues to lines 1124-1126, which run one last GroupAFQ, which would generate the combined tract profile. I worry that this is not super robust, though, and could potentially run into some funky race conditions. In my own test on hyak, I did not get the combined tract profile did not work. So, maybe we need to do something a bit more direct to create the group-level derivatives.

[teresamg]

1120 is checking whether the process is the last process running by trawling for each subject's *_desc-profiles_dwi.csv. If all are present, it exports the GroupAFQ object to create the final tract_profiles.csv.

[36000]

OK that makes sense

[36000]

I wonder if pydra has some way to do this through their API

[teresamg]

Ariel, if all csvs are present, perhaps you can print "output_dirs" and double check that none of the paths are wonky or null? Is there anything unusual in test.out or test.err?

[arokem]

Oh yeah, good call:

slurmstepd: error: *** JOB 18301898 ON g3030 CANCELLED AT 2024-05-14T22:33:02 DUE TO TIME LIMIT ***

I will try this again with a longer max time limit than two hours.

[arokem]

I think it would be tricky to do this through pydra, because we'd have to somehow leave the submitting process running somehow for the duration of all the sub-processes, and that may become very cumbersome for very large datasets. I think that assuming we don't leave the program running until all tasks are completed, we have two options:

1. Something like what is being done here.
2. Don't merge and let users do that separately in a separate program.

I suggest that if my current test works and we get what we expect that we go ahead and merge this PR as is, including this bit of code. I can follow up with a documentation example, based on my current experiments with the HBN data. Users need to know that the final merge into a tract profiles file will fail if any of the sub-tasks fail, so maybe we just need to clarify this in the documentation.

Or is there anything else we need to address before merging?

[36000]

I agree. This looks good to me!

[teresamg]

I'm not sure what you mean about the submitting process... it ends as soon as the jobs are submitted, not completed, and its the last job that does the final merge (all of them check for it). Plus GroupAFQ also fails to produce tract_profiles.csv if anything fails.

[arokem]

I don't know if this is directly related to the content of this PR, but I am now getting an error in visualizing the standard set of bundles, where the visualization code is raising:

INFO:AFQ:Generating colorful lines from tractography...
KeyError: 'Forceps Minor'

Presumably because it's looking for a tract that is now no longer part of the default set of tracts (because we're using the more granular set of CC tracts).

@36000 : any chance this is related to changes you introduced here to how data is passed between GroupAFQ and ParticipantAFQ?

If you think this is unrelated, I think we can probably merge this PR, and fix this issue elsewhere.

[36000]

Not sure what would cause this, but I think it is unrelated, as removing overlapping bundle definitions happens in the init method of the BundleDict, so I don't think a race condition is causing this error.