"""
.. inheritance-diagram:: pyopus.simulator.base
:parts: 1
**Base class for simulator objects (PyOPUS subsystem name: SI)**
A **system** is a physical object (e.g. a circuit) whose characteristics and
responses are evaluated (simulated) by a simulator.
Simulators are divided into two groups: **batch mode simulators** and
**interactive simulators**. An interactive simulator presents the user with a
prompt where commands of a simulator-specific command language can be entered
to control the simulator (load input file, perform simulation, save result
file, ...).
Interactive simulators can be used as batch simulators if they are presented
with a script written in the simulator control language at their standard
input.
To tell a batch mode simulator what to simulate one must provide a
**job list**. A job list is a list of jobs where every job specifies one
simulation.
A job is a dictionary with the following members:
* ``name`` - the name of the job
* ``definitions`` - a list of system description modules that constitute the
system to be simulated
* ``params`` - a dictionary with the system parameters used for this job.
Keys are parameter names while values are parameter values
* ``options`` - a dictionary with the simulator options used for this job.
Keys are option names while values are option values
* ``variables`` - a dictionary containing the Python variables that are
available during the evaluation of ``saves`` and ``command``.
* ``saves`` - a list of strings giving Python expressions that evaluate to
lists of save directives specifying what simulated quantities should be
stored in simulator output
* ``command`` - a string giving a Python expression that evaluates to the
simulator command for the analysis performed by this job
The order in which jobs are given may not be optimal for fastest simulation
because it can require an excessive number of system parameter/system
description changes. The process of **job optimization** reorders and groups
the jobs into **job groups** in such manner that the number of these changes
is minimized. the sequence of job groups is called **optimized job sequence**
Optimal job ordering and grouping as well as the way individual groups of jobs
are handled in simulation depends on the simulator.
A batch simulator offers the capability to run the jobs in one job group at a
time. For every job group the simulator is started, jobs simulated, and the
simulator shut down. Every time a job group is run the user can provide a
dictionary of *inputParameters* that complement the parameters defined in the
``params`` dictionaries of the jobs. If both the *inputParameters* and the
``params`` dictionary of a job list the value of the same parameter, the value
in the ``params`` dictionary takes precedence over the value defined in the
*inputParameters* dictionary. After the simulation in finished the results can
be retrieved from the simulator object.
Interactive simulators offer the user the possibility to send commands to the
simulator. After the execution of every command the output produced by the
simulator is returned. Collecting simulator results is left over to the user.
The simulator can be shut down and restarted whenever the user decides.
"""
# Abstract siumulator interface
from ..misc.identify import locationID
from ..misc.debug import DbgMsgOut, DbgMsg
from .. import PyOpusError
from os import remove, listdir
import numpy as np
from ..evaluator import measure as m
__all__ = [ 'Simulator', 'SimulationResults', 'BlankSimulationResults' ]
# TODO: stop changing input structures by adding missing defaults
[docs]class Simulator(object):
"""
Base class for simulator classes.
Every simulator is equipped with a *simulatorID* unique for every simulator
in every Python process on every host. The format of the simulatorID is
``<locationID>_simulatorObjectNumber``
``<locationID>`` is generated by the :func:`pyopus.misc.locationID`
function. It containes the information on the hosts and the task within
which the simulator object was created.
A different simulator object number is assigned to every simulator object
of a given type in one Python interpreter.
All intermediate files resulting from the actions performed by a simulator
object have their name prefixed by *simulatorID*.
Every simulator can hold result groups from several jobs. One of these
groups is the active result group.
The *binary* argument gives the full path to the simulator executable.
*args* is a list of additional command line parameters passed to the
simulator executable at simulator startup.
If *debug* is greater than 0 debug messages are printed to standard output.
"""
# Class variable that counts simulator instances
instanceNumber=0
# Define instance variables and initialize instance
def __init__(self, binary, args=[], debug=0):
# Simulator binary and command line arguments
self.binary=binary
self.cmdline=args
# Debug mode fkag
self.debug=debug
# Job list and sequence
self.jobList=None
self.jobSequence=None
# Input parameters dictionary
self.inputParameters={}
# TODO: remove
# Results of the simulation
self.results={}
self.activeResult=None
# jobGroup cleanup pattern list
self.jobGroupCleanupPatterns={}
self.generateSimulatorID()
[docs] def generateSimulatorID(self):
"""
Generates a new ID for simulator.
"""
# Build a unique ID for this simulator
self.simulatorID=locationID()+"_"+str(Simulator.instanceNumber)
# Increase simulator instance number
Simulator.instanceNumber+=1
[docs] @classmethod
def findSimulator(cls):
"""
Finds the simulator. Returns ``None`` if simulator is not found.
"""
return None
[docs] def cleanup(self):
"""
Removes all intermediate files created as a result of actions performed
by this simulator object (files with filenames that have *simulatorID*
for prefix).
"""
# Get directory
dirEntries=listdir('.')
# Find all directory entries starting with simulatorID and delete them
for entry in dirEntries:
if entry.find(self.simulatorID)==0:
if self.debug:
DbgMsgOut("SI", "Removing "+entry)
try:
remove(entry)
except:
pass
#
# For interactive simulators
#
[docs] def simulatorRunning(self):
"""
Returns ``True`` if the interactive simulator is running.
"""
return False
[docs] def startSimulator(self):
"""
Starts the interactive simulator if it is not already running.
"""
pass
[docs] def stopSimulator(self):
"""
Stops a running interactive simulator.
This is done by politely asking the simulator to exit.
"""
pass
#
# Job list and job list optimization
#
[docs] def setJobList(self, jobList, optimize=True):
"""
Sets *jobList* to be the job list for batch simulation. If the
``options``, ``params``, ``saves``, or ``variables`` member is
missing in any of the jobs, an empty dictionary/list is added to
that job.
The job list is marked as fresh meaning that a new set of simulator
input files needs to be created. Files are created the first time a
job group is run.
If *optimize* is ``True`` the optimized job sequence is computed by
calling the :meth:`optimizedJobSequence` method. If *optimize* is
``True`` an unoptimized job sequence is produced by calling the
:meth:`unoptimizedJobSequence` method.
"""
# Add missing members
for job in jobList:
if 'options' not in job:
job['options']={}
if 'params' not in job:
job['params']={}
if 'saves' not in job:
job['saves']=[]
if 'variables' not in job:
job['variables']={}
# Store it
self.jobList=jobList
# Optimize if required
if optimize:
self.jobSequence=self.optimizedJobSequence()
else:
self.jobSequence=self.unoptimizedJobSequence()
# Converter from jobIndex to jobGroupIndex
self.jobGroupIndex={}
for jobGroupNdx in range(len(self.jobSequence)):
jobGroup=self.jobSequence[jobGroupNdx]
for jobIndex in jobGroup:
self.jobGroupIndex[jobIndex]=jobGroupNdx
[docs] def unoptimizedJobSequence(self):
"""
Returns the unoptimized job sequence. A job sequence is a list of
lists containing indices in the job list passed to the
:meth:`setJobList` method. Every inner list corresponds to one job
group.
"""
raise PyOpusError(DbgMsg("SI", "Job list optimization procedure is not defined."))
[docs] def optimizedJobSequence(self):
"""
Returns the optimized job sequence. Usually the order of job groups
and jobs within them is identical to the order of jobs in the job list
set with the :meth:`setJobList` method.
"""
raise PyOpusError(DbgMsg("SI", "The procedure for obtaining the optimized job list is not defined."))
[docs] def jobGroupCount(self):
"""
Returns the number of job groups.
"""
raise PyOpusError(DbgMsg("SI", "The method is not defined."))
[docs] def jobGroup(self, i):
"""
Returns a structure describing the *i*-th job group.
"""
raise PyOpusError(DbgMsg("SI", "The method is not defined."))
#
# Batch simulation
#
[docs] def runJobGroup(self, i):
"""
Runs the jobs in the *i*-th job group.
Returns a tuple of the form (*jobIndices*, *status*) where *jobIndices*
is a list of indices corresponding to the jobs that ran. *status* is
``True`` if everything is OK.
"""
raise PyOpusError(DbgMsg("SI", "The method is not defined."))
[docs] def cleanupResults(self, i):
"""
Removes all result files that were produced during the simulation of
the *i*-th job group. Simulator input files are left untouched.
"""
# The derived class has to override this method
raise PyOpusError(DbgMsg("SI", "The method is not defined."))
[docs] def readResults(self, jobIndex, runOK=None):
"""
Read results of job with given *jobIndex*.
*runOK* specifies the status returned by the :meth:`runJobGroup`
method which produced the results.
"""
raise PyOpusError(DbgMsg("SI", "The method is not defined."))
[docs]class SimulationResults(object):
"""
Base class for simulation results.
"""
def __init__(self, params={}, variables={}, results={}):
self.paramsDict={}
self.variablesDict={}
self.resultsDict={}
self.paramsDict.update(params)
self.variablesDict.update(variables)
self.resultsDict.update(results)
self._compile()
def _compile(self):
# Build a results dictionary with corner for first key
# and measure for second key
resultsDictRev={}
for measure, measDict in self.resultsDict.items():
for corner, value in measDict.items():
if corner not in resultsDictRev:
resultsDictRev[corner]={}
resultsDictRev[corner][measure]=value
self.resultsDictRev=resultsDictRev
[docs] def driverTable(self):
"""
Returns a dictionary of available driver functions for
accessing simulation results.
They are used for measurement evaluation in
:class:`~pyopus.evaluator.performance.PerformanceEvaluator` objects
where they appear as a part of the environment for evaluating the
measures.
The following functions (dictionary key names) are usually available:
* ``ipath`` - a reference to the :func:`ipath` function for constructing
object paths in hierarchical designs
* ``vectorNames`` - returns the list of available raw vectors
* ``vector`` - retrieve a vector by its raw name
* ``v`` - retrieve a node potential or potential difference between two
nodes
* ``i`` - retrieve a current flowing through an instance (usually a
voltage source or inductor)
* ``p`` - retrieve an element instance property (assuming that it was
saves during simulation with a correspondign save directive)
* ``ns`` - retrieve a noise spectrum
* ``scaleName`` - retrieve the name of the vector holding the scale
corresponding to a vector or the default scale of a results group
* ``scale`` - retrieve the scale corresponding to a vector or the
default scale of a results group
"""
return {}
[docs] def evalEnvironment(self):
"""
Returns the environment in which measures will be evaluated.
The environment consists of following entries:
* Python variables defined for the simulation that produced the results.
* ``m`` - :mod:`pyopus.performance.measure` module
* ``np`` - :mod:`numpy` module
* ``param`` - parameter values dictionary. These parameters were used
during simulation that produced the results
* driver functions returned by the :meth:`driverTable` method.
The environment is constructed in this order. Later entries override
earlier entries if entry name is the same.
"""
env={}
env.update(self.variablesDict)
env['m']=m
env['np']=np
env['param']=self.paramsDict
env.update(self.driverTable())
return env
[docs] def param(self, name=None):
"""
Returns a stored PyOPUS parameter with given *name*.
If name is not specified returns the dictionary of all stored
parameters.
"""
if name is None:
return self.paramsDict
else:
if name not in self.paramsDict:
raise PyOpusError("Parameter '%s' not found." % (name))
return self.paramsDict[name]
[docs] def var(self, name=None):
"""
Returns a stored PyOPUS variable with given *name*.
If name is not specified returns the dictionary of all stored
variables.
"""
if name is None:
return self.variablesDict
else:
if name not in self.variablesDict:
raise PyOpusError("Variable '%s' not found." % (name))
return self.variablesDict[name]
[docs] def result(self, measure=None, corner=None):
"""
Returns a stored PyOPUS measure value named *name* for given *corner*.
If only *measure* is specified returns a dictionary holding values
of that measure across all corners. If measure does not exist and
error is raised.
If only *corner* is specified returns a dictionary holding all
measures for that particular corner. If no such corner exists in
the results dictionary an error is raised.
if neither *measure* nor *corner* is specified, returns the dictionary
holding all stored measure values across all corners with first index
being the measure name and the second one the corner name.
"""
if measure is None and corner is None:
return self.resultsDict
elif corner is None:
if measure not in self.resultsDict:
raise PyOpusError("Measure '%s' not found." % (measure))
return self.resultsDict[measure]
elif measure is None:
if corner not in self.resultsDictRev:
raise PyOpusError("Corner '%s' not found." % (corner))
return self.resultsDictRev[corner]
else:
if measure not in self.resultsDict:
raise PyOpusError("Measure '%s' not found." % (measure))
if corner not in self.resultsDictRev:
raise PyOpusError("Corner '%s' not found." % (corner))
return self.resultsDict[measure][corner]
[docs] def vector(self, name):
"""
Returns vector named *name* from results.
"""
raise PyOpusError("The method is not defined.")
[docs]class BlankSimulationResults(SimulationResults):
"""
Results object used for evaluating measures that are not associated
with an analysis (i.e. dependent measures).
"""
def __init__(self, params={}, variables={}, results={}):
SimulationResults.__init__(self, params, variables, results)
# For pickling
def __getstate__(self):
state=self.__dict__.copy()
del state['resultsDictRev']
return state
# For unpickling
def __setstate__(self, state):
self.__dict__.update(state)
self._compile()
[docs] def evalEnvironment(self):
"""
Returns the environment in which measures will be evaluated.
The environment consists of following entries:
* Python variables defined for the simulation that produced the results.
* ``m`` - :mod:`pyopus.performance.measure` module
* ``np`` - :mod:`numpy` module
* ``param`` - parameter values dictionary. These parameters were used
during simulation that produced the results
* ``result`` - double dictionary with measure results. First key is
measure, second key is corner.
* driver functions returned by the :meth:`driverTable` method.
The environment is constructed in this order. Later entries override
earlier entries if entry name is the same.
"""
env={}
env.update(self.variablesDict)
env['m']=m
env['np']=np
env['param']=self.paramsDict
env['result']=self.resultsDict
return env