🥅 Guardrail BBR only

CPAC/pipeline/nipype_pipeline_engine/__init__.py

@@ -1,25 +1,24 @@
-'''Module to import Nipype Pipeline engine and override some Classes.
-See https://fcp-indi.github.io/docs/developer/nodes
-for C-PAC-specific documentation.
-See https://nipype.readthedocs.io/en/latest/api/generated/nipype.pipeline.engine.html
-for Nipype's documentation.
-
-Copyright (C) 2022  C-PAC Developers
+# Copyright (C) 2022  C-PAC Developers
 
-This file is part of C-PAC.
+# This file is part of C-PAC.
 
-C-PAC is free software: you can redistribute it and/or modify it under
-the terms of the GNU Lesser General Public License as published by the
-Free Software Foundation, either version 3 of the License, or (at your
-option) any later version.
+# C-PAC is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
 
-C-PAC is distributed in the hope that it will be useful, but WITHOUT
-ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
-FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
-License for more details.
+# C-PAC is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
+# License for more details.
 
-You should have received a copy of the GNU Lesser General Public
-License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.'''  # noqa: E501
+# You should have received a copy of the GNU Lesser General Public
+# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
+'''Module to import Nipype Pipeline engine and override some Classes.
+See https://fcp-indi.github.io/docs/developer/nodes
+for C-PAC-specific documentation.
+See https://nipype.readthedocs.io/en/latest/api/generated/nipype.pipeline.engine.html
+for Nipype's documentation.'''  # noqa: E501  # pylint: disable=line-too-long
 from nipype.pipeline import engine as pe
 # import everything in nipype.pipeline.engine.__all__
 from nipype.pipeline.engine import *  # noqa: F401,F403

CPAC/pipeline/nipype_pipeline_engine/engine.py

@@ -1,43 +1,57 @@
-'''Module to import Nipype Pipeline engine and override some Classes.
-See https://fcp-indi.github.io/docs/developer/nodes
-for C-PAC-specific documentation.
-See https://nipype.readthedocs.io/en/latest/api/generated/nipype.pipeline.engine.html
-for Nipype's documentation.
+# STATEMENT OF CHANGES:
+#     This file is derived from sources licensed under the Apache-2.0 terms,
+#     and this file has been changed.
 
-STATEMENT OF CHANGES:
-    This file is derived from sources licensed under the Apache-2.0 terms,
-    and this file has been changed.
+# CHANGES:
+#     * Supports just-in-time dynamic memory allocation
+#     * Skips doctests that require files that we haven't copied over
+#     * Applies a random seed
+#     * Supports overriding memory estimates via a log file and a buffer
+#     * Adds quotation marks around strings in dotfiles
 
-CHANGES:
-    * Supports just-in-time dynamic memory allocation
-    * Skips doctests that require files that we haven't copied over
-    * Applies a random seed
-    * Supports overriding memory estimates via a log file and a buffer
+# ORIGINAL WORK'S ATTRIBUTION NOTICE:
+#     Copyright (c) 2009-2016, Nipype developers
 
-ORIGINAL WORK'S ATTRIBUTION NOTICE:
-    Copyright (c) 2009-2016, Nipype developers
+#     Licensed under the Apache License, Version 2.0 (the "License");
+#     you may not use this file except in compliance with the License.
+#     You may obtain a copy of the License at
 
-    Licensed under the Apache License, Version 2.0 (the "License");
-    you may not use this file except in compliance with the License.
-    You may obtain a copy of the License at
+#         http://www.apache.org/licenses/LICENSE-2.0
 
-        http://www.apache.org/licenses/LICENSE-2.0
+#     Unless required by applicable law or agreed to in writing, software
+#     distributed under the License is distributed on an "AS IS" BASIS,
+#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#     See the License for the specific language governing permissions and
+#     limitations under the License.
 
-    Unless required by applicable law or agreed to in writing, software
-    distributed under the License is distributed on an "AS IS" BASIS,
-    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-    See the License for the specific language governing permissions and
-    limitations under the License.
+#     Prior to release 0.12, Nipype was licensed under a BSD license.
 
-    Prior to release 0.12, Nipype was licensed under a BSD license.
+# Modifications Copyright (C) 2022 C-PAC Developers
 
-Modifications Copyright (C) 2022  C-PAC Developers
+# This file is part of C-PAC.
 
-This file is part of C-PAC.'''  # noqa: E501
+# C-PAC is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+
+# C-PAC is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
+# License for more details.
+
+# You should have received a copy of the GNU Lesser General Public
+# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
+'''Module to import Nipype Pipeline engine and override some Classes.
+See https://fcp-indi.github.io/docs/developer/nodes
+for C-PAC-specific documentation.
+See https://nipype.readthedocs.io/en/latest/api/generated/nipype.pipeline.engine.html
+for Nipype's documentation.'''  # noqa: E501  # pylint: disable=line-too-long
 import os
 import re
