A suggested (limited) workaround for MapNode of a small workflow

[Michael]

I often myself wanting to use the same workflow (WF_Node1, WF_Node2, ...) for a series of input from a node InputSeries and to collect all resulting outputs :

                                                              InputSeries

 

                WF                                                  WF                                 WF                                 WF

(WF_Node1, WF_Node2, ...)    (WF_Node1, WF_Node2, ...)                        (...)                                 (...)

                                                   Collection of output from each WF

                                                             (Continue pipeline)

The solution is a "MapNode of a workflow" however not implemented yet in NiPype. I still didn't find any elegant solution to this issue and I don't know if it would be useful to anyone but i found a limited workaround which could be used for small Workflow not including any MapNode: defining each Node of WF as a MapNode except for a last WF_MergeNode  

                                                              InputSeries

 

                                                                      WF

                                               (WF_MapNode1, WF_MapNode2, ...)

                                                              WF_MergeNode

                                                             (Continue pipeline)

[Fred Loney]

A succession of MapNodes achieves the purpose, but that is a pain and limits parallelization since each MapNode must execute to completion before aggregating the output and enabling the next MapNode.

I recently implemented a Nipype enhancement that collects outputs of an iterated path using a Join node. See the attached example.

A limitation of the Join node is that it requires an iterated predecessor. Unlike MapNode, the iteration input cannot be set dynamically by a connection. This is a fundamental Nipype constraint that arises from the necessity to fully expand the workflow graph prior to execution. In practice, however, it is easier to set the iterables input once and use plain Nodes rather than selectively propagate cascading MapNodes.

The join enhancement is undergoing testing and will be submitted as a patch in the near future. Please let me know if you have an interest in this solution.

Fred

[Michael]

Many thanks Fred, i will have a look into it and i'm sure it's of interest. 

--
 
---
You received this message because you are subscribed to a topic in the Google Groups "NiPy Users" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/nipy-user/AWIuYf5Ef0g/unsubscribe.
To unsubscribe from this group and all its topics, send an email to nipy-user+...@googlegroups.com.
For more options, visit https://groups.google.com/groups/opt_out.
 
 

[Michael]

Hi Fred,

After looking into it it seems like an excellent solution for most if not all my issues with "MapWorklows". In your example, is it then possible to also join the output of realign as below? And would a second node join2 need to be created for this purpose?

join2 = pe.Node(Join(), joinsource='inputspec', name='join2')

wf.connect(realign, 'out_file', join2, 'in')  

This would mean we could both collect the outputs of iterables at different steps (after img2flt, after realign, ...) but also continue to act on the iterated object separately (such as with strip), is that right? This would really be extremely useful.

How to access the code for the Join Node? When would the patch be available?

To unsubscribe from this group and all its topics, send an email to nipy-user+unsubscribe@googlegroups.com.

[Fred Loney]

Hi Michael,

Yes, it is possible to collect several joins in the workflow. The example modified for your case is shown in the attached file. Note that the syntax is modified so that any interface can be specified as a JoinNode, similar to MapNode. The joinfield parameter is added to specify which field is joined.

The join enhancement works in lock-step with another enhancement, an iterable itersource parameter. Togther, these enhancements provide arbitrary fork and join off of an upstream iterable. These changes touch the heart of Nipype workflow expansion, so are being extensively tested. I plan to post a formal enhancement request and issue a git pull request in two weeks. Whether and when the Nipype developers choose to merge the changes is out of my control. You can access a preliminary version by cloning the git repository at https://github.com/FredLoney/nipype and selecting the join_itersource_synchronize branch as follows:

cd /path/to/my/workspace
git clone https://github.com/FredLoney/nipype
git checkout join_itersource_synchronize

Note that this fork repository is not guaranteed to be stable and is not supported or endorsed by the Nipype development team.

Fred

To unsubscribe from this group and all its topics, send an email to nipy-user+...@googlegroups.com.

[Satrajit Ghosh]

hi fred,

we look forward to seeing the PR.

cheers,

satra

--
 
---
You received this message because you are subscribed to the Google Groups "NiPy Users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to nipy-user+...@googlegroups.com.

[Michael]

I looked at your example and that's just great, this is an extremely useful feature and i can't wait to implement it. I'll hold on for a couple of weeks before modifying the whole pipeline in case important changes happen to the PR during this time however thanks to the git link you provided i can prepare the pipeline upgrade and do some testing with these new features, thank you very much for this!