def __init__(self, pid):

self.pid = pid

self.process = psutil.Process(pid)

self.returncode = 0

def poll(self):

return None if self.process.is_running() else True

def kill(self):

return os.kill(self.pid, signal.SIGKILL)

def wait(self, timeout):

def __init__(self):

# Get the path to the OMC executable, if not installed this will be None

omc_env_home = os.environ.get('OPENMODELICAHOME')

if omc_env_home:

self.omhome = omc_env_home

else:

path_to_omc = spawn.find_executable("omc")

if path_to_omc is None:

raise ValueError("Cannot find OpenModelica executable, please install from openmodelica.org")

self.omhome = os.path.split(os.path.split(os.path.realpath(path_to_omc))[0])[0]

def _get_omc_path(self):

try:

return os.path.join(self.omhome, 'bin', 'omc')

except BaseException:

logger.error("The OpenModelica compiler is missing in the System path (%s), please install it" % os.path.join(self.omhome, 'bin', 'omc'))

def __init__(self, readonly=False):

self.readonly = readonly

self.omc_cache = {}

self._omc_process = None

self._omc_command = None

self._omc = None

self._dockerCid = None

self._serverIPAddress = "127.0.0.1"

self._interactivePort = None

# FIXME: this code is not well written... need to be refactored

self._temp_dir = tempfile.gettempdir()

# generate a random string for this session

self._random_string = uuid.uuid4().hex

# omc log file

self._omc_log_file = None

try:

self._currentUser = getpass.getuser()

if not self._currentUser:

self._currentUser = "nobody"

except KeyError:

# We are running as a uid not existing in the password database... Pretend we are nobody

self._currentUser = "nobody"

def __del__(self):

try:

self.sendExpression("quit()")

except:

pass

self._omc_log_file.close()

if sys.version_info.major >= 3:

try:

self._omc_process.wait(timeout=2.0)

except:

if self._omc_process:

self._omc_process.kill()

else:

for i in range(0,100):

time.sleep(0.02)

if self._omc_process and (self._omc_process.poll() is not None):

break

# kill self._omc_process process if it is still running/exists

if self._omc_process is not None and self._omc_process.returncode is None:

print("OMC did not exit after being sent the quit() command; killing the process with pid=%s" % str(self._omc_process.pid))

if sys.platform=="win32":

self._omc_process.kill()

self._omc_process.wait()

else:

os.killpg(os.getpgid(self._omc_process.pid), signal.SIGTERM)

self._omc_process.kill()

self._omc_process.wait()

def _create_omc_log_file(self, suffix):

if sys.platform == 'win32':

self._omc_log_file = open(os.path.join(self._temp_dir, "openmodelica.{0}.{1}.log".format(suffix, self._random_string)), 'w')

else:

# this file must be closed in the destructor

self._omc_log_file = open(os.path.join(self._temp_dir, "openmodelica.{0}.{1}.{2}.log".format(self._currentUser, suffix, self._random_string)), 'w')

def _start_omc_process(self, timeout):

if sys.platform == 'win32':

omhome_bin = os.path.join(self.omhome, 'bin').replace("\\", "/")

my_env = os.environ.copy()

my_env["PATH"] = omhome_bin + os.pathsep + my_env["PATH"]

self._omc_process = subprocess.Popen(self._omc_command, stdout=self._omc_log_file, stderr=self._omc_log_file, env=my_env)

else:

# set the user environment variable so omc running from wsgi has the same user as OMPython

my_env = os.environ.copy()

my_env["USER"] = self._currentUser

# Because we spawned a shell, and we need to be able to kill OMC, create a new process group for this

self._omc_process = subprocess.Popen(self._omc_command, shell=True, stdout=self._omc_log_file, stderr=self._omc_log_file, env=my_env, preexec_fn=os.setsid)

if self._docker:

for i in range(0,40):

try:

with open(self._dockerCidFile, "r") as fin:

self._dockerCid = fin.read().strip()

