+# ==============================================================================
+class FullOperator(object):
+ """
+ Classe générale d'interface de type opérateur complet
+ (Direct, Linéaire Tangent, Adjoint)
+ """
+ def __init__(self,
+ name = "GenericFullOperator",
+ asMatrix = None,
+ asOneFunction = None, # 1 Fonction
+ asThreeFunctions = None, # 3 Fonctions in a dictionary
+ asScript = None, # 1 or 3 Fonction(s) by script
+ asDict = None, # Parameters
+ appliedInX = None,
+ extraArguments = None,
+ performancePrf = None,
+ inputAsMF = False,# Fonction(s) as Multi-Functions
+ scheduledBy = None,
+ toBeChecked = False,
+ ):
+ ""
+ self.__name = str(name)
+ self.__check = bool(toBeChecked)
+ self.__extraArgs = extraArguments
+ #
+ self.__FO = {}
+ #
+ __Parameters = {}
+ if (asDict is not None) and isinstance(asDict, dict):
+ __Parameters.update( asDict )
+ # Priorité à EnableMultiProcessingInDerivatives=True
+ if "EnableMultiProcessing" in __Parameters and __Parameters["EnableMultiProcessing"]:
+ __Parameters["EnableMultiProcessingInDerivatives"] = True
+ __Parameters["EnableMultiProcessingInEvaluation"] = False
+ if "EnableMultiProcessingInDerivatives" not in __Parameters:
+ __Parameters["EnableMultiProcessingInDerivatives"] = False
+ if __Parameters["EnableMultiProcessingInDerivatives"]:
+ __Parameters["EnableMultiProcessingInEvaluation"] = False
+ if "EnableMultiProcessingInEvaluation" not in __Parameters:
+ __Parameters["EnableMultiProcessingInEvaluation"] = False
+ if "withIncrement" in __Parameters: # Temporaire
+ __Parameters["DifferentialIncrement"] = __Parameters["withIncrement"]
+ # Le défaut est équivalent à "ReducedOverallRequirements"
+ __reduceM, __avoidRC = True, True
+ if performancePrf is not None:
+ if performancePrf == "ReducedAmountOfCalculation":
+ __reduceM, __avoidRC = False, True
+ elif performancePrf == "ReducedMemoryFootprint":
+ __reduceM, __avoidRC = True, False
+ elif performancePrf == "NoSavings":
+ __reduceM, __avoidRC = False, False
+ #
+ if asScript is not None:
+ __Matrix, __Function = None, None
+ if asMatrix:
+ __Matrix = Interfaces.ImportFromScript(asScript).getvalue( self.__name )
+ elif asOneFunction:
+ __Function = { "Direct":Interfaces.ImportFromScript(asScript).getvalue( "DirectOperator" ) }
+ __Function.update({"useApproximatedDerivatives":True})
+ __Function.update(__Parameters)
+ elif asThreeFunctions:
+ __Function = {
+ "Direct" :Interfaces.ImportFromScript(asScript).getvalue( "DirectOperator" ),
+ "Tangent":Interfaces.ImportFromScript(asScript).getvalue( "TangentOperator" ),
+ "Adjoint":Interfaces.ImportFromScript(asScript).getvalue( "AdjointOperator" ),
+ }
+ __Function.update(__Parameters)
+ else:
+ __Matrix = asMatrix
+ if asOneFunction is not None:
+ if isinstance(asOneFunction, dict) and "Direct" in asOneFunction:
+ if asOneFunction["Direct"] is not None:
+ __Function = asOneFunction
+ else:
+ raise ValueError("The function has to be given in a dictionnary which have 1 key (\"Direct\")")
+ else:
+ __Function = { "Direct":asOneFunction }
+ __Function.update({"useApproximatedDerivatives":True})
+ __Function.update(__Parameters)
+ elif asThreeFunctions is not None:
+ if isinstance(asThreeFunctions, dict) and \
+ ("Tangent" in asThreeFunctions) and (asThreeFunctions["Tangent"] is not None) and \
+ ("Adjoint" in asThreeFunctions) and (asThreeFunctions["Adjoint"] is not None) and \
+ (("useApproximatedDerivatives" not in asThreeFunctions) or not bool(asThreeFunctions["useApproximatedDerivatives"])):
+ __Function = asThreeFunctions
+ elif isinstance(asThreeFunctions, dict) and \
+ ("Direct" in asThreeFunctions) and (asThreeFunctions["Direct"] is not None):
+ __Function = asThreeFunctions
+ __Function.update({"useApproximatedDerivatives":True})
+ else:
+ raise ValueError(
+ "The functions has to be given in a dictionnary which have either"+\
+ " 1 key (\"Direct\") or"+\
+ " 3 keys (\"Direct\" (optionnal), \"Tangent\" and \"Adjoint\")")
+ if "Direct" not in asThreeFunctions:
+ __Function["Direct"] = asThreeFunctions["Tangent"]
+ __Function.update(__Parameters)
+ else:
+ __Function = None
+ #
+ if appliedInX is not None and isinstance(appliedInX, dict):
+ __appliedInX = appliedInX
+ elif appliedInX is not None:
+ __appliedInX = {"HXb":appliedInX}
+ else:
+ __appliedInX = None
+ #
+ if scheduledBy is not None:
+ self.__T = scheduledBy
+ #
+ if isinstance(__Function, dict) and \
+ ("useApproximatedDerivatives" in __Function) and bool(__Function["useApproximatedDerivatives"]) and \
+ ("Direct" in __Function) and (__Function["Direct"] is not None):
+ if "CenteredFiniteDifference" not in __Function: __Function["CenteredFiniteDifference"] = False
+ if "DifferentialIncrement" not in __Function: __Function["DifferentialIncrement"] = 0.01
+ if "withdX" not in __Function: __Function["withdX"] = None
+ if "withReducingMemoryUse" not in __Function: __Function["withReducingMemoryUse"] = __reduceM
+ if "withAvoidingRedundancy" not in __Function: __Function["withAvoidingRedundancy"] = __avoidRC
+ if "withToleranceInRedundancy" not in __Function: __Function["withToleranceInRedundancy"] = 1.e-18
+ if "withLengthOfRedundancy" not in __Function: __Function["withLengthOfRedundancy"] = -1
+ if "NumberOfProcesses" not in __Function: __Function["NumberOfProcesses"] = None
+ if "withmfEnabled" not in __Function: __Function["withmfEnabled"] = inputAsMF
+ from daCore import NumericObjects
+ FDA = NumericObjects.FDApproximation(
+ name = self.__name,
+ Function = __Function["Direct"],
+ centeredDF = __Function["CenteredFiniteDifference"],
+ increment = __Function["DifferentialIncrement"],
+ dX = __Function["withdX"],
+ extraArguments = self.__extraArgs,
+ reducingMemoryUse = __Function["withReducingMemoryUse"],
+ avoidingRedundancy = __Function["withAvoidingRedundancy"],
+ toleranceInRedundancy = __Function["withToleranceInRedundancy"],
+ lengthOfRedundancy = __Function["withLengthOfRedundancy"],
+ mpEnabled = __Function["EnableMultiProcessingInDerivatives"],
+ mpWorkers = __Function["NumberOfProcesses"],
+ mfEnabled = __Function["withmfEnabled"],
+ )
+ self.__FO["Direct"] = Operator(
+ name = self.__name,
+ fromMethod = FDA.DirectOperator,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs,
+ enableMultiProcess = __Parameters["EnableMultiProcessingInEvaluation"] )
+ self.__FO["Tangent"] = Operator(
+ name = self.__name+"Tangent",
+ fromMethod = FDA.TangentOperator,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs )
+ self.__FO["Adjoint"] = Operator(
+ name = self.__name+"Adjoint",
+ fromMethod = FDA.AdjointOperator,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs )
+ 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(
+ name = self.__name,
+ fromMethod = __Function["Direct"],
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs,
+ enableMultiProcess = __Parameters["EnableMultiProcessingInEvaluation"] )
+ self.__FO["Tangent"] = Operator(
+ name = self.__name+"Tangent",
+ fromMethod = __Function["Tangent"],
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs )
+ self.__FO["Adjoint"] = Operator(
+ name = self.__name+"Adjoint",
+ fromMethod = __Function["Adjoint"],
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ extraArguments = self.__extraArgs )
+ elif asMatrix is not None:
+ if isinstance(__Matrix, str):
+ __Matrix = PlatformInfo.strmatrix2liststr( __Matrix )
+ __matrice = numpy.asarray( __Matrix, dtype=float )
+ self.__FO["Direct"] = Operator(
+ name = self.__name,
+ fromMatrix = __matrice,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF,
+ enableMultiProcess = __Parameters["EnableMultiProcessingInEvaluation"] )
+ self.__FO["Tangent"] = Operator(
+ name = self.__name+"Tangent",
+ fromMatrix = __matrice,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF )
+ self.__FO["Adjoint"] = Operator(
+ name = self.__name+"Adjoint",
+ fromMatrix = __matrice.T,
+ reducingMemoryUse = __reduceM,
+ avoidingRedundancy = __avoidRC,
+ inputAsMultiFunction = inputAsMF )
+ del __matrice
+ else:
+ raise ValueError(
+ "The %s object is improperly defined or undefined,"%self.__name+\
+ " it requires at minima either a matrix, a Direct operator for"+\
+ " approximate derivatives or a Tangent/Adjoint operators pair."+\
+ " Please check your operator input.")
+ #
+ if __appliedInX is not None:
+ self.__FO["AppliedInX"] = {}
+ for key in __appliedInX:
+ if isinstance(__appliedInX[key], str):
+ __appliedInX[key] = PlatformInfo.strvect2liststr( __appliedInX[key] )
+ self.__FO["AppliedInX"][key] = numpy.ravel( __appliedInX[key] ).reshape((-1,1))
+ else:
+ self.__FO["AppliedInX"] = None
+
+ def getO(self):
+ return self.__FO
+
+ def __repr__(self):
+ "x.__repr__() <==> repr(x)"
+ return repr(self.__FO)
+
+ def __str__(self):
+ "x.__str__() <==> str(x)"
+ return str(self.__FO)
+