#!/usr/bin/python
# -*- coding: <UTF-8> -*-
"""
Python Version : 3.6
$LastChangedRevision$
$LastChangedDate:: $
Copyright (c) 2018, Linz Center of Mechatronics GmbH (LCM) http://www.lcm.at/
All rights reserved.
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
"""
import symspace.wrapper.symwrapper as sw
from symspace.jobinterface.portDescWrapper import SymJobDesc
from symspace.generic_gui_qt import GUIMessageThread
import json
import os
import numpy as np
import pickle
import uuid
import shutil
import time
# temporary to include TolData package
# try:
# # curPath = CurProject.getDirectory()
# curPath = r'C:\Users\AK112122\MyDocuments\10_Projekte\Toleranzanalyse\SyM_wokrflow'.replace('\\', '/')
# if curPath not in sys.path:
# sys.path.append(curPath)
# print(f"path {curPath} added!")
# except:
# pass
from symspace.symtolerance2.symtolsettings import SymTolSettings
from symspace.symtolerance2.toldata import TolData
from symspace.symtolerance2.symtoltools import tprint, ControlFile, RunEveryN, setStandardLogger
import logging
# TODO:
# - check if base project is modified (between initialization of SymTol and createSimProject)!!!
# - make some interface to add dict elements to every class method which is used to update the settings dataclass
# - load settings from pickled tolData (see loadTolData())
# - log functionality, especially error logging!
[docs]class SymTol:
"""
The :class:`SymTol` class is used to perform a tolerance analysis based on an existing |sym| project.
.. note::
Tolerance analysis has pre-alpha status and is not integrated into the |sym| core and not accessible via GUI
interaction. Setting up and running ana analysis has to be done using the |sym| Python Console. No automatic
postprocessing is available at the moment!
.. note::
In this introductory documentation the most common and necessary steps to perform a tolerance analysis are
described. This is not a full documentation of all features. More details might be presented at the available
methods.
**Basic workflow**
.. image:: _rstcontent/pics/workflow.png
:scale: 30%
:align: right
Initialization
Starting from an existing |sym| project, this base project has to be prepared for tolerance analysis, i.e.,
define which fields are affected by tolerances, what the tolerance distribution looks like, and what the
results are to be investigated during analysis. This |sym| project can then be used to initialize a |sym|
tolerance analysis.
Configuration
Before starting an analysis, several settings can be modified. For example, number of simulation, number of
indizes of the temporary |sym| project which is later created for simulation, autosave interval, ...
Start an analysis
If an analysis is started, a temporary |sym| project is created based on the initialization project. This
temporary project gets a certain number of indizes, which can be defined in the configuration of the tolerance
analysis problem. Then, tolerance affected values are assigned to each index and all indizes are regenerated.
If an index is finnished sucessfully, variables and results are collected, new tolerance affected values are
assigned, and the corresponding index is regenerated. This loop runs, until a predefined number of tolerance
affected evaluations is finnished
Clean up
At the end, a clean up can be performed to remove the temporary project. This has to be triggered manually!
|clearfloat|
**Definition of tolerance-affected variables**
Tolerances can be defined in the `Comment` box of the respective parameter which can be found in the `Details`-tab
of the |sym| GUI. The comment has to start with the :code:`@Tolerance` string, further details are defined as json
string.
For example, a variable whichs' value has uniform distribution can be defined writing
.. code-block:: none
@Tolerance{"var":{"uniform":{"lower_tol":-0.1, "upper_tol":0.7}}}
in the `Comment` area of the corresponding |sym| parameter. The key :code:`"var"` defines that this parameter is
a variable which has to be modified during tolerance analysis. For a list of available distributions see
:func:`_findAtToleranceData_r`.
**Definition of results**
Results which have to be observed during tolerance analysis are defined in a similar manner, using the :code:`"res"`
keyword:
.. code-block:: none
@Tolerance{"res":{}}
For more information, also see :func:`_findAtToleranceData_r`.
**Running an analysis**
When the |sym| project is prepared for tolerance analysis (setting variables and results), assign the
:code:`CurProject` handle to the |sym| Python Console
(:menuselection:`Debug --> Python --> Assign Project Handle in Python Console`, :guibutton:`strg` + :guibutton:`P`).
Then initialize the tolerance project in the |sym| Python Console(for other initialization methos see :func:`__init__`).
.. code-block:: python
>>> stp = SymTol(CurProject) # create a SymTolProject
After creation of a :class:`SymTol` instance, different settings can be defined
(for more information, see the :mod:`symspace.symtolerance2.symtolsettings` module). The :class:`SymTol` object
has a :attr:`config` attribute, where all settings can be defined. Some useful configurations
might be:
.. code-block:: python
>>> stp.config.samplesCount = 100 # number of samples to be evaluated
>>> stp.config.simProject.indices = 20 # number of indizes of the temporary SyMSpace project
>>> stp.config.simControl.tolDataPickleInterval = 10 # autosave functionality (variables and results are
>>> # saved after n evaluated samples, overwriting
>>> # previouse autosave files)
After adopting the settings, one can start a tolerance analysis using
.. code-block:: python
>>> stp.start()
Now 100 samples are evaluated, where every sample is automatically defined randomly with the specified distribution
as defined by the :code:`@Tolerance{}` key.
.. note::
Unfortunately, Python threading seems not to be possible from the |sym| Python Console.
Thus, during performing the tolerance analysis the |sym| Python Console is locked. If
you want to interrupt the evaluation of the analysis - might be helpful especially during testing phase - a
workaround using a simulation control file :file:`<SyM-project-filename>.ctrl` is created. This file has two
flags defined, namely :code:`stop=False` and :code:`finish=False`. Set either of these to :code:`True` and save
the file to stop or finish the analysis before the defined number of evaluations is reached. When setting
:code:`finish` to :code:`True`, all indizes which are already under progress are finished. setting :code:`stop`
to :code:`True`, the analysis is stopped imediatly and all running jobs are removed.
When the defined number of samples is evaluated successfully, the results and also the corresponding variables can
accessed via the :attr:`results` and :attr:`variables` attributes of the :class:`SymTol` object.
.. code-block:: python
>>> stp.results
>>> stp.variables
Both, results and variables, are stored in a 3-dimensional numpy array, where the different samples can be accessed
via the 3rd dimension (axis=2). The variable and result data of a single sample can be 2-dimensional and is stored
in axis=0 and axis=1.
If additional samples are necessary, the analysis can be started as often as necessary. Results and variables are
appended to the existing data.
.. note::
If a tolerance analysis is initialized from a project where an existing analysis is available, these data can be
loaded and the analysis can be continued.
.. warning::
If the analysis should not be continued, but a new analysis should be performed, the axisting tolerance-data
file will be overwritten. If you want to keep this data, make a copy of the original file under a different name!
**Clean-up tolerance analysis project**
Before you close the |sym| - Python Console, a clean-up can be performed. This removes all temporary files created
during the tolerance analysis.
.. code-block:: python
>>> stp.cleanup()
**Post-processing**
For postprocessing of scalar objectives the :class:`symtolerance2.toldata.TolData` class can be used. But
postprocessing is not integrated into the main workflow so far and the :class:`symtolerance2.toldata.TolData` class
has pre-alpha status!
"""
[docs] def __init__(self, sym_project):
"""
:class:`SymTol` initialization function.
Parameters
----------
sym_project: :class:`symspace.wrapper.symwrapper.SymProject` | str (absolute path to mop file)
Assign a :class:`SymProject` handle from the :mod:`symspace.wrapper.symwrapper` module OR the
absolute path to the :code:`.mop` file of the |sym| project which should be loaded.
"""
# load base project
if isinstance(sym_project, sw.SymProject):
self.baseProject = sym_project
elif sym_project == 'dummy':
self.baseProject = 'dummy'
elif isinstance(sym_project, str):
# try to load project
if os.path.exists(sym_project) and sym_project.endswith('.mop'):
self.baseProject = sw.SymProject.load(os.path.abspath(sym_project))
elif hasattr(sym_project, 'read'):
# should be a file-handle
# TODO: implement something to load a previously done tolerance analysis,
# e.g., all settings stored as pickle file and can be loaded
# maybe this is not necessary if restarting tolerance analysis when just checking
# if .tol folder exists ....
print(f"Error: Loading from file handle is not yet implemented!")
pass
else:
print(f"Error: SymProject has to be from type {type(sw.SymProject)} or a valid path to a SyMSPace-Project (.mop file)!")
"""
Initialize configuration/settings Instance
"""
# temporary to check for dummy
if isinstance(self.baseProject, sw.SymProject):
self._config = SymTolSettings(os.path.join(self.baseProject.getDirectory(), self.baseProject.getFile()))
else: #self.baseProject == 'dummy'
self._config = SymTolSettings('dummy_directory', 'dummy_file')
"""
Initialize logging
Here the logging module is initialized. Different logger (using this initialization) can be created using
new_logger = logging.getLogger('new_name')
"""
setStandardLogger(self.__getBaseFileName()+'_error.log')
logger = logging.getLogger(self.__class__.__name__)
logger.info('Logger initialized sucessfully!')
logger.info('Errors are written to ' + self.__getBaseFileName() + '_error.log')
"""
Parse SyMSpace tree to find tolerance related data
"""
self.AtToleranceData = []
# temporary to check for dummy
if isinstance(self.baseProject, sw.SymProject):
self._findAtToleranceData_r(self.baseProject)
# self.printAtToleranceData()
"""
Prepare simulation project
"""
self.simProject = None # will hold sw.SymProject which is used for tolerance analysis
self._next_index_cnt = 0 # used for control of the order of regenerating the indices
# (see self.__set_next_index())
self._index_status = [] # used to finish indices at the end
self._index_runtime = [] # used to check if an index has timeout
self._index_assigned_cnt = 0 # increase if sample/index is assigned, decrease if index has error
# used for simulation control
"""
Initialization of other members
"""
self._results = {} # initialize results structure with entries of list type for dynamic appending data
self._variables = {} # initialize variables structure with entries of list type for dynamic appending data
self._experiments = {} # use this dictionary to store information about experiments which are done ...
# not sure if this "field" will stay .... maybe only temporary
# is also stored with self.pickleTolData() and loaded by self.loadTolData()
# keys:
# -----
# - nominalSet: initialized and set to True in self.startNominalSet()
"""
check for existing tolData file
"""
tolDataDefaultName = self.__getTolDataDefaultFileName()
if os.path.exists(tolDataDefaultName):
msg_elements = {'icon': 'question', 'text': f'Do you want to load variable and result data stored in {os.path.split(tolDataDefaultName)[-1]}?', 'buttons': 'yes_no'}
msg = GUIMessageThread('Loading TolData', msg_elements)
msg.start()
if msg.accept:
self.loadTolData()
"""
Initialize a control elements to interrupt a running analysis
(workaround, since threading seems not to work)
"""
self.ctrlFile = None # initialization at createSimProject
self.stop = False # can be set to True via ctrlFile
self.finish = False # can be set to True via ctrlFile
self._JobIDIdentifier = [] # to clean up after stop
[docs] def loadTolData(self, fname=None):
"""
Load variables and results from previous tolerance analysis, stored by :func:`pickleTolData`.
Parameters
----------
fname : str
full path to pickle file stored by the :func:`pickleTolData` function; if default (`None`), default
file name from :func:`pickleTolData` is used.
"""
# TODO: also load settings, which are stored by pickleTolData ! It might be good to serialize config to ordinary
# dict, otherwise settings class is necessary to laod data!
logger = logging.getLogger(self.loadTolData.__name__)
if fname is None:
fname = self.__getTolDataDefaultFileName()
logger.info('searching for default filename to load tolerance related data')
try:
if os.path.exists(fname):
toldata = pickle.load(open(fname, 'rb'))
if '_experiments' in toldata.keys():
self._experiments = toldata['_experiments']
res_data = toldata['results']
var_data = toldata['variables']
size = res_data[list(res_data.keys())[0]].shape[2]
for res in res_data.keys():
for idx in range(res_data[res].shape[2]):
tmp = res_data[res][:, :, idx]
if tmp.shape == (1, 1):
tmp = tmp[0, 0] # store as scalar!
self._results.setdefault(res, []).append(tmp)
for var in var_data.keys():
for idx in range(var_data[var].shape[2]):
tmp = var_data[var][:, :, idx]
if tmp.shape == (1, 1):
tmp = tmp[0, 0] # store as scalar!
self._variables.setdefault(var, []).append(tmp)
logger.info(f"{size} samples loaded sucessfully from {fname}!")
logger.info(f"{self.sampleCount} samples available in total!")
# print('\n'+'*'*30)
# tprint(f"{size} samples loaded sucessfully from {fname}!")
# tprint(f"{self.sampleCount} samples available in total!")
else:
#print(f"file {fname} does not exist!")
logger.error(f"file {fname} does not exist!")
except IOError as e:
errno, strerror = e.args
#tprint(f"{self.__class__.__name__}.loadTolData(): I/O error({errno}): {strerror}", level='ERROR')
logger.error(f"I/O error({errno}): {strerror}")
except:
logger.error('Unexpected Error!')
#tprint(f"{self.__class__.__name__}.loadTolData(): Unexpected Error!", level='ERROR')
[docs] def findAtToleranceData(self):
# TODO: copy documentation from below and make project clean, i.e., also reset other things!?
self.AtToleranceData = []
self._findAtToleranceData_r(self.baseProject)
def _findAtToleranceData_r(self, sym_element):
"""
Recursively searches the tree for tolerance relevant fields, where :code:`sym_element` is the tree root. To
search the whole |sym| project, pass the projects :class:`symspace.wrapper.symwrapper.SymProject` handle to the
function. Relevant fields are those which have the :code:`@Tolerance{}` keyword in the
:guibutton:`Comment:__________` area of the :guibutton:`Details` tab. For more information about possible
keywords see :func:`setToleranceVariables`.
Found tolerance related information is stored in the :attr:`AtToleranceData` attribute of the :class:`SymTol`
instance and can be printed using :func:`printAtToleranceData`.
Parameters
----------
sym_element : :class:`symspace.wrapper.symwrapper.SymProject` | :class:`symspace.wrapper.symwrapper.SymContainer` | :class:`symspace.wrapper.symwrapper.SymCollector` | :class:`symspace.wrapper.symwrapper.SymFunction` | :class:`symspace.wrapper.symwrapper.SymField` | :class:`symspace.wrapper.symwrapper.SymFormula`
|sym| tree element which serves as root for the recursive search.
"""
# TODO: add check if @Tolerance{} comments are valid!
# maybe define dataclasses for possible settings and check if available and
# when available if all parameters are available and from proper type
logger = logging.getLogger(self._findAtToleranceData_r.__name__)
if isinstance(sym_element, (sw.SymProject, sw.SymContainer, sw.SymCollector,
sw.SymFunction)): # maybe more, every element which has sub-params (.getParam())
for elem in sym_element.getParam():
self._findAtToleranceData_r(elem)
else:
# should be SymField, SymFormula, ... maybe more
comment = sym_element.getComment()
identifier = sym_element.getIdentifier()
if comment.startswith('@Tolerance'):
# TODO: could check if var is valid input field, res has to be read-only, ...
# see also self.createSimProject()!
# TODO: check how many var fields, how many res fields, print info ...
logger.info(f"Tolerance {identifier} found ... {comment}")
# print(f"found: {identifier} ... {comment}")
json_str = comment.split('@Tolerance', 1)[1]
try:
tol = json.loads(json_str)
except ValueError:
logger.error(f"Comment string of '{identifier}' cannot be loaded by json parser!")
#print(f"\n+++ Comment string of '{identifier}' can not be loaded by json parser! +++\n")
# TODO: better error handling!
tol['identifier'] = identifier
base_value = self.baseProject.getObject(identifier).getData()
tol['base_value'] = base_value
self.AtToleranceData.append(tol)
[docs] def printAtToleranceData(self):
"""
Print all tolerance related data found in |sym| project (identified by the :code:`@Tolerance{}` keyword in the
:guibutton:`Comment:__________` field).
"""
s_rep = [[f"{k}:\t{v}" for k, v in d.items()] for d in self.AtToleranceData]
print(f"\nFollowing tolerance-related fields found:")
for p in s_rep:
print()
print("\n".join(p))
[docs] def createSimProject(self):
"""
Create and prepare the |sym| project which will be used for tolerance analyses.
By default the original |sym| project is cloned and stored as :file:`<SyMProject_name>_tol.mop`. If the
original |sym| project shall be used, *what is not recommended*, set
.. code-block:: python
<SymTol>.config.simProject.clone=False
The (temporary) simulation project can be removed using :func:`cleanUpProject`.
.. note::
If a |sym| formula is set to be tolerance affected, the formula is converted to a ParamField during
creation of the (temporary) simulation project (a formula field can be useful to define nominal values
for vectors or matrices). If a tolerance affected |sym| field has its `constant` flag set, this is removed
during creation of the simulation project.
"""
logger = logging.getLogger(self.createSimProject.__name__)
if not self._config.simProject.clone and self.simProject is None:
"""use base project which is not assigned so far"""
logger.warning('Using base project for tolerance variations - THIS IS NOT RECOMMENDED!')
logger.info(f'See {self.__class__.__name__}.cleanUpProject() to remove indices.')
#tprint(f"Using base project for tolerance variations - THIS IS NOT RECOMMENDED!"
# f"\n\t\t\tSee {self.__class__.__name__}.cleanUpProject() to remove indices.")
self.simProject = self.baseProject
elif self._config.simProject.clone:
""" create simProject """
base_file_parts = self.baseProject.getFile().rsplit('.', 1)
sim_project_name = base_file_parts[0] + '_tol.' + base_file_parts[1]
if isinstance(self.simProject, sw.SymProject) and self.simProject.getFile() == sim_project_name:
#tprint(f"Proper simulation project {self.simProject.getFile()} already assigned")
logger.info(f"Proper simulation project {self.simProject.getFile()} already assigned")
elif os.path.isfile(os.path.join(self.baseProject.getDirectory(), sim_project_name)):
logger.info(f"Temporary SyMSpace Project {sim_project_name} already exists:")
logger.warning("Existing simulation project is loaded. If the base project has changed, this can lead to wrong results!")
logger.info(f"If temporary simulation project should be newly created, run {self.__class__.__name__}.{self.cleanUpProject.__name__}() and rerun {self.__class__.__name__}.{self.createSimProject.__name__}().")
#tprint(f"Temporary SyMSpace Project {sim_project_name} already exists:")
#tprint(f"Existing project is loaded. If temporary simulation project should be newly created, run SymTol.cleanUpProject() and rerun SymTol.createSimProject()")
self.simProject = sw.SymProject.load(os.path.join(self.baseProject.getDirectory(), sim_project_name))
else:
logger.info(f"Create temporary SyMSpace Project: {sim_project_name}")
#tprint(f"create temporary SyMSpace Project: {sim_project_name}")
self.simProject = sw.SymProject.load(self._config.baseProjectFile)
# use the override=True option! Otherwise, the simProject is not created if .mop.files directory is
# still existing (even if the .mop file is not there). In this case, simProject points to a second
# instance of the baseProject using the baseProject.mop.files directory, what least to unpredictable
# behaviour !!!
self.simProject.saveAs(os.path.join(self.baseProject.getDirectory(), sim_project_name), override=True)
# Don't reload baseProject! Otherwise, the sel.baseProject handle does point to a different instance
# of the project and not to the one opened in SyMSpace (which was probably used to initialize the
# SymTol project!)
# do NOT: # self.baseProject = sw.SymProject.load(self._config.baseProjectFile)
"""
check:
- if var entries are formulas -> convert to fields
- if var entries are set as constant -> remove constant flag
"""
""" also see self.extractTolDataProject()"""
# TODO: probably other cases will arise ...
for tol in self.AtToleranceData:
if 'var' in tol.keys():
identifier = tol['identifier']
obj = self.simProject.getObject(identifier)
if isinstance(obj, sw.SymFormula):
if not self._config.simProject.clone:
logger.error(f"Formula {identifier} has to be converted to field to set tolerance values.")
logger.info("This is done automatically when using a clone of the base project for simulation. Using the base project, please convert manually.")
#tprint(
# f"ERROR: formula has to be converted to field to _config tolerance values, this is done automatically when not using the base project. Using the base project, please convert manually!")
else:
obj.javaObj.convertToParamField()
logger.warning(f"Converting {identifier} from SymFormula to SymField!")
#tprint(f"converting {identifier} from SymFormula to SymField!")
if isinstance(obj, sw.SymField) and obj.getConstant():
obj.setConstant(False)
logger.warning(f"Setting {identifier}'s constant flag to False!")
#tprint(f"INFO: setting {identifier}'s constant flag to False!")
""" check indizes: """
self.__updateSimProjectIndices()
""" save simulation project """
self.simProject.setIndex(0)
self.simProject.save()
""" create control file"""
self.ctrlFile = ControlFile(self.simProject)
self.stop = False
self.finish = False
[docs] def getWorstCaseIndex(self):
"""
Evaluate the index where the worst case design is stored within the :func:`results` and :func:`variables`
properties of the :class:`SymTol` instance. The worst case design is evaluated for each result (objective)
and is based on the available data (it is not the global worst case configuration!!)
Returns
-------
dict:
:code:`dict['SymIdentifier'] = <worst case index>` with respect to the |sym| field defined by
code:`<'SymIdentifier'>`
"""
#TODO: make it more general, e.g.:
# getTolDataIndex(q=None, p=None) to select an index based on quantile or
# probability value. But this should propbably make use of the TolData class!
wc_index = {}
for identifier in self._results:
wc_index[identifier] = np.argsort(self._results[identifier])[-1]
return wc_index
[docs] def start(self, samplesCount=None):
"""
Start tolerance analysis. Tolerance variables and results can be accessed via :attr:`variables`
and :attr:`results`. Check if :code:`<SymTol>.config.simControl.indexTimeout` and other :code:`simControl`
settings are defined properly!
Parameters
----------
samplesCount: int | None
Parameter can be used to override :code:`<SymTol>.config.samplesCount` value.
"""
logger = logging.getLogger(self.start.__name__)
logger.info('*** Starting tolerance analysis ***')
#print('\n' + '*'*30)
#tprint('starting tolerance analysis ...\n')
if isinstance(samplesCount, int):
self._config.samplesCount = samplesCount
logger.info(f"SymTol.config.samplesCount modified to {self._config.samplesCount}.")
#tprint(f"config.samplesCount modified to {self._config.samplesCount} ...")
logger.info(f"Evaluating {self._config.samplesCount} samples.")
#tprint(f"Evalutating {self._config.samplesCount} samples ...")
self.createSimProject()
if self._config.storeIndices:
# TODO: maybe better check this earlier (also done in self.startNominalSet())
self.__setStoreDirectory()
""" initialize all indices """
self._index_assigned_cnt = 0
if self._config.simProject.initializeIndices:
for i in range(self.simProject.getLength()):
self.simProject.setIndex(i)
self.setTolerancesVariables()
""" check if tolerance data should be pickled """
store_TolData = False
if self._config.simControl.tolDataPickleInterval > 0:
store_TolData = True
pickleTolDataEveryN = RunEveryN(self._config.simControl.tolDataPickleInterval, self.pickleTolData)
self.simProject.setIndex(0)
"""check if runSim == True, otherwise set it. runSim == False might lead to infinite loop,
since indices might not get status OK"""
if not self._config.simControl.runSim:
self._config.simControl.runSim = True
logger.warning("SymTol.config.simControl.runSim is set to True.")
#print("<SymTol>.config.simControl.runSim is set to True!")
"""initialize timeout list"""
self._index_runtime[:] = time.time()
ready = 0
debug_cnt = 0
ctrl_interval = 10 # sec.
ctrl_starttime = time.time()
# max_evals_debug = 30
while ready < self._config.samplesCount and not self.stop: # and debug_cnt < max_evals_debug: # self._config.simControl.runSim:
if self.finish:
self._config.samplesCount = self._index_assigned_cnt # prevent from creation of new individuals
logger.info(f"Simulation will stop after remaining {self._config.samplesCount-ready} samples are finished.")
#tprint(f"Simulation will stop after remaining {self._config.samplesCount-ready} samples are finished.")
if self.simProject.getError(): # or self.__has_index_timeout():
t_start = self._index_runtime[self.simProject.getIndex()]
logger.error(f"Index {self.simProject.getIndex()}, DS: {self.simProject.getDSId()} (time: {(time.time()-t_start)/60:.0f} min)")
#tprint(f"error in index {self.simProject.getIndex()}, DS: {self.simProject.getDSId()} (time: {(time.time()-t_start)/60:.0f} min)", level='ERROR')
jEventLogBuffer = list(self.simProject.javaObj.getEventLog().getBuffer()) # List of 'PyJObject's
error_list = [x.getDescription() + " --> " + list(x.getStack())[0].toString() for x in jEventLogBuffer if x.getType().toString() == "Error"]
# todo: maybe get whole stack, not only last message ...
for e in error_list:
logger.error(e)
"""decrease _index_assigned_cnt to get sampleCount samples with OK at the end!"""
self._index_assigned_cnt -= 1
if not self.finish:
"""regenerate True to cancle running job!"""
self.setTolerancesVariables(regenerate=True)
elif self.simProject.getIncomplete():
""" regenerate -> next index: done at the end!"""
pass
elif not self.__has_index_status('paused'):
"""finished with OK status and not paused!"""
self.getResultValues(sym_project=self.simProject)
self.getVariableValues(sym_project=self.simProject)
ready += 1
logger.info(f" :-) {ready} of {self._config.samplesCount} samples finished successfully.")
# tprint(f"{ready} of {self._config.samplesCount} samples finished successfully!")
if self._config.storeIndices:
self.pickleSimIndex(fname=os.path.join(self._config.storeIndicesDirectory, str(uuid.uuid4()) + '.p'))
if store_TolData:
pickleTolDataEveryN()
self.setTolerancesVariables()
if not SymJobDesc.isJobRunning(self.simProject) and self.__has_index_status('active'):
logger.info(f"Regenerating ... (Idx: {self.simProject.getIndex()}, DS: {self.simProject.getDSId()}, status: {self._index_status[self.simProject.getIndex()]}, running {self.__get_index_runtime() / 60:.0f} minutes)")
#tprint(f"regenerating ... (Idx: {self.simProject.getIndex()}, DS: {self.simProject.getDSId()}, status: {self._index_status[self.simProject.getIndex()]}, running {self.__get_index_runtime() / 60:.0f} minutes)")
self.simProject.regenerate(RunSim=self._config.simControl.runSim,
RemoteCompute=self._config.simControl.remoteCompute,
Optimize=self._config.simControl.optimize,
Timeout=self._config.simControl.jobTimeout*60) # Timeout
self.__reset_index_runtime() # at the moment, index timeout is not used to abort an index - just for information printing ...
self._next_index_cnt += 1 # see self.__set_next_index()
if ready == 0:
"""until one index is finished, check for/collect JobIDIdentifiers"""
self.__findJobIDIdentifier(self.simProject)
# print(self._JobIDIdentifier)
""" print information about running analysis/ interrupt analysis if necessary (controlFile) """
debug_cnt += 1
timeinfo = ""
if debug_cnt > self._config.simProject.indices*500: # with this limit definition every index is printed one after the other
if self.__has_index_status('active'):
"""e.g.: no timeout if index is paused, runtime makes no sense, ..."""
timeinfo = f", running {self.__get_index_runtime() / 60:.0f} minutes"
logger.info(f"Still working ... (Idx: {self.simProject.getIndex()}, DS: {self.simProject.getDSId()}, status: {self._index_status[self.simProject.getIndex()]}{timeinfo})")
#tprint(f"processing ... (Idx: {self.simProject.getIndex()}, DS: {self.simProject.getDSId()}, status: {self._index_status[self.simProject.getIndex()]}{timeinfo})")
debug_cnt = 0
""" check for modified control file in a certain time interval """
if time.time()-ctrl_starttime > ctrl_interval:
if self.ctrlFile.modified():
ctrls_d = self.ctrlFile.getControls()
self.stop = ctrls_d['stop']
self.finish = ctrls_d['finish']
if self.stop:
logger.warning(f"SymTol.stop set to {self.stop}")
if self.finish:
logger.warning(f"SymTol.finish set to {self.finish}")
# tprint(f"stop set to {self.stop}, finish set to {self.finish}\n")
ctrl_starttime = time.time()
"""set next index"""
self.__set_next_index()
if self.stop:
logger.info("Removing jobs from job manager ... this can take some time ...")
#tprint("removing jobs from job manager ... this can take some time ...")
#self._JobIDIdentifier = [] # reset value!
time.sleep(10)
self.__findJobIDIdentifier(self.simProject)
for idx in range(self.simProject.getLength()):
self.simProject.setIndex(idx)
logger.info(f"Check index {self.simProject.getIndex()} for open jobs ...")
#tprint(f"check index {self.simProject.getIndex()}")
time.sleep(0.5)
jobIDs = self.__getJobIds()
for jobID in jobIDs:
if jobID is not None:
"""can be None if index is paused --> cold also be used to check instead of checking for None"""
SymJobDesc.stop(jobID)
SymJobDesc.remove(jobID)
logger.info(f" +++ {ready:<10} samples calculated successfully +++")
logger.info(f" +++ {self.sampleCount:<10} samples available in total +++")
#tprint(f"\n\t\t+++ {ready} samples calculated successfully +++")
#tprint(f"\n\t\t+++ {self.sampleCount} samples available in total +++")
""" reset index cnt if new start() or restart()"""
self._next_index_cnt = 0
self.pickleTolData()
if self._config.simProject.cleanUp:
self.cleanUpProject()
[docs] def startNominalSet(self):
"""
Run the nominal |sym| model without tolerances applied to the respective parameters. If the status of
index 0 of the base Project is `OK` (i.e., not `incomplete` nor `error`), values from this index will be used
as nominal set. Otherwise, index 0 of the simulation project (:attr:<SymTol>.simProject`) is set to
nominal values and calculated!
Nominal values are stored at index 0 of :attr:results` and :attr:`variables` and pickled |sym| tree is stored
with filename :file:`0_nominal.p` in the common store directory (:file:`<>.tol.files`).
"""
logger = logging.getLogger(self.startNominalSet.__name__)
if ('nominalSet' in self._experiments.keys()) and self._experiments['nominalSet']:
logger.warning('SymTol.startNominalSet() skipped, because nominal set has already been added to the experiments.')
#tprint('SymTol.startNominalSet() skipped, because nominal set has already been added to the experiments!')
else:
if self._config.storeIndices:
# TODO: maybe better check this earlier (also done in self.start())
self.__setStoreDirectory()
self.baseProject.setIndex(0)
if not (self.baseProject.getIncomplete() or self.baseProject.getError()):
""" index 0 base project is OK, use this values """
self.getResultValues(sym_project=self.baseProject, insert_pos=0)
self.getVariableValues(sym_project=self.baseProject, insert_pos=0)
logger.info("No recalculation needed - nominal results added from baseProject.")
#tprint("no recalculation needed - nominal results added from baseProject!")
"""nominal set added to data - for further reference"""
self._experiments['nominalSet'] = True
if self._config.storeIndices:
pickle.dump(self.baseProject.toDict(detailed=True),
open(os.path.join(self._config.storeIndicesDirectory, '0_nominal' + '.p'), 'wb'))
else:
self.createSimProject()
""" _config idx 0 of simProject to base values and regenerate """
self.simProject.setIndex(0)
self.setBaseValues()
debug_cnt = 0
while self.simProject.getIncomplete() and not(self.simProject.getError() or self.__has_index_timeout()):
debug_cnt += 1
if debug_cnt > 50000:
logger.info("Still working to recalculate nominal set ...")
#tprint("running ...")
debug_cnt = 0
if not SymJobDesc.isJobRunning(self.simProject):
self.simProject.regenerate(RunSim=self._config.simControl.runSim,
RemoteCompute=self._config.simControl.remoteCompute,
Optimize=self._config.simControl.optimize,
Timeout=self._config.simControl.jobTimeout * 60)
logger.info("Regenerate nominal dataset.")
# tprint(f"regenerate nominal dataset {self.simProject.getIndex()}")
if not self.simProject.getError():
self.getResultValues(insert_pos=0)
self.getVariableValues(insert_pos=0)
logger.info(":-) Nominal set calculated successfully.")
#tprint("nominal results calculated successfully!")
"""nominal set added to data - for further reference"""
self._experiments['nominalSet'] = True
if self._config.storeIndicesDirectory is not None:
pickle.dump(self.simProject.toDict(detailed=True),
open(os.path.join(self._config.storeIndicesDirectory, '0_nominal' + '.p'), 'wb'))
else:
logger.error("Nominal dataset finished with ERROR!")
jEventLogBuffer = list(self.simProject.javaObj.getEventLog().getBuffer()) # List of 'PyJObject's
error_list = [x.getDescription() + " --> " + list(x.getStack())[0].toString() for x in jEventLogBuffer if x.getType().toString() == "Error"]
# todo: maybe get whole stack, not only last message ...
for e in error_list:
logger.error(e)
#tprint("Nominal dataset finished with ERROR!")
if self._config.simProject.cleanUp:
self.cleanUpProject()
[docs] def setTolerancesVariables(self, regenerate=False):
"""
Set values to the tolerance affected variables as defined in the :guibutton:`Comment:________` area of the
corresponding field in the |sym| tree using the keyword :code:`@Tolerance{"var":{}}`.
The general structure of the comment-string to define a tolerance affected variable is:
.. code-block:: none
@Tolerance{"var":{"<distribution>":{["<setting>":<value>]}}}
Available distributions are:
====================== ================================================= ====================================
"<distribution>" ["<setting>":<value>] Comment
====================== ================================================= ====================================
:code:`"uniform"` :code:`"lower_tol":<float>,"upper_tol":<float>,` uniform distribution with lower and
upper boundary; :code:`lower_tol`
has to be defined with negative sign.
:code:`"norm"` :code:`"sigma":<float>` normal (Gauss) distribution with
standard deviation :code:`sigma`
====================== ================================================= ====================================
For example, a parameter that has a tolerance range of [-0.1,0.15] with uniform distribution,
the comment (json) string looks:
.. code-block:: none
@Tolerance{"var":{"uniform":{"lower_tol":-0.1, "upper_tol":0.15}}}
.. note::
As nominal value, the field value (Data) of the corresponding |sym| parameter is used.
Parameters
----------
regenerate: bool
if set to True, |sym| project is regenerated after assigning new toelrance parameters, default False
"""
logger = logging.getLogger(self.setTolerancesVariables.__name__)
if self._index_assigned_cnt < self._config.samplesCount:
# todo: if constraints are implemented, only use proper variables ... e.g.:
# for tol in [self.AtToleranceData[x] for x in range(len(self.AtToleranceData)) if self.AtToleranceData[x]['identifier'] not in self._constraints:
for tol in self.AtToleranceData:
if 'var' in tol.keys():
# its a statistic variable:
distribution = list(tol['var'].keys())[0].lower()
"""
uniform, normal
"""
settings = tol['var'][distribution]
identifier = tol['identifier']
base_value = tol['base_value']
if 'uniform'.startswith(distribution):
lower_tol = settings['lower_tol']
upper_tol = settings['upper_tol']
delta_tol = upper_tol - lower_tol # lower tol with neg. sign defined!
delta = np.random.random(np.array(base_value).shape) * delta_tol + lower_tol
data = base_value + delta
elif 'normal'.startswith(distribution):
sigma = settings['sigma']
data = np.random.normal(base_value, sigma)
else:
data = None
logger.error(f"Distribution '{distribution}' not implemented!")
#print(f"distribution '{distribution}' not implemented!")
self.simProject.getObject(identifier).setData(data)
logger.info(f"Tolerance variables set to index {self.simProject.getIndex()} (count: {self._index_assigned_cnt})")
"""refresh to update status!"""
self.simProject.refresh()
if regenerate:
logger.info(f"Regenerating index {self.simProject.getIndex()} of simulation project.")
self.simProject.regenerate(RunSim=self._config.simControl.runSim,
RemoteCompute=self._config.simControl.remoteCompute,
Optimize=self._config.simControl.optimize,
Timeout=self._config.simControl.jobTimeout * 60) # Timeout
"""set status of current index"""
self.__set_index_status('active')
"""increase _index_assigned_cnt"""
self._index_assigned_cnt += 1
"""updating start-time of newly assigned index"""
self.__reset_index_runtime()
# tprint(f"tolerance variables set to index {self.simProject.getIndex()} (count: {self._index_assigned_cnt})")
else:
"""set index to paused"""
if self.__has_index_status('active'):
logger.info(f"Tolerance variables not set to index {self.simProject.getIndex()}, because enough samples are in the job queue!")
#tprint(f"tolerance variables not set to index {self.simProject.getIndex()}, index paused because enough samples in job queue!")
self.__set_index_status('paused')
[docs] def setBaseValues(self):
"""
sets the nominal values to the actual index of the (temporary) |sym| project used for tolerance analysis
"""
for tol in self.AtToleranceData:
if 'var' in tol.keys():
# its a statistic variable:
identifier = tol['identifier']
base_value = tol['base_value']
self.simProject.getObject(identifier).setData(base_value)
self.simProject.refresh()
"""updating start-time of newly assigned index"""
self._index_runtime[self.simProject.getIndex()] = time.time()
[docs] def getResultValues(self, sym_project=None, insert_pos=None):
"""
Collect all tolerance related results from the current index of the |sym| - project `sym_project` and
inserts these at position `insert_pos` to the :attr:`results`.
Definition of `sym_project` and `insert_pos` is mainly to add nominal results at position (:,:,0).
Parameters
----------
sym_project : :class:`symspace.wrapper.symwrapper.SymProject`
|sym| project where results are collected from, default is simulation project (attr:`<SymTol>.simProject`)
insert_pos : int
position (of axis 2/ 3rd dimension) in :attr:`results` where values are inserted, default is appending
at the end.
"""
logger = logging.getLogger(self.getResultValues.__name__)
if sym_project is None:
sym_project = self.simProject
logger.info(f"Collecting results from SyM-Project {sym_project.getFile()}: index {sym_project.getIndex()}")
#tprint(f"collecting results from SyM-Project {sym_project.getFile()}: index {sym_project.getIndex()}")
for tol in self.AtToleranceData:
if 'res' in tol.keys():
""" it is a statistical result """
identifier = tol['identifier']
if isinstance(insert_pos, int):
"""mainly to insert nominal value on first position"""
self._results.setdefault(identifier, []).insert(insert_pos,
sym_project.getObject(identifier).getData())
else:
self._results.setdefault(identifier, []).append(sym_project.getObject(identifier).getData())
# self.__set_index_status('OK')
[docs] def getVariableValues(self, sym_project=None, insert_pos=None):
"""
Collect all tolerance related variables from the current index of the |sym| - project `sym_project` and
inserts these at position `insert_pos` to the :attr:`variables`.
Definition of `sym_project` and `insert_pos` is mainly to add nominal results at position (:,:,0).
Parameters
----------
sym_project : :class:`symspace.wrapper.symwrapper.SymProject`
|sym| project where results are collected from, default is simulation project (attr:`<SymTol>.simProject`)
insert_pos : int
position (of axis 2/ 3rd dimension) in :attr:`variables` where values are inserted, default is appending
at the end.
"""
logger = logging.getLogger(self.getVariableValues.__name__)
if sym_project is None:
sym_project = self.simProject
logger.info(f"Collecting variables from SyM-Project {sym_project.getFile()}: index {sym_project.getIndex()}")
#tprint(f"collecting variables from SyM-Project {sym_project.getFile()}: index {sym_project.getIndex()}")
for tol in self.AtToleranceData:
if 'var' in tol.keys():
""" it is a statistical variable """
identifier = tol['identifier']
if isinstance(insert_pos, int):
"""mainly to insert nominal value on first position"""
self._variables.setdefault(identifier, []).insert(insert_pos,
sym_project.getObject(identifier).getData())
else:
self._variables.setdefault(identifier, []).append(sym_project.getObject(identifier).getData())
[docs] def pickleSimIndex(self, fname=None):
"""
Automatically pickle the whole |sym| tree of the current index of the simulation project.
If field data is array, struct, ... it is stored as serialized pickle string and has to be loaded using
`pickle.loads(base64.b64decode(<serialized string>))`
Parameters
----------
fname : str
full path of the pickle file, if not defined, project path and name with time code is used!
"""
logger = logging.getLogger(self.pickleSimIndex.__name__)
if self.simProject is not None:
if fname is None:
fname = os.path.join(self.simProject.getDirectory(),
self.simProject.getFile().rsplit('.', 1)[0] + time.strftime('_%m%d%H%M%S') + '.p')
pickle.dump(self.simProject.toDict(detailed=True), open(fname, 'wb'))
logger.info(f"Index {self.simProject.getIndex()} of {self.simProject.getFile()} pickled to {fname}")
# tprint(f"Index {self.simProject.getIndex()} of {self.simProject.getFile()} pickled to {fname}")
else:
logger.warning(f"Simulation project not available! See {self.__class__.__name__}.{self.createSimProject.__name__}().")
#tprint("Simulation project not available! See SymTol.createSimProject().")
[docs] def pickleTolData(self, fname=None):
"""
Automatically pickle tolerance-affected data, i.e., :attr:`results` and :attr:`variables`. Default filename
is the name of the |sym|-project with :file:`_TolData` appended (:file:`<SyM_project_name>_TolData.p`).
Parameters
----------
fname : str
full path of the pickle file, if not defined, project path and name with :file:`_TolData` suffix is used!
Returns
-------
dict :
dictionary with keys :code:`'results'` and :code:`'variables'`, holding the data returned by :attr:`results`
and :attr:`variables` respectively.
"""
logger = logging.getLogger(self.pickleTolData.__name__)
if self.sampleCount > 0:
tolData = dict(results=self.results, variables=self.variables, config=self.config, _experiments=self._experiments)
# the _experiments part is more temporary for now - not sure if this makes sense ...
# todo: config might be better to serialize as dict, otherwise the settings classes are used to load the pickle ....
if fname is None:
fname = self.__getTolDataDefaultFileName()
# self.tmp_cnt += 1
pickle.dump(tolData, open(fname, 'wb'))
logger.info(f"Tolerance related data ({self.sampleCount} samples) written to {fname}")
# tprint(f"Tolerance related data ({self.sampleCount} samples) written to {fname}")
else:
logger.warning("Nothing to pickle.")
# tprint("SymTol.pickleTolData(): nothing to pickle!")
[docs] def cleanUpProject(self):
"""
Cleans up the |sym| project which is cereated for temporary use of tolerance analysis. If original |sym|
project is used, all indices are removed (except first one)
"""
"""
Warning: when using baseProject, all indices are removed except 1st one!
"""
logger = logging.getLogger(self.cleanUpProject.__name__)
logger.info("Cleaning up temporary simulation project files ...")
#print('\n'+'*'*30)
#tprint("cleaning up temporary simulation project files ...")
if self._config.simProject.clone:
self.simProject.close()
"""remove temporary SyM-project folder"""
try:
#shutil.rmtree(os.path.join(self.simProject.getDirectory(), self.simProject.getFile() + '.Files'),
# ignore_errors=True)
shutil.rmtree(os.path.join(self.simProject.getDirectory(), self.simProject.getFile() + '.Files'))
logger.info("Simuation project folder removed successfully")
except IOError as e:
errno, strerror = e.args
logger.error(f"I/O error({errno}): {strerror}")
except:
logger.error('Unexpected error when removing simulation project folder.')
"""remove temporary SyM-project"""
try:
os.remove(os.path.join(self.simProject.getDirectory(), self.simProject.getFile()))
logger.info("Simulation project removed successfully")
except IOError as e:
errno, strerror = e.args
logger.error(f"I/O error({errno}): {strerror}")
except:
logger.error('Unexpected error when removing simulation project.')
self.simProject = None
"""remove control file"""
if isinstance(self.ctrlFile, ControlFile):
self.ctrlFile.remove()
else:
# Warning: all indices are removed except 1st one ....!
# input() command to get user feedback hangs SyMSpace console!!!
for i in range(self.baseProject.getLength() - 1):
self.baseProject.reduce()
self.setBaseValues()
self.baseProject.regenerate()
self.baseProject.save()
logger.info("Base project reverted to initial state.")
@property
def config(self):
"""
Access all configuration settings available for tolerance analysis.
Returns
-------
:class:`symtolerance2.symtolsettings.SymTolSettings`:
Data class holding all configuration settings. For more information about available settings see the
:class:`symtolerance2.symtolsettings.SymTolSettings` documentation.
"""
return self._config
@property
def results(self):
"""
Get all results of the tolerance analysis, i.e., values of all |sym| parameters which are marked with
:code:`@Tolerance{"res":{}}` in the parameters' :guibutton:`Comments:________` area. The results of a certain
parameter can be accessed from this results dictionary via its |sym| identifier (key). The data itself is
stored as 3-dimensional numpy.ndarray, where the different tolerance affected samples are stored in the 3rd
dimension (axis=2). Values of the nominal dataset are, if available, at index (:,:,0). The variable set leading
to a certain result (certain index in 3rd dimension) can be accessed via :func:`variables`.
.. note::
So far, only 2-dimensional data :math:`\in \mathbb{R}^{2}` can be processed as results of a tolerance analysis
Returns
-------
dict
Dictionary where keys correspond to |sym| identifier of the result fields and values are of type
numpy.ndarray, holding tolerance information in the third axes. Values of the nominal dataset are,
if available, at index (:,:,0).
"""
""" Internal:
In this function self._results is converted to dictionary with 3dimensional arrays which hold tolerance
information in 3rd axis.
"""
logger = logging.getLogger('results')
res = {}
for k, v in self._results.items():
v_np = np.array(v)
if len(v_np.shape) == 1:
# scalar objective
res[k] = np.empty((1, 1, v_np.shape[0]))
res[k][0, 0, :] = v_np
elif len(v_np.shape) == 3:
# array objective
res[k] = np.swapaxes(np.swapaxes(np.array(v), 0, 1), 1, 2)
else:
# TODO: has to be checked
logger.error(f'Dimension of {k} is not implemented to be a result so far -> debug and extend ({self.__class__.__name__}.result()).')
#tprint(
# f'return dimension of {k} is not implemented so far -> debug and extend ({self.__class__.__name__}.result())!')
res[k] = f'Dimension of {k} is not implemented to be a result so far -> debug and extend ({self.__class__.__name__}.result()).'
return res
@property
def variables(self):
"""
Get all variables of the tolerance analysis, i.e., values of all |sym| parameters which are marked with
:code:`@Tolerance{"var":{}}` in the parameters' :guibutton:`Comments:________` area. The values of a certain
parameter can be accessed from this results dictionary via its |sym| identifier (key). The data itself is
stored as 3-dimensional numpy.ndarray, where the tolerance affected variable values are stored in the 3rd
dimension (axis=2). The nominal values are, if available, at index (:,:,0). The results caused by a certain
tolerance setting (certain index in 3rd dimension) can be accessed via :func:`results`.
.. note::
So far, only 2-dimensional data :math:`\in \mathbb{R}^{2}` can be processed as variables of a tolerance analysis.
Returns
-------
dict
Dictionary where keys correspond to |sym| identifier of the variable fields and values are of type
numpy.ndarray, holding tolerance information in the third axes. Values of the nominal dataset are,
if available, at index (:,:,0).
"""
""" Internal:
In this function self._results is converted to dictionary with 3dimensional arrays which hold tolerance
information in 3rd axis.
"""
logger = logging.getLogger('variables')
var = {}
for k, v in self._variables.items():
v_np = np.array(v)
if len(v_np.shape) == 1:
# scalar objective
var[k] = np.empty((1, 1, v_np.shape[0]))
var[k][0, 0, :] = v_np
elif len(v_np.shape) == 3:
# array objective
var[k] = np.swapaxes(np.swapaxes(np.array(v), 0, 1), 1, 2)
else:
# TODO: has to be checked
logger.error(
f'Dimension of {k} is not implemented to be a variable so far -> debug and extend ({self.__class__.__name__}.variables()).')
# tprint(
# f'return dimension of {k} is not implemented so far -> debug and extend ({self.__class__.__name__}.variables())!')
var[k] = f'Dimension of {k} is not implemented to be a variable so far -> debug and extend ({self.__class__.__name__}.variables()).'
return var
@property
def sampleCount(self):
"""
Get the number of tolerance-affected samples available in the project. If the nominal problem is calculated
(:func:`startNominalSet`), this is included in the sample count value.
Returns
-------
int :
Number of samples available in the project
"""
if len(self.results) == 0:
"""no samples available"""
return 0
else:
return self.results[list(self.results.keys())[0]].shape[2]
# def updateResultStatistics(self):
# # TODO!!!
# pass
#
# def showStatistics(self):
# # TODO!!!
# pass
#
# def plotStatistics(self):
# # TODO!!!
# pass
def __findJobIDIdentifier(self, sym_element):
"""
Recursively searches the tree for functions which might be distributed via the job manager
Parameters
----------
sym_element : :class:`symspace.wrapper.symwrapper.SymProject` | :class:`symspace.wrapper.symwrapper.SymContainer` | :class:`symspace.wrapper.symwrapper.SymCollector` | :class:`symspace.wrapper.symwrapper.SymFunction` | :class:`symspace.wrapper.symwrapper.SymField` | :class:`symspace.wrapper.symwrapper.SymFormula`
|sym| tree element which serves as root for the recursive search.
"""
logger = logging.getLogger(self.__findJobIDIdentifier.__name__)
if isinstance(sym_element, (sw.SymProject, sw.SymContainer, sw.SymCollector,
sw.SymFunction)): # maybe more, every element which has sub-params (.getParam())
if isinstance(sym_element, sw.SymFunction):
# only function can be distributed via condor ?!
id = sym_element.javaObj.getTmpDSData("JobID")
if (not (id is None)) & (not (sym_element.getIdentifier() in self._JobIDIdentifier)):
self._JobIDIdentifier.append(sym_element.getIdentifier())
logger.debug(f"jobs created by: {sym_element.getIdentifier()}")
#tprint(f"jobs created by: {sym_element.getIdentifier()}")
for elem in sym_element.getParam():
self.__findJobIDIdentifier(elem)
def __getJobIds(self):
"""
Returns
-------
list :
list of all jobIDs of the current index of :code:`self.simProject`
"""
logger = logging.getLogger(self.__getJobIds.__name__)
id_lst = []
for identifier in self._JobIDIdentifier:
id_a = np.array(self.simProject.getObject(identifier).javaObj.getTmpDSData("JobID")).flatten()
#if len(id) != 1:
# tprint(f"{len(id)} JobIDs available for function {identifier}! Only first one is used!")
#id_lst.append(id[0])
logger.debug(f"{len(id_a)} JobIDs available for function {identifier}")
id_lst += id_a.tolist()
return id_lst
def __getTolDataDefaultFileName(self):
"""
get the default filename to store :attr:`variables` and :attr:`results` as pickle file
Returns
-------
str
full path with default filename
"""
return self.__getBaseFileName() + '_TolData.p'
def __getBaseFileName(self):
return os.path.join(self.baseProject.getDirectory(), self.baseProject.getFile().rsplit('.', 1)[0])
def __updateSimProjectIndices(self):
"""
Updates number of indices of the |sym| project used for tolerance analysis.
"""
logger = logging.getLogger(self.__updateSimProjectIndices.__name__)
indices = self._config.simProject.indices
if isinstance(self.simProject, sw.SymProject):
if self.simProject.getLength() != indices:
# TODO: implement some intelligence, i.e.: check how many are available ....
for i in range(self.simProject.getLength() - 1):
self.simProject.reduce()
for i in range(indices - 1):
self.simProject.duplicate()
logger.info("SimProject indices updated!")
#tprint("SimProject indices updated!")
else:
logger.info("Number of indices of SimProject ok.")
#tprint("Number of indices of SimProject ok.")
else:
# should not happen ... :-)
logger.error(f"Simulation project not assigned. Run {self.__class__.__name__}.{self.createSimProject.__name__}().")
#tprint("self.__updateSimProjectIndices: SimProject not assigned!")
"""set tracking lists to proper length"""
if len(self._index_runtime) != indices:
self._index_runtime = np.array([None] * indices)
self._index_status = ['undefined']*indices
def __setStoreDirectory(self):
"""
Sets and, if necessary, creates the directory where analysed index files are stored.
"""
logger = logging.getLogger(self.__setStoreDirectory.__name__)
if (self._config.storeIndicesDirectory is None) or (self._config.storeIndicesDirectory == 'default'):
self._config.storeIndicesDirectory = self._config.baseProjectFile.rsplit('.mop', 1)[0] + '.tol.Files'
if not os.path.isdir(self._config.storeIndicesDirectory):
try:
os.mkdir(self._config.storeIndicesDirectory)
logger.info(f"Directory {self._config.storeIndicesDirectory} created.")
# print(f"directory {self._config.storeIndicesDirectory} created!")
logger.info(f"Tolerance data will be stored to:\n\t{self._config.storeIndicesDirectory}")
except IOError as e:
errno, strerror = e.args
logger.error(f"I/O error({errno}): {strerror}")
except:
logger.error(f'Unexpected error when creating {self._config.storeIndicesDirectory} directory.')
#print(f"Tolerance data will be stored to:\n\t{self._config.storeIndicesDirectory}")
def __set_next_index(self):
"""
Sets the next index of the simulation project with some intelligence (like SyM - optimizer)
Returns
-------
None
"""
logger = logging.getLogger(self.__set_next_index.__name__)
if self._next_index_cnt < self._config.simControl.batch:
self.simProject.setIndex(np.mod(self.simProject.getIndex() + 1, self.simProject.getLength()))
# self._next_index_cnt += 1 only _config when regenerating and index!
else:
# in start() it is continuously checked for JobIDIdentifiers, and if no jobID available
# first time in here, it is assumed that there is no CondorJob -> set batch to indices!
if (len(self._JobIDIdentifier) == 0) & (self._config.simControl.batch < self._config.simProject.indices):
# no jobmanager jobs and batch < simProject indices -> set batch to indizes:
self._config.simControl.batch = self._config.simProject.indices
str_lst = ["Modifying batchsize.",
"Batchsize for simulation control",
"is smaller than number of indices of the",
"simulation project and no jobmanager jobs",
"detected.",
"SymTol.config.simControl.batch is set to SymTol.config.simProject.indices."]
logger.warning('\n\t\t\t'.join(str_lst))
#tprint('\n\t\t\t'.join(str_lst))
else:
self._next_index_cnt = 0
self.simProject.setIndex(0)
def __set_index_status(self, status):
"""
set :attr:`self._index_status` for the current index. `status` can be: :code:`'active','paused'`
:attr:`self._index_status` is used for simulation control, i.e., to check if an index has to be regenerated
Parameters
----------
status: str
`status` can be: :code:`'active','paused'`
"""
logger = logging.getLogger(self.__set_index_status.__name__)
mapping = dict(a='active', p='paused')
stat = status.lower()[0]
old = self._index_status[self.simProject.getIndex()]
if stat in mapping.keys():
self._index_status[self.simProject.getIndex()] = stat
if stat != old:
logger.debug(f"status of index {self.simProject.getIndex()} changed to {mapping[stat]}")
#tprint(f"status of index {self.simProject.getIndex()} changed to {mapping[stat]}")
# print(self._index_status)
else:
tprint(f"status: '{status}' not available!")
self._index_status[self.simProject.getIndex()] = 'undefined'
def __has_index_status(self, status):
"""
returns true, if the current index status corresponds with one of `status`
Parameters
----------
status : tuple(str,)
tuple of status values to be checked for
Returns
-------
bool :
True if current index status is in `status` list
"""
if isinstance(status, str):
status = (status,)
status = tuple(x.lower()[0] for x in status)
stat = self._index_status[self.simProject.getIndex()]
return stat in status
def __get_num_indices_with_status(self, status):
"""
-> NOT USED AT THE MOMENT
get number of indices which have the defined status
Parameters
----------
status : tuple(str,)
tuple of status values to be checked for
Returns
-------
int :
number of indices which have the defined status
"""
if isinstance(status, str):
status = (status,)
status = tuple(x.lower()[0] for x in status)
return np.sum([1 if stat in status else 0 for stat in self._index_status])
def __has_index_timeout(self):
"""return True if index has timeout
Note: this is not Job-Timeout!
"""
timeout = False
if self.__has_index_status('active'):
"""no timeout if index is paused!"""
t_start = self._index_runtime[self.simProject.getIndex()]
if (time.time()-t_start)/60 > self._config.simControl.indexTimeout:
timeout = True
return timeout
def __get_index_runtime(self):
"""returns runtime of an index - its not Job-runtime on Condor!"""
t_start = self._index_runtime[self.simProject.getIndex()]
return time.time() - t_start
def __reset_index_runtime(self):
self._index_runtime[self.simProject.getIndex()] = time.time()
def __str__(self):
header_str = f"{self.__class__.__name__}({self.baseProject.getFile()})\n"
#ret_str = '#'*len(header_str)+'\n'
ret_str = '#' * 35 + '\n'
ret_str += header_str
#ret_str += '-' * len(header_str) + '\n\n'
ret_str += '#' * 35 + '\n\n'
ret_str += 'The following tolerance related fields are found:\n'
ret_str += '=' * 35 + '\n'
ret_str += '\n--------------------------------\n'.join(['\n'.join([f"{k}:\t{v}" for k, v in d.items()]) for d in self.AtToleranceData])
ret_str += '\n\n' + '*'*5 + f' {self.sampleCount} tolerance affected designs evaluated so far ' + '*'*5
return ret_str
def __repr__(self):
return f"{self.__class__.__name__}({self._config.baseProjectFile})"
# def XloadFromTolFiles(self):
# """
# Load variables and results from :file:`.tol.files` directory if this exists. This can also be used to
# load and investigate data without creation of a temporary |sym| - project. The loaded data is stored and can
# be accessed via the :attr:`results` and :attr:`variables` attributes.
#
# .. warning::
# This is not yet implemented!!!
#
# """
# # TODO: load variables and results from .tol.files directory if this exists an contains data.
# # Might also be used to load and investigate data without creation of a temporary simProject
#
# print('Sorry, this function is not yet implemented!')
# pass
# def XstoreTolProject(self):
# """
# Store the tolerance-analysis project: TBD
#
# .. warning::
# This is not yet implemented!!!
#
# """
# # TODO: zip simProjecta, settings, and tol.files
#
# print('Sorry, this function is not yet implemented!')
# pass
# @classmethod
# def XloadTolProject(cls, fname):
# """
# Load a tolerance-analysis project: TBD
#
# .. warning::
# This is not yet implemented!!!
#
# """
# # TODO: load simProject, settings, tol.files data
# # also add in self.__init__()
# print('Sorry, this function is not yet implemented!')
# pass
def init_test():
X = SymTol(CurProject)
X.config.samplesCount = 5
X.config.simProject.indices = 7
X.config.simControl.indexTimeout = 1
#X.config.simControl.batch = 3
#X.config.storeIndices = True
X.loadTolData()
return X