Merge 630d0fb into 347ccad

tests/test_wflow.py

@@ -17,7 +17,7 @@ def test_serialize():
     json.dumps(data)
 
 def test_deserialize():
-    data = {'rules': [], 'applied': [], 'dag': {'nodes': [], 'edges': []}, 'stepsbystage': {}, 'bookkeeping': {}}
+    data = {'rules': [], 'applied': [], 'dag': {'nodes': [], 'edges': []}, 'stepsbystage': {}, 'bookkeeping': {}, 'values': {}}
     wflow = YadageWorkflow.fromJSON(data)
     assert data == wflow.json()
 

yadage/handlers/expression_handlers.py

@@ -96,3 +96,12 @@ def stage_output_selector(stageview, selection):
                                  selection.get('unwrap', False))
         log.debug('selected outputs %s', outputs)
         return outputs
+
+@expression('fromvalue')
+def value_resolver(view, expression):
+    value = view.getValue(expression['key'])
+    log.info('resolved to %s', value)
+    if isinstance(value,dict) and 'expression_type' in value:
+        return handlers[value['expression_type']](view, value)
+    else:
+        return value

yadage/wflow.py

@@ -18,6 +18,7 @@ def __init__(self):
         super(YadageWorkflow, self).__init__()
         self.stepsbystage = {}
         self.bookkeeping = {}
+        self.values = {}
 
     def view(self, offset=''):
         return WorkflowView(self, offset)
@@ -30,6 +31,7 @@ def json(self):
                            )
         data['bookkeeping'] = self.bookkeeping
         data['stepsbystage'] = self.stepsbystage
+        data['values'] = self.values
         return data
 
     @classmethod
@@ -43,6 +45,7 @@ def fromJSON(cls, data,
         instance.applied_rules = [OffsetStage.fromJSON(x,state_provider_deserializer) for x in data['applied'] ]
         instance.bookkeeping = data['bookkeeping']
         instance.stepsbystage = data['stepsbystage']
+        instance.values = data['values']
 
         instance.dag = adage.serialize.dag_from_json(
             data['dag'],

yadage/wflowview.py

@@ -27,6 +27,7 @@ def __init__(self, workflowobj, offset=''):
         self.offset = offset
         self.steps = JsonPointer(self.offset).resolve(workflowobj.stepsbystage)
         self.bookkeeper = JsonPointer(self.offset).resolve(workflowobj.bookkeeping)
+        self.values = JsonPointer(self.offset).resolve(workflowobj.values)
 
     def view(self, offset):
         '''
@@ -108,6 +109,15 @@ def addRule(self, rule, offset=''):
         thisoffset.resolve(self.bookkeeper)['_meta']['stages'] += [offsetstage.identifier]
         return offsetstage.identifier
 
+    def addValue(self, key, value):
+        v = self.values.setdefault('_values', {})
+        if key in v:
+            raise RuntimeError('cannot overwrite value')
+        v[key] = value
+
+    def getValue(self, key):
+        return self.values['_values'][key]
+
     def addStep(self, task, stage, depends_on=None):
         '''
         adds a node to the DAG connecting it to the passed depending nodes
@@ -146,6 +156,10 @@ def addWorkflow(self, rules, stage=None):
             offset = JsonPointer.from_parts([stage, len(self.steps[stage]) - 1]).path
             self.steps[stage][-1]['_offset'] = offset
 
+            self.values.setdefault(stage,[]).append({})
+            offset = JsonPointer.from_parts([stage, len(self.values[stage]) - 1]).path
+            self.values[stage][-1] = {}
+
         for rule in rules:
             self.addRule(rule, offset)