-from logging import getLogger
 from inspect import Parameter, Signature, signature
+from logging import getLogger
+from typing import Iterable, Tuple, Union
 from nibabel import load
 from nipype import logging
 from nipype.interfaces.utility import Function
@@ -53,6 +67,7 @@
 UNDEFINED_SIZE = (42, 42, 42, 1200)
 
 random_state_logger = getLogger('random')
+logger = getLogger("nipype.workflow")
 
 
 def _check_mem_x_path(mem_x_path):
@@ -399,10 +414,9 @@ def run(self, updatehash=False):
         if self.seed is not None:
             self._apply_random_seed()
             if self.seed_applied:
-                random_state_logger.info('%s',
-                                         '%s  # (Atropos constant)' %
-                                         self.name if 'atropos' in
-                                         self.name else self.name)
+                random_state_logger.info('%s\t%s', '# (Atropos constant)' if
+                                         'atropos' in self.name else
+                                         str(self.seed), self.name)
         return super().run(updatehash)
 
 
@@ -483,6 +497,40 @@ def _configure_exec_nodes(self, graph):
                                         TypeError):
                                     self._handle_just_in_time_exception(node)
 
+    def connect_retries(self, nodes: Iterable['Node'],
+                        connections: Iterable[Tuple['Node', Union[str, tuple],
+                                                    str]]) -> None:
+        """Method to generalize making the same connections to try and
+        retry nodes.
+
+        For each 3-tuple (``conn``) in ``connections``, will do
+        ``wf.connect(conn[0], conn[1], node, conn[2])`` for each ``node``
+        in ``nodes``
+
+        Parameters
+        ----------
+        nodes : iterable of Nodes
+
+        connections : iterable of 3-tuples of (Node, str or tuple, str)
+        """
+        wrong_conn_type_msg = (r'connect_retries `connections` argument '
+                               'must be an iterable of (Node, str or '
+                               'tuple, str) tuples.')
+        if not isinstance(connections, (list, tuple)):
+            raise TypeError(f'{wrong_conn_type_msg}: Given {connections}')
+        for node in nodes:
+            if not isinstance(node, Node):
+                raise TypeError('connect_retries requires an iterable '
+                                r'of nodes for the `nodes` parameter: '
+                                f'Given {node}')
+            for conn in connections:
+                if not all((isinstance(conn, (list, tuple)), len(conn) == 3,
+                            isinstance(conn[0], Node),
+                            isinstance(conn[1], (tuple, str)),
+                            isinstance(conn[2], str))):
+                    raise TypeError(f'{wrong_conn_type_msg}: Given {conn}')
+                self.connect(*conn[:2], node, conn[2])
+
     def _handle_just_in_time_exception(self, node):
         # pylint: disable=protected-access
         if hasattr(self, '_local_func_scans'):
@@ -492,6 +540,32 @@ def _handle_just_in_time_exception(self, node):
             # TODO: handle S3 files
             node._apply_mem_x(UNDEFINED_SIZE)  # noqa: W0212
 