except:

pass

if self._dockerCid:

break

time.sleep(timeout / 40.0)

try:

os.remove(self._dockerCidFile)

except:

pass

if self._dockerCid is None:

logger.error("Docker did not start. Log-file says:\n%s" % (open(self._omc_log_file.name).read()))

raise Exception("Docker did not start (timeout=%f might be too short especially if you did not docker pull the image before this command)." % timeout)

if self._docker or self._dockerContainer:

if self._dockerNetwork == "separate":

self._serverIPAddress = json.loads(subprocess.check_output(["docker", "inspect", self._dockerCid]).decode().strip())[0]["NetworkSettings"]["IPAddress"]

for i in range(0,40):

if sys.platform == 'win32':

break

dockerTop = subprocess.check_output(["docker", "top", self._dockerCid]).decode().strip()

self._omc_process = None

for line in dockerTop.split("\n"):

columns = line.split()

if self._random_string in line:

try:

self._omc_process = DummyPopen(int(columns[1]))

except psutil.NoSuchProcess:

raise Exception("Could not find PID %d - is this a docker instance spawned without --pid=host?\nLog-file says:\n%s" % (self._random_string, dockerTop, open(self._omc_log_file.name).read()))

break

if self._omc_process is not None:

break

time.sleep(timeout / 40.0)

if self._omc_process is None:

raise Exception("Docker top did not contain omc process %s:\n%s\nLog-file says:\n%s" % (self._random_string, dockerTop, open(self._omc_log_file.name).read()))

return self._omc_process

def _getuid(self):

"""

return 1000 if sys.platform == 'win32' else os.getuid()

def _set_omc_command(self, omc_path_and_args_list):

"""Define the command that will be called by the subprocess module.

if (self._docker or self._dockerContainer) and sys.platform == "win32":

extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"]

if not self._interactivePort:

raise Exception("docker on Windows requires knowing which port to connect to. For dockerContainer=..., the container needs to have already manually exposed this port when it was started (-p 127.0.0.1:n:n) or you get an error later.")

else:

extraFlags = []

if self._docker:

if sys.platform == "win32":

p = int(self._interactivePort)

dockerNetworkStr = ["-p", "127.0.0.1:%d:%d" % (p,p)]

elif self._dockerNetwork == "host" or self._dockerNetwork is None:

dockerNetworkStr = ["--network=host"]

elif self._dockerNetwork == "separate":

dockerNetworkStr = []

extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"]

else:

raise Exception('dockerNetwork was set to %s, but only \"host\" or \"separate\" is allowed')

self._dockerCidFile = self._omc_log_file.name + ".docker.cid"

omcCommand = ["docker", "run", "--cidfile", self._dockerCidFile, "--rm", "--env", "USER=%s" % self._currentUser, "--user", str(self._getuid())] + self._dockerExtraArgs + dockerNetworkStr + [self._docker, self._dockerOpenModelicaPath]

elif self._dockerContainer:

omcCommand = ["docker", "exec", "--env", "USER=%s" % self._currentUser, "--user", str(self._getuid())] + self._dockerExtraArgs + [self._dockerContainer, self._dockerOpenModelicaPath]

self._dockerCid = self._dockerContainer

else:

omcCommand = [self._get_omc_path()]

if self._interactivePort:

extraFlags = extraFlags + ["--interactivePort=%d" % int(self._interactivePort)]

omc_path_and_args_list = omcCommand + omc_path_and_args_list + extraFlags

if sys.platform == 'win32':

self._omc_command = omc_path_and_args_list

else:

self._omc_command = ' '.join([shlex.quote(a) if (sys.version_info > (3, 0)) else a for a in omc_path_and_args_list])

return self._omc_command

@abc.abstractmethod

def _connect_to_omc(self, timeout):

pass

# FIXME: we should have one function which interacts with OMC. Either execute OR sendExpression.

# Execute uses OMParser.check_for_values and sendExpression uses OMTypedParser.parseString.

