# -*- coding: utf-8 -*-
#
-# Copyright (C) 2008-2017 EDF R&D
+# Copyright (C) 2008-2024 EDF R&D
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# Author: Jean-Philippe Argaud, jean-philippe.argaud@edf.fr, EDF R&D
"""
- Normalized interface for ADAO scripting (generic API)
+ Normalized interface for ADAO scripting (generic API).
"""
__author__ = "Jean-Philippe ARGAUD"
__all__ = ["Aidsm"]
-import os, sys
-from daCore import ExtendedLogging ; ExtendedLogging.ExtendedLogging() # A importer en premier
-import logging
-import numpy
+import os
+import sys
+import inspect
#
from daCore.BasicObjects import State, Covariance, FullOperator, Operator
from daCore.BasicObjects import AlgorithmAndParameters, DataObserver
-from daCore.BasicObjects import DiagnosticAndParameters, ImportFromScript
-from daCore.BasicObjects import CaseLogger, GenericCaseViewer
-from daCore.Templates import ObserverTemplates
+from daCore.BasicObjects import RegulationAndParameters, CaseLogger
+from daCore.BasicObjects import UserScript, ExternalParameters
from daCore import PlatformInfo
+from daCore import version
-# ==============================================================================
-class DICViewer(GenericCaseViewer):
- """
- Etablissement des commandes de creation d'un cas DIC
- """
- def __init__(self, __name="", __objname="case", __content=None):
- "Initialisation et enregistrement de l'entete"
- GenericCaseViewer.__init__(self, __name, __objname, __content)
- self._addLine("# -*- coding: utf-8 -*-")
- self._addLine("#\n# Input for ADAO converter to YACS\n#")
- self._addLine("from numpy import array, matrix")
- self._addLine("#")
- self._addLine("study_config = {}")
- self._addLine("study_config['StudyType'] = 'ASSIMILATION_STUDY'")
- self._addLine("study_config['Name'] = '%s'"%self._name)
- self._addLine("observers = {}")
- self._addLine("study_config['Observers'] = observers")
- self._addLine("#")
- self._addLine("inputvariables_config = {}")
- self._addLine("inputvariables_config['Order'] =['adao_default']")
- self._addLine("inputvariables_config['adao_default'] = -1")
- self._addLine("study_config['InputVariables'] = inputvariables_config")
- self._addLine("#")
- self._addLine("outputvariables_config = {}")
- self._addLine("outputvariables_config['Order'] = ['adao_default']")
- self._addLine("outputvariables_config['adao_default'] = -1")
- self._addLine("study_config['OutputVariables'] = outputvariables_config")
- if __content is not None:
- for command in __content:
- self._append(*command)
- def _append(self, __command=None, __keys=None, __local=None, __pre=None, __switchoff=False):
- "Transformation d'une commande individuelle en un enregistrement"
- if __command == "set": __command = __local["Concept"]
- else: __command = __command.replace("set", "", 1)
- #
- __text = None
- if __command in (None, 'execute', 'executePythonScheme', 'executeYACSScheme', 'get'):
- return
- elif __command in ['Debug', 'setDebug']:
- __text = "#\nstudy_config['Debug'] = '1'"
- elif __command in ['NoDebug', 'setNoDebug']:
- __text = "#\nstudy_config['Debug'] = '0'"
- elif __command in ['Observer', 'setObserver']:
- __obs = __local['Variable']
- self._numobservers += 1
- __text = "#\n"
- __text += "observers['%s'] = {}\n"%__obs
- if __local['String'] is not None:
- __text += "observers['%s']['nodetype'] = '%s'\n"%(__obs, 'String')
- __text += "observers['%s']['String'] = \"\"\"%s\"\"\"\n"%(__obs, __local['String'])
- if __local['Script'] is not None:
- __text += "observers['%s']['nodetype'] = '%s'\n"%(__obs, 'Script')
- __text += "observers['%s']['Script'] = \"%s\"\n"%(__obs, __local['Script'])
- if __local['Template'] is not None:
- __text += "observers['%s']['nodetype'] = '%s'\n"%(__obs, 'String')
- __text += "observers['%s']['String'] = \"\"\"%s\"\"\"\n"%(__obs, ObserverTemplates[__local['Template']])
- if __local['Info'] is not None:
- __text += "observers['%s']['info'] = \"\"\"%s\"\"\"\n"%(__obs, __local['Info'])
- else:
- __text += "observers['%s']['info'] = \"\"\"%s\"\"\"\n"%(__obs, __obs)
- __text += "observers['%s']['number'] = %s"%(__obs, self._numobservers)
- elif __local is not None: # __keys is not None and
- numpy.set_printoptions(precision=15,threshold=1000000,linewidth=1000*15)
- __text = "#\n"
- __text += "%s_config = {}\n"%__command
- if 'self' in __local: __local.pop('self')
- __to_be_removed = []
- for __k,__v in __local.items():
- if __v is None: __to_be_removed.append(__k)
- for __k in __to_be_removed:
- __local.pop(__k)
- for __k,__v in __local.items():
- if __k == "Concept": continue
- if __k in ['ScalarSparseMatrix','DiagonalSparseMatrix','Matrix'] and 'Script' in __local: continue
- if __k == 'Algorithm':
- __text += "study_config['Algorithm'] = %s\n"%(repr(__v))
- elif __k == 'Script':
- __k = 'Vector'
- for __lk in ['ScalarSparseMatrix','DiagonalSparseMatrix','Matrix']:
- if __lk in __local and __local[__lk]: __k = __lk
- if __command == "AlgorithmParameters": __k = "Dict"
- __text += "%s_config['Type'] = '%s'\n"%(__command,__k)
- __text += "%s_config['From'] = '%s'\n"%(__command,'Script')
- __text += "%s_config['Data'] = '%s'\n"%(__command,repr(__v))
- __text = __text.replace("''","'")
- elif __k in ('Stored', 'Checked'):
- if bool(__v):
- __text += "%s_config['%s'] = '%s'\n"%(__command,__k,int(bool(__v)))
- elif __k in ('AvoidRC', 'noDetails'):
- if not bool(__v):
- __text += "%s_config['%s'] = '%s'\n"%(__command,__k,int(bool(__v)))
- else:
- if __k is 'Parameters': __k = "Dict"
- __text += "%s_config['Type'] = '%s'\n"%(__command,__k)
- __text += "%s_config['From'] = '%s'\n"%(__command,'String')
- __text += "%s_config['Data'] = \"\"\"%s\"\"\"\n"%(__command,repr(__v))
- __text += "study_config['%s'] = %s_config"%(__command,__command)
- numpy.set_printoptions(precision=8,threshold=1000,linewidth=75)
- if __switchoff:
- self._switchoff = True
- if __text is not None: self._addLine(__text)
- if not __switchoff:
- self._switchoff = False
- def _finalize(self):
- self.__loadVariablesByScript()
- self._addLine("#")
- self._addLine("Analysis_config = {}")
- self._addLine("Analysis_config['From'] = 'String'")
- self._addLine("Analysis_config['Data'] = \"\"\"import numpy")
- self._addLine("xa=numpy.ravel(ADD.get('Analysis')[-1])")
- self._addLine("print 'Analysis:',xa\"\"\"")
- self._addLine("study_config['UserPostAnalysis'] = Analysis_config")
- def __loadVariablesByScript(self):
- exec("\n".join(self._lineSerie))
- if "Algorithm" in study_config and len(study_config['Algorithm'])>0:
- self.__hasAlgorithm = True
- else:
- self.__hasAlgorithm = False
- if not self.__hasAlgorithm and \
- "AlgorithmParameters" in study_config and \
- isinstance(study_config['AlgorithmParameters'], dict) and \
- "From" in study_config['AlgorithmParameters'] and \
- "Data" in study_config['AlgorithmParameters'] and \
- study_config['AlgorithmParameters']['From'] == 'Script':
- __asScript = study_config['AlgorithmParameters']['Data']
- __var = ImportFromScript(__asScript).getvalue( "Algorithm" )
- __text = "#\nstudy_config['Algorithm'] = '%s'"%(__var,)
- self._addLine(__text)
- if self.__hasAlgorithm and \
- "AlgorithmParameters" in study_config and \
- isinstance(study_config['AlgorithmParameters'], dict) and \
- "From" not in study_config['AlgorithmParameters'] and \
- "Data" not in study_config['AlgorithmParameters']:
- __text = "#\n"
- __text += "AlgorithmParameters_config['Type'] = 'Dict'\n"
- __text += "AlgorithmParameters_config['From'] = 'String'\n"
- __text += "AlgorithmParameters_config['Data'] = '{}'\n"
- self._addLine(__text)
- del study_config
-
-class XMLViewer(GenericCaseViewer):
- """
- Etablissement des commandes de creation d'un cas DIC
- """
- def __init__(self, __name="", __objname="case", __content=None):
- "Initialisation et enregistrement de l'entete"
- GenericCaseViewer.__init__(self, __name, __objname, __content)
- raise NotImplementedError()
+from daCore import ExtendedLogging
+ExtendedLogging.ExtendedLogging() # A importer en premier
+import logging # noqa: E402
# ==============================================================================
class Aidsm(object):
""" ADAO Internal Data Structure Model """
- def __init__(self, name = "", viewers={"DIC":DICViewer, "XML":XMLViewer}):
- self.__name = str(name)
- self.__case = CaseLogger(self.__name, "case", viewers)
+ __slots__ = (
+ "__name", "__objname", "__directory", "__case", "__parent",
+ "__adaoObject", "__StoredInputs", "__PostAnalysis", "__Concepts",
+ )
+
+ def __init__(self, name="", addViewers=None):
+ self.__name = str(name)
+ self.__objname = None
+ self.__directory = None
+ self.__case = CaseLogger(self.__name, "case", addViewers)
#
self.__adaoObject = {}
self.__StoredInputs = {}
+ self.__PostAnalysis = []
#
- self.__Concepts = [
+ self.__Concepts = [ # Liste exhaustive
"AlgorithmParameters",
"Background",
+ "BackgroundError",
"CheckingPoint",
"ControlInput",
- "Observation",
- "Controls",
- "BackgroundError",
- "ObservationError",
- "EvolutionError",
- "ObservationOperator",
- "EvolutionModel",
"ControlModel",
"Debug",
+ "Directory",
+ "EvolutionError",
+ "EvolutionModel",
+ "Name",
"NoDebug",
+ "Observation",
+ "ObservationError",
+ "ObservationOperator",
"Observer",
- ]
+ "RegulationParameters",
+ "SupplementaryParameters",
+ "UserPostAnalysis",
+ ]
#
for ename in self.__Concepts:
self.__adaoObject[ename] = None
- for ename in ("ObservationOperator", "EvolutionModel", "ControlModel", "Observer"):
+ for ename in ("ObservationOperator", "EvolutionModel", "ControlModel"):
self.__adaoObject[ename] = {}
- for ename in ("Diagnostic", "Observer"):
+ for ename in ("BackgroundError", "ObservationError"):
+ self.__adaoObject[ename] = Covariance(ename, asEyeByScalar = 1.)
+ for ename in ("EvolutionError",):
+ self.__adaoObject[ename] = Covariance(ename, asEyeByScalar = 1.e-16)
+ for ename in ("Observer", "UserPostAnalysis"):
self.__adaoObject[ename] = []
- self.__StoredInputs[ename] = []
+ self.__StoredInputs[ename] = [] # Vide par defaut
+ self.__StoredInputs["Name"] = self.__name
+ self.__StoredInputs["Directory"] = self.__directory
#
# Récupère le chemin du répertoire parent et l'ajoute au path
# (Cela complète l'action de la classe PathManagement dans PlatformInfo,
# qui est activée dans Persistence)
- self.__parent = os.path.abspath(os.path.join(os.path.dirname(__file__),".."))
+ self.__parent = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, self.__parent)
- sys.path = PlatformInfo.uniq( sys.path ) # Conserve en unique exemplaire chaque chemin
+ sys.path = PlatformInfo.uniq( sys.path ) # Conserve en unique exemplaire chaque chemin
def set(self,
- Concept = None, # Premier argument
+ Concept = None, # Premier argument
Algorithm = None,
AppliedInXb = None,
- AvoidRC = True,
- BaseType = None,
Checked = False,
- Diagnostic = None,
+ ColMajor = False,
+ ColNames = None,
+ CrossObs = False,
+ DataFile = None,
DiagonalSparseMatrix = None,
- Identifier = None,
+ ExtraArguments = None,
Info = None,
+ InputFunctionAsMulti = False,
Matrix = None,
ObjectFunction = None,
ObjectMatrix = None,
OneFunction = None,
Parameters = None,
+ PerformanceProfile = None,
ScalarSparseMatrix = None,
Scheduler = None,
Script = None,
Stored = False,
String = None,
+ SyncObs = True,
Template = None,
ThreeFunctions = None,
- Unit = None,
Variable = None,
Vector = None,
VectorSerie = None,
):
- "Interface unique de definition de variables d'entrees par argument"
- self.__case.register("set",dir(),locals(),None,True)
+ "Interface unique de définition de variables d'entrées par argument"
+ self.__case.register("set", dir(), locals(), None, True)
try:
- if Concept in ("Background", "CheckingPoint", "ControlInput", "Observation", "Controls"):
- commande = getattr(self,"set"+Concept)
- commande(Vector, VectorSerie, Script, Scheduler, Stored, Checked )
+ if Concept in ("Background", "CheckingPoint", "ControlInput", "Observation"):
+ commande = getattr(self, "set" + Concept)
+ commande(Vector, VectorSerie, Script, DataFile, ColNames, ColMajor, Stored, Scheduler, Checked )
elif Concept in ("BackgroundError", "ObservationError", "EvolutionError"):
- commande = getattr(self,"set"+Concept)
+ commande = getattr(self, "set" + Concept)
commande(Matrix, ScalarSparseMatrix, DiagonalSparseMatrix,
- ObjectMatrix, Script, Stored, Checked )
+ Script, Stored, ObjectMatrix, Checked )
elif Concept == "AlgorithmParameters":
self.setAlgorithmParameters( Algorithm, Parameters, Script )
+ elif Concept == "RegulationParameters":
+ self.setRegulationParameters( Algorithm, Parameters, Script )
+ elif Concept == "Name":
+ self.setName(String)
+ elif Concept == "Directory":
+ self.setDirectory(String)
elif Concept == "Debug":
self.setDebug()
elif Concept == "NoDebug":
self.setNoDebug()
elif Concept == "Observer":
- self.setObserver( Variable, Template, String, Script, ObjectFunction, Scheduler, Info )
- elif Concept == "Diagnostic":
- self.setDiagnostic( Diagnostic, Identifier, Parameters, Script, Unit, BaseType )
+ self.setObserver(
+ Variable, Template, String, Script, Info,
+ ObjectFunction, CrossObs, SyncObs, Scheduler )
+ elif Concept == "UserPostAnalysis":
+ self.setUserPostAnalysis( Template, String, Script )
+ elif Concept == "SupplementaryParameters":
+ self.setSupplementaryParameters( Parameters, Script )
elif Concept == "ObservationOperator":
self.setObservationOperator(
Matrix, OneFunction, ThreeFunctions, AppliedInXb,
- Parameters, Script, AvoidRC, Stored, Checked )
+ Parameters, Script, ExtraArguments,
+ Stored, PerformanceProfile, InputFunctionAsMulti, Checked )
elif Concept in ("EvolutionModel", "ControlModel"):
- commande = getattr(self,"set"+Concept)
+ commande = getattr(self, "set" + Concept)
commande(
Matrix, OneFunction, ThreeFunctions,
- Parameters, Script, Scheduler, AvoidRC, Stored, Checked )
-
+ Parameters, Script, Scheduler, ExtraArguments,
+ Stored, PerformanceProfile, InputFunctionAsMulti, Checked )
else:
raise ValueError("the variable named '%s' is not allowed."%str(Concept))
except Exception as e:
- if isinstance(e, SyntaxError): msg = "at %s: %s"%(e.offset, e.text)
- else: msg = ""
- raise ValueError("during settings, the following error occurs:\n\n%s %s\n\nSee also the potential messages, which can show the origin of the above error, in the launching terminal."%(str(e),msg))
+ if isinstance(e, SyntaxError):
+ msg = " at %s: %s"%(e.offset, e.text)
+ else:
+ msg = ""
+ raise ValueError((
+ "during settings, the following error occurs:\n" + \
+ "\n%s%s\n\nSee also the potential messages, " + \
+ "which can show the origin of the above error, " + \
+ "in the launching terminal.")%(str(e), msg))
# -----------------------------------------------------------
- def setBackground(self,
+ def setBackground(
+ self,
Vector = None,
VectorSerie = None,
Script = None,
- Scheduler = None,
+ DataFile = None,
+ ColNames = None,
+ ColMajor = False,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ Scheduler = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "Background"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = State(
name = Concept,
asVector = Vector,
asPersistentVector = VectorSerie,
- asScript = Script,
+ asScript = self.__with_directory(Script),
+ asDataFile = DataFile,
+ colNames = ColNames,
+ colMajor = ColMajor,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setCheckingPoint(self,
+ def setCheckingPoint(
+ self,
Vector = None,
VectorSerie = None,
Script = None,
- Scheduler = None,
+ DataFile = None,
+ ColNames = None,
+ ColMajor = False,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ Scheduler = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "CheckingPoint"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = State(
name = Concept,
asVector = Vector,
asPersistentVector = VectorSerie,
- asScript = Script,
+ asScript = self.__with_directory(Script),
+ asDataFile = DataFile,
+ colNames = ColNames,
+ colMajor = ColMajor,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setControlInput(self,
+ def setControlInput(
+ self,
Vector = None,
VectorSerie = None,
Script = None,
- Scheduler = None,
+ DataFile = None,
+ ColNames = None,
+ ColMajor = False,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ Scheduler = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "ControlInput"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = State(
name = Concept,
asVector = Vector,
asPersistentVector = VectorSerie,
- asScript = Script,
+ asScript = self.__with_directory(Script),
+ asDataFile = DataFile,
+ colNames = ColNames,
+ colMajor = ColMajor,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setObservation(self,
+ def setObservation(
+ self,
Vector = None,
VectorSerie = None,
Script = None,
- Scheduler = None,
+ DataFile = None,
+ ColNames = None,
+ ColMajor = False,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
- Concept = "Observation"
- self.__case.register("set"+Concept, dir(), locals())
- self.__adaoObject[Concept] = State(
- name = Concept,
- asVector = Vector,
- asPersistentVector = VectorSerie,
- asScript = Script,
- scheduledBy = Scheduler,
- toBeChecked = Checked,
- )
- if Stored:
- self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
- return 0
-
- def setControls(self,
- Vector = (), # Valeur par defaut pour un vecteur vide
- VectorSerie = None,
- Script = None,
Scheduler = None,
- Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
- Concept = "Controls"
- self.__case.register("set"+Concept, dir(), locals())
+ Checked = False ):
+ "Définition d'un concept de calcul"
+ Concept = "Observation"
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = State(
name = Concept,
asVector = Vector,
asPersistentVector = VectorSerie,
- asScript = Script,
+ asScript = self.__with_directory(Script),
+ asDataFile = DataFile,
+ colNames = ColNames,
+ colMajor = ColMajor,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setBackgroundError(self,
+ def setBackgroundError(
+ self,
Matrix = None,
ScalarSparseMatrix = None,
DiagonalSparseMatrix = None,
- ObjectMatrix = None,
Script = None,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ ObjectMatrix = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "BackgroundError"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = Covariance(
name = Concept,
asCovariance = Matrix,
asEyeByScalar = ScalarSparseMatrix,
asEyeByVector = DiagonalSparseMatrix,
asCovObject = ObjectMatrix,
- asScript = Script,
+ asScript = self.__with_directory(Script),
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setObservationError(self,
+ def setObservationError(
+ self,
Matrix = None,
ScalarSparseMatrix = None,
DiagonalSparseMatrix = None,
- ObjectMatrix = None,
Script = None,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ ObjectMatrix = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "ObservationError"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = Covariance(
name = Concept,
asCovariance = Matrix,
asEyeByScalar = ScalarSparseMatrix,
asEyeByVector = DiagonalSparseMatrix,
asCovObject = ObjectMatrix,
- asScript = Script,
+ asScript = self.__with_directory(Script),
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setEvolutionError(self,
+ def setEvolutionError(
+ self,
Matrix = None,
ScalarSparseMatrix = None,
DiagonalSparseMatrix = None,
- ObjectMatrix = None,
Script = None,
Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ ObjectMatrix = None,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "EvolutionError"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = Covariance(
name = Concept,
asCovariance = Matrix,
asEyeByScalar = ScalarSparseMatrix,
asEyeByVector = DiagonalSparseMatrix,
asCovObject = ObjectMatrix,
- asScript = Script,
+ asScript = self.__with_directory(Script),
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setObservationOperator(self,
- Matrix = None,
- OneFunction = None,
- ThreeFunctions = None,
- AppliedInXb = None,
- Parameters = None,
- Script = None,
- AvoidRC = True,
- Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ def setObservationOperator(
+ self,
+ Matrix = None,
+ OneFunction = None,
+ ThreeFunctions = None,
+ AppliedInXb = None,
+ Parameters = None,
+ Script = None,
+ ExtraArguments = None,
+ Stored = False,
+ PerformanceProfile = None,
+ InputFunctionAsMulti = False,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "ObservationOperator"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = FullOperator(
name = Concept,
asMatrix = Matrix,
asOneFunction = OneFunction,
asThreeFunctions = ThreeFunctions,
- asScript = Script,
+ asScript = self.__with_directory(Script),
asDict = Parameters,
appliedInX = AppliedInXb,
- avoidRC = AvoidRC,
+ extraArguments = ExtraArguments,
+ performancePrf = PerformanceProfile,
+ inputAsMF = InputFunctionAsMulti,
scheduledBy = None,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setEvolutionModel(self,
- Matrix = None,
- OneFunction = None,
- ThreeFunctions = None,
- Parameters = None,
- Script = None,
- Scheduler = None,
- AvoidRC = True,
- Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ def setEvolutionModel(
+ self,
+ Matrix = None,
+ OneFunction = None,
+ ThreeFunctions = None,
+ Parameters = None,
+ Script = None,
+ Scheduler = None,
+ ExtraArguments = None,
+ Stored = False,
+ PerformanceProfile = None,
+ InputFunctionAsMulti = False,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "EvolutionModel"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = FullOperator(
name = Concept,
asMatrix = Matrix,
asOneFunction = OneFunction,
asThreeFunctions = ThreeFunctions,
- asScript = Script,
+ asScript = self.__with_directory(Script),
asDict = Parameters,
appliedInX = None,
- avoidRC = AvoidRC,
+ extraArguments = ExtraArguments,
+ performancePrf = PerformanceProfile,
+ inputAsMF = InputFunctionAsMulti,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setControlModel(self,
- Matrix = None,
- OneFunction = None,
- ThreeFunctions = None,
- Parameters = None,
- Script = None,
- Scheduler = None,
- AvoidRC = True,
- Stored = False,
- Checked = False):
- "Definition d'un concept de calcul"
+ def setControlModel(
+ self,
+ Matrix = None,
+ OneFunction = None,
+ ThreeFunctions = None,
+ Parameters = None,
+ Script = None,
+ Scheduler = None,
+ ExtraArguments = None,
+ Stored = False,
+ PerformanceProfile = None,
+ InputFunctionAsMulti = False,
+ Checked = False ):
+ "Définition d'un concept de calcul"
Concept = "ControlModel"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = FullOperator(
name = Concept,
asMatrix = Matrix,
asOneFunction = OneFunction,
asThreeFunctions = ThreeFunctions,
- asScript = Script,
+ asScript = self.__with_directory(Script),
asDict = Parameters,
appliedInX = None,
- avoidRC = AvoidRC,
+ extraArguments = ExtraArguments,
+ performancePrf = PerformanceProfile,
+ inputAsMF = InputFunctionAsMulti,
scheduledBy = Scheduler,
toBeChecked = Checked,
- )
+ )
if Stored:
self.__StoredInputs[Concept] = self.__adaoObject[Concept].getO()
return 0
- def setDebug(self, level = 10):
+ def setName(self, String=None):
+ "Définition d'un concept de calcul"
+ self.__case.register("setName", dir(), locals())
+ if String is not None:
+ self.__name = str(String)
+ else:
+ self.__name = None
+ self.__StoredInputs["Name"] = self.__name
+
+ def setDirectory(self, String=None):
+ "Définition d'un concept de calcul"
+ self.__case.register("setDirectory", dir(), locals())
+ if os.path.isdir(os.path.abspath(str(String))):
+ self.__directory = os.path.abspath(str(String))
+ else:
+ self.__directory = None
+ self.__StoredInputs["Directory"] = self.__directory
+
+ def setDebug(self, __level = 10):
"NOTSET=0 < DEBUG=10 < INFO=20 < WARNING=30 < ERROR=40 < CRITICAL=50"
- self.__case.register("setDebug",dir(),locals())
+ self.__case.register("setDebug", dir(), locals())
log = logging.getLogger()
- log.setLevel( level )
- self.__StoredInputs["Debug"] = level
+ log.setLevel( __level )
+ logging.debug("Mode debug initialisé avec %s %s"%(version.name, version.version))
+ self.__StoredInputs["Debug"] = __level
self.__StoredInputs["NoDebug"] = False
return 0
def setNoDebug(self):
"NOTSET=0 < DEBUG=10 < INFO=20 < WARNING=30 < ERROR=40 < CRITICAL=50"
- self.__case.register("setNoDebug",dir(),locals())
+ self.__case.register("setNoDebug", dir(), locals())
log = logging.getLogger()
log.setLevel( logging.WARNING )
self.__StoredInputs["Debug"] = logging.WARNING
self.__StoredInputs["NoDebug"] = True
return 0
- def setAlgorithmParameters(self,
+ def setAlgorithmParameters(
+ self,
Algorithm = None,
Parameters = None,
- Script = None):
- "Definition d'un concept de calcul"
+ Script = None ):
+ "Définition d'un concept de calcul"
Concept = "AlgorithmParameters"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept] = AlgorithmAndParameters(
name = Concept,
asAlgorithm = Algorithm,
asDict = Parameters,
- asScript = Script,
- )
+ asScript = self.__with_directory(Script),
+ )
return 0
- def updateAlgorithmParameters(self,
+ def updateAlgorithmParameters(
+ self,
Parameters = None,
- Script = None):
- "Mise a jour d'un concept de calcul"
- if "AlgorithmParameters" not in self.__adaoObject:
- raise ValueError("No algorithm registred, ask for one before updating parameters")
- self.__adaoObject["AlgorithmParameters"].updateParameters(
+ Script = None ):
+ "Mise à jour d'un concept de calcul"
+ Concept = "AlgorithmParameters"
+ if Concept not in self.__adaoObject or self.__adaoObject[Concept] is None:
+ raise ValueError("\n\nNo algorithm registred, set one before updating parameters or executing\n")
+ self.__adaoObject[Concept].updateParameters(
asDict = Parameters,
- asScript = Script,
- )
+ asScript = self.__with_directory(Script),
+ )
+ # RaJ du register
return 0
- def setObserver(self,
+ def setRegulationParameters(
+ self,
+ Algorithm = None,
+ Parameters = None,
+ Script = None ):
+ "Définition d'un concept de calcul"
+ Concept = "RegulationParameters"
+ self.__case.register("set" + Concept, dir(), locals())
+ self.__adaoObject[Concept] = RegulationAndParameters(
+ name = Concept,
+ asAlgorithm = Algorithm,
+ asDict = Parameters,
+ asScript = self.__with_directory(Script),
+ )
+ return 0
+
+ def setSupplementaryParameters(
+ self,
+ Parameters = None,
+ Script = None ):
+ "Définition d'un concept de calcul"
+ Concept = "SupplementaryParameters"
+ self.__case.register("set" + Concept, dir(), locals())
+ self.__adaoObject[Concept] = ExternalParameters(
+ name = Concept,
+ asDict = Parameters,
+ asScript = self.__with_directory(Script),
+ )
+ return 0
+
+ def updateSupplementaryParameters(
+ self,
+ Parameters = None,
+ Script = None ):
+ "Mise à jour d'un concept de calcul"
+ Concept = "SupplementaryParameters"
+ if Concept not in self.__adaoObject or self.__adaoObject[Concept] is None:
+ self.__adaoObject[Concept] = ExternalParameters(name = Concept)
+ self.__adaoObject[Concept].updateParameters(
+ asDict = Parameters,
+ asScript = self.__with_directory(Script),
+ )
+ return 0
+
+ def setObserver(
+ self,
Variable = None,
Template = None,
String = None,
Script = None,
+ Info = None,
ObjectFunction = None,
- Scheduler = None,
- Info = None):
- "Definition d'un concept de calcul"
+ CrossObs = False,
+ SyncObs = True,
+ Scheduler = None ):
+ "Définition d'un concept de calcul"
Concept = "Observer"
- self.__case.register("set"+Concept, dir(), locals())
+ self.__case.register("set" + Concept, dir(), locals())
self.__adaoObject[Concept].append( DataObserver(
name = Concept,
onVariable = Variable,
asTemplate = Template,
asString = String,
- asScript = Script,
+ asScript = self.__with_directory(Script),
asObsObject = ObjectFunction,
withInfo = Info,
+ crossObs = CrossObs,
+ syncObs = SyncObs,
scheduledBy = Scheduler,
withAlgo = self.__adaoObject["AlgorithmParameters"]
- ))
+ ))
return 0
- def removeObserver(self,
+ def removeObserver(
+ self,
Variable = None,
- ObjectFunction = None,
- ):
- """
- Permet de retirer un observer à une ou des variable nommée.
- """
+ ObjectFunction = None ):
+ "Permet de retirer un observer à une ou des variable nommées"
if "AlgorithmParameters" not in self.__adaoObject:
raise ValueError("No algorithm registred, ask for one before removing observers")
#
else:
return self.__adaoObject["AlgorithmParameters"].removeObserver( ename, ObjectFunction )
- # -----------------------------------------------------------
-
- def setDiagnostic(self,
- Diagnostic = None,
- Identifier = None,
- Parameters = None,
- Script = None,
- Unit = None,
- BaseType = None):
- "Definition d'un concept de calcul"
- Concept = "Diagnostic"
- self.__case.register("set"+Concept, dir(), locals())
- self.__adaoObject[Concept].append( DiagnosticAndParameters(
- name = Concept,
- asDiagnostic = Diagnostic,
- asIdentifier = Identifier,
- asDict = Parameters,
- asScript = Script,
- asUnit = Unit,
- asBaseType = BaseType,
- asExistingDiags = self.__StoredInputs[Concept],
- ))
- self.__StoredInputs[Concept].append(str(Identifier))
+ def setUserPostAnalysis(
+ self,
+ Template = None,
+ String = None,
+ Script = None ):
+ "Définition d'un concept de calcul"
+ Concept = "UserPostAnalysis"
+ self.__case.register("set" + Concept, dir(), locals())
+ self.__adaoObject[Concept].append( repr(UserScript(
+ name = Concept,
+ asTemplate = Template,
+ asString = String,
+ asScript = self.__with_directory(Script),
+ )))
return 0
+ # -----------------------------------------------------------
+
def get(self, Concept=None, noDetails=True ):
- "Recuperation d'une sortie du calcul"
- """
- Permet d'accéder aux informations stockées, diagnostics et résultats
- disponibles après l'exécution du calcul. Attention, quand un diagnostic
- porte le même nom qu'une variable stockée (paramètre ou résultat),
- c'est la variable stockée qui est renvoyée, et le diagnostic est
- inatteignable.
- """
+ "Récupération d'une sortie du calcul"
if Concept is not None:
try:
- self.__case.register("get", dir(), locals(), Concept) # Break pickle in Python 2
- except:
+ self.__case.register("get", dir(), locals(), Concept) # Break pickle in Python 2
+ except Exception:
pass
if Concept in self.__StoredInputs:
return self.__StoredInputs[Concept]
elif Concept == "AlgorithmRequiredParameters" and self.__adaoObject["AlgorithmParameters"] is not None:
return self.__adaoObject["AlgorithmParameters"].getAlgorithmRequiredParameters(noDetails)
#
- elif Concept in self.__StoredInputs["Diagnostic"]:
- indice = self.__StoredInputs["Diagnostic"].index(Concept)
- return self.__adaoObject["Diagnostic"][indice].get()
+ elif Concept == "AlgorithmRequiredInputs" and self.__adaoObject["AlgorithmParameters"] is not None:
+ return self.__adaoObject["AlgorithmParameters"].getAlgorithmInputArguments()
+ #
+ elif Concept == "AlgorithmAttributes" and self.__adaoObject["AlgorithmParameters"] is not None:
+ return self.__adaoObject["AlgorithmParameters"].getAlgorithmAttributes()
+ #
+ elif self.__adaoObject["SupplementaryParameters"] is not None and Concept == "SupplementaryParameters":
+ return self.__adaoObject["SupplementaryParameters"].get()
+ #
+ elif self.__adaoObject["SupplementaryParameters"] is not None and Concept in self.__adaoObject["SupplementaryParameters"]:
+ return self.__adaoObject["SupplementaryParameters"].get( Concept )
#
else:
- raise ValueError("The requested key \"%s\" does not exists as an input, a diagnostic or a stored variable."%Concept)
+ raise ValueError("The requested key \"%s\" does not exists as an input or a stored variable."%Concept)
else:
allvariables = {}
- allvariables.update( {"AlgorithmParameters":self.__adaoObject["AlgorithmParameters"].get()} )
- allvariables.update( self.__adaoObject["AlgorithmParameters"].get() )
+ allvariables.update( {"AlgorithmParameters": self.__adaoObject["AlgorithmParameters"].get()} )
+ if self.__adaoObject["SupplementaryParameters"] is not None:
+ allvariables.update( {"SupplementaryParameters": self.__adaoObject["SupplementaryParameters"].get()} )
+ # allvariables.update( self.__adaoObject["AlgorithmParameters"].get() )
allvariables.update( self.__StoredInputs )
- allvariables.pop('Diagnostic', None)
allvariables.pop('Observer', None)
+ allvariables.pop('UserPostAnalysis', None)
return allvariables
# -----------------------------------------------------------
- def get_available_variables(self):
+ def __get_available_variables(self):
"""
Renvoie les variables potentiellement utilisables pour l'étude,
initialement stockées comme données d'entrées ou dans les algorithmes,
préalablement choisi sinon la méthode renvoie "None".
"""
if len(list(self.__adaoObject["AlgorithmParameters"].keys())) == 0 and \
- len(list(self.__StoredInputs.keys())) == 0:
+ len(list(self.__StoredInputs.keys())) == 0:
return None
else:
variables = []
if len(list(self.__adaoObject["AlgorithmParameters"].keys())) > 0:
variables.extend(list(self.__adaoObject["AlgorithmParameters"].keys()))
+ if self.__adaoObject["SupplementaryParameters"] is not None and \
+ len(list(self.__adaoObject["SupplementaryParameters"].keys())) > 0:
+ variables.extend(list(self.__adaoObject["SupplementaryParameters"].keys()))
if len(list(self.__StoredInputs.keys())) > 0:
variables.extend( list(self.__StoredInputs.keys()) )
- variables.remove('Diagnostic')
variables.remove('Observer')
+ variables.remove('UserPostAnalysis')
variables.sort()
return variables
- def get_available_algorithms(self):
+ def __get_available_algorithms(self):
"""
Renvoie la liste des algorithmes potentiellement utilisables, identifiés
par les chaînes de caractères.
"""
files = []
for directory in sys.path:
- if os.path.isdir(os.path.join(directory,"daAlgorithms")):
- for fname in os.listdir(os.path.join(directory,"daAlgorithms")):
- root, ext = os.path.splitext(fname)
- if ext == '.py' and root != '__init__':
- files.append(root)
- files.sort()
- return files
-
- def get_available_diagnostics(self):
- """
- Renvoie la liste des diagnostics potentiellement utilisables, identifiés
- par les chaînes de caractères.
- """
- files = []
- for directory in sys.path:
- if os.path.isdir(os.path.join(directory,"daDiagnostics")):
- for fname in os.listdir(os.path.join(directory,"daDiagnostics")):
- root, ext = os.path.splitext(fname)
- if ext == '.py' and root != '__init__':
- files.append(root)
+ trypath = os.path.join(directory, "daAlgorithms")
+ if os.path.isdir(trypath):
+ for fname in os.listdir(trypath):
+ if os.path.isfile(os.path.join(trypath, fname)):
+ root, ext = os.path.splitext(fname)
+ if ext != ".py":
+ continue
+ with open(os.path.join(trypath, fname)) as fc:
+ iselal = bool("class ElementaryAlgorithm" in fc.read())
+ if iselal and ext == '.py' and root != '__init__':
+ files.append(root)
files.sort()
return files
- # -----------------------------------------------------------
-
- def get_algorithms_main_path(self):
+ def __get_algorithms_main_path(self):
"""
Renvoie le chemin pour le répertoire principal contenant les algorithmes
- dans un sous-répertoire "daAlgorithms"
"""
return self.__parent
- def add_algorithms_path(self, Path=None):
+ def __add_algorithms_path(self, Path=None):
"""
Ajoute au chemin de recherche des algorithmes un répertoire dans lequel
se trouve un sous-répertoire "daAlgorithms"
-
- Remarque : si le chemin a déjà été ajouté pour les diagnostics, il n'est
- pas indispensable de le rajouter ici.
- """
- if not os.path.isdir(Path):
- raise ValueError("The given "+Path+" argument must exist as a directory")
- if not os.path.isdir(os.path.join(Path,"daAlgorithms")):
- raise ValueError("The given \""+Path+"\" argument must contain a subdirectory named \"daAlgorithms\"")
- if not os.path.isfile(os.path.join(Path,"daAlgorithms","__init__.py")):
- raise ValueError("The given \""+Path+"/daAlgorithms\" path must contain a file named \"__init__.py\"")
- sys.path.insert(0, os.path.abspath(Path))
- sys.path = PlatformInfo.uniq( sys.path ) # Conserve en unique exemplaire chaque chemin
- return 0
-
- def get_diagnostics_main_path(self):
- """
- Renvoie le chemin pour le répertoire principal contenant les diagnostics
- dans un sous-répertoire "daDiagnostics"
- """
- return self.__parent
-
- def add_diagnostics_path(self, Path=None):
- """
- Ajoute au chemin de recherche des algorithmes un répertoire dans lequel
- se trouve un sous-répertoire "daDiagnostics"
-
- Remarque : si le chemin a déjà été ajouté pour les algorithmes, il n'est
- pas indispensable de le rajouter ici.
"""
if not os.path.isdir(Path):
- raise ValueError("The given "+Path+" argument must exist as a directory")
- if not os.path.isdir(os.path.join(Path,"daDiagnostics")):
- raise ValueError("The given \""+Path+"\" argument must contain a subdirectory named \"daDiagnostics\"")
- if not os.path.isfile(os.path.join(Path,"daDiagnostics","__init__.py")):
- raise ValueError("The given \""+Path+"/daDiagnostics\" path must contain a file named \"__init__.py\"")
+ raise ValueError("The given " + Path + " argument must exist as a directory")
+ if not os.path.isdir(os.path.join(Path, "daAlgorithms")):
+ raise ValueError("The given \"" + Path + "\" argument must contain a subdirectory named \"daAlgorithms\"")
+ if not os.path.isfile(os.path.join(Path, "daAlgorithms", "__init__.py")):
+ raise ValueError("The given \"" + Path + "/daAlgorithms\" path must contain a file named \"__init__.py\"")
sys.path.insert(0, os.path.abspath(Path))
- sys.path = PlatformInfo.uniq( sys.path ) # Conserve en unique exemplaire chaque chemin
+ sys.path = PlatformInfo.uniq( sys.path ) # Conserve en unique exemplaire chaque chemin
return 0
# -----------------------------------------------------------
- def execute(self, Executor=None, FileName=None):
+ def execute(self, Executor=None, SaveCaseInFile=None, nextStep=False):
"Lancement du calcul"
- self.__case.register("execute",dir(),locals(),None,True)
- Operator.CM.clearCache()
- #~ try:
- if Executor == "YACS": self.__executeYACSScheme( FileName )
- else: self.__executePythonScheme( FileName )
- #~ except Exception as e:
- #~ if isinstance(e, SyntaxError): msg = "at %s: %s"%(e.offset, e.text)
- #~ else: msg = ""
- #~ raise ValueError("during execution, the following error occurs:\n\n%s %s\n\nSee also the potential messages, which can show the origin of the above error, in the launching terminal.\n"%(str(e),msg))
+ self.__case.register("execute", dir(), locals(), None, True)
+ self.updateAlgorithmParameters(Parameters={"nextStep": bool(nextStep)})
+ if not nextStep:
+ Operator.CM.clearCache()
+ try:
+ if Executor == "YACS":
+ self.__executeYACSScheme( SaveCaseInFile )
+ else:
+ self.__executePythonScheme( SaveCaseInFile )
+ except Exception as e:
+ if isinstance(e, SyntaxError):
+ msg = "at %s: %s"%(e.offset, e.text)
+ else:
+ msg = ""
+ raise ValueError((
+ "during execution, the following error occurs:\n" + \
+ "\n%s %s\n\nSee also the potential messages, " + \
+ "which can show the origin of the above error, " + \
+ "in the launching terminal.\n")%(str(e), msg))
return 0
def __executePythonScheme(self, FileName=None):
if FileName is not None:
self.dump( FileName, "TUI")
self.__adaoObject["AlgorithmParameters"].executePythonScheme( self.__adaoObject )
+ if "UserPostAnalysis" in self.__adaoObject and len(self.__adaoObject["UserPostAnalysis"]) > 0:
+ self.__objname = self.__retrieve_objname()
+ for __UpaOne in self.__adaoObject["UserPostAnalysis"]:
+ __UpaOne = eval(str(__UpaOne))
+ exec(__UpaOne, {}, {'self': self, 'ADD': self, 'case': self, 'adaopy': self, self.__objname: self})
return 0
def __executeYACSScheme(self, FileName=None):
"Lancement du calcul"
self.__case.register("executeYACSScheme", dir(), locals())
- if FileName is not None:
- self.dump( FileName, "DIC")
+ self.dump( FileName, "YACS")
self.__adaoObject["AlgorithmParameters"].executeYACSScheme( FileName )
return 0
def dump(self, FileName=None, Formater="TUI"):
"Restitution normalisée des commandes"
- return self.__case.dump(FileName, Formater)
+ __Upa = "\n".join(self.__PostAnalysis)
+ return self.__case.dump(FileName, Formater, __Upa)
- def load(self, FileName=None, Formater="TUI"):
+ def load(self, FileName=None, Content=None, Object=None, Formater="TUI"):
"Chargement normalisé des commandes"
- __commands = self.__case.load(FileName, Formater)
- from numpy import array, matrix
+ __commands = self.__case.load(FileName, Content, Object, Formater)
+ from numpy import array, matrix # noqa: F401
for __command in __commands:
- exec("self."+__command)
- return 0
+ if (__command.find("set") > -1 and __command.find("set_") < 0) or 'UserPostAnalysis' in __command:
+ exec("self." + __command, {}, locals())
+ else:
+ self.__PostAnalysis.append(__command)
+ return self
+
+ def convert(self,
+ FileNameFrom=None, ContentFrom=None, ObjectFrom=None, FormaterFrom="TUI",
+ FileNameTo=None, FormaterTo="TUI" ):
+ "Conversion normalisée des commandes"
+ return self.load(
+ FileName=FileNameFrom, Content=ContentFrom, Object=ObjectFrom, Formater=FormaterFrom
+ ).dump(
+ FileName=FileNameTo, Formater=FormaterTo
+ )
def clear(self):
"Effacement du contenu du cas en cours"
# -----------------------------------------------------------
+ def __with_directory(self, __filename=None):
+ if os.path.exists(str(__filename)):
+ __fullpath = __filename
+ elif os.path.exists(os.path.join(str(self.__directory), str(__filename))):
+ __fullpath = os.path.join(self.__directory, str(__filename))
+ else:
+ __fullpath = __filename
+ return __fullpath
+
+ def __retrieve_objname(self):
+ "Ne pas utiliser dans le __init__, la variable appelante n'existe pas encore"
+ __names = []
+ for level in reversed(inspect.stack()):
+ __names += [name for name, value in level.frame.f_locals.items() if value is self]
+ __names += [name for name, value in globals().items() if value is self]
+ while 'self' in __names:
+ __names.remove('self') # Devrait toujours être trouvé, donc pas d'erreur
+ if len(__names) > 0:
+ self.__objname = __names[0]
+ else:
+ self.__objname = "ADD"
+ return self.__objname
+
def __dir__(self):
"Clarifie la visibilité des méthodes"
- return ['set', 'get', 'execute', '__doc__', '__init__', '__module__']
+ return ['set', 'get', 'execute', 'dump', 'load', '__doc__', '__init__', '__module__']
+
+ def __str__(self):
+ "Représentation pour impression (mais pas repr)"
+ msg = self.dump(None, "SimpleReportInPlainTxt")
+ return msg
+
+ def sysinfo(self, title=""):
+ msg = PlatformInfo.PlatformInfo().getAllInformation(" ", title)
+ return msg
+
+ def callinfo(self, __prefix=" "):
+ msg = ""
+ for oname in ["ObservationOperator", "EvolutionModel"]:
+ if hasattr(self.__adaoObject[oname], "nbcalls"):
+ ostats = self.__adaoObject[oname].nbcalls()
+ msg += "\n%sNumber of calls for the %s:"%(__prefix, oname)
+ for otype in ["Direct", "Tangent", "Adjoint"]:
+ if otype in ostats:
+ msg += "\n%s%30s : %s"%(__prefix, "%s evaluation"%(otype,), ostats[otype][0])
+ msg += "\n"
+ return msg
def prepare_to_pickle(self):
"Retire les variables non pickelisables, avec recopie efficace"
if self.__adaoObject['AlgorithmParameters'] is not None:
for k in self.__adaoObject['AlgorithmParameters'].keys():
- if k == "Algorithm": continue
+ if k == "Algorithm":
+ continue
if k in self.__StoredInputs:
- raise ValueError("the key \"%s\s to be transfered for pickling will overwrite an existing one.")
+ raise ValueError("The key \"%s\" to be transfered for pickling will overwrite an existing one."%(k,))
if self.__adaoObject['AlgorithmParameters'].hasObserver( k ):
self.__adaoObject['AlgorithmParameters'].removeObserver( k, "", True )
self.__StoredInputs[k] = self.__adaoObject['AlgorithmParameters'].pop(k, None)
- del self.__adaoObject # Break pickle in Python 2
- del self.__case # Break pickle in Python 2
- return 0
+ if sys.version_info[0] == 2:
+ del self.__adaoObject # Because it breaks pickle in Python 2. Not required for Python 3
+ del self.__case # Because it breaks pickle in Python 2. Not required for Python 3
+ if sys.version_info.major < 3:
+ return 0
+ else:
+ return self.__StoredInputs
# ==============================================================================
if __name__ == "__main__":
- print('\n AUTODIAGNOSTIC \n')
+ print("\n AUTODIAGNOSTIC\n")
