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, session: OMCSessionZMQ, readonly: bool = False):

if not isinstance(session, OMCSessionZMQ):

raise OMCSessionException("Invalid session definition!")

self._session = session

self._readonly = readonly

self._omc_cache: dict[tuple[str, bool], Any] = {}

def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: bool = True):

if opt is None:

expression = question

elif isinstance(opt, list):

expression = f"{question}({','.join([str(x) for x in opt])})"

else:

raise OMCSessionException(f"Invalid definition of options for {repr(question)}: {repr(opt)}")

p = (expression, parsed)

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

# can use cache if readonly

if p in self._omc_cache:

return self._omc_cache[p]

try:

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

except OMCSessionException as ex:

raise OMCSessionException("OMC _ask() failed: %s (parsed=%s)", (expression, parsed)) from ex

# 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(question='loadFile', opt=[f'"{filename}"'])

def loadModel(self, className):

return self._ask(question='loadModel', opt=[className])

def isModel(self, className):

return self._ask(question='isModel', opt=[className])

def isPackage(self, className):

return self._ask(question='isPackage', opt=[className])

def isPrimitive(self, className):

return self._ask(question='isPrimitive', opt=[className])

def isConnector(self, className):

return self._ask(question='isConnector', opt=[className])

def isRecord(self, className):

return self._ask(question='isRecord', opt=[className])

def isBlock(self, className):

return self._ask(question='isBlock', opt=[className])

def isType(self, className):

return self._ask(question='isType', opt=[className])

def isFunction(self, className):

return self._ask(question='isFunction', opt=[className])

def isClass(self, className):

return self._ask(question='isClass', opt=[className])

def isParameter(self, className):

return self._ask(question='isParameter', opt=[className])

def isConstant(self, className):

return self._ask(question='isConstant', opt=[className])

def isProtected(self, className):

return self._ask(question='isProtected', opt=[className])

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

return self._ask(question='getPackages', opt=[className])

def getClassRestriction(self, className):

return self._ask(question='getClassRestriction', opt=[className])

def getDerivedClassModifierNames(self, className):

return self._ask(question='getDerivedClassModifierNames', opt=[className])

def getDerivedClassModifierValue(self, className, modifierName):

return self._ask(question='getDerivedClassModifierValue', opt=[className, modifierName])

def typeNameStrings(self, className):

return self._ask(question='typeNameStrings', opt=[className])

def getComponents(self, className):

return self._ask(question='getComponents', opt=[className])

def getClassComment(self, className):

try:

return self._ask(question='getClassComment', opt=[className])

except pyparsing.ParseException as ex:

logger.warning("Method 'getClassComment(%s)' failed; OMTypedParser error: %s",

className, ex.msg)

return 'No description available'

except OMCSessionException:

raise

def getNthComponent(self, className, comp_id):

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

return self._ask(question='getNthComponent', opt=[className, comp_id])

def getNthComponentAnnotation(self, className, comp_id):

return self._ask(question='getNthComponentAnnotation', opt=[className, comp_id])

def getImportCount(self, className):

return self._ask(question='getImportCount', opt=[className])

def getNthImport(self, className, importNumber):

# [Path, id, kind]

return self._ask(question='getNthImport', opt=[className, importNumber])

def getInheritanceCount(self, className):

return self._ask(question='getInheritanceCount', opt=[className])

def getNthInheritedClass(self, className, inheritanceDepth):

return self._ask(question='getNthInheritedClass', opt=[className, inheritanceDepth])

def getParameterNames(self, className):

try:

return self._ask(question='getParameterNames', opt=[className])

except KeyError as ex:

logger.warning('OMPython error: %s', ex)

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

return []

except OMCSessionException:

raise

def getParameterValue(self, className, parameterName):

try:

return self._ask(question='getParameterValue', opt=[className, parameterName])

except pyparsing.ParseException as ex:

logger.warning("Method 'getParameterValue(%s, %s)' failed; OMTypedParser error: %s",

className, parameterName, ex.msg)

return ""

except OMCSessionException:

raise

def getComponentModifierNames(self, className, componentName):

return self._ask(question='getComponentModifierNames', opt=[className, componentName])

def getComponentModifierValue(self, className, componentName):

