Alfoa/hybrid model for batching and ensemble model

ravenframework/JobHandler.py

@@ -22,7 +22,6 @@
 import copy
 import sys
 import threading
-from random import randint
 import socket
 import re
 
@@ -47,6 +46,69 @@
 # FIXME: Finished jobs can bog down the queue waiting for other objects to take
 # them away. Can we shove them onto a different list and free up the job queue?
 
+class IdentifiersFactory(BaseType):
+  """
+    Identifier Factory. This class contains the memory of identifiers used to execute
+    JOBS in the job handler. The identifiers are removed from the Factory once out of
+    scope (i.e. once the job is collected)
+  """
+  def __init__(self):
+    """
+      Constructor
+      @ In, None
+      @ Out, None
+    """
+    self.__IDENTIFIERS_FACTORY = {} # {identifier:uniqueHandler}
+    self.__counter = 0
+
+  def __len__(self):
+    """
+      length (number of identifiers)
+    """
+    return len(self.__IDENTIFIERS_FACTORY)
+
+  def addIdentifier(self, identifier: str, uniqueHandler: str | None) -> None:
+    """
+      Add identifier in factory
+      @ In, identifier, str, new identifier to add
+      @ In, uniqueHandler, str, optional, the `uniqueHandler` if associated with this identifier
+      @ Out, None
+    """
+    if identifier in self.__IDENTIFIERS_FACTORY:
+      self.raiseAnError(RuntimeError, f"Identifier {identifier} is still in use and cannot be re-used yet!")
+
+    self.__IDENTIFIERS_FACTORY[identifier] = uniqueHandler
+    self.__counter += 1
+
+  def removeIdentifier(self, identifier: str) -> None:
+    """
+      Remove identifier in factory
+      @ In, identifier, str, new identifier to add
+      @ Out, None
+    """
+    if identifier not in self.__IDENTIFIERS_FACTORY:
+      self.raiseAnError(RuntimeError, f"Identifier {identifier} is not present in identifier factory. It cannot be removed!")
+
+    self.__IDENTIFIERS_FACTORY.pop(identifier)
+
+  def checkIfIdentifierIsInUse(self, identifier: str) -> bool:
+    """
+      This method is a utility method used to check if an identifier is in use.
+      @ In, identifier, str, the  identifier to check
+      @ Out, checkIfIdentifierIsInUse, bool, is the Identifier in use?
+    """
+    return identifier in list(self.__IDENTIFIERS_FACTORY.keys())
+
+  def clear(self) -> None:
+    """
+      Clear
+      @ In, None
+      @ Out, None
+    """
+    self.__IDENTIFIERS_FACTORY = {}
+
+IDENTIFIERS_COLLECTOR = IdentifiersFactory()
+
 class JobHandler(BaseType):
   """
     JobHandler class. This handles the execution of any job in the RAVEN
@@ -162,7 +224,7 @@ def initialize(self):
     # initialize PBS
     with self.__queueLock:
       self.__running       = [None]*self.runInfoDict['batchSize']
-      self.__clientRunning = [None]*self.runInfoDict['batchSize']
+      self.__clientRunning = [None]*self.runInfoDict['batchSize'] * 2
     self._parallelLib = ParallelLibEnum.shared
     if self.runInfoDict['parallelMethod'] is not None and self.runInfoDict['parallelMethod'] != ParallelLibEnum.distributed:
       self._parallelLib = self.runInfoDict['parallelMethod']
@@ -640,6 +702,7 @@ def addJob(self, args, functionToRun, identifier, metadata=None, forceUseThreads
         clientQueue
       @ Out, None
     """
+    global IDENTIFIERS_COLLECTOR
     assert "original_function" in dir(functionToRun), "to parallelize a function, it must be" \
            " decorated with RAVEN Parallel decorator"
     if self._server is None or forceUseThreads:
