added extra_dimension, added reset_failed ff_function

Author: Michele Berselli

docs/ff_functions.rst

@@ -158,3 +158,18 @@ The function ``reset_all(metawfr_uuid<str>, ff_key<key>, verbose=False)`` allows
     #   -> key generated by ff_utils to authenticate on the portal
 
     reset_metawfr.reset_all(metawfr_uuid, ff_key, verbose=False)
+
+
+The function ``reset_failed(metawfr_uuid<str>, ff_key<key>, verbose=False)`` allows to re-set all workflow-runs in meta-workflow-run with status ``failed``.
+
+.. code-block:: python
+
+    from magma_ff import reset_metawfr
+
+    # UUIDs
+    metawfr_uuid = '' # uuid for meta-workflow-run
+
+    # ff_key
+    #   -> key generated by ff_utils to authenticate on the portal
+
+    reset_metawfr.reset_failed(metawfr_uuid, ff_key, verbose=False)

docs/meta-workflow.rst

@@ -100,6 +100,7 @@ Structure
               #   These are optional fields that can be used to change input argument dimension
               "scatter": 2, # dimension to scatter list argument if any
               "gather": 1, # increment for input argument dimension if previous steps were scattered
+              "extra_dimension": 1, # additional increment for input dimension on top of gather
 
               # These are optional fields
               #   It is possible to skip these fields or add custom ones

magma/inputgenerator.py

@@ -259,9 +259,10 @@ def _match_argument_file(self, arg_obj, run_obj):
                 #end if
             #end for
             gather = getattr(arg_obj, 'gather', 0)
+            gather += getattr(arg_obj, 'extra_dimension', 0) # extra input dimension
             if gather == 1:  # gather 1 dimension
                 arg_obj.files = file_
-            elif gather > 1:  # gather 2 dimensions
+            elif gather > 1:  # gather 2+ dimensions
                 arg_obj.files = file_
                 for i in range(1, gather):
                     arg_obj.files = [arg_obj.files]

magma_ff/reset_metawfr.py

@@ -79,6 +79,8 @@ def reset_shards(metawfr_uuid, shards_name, ff_key, verbose=False):
 ################################################
 def reset_status(metawfr_uuid, status, step_name, ff_key, verbose=False):
     """
+        reset shards with status in status and with name in step_name
+
             metawfr_uuid, uuid for meta-workflow-run
             status, status or list of status to reset
             step_name, name or list of names for step-workflows that need to be reset
@@ -122,6 +124,8 @@ def reset_status(metawfr_uuid, status, step_name, ff_key, verbose=False):
 ################################################
 def reset_all(metawfr_uuid, ff_key, verbose=False):
     """
+        reset all shards
+
             metawfr_uuid, uuid for meta-workflow-run
     """
     # Get meta-workflow-run json from the portal
@@ -147,3 +151,38 @@ def reset_all(metawfr_uuid, ff_key, verbose=False):
         print(res_post)
     #end if
 #end def
+
+################################################
+#   reset_failed
+################################################
+def reset_failed(metawfr_uuid, ff_key, verbose=False):
+    """
+        reset all shards with status failed
+
+            metawfr_uuid, uuid for meta-workflow-run
+    """
+    # Get meta-workflow-run json from the portal
+    run_json = ff_utils.get_metadata(metawfr_uuid, add_on='?frame=raw&datastore=database', key=ff_key)
+    # Create MetaWorkflowRun object for meta-workflow-run
+    run_obj = MetaWorkflowRun(run_json)
+
+    # Get shards to reset
+    to_reset = []
+    for shard_name, obj in run_obj.runs.items():
+        if obj.status == 'failed':
+            to_reset.append(shard_name)
+        #end if
+    #end for
+
+    # Create RunUpdate object for new meta-workflow-run
+    runupd_obj = RunUpdate(run_obj)
+
+    # Reset shards
+    patch_dict = runupd_obj.reset_shards(to_reset)
+
+    # Patch
+    res_post = ff_utils.patch_metadata(patch_dict, metawfr_uuid, key=ff_key)
+    if verbose:
+        print(res_post)
+    #end if
+#end def

test/files/CGAP_WGS_trio.json

@@ -109,6 +109,7 @@
           "argument_name": "input_bams",
           "argument_type": "file",
           "gather": 1,
+          "extra_dimension": 1,
           "mount": true,
           "source": "workflow_add-readgroups-check",
           "source_argument_name": "bam_w_readgroups"

test/test_utils.py

@@ -180,7 +180,7 @@ def test_inputgen_WGS_trio_scatter():
           'env': 'env'
         },
         'input_files': [
-            {'workflow_argument_name': 'input_bams', 'uuid': ['uuid-bam_w_readgroups-2:0', 'uuid-bam_w_readgroups-2:1', 'uuid-bam_w_readgroups-2:2'], 'mount': True}]}
+            {'workflow_argument_name': 'input_bams', 'uuid': [['uuid-bam_w_readgroups-2:0', 'uuid-bam_w_readgroups-2:1', 'uuid-bam_w_readgroups-2:2']], 'mount': True}]}
         ]
     shard = [
         'workflow_bwa-mem_no_unzip-check:0:0',