# We should have one parser. Then we can get rid of one of these functions.

@abc.abstractmethod

def execute(self, command):

pass

def clearOMParserResult(self):

OMParser.result = {}

# FIXME: we should have one function which interacts with OMC. Either execute OR sendExpression.

# Execute uses OMParser.check_for_values and sendExpression uses OMTypedParser.parseString.

# We should have one parser. Then we can get rid of one of these functions.

@abc.abstractmethod

def sendExpression(self, command, parsed=True):

"""

pass

def ask(self, question, opt=None, parsed=True):

p = (question, opt, parsed)

if self.readonly and question != 'getErrorString':

# can use cache if readonly

if p in self.omc_cache:

return self.omc_cache[p]

if opt:

expression = '{0}({1})'.format(question, opt)

else:

expression = question

logger.debug('OMC ask: {0} - parsed: {1}'.format(expression, parsed))

try:

if parsed:

res = self.execute(expression)

else:

res = self.sendExpression(expression, parsed=False)

except Exception as e:

logger.error("OMC failed: {0}, {1}, parsed={2}".format(question, opt, parsed))

raise e

# save response

self.omc_cache[p] = res

return res

# TODO: Open Modelica Compiler API functions. Would be nice to generate these.

def loadFile(self, filename):

return self.ask('loadFile', '"{0}"'.format(filename))

def loadModel(self, className):

return self.ask('loadModel', className)

def isModel(self, className):

return self.ask('isModel', className)

def isPackage(self, className):

return self.ask('isPackage', className)

def isPrimitive(self, className):

return self.ask('isPrimitive', className)

def isConnector(self, className):

return self.ask('isConnector', className)

def isRecord(self, className):

return self.ask('isRecord', className)

def isBlock(self, className):

return self.ask('isBlock', className)

def isType(self, className):

return self.ask('isType', className)

def isFunction(self, className):

return self.ask('isFunction', className)

def isClass(self, className):

return self.ask('isClass', className)

def isParameter(self, className):

return self.ask('isParameter', className)

def isConstant(self, className):

return self.ask('isConstant', className)

def isProtected(self, className):

return self.ask('isProtected', className)

def getPackages(self, className="AllLoadedClasses"):

return self.ask('getPackages', className)

def getClassRestriction(self, className):

return self.ask('getClassRestriction', className)

def getDerivedClassModifierNames(self, className):

return self.ask('getDerivedClassModifierNames', className)

def getDerivedClassModifierValue(self, className, modifierName):

return self.ask('getDerivedClassModifierValue', '{0}, {1}'.format(className, modifierName))

def typeNameStrings(self, className):

return self.ask('typeNameStrings', className)

def getComponents(self, className):

return self.ask('getComponents', className)

def getClassComment(self, className):

try:

return self.ask('getClassComment', className)

except pyparsing.ParseException as ex:

logger.warning("Method 'getClassComment' failed for {0}".format(className))

logger.warning('OMTypedParser error: {0}'.format(ex.message))

return 'No description available'

def getNthComponent(self, className, comp_id):

""" returns with (type, name, description) """

return self.ask('getNthComponent', '{0}, {1}'.format(className, comp_id))

def getNthComponentAnnotation(self, className, comp_id):

return self.ask('getNthComponentAnnotation', '{0}, {1}'.format(className, comp_id))

def getImportCount(self, className):

return self.ask('getImportCount', className)

def getNthImport(self, className, importNumber):

# [Path, id, kind]

return self.ask('getNthImport', '{0}, {1}'.format(className, importNumber))

def getInheritanceCount(self, className):

return self.ask('getInheritanceCount', className)

def getNthInheritedClass(self, className, inheritanceDepth):

return self.ask('getNthInheritedClass', '{0}, {1}'.format(className, inheritanceDepth))

def getParameterNames(self, className):

try:

return self.ask('getParameterNames', className)

except KeyError as ex:

logger.warning('OMPython error: {0}'.format(ex))

# FIXME: OMC returns with a different structure for empty parameter set

return []

def getParameterValue(self, className, parameterName):

try:

return self.ask('getParameterValue', '{0}, {1}'.format(className, parameterName))

except pyparsing.ParseException as ex:

logger.warning('OMTypedParser error: {0}'.format(ex.message))

return ""

def getComponentModifierNames(self, className, componentName):

return self.ask('getComponentModifierNames', '{0}, {1}'.format(className, componentName))

def getComponentModifierValue(self, className, componentName):

try:

# FIXME: OMPython exception UnboundLocalError exception for 'Modelica.Fluid.Machines.ControlledPump'

return self.ask('getComponentModifierValue', '{0}, {1}'.format(className, componentName))

except pyparsing.ParseException as ex:

logger.warning('OMTypedParser error: {0}'.format(ex.message))

result = self.ask('getComponentModifierValue', '{0}, {1}'.format(className, componentName), parsed=False)

try:

answer = OMParser.check_for_values(result)

OMParser.result = {}

return answer[2:]

except (TypeError, UnboundLocalError) as ex:

logger.warning('OMParser error: {0}'.format(ex))

return result

def getExtendsModifierNames(self, className, componentName):

return self.ask('getExtendsModifierNames', '{0}, {1}'.format(className, componentName))

def getExtendsModifierValue(self, className, extendsName, modifierName):

try:

# FIXME: OMPython exception UnboundLocalError exception for 'Modelica.Fluid.Machines.ControlledPump'

return self.ask('getExtendsModifierValue', '{0}, {1}, {2}'.format(className, extendsName, modifierName))

except pyparsing.ParseException as ex:

logger.warning('OMTypedParser error: {0}'.format(ex.message))

result = self.ask('getExtendsModifierValue', '{0}, {1}, {2}'.format(className, extendsName, modifierName), parsed=False)

try:

answer = OMParser.check_for_values(result)

OMParser.result = {}

return answer[2:]

except (TypeError, UnboundLocalError) as ex:

logger.warning('OMParser error: {0}'.format(ex))

return result

def getNthComponentModification(self, className, comp_id):

# FIXME: OMPython exception Results KeyError exception

# get {$Code(....)} field

# \{\$Code\((\S*\s*)*\)\}

value = self.ask('getNthComponentModification', '{0}, {1}'.format(className, comp_id), parsed=False)

value = value.replace("{$Code(", "")

return value[:-3]

# return self.re_Code.findall(value)

# function getClassNames

# input TypeName class_ = $Code(AllLoadedClasses);

# input Boolean recursive = false;

# input Boolean qualified = false;

# input Boolean sort = false;

# input Boolean builtin = false "List also builtin classes if true";

# input Boolean showProtected = false "List also protected classes if true";

# output TypeName classNames[:];

# end getClassNames;

def getClassNames(self, className=None, recursive=False, qualified=False, sort=False, builtin=False,

showProtected=False):

if className:

value = self.ask('getClassNames',

'{0}, recursive={1}, qualified={2}, sort={3}, builtin={4}, showProtected={5}'.format(

className, str(recursive).lower(), str(qualified).lower(), str(sort).lower(),

str(builtin).lower(), str(showProtected).lower()))

else:

value = self.ask('getClassNames',

'recursive={0}, qualified={1}, sort={2}, builtin={3}, showProtected={4}'.format(

str(recursive).lower(), str(qualified).lower(), str(sort).lower(),

str(builtin).lower(), str(showProtected).lower()))

def __init__(self, readonly=False, serverFlag='--interactive=corba', timeout = 10.0, docker = None, dockerContainer = None, dockerExtraArgs = [], dockerOpenModelicaPath = "omc", dockerNetwork = None):

OMCSessionHelper.__init__(self)

OMCSessionBase.__init__(self, readonly)

self._create_omc_log_file("objid")

# Locating and using the IOR

if sys.platform != 'win32' or docker or dockerContainer:

self._port_file = "openmodelica." + self._currentUser + ".objid." + self._random_string

else:

self._port_file = "openmodelica.objid." + self._random_string

self._port_file = os.path.join("/tmp" if (docker or dockerContainer) else self._temp_dir, self._port_file).replace("\\", "/")

# set omc executable path and args

self._docker = docker

self._dockerContainer = dockerContainer

self._dockerExtraArgs = dockerExtraArgs

self._dockerOpenModelicaPath = dockerOpenModelicaPath

self._dockerNetwork = dockerNetwork

self._timeout = timeout

self._create_omc_log_file("port")

self._set_omc_command([serverFlag, "+c={0}".format(self._random_string)])

# start up omc executable, which is waiting for the CORBA connection

self._start_omc_process(timeout)

# connect to the running omc instance using CORBA

self._connect_to_omc(timeout)

def __del__(self):

OMCSessionBase.__del__(self)

def _connect_to_omc(self, timeout):

# add OPENMODELICAHOME\lib\python to PYTHONPATH so python can load omniORB imports

sys.path.append(os.path.join(self.omhome, 'lib', 'python'))

# import the skeletons for the global module

try:

from omniORB import CORBA

from OMPythonIDL import _OMCIDL

except ImportError:

self._omc_process.kill()

raise

self._omc_corba_uri = "file:///" + self._port_file

# See if the omc server is running

attempts = 0

while True:

if self._dockerCid:

try:

self._ior = subprocess.check_output(["docker", "exec", self._dockerCid, "cat", self._port_file], stderr=subprocess.DEVNULL if (sys.version_info > (3, 0)) else subprocess.STDOUT).decode().strip()

break

except subprocess.CalledProcessError:

pass

if os.path.isfile(self._port_file):

# Read the IOR file

with open(self._port_file, 'r') as f_p:

self._ior = f_p.readline()

break

attempts += 1

if attempts == 80:

name = self._omc_log_file.name

self._omc_log_file.close()

with open(name) as fin:

contents = fin.read()

self._omc_process.kill()

raise Exception("OMC Server is down (timeout=%f). Please start it! If the OMC version is old, try OMCSession(..., serverFlag='-d=interactiveCorba') or +d=interactiveCorba. Log-file says:\n%s" % (timeout, contents))

time.sleep(timeout / 80.0)

while True:

if self._dockerCid:

try:

self._port = subprocess.check_output(["docker", "exec", self._dockerCid, "cat", self._port_file]).decode().strip()

break

except:

pass

else:

if os.path.isfile(self._port_file):

# Read the port file

with open(self._port_file, 'r') as f_p:

self._port = f_p.readline()

os.remove(self._port_file)

break

attempts += 1

if attempts == 80.0:

name = self._omc_log_file.name

self._omc_log_file.close()

logger.error("OMC Server is down (timeout=%f). Please start it! Log-file says:\n%s" % open(name).read())

raise Exception("OMC Server is down. Could not open file %s" % (timeout,self._port_file))

time.sleep(timeout / 80.0)

logger.info("OMC Server is up and running at {0}".format(self._omc_corba_uri))

# initialize the ORB with maximum size for the ORB set

sys.argv.append("-ORBgiopMaxMsgSize")

sys.argv.append("2147483647")

self._orb = CORBA.ORB_init(sys.argv, CORBA.ORB_ID)

# Find the root POA

self._poa = self._orb.resolve_initial_references("RootPOA")

# Convert the IOR into an object reference

self._obj_reference = self._orb.string_to_object(self._ior)

# Narrow the reference to the OmcCommunication object

self._omc = self._obj_reference._narrow(_OMCIDL.OmcCommunication)

# Check if we are using the right object

if self._omc is None:

logger.error("Object reference is not valid")

raise Exception

def execute(self, command):

## check for process is running

p=self._omc_process.poll()

if (p == None):

result = self._omc.sendExpression(command)

if command == "quit()":

self._omc = None

return result

else:

answer = OMParser.check_for_values(result)

return answer

else:

raise Exception("Process Exited, No connection with OMC. Create a new instance of OMCSession")

def sendExpression(self, command, parsed=True):

## check for process is running

p=self._omc_process.poll()

if (p== None):

result = self._omc.sendExpression(str(command))

if command == "quit()":

self._omc = None

return result

else:

if parsed is True:

answer = OMTypedParser.parseString(result)

return answer

else:

return result

else:

def __init__(self, readonly=False, timeout = 10.00, docker = None, dockerContainer = None, dockerExtraArgs = [], dockerOpenModelicaPath = "omc", dockerNetwork = None, port = None):

OMCSessionHelper.__init__(self)

OMCSessionBase.__init__(self, readonly)

# Locating and using the IOR

if sys.platform != 'win32' or docker or dockerContainer:

self._port_file = "openmodelica." + self._currentUser + ".port." + self._random_string

else:

self._port_file = "openmodelica.port." + self._random_string

self._docker = docker

self._dockerContainer = dockerContainer

self._dockerExtraArgs = dockerExtraArgs

self._dockerOpenModelicaPath = dockerOpenModelicaPath

self._dockerNetwork = dockerNetwork

self._create_omc_log_file("port")

self._timeout = timeout

self._port_file = os.path.join("/tmp" if docker else self._temp_dir, self._port_file).replace("\\", "/")

self._interactivePort = port

# set omc executable path and args

self._set_omc_command([

"--interactive=zmq",

"--locale=C",

"-z={0}".format(self._random_string)

])

# start up omc executable, which is waiting for the ZMQ connection

self._start_omc_process(timeout)

# connect to the running omc instance using ZMQ

self._connect_to_omc(timeout)

def __del__(self):

OMCSessionBase.__del__(self)

def _connect_to_omc(self, timeout):

self._omc_zeromq_uri = "file:///" + self._port_file

# See if the omc server is running

attempts = 0

self._port = None

while True:

if self._dockerCid:

try:

self._port = subprocess.check_output(["docker", "exec", self._dockerCid, "cat", self._port_file], stderr=subprocess.DEVNULL if (sys.version_info > (3, 0)) else subprocess.STDOUT).decode().strip()

break

except:

pass

else:

if os.path.isfile(self._port_file):

# Read the port file

with open(self._port_file, 'r') as f_p:

self._port = f_p.readline()

os.remove(self._port_file)

break

attempts += 1

if attempts == 80.0:

name = self._omc_log_file.name

self._omc_log_file.close()

logger.error("OMC Server did not start. Please start it! Log-file says:\n%s" % open(name).read())

raise Exception("OMC Server did not start (timeout=%f). Could not open file %s" % (timeout,self._port_file))

time.sleep(timeout / 80.0)

self._port = self._port.replace("0.0.0.0", self._serverIPAddress)

logger.info("OMC Server is up and running at {0} pid={1} cid={2}".format(self._omc_zeromq_uri, self._omc_process.pid, self._dockerCid))

# Create the ZeroMQ socket and connect to OMC server

import zmq

context = zmq.Context.instance()

self._omc = context.socket(zmq.REQ)

self._omc.setsockopt(zmq.LINGER, 0) # Dismisses pending messages if closed

self._omc.setsockopt(zmq.IMMEDIATE, True) # Queue messages only to completed connections

self._omc.connect(self._port)

def execute(self, command):

## check for process is running

return self.sendExpression(command, parsed=False)

def sendExpression(self, command, parsed=True):

## check for process is running

p=self._omc_process.poll()

if (p == None):

attempts = 0

while True:

try:

self._omc.send_string(str(command), flags=zmq.NOBLOCK)

break

except zmq.error.Again:

pass

attempts += 1

if attempts == 50.0:

name = self._omc_log_file.name

self._omc_log_file.close()

raise Exception("No connection with OMC (timeout=%f). Log-file says: \n%s" % (self._timeout, open(name).read()))

time.sleep(self._timeout / 50.0)

if command == "quit()":

self._omc.close()

self._omc = None

return None

else:

result = self._omc.recv_string()

if parsed is True:

answer = OMTypedParser.parseString(result)

return answer

else:

return result

else:

def __init__(self, fileName=None, modelName=None, lmodel=[], useCorba=False, commandLineOptions=None, variableFilter=None): # 1

"""