@@ -686,6 +749,8 @@ def addJob(self, args, functionToRun, identifier, metadata=None, forceUseThreads
       self.__batching[groupId]["ids"].append(identifier)
     # add the runner in the Queue
     self.reAddJob(internalJob)
+    # update identifier factory
+    IDENTIFIERS_COLLECTOR.addIdentifier(identifier, uniqueHandler)
 
   def reAddJob(self, runner):
     """
@@ -912,6 +977,7 @@ def getFinished(self, removeFinished=True, jobIdentifier='', uniqueHandler="any"
         finished =    [job1, job2, [job3.1, job3.2], job4 ] (job3.1/3.2 belong to the same groupID)
                    or [job1, job2, job3, job4]
     """
+    global IDENTIFIERS_COLLECTOR
     # If the user does not specify a jobIdentifier, then set it to the empty
     # string because every job will match this starting string.
     if jobIdentifier is None:
@@ -955,6 +1021,7 @@ def getFinished(self, removeFinished=True, jobIdentifier='', uniqueHandler="any"
       if removeFinished:
         for i in reversed(runsToBeRemoved):
           self.__finished[i].trackTime('collected')
+          IDENTIFIERS_COLLECTOR.removeIdentifier(self.__finished[i].identifier)
           del self.__finished[i]
 
       # end with self.__queueLock
@@ -1140,8 +1207,10 @@ def shutdown(self):
     @ In, None
     @ Out, None
     """
+    global IDENTIFIERS_COLLECTOR
     self.completed = True
     self.__shutdownParallel()
+    IDENTIFIERS_COLLECTOR.clear()
 
   def terminateAll(self):
     """

ravenframework/Models/EnsembleModel.py

@@ -136,7 +136,7 @@ def localInputAndChecks(self,xmlNode):
           self.raiseAnError(IOError, "Input XML node for Model" + modelName +" has not been inputted!")
         if len(self.modelsInputDictionary[modelName].values()) > allowedEntriesLen:
           self.raiseAnError(IOError, "TargetEvaluation, Input and metadataToTransfer XML blocks are the only XML sub-blocks allowed!")
-        if child.attrib['type'].strip() == "Code":
+        if child.attrib['type'].strip() in ["Code", 'HybridModel', 'LogicalModel']:
           self.createWorkingDir = True
       if child.tag == 'settings':
         self.__readSettings(child)
@@ -246,6 +246,7 @@ def initialize(self,runInfo,inputs,initDict=None):
     for modelClass, modelType, modelName, modelInstance in self.assemblerDict['Model']:
       if not isThereACode:
         isThereACode = modelType == 'Code'
+
       self.modelsDictionary[modelName]['Instance'] = modelInstance
       inputInstancesForModel = []
       for inputName in self.modelsInputDictionary[modelName]['Input']:
@@ -267,6 +268,12 @@ def initialize(self,runInfo,inputs,initDict=None):
 
       # initialize model
       self.modelsDictionary[modelName]['Instance'].initialize(runInfo,inputInstancesForModel,initDict)
+      if modelType in ['HybridModel', 'LogicalModel']:
+        for submodelInst in self.modelsDictionary[modelName]['Instance'].modelInstances.values():
+          if not isThereACode:
+            isThereACode = submodelInst.type == 'Code'
+
+
       # retrieve 'TargetEvaluation' DataObjects
       targetEvaluation = self.retrieveObjectFromAssemblerDict('TargetEvaluation',self.modelsInputDictionary[modelName]['TargetEvaluation'], True)
       # assert acceptable TargetEvaluation types are used

ravenframework/Models/HybridModels/HybridModelBase.py

@@ -161,9 +161,32 @@ def submit(self,myInput,samplerType,jobHandler,**kwargs):
     ## Hybrid models need access to the job handler, so let's stuff it in our
     ## catch all kwargs where evaluateSample can pick it up, not great, but
     ## will suffice until we can better redesign this whole process.
-    prefix = kwargs['prefix']
-    kwargs['jobHandler'] = jobHandler
-    jobHandler.addClientJob((self, myInput, samplerType, kwargs), self.__class__.evaluateSample, prefix, kwargs)
+
+    nRuns = 1
+    batchMode =  kwargs.get("batchMode", False)
+    if batchMode:
+      nRuns = kwargs["batchInfo"]['nRuns']
+
+    for index in range(nRuns):
+      if batchMode:
+        kw =  kwargs['batchInfo']['batchRealizations'][index]
+        kw['batchMode'] = False
+      else:
+        kw = kwargs
+
+      kw['jobHandler'] = jobHandler
+
+      prefix = kw.get("prefix")
+      uniqueHandler = kw.get("uniqueHandler",'any')
+      ## These kw are updated by createNewInput, so the job either should not
+      ## have access to the metadata, or it needs to be updated from within the
+      ## evaluateSample function, which currently is not possible since that
+      ## function does not know about the job instance.
+      metadata = kw
+
+      jobHandler.addClientJob((self, myInput, samplerType, kw), self.__class__.evaluateSample, prefix, metadata=metadata,
+                  uniqueHandler=uniqueHandler,
+                  groupInfo={'id': kwargs['batchInfo']['batchId'], 'size': nRuns} if batchMode else None)
 
   @Parallel()
   def evaluateSample(self, myInput, samplerType, kwargs):
@@ -187,7 +210,9 @@ def evaluateSample(self, myInput, samplerType, kwargs):
     # assure rlz has all metadata
     rlz = dict((var,np.atleast_1d(kwargsToKeep[var])) for var in kwargsToKeep.keys())
     # update rlz with input space from inRun and output space from result
-    rlz.update(dict((var,np.atleast_1d(kwargsToKeep['SampledVars'][var] if var in kwargs['SampledVars'] else result[var])) for var in set(itertools.chain(result.keys(),kwargsToKeep['SampledVars'].keys()))))
+    rlz.update(dict((var,np.atleast_1d(kwargsToKeep['SampledVars'][var]
+                                       if var in kwargs['SampledVars'] else result[var]))
+                    for var in set(itertools.chain(result.keys(),kwargsToKeep['SampledVars'].keys()))))
     return rlz
 
   @abc.abstractmethod

ravenframework/Models/HybridModels/LogicalModel.py

@@ -200,7 +200,7 @@ def _externalRun(self, inRun, jobHandler):
     # TODO: execute control function, move this to createNewInput
     modelToRun = inputKwargs.pop('modelToRun')
     inputKwargs['prefix'] = modelToRun + utils.returnIdSeparator() + identifier
-    inputKwargs['uniqueHandler'] = self.name + identifier
+    inputKwargs['uniqueHandler'] = self.name + utils.returnIdSeparator() +  identifier
 
     moveOn = False
     while not moveOn:

ravenframework/Models/Model.py

@@ -503,3 +503,11 @@ def getSerializationFiles(self):
     serializationFiles = set()
     return serializationFiles
 
+  def constructModelBasedIdentifier(self, jobId):
+    """
+
+      @ In, jobId, str, jobId (the default identifier coming from Samplers/Optimizers)
+      @ Out, new, set, set of filenames that are needed
+    """
+    serializationFiles = set()
+    return serializationFiles

tests/framework/hybridModel/test_logical_ensemble_model_and_code.xml

@@ -0,0 +1,138 @@
+<?xml version="1.0" ?>
+<Simulation verbosity="debug">
+  <TestInfo>
+    <name>abcd</name>
+    <author>abcd</author>
+    <created>2020-05-07</created>
+    <classesTested>Models.LogicalModel</classesTested>
+    <description>
+
+    </description>
+  </TestInfo>
+
+  <RunInfo>
+    <JobName>testLogicalCodeAndExtModelInEnsembleModel</JobName>
+    <Sequence>logicalAndExtModel</Sequence>
+    <WorkingDir>logicalModelCodeAndExtModelInEnsembleModel</WorkingDir>
+    <batchSize>1</batchSize>
+  </RunInfo>
+
+  <Files>
+    <Input name="gen.one" type="">/Users/aalfonsi/projects/raven/tests/framework/hybridModel/logicalCode/gen.one</Input>
+    <Input name="gen.two" type="">/Users/aalfonsi/projects/raven/tests/framework/hybridModel/logicalCode/gen.two</Input>
+  </Files>
+
+  <Functions>
+    <External file="/Users/aalfonsi/projects/raven/tests/framework/hybridModel/logicalCode/control" name="control">
+      <variables>x, y</variables>
+    </External>
+  </Functions>
+
+  <Models>
+    <Code name="poly" subType="GenericCode">
+      <executable>/Users/aalfonsi/projects/raven/tests/framework/hybridModel/logicalCode/poly_code.py</executable>
+      <clargs arg="python" type="prepend"/>
+      <clargs arg="-i" extension=".one" type="input"/>
+      <fileargs arg="aux" extension=".two" type="input"/>
+      <fileargs arg="output" type="output"/>
+    </Code>
+    <Code name="exp" subType="GenericCode">
+      <executable>/Users/aalfonsi/projects/raven/tests/framework/hybridModel/logicalCode/exp_code.py</executable>
+      <clargs arg="python" type="prepend"/>
+      <clargs arg="-i" extension=".one" type="input"/>
+      <fileargs arg="aux" extension=".two" type="input"/>
+      <fileargs arg="output" type="output"/>
+    </Code>
+    <LogicalModel name="logical" subType="">
+      <Model class="Models" type="Code">poly</Model>
+      <Model class="Models" type="Code">exp</Model>
+      <ControlFunction class="Functions" type="External">control</ControlFunction>
+    </LogicalModel>
+
+    <ExternalModel ModuleToLoad="steadyState" name="steadyState" subType="">
+      <inputs>x</inputs>
+      <outputs>y</outputs>
+    </ExternalModel>
+
+    <EnsembleModel name="ExternalModelAndLogical" subType="">
+      <Model class="Models" type="LogicalModel">
+          logical
+        <Input class="Files" type="">gen.one</Input>
+        <Input class="Files" type="">gen.two</Input>
+        <TargetEvaluation class="DataObjects" type="PointSet">samplesLogical</TargetEvaluation>
+      </Model>
+      <Model class="Models" type="ExternalModel">
+          steadyState
+        <Input class="DataObjects" type="PointSet">inputHolder</Input>
+        <TargetEvaluation class="DataObjects" type="PointSet">samplesSteadyState</TargetEvaluation>
+      </Model>
+    </EnsembleModel>
+
+
+  </Models>
+
+  <Distributions>
+    <Uniform name="xd">
+      <lowerBound>0.0</lowerBound>
+      <upperBound>1.0</upperBound>
+    </Uniform>
+  </Distributions>
+
+  <Samplers>
+    <Stratified name="LHS">
+      <variable name="x">
+        <distribution>xd</distribution>
+        <grid construction="equal" steps="15" type="CDF">0.3 0.9</grid>
+      </variable>
+    </Stratified>
+  </Samplers>
+
+  <Steps>
+    <MultiRun name="logicalAndExtModel">
+      <Input class="Files" type="">gen.one</Input>
+      <Input class="Files" type="">gen.two</Input>
+      <Input class="DataObjects" type="PointSet">inputHolder</Input>
+
+      <Model class="Models" type="EnsembleModel">ExternalModelAndLogical</Model>
+      <Sampler class="Samplers" type="Stratified">LHS</Sampler>
+
+      <Output class="DataObjects" type="PointSet">finalResults</Output>
+      <Output class="DataObjects" type="PointSet">samplesLogical</Output>
+      <Output class="OutStreams" type="Print">samplesLogical</Output>
+      <Output class="DataObjects" type="PointSet">samplesSteadyState</Output>
+      <Output class="OutStreams" type="Print">samplesSteadyState</Output>
+    </MultiRun>
+  </Steps>
+
+  <DataObjects>
+    <PointSet name="inputHolder"/>
+
+    <PointSet name="samplesSteadyState">
+      <Input>x</Input>
+      <Output>y</Output>
+    </PointSet>
+    <PointSet name="samplesLogical">
+      <Input>x,y</Input>
+      <Output>poly</Output>
+    </PointSet>
+
+    <PointSet name="finalResults">
+      <Input>x,y</Input>
+      <Output>poly</Output>
+    </PointSet>
+  </DataObjects>
+
+  <OutStreams>
+    <Print name="samplesLogical">
+      <type>csv</type>
+      <source>samplesLogical</source>
+      <what>input, output</what>
+    </Print>
+    <Print name="samplesSteadyState">
+      <type>csv</type>
+      <source>samplesSteadyState</source>
+      <what>input, output</what>
+    </Print>
+  </OutStreams>
+
+</Simulation>