# Please use the
# http://google-styleguide.googlecode.com/svn/trunk/pyguide.html
# as documentation format for this class
# disable some pylint checks.
# pylint: disable=broad-except, too-many-arguments
# pylint: disable=too-many-lines, too-many-public-methods
""" This module provides the XML-RPC communication with 20-sim 4C.
Example:
>>> from controllab import XXSim4CWrapper
"""
import collections
import math
import os.path
from .tool import ExeTool
from .wrapper import Wrapper, WrapperError
from distutils.util import strtobool
LogInfo = collections.namedtuple(
'LogInfo',
['filename', 'path', 'Overwrite', 'addTimeStamp'])
ProjectInfo = collections.namedtuple(
'ProjectInfo', ['projectname', 'identifier'])
TargetInfo = collections.namedtuple(
'TargetInfo', ['targetname', 'identifier', 'filename'])
ModelInfo = collections.namedtuple(
'ModelInfo', ['modelname', 'identifier', 'filename'])
TimeSettings = collections.namedtuple(
'TimeSettings', ['starttime', 'finishtime', 'interval'])
TemplateInfo = collections.namedtuple(
'TemplateInfo', ['identifier', 'name', 'version', 'file', 'path'])
[docs]class XXSim4CWrapper(Wrapper):
"""20-sim 4C scripting interface for Python.
This is a Python wrapper for 20-sim 4C's XML-RPC interface.
Attributes:
errorlevel: The level of errors that should be raised.
"""
DEFAULT_PORT = 5480
def _get_default_exetool(self, version=None, **kwargs):
""" Get the default ExeTool for XXSim4CWrapper. Optionally a different
version of the default tool can be started.
Args:
version (optional) : Will be passed on to ExeTool.__init__
Default: '2.2'
Returns:
ExeTool : An ExeTool representing 20-sim 4C.
"""
return ExeTool('XXSIM4C', version or '2.2', **kwargs)
def _set_xrc_server(self, xrc): # pylint: disable=arguments-differ
""" Correctly sets the self.server attribute.
Args:
xrc (xmlrpclib.ServerProxy) : The serverproxy
Example:
>>> import xmlrpc.client as xmlrpclib
>>> _set_xrc_server(self,
... xmlrpclib.Server_proxy('http://localhost:5480'))
"""
self.server = xrc.xxsim4C
[docs] def connect(self,
uri='http://localhost:{}'.format(DEFAULT_PORT),
version=ExeTool.DEFAULT_XXSIM4C_VERSION, **kwargs):
"""Create a connection with 20-sim 4C using XML-RPC
Args:
uri (str): The URI to connect to.
Defaults to 'http://localhost:5480' for connecting to 20-sim 4C.
Returns:
bool: True if successful, False otherwise.
Example:
>>> connect('http://localhost:5480')
True
"""
""" Check if we are running from 20-sim 4C """
try:
if os.environ['RUNNING_FROM_20SIM4C'] == '1':
running_from_20sim4C = True
if running_from_20sim4C:
ivc_HTTP_port = int(os.environ['XXSIM4C_HTTP_SCRIPT_PORT'])
""" override the uri """
uri='http://localhost:{}'.format(ivc_HTTP_port)
except:
pass
return super(XXSim4CWrapper, self).connect(uri, version=version, **kwargs)
[docs] def get_version(self):
"""Get the version number of the 20-sim 4C instance to which a connection was made.
Returns:
dict: A dictionary with the following fields:
* ``major``: The major number of the version number.
* ``minor``: The minor number of the version number.
* ``patch``: The patch number of the version number.
* ``build``: The build number of the version number.
False: On failure.
Example:
>>> xxsim4c.get_version()
{'major': 2,
'minor': 2,
'patch': 0,
'build': 4700}
"""
try:
# Obtain the version information via XML-RPC call.
version_struct = self.server.getVersion()
# Obtain the version string.
version_number_string = version_struct['version']
# Split the version number string.
version_numbers_array = version_number_string.split(".")
# Return a dictionary with all version data.
return {'major':int(version_numbers_array[0]),
'minor':int(version_numbers_array[1]),
'patch':int(version_numbers_array[2]),
'build':int(version_numbers_array[3])}
except Exception as error:
self.errorhandler(error)
return False
[docs] def open_project(self, project):
"""Opens a project file from its path.
Args:
project (str): Name of the file or project to load.
Returns:
int: The identifier of the project (1 or higher).
It can be used to select the active project
Returns False on error.
The project is activated and initialized after opening.
Example:
>>> project_id = open_project('c:\\temp\\myproject.emx')
"""
try:
# Check whether the provided project exists
if os.path.isabs(project) and not os.path.isfile(project):
raise FileNotFoundError('File not found: {}'.format(project))
# Open the project
reply = self.server.project.open({'name': project})
# the actual XML-RPC call
if reply is not None:
return reply['identifier']
except Exception as error:
self.errorhandler(error)
return False
[docs] def close_project(self):
""" Closes active 20-sim 4C project without saving changes
Returns:
bool: True when closed properly, False otherwise
Example:
>>> result = close_project(True)
"""
try:
# the actual XML-RPC call
return self.server.project.close()
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_project(self):
"""Returns a list of the opened project and their project identifier
Args:
None
Returns:
list: List of ``ProjectInfo``
"""
try:
# the actual XML-RPC call
reply = self.server.project.getProject()
projects = []
for project in reply:
# Ensure identifier is int instead of string.
project['identifier'] = int(project['identifier'])
# Create ProjectInfo tuple from dict and append to list.
projects.append(ProjectInfo(**project))
return projects
except Exception as error:
self.errorhandler(error)
return False
# TODO: XMLRPC NOT AVAILABLE
# def save_model(self, path):
# """ Save the changes made to the active model.
#
# Args:
# path (str): Save the active model at this location.
# *Note*: When given an empty string, the model is saved
# under its current name (if any)
# Returns:
# bool: True on success, False otherwise
#
# Example:
#
# >>> save_model('c:\\temp\\mymodel.emx')
# True
# """
# try:
# return self.server.save_model({'name': fileName})
# except Exception as error: self.errorhandler(error)
[docs] def get_targets(self):
"""Retrieves list of all the targets in use in the open project
Args:
None
Returns:
list: List of xxsim4c.TargetInfo
"""
try:
# the actual XML-RPC call
reply = self.server.project.getTargets()
targets = []
for target in reply:
# Turn identifier of target into int.
target['identifier'] = int(target['identifier'])
# Create tuple from dict and append to targets list.
targets.append(TargetInfo(**target))
return targets
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_active_target(self):
""" Retrieves the active target identifier
Args:
None
Returns:
collections.namedtuple: (targetname, filename, identifier)
bool: False on failure.
Example:
>>> targetinfo = getActive_target()
"""
try:
# the actual XML-RPC call
reply = self.server.project.getActiveTarget()
# Make identifier int instead of str.
reply['identifier'] = int(reply['identifier'])
# Return TargetInfo tuple from reply dict.
return TargetInfo(**reply)
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_active_target(self, identifier):
""" Sets the active target
Args:
Identifier (int): The identifier of the target.
Returns:
bool: True on success, False on failure.
Example:
>>> targetinfo = set_active_target(1)
"""
identifier = {'identifier': int(identifier)}
try:
# the actual XML-RPC call
return self.server.project.setActiveTarget(identifier)
except Exception as error:
self.errorhandler(error)
return False
[docs] def is_target_online(self):
"""Checks whether the active target is online or not
Returns:
bool: True when the target is on-line. False when the target is
off-line or when 20-sim 4C is still trying to connect.
"""
try:
# the actual XML-RPC call
return self.server.target.isOnline()
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_models(self):
"""Returns a list of all opened models and their model identifier
Args:
None
Returns:
list: List of ``xxsim4c.ModelInfo``
"""
try:
reply = self.server.project.getModels() # the actual XML-RPC call
models = []
for model in reply:
# Make identifier int instead of str.
model['identifier'] = int(model['identifier'])
# Make tuple from dict and append to models.
models.append(ModelInfo(model))
return models
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_active_model(self):
""" Returns name and unique identifier of active model
Args:
None
Returns:
xxsim4c.ModelInfo : The model information
Example:
>>> modelinfo = getActive_model()
"""
try:
# the actual XML-RPC call
reply = self.server.project.getActiveModel()
# Make identifier int instead of str.
reply['identifier'] = int(reply['identifier'])
# Make ModelInfo tuple from dict and return it.
return ModelInfo(reply)
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_active_model(self, identifier):
""" Returns name and unique identifier of active model
Args:
Identifier (integer)
Returns:
bool: True on success. False on failure.
Example:
>>> set_active_model(1)
True
"""
identifier = {'identifier': int(identifier)}
try:
# the actual XML-RPC call
return self.server.project.setActiveModel(identifier)
except Exception as error:
self.errorhandler(error)
return False
[docs] def start(self, log):
"""Start a model, model needs to be ready and target needs to be online
Args:
log : bool that enables or disables logging
Returns:
bool: True on success, False otherwise
Example:
>>> start = start(True)
"""
logging = {'enableLog': log}
try:
# the actual XML-RPC call
return self.server.run.start(logging)
except Exception as error:
self.errorhandler(error)
return False
[docs] def stop(self):
"""Stops a model, if model is not running true will be returned as well
Args:
None
Returns:
bool: True on success, False otherwise
Example:
>>> stopped = stop()
"""
try:
# the actual XML-RPC call
return self.server.run.stop()
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_url(self):
""" Gets the URL of the target
Args:
None
Returns:
URL of the target
Example:
>>> URL = get_url()
"""
try:
# the actual XML-RPC call
return self.server.configure.getURL()
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_url(self, url):
""" Sets the url of the target
Args:
url (string)
Returns:
bool: False on failure.
Example:
>>> url = set_url('192.168.1.123')
"""
try:
# the actual XML-RPC call
return self.server.configure.setURL(url)
except Exception as error:
self.errorhandler(error)
return False
[docs] def compile(self):
"""Compiles the 20-sim 4C sources to a target application/binary
Args:
None
Returns:
bool: True on success, False otherwise
Example:
>>> compile=compile()
"""
try:
# the actual XML-RPC call
return self.server.compile()
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_settings(self, settings=None):
"""Queries the active 20-sim 4C model for the settings
Args:
settings (list, optional) : The list of names of the settings that
should be retrieved. If left blank, all settings will be retrieved.
When querying a single setting, the value can be passed as string.
Returns:
The settings.
"""
# Start with an empty key list
settings = settings or []
if isinstance(settings, str):
settings = [settings]
try:
# the actual XML-RPC call
return self.server.getSettings({'keys': settings})
except Exception as error:
self.errorhandler(error)
return False
def get_setting_value(self, setting):
settingsResult = self.get_settings(setting)
if isinstance(settingsResult, bool):
return None
# Convert the returned string values to proper Python data types
retval = []
for setting in settingsResult:
if setting['type'] == 'double':
retval.append(float(setting['value']))
elif setting['type'] == 'boolean':
retval.append(bool(strtobool(setting['value'])))
elif setting['type'] == 'int':
retval.append(int(setting['value']))
else:
retval.append(setting['value'])
if len(retval) == 1:
return retval[0]
else:
return retval
[docs] def get_log_file(self):
""" Retrieves the properties of the log file.
Args:
None
Returns:
LogInfo : The information about the log file.
None in case of failure.
Example:
>>> logfile = get_log_file()
"""
try:
# Get dict from XMLRPC, make LogInfo tuple and return it.
return LogInfo(self.server.log.getLogFile())
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_log_file(self, logname, path, overwrite=False, timestamp=True):
""" Sets the LogFile
Args:
logname (str): The file name of the log file.
path (str): The path
overwrite(bool): Overwrite existing logfile.
timestamp (bool): adds time stamp to log.
Returns:
bool: False on failure.
Example:
>>> xxsim4C.set.log_file('test', r'D:\temp\', True)
True
"""
overwrite = bool(overwrite)
try:
return self.server.setSettings(
['log.filename',
'log.path',
'log.overWriteFile',
'log.addTimeStamp'],
[logname, path, overwrite, timestamp])
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_run_settings(self):
""" Retrieves the run settings.
Args:
None
Returns:
RunSettings : The run settings, None on failure
Example:
>>> run_settings = getRun_settings()
"""
try:
reply = self.server.getSettings() # the actual XML-RPC call
# Create and return tuple.
return TimeSettings(
starttime=float(reply['run.startTime']),
finishtime=float(reply['run.finishTime']),
interval=float(reply['run.discreteTimeInterval']))
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_run_settings(
self, start, finish, sampletime=None, frequency=None, omega=None):
""" Sets the LogFile
Args:
``Start`` , ``Finish`` and ``DiscreteTimeInterval``
Returns:
bool: False on failure.
Example:
>>> set_run_settings('0','1','0.01')
True
"""
if sampletime is None:
if frequency is not None:
sampletime = 1 / frequency
elif omega is not None:
sampletime = 2 * math.pi / omega
else:
self.errorhandler(
WrapperError(
'Please specify sampletime, frequency or omega.'))
try:
return self.set_settings(['run.startTime',
'run.finishTime',
'run.discreteTimeInterval'],
[start,
finish,
sampletime])
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_log_settings(self):
""" Retrieves the settings for logging.
Args:
None
Returns:
TimeSettings : The timing settings for the logging.
Example:
>>> logsettings = get_log_settings()
"""
try:
reply = self.server.getSettings() # the actual XML-RPC call
# Convert values in reply from string to numbers.
return TimeSettings(
starttime=int(reply['log.startTime']),
finishtime=int(reply['log.finishTime']),
interval=float(reply['log.discreteTimeInterval']))
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_log_settings(
self, start, finish, sampletime=None, frequency=None, omega=None):
""" Sets the LogFile
Args:
start (int) : Start time for logging
finish (int) : Finish time for logging
sampletime (float, optional) : Sample time for logging.
frequency (float, optional) : Logging frequency.
omega (float, optional) : Omega for logging.
One of sampletime, frequency and omega must be provided. If multiple
are provided, the first one in order (sampletime, frequency, omega)
will be used.
Returns:
bool: False on failure.
Example:
>>> log_settings = set_log_settings(0,1,'sampletime'=0.01)
"""
if sampletime is None:
if frequency is not None:
sampletime = 1 / frequency
elif omega is not None:
sampletime = 2 * math.pi / omega
else:
self.errorhandler(
WrapperError(
'Please specify sampletime, frequency or omega.'))
return False
if self.get_run_sampletime() < sampletime:
self.errorhandler(
WrapperError(
'Unable to set log settings. Sampletime of log must be '
'smaller than sampletime of run'))
return False
try:
return self.set_settings(['log.startTime',
'log.finishTime',
'log.discreteTimeInterval'],
[start, finish, sampletime])
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_variable_flag(self, variables, log, monitor):
# Todo write description, test
""" set flags for logging and monitoring variables
Args:
variables (list of str) : The variable names.
log (bool) : True to enable logging, False to disable.
monitor (bool) : True to enable monitoring, False to disable.
Returns:
bool
Example:
>>> Flag=set_variable_flag([var1,var2],
... ['log','monitor'],
... ['True','False'])
"""
# check for single or list variables
if isinstance(variables, str):
variables = [variables]
try:
# the actual XML-RPC call
return self.server.model.setVariableFlag(
{'variablenames': variables,
'flag': [{'key': 'log', 'value': log},
{'key': 'monitor', 'value': monitor}]})
except Exception as error:
self.errorhandler(error)
return False
[docs] def query_templates(self):
""" Queries the available Templates
Args:
None
Returns:
TemplateInfo : The information about the templates
Example:
>>> Templates=query_templates()
"""
try:
# Create TemplateInfo from dict returned by XMLRPC call.
return TemplateInfo(**self.server.configure.queryTemplates())
except Exception as error:
self.errorhandler(error)
return False
[docs] def select_template(self, target_file):
""" Selects and sets a templates
Args:
target_file (str): the file name of the Target Configuration
(TCF) of the template.
Returns:
bool : True indicates success.
Example:
>>>select_template('test')
True
"""
try:
# the actual XML-RPC call
return self.server.configure.selectTemplate(target_file)
except Exception as error:
self.errorhandler(error)
return False
[docs] def set_run_start(self, start):
""" Set the run start time.
Equivalent of: set_settings('run.startTime', start)
"""
return self.set_settings('run.startTime', start)
[docs] def set_run_stop(self, stop):
""" Set the run finish time.
Equivalent of: set_settings('run.finishTime', start)
"""
return self.set_settings('run.finishTime', stop)
[docs] def set_run_endless(self, state=False):
""" Set the run endless setting.
Equivalent of: set_settings('run.endless', state)
Args:
state (bool, optional) : Default false (run until finish time.)
"""
return self.set_settings('run.endless', state)
[docs] def get_run_endless(self):
""" Get the run endless setting.
Equivalent of: get_settings('run.endless')
"""
return self.get_setting_value('run.endless')
[docs] def set_run_frequency(self, frequency):
""" Set the run frequency.
Equivalent of:
self.set_settings('run.discreteTimeInterval', 1 / frequency)
Args:
frequency (float) : The run frequency.
"""
return self.set_settings('run.discreteTimeInterval', 1 / frequency)
[docs] def set_run_omega(self, omega):
""" Set the run omega.
Equivalent of:
self.set_settings('run.discreteTimeInterval', 2 * math.pi / omega)
Args:
omega (float) : The run omega.
"""
return self.set_settings(
'run.discreteTimeInterval', 2 * math.pi / omega)
[docs] def set_run_sampletime(self, sampletime):
""" Set the run sampletime.
Equivalent of:
self.set_settings('run.discreteTimeInterval', sampletime)
Args:
sampletime (float) : The run sampletime.
"""
return self.set_settings('run.discreteTimeInterval', sampletime)
[docs] def set_log_start(self, start):
""" Set the log start time.
Equivalent of:
self.set_settings('log.starttime', start)
Args:
start (float) : The log start time.
"""
return self.set_settings('log.startTime', start)
[docs] def set_log_stop(self, stop):
""" Set the log finish time.
Equivalent of:
self.set_settings('log.finishTime', stop)
Args:
start (float) : The log finish time.
"""
return self.set_settings('log.finishTime', stop)
[docs] def set_log_frequency(self, frequency):
""" Set the log frequency.
Equivalent of:
self.set_settings('log.discreteTimeInterval', 1 / frequency)
Args:
frequency (float) : The log frequency.
"""
return self.set_settings('log.discreteTimeInterval', 1 / frequency)
[docs] def set_log_omega(self, omega):
""" Set the log omega.
Equivalent of:
self.set_settings('log.discreteTimeInterval', 2 * math.pi / omega)
Args:
omega (float) : The log omega.
"""
return self.set_settings(
'log.discreteTimeInterval', 2 * math.pi / omega)
[docs] def set_log_sampletime(self, sampletime):
""" Set the log sampletime.
Equivalent of:
self.set_settings('log.discreteTimeInterval', sampletime)
Args:
sampletime (float) : The log sampletime.
"""
return self.set_settings('log.discreteTimeInterval', sampletime)
[docs] def get_run_start(self):
""" Get the run start time. """
return self.get_setting_value('run.startTime')
[docs] def get_run_stop(self):
""" Get the run finish time. """
return self.get_setting_value('run.finishTime')
[docs] def get_run_sampletime(self):
""" Get the run sampletime """
return self.get_setting_value('run.discreteTimeInterval')
[docs] def get_run_frequency(self):
""" Get the run frequency.
Returns:
float : Frequency as calculated from the sampletime.
"""
return 1 / self.get_setting_value('run.discreteTimeInterval')
[docs] def get_run_omega(self):
""" Get the run omega.
Returns:
float : Omega as calculated from the sampletime.
"""
return 2 * math.pi / self.get_setting_value('run.discreteTimeInterval')
[docs] def get_log_start(self):
""" Get the log start time. """
return self.get_setting_value('log.startTime')
[docs] def get_log_stop(self):
""" Get the log finish time. """
return self.get_setting_value('log.finishTime')
[docs] def get_log_sampletime(self):
""" Get the log sampletime. """
return self.get_setting_value('log.discreteTimeInterval')
[docs] def get_log_frequency(self):
""" Get the log frequency.
Returns:
float : Frequency as calculated from the sampletime.
"""
return 1 / self.get_setting_value('log.discreteTimeInterval')
[docs] def get_log_omega(self):
""" Get the log omega.
Returns:
float : Omega as calculated from the sampletime.
"""
return 2 * math.pi / self.get_setting_value('log.discreteTimeInterval')
[docs] def get_log_filename(self):
""" Get the base file name of the log file. """
return self.get_setting_value('log.filename')
[docs] def get_log_last_filename(self):
""" Get the file name of the last stored log file. """
return self.get_setting_value('log.lastfilename')
[docs] def set_log_filename(self, filename):
""" Set the file name of the log file.
Args filename (str) : The file name
"""
return self.set_settings('log.filename', filename)
[docs] def get_log_path(self):
""" Get the path of the log file. """
return self.get_setting_value('log.path')
[docs] def set_log_path(self, path):
""" Set the path of the log file.
Args path (str) : The path
"""
return self.set_settings('log.path', path)
[docs] def get_log_overwrite(self):
""" Get the overwrite setting for the log. """
return self.get_setting_value('log.overwriteFile')
[docs] def set_log_overwrite(self, overwrite=False):
""" Set the overwrite setting for the log.
Args:
overwrite (bool, optional): True to overwrite, default is False.
"""
return self.set_settings('log.overwriteFile', overwrite)
[docs] def get_log_add_time_stamp(self):
""" Get the addTimeStamp property for the log.
Returns:
bool: True if the loging is timestamped.
"""
return self.get_setting_value('log.addTimeStamp')
[docs] def set_log_add_time_stamp(self, timestamp=True):
""" Set the addTimeStamp property for the log.
Args:
timestamp (bool, optional): True to add timestamps to log messages.
Default is True.
"""
return self.set_settings('log.addTimeStamp', timestamp)
[docs] def get_log_show_plot(self):
""" Get the showPlot property for the log.
Returns:
bool: True if the 'Show plot after run' option is enabled.
"""
return self.get_setting_value('log.showPlot')
[docs] def set_log_show_plot(self, show_plot=True):
""" Sets the showPlot property for the log.
Args:
show_plot (bool, optional): True shows the plot with the log results
after logging is ready. Default is True.
"""
return self.set_settings('log.showPlot', show_plot)
[docs] def get_log_variables(self):
""" Returns a list with all logVariables."""
filters = [{'key': 'flag', 'value': 'logvariable'}]
try:
# the actual XML-RPC call
variables = self.server.model.getVariables(
{'variables': [], 'filters': filters})
return variables
except Exception as error:
self.errorhandler(error)
return False
[docs] def get_monitor_variables(self):
""" Returns a list with all monitorVariables. """
filters = [{'key': 'flag', 'value': 'monitorvariable'}]
try:
# the actual XML-RPC call
variables = self.server.model.getVariables(
{'variables': [], 'filters': filters})
return variables
except Exception as error:
self.errorhandler(error)
return False
# Missing functions:
# get/set start/finish time and frequency (in log window)
# getAvailableLogFreq lists the available Freq for logging
# Get List of models below a certain target (Low priority)
# addNewTarget: Adds a new target to the project. (Long term)
# GET and SET TEMPLATE:
# Gets all possible templates / Selects a new template (Long Term)
# Discover targets and set IP (Long term)
# connect Inputs/Outputs (Long Term)
# Get//SetActive log variables (please also allow 'all') (Long Term)
# functions that should be expanded:
# GetTarget -> IP (or DNS) adress should also be exported!
# configure(basic)
# connect (basic)