if fileName is None and modelName is None and not lmodel: # all None

if useCorba:

self.getconn = OMCSession()

else:

self.getconn = OMCSessionZMQ()

return

if fileName is None:

return "File does not exist"

self.tree = None

self.quantitiesList=[]

self.paramlist={}

self.inputlist={}

self.outputlist={}

self.continuouslist={}

self.simulateOptions={}

self.overridevariables={}

self.simoptionsoverride={}

self.linearOptions={'startTime':0.0, 'stopTime': 1.0, 'stepSize':0.002, 'tolerance':1e-8}

self.optimizeOptions={'startTime':0.0, 'stopTime': 1.0, 'numberOfIntervals':500, 'stepSize':0.002, 'tolerance':1e-8}

self.linearinputs = [] # linearization input list

self.linearoutputs = [] # linearization output list

self.linearstates = [] # linearization states list

self.tempdir = ""

if useCorba:

self.getconn = OMCSession()

else:

self.getconn = OMCSessionZMQ()

## set commandLineOptions if provided by users

if commandLineOptions is not None:

exp="".join(["setCommandLineOptions(","\"",commandLineOptions,"\"",")"])

self.getconn.sendExpression(exp)

self.xmlFile = None

self.lmodel = lmodel # may be needed if model is derived from other model

