src/daSalome/daYacsIntegration/daOptimizerLoop.py

   1 #-*-coding:iso-8859-1-*-
   2
   3 import SALOMERuntime
   4 import pickle
   5 import numpy
   6 import threading
   7
   8 from daCore.AssimilationStudy import AssimilationStudy
   9
  10 class OptimizerHooks:
  11
  12   def __init__(self, optim_algo):
  13     self.optim_algo = optim_algo
  14
  15     # Gestion du compteur
  16     self.sample_counter = 0
  17     self.counter_lock = threading.Lock()
  18
  19   def Direct(self, X, sync = true):
  20     print "Call Direct OptimizerHooks"
  21     if sync == true:
  22       # 1: Get a unique sample number
  23       self.counter_lock.acquire()
  24       self.sample_counter += 1
  25       local_counter = self.sample_counter
  26
  27       # 2: Put sample in the job pool
  28       matrix_to_pool = pickle.dumps(X)
  29       self.optim_algo.pool.pushInSample(local_counter, matrix_to_pool)
  30
  31       # 3: Wait
  32       while 1:
  33         self.optim_algo.signalMasterAndWait()
  34         if self.optim_algo.isTerminationRequested():
  35           self.optim_algo.pool.destroyAll()
  36           return
  37         else:
  38           # Get current Id
  39           sample_id = self.pool.getCurrentId()
  40           if sample_id == local_counter:
  41             # 4: Data is ready
  42             matrix_from_pool = self.optim_algo.pool.getOutSample(local_counter).getStringValue()
  43
  44             # 5: Release lock
  45             # Have to be done before but need a new implementation
  46             # of the optimizer loop
  47             self.counter_lock.release()
  48
  49             # 6: return results
  50             Y = pickle.loads(matrix_from_pool)
  51             return Y
  52     else:
  53       print "sync false is not yet implemented"
  54       raise ValueError("sync == false not yet implemented")
  55
  56   def Tangent(self, X, sync = true):
  57     print "Call Tangent OptimizerHooks"
  58     if sync == true:
  59       # 1: Get a unique sample number
  60       self.counter_lock.acquire()
  61       self.sample_counter += 1
  62       local_counter = self.sample_counter
  63
  64       # 2: Put sample in the job pool
  65       matrix_to_pool = pickle.dumps(X)
  66       self.optim_algo.pool.pushInSample(local_counter, matrix_to_pool)
  67
  68       # 3: Wait
  69       while 1:
  70         self.optim_algo.signalMasterAndWait()
  71         if self.optim_algo.isTerminationRequested():
  72           self.optim_algo.pool.destroyAll()
  73           return
  74         else:
  75           # Get current Id
  76           sample_id = self.pool.getCurrentId()
  77           if sample_id == local_counter:
  78             # 4: Data is ready
  79             matrix_from_pool = self.optim_algo.pool.getOutSample(local_counter).getStringValue()
  80
  81             # 5: Release lock
  82             # Have to be done before but need a new implementation
  83             # of the optimizer loop
  84             self.counter_lock.release()
  85
  86             # 6: return results
  87             Y = pickle.loads(matrix_from_pool)
  88             return Y
  89     else:
  90       print "sync false is not yet implemented"
  91       raise ValueError("sync == false not yet implemented")
  92
  93   def Adjoint(self, (X, Y), sync = true):
  94     print "Call Adjoint OptimizerHooks"
  95     if sync == true:
  96       # 1: Get a unique sample number
  97       self.counter_lock.acquire()
  98       self.sample_counter += 1
  99       local_counter = self.sample_counter
 100
 101       # 2: Put sample in the job pool
 102       matrix_to_pool = pickle.dumps(Y)
 103       self.optim_algo.pool.pushInSample(local_counter, matrix_to_pool)
 104
 105       # 3: Wait
 106       while 1:
 107         self.optim_algo.signalMasterAndWait()
 108         if self.optim_algo.isTerminationRequested():
 109           self.optim_algo.pool.destroyAll()
 110           return
 111         else:
 112           # Get current Id
 113           sample_id = self.pool.getCurrentId()
 114           if sample_id == local_counter:
 115             # 4: Data is ready
 116             matrix_from_pool = self.optim_algo.pool.getOutSample(local_counter).getStringValue()
 117
 118             # 5: Release lock
 119             # Have to be done before but need a new implementation
 120             # of the optimizer loop
 121             self.counter_lock.release()
 122
 123             # 6: return results
 124             Z = pickle.loads(matrix_from_pool)
 125             return Z
 126     else:
 127       print "sync false is not yet implemented"
 128       raise ValueError("sync == false not yet implemented")
 129
 130 class AssimilationAlgorithm_asynch_3DVAR(SALOMERuntime.OptimizerAlgASync):
 131
 132   def __init__(self):
 133     SALOMERuntime.RuntimeSALOME_setRuntime()
 134     SALOMERuntime.OptimizerAlgASync.__init__(self, None)
 135     self.runtime = SALOMERuntime.getSALOMERuntime()
 136
 137     # Definission des types d'entres et de sorties pour le code de calcul
 138     self.tin  = self.runtime.getTypeCode("pyobj")
 139     self.tout = self.runtime.getTypeCode("pyobj")
 140
 141     self.optim_hooks = OptimizerHooks(self)
 142
 143   # input vient du port algoinit, input est un Any YACS !
 144   def initialize(self,input):
 145     print "Algorithme initialize"
 146
 147   def startToTakeDecision(self):
 148     print "Algorithme startToTakeDecision"
 149
 150     TODO !!
 151
 152     precision = 1.e-13
 153     dimension = 3
 154
 155     xt = numpy.matrix(numpy.arange(dimension)).T
 156     Eo = numpy.matrix(numpy.zeros((dimension,))).T
 157     Eb = numpy.matrix(numpy.zeros((dimension,))).T
 158     H  = numpy.matrix(numpy.core.identity(dimension))
 159     xb = xt + Eb
 160     yo = FunctionH( xt ) + Eo
 161     xb = xb.A1
 162     yo = yo.A1
 163     R  = numpy.matrix(numpy.core.identity(dimension)).T
 164     B  = numpy.matrix(numpy.core.identity(dimension)).T
 165
 166     ADD = AssimilationStudy()
 167     ADD.setBackground         (asVector     = xb )
 168     ADD.setBackgroundError    (asCovariance = B )
 169     ADD.setObservation        (asVector     = yo )
 170     ADD.setObservationError   (asCovariance = R )
 171     ADD.setObservationOperator(asFunction   = {"Tangent":FunctionH,
 172                                                "Adjoint":AdjointH} )
 173     ADD.setControls()
 174     ADD.setAlgorithm(choice="3DVAR")
 175     ADD.analyze()
 176
 177     xa = numpy.array(ADD.get("Analysis").valueserie(0))
 178     d  = numpy.array(ADD.get("Innovation").valueserie(0))
 179     if max(abs(xa - xb)) > precision:
 180         raise ValueError("Résultat du test erroné (1)")
 181     elif max(abs(d)) > precision:
 182         raise ValueError("Résultat du test erroné (2)")
 183     else:
 184         print "    Test correct, erreur maximale inférieure à %s"%precision
 185         print
 186     # On a fini !
 187     self.pool.destroyAll()
 188
 189   # Obligatoire ???
 190   def finish(self):
 191     print "Algorithme finish"
 192   def parseFileToInit(self,fileName):
 193     print "Algorithme parseFileToInit"
 194
 195   # Fonctions qui ne changent pas
 196   def setPool(self,pool):
 197     self.pool=pool
 198   def getTCForIn(self):
 199     return self.tin
 200   def getTCForOut(self):
 201     return self.tout