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,70 @@
 # 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, **kwargs):
+    #"""
+      #Constructor
+      #@ In, None
+      #@ Out, None
+    #"""
+    #super().__init__(**kwargs)
+    #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 +225,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 +703,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 +750,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(internalJob.identifier, uniqueHandler)
 
   def reAddJob(self, runner):
     """
@@ -747,6 +813,8 @@ def addFinishedJob(self, data, metadata=None, uniqueHandler="any", profile=False
       @ In, profile, bool, optional, if True then at de-construction timing statements will be printed
       @ Out, None
     """
+    #global IDENTIFIERS_COLLECTOR
+
     # create a placeholder runner
     run = Runners.factory.returnInstance('PassthroughRunner', data, None,
                                          metadata=metadata,
@@ -755,6 +823,8 @@ def addFinishedJob(self, data, metadata=None, uniqueHandler="any", profile=False
     # place it on the finished queue
     with self.__queueLock:
       self.__finished.append(run)
+      # update identifier factory
+      #IDENTIFIERS_COLLECTOR.addIdentifier(run.identifier, uniqueHandler)
 
   def isFinished(self, uniqueHandler=None):
     """
@@ -912,6 +982,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 +1026,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 +1212,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:

tests/framework/hybridModel/gold/logicalModelCodeAndExtModelInEnsembleModel/finalResults.csv

@@ -0,0 +1,16 @@
+x,y,poly
+0.73162765242,1.73162765242,0.0720237169458
+0.492704456326,1.49270445633,1.29349617899
+0.555294141261,1.55529414126,0.197763300797
+0.765845655665,1.76584565567,0.0548282569708
+0.503382503605,1.50338250361,0.246628937725
+0.587434256432,1.58743425643,0.170210492765
+0.639866882027,1.63986688203,0.129695862661
+0.823657390038,1.82365739004,0.0310967160881
+0.458268852019,1.45826885202,1.34107648262
+0.419415902449,1.41941590245,1.40084817762
+0.379586681845,1.37958668184,1.46948600818
+0.819461917952,1.81946191795,0.0325939990697
+0.324619247789,1.32461924779,1.5779699205
+0.690076483411,1.69007648341,0.0960525861347
+0.875131506812,1.87513150681,0.0155921405911

tests/framework/hybridModel/logicalModelCodeAndExtModelInEnsembleModel/gen.one

@@ -0,0 +1,9 @@
+#############################################
+#                                           #
+#  Dummy Input File for Poly Python Module  #
+#                                           #
+#############################################
+
+case = $RAVEN-output$
+auxfile = $RAVEN-aux$
+x = $RAVEN-x$

tests/framework/hybridModel/logicalModelCodeAndExtModelInEnsembleModel/gen.two

@@ -0,0 +1,7 @@
+#############################################
+#                                           #
+#  Dummy Input File for Poly Python Module  #
+#                                           #
+#############################################
+
+y = $RAVEN-y$