self.modelName = modelName # Model class name

self.fileName = fileName # Model file/package name

self.inputFlag = False # for model with input quantity

self.simulationFlag = False # if the model is simulated?

self.outputFlag = False

self.csvFile = '' # for storing inputs condition

self.resultfile="" # for storing result file

self.variableFilter = variableFilter

if not os.path.exists(self.fileName): # if file does not eixt

print("File Error:" + os.path.abspath(self.fileName) + " does not exist!!!")

return

## set default command Line Options for linearization as

## linearize() will use the simulation executable and runtime

## flag -l to perform linearization

self.getconn.sendExpression("setCommandLineOptions(\"--linearizationDumpLanguage=python\")")

self.getconn.sendExpression("setCommandLineOptions(\"--generateSymbolicLinearization\")")

self.loadingModel()

def __del__(self):

OMCSessionBase.__del__(self)

# for loading file/package, loading model and building model

def loadingModel(self):

# load file

loadFileExp="".join(["loadFile(","\"",self.fileName,"\"",")"]).replace("\\","/")

loadMsg = self.getconn.sendExpression(loadFileExp)

if not loadMsg:

return print(self.getconn.sendExpression("getErrorString()"))

# load Modelica standard libraries or Modelica files if needed

for element in self.lmodel:

if element is not None:

