adds a scoped object to store values under scope-wide names

Dockerfile

@@ -2,7 +2,7 @@ FROM fedora
 ARG VCS_REF=none
 LABEL org.label-schema.vcs-ref=$VCS_REF \
       org.label-schema.vcs-url="https://github.com/diana-hep/yadage"
-RUN dnf install -y gcc gcc-c++ graphviz-devel ImageMagick python-devel libffi-devel openssl openssl-devel unzip nano autoconf automake libtool redhat-rpm-config make; dnf clean all 
+RUN dnf install -y gcc gcc-c++ graphviz-devel ImageMagick python python-devel libffi-devel openssl openssl-devel unzip nano autoconf automake libtool redhat-rpm-config make; dnf clean all 
 COPY . /yadage
 WORKDIR /yadage
 RUN curl https://bootstrap.pypa.io/get-pip.py | python -

tests/conftest.py

@@ -96,6 +96,14 @@ def singlestage_cases(tmpdir,localfs_state_provider):
     wflow = YadageWorkflow.createFromJSON(data,localfs_state_provider)
     return wflow
 
+@pytest.fixture()
+def value_registering_workflow(tmpdir,localfs_state_provider):
+    '''a workflow object with horizontally scalable map stage scheduling sub-workflows'''
+    data  = yadage.workflow_loader.workflow('workflow.yml','tests/testspecs/registering_values')
+    wflow = YadageWorkflow.createFromJSON(data,localfs_state_provider)
+    return wflow
+
+
 @pytest.fixture()
 def multiproc_backend():
     backend = setupbackend_fromstring('multiproc:4')

tests/test_schedulers.py

@@ -43,6 +43,28 @@ def test_singlestep_cases_first(singlestage_cases, foregroundasync_backend):
 
     assert wflow.dag.getNodeByName('hello_world').task.spec == right_taskspec
 
+def test_value_registration(value_registering_workflow, foregroundasync_backend):
+    wflow = value_registering_workflow
+
+    inputdata = {'msg': 'Hello World'}
+    wflow.view().init(inputdata)
+
+    assert wflow.rules[0].applicable(wflow) == False
+    wflow.view().rules[-1].apply(wflow) #apply init
+    assert len(wflow.dag.nodes()) == 1 #init applied
+
+    assert wflow.rules[0].applicable(wflow) == True #first stage is good
+    wflow.rules[0].apply(wflow) #apply first stage
+
+    value =  wflow.view().getValue('ascopedfile')
+    assert value
+    assert type(value) == dict
+    assert value['expression_type'] == 'stage-output-selector'
+    assert value['stages'] == 'hello'
+    assert value['output'] == 'output'
+    assert value['unwrap'] == True
+
+
 def test_singlestep_cases_second(singlestage_cases):
     wflow = singlestage_cases
     inputdata = {'par': 1}

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()
 

tests/testspecs/registering_values/workflow.yml

@@ -0,0 +1,41 @@
+stages:
+- name: hello
+  dependencies: [init]
+  scheduler:
+    scheduler_type: singlestep-stage
+    parameters:
+      msg: {stages: init, output: msg, unwrap: true}
+      outputfile: "{workdir}/hello.txt"
+    register_values:
+      ascopedfile: {
+        expression_type: "stage-output-selector",
+        stages: "hello", output: "output", unwrap: true
+      }
+    step:
+      process:
+        process_type: 'string-interpolated-cmd'
+        cmd: echo {msg} > {outputfile}
+      environment:
+        environment_type: 'localproc-env'
+      publisher:
+        publisher_type: 'interpolated-pub'
+        publish:
+          output: '{outputfile}'
+
+- name: anotherstage
+  dependencies: [hello]
+  scheduler:
+    scheduler_type: singlestep-stage
+    parameters:
+      inputfile: {expression_type: fromvalue, key: ascopedfile}
+      outputfile: "{workdir}/world.txt"
+    step:
+      process:
+        process_type: 'string-interpolated-cmd'
+        cmd: cat {inputfile} | tr '[a-z]' '[A-Z]'  > {outputfile}
+      environment:
+        environment_type: 'localproc-env'
+      publisher:
+        publisher_type: 'interpolated-pub'
+        publish:
+          output: '{outputfile}'

yadage/handlers/expression_handlers.py

@@ -96,3 +96,14 @@ 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.debug('resolved to %s', value)
+    if isinstance(value,dict) and 'expression_type' in value:
+        log.debug('looking up expression %s', value)
+        return handlers[value['expression_type']](view, value)
+    else:
+        log.info('not an expression value')
+        return value

yadage/handlers/scheduler_handlers.py

@@ -109,6 +109,11 @@ def step_or_stages(name, spec, state_provider, inputs, parameters, dependencies)
         return None, None
     raise RuntimeError('do not know what kind of stage spec we are dealing with. %s', spec.keys())
 
+def register_expressions(stage, expressions):
+    if not expressions: return
+    for key, expression in expressions.items():
+        stage.view.addValue(key, expression)
+
 def addStepOrWorkflow(name, stage, parameters, inputs, spec):
     '''
     adds a step or a sub-workflow based on a init step
@@ -163,6 +168,7 @@ def singlestep_stage(stage, spec):
     finalized, inputs = finalize_input(parameters, stage.view)
     finalized = TypedLeafs(finalized, getattr(stage.state_provider,'datamodel',None))
     addStepOrWorkflow(stage.name, stage, finalized, inputs, spec)
+    register_expressions(stage, spec.get('register_values'))
 
 def chunk(alist, chunksize):
     '''split a list into equal-sized chunks of size chunksize'''
@@ -257,6 +263,8 @@ def multistep_stage(stage, spec):
         finalized, inputs = finalize_input(pars, stage.view)
         finalized = TypedLeafs(finalized, getattr(stage.state_provider,'datamodel',None))
         addStepOrWorkflow(singlename, stage, finalized, inputs, spec)
+    register_expressions(stage, spec.get('register_values'))
+
 
 def process_noderef(leafobj,resultscript,view):
     n = view.dag.getNode(leafobj['_nodeid'])
@@ -300,7 +308,7 @@ def jq_stage(stage, spec):
 
         log.info('finalized to: %s',after_post)
         addStepOrWorkflow(singlename, stage, after_post, inputs, spec)
-
+    register_expressions(stage, spec.get('register_values'))
 
 @scheduler('init-stage')
 def init_stage(stage, spec):
@@ -329,3 +337,4 @@ def init_stage(stage, spec):
     task.s(**spec['parameters'])
     task.used_inputs(inputs)
     stage.addStep(task)
+    register_expressions(stage, spec.get('register_values'))

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)