return self._ask(question='getComponentModifierValue', opt=[className, componentName])

def getExtendsModifierNames(self, className, componentName):

return self._ask(question='getExtendsModifierNames', opt=[className, componentName])

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

return self._ask(question='getExtendsModifierValue', opt=[className, extendsName, modifierName])

def getNthComponentModification(self, className, comp_id):

# FIXME: OMPython exception Results KeyError exception

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

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

value = self._ask(question='getNthComponentModification', opt=[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):

opt = [className] if className else [] + [f'recursive={str(recursive).lower()}',

f'qualified={str(qualified).lower()}',

f'sort={str(sort).lower()}',

f'builtin={str(builtin).lower()}',

f'showProtected={str(showProtected).lower()}']

def __init__(

self,

timeout: float = 10.00,

omhome: Optional[str] = None,

omc_process: Optional[OMCProcess] = None,

) -> None:

"""

self._timeout = timeout

if omc_process is None:

omc_process = OMCProcessLocal(omhome=omhome, timeout=timeout)

elif not isinstance(omc_process, OMCProcess):

raise OMCSessionException("Invalid definition of the OMC process!")

self.omc_process = omc_process

port = self.omc_process.get_port()

if not isinstance(port, str):

raise OMCSessionException(f"Invalid content for port: {port}")

# Create the ZeroMQ socket and connect to OMC server

context = zmq.Context.instance()

omc = context.socket(zmq.REQ)

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

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

omc.connect(port)

self.omc_zmq: Optional[zmq.Socket[bytes]] = omc

# variables to store compiled re expressions use in self.sendExpression()

self._re_log_entries: Optional[re.Pattern[str]] = None

self._re_log_raw: Optional[re.Pattern[str]] = None

def __del__(self):

if isinstance(self.omc_zmq, zmq.Socket):

try:

self.sendExpression("quit()")

except OMCSessionException:

pass

del self.omc_zmq

self.omc_zmq = None

def execute(self, command: str):

warnings.warn("This function is depreciated and will be removed in future versions; "

"please use sendExpression() instead", DeprecationWarning, stacklevel=2)

return self.sendExpression(command, parsed=False)

def sendExpression(self, command: str, parsed: bool = True) -> Any:

if self.omc_zmq is None:

raise OMCSessionException("No OMC running. Create a new instance of OMCSessionZMQ!")

logger.debug("sendExpression(%r, parsed=%r)", command, parsed)

attempts = 0

while True:

try:

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

break

except zmq.error.Again:

pass

attempts += 1

if attempts >= 50:

raise OMCSessionException(f"No connection with OMC (timeout={self._timeout}). "

f"Log-file says: \n{self.omc_process.get_log()}")

time.sleep(self._timeout / 50.0)

if command == "quit()":

self.omc_zmq.close()

self.omc_zmq = None

return None

result = self.omc_zmq.recv_string()

if command == "getErrorString()":

# no error handling if 'getErrorString()' is called

if parsed:

logger.warning("Result of 'getErrorString()' cannot be parsed!")

return result

if command == "getMessagesStringInternal()":

# no error handling if 'getMessagesStringInternal()' is called

if parsed:

logger.warning("Result of 'getMessagesStringInternal()' cannot be parsed!")

return result

# always check for error

self.omc_zmq.send_string('getMessagesStringInternal()', flags=zmq.NOBLOCK)

error_raw = self.omc_zmq.recv_string()

# run error handling only if there is something to check

if error_raw != "{}\n":

if not self._re_log_entries:

self._re_log_entries = re.compile(pattern=r'record OpenModelica\.Scripting\.ErrorMessage'

'(.*?)'

r'end OpenModelica\.Scripting\.ErrorMessage;',

flags=re.MULTILINE | re.DOTALL)

if not self._re_log_raw:

self._re_log_raw = re.compile(

pattern=r"\s+message = \"(.*?)\",\n" # message

r"\s+kind = .OpenModelica.Scripting.ErrorKind.(.*?),\n" # kind

r"\s+level = .OpenModelica.Scripting.ErrorLevel.(.*?),\n" # level

r"\s+id = (.*?)" # id

"(,\n|\n)", # end marker

flags=re.MULTILINE | re.DOTALL)

# extract all ErrorMessage records

log_entries = self._re_log_entries.findall(string=error_raw)

for log_entry in reversed(log_entries):

log_raw = self._re_log_raw.findall(string=log_entry)

if len(log_raw) != 1 or len(log_raw[0]) != 5:

logger.warning("Invalid ErrorMessage record returned by 'getMessagesStringInternal()':"

f" {repr(log_entry)}!")

continue

log_message = log_raw[0][0].encode().decode('unicode_escape')

log_kind = log_raw[0][1]

log_level = log_raw[0][2]

log_id = log_raw[0][3]

msg = (f"[OMC log for 'sendExpression({command}, {parsed})']: "

f"[{log_kind}:{log_level}:{log_id}] {log_message}")

# response according to the used log level

# see: https://build.openmodelica.org/Documentation/OpenModelica.Scripting.ErrorLevel.html

if log_level == 'error':

raise OMCSessionException(msg)

elif log_level == 'warning':

logger.warning(msg)

elif log_level == 'notification':

logger.info(msg)

else: # internal

logger.debug(msg)

if parsed is False:

return result

try:

return om_parser_typed(result)

except pyparsing.ParseException as ex:

logger.warning('OMTypedParser error: %s. Returning the basic parser result.', ex.msg)

try:

return om_parser_basic(result)

except (TypeError, UnboundLocalError) as ex:

def __init__(

self,

timeout: float = 10.00,

**kwargs,

) -> None:

super().__init__(**kwargs)

# store variables

self._timeout = timeout

# omc process

self._omc_process: Optional[subprocess.Popen] = None

# omc ZMQ port to use

self._omc_port: Optional[str] = None

# generate a random string for this session

self._random_string = uuid.uuid4().hex

# omc port and log file

self._omc_filebase = f"openmodelica.{self._random_string}"

# get a temporary directory

self._temp_dir = pathlib.Path(tempfile.gettempdir())

# setup log file - this file must be closed in the destructor

logfile = self._temp_dir / (self._omc_filebase + ".log")

self._omc_loghandle: Optional[io.TextIOWrapper] = None

try:

self._omc_loghandle = open(file=logfile, mode="w+", encoding="utf-8")

except OSError as ex:

raise OMCSessionException(f"Cannot open log file {logfile}.") from ex

self._re_portfile_path = re.compile(pattern=r'\nDumped server port in file: (.*?)($|\n)',

flags=re.MULTILINE | re.DOTALL)

def __del__(self):

if self._omc_loghandle is not None:

try:

self._omc_loghandle.close()

except (OSError, IOError):

pass

self._omc_loghandle = None

if isinstance(self._omc_process, subprocess.Popen):

try:

self._omc_process.wait(timeout=2.0)

except subprocess.TimeoutExpired:

if self._omc_process:

logger.warning("OMC did not exit after being sent the quit() command; "

"killing the process with pid=%s", self._omc_process.pid)

self._omc_process.kill()

self._omc_process.wait()

finally:

self._omc_process = None

def get_port(self) -> Optional[str]:

if not isinstance(self._omc_port, str):

raise OMCSessionException(f"Invalid port to connect to OMC process: {self._omc_port}")

return self._omc_port

def get_log(self) -> str:

if self._omc_loghandle is None:

raise OMCSessionException("Log file not available!")

self._omc_loghandle.seek(0)

log = self._omc_loghandle.read()

return log

def _get_portfile_path(self) -> Optional[pathlib.Path]:

omc_log = self.get_log()

portfile = self._re_portfile_path.findall(string=omc_log)

portfile_path = None

if portfile:

portfile_path = pathlib.Path(portfile[-1][0])

def __init__(

self,

omc_port: str,

) -> None:

super().__init__()

def __init__(

self,

timeout: float = 10.00,

omhome: Optional[str] = None,

) -> None:

super().__init__(timeout=timeout)

# where to find OpenModelica

self._omhome = self._omc_home_get(omhome=omhome)

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

self._omc_process = self._omc_process_get()

# connect to the running omc instance using ZMQ

self._omc_port = self._omc_port_get()

@staticmethod

def _omc_home_get(omhome: Optional[str] = None) -> pathlib.Path:

# use the provided path

if omhome is not None:

return pathlib.Path(omhome)

# check the environment variable

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

if omhome is not None:

return pathlib.Path(omhome)

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

path_to_omc = shutil.which("omc")

if path_to_omc is not None:

return pathlib.Path(path_to_omc).parents[1]

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

def _omc_process_get(self) -> subprocess.Popen:

my_env = os.environ.copy()

my_env["PATH"] = (self._omhome / "bin").as_posix() + os.pathsep + my_env["PATH"]

omc_command = [

(self._omhome / "bin" / "omc").as_posix(),

"--locale=C",

"--interactive=zmq",

f"-z={self._random_string}"]

omc_process = subprocess.Popen(omc_command,

stdout=self._omc_loghandle,

stderr=self._omc_loghandle,

env=my_env)

return omc_process

def _omc_port_get(self) -> str:

port = None

# See if the omc server is running

attempts = 0

while True:

omc_portfile_path = self._get_portfile_path()

if omc_portfile_path is not None and omc_portfile_path.is_file():

# Read the port file

with open(file=omc_portfile_path, mode='r', encoding="utf-8") as f_p:

port = f_p.readline()

break

if port is not None:

break

attempts += 1

if attempts == 80.0:

raise OMCSessionException(f"OMC Server did not start (timeout={self._timeout}). "

f"Could not open file {omc_portfile_path}. "

f"Log-file says:\n{self.get_log()}")

time.sleep(self._timeout / 80.0)

logger.info(f"Local OMC Server is up and running at ZMQ port {port} "

f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}")

def __init__(

self,

timeout: float = 10.00,

dockerExtraArgs: Optional[list] = None,

dockerOpenModelicaPath: str = "omc",

dockerNetwork: Optional[str] = None,

port: Optional[int] = None,

) -> None:

super().__init__(timeout=timeout)

if dockerExtraArgs is None:

dockerExtraArgs = []

self._dockerExtraArgs = dockerExtraArgs

self._dockerOpenModelicaPath = dockerOpenModelicaPath

self._dockerNetwork = dockerNetwork

self._interactivePort = port

self._dockerCid: Optional[str] = None

self._docker_process: Optional[DummyPopen] = None

def _docker_process_get(self, docker_cid: str) -> Optional[DummyPopen]:

if sys.platform == 'win32':

raise NotImplementedError("Docker not supported on win32!")

docker_process = None

for idx in range(0, 40):

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

docker_process = None

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

columns = line.split()

if self._random_string in line:

try:

docker_process = DummyPopen(int(columns[1]))

except psutil.NoSuchProcess as ex:

raise OMCSessionException(f"Could not find PID {dockerTop} - "

"is this a docker instance spawned without --pid=host?") from ex

if docker_process is not None:

break

time.sleep(self._timeout / 40.0)

return docker_process

@staticmethod

def _getuid() -> int:

"""

# mypy complained about os.getuid() not being available on

# Windows, hence the type: ignore comment.

return 1000 if sys.platform == 'win32' else os.getuid() # type: ignore

def _omc_port_get(self) -> str:

port = None

if not isinstance(self._dockerCid, str):

raise OMCSessionException(f"Invalid docker container ID: {self._dockerCid}")

# See if the omc server is running

attempts = 0

while True:

omc_portfile_path = self._get_portfile_path()

if omc_portfile_path is not None:

try:

output = subprocess.check_output(args=["docker",

"exec", self._dockerCid,

"cat", omc_portfile_path.as_posix()],

stderr=subprocess.DEVNULL)

port = output.decode().strip()

except subprocess.CalledProcessError:

pass

if port is not None:

break

attempts += 1

if attempts == 80.0:

raise OMCSessionException(f"Docker based OMC Server did not start (timeout={self._timeout}). "

f"Could not open port file {omc_portfile_path}. "

f"Log-file says:\n{self.get_log()}")

time.sleep(self._timeout / 80.0)

logger.info(f"Docker based OMC Server is up and running at port {port}")

return port

def get_server_address(self) -> Optional[str]:

if self._dockerNetwork == "separate" and isinstance(self._dockerCid, str):

output = subprocess.check_output(["docker", "inspect", self._dockerCid]).decode().strip()

return json.loads(output)[0]["NetworkSettings"]["IPAddress"]

return None

def get_docker_container_id(self) -> str:

if not isinstance(self._dockerCid, str):

raise OMCSessionException(f"Invalid docker container ID: {self._dockerCid}!")

def __init__(

self,

timeout: float = 10.00,

docker: Optional[str] = None,

dockerExtraArgs: Optional[list] = None,

dockerOpenModelicaPath: str = "omc",

dockerNetwork: Optional[str] = None,

port: Optional[int] = None,

) -> None:

super().__init__(

timeout=timeout,

dockerExtraArgs=dockerExtraArgs,

dockerOpenModelicaPath=dockerOpenModelicaPath,

dockerNetwork=dockerNetwork,

port=port,

)

if docker is None:

raise OMCSessionException("Argument docker must be set!")

self._docker = docker

# start up omc executable in docker container waiting for the ZMQ connection

self._omc_process, self._docker_process, self._dockerCid = self._docker_omc_start()

# connect to the running omc instance using ZMQ

self._omc_port = self._omc_port_get()

def __del__(self) -> None:

super().__del__()

if isinstance(self._docker_process, DummyPopen):

try:

self._docker_process.wait(timeout=2.0)

except subprocess.TimeoutExpired:

if self._docker_process:

logger.warning("OMC did not exit after being sent the quit() command; "

"killing the process with pid=%s", self._docker_process.pid)

self._docker_process.kill()

self._docker_process.wait(timeout=2.0)

finally:

self._docker_process = None

def _docker_omc_cmd(

self,

omc_path_and_args_list: list[str],

docker_cid_file: pathlib.Path,

) -> list:

"""

extraFlags = []

if sys.platform == "win32":

extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"]

if not self._interactivePort:

raise OMCSessionException("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.")

if sys.platform == "win32":

if isinstance(self._interactivePort, str):

port = int(self._interactivePort)

elif isinstance(self._interactivePort, int):

port = self._interactivePort

else:

raise OMCSessionException("Missing or invalid interactive port!")

dockerNetworkStr = ["-p", f"127.0.0.1:{port}:{port}"]

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

dockerNetworkStr = ["--network=host"]

elif self._dockerNetwork == "separate":

dockerNetworkStr = []

extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"]

else:

raise OMCSessionException(f'dockerNetwork was set to {self._dockerNetwork}, '

'but only \"host\" or \"separate\" is allowed')

if isinstance(self._interactivePort, int):

extraFlags = extraFlags + [f"--interactivePort={int(self._interactivePort)}"]

omc_command = ([

"docker", "run",

"--cidfile", docker_cid_file.as_posix(),

"--rm",

"--user", str(self._getuid()),

]

+ self._dockerExtraArgs

+ dockerNetworkStr

+ [self._docker, self._dockerOpenModelicaPath]

+ omc_path_and_args_list

+ extraFlags)

return omc_command

def _docker_omc_start(self) -> Tuple[subprocess.Popen, DummyPopen, str]:

my_env = os.environ.copy()

docker_cid_file = self._temp_dir / (self._omc_filebase + ".docker.cid")

omc_command = self._docker_omc_cmd(

omc_path_and_args_list=["--locale=C",

"--interactive=zmq",

f"-z={self._random_string}"],

docker_cid_file=docker_cid_file,

)

omc_process = subprocess.Popen(omc_command,

stdout=self._omc_loghandle,

stderr=self._omc_loghandle,

env=my_env)

if not isinstance(docker_cid_file, pathlib.Path):

raise OMCSessionException(f"Invalid content for docker container ID file path: {docker_cid_file}")

docker_cid = None

for idx in range(0, 40):

try:

with open(file=docker_cid_file, mode="r", encoding="utf-8") as fh:

docker_cid = fh.read().strip()

except IOError:

pass

if docker_cid:

break

time.sleep(self._timeout / 40.0)

if docker_cid is None:

logger.error(f"Docker did not start. Log-file says:\n{self.get_log()}")

raise OMCSessionException(f"Docker did not start (timeout={self._timeout} might be too short "

"especially if you did not docker pull the image before this command).")

docker_process = self._docker_process_get(docker_cid=docker_cid)

if docker_process is None:

raise OMCSessionException(f"Docker top did not contain omc process {self._random_string}. "

f"Log-file says:\n{self.get_log()}")

def __init__(

self,

timeout: float = 10.00,

dockerContainer: Optional[str] = None,

dockerExtraArgs: Optional[list] = None,

dockerOpenModelicaPath: str = "omc",

dockerNetwork: Optional[str] = None,

port: Optional[int] = None,

) -> None:

super().__init__(

timeout=timeout,

dockerExtraArgs=dockerExtraArgs,

dockerOpenModelicaPath=dockerOpenModelicaPath,

dockerNetwork=dockerNetwork,

port=port,

)

if not isinstance(dockerContainer, str):

raise OMCSessionException("Argument dockerContainer must be set!")

self._dockerCid = dockerContainer

# start up omc executable in docker container waiting for the ZMQ connection

self._omc_process, self._docker_process = self._docker_omc_start()

# connect to the running omc instance using ZMQ

self._omc_port = self._omc_port_get()

def __del__(self) -> None:

super().__del__()

# docker container ID was provided - do NOT kill the docker process!

self._docker_process = None

def _docker_omc_cmd(self, omc_path_and_args_list) -> list:

"""

extraFlags: list[str] = []

if sys.platform == "win32":

extraFlags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"]

if not self._interactivePort:

raise OMCSessionException("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.")

if isinstance(self._interactivePort, int):

extraFlags = extraFlags + [f"--interactivePort={int(self._interactivePort)}"]

omc_command = ([

"docker", "exec",

"--user", str(self._getuid()),

]

+ self._dockerExtraArgs

+ [self._dockerCid, self._dockerOpenModelicaPath]

+ omc_path_and_args_list

+ extraFlags)

return omc_command

def _docker_omc_start(self) -> Tuple[subprocess.Popen, DummyPopen]:

my_env = os.environ.copy()

omc_command = self._docker_omc_cmd(

omc_path_and_args_list=["--locale=C",

"--interactive=zmq",

f"-z={self._random_string}"],

)

omc_process = subprocess.Popen(omc_command,

stdout=self._omc_loghandle,

stderr=self._omc_loghandle,

env=my_env)

docker_process = None

if isinstance(self._dockerCid, str):

docker_process = self._docker_process_get(docker_cid=self._dockerCid)

if docker_process is None:

raise OMCSessionException(f"Docker top did not contain omc process {self._random_string} "

f"/ {self._dockerCid}. Log-file says:\n{self.get_log()}")

def __init__(

self,

timeout: float = 10.00,

wsl_omc: str = 'omc',

wsl_distribution: Optional[str] = None,

wsl_user: Optional[str] = None,

) -> None:

super().__init__(timeout=timeout)

# get wsl base command

self._wsl_cmd = ['wsl']

if isinstance(wsl_distribution, str):

self._wsl_cmd += ['--distribution', wsl_distribution]

if isinstance(wsl_user, str):

self._wsl_cmd += ['--user', wsl_user]

self._wsl_cmd += ['--']

# where to find OpenModelica

self._wsl_omc = wsl_omc

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

self._omc_process = self._omc_process_get()

# connect to the running omc instance using ZMQ

self._omc_port = self._omc_port_get()

def _omc_process_get(self) -> subprocess.Popen:

my_env = os.environ.copy()

omc_command = self._wsl_cmd + [

self._wsl_omc,

"--locale=C",

"--interactive=zmq",

f"-z={self._random_string}"]

omc_process = subprocess.Popen(omc_command,

stdout=self._omc_loghandle,

stderr=self._omc_loghandle,

env=my_env)

return omc_process

def _omc_port_get(self) -> str:

omc_portfile_path: Optional[pathlib.Path] = None

port = None

# See if the omc server is running

attempts = 0

while True:

try:

omc_portfile_path = self._get_portfile_path()

if omc_portfile_path is not None:

output = subprocess.check_output(

args=self._wsl_cmd + ["cat", omc_portfile_path.as_posix()],

stderr=subprocess.DEVNULL,

)

port = output.decode().strip()

except subprocess.CalledProcessError:

pass

if port is not None:

break