loadmodelError = ''

if isinstance(element, str):

if element.endswith(".mo"):

loadModelResult = self.requestApi("loadFile", element)

loadmodelError = self.requestApi('getErrorString')

else:

loadModelResult = self.requestApi("loadModel", element)

loadmodelError = self.requestApi('getErrorString')

elif isinstance(element, tuple):

if not element[1]:

libname = "".join(["loadModel(", element[0], ")"])

else:

libname = "".join(["loadModel(", element[0], ", ", "{", "\"", element[1], "\"", "}", ")"])

loadmodelError = self.sendExpression(libname)

loadmodelError = self.sendExpression("getErrorString()")

else:

print("| info | loadLibrary() failed, Unknown type detected: ", element , " is of type ", type(element), ", The following patterns are supported\n1)[\"Modelica\"]\n2)[(\"Modelica\",\"3.2.3\"), \"PowerSystems\"]\n")

if loadmodelError:

print(loadmodelError)

# create a unique temp directory for each session and build the model in that directory

self.tempdir = tempfile.mkdtemp()

if not os.path.exists(self.tempdir):

return print(self.tempdir, " cannot be created")

exp="".join(["cd(","\"",self.tempdir,"\"",")"]).replace("\\","/")

self.getconn.sendExpression(exp)

self.buildModel()

def buildModel(self, variableFilter=None):

if variableFilter is not None:

self.variableFilter = variableFilter

if self.variableFilter is not None:

varFilter = "variableFilter=" + "\"" + self.variableFilter + "\""

else:

varFilter = "variableFilter=" + "\".*""\""

# print(varFilter)

# buildModelResult=self.getconn.sendExpression("buildModel("+ mName +")")

buildModelResult = self.requestApi("buildModel", self.modelName, properties=varFilter)

buildModelError = self.requestApi("getErrorString")

# Issue #145. Always print the getErrorString since it might contains build warnings.

if buildModelError:

print(buildModelError)

if ('' in buildModelResult):

return

self.xmlFile=os.path.join(os.path.dirname(buildModelResult[0]),buildModelResult[1]).replace("\\","/")

self.xmlparse()

def sendExpression(self,expr,parsed=True):

return self.getconn.sendExpression(expr,parsed)

# request to OMC

def requestApi(self, apiName, entity=None, properties=None): # 2

if (entity is not None and properties is not None):

exp = '{}({}, {})'.format(apiName, entity, properties)

elif entity is not None and properties is None:

if (apiName == "loadFile" or apiName == "importFMU"):

exp = '{}("{}")'.format(apiName, entity)

else:

exp = '{}({})'.format(apiName, entity)

