import logging
import copy
import numpy
+from functools import partial
from daCore import Persistence
from daCore import PlatformInfo
from daCore import Interfaces
NbCallsOfCached = 0
CM = CacheManager()
#
- def __init__(self, fromMethod=None, fromMatrix=None, avoidingRedundancy = True):
+ def __init__(self, fromMethod=None, fromMatrix=None, avoidingRedundancy = True, inputAsMultiFunction = False):
"""
- On construit un objet de ce type en fournissant à l'aide de l'un des
- deux mots-clé, soit une fonction python, soit une matrice.
+ On construit un objet de ce type en fournissant, à l'aide de l'un des
+ deux mots-clé, soit une fonction ou un multi-fonction python, soit une
+ matrice.
Arguments :
- fromMethod : argument de type fonction Python
- fromMatrix : argument adapté au constructeur numpy.matrix
- avoidingRedundancy : évite ou pas les calculs redondants
+ - inputAsMultiFunction : fonction explicitement définie ou pas en multi-fonction
"""
self.__NbCallsAsMatrix, self.__NbCallsAsMethod, self.__NbCallsOfCached = 0, 0, 0
self.__AvoidRC = bool( avoidingRedundancy )
- if fromMethod is not None:
+ self.__inputAsMF = bool( inputAsMultiFunction )
+ if fromMethod is not None and self.__inputAsMF:
self.__Method = fromMethod # logtimer(fromMethod)
self.__Matrix = None
self.__Type = "Method"
+ elif fromMethod is not None and not self.__inputAsMF:
+ self.__Method = partial( MultiFonction, _sFunction=fromMethod)
+ self.__Matrix = None
+ self.__Type = "Method"
elif fromMatrix is not None:
self.__Method = None
self.__Matrix = numpy.matrix( fromMatrix, numpy.float )
"Renvoie le type"
return self.__Type
- def appliedTo(self, xValue, HValue = None):
+ def appliedTo(self, xValue, HValue = None, argsAsSerie = False):
"""
Permet de restituer le résultat de l'application de l'opérateur à un
argument xValue. Cette méthode se contente d'appliquer, son argument
Arguments :
- xValue : argument adapté pour appliquer l'opérateur
"""
- if HValue is not None:
- HxValue = numpy.asmatrix( numpy.ravel( HValue ) ).T
- if self.__AvoidRC:
- Operator.CM.storeValueInX(xValue,HxValue)
+ if argsAsSerie:
+ _xValue = xValue
+ _HValue = HValue
else:
- if self.__AvoidRC:
- __alreadyCalculated, __HxV = Operator.CM.wasCalculatedIn(xValue)
- else:
- __alreadyCalculated = False
- #
- if __alreadyCalculated:
- self.__addOneCacheCall()
- HxValue = __HxV
+ _xValue = (xValue,)
+ if HValue is not None:
+ _HValue = (HValue,)
else:
- if self.__Matrix is not None:
- self.__addOneMatrixCall()
- HxValue = self.__Matrix * xValue
- else:
- self.__addOneMethodCall()
- HxValue = self.__Method( xValue )
+ _HValue = HValue
+ PlatformInfo.isIterable( _xValue, True, " in Operator.appliedTo" )
+ #
+ if _HValue is not None:
+ assert len(_xValue) == len(_HValue), "Incompatible number of elements in xValue and HValue"
+ HxValue = []
+ for i in range(len(_HValue)):
+ HxValue.append( numpy.asmatrix( numpy.ravel( _HValue[i] ) ).T )
if self.__AvoidRC:
- Operator.CM.storeValueInX(xValue,HxValue)
- #
- return HxValue
+ Operator.CM.storeValueInX(_xValue[i],HxValue[-1])
+ else:
+ HxValue = []
+ _xserie = []
+ _hindex = []
+ for i, xv in enumerate(_xValue):
+ if self.__AvoidRC:
+ __alreadyCalculated, __HxV = Operator.CM.wasCalculatedIn(xv)
+ else:
+ __alreadyCalculated = False
+ #
+ if __alreadyCalculated:
+ self.__addOneCacheCall()
+ _hv = __HxV
+ else:
+ if self.__Matrix is not None:
+ self.__addOneMatrixCall()
+ _hv = self.__Matrix * xv
+ else:
+ self.__addOneMethodCall()
+ _xserie.append( xv )
+ _hindex.append( i )
+ _hv = None
+ HxValue.append( _hv )
+ #
+ if len(_xserie)>0 and self.__Matrix is None:
+ _hserie = self.__Method( _xserie ) # Calcul MF
+ if not hasattr(_hserie, "pop"):
+ raise TypeError("The user input multi-function doesn't seem to return sequence results, behaving like a mono-function. It has to be checked.")
+ for i in _hindex:
+ _xv = _xserie.pop(0)
+ _hv = _hserie.pop(0)
+ HxValue[i] = _hv
+ if self.__AvoidRC:
+ Operator.CM.storeValueInX(_xv,_hv)
+ #
+ if argsAsSerie: return HxValue
+ else: return HxValue[-1]
def appliedControledFormTo(self, paire ):
"""
self.__addOneMethodCall()
return self.__Method( xValue )
- def appliedInXTo(self, paire ):
+ def appliedInXTo(self, paires, argsAsSerie = False ):
"""
Permet de restituer le résultat de l'application de l'opérateur à un
argument xValue, sachant que l'opérateur est valable en xNominal.
est construit pour etre ensuite appliqué
- xValue : argument adapté pour appliquer l'opérateur
"""
- assert len(paire) == 2, "Incorrect number of arguments"
- xNominal, xValue = paire
+ if argsAsSerie: _nxValue = paires
+ else: _nxValue = (paires,)
+ PlatformInfo.isIterable( _nxValue, True, " in Operator.appliedInXTo" )
+ #
if self.__Matrix is not None:
- self.__addOneMatrixCall()
- return self.__Matrix * xValue
+ HxValue = []
+ for paire in _nxValue:
+ _xNominal, _xValue = paire
+ self.__addOneMatrixCall()
+ HxValue.append( self.__Matrix * _xValue )
else:
- self.__addOneMethodCall()
- return self.__Method( (xNominal, xValue) )
+ self.__addOneMethodCall( len(_nxValue) )
+ HxValue = self.__Method( _nxValue ) # Calcul MF
+ #
+ if argsAsSerie: return HxValue
+ else: return HxValue[-1]
def asMatrix(self, ValueForMethodForm = "UnknownVoidValue"):
"""
asDict = None, # Parameters
appliedInX = None,
avoidRC = True,
+ inputAsMF = False,# Fonction(s) as Multi-Functions
scheduledBy = None,
toBeChecked = False,
):
if "withLenghtOfRedundancy" not in __Function: __Function["withLenghtOfRedundancy"] = -1
if "withmpEnabled" not in __Function: __Function["withmpEnabled"] = False
if "withmpWorkers" not in __Function: __Function["withmpWorkers"] = None
+ if "withmfEnabled" not in __Function: __Function["withmfEnabled"] = inputAsMF
from daNumerics.ApproximatedDerivatives import FDApproximation
FDA = FDApproximation(
Function = __Function["Direct"],
lenghtOfRedundancy = __Function["withLenghtOfRedundancy"],
mpEnabled = __Function["withmpEnabled"],
mpWorkers = __Function["withmpWorkers"],
+ mfEnabled = __Function["withmfEnabled"],
)
- self.__FO["Direct"] = Operator( fromMethod = FDA.DirectOperator, avoidingRedundancy = avoidRC )
- self.__FO["Tangent"] = Operator( fromMethod = FDA.TangentOperator, avoidingRedundancy = avoidRC )
- self.__FO["Adjoint"] = Operator( fromMethod = FDA.AdjointOperator, avoidingRedundancy = avoidRC )
+ self.__FO["Direct"] = Operator( fromMethod = FDA.DirectOperator, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF)
+ self.__FO["Tangent"] = Operator( fromMethod = FDA.TangentOperator, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
+ self.__FO["Adjoint"] = Operator( fromMethod = FDA.AdjointOperator, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
elif isinstance(__Function, dict) and \
("Direct" in __Function) and ("Tangent" in __Function) and ("Adjoint" in __Function) and \
(__Function["Direct"] is not None) and (__Function["Tangent"] is not None) and (__Function["Adjoint"] is not None):
- self.__FO["Direct"] = Operator( fromMethod = __Function["Direct"], avoidingRedundancy = avoidRC )
- self.__FO["Tangent"] = Operator( fromMethod = __Function["Tangent"], avoidingRedundancy = avoidRC )
- self.__FO["Adjoint"] = Operator( fromMethod = __Function["Adjoint"], avoidingRedundancy = avoidRC )
+ self.__FO["Direct"] = Operator( fromMethod = __Function["Direct"], avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
+ self.__FO["Tangent"] = Operator( fromMethod = __Function["Tangent"], avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
+ self.__FO["Adjoint"] = Operator( fromMethod = __Function["Adjoint"], avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
elif asMatrix is not None:
__matrice = numpy.matrix( __Matrix, numpy.float )
- self.__FO["Direct"] = Operator( fromMatrix = __matrice, avoidingRedundancy = avoidRC )
- self.__FO["Tangent"] = Operator( fromMatrix = __matrice, avoidingRedundancy = avoidRC )
- self.__FO["Adjoint"] = Operator( fromMatrix = __matrice.T, avoidingRedundancy = avoidRC )
+ self.__FO["Direct"] = Operator( fromMatrix = __matrice, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
+ self.__FO["Tangent"] = Operator( fromMatrix = __matrice, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
+ self.__FO["Adjoint"] = Operator( fromMatrix = __matrice.T, avoidingRedundancy = avoidRC, inputAsMultiFunction = inputAsMF )
del __matrice
else:
raise ValueError("Improperly defined observation operator, it requires at minima either a matrix, a Direct for approximate derivatives or a Tangent/Adjoint pair.")
#
if __appliedInX is not None:
self.__FO["AppliedInX"] = {}
- if not isinstance(__appliedInX, dict):
- raise ValueError("Error: observation operator defined by \"AppliedInX\" need a dictionary as argument.")
for key in list(__appliedInX.keys()):
if type( __appliedInX[key] ) is type( numpy.matrix([]) ):
# Pour le cas où l'on a une vraie matrice
def __mul__(self, other):
"x.__mul__(y) <==> x*y"
- if self.ismatrix() and isinstance(other,numpy.matrix):
+ if self.ismatrix() and isinstance(other, (int, numpy.matrix, float)):
return self.__C * other
elif self.ismatrix() and isinstance(other, (list, numpy.ndarray, tuple)):
if numpy.ravel(other).size == self.shape[1]: # Vecteur
def __rmul__(self, other):
"x.__rmul__(y) <==> y*x"
- if self.ismatrix() and isinstance(other,numpy.matrix):
+ if self.ismatrix() and isinstance(other, (int, numpy.matrix, float)):
return other * self.__C
elif self.isvector() and isinstance(other,numpy.matrix):
if numpy.ravel(other).size == self.shape[0]: # Vecteur
elif numpy.asmatrix(other).shape[1] == self.shape[0]: # Matrice
return numpy.asmatrix(numpy.array(other) * self.__C)
else:
- raise ValueError("operands could not be broadcast together with shapes %s %s in %s matrix"%(self.shape,numpy.ravel(other).shape,self.__name))
+ raise ValueError("operands could not be broadcast together with shapes %s %s in %s matrix"%(numpy.ravel(other).shape,self.shape,self.__name))
elif self.isscalar() and isinstance(other,numpy.matrix):
return other * self.__C
elif self.isobject():
raise ValueError("Loading as \"%s\" is not available"%__format)
return __formater.load(__filename, __content, __object)
+# ==============================================================================
+def MultiFonction( __xserie, _sFunction = lambda x: x ):
+ """
+ Pour une liste ordonnée de vecteurs en entrée, renvoie en sortie la liste
+ correspondante de valeurs de la fonction en argument
+ """
+ if not PlatformInfo.isIterable( __xserie ):
+ raise ValueError("MultiFonction not iterable unkown input type: %s"%(type(__xserie),))
+ #
+ __multiHX = []
+ for __xvalue in __xserie:
+ __multiHX.append( _sFunction( __xvalue ) )
+ #
+ return __multiHX
+
# ==============================================================================
def CostFunction3D(_x,
_Hm = None, # Pour simuler Hm(x) : HO["Direct"].appliedTo