add find_required_input_source and SkipOperation

Author: cgc

examples/flywheel_analyzer_engage.py

@@ -60,7 +60,8 @@ def define_analysis(gear_name, acquisition_label, create_inputs):
 
 
 def reactivity_inputs(acquisition_label, acquisitions, session, **kwargs):
-    functional = fa.find(acquisitions, label=acquisition_label)
+    functional = fa.find_required_input_source(acquisitions, label=acquisition_label)
+    # using plain find() here b/c this T1w might be missing
     structural = fa.find(acquisitions, label='T1w 1mm')
     if not structural:
         assert session['_id'] in second_to_first_visit_id,\
@@ -70,6 +71,7 @@ def reactivity_inputs(acquisition_label, acquisitions, session, **kwargs):
         first_visit_acquisitions = client.request(
             'sessions/{}/acquisitions'.format(first_visit_session_id)).json()
         structural = fa.find(first_visit_acquisitions, label='T1w 1mm')
+        assert structural, 'Session {} is missing a structural.'.format(session['_id'])
 
     return dict(
         functional=functional.find_file('*.nii.gz'),
@@ -78,7 +80,7 @@ def reactivity_inputs(acquisition_label, acquisitions, session, **kwargs):
 
 
 def connectivity_inputs(acquisition_label, analyses, **kwargs):
-    reactivity = fa.find(
+    reactivity = fa.find_required_input_source(
         analyses, label=analysis_label('reactivity-preprocessing', acquisition_label))
 
     return dict(
@@ -89,11 +91,11 @@ def connectivity_inputs(acquisition_label, analyses, **kwargs):
 
 
 def first_level_model_inputs(acquisition_label, analyses, acquisitions, **kwargs):
-    reactivity = fa.find(
+    reactivity = fa.find_required_input_source(
         analyses, label=analysis_label('reactivity-preprocessing', acquisition_label))
-    connectivity = fa.find(
+    connectivity = fa.find_required_input_source(
         analyses, label=analysis_label('connectivity-preprocessing', acquisition_label))
-    behavioral = fa.find(
+    behavioral = fa.find_required_input_source(
         acquisitions, label='Behavioral and Physiological')
 
     return dict(

scitran_client/flywheel_analyzer.py

@@ -117,6 +117,16 @@ def find(items, _constructor_=FlywheelFileContainer, **kwargs):
     return result and _constructor_(result)
 
 
+def find_required_input_source(items, **kwargs):
+    '''Finds a match to `kwargs` in `items` by using `find()`. If this match is not
+    found, the current operation will be skipped.
+    '''
+    result = find(items, **kwargs)
+    if not result:
+        raise SkipOperation('could not find match to {}'.format(kwargs))
+    return result
+
+
 def find_project(**kwargs):
     '''Finds a project that matches the key, value pairs in `kwargs`.
 
@@ -130,6 +140,19 @@ class ShuttingDownException(Exception):
     shutting_down = False
 
 
+class SkipOperation(Exception):
+    '''
+    SkipOperation can be thrown from a `create_inputs` function to skip the execution of that
+    operation. This is a way to more dynamically create operation graphs by discarding nodes
+    at runtime.
+
+    For example, if every session has a variable number of functional acquisitions that need to be
+    processed, you can define operations for the max number of per-session functional acquisitions,
+    and throw SkipOperation for all operations corresponding to acquisitions missing for a session.
+    '''
+    pass
+
+
 def request(*args, **kwargs):
     # HACK client is a module variable for now. In the future, we should pass client around.
     assert 'client' in state, 'client must be installed in state before using request. See `installed_client`.'
@@ -208,7 +231,12 @@ def _analyze_session(operations, gears_by_name, session):
             # have completed analysis
             if not acquisitions:
                 acquisitions = request('sessions/{}/acquisitions'.format(session_id))
-            job_inputs = create_inputs(analyses=analyses, acquisitions=acquisitions, session=session)
+            try:
+                job_inputs = create_inputs(analyses=analyses, acquisitions=acquisitions, session=session)
+            except SkipOperation:
+                # we skip to the next operation
+                continue
+
             job_config = _defaults_for_gear(gears_by_name[gear_name])
 
             # When create_inputs returns a tuple, we unpack it into job_inputs and job_config.