+    def nodes_and_guardrails(self, *nodes, registered, add_clones=True):
+        """Returns a two tuples of Nodes: (try, retry) and their
+        respective guardrails
+
+        Parameters
+        ----------
+        nodes : any number of Nodes
+
+        Returns
+        -------
+        nodes : tuple of Nodes
+
+        guardrails : tuple of Nodes
+        """
+        from CPAC.registration.guardrails import registration_guardrail_node, \
+                                                 retry_clone
+        nodes = list(nodes)
+        if add_clones is True:
+            nodes.extend([retry_clone(node) for node in nodes])
+        guardrails = [None] * len(nodes)
+        for i, node in enumerate(nodes):
+            guardrails[i] = registration_guardrail_node(
+                f'guardrail_{node.name}', i)
+            self.connect(node, registered, guardrails[i], 'registered')
+        return tuple(nodes), tuple(guardrails)
+
 
 def get_data_size(filepath, mode='xyzt'):
     """Function to return the size of a functional image (x * y * z * t)

CPAC/pipeline/random_state/__init__.py

@@ -1,6 +1,6 @@
 '''Random state for C-PAC'''
-from .seed import random_seed, random_seed_flags, set_up_random_state, \
-                  set_up_random_state_logger
+from .seed import MAX_SEED, random_seed, random_seed_flags, \
+                  set_up_random_state, set_up_random_state_logger
 
-__all__ = ['random_seed', 'random_seed_flags', 'set_up_random_state',
-           'set_up_random_state_logger']
+__all__ = ['MAX_SEED', 'random_seed', 'random_seed_flags',
+           'set_up_random_state', 'set_up_random_state_logger']

CPAC/pipeline/random_state/seed.py

@@ -1,5 +1,20 @@
-'''Functions to set, check, and log random seed'''
-import os
+# Copyright (C) 2022  C-PAC Developers
+
+# This file is part of C-PAC.
+
+# C-PAC is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+
+# C-PAC is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
+# License for more details.
+
+# You should have received a copy of the GNU Lesser General Public
+# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
+"""Functions to set, check, and log random seed"""
 import random
 from logging import getLogger
 
@@ -13,9 +28,26 @@
 from CPAC.utils.interfaces.ants import AI
 from CPAC.utils.monitoring.custom_logging import set_up_logger
 
+MAX_SEED = np.iinfo(np.int32).max
 _seed = {'seed': None}
 
 
+def increment_seed(node):
+    """Increment the random seed for a given node
+
+    Parameters
+    ----------
+    node : Node
+
+    Returns
+    -------
+    node : Node
+    """
+    if isinstance(node.seed, int):
+        node.seed = seed_plus_1()
+    return node
+
+
 def random_random_seed():
     '''Returns a random postive integer up to 2147483647
 
@@ -29,10 +61,10 @@ def random_random_seed():
 
     Examples
     --------
-    >>> 0 < random_random_seed() <= np.iinfo(np.int32).max
+    >>> 0 < random_random_seed() <= MAX_SEED
     True
     '''
-    return random.randint(1, np.iinfo(np.int32).max)
+    return random.randint(1, MAX_SEED)
 
 
 def random_seed():
@@ -46,7 +78,7 @@ def random_seed():
     -------
     seed : int or None
     '''
-    if _seed['seed'] == 'random':
+    if _seed['seed'] in ['random', None]:
         _seed['seed'] = random_random_seed()
     return _seed['seed']
 
@@ -137,6 +169,24 @@ def _reusable_flags():
     }
 
 
+def seed_plus_1(seed=None):
+    '''Increment seed, looping back to 1 at MAX_SEED
+
+    Parameters
+    ----------
+    seed : int, optional
+        Uses configured seed if not specified
+
+    Returns
+    -------
+    int
+    '''
+    seed = random_seed() if seed is None else int(seed)
+    if seed < MAX_SEED:  # increment random seed
+        return seed + 1
+    return 1  # loop back to 1
+
+
 def set_up_random_state(seed):
     '''Set global random seed
 
@@ -160,20 +210,22 @@ def set_up_random_state(seed):
     >>> set_up_random_state(0)
     Traceback (most recent call last):
     ValueError: Valid random seeds are positive integers up to 2147483647, "random", or None, not 0
-    >>> set_up_random_state(None)
-
+    >>> 1 <= set_up_random_state(None) <= MAX_SEED
+    True
     '''  # noqa: E501  # pylint: disable=line-too-long
     if seed is not None:
         if seed == 'random':
             seed = random_random_seed()
         else:
             try:
                 seed = int(seed)
-                assert 0 < seed <= np.iinfo(np.int32).max
-            except(ValueError, TypeError, AssertionError):
-                raise ValueError('Valid random seeds are positive integers up to '
-                                    f'2147483647, "random", or None, not {seed}')
-
+                assert 0 < seed <= MAX_SEED
+            except (ValueError, TypeError, AssertionError) as error:
+                raise ValueError(
+                    'Valid random seeds are positive integers up '
+                    f'to {MAX_SEED}, "random", or None, not {seed}'
+                ) from error
+
     _seed['seed'] = seed
     return random_seed()
 
@@ -185,5 +237,6 @@ def set_up_random_state_logger(log_dir):
     ----------
     log_dir : str
     '''
-    set_up_logger('random', level='info', log_dir=log_dir)
-    getLogger('random').info('seed: %s', random_seed())
+    set_up_logger('random', filename='random.tsv', level='info',
+                  log_dir=log_dir)
+    getLogger('random').info('seed\tnode')