else:

exp = '{}()'.format(apiName)

try:

res = self.getconn.sendExpression(exp)

except Exception as e:

print(e)

res = None

return res

def xmlparse(self):

if(os.path.exists(self.xmlFile)):

self.tree = ET.parse(self.xmlFile)

self.root = self.tree.getroot()

rootCQ = self.root

for attr in rootCQ.iter('DefaultExperiment'):

self.simulateOptions["startTime"]= attr.get('startTime')

self.simulateOptions["stopTime"] = attr.get('stopTime')

self.simulateOptions["stepSize"] = attr.get('stepSize')

self.simulateOptions["tolerance"] = attr.get('tolerance')

self.simulateOptions["solver"] = attr.get('solver')

self.simulateOptions["outputFormat"] = attr.get('outputFormat')

for sv in rootCQ.iter('ScalarVariable'):

scalar={}

scalar["name"] = sv.get('name')

scalar["changeable"] = sv.get('isValueChangeable')

scalar["description"] = sv.get('description')

scalar["variability"] = sv.get('variability')

scalar["causality"] = sv.get('causality')

scalar["alias"] = sv.get('alias')

scalar["aliasvariable"] = sv.get('aliasVariable')

ch = list(sv)

start = None

min = None

max = None

for att in ch:

start = att.get('start')

min = att.get('min')

max = att.get('max')

scalar["start"] =start

scalar["min"] = min

scalar["max"] = max

if(scalar["variability"]=="parameter"):

if scalar["name"] in self.overridevariables:

self.paramlist[scalar["name"]] = self.overridevariables[scalar["name"]]

else:

self.paramlist[scalar["name"]] = scalar["start"]

if(scalar["variability"]=="continuous"):

self.continuouslist[scalar["name"]]=scalar["start"]

if(scalar["causality"]=="input"):

self.inputlist[scalar["name"]]=scalar["start"]

if(scalar["causality"]=="output"):

self.outputlist[scalar["name"]]=scalar["start"]

self.quantitiesList.append(scalar)

else:

print("Error: ! XML file not generated: " + self.xmlFile)