From 043fb276ca9995afa8702b51ca3d73b828a1cafb Mon Sep 17 00:00:00 2001 From: syntron Date: Sun, 8 Feb 2026 20:52:18 +0100 Subject: [PATCH 1/3] (E001) update tests (v4.x.x) [test_*] reorder imports [tests_ModelicaDoE*] fix pylint hint * use .items() [tests_*] use OMSessionABC.get_version() [test_ModelicaSystemCmd] use get_model_name() instead of access to private variable _model_name [test_ModelicaSystemOMC] read file using utf-8 encoding / linter fix [test_ModelicaSystemRunner] update test case * ModelicaSystemRunner & OMCPath * ModelicaSystemRunner & OMPathRunnerLocal * ModelicaSystemRunner & OMPathRunnerBash * ModelicaSystemRunner & OMPathRunnerBash using docker * ModelicaSystemRunner & OMPathRunnerBash using WSL (not tested!) [test_OMCPath] update test case * OMCPath & OMCSessionZMQ * OMCPath & OMCSessionLocal * OMCPath & OMCSessionDocker * OMCPath & OMCSessionWSL (not tested!) * OMPathLocal & OMCSessionRunner * OMPathBash & OMCSessionRunner * OMPathBash & OMCSessionRunner in docker * OMPathBash & OMCSessionRunner in WSL (not tested!) add workflow to run unittests in ./tests [test_OMParser] use only the public interface => om_parser_basic() [test_OMTypedParser] rename file / use om_parser_typed() update tests - do NOT run test_FMIRegression.py reason: * it is only a test for OMC / not OMPython specific * furthermore, it is run automatically via cron job (= FMITest) [test_ModelExecutionCmd] rename from test_ModelicaSystemCmd --- OMPython/__init__.py | 2 + tests/test_FMIExport.py | 2 +- ...SystemCmd.py => test_ModelExecutionCmd.py} | 2 +- tests/test_ModelicaDoEOMC.py | 6 +- tests/test_ModelicaDoERunner.py | 6 +- tests/test_ModelicaSystemOMC.py | 2 +- tests/test_ModelicaSystemRunner.py | 176 +++++++++++++++++- tests/test_OMCPath.py | 96 +++++++--- tests/test_OMParser.py | 53 ++++-- tests/test_OMTypedParser.py | 65 +++++++ tests/test_ZMQ.py | 1 + tests/test_docker.py | 2 + tests/test_typedParser.py | 53 ------ 13 files changed, 365 insertions(+), 101 deletions(-) rename tests/{test_ModelicaSystemCmd.py => test_ModelExecutionCmd.py} (97%) create mode 100644 tests/test_OMTypedParser.py delete mode 100644 tests/test_typedParser.py diff --git a/OMPython/__init__.py b/OMPython/__init__.py index c12f8524..22c88137 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -30,6 +30,7 @@ OMPathABC, OMCPath, + OMSessionABC, OMSessionRunner, OMCSessionABC, @@ -77,6 +78,7 @@ 'OMPathABC', 'OMCPath', + 'OMSessionABC', 'OMSessionRunner', 'OMCSessionABC', diff --git a/tests/test_FMIExport.py b/tests/test_FMIExport.py index c7ab038a..65ac2766 100644 --- a/tests/test_FMIExport.py +++ b/tests/test_FMIExport.py @@ -1,6 +1,6 @@ -import shutil import os import pathlib +import shutil import OMPython diff --git a/tests/test_ModelicaSystemCmd.py b/tests/test_ModelExecutionCmd.py similarity index 97% rename from tests/test_ModelicaSystemCmd.py rename to tests/test_ModelExecutionCmd.py index 3d35376b..db5aadeb 100644 --- a/tests/test_ModelicaSystemCmd.py +++ b/tests/test_ModelExecutionCmd.py @@ -29,7 +29,7 @@ def mscmd_firstorder(model_firstorder): cmd_local=mod.get_session().model_execution_local, cmd_windows=mod.get_session().model_execution_windows, cmd_prefix=mod.get_session().model_execution_prefix(cwd=mod.getWorkDirectory()), - model_name=mod._model_name, + model_name=mod.get_model_name(), ) return mscmd diff --git a/tests/test_ModelicaDoEOMC.py b/tests/test_ModelicaDoEOMC.py index 143932fc..9d6afc63 100644 --- a/tests/test_ModelicaDoEOMC.py +++ b/tests/test_ModelicaDoEOMC.py @@ -159,6 +159,6 @@ def _run_ModelicaDoEOMC(doe_mod): f"y[{row['p']}]": float(row['b']), } - for var in var_dict: - assert var in sol['data'] - assert np.isclose(sol['data'][var][-1], var_dict[var]) + for key, val in var_dict.items(): + assert key in sol['data'] + assert np.isclose(sol['data'][key][-1], val) diff --git a/tests/test_ModelicaDoERunner.py b/tests/test_ModelicaDoERunner.py index 2d41315f..e29e7e05 100644 --- a/tests/test_ModelicaDoERunner.py +++ b/tests/test_ModelicaDoERunner.py @@ -153,6 +153,6 @@ def _check_runner_result(mod, doe_mod): 'b': float(row['b']), } - for var in var_dict: - assert var in sol['data'] - assert np.isclose(sol['data'][var][-1], var_dict[var]) + for key, val in var_dict.items(): + assert key in sol['data'] + assert np.isclose(sol['data'][key][-1], val) diff --git a/tests/test_ModelicaSystemOMC.py b/tests/test_ModelicaSystemOMC.py index 8dd17ef0..c63b92e1 100644 --- a/tests/test_ModelicaSystemOMC.py +++ b/tests/test_ModelicaSystemOMC.py @@ -495,7 +495,7 @@ def test_simulate_inputs(tmp_path): } mod.setInputs(**inputs) csv_file = mod._createCSVData() - assert pathlib.Path(csv_file).read_text() == """time,u1,u2,end + assert pathlib.Path(csv_file).read_text(encoding='utf-8') == """time,u1,u2,end 0.0,0.0,0.0,0 0.25,0.25,0.5,0 0.5,0.5,1.0,0 diff --git a/tests/test_ModelicaSystemRunner.py b/tests/test_ModelicaSystemRunner.py index ec9d734d..a207368c 100644 --- a/tests/test_ModelicaSystemRunner.py +++ b/tests/test_ModelicaSystemRunner.py @@ -1,9 +1,22 @@ +import sys + import numpy as np import pytest import OMPython +skip_on_windows = pytest.mark.skipif( + sys.platform.startswith("win"), + reason="OpenModelica Docker image is Linux-only; skipping on Windows.", +) + +skip_python_older_312 = pytest.mark.skipif( + sys.version_info < (3, 12), + reason="OMCPath(non-local) only working for Python >= 3.12.", +) + + @pytest.fixture def model_firstorder_content(): return """ @@ -37,7 +50,7 @@ def param(): } -def test_runner(model_firstorder, param): +def test_ModelicaSystemRunner_OMC(model_firstorder, param): # create a model using ModelicaSystem mod = OMPython.ModelicaSystemOMC() mod.model( @@ -71,6 +84,167 @@ def test_runner(model_firstorder, param): _check_result(mod=mod, resultfile=resultfile_modr, param=param) +def test_ModelicaSystemRunner_local(model_firstorder, param): + # create a model using ModelicaSystem + mod = OMPython.ModelicaSystemOMC() + mod.model( + model_file=model_firstorder, + model_name="M", + ) + + resultfile_mod = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_mod.mat" + _run_simulation(mod=mod, resultfile=resultfile_mod, param=param) + + # run the model using only the runner class + omcs = OMPython.OMSessionRunner( + version=mod.get_session().get_version(), + ompath_runner=OMPython.OMPathRunnerLocal, + ) + modr = OMPython.ModelicaSystemRunner( + session=omcs, + work_directory=mod.getWorkDirectory(), + ) + modr.setup( + model_name="M", + ) + + resultfile_modr = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_modr.mat" + _run_simulation(mod=modr, resultfile=resultfile_modr, param=param) + + # cannot check the content as runner does not have the capability to open a result file + assert resultfile_mod.size() == resultfile_modr.size() + + # check results + _check_result(mod=mod, resultfile=resultfile_mod, param=param) + _check_result(mod=mod, resultfile=resultfile_modr, param=param) + + +@skip_on_windows +def test_ModelicaSystemRunner_bash(model_firstorder, param): + # create a model using ModelicaSystem + mod = OMPython.ModelicaSystemOMC() + mod.model( + model_file=model_firstorder, + model_name="M", + ) + + resultfile_mod = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_mod.mat" + _run_simulation(mod=mod, resultfile=resultfile_mod, param=param) + + # run the model using only the runner class + omcsr = OMPython.OMSessionRunner( + version=mod.get_session().get_version(), + ompath_runner=OMPython.OMPathRunnerBash, + ) + modr = OMPython.ModelicaSystemRunner( + session=omcsr, + work_directory=mod.getWorkDirectory(), + ) + modr.setup( + model_name="M", + ) + + resultfile_modr = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_modr.mat" + _run_simulation(mod=modr, resultfile=resultfile_modr, param=param) + + # cannot check the content as runner does not have the capability to open a result file + assert resultfile_mod.size() == resultfile_modr.size() + + # check results + _check_result(mod=mod, resultfile=resultfile_mod, param=param) + _check_result(mod=mod, resultfile=resultfile_modr, param=param) + + +@skip_on_windows +@skip_python_older_312 +def test_ModelicaSystemRunner_bash_docker(model_firstorder, param): + omcs = OMPython.OMCSessionDocker(docker="openmodelica/openmodelica:v1.25.0-minimal") + omversion = omcs.sendExpression("getVersion()") + assert isinstance(omversion, str) and omversion.startswith("OpenModelica") + + # create a model using ModelicaSystem + mod = OMPython.ModelicaSystemOMC( + session=omcs, + ) + mod.model( + model_file=model_firstorder, + model_name="M", + ) + + resultfile_mod = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_mod.mat" + _run_simulation(mod=mod, resultfile=resultfile_mod, param=param) + + # run the model using only the runner class + omcsr = OMPython.OMSessionRunner( + version=mod.get_session().get_version(), + cmd_prefix=omcs.model_execution_prefix(cwd=mod.getWorkDirectory()), + ompath_runner=OMPython.OMPathRunnerBash, + model_execution_local=False, + ) + modr = OMPython.ModelicaSystemRunner( + session=omcsr, + work_directory=mod.getWorkDirectory(), + ) + modr.setup( + model_name="M", + ) + + resultfile_modr = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_modr.mat" + _run_simulation(mod=modr, resultfile=resultfile_modr, param=param) + + # cannot check the content as runner does not have the capability to open a result file + assert resultfile_mod.size() == resultfile_modr.size() + + # check results + _check_result(mod=mod, resultfile=resultfile_mod, param=param) + _check_result(mod=mod, resultfile=resultfile_modr, param=param) + + +@pytest.mark.skip(reason="Not able to run WSL on github") +@skip_python_older_312 +def test_ModelicaSystemDoE_WSL(tmp_path, model_doe, param_doe): + omcs = OMPython.OMCSessionWSL() + omversion = omcs.sendExpression("getVersion()") + assert isinstance(omversion, str) and omversion.startswith("OpenModelica") + + # create a model using ModelicaSystem + mod = OMPython.ModelicaSystemOMC( + session=omcs, + ) + mod.model( + model_file=model_firstorder, + model_name="M", + ) + + resultfile_mod = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_mod.mat" + _run_simulation(mod=mod, resultfile=resultfile_mod, param=param) + + # run the model using only the runner class + omcsr = OMPython.OMSessionRunner( + version=mod.get_session().get_version(), + cmd_prefix=omcs.model_execution_prefix(cwd=mod.getWorkDirectory()), + ompath_runner=OMPython.OMPathRunnerBash, + model_execution_local=False, + ) + modr = OMPython.ModelicaSystemRunner( + session=omcsr, + work_directory=mod.getWorkDirectory(), + ) + modr.setup( + model_name="M", + ) + + resultfile_modr = mod.getWorkDirectory() / f"{mod.get_model_name()}_res_modr.mat" + _run_simulation(mod=modr, resultfile=resultfile_modr, param=param) + + # cannot check the content as runner does not have the capability to open a result file + assert resultfile_mod.size() == resultfile_modr.size() + + # check results + _check_result(mod=mod, resultfile=resultfile_mod, param=param) + _check_result(mod=mod, resultfile=resultfile_modr, param=param) + + def _run_simulation(mod, resultfile, param): simOptions = {"stopTime": param['stopTime'], "stepSize": 0.1, "tolerance": 1e-8} mod.setSimulationOptions(**simOptions) diff --git a/tests/test_OMCPath.py b/tests/test_OMCPath.py index df01b86a..e15c75ff 100644 --- a/tests/test_OMCPath.py +++ b/tests/test_OMCPath.py @@ -15,42 +15,98 @@ ) -def test_OMCPath_OMCProcessLocal(): - omcs = OMPython.OMCSessionLocal() +# TODO: based on compatibility layer +def test_OMCPath_OMCSessionZMQ(): + om = OMPython.OMCSessionZMQ() - _run_OMCPath_checks(omcs) + _run_OMPath_checks(om) + _run_OMPath_write_file(om) - del omcs + +def test_OMCPath_OMCSessionLocal(): + oms = OMPython.OMCSessionLocal() + + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) @skip_on_windows @skip_python_older_312 -def test_OMCPath_OMCProcessDocker(): +def test_OMCPath_OMCSessionDocker(): omcs = OMPython.OMCSessionDocker(docker="openmodelica/openmodelica:v1.25.0-minimal") omversion = omcs.sendExpression("getVersion()") assert isinstance(omversion, str) and omversion.startswith("OpenModelica") - _run_OMCPath_checks(omcs) - - del omcs + _run_OMPath_checks(omcs) + _run_OMPath_write_file(omcs) @pytest.mark.skip(reason="Not able to run WSL on github") @skip_python_older_312 -def test_OMCPath_OMCProcessWSL(): - omcs = OMPython.OMCSessionWSL( +def test_OMCPath_OMCSessionWSL(): + oms = OMPython.OMCSessionWSL( wsl_omc='omc', wsl_user='omc', timeout=30.0, ) - _run_OMCPath_checks(omcs) + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) + + +@skip_python_older_312 +def test_OMPathLocal_OMSessionRunner(): + oms = OMPython.OMSessionRunner() + + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) + + +@skip_on_windows +@skip_python_older_312 +def test_OMPathBash_OMSessionRunner(): + oms = OMPython.OMSessionRunner( + ompath_runner=OMPython.OMPathRunnerBash, + ) + + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) + + +@skip_on_windows +@skip_python_older_312 +def test_OMPathBash_OMSessionRunner_Docker(): + oms_docker = OMPython.OMCSessionDocker(docker="openmodelica/openmodelica:v1.25.0-minimal") + omversion = oms_docker.sendExpression("getVersion()") + assert isinstance(omversion, str) and omversion.startswith("OpenModelica") + + oms = OMPython.OMSessionRunner( + cmd_prefix=oms_docker.get_cmd_prefix(), + ompath_runner=OMPython.OMPathRunnerBash, + ) + + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) - del omcs +@pytest.mark.skip(reason="Not able to run WSL on github") +@skip_python_older_312 +def test_OMPathBash_OMSessionRunner_WSL(): + oms_docker = OMPython.OMCSessionWSL() + omversion = oms_docker.sendExpression("getVersion()") + assert isinstance(omversion, str) and omversion.startswith("OpenModelica") + + oms = OMPython.OMSessionRunner( + cmd_prefix=oms_docker.get_cmd_prefix(), + ompath_runner=OMPython.OMPathRunnerBash, + ) + + _run_OMPath_checks(oms) + _run_OMPath_write_file(oms) -def _run_OMCPath_checks(omcs: OMPython.OMCSessionABC): - p1 = omcs.omcpath_tempdir() + +def _run_OMPath_checks(om: OMPython.OMSessionABC): + p1 = om.omcpath_tempdir() p2 = p1 / 'test' p2.mkdir() assert p2.is_dir() @@ -59,8 +115,8 @@ def _run_OMCPath_checks(omcs: OMPython.OMCSessionABC): assert p3.write_text('test') assert p3.is_file() assert p3.size() > 0 - p3 = p3.resolve().absolute() - assert str(p3) == str((p2 / 'test.txt').resolve().absolute()) + p3 = p3.resolve() + assert str(p3) == str((p2 / 'test.txt').resolve()) assert p3.read_text() == "test" assert p3.is_file() assert p3.parent.is_dir() @@ -68,15 +124,11 @@ def _run_OMCPath_checks(omcs: OMPython.OMCSessionABC): assert p3.is_file() is False -def test_OMCPath_write_file(tmpdir): - omcs = OMPython.OMCSessionLocal() - +def _run_OMPath_write_file(om: OMPython.OMSessionABC): data = "abc # \\t # \" # \\n # xyz" - p1 = omcs.omcpath_tempdir() + p1 = om.omcpath_tempdir() p2 = p1 / 'test.txt' p2.write_text(data=data) assert data == p2.read_text() - - del omcs diff --git a/tests/test_OMParser.py b/tests/test_OMParser.py index 875604e5..9dca784d 100644 --- a/tests/test_OMParser.py +++ b/tests/test_OMParser.py @@ -1,6 +1,6 @@ -from OMPython import OMParser +import OMPython -typeCheck = OMParser.typeCheck +parser = OMPython.OMParser.om_parser_basic def test_newline_behaviour(): @@ -8,31 +8,38 @@ def test_newline_behaviour(): def test_boolean(): - assert typeCheck('TRUE') is True - assert typeCheck('True') is True - assert typeCheck('true') is True - assert typeCheck('FALSE') is False - assert typeCheck('False') is False - assert typeCheck('false') is False + assert parser('TRUE') is True + assert parser('True') is True + assert parser('true') is True + assert parser('FALSE') is False + assert parser('False') is False + assert parser('false') is False def test_int(): - assert typeCheck('2') == 2 - assert type(typeCheck('1')) == int - assert type(typeCheck('123123123123123123232323')) == int - assert type(typeCheck('9223372036854775808')) == int + assert parser('2') == 2 + assert type(parser('1')) == int + assert type(parser('123123123123123123232323')) == int + assert type(parser('9223372036854775808')) == int def test_float(): - assert type(typeCheck('1.2e3')) == float + assert type(parser('1.2e3')) == float -# def test_dict(): -# assert type(typeCheck('{"a": "b"}')) == dict +def test_dict(): + # TODO: why does it fail? + # assert type(parser('{"a": "b"}')) == dict + pass def test_ident(): - assert typeCheck('blabla2') == "blabla2" + assert parser('blabla2') == "blabla2" + + +def test_empty(): + # TODO: this differs from OMTypedParser + assert parser('') == {} def test_str(): @@ -41,3 +48,17 @@ def test_str(): def test_UnStringable(): pass + + +# def test_everything(): +# # this test used to be in OMTypedParser.py's main() +# testdata = """ +# (1.0,{{1,true,3},{"4\\" +# ",5.9,6,NONE ( )},record ABC +# startTime = ErrorLevel.warning, +# 'stop*Time' = SOME(1.0) +# end ABC;}) +# """ +# expected = (1.0, ((1, True, 3), ('4"\n', 5.9, 6, None), {"'stop*Time'": 1.0, 'startTime': 'ErrorLevel.warning'})) +# results = parser(testdata) +# assert results == expected diff --git a/tests/test_OMTypedParser.py b/tests/test_OMTypedParser.py new file mode 100644 index 00000000..94a14210 --- /dev/null +++ b/tests/test_OMTypedParser.py @@ -0,0 +1,65 @@ +import OMPython + +parser = OMPython.OMTypedParser.om_parser_typed + + +def test_newline_behaviour(): + pass + + +def test_boolean(): + # TODO: why does these fail? + # assert parser('TRUE') is True + # assert parser('True') is True + assert parser('true') is True + # TODO: why does these fail? + # assert parser('FALSE') is False + # assert parser('False') is False + assert parser('false') is False + + +def test_int(): + assert parser('2') == 2 + assert type(parser('1')) == int + assert type(parser('123123123123123123232323')) == int + assert type(parser('9223372036854775808')) == int + + +def test_float(): + assert type(parser('1.2e3')) == float + + +def test_dict(): + # TODO: why does it fail? + # assert type(parser('{"a": "b"}')) == dict + pass + + +def test_ident(): + assert parser('blabla2') == "blabla2" + + +def test_empty(): + assert parser('') is None + + +def test_str(): + pass + + +def test_UnStringable(): + pass + + +def test_everything(): + # this test used to be in OMTypedParser.py's main() + testdata = """ + (1.0,{{1,true,3},{"4\\" +",5.9,6,NONE ( )},record ABC + startTime = ErrorLevel.warning, + 'stop*Time' = SOME(1.0) +end ABC;}) + """ + expected = (1.0, ((1, True, 3), ('4"\n', 5.9, 6, None), {"'stop*Time'": 1.0, 'startTime': 'ErrorLevel.warning'})) + results = parser(testdata) + assert results == expected diff --git a/tests/test_ZMQ.py b/tests/test_ZMQ.py index 1302a79d..89a8387b 100644 --- a/tests/test_ZMQ.py +++ b/tests/test_ZMQ.py @@ -1,5 +1,6 @@ import pathlib import os + import pytest import OMPython diff --git a/tests/test_docker.py b/tests/test_docker.py index a1acfbe1..50d2763a 100644 --- a/tests/test_docker.py +++ b/tests/test_docker.py @@ -1,5 +1,7 @@ import sys + import pytest + import OMPython skip_on_windows = pytest.mark.skipif( diff --git a/tests/test_typedParser.py b/tests/test_typedParser.py deleted file mode 100644 index 8e74a556..00000000 --- a/tests/test_typedParser.py +++ /dev/null @@ -1,53 +0,0 @@ -from OMPython import OMTypedParser - -typeCheck = OMTypedParser.om_parser_typed - - -def test_newline_behaviour(): - pass - - -def test_boolean(): - assert typeCheck('true') is True - assert typeCheck('false') is False - - -def test_int(): - assert typeCheck('2') == 2 - assert type(typeCheck('1')) == int - assert type(typeCheck('123123123123123123232323')) == int - assert type(typeCheck('9223372036854775808')) == int - - -def test_float(): - assert type(typeCheck('1.2e3')) == float - - -def test_ident(): - assert typeCheck('blabla2') == "blabla2" - - -def test_empty(): - assert typeCheck('') is None - - -def test_str(): - pass - - -def test_UnStringable(): - pass - - -def test_everything(): - # this test used to be in OMTypedParser.py's main() - testdata = """ - (1.0,{{1,true,3},{"4\\" -",5.9,6,NONE ( )},record ABC - startTime = ErrorLevel.warning, - 'stop*Time' = SOME(1.0) -end ABC;}) - """ - expected = (1.0, ((1, True, 3), ('4"\n', 5.9, 6, None), {"'stop*Time'": 1.0, 'startTime': 'ErrorLevel.warning'})) - results = typeCheck(testdata) - assert results == expected From 71c74f26717e2c8f87f548d9733a055f2238d334 Mon Sep 17 00:00:00 2001 From: syntron Date: Fri, 13 Feb 2026 21:16:47 +0100 Subject: [PATCH 2/3] (E002) prepare restructure [ModelicaSystemCmd] add missing docstring [OMCSession] spelling fixes [OMCSessionCmd] add warning about depreciated class [OMCSessionABC] remove duplicated code; see OMSessionABC [OMSessionRunnerABC] define class [OMCSessionZMQ] call super()__init__() [OMCPath] fix forward dependency on OMCSessionLocal [OMSessionException] rename from OMCSessionException [__init__] fix imports --- OMPython/ModelicaSystem.py | 8 +- OMPython/OMCSession.py | 216 ++++++++++++++++++++----------------- OMPython/__init__.py | 6 +- 3 files changed, 122 insertions(+), 108 deletions(-) diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index 01e5bfbd..3159bb31 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -25,7 +25,7 @@ ModelExecutionData, ModelExecutionException, - OMCSessionException, + OMSessionException, OMCSessionLocal, OMPathABC, @@ -1688,7 +1688,7 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: """ try: retval = self._session.sendExpression(expr=expr, parsed=parsed) - except OMCSessionException as ex: + except OMSessionException as ex: raise ModelicaSystemError(f"Error executing {repr(expr)}: {ex}") from ex logger.debug(f"Result of executing {repr(expr)}: {textwrap.shorten(repr(retval), width=100)}") @@ -2829,7 +2829,9 @@ def _prepare_structure_parameters( class ModelicaSystemCmd(ModelExecutionCmd): - # TODO: docstring + """ + Compatibility class; in the new version it is renamed as MOdelExecutionCmd. + """ def __init__( self, diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index 04b5d9cc..fcd39c2e 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -59,18 +59,31 @@ def wait(self, timeout): pass -class OMCSessionException(Exception): +class OMSessionException(Exception): """ Exception which is raised by any OMC* class. """ +class OMCSessionException(OMSessionException): + """ + Just a compatibility layer ... + """ + + class OMCSessionCmd: """ Implementation of Open Modelica Compiler API functions. Depreciated! """ def __init__(self, session: OMSessionABC, readonly: bool = False): + warnings.warn( + message="The class OMCSessionCMD is depreciated and will be removed in future versions; " + "please use OMCSession*.sendExpression(...) instead!", + category=DeprecationWarning, + stacklevel=2, + ) + if not isinstance(session, OMSessionABC): raise OMCSessionException("Invalid OMC process definition!") self._session = session @@ -84,7 +97,7 @@ def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: bool = Tr 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)}") + raise OMSessionException(f"Invalid definition of options for {repr(question)}: {repr(opt)}") p = (expression, parsed) @@ -95,8 +108,8 @@ def _ask(self, question: str, opt: Optional[list[str]] = None, parsed: bool = Tr try: res = self._session.sendExpression(expression, parsed=parsed) - except OMCSessionException as ex: - raise OMCSessionException(f"OMC _ask() failed: {expression} (parsed={parsed})") from ex + except OMSessionException as ex: + raise OMSessionException(f"OMC _ask() failed: {expression} (parsed={parsed})") from ex # save response self._omc_cache[p] = res @@ -411,7 +424,7 @@ def is_file(self) -> bool: """ retval = self.get_session().sendExpression(expr=f'regularFileExists("{self.as_posix()}")') if not isinstance(retval, bool): - raise OMCSessionException(f"Invalid return value for is_file(): {retval} - expect bool") + raise OMSessionException(f"Invalid return value for is_file(): {retval} - expect bool") return retval def is_dir(self) -> bool: @@ -420,14 +433,14 @@ def is_dir(self) -> bool: """ retval = self.get_session().sendExpression(expr=f'directoryExists("{self.as_posix()}")') if not isinstance(retval, bool): - raise OMCSessionException(f"Invalid return value for is_dir(): {retval} - expect bool") + raise OMSessionException(f"Invalid return value for is_dir(): {retval} - expect bool") return retval def is_absolute(self) -> bool: """ Check if the path is an absolute path. Special handling to differentiate Windows and Posix definitions. """ - if isinstance(self._session, OMCSessionLocal) and platform.system() == 'Windows': + if self._session.model_execution_windows and self._session.model_execution_local: return pathlib.PureWindowsPath(self.as_posix()).is_absolute() return pathlib.PurePosixPath(self.as_posix()).is_absolute() @@ -437,7 +450,7 @@ def read_text(self) -> str: """ retval = self.get_session().sendExpression(expr=f'readFile("{self.as_posix()}")') if not isinstance(retval, str): - raise OMCSessionException(f"Invalid return value for read_text(): {retval} - expect str") + raise OMSessionException(f"Invalid return value for read_text(): {retval} - expect str") return retval def write_text(self, data: str) -> int: @@ -464,7 +477,7 @@ def mkdir(self, parents: bool = True, exist_ok: bool = False) -> None: raise FileExistsError(f"Directory {self.as_posix()} already exists!") if not self._session.sendExpression(expr=f'mkdir("{self.as_posix()}")'): - raise OMCSessionException(f"Error on directory creation for {self.as_posix()}!") + raise OMSessionException(f"Error on directory creation for {self.as_posix()}!") def cwd(self) -> OMPathABC: """ @@ -486,7 +499,7 @@ def resolve(self, strict: bool = False) -> OMPathABC: Resolve the path to an absolute path. This is done based on available OMC functions. """ if strict and not (self.is_file() or self.is_dir()): - raise OMCSessionException(f"Path {self.as_posix()} does not exist!") + raise OMSessionException(f"Path {self.as_posix()} does not exist!") if self.is_file(): pathstr_resolved = self._omc_resolve(self.parent.as_posix()) @@ -495,10 +508,10 @@ def resolve(self, strict: bool = False) -> OMPathABC: pathstr_resolved = self._omc_resolve(self.as_posix()) omcpath_resolved = self._session.omcpath(pathstr_resolved) else: - raise OMCSessionException(f"Path {self.as_posix()} is neither a file nor a directory!") + raise OMSessionException(f"Path {self.as_posix()} is neither a file nor a directory!") if not omcpath_resolved.is_file() and not omcpath_resolved.is_dir(): - raise OMCSessionException(f"OMCPath resolve failed for {self.as_posix()} - path does not exist!") + raise OMSessionException(f"OMCPath resolve failed for {self.as_posix()} - path does not exist!") return omcpath_resolved @@ -514,12 +527,12 @@ def _omc_resolve(self, pathstr: str) -> str: try: retval = self.get_session().sendExpression(expr=expr, parsed=False) if not isinstance(retval, str): - raise OMCSessionException(f"Invalid return value for _omc_resolve(): {retval} - expect str") + raise OMSessionException(f"Invalid return value for _omc_resolve(): {retval} - expect str") result_parts = retval.split('\n') pathstr_resolved = result_parts[1] pathstr_resolved = pathstr_resolved[1:-1] # remove quotes - except OMCSessionException as ex: - raise OMCSessionException(f"OMCPath resolve failed for {pathstr}!") from ex + except OMSessionException as ex: + raise OMSessionException(f"OMCPath resolve failed for {pathstr}!") from ex return pathstr_resolved @@ -528,13 +541,13 @@ def size(self) -> int: Get the size of the file in bytes - this is an extra function and the best we can do using OMC. """ if not self.is_file(): - raise OMCSessionException(f"Path {self.as_posix()} is not a file!") + raise OMSessionException(f"Path {self.as_posix()} is not a file!") res = self._session.sendExpression(expr=f'stat("{self.as_posix()}")') if res[0]: return int(res[1]) - raise OMCSessionException(f"Error reading file size for path {self.as_posix()}!") + raise OMSessionException(f"Error reading file size for path {self.as_posix()}!") class OMPathRunnerABC(OMPathABC, metaclass=abc.ABCMeta): """ @@ -618,10 +631,10 @@ def resolve(self, strict: bool = False) -> OMPathABC: def size(self) -> int: """ - Get the size of the file in bytes - implementation baseon on pathlib.Path. + Get the size of the file in bytes - implementation based on pathlib.Path. """ if not self.is_file(): - raise OMCSessionException(f"Path {self.as_posix()} is not a file!") + raise OMSessionException(f"Path {self.as_posix()} is not a file!") path = self._path() return path.stat().st_size @@ -729,7 +742,7 @@ def mkdir(self, parents: bool = True, exist_ok: bool = False) -> None: try: subprocess.run(cmdl, check=True) except subprocess.CalledProcessError as exc: - raise OMCSessionException(f"Error on directory creation for {self.as_posix()}!") from exc + raise OMSessionException(f"Error on directory creation for {self.as_posix()}!") from exc def cwd(self) -> OMPathABC: """ @@ -776,10 +789,10 @@ def resolve(self, strict: bool = False) -> OMPathABC: def size(self) -> int: """ - Get the size of the file in bytes - implementation baseon on pathlib.Path. + Get the size of the file in bytes - implementation based on pathlib.Path. """ if not self.is_file(): - raise OMCSessionException(f"Path {self.as_posix()} is not a file!") + raise OMSessionException(f"Path {self.as_posix()} is not a file!") cmdl = self.get_session().get_cmd_prefix() cmdl += ['bash', '-c', f'stat -c %s "{self.as_posix()}"'] @@ -790,7 +803,7 @@ def size(self) -> int: try: return int(stdout) except ValueError as exc: - raise OSError(f"Invalid return value for filesize ({self.as_posix()}): {stdout}") from exc + raise OSError(f"Invalid return value for file size ({self.as_posix()}): {stdout}") from exc else: raise OSError(f"Cannot get size for file {self.as_posix()}") @@ -959,7 +972,7 @@ def set_timeout(self, timeout: Optional[float] = None) -> float: retval = self._timeout if timeout is not None: if timeout <= 0.0: - raise OMCSessionException(f"Invalid timeout value: {timeout}s!") + raise OMSessionException(f"Invalid timeout value: {timeout}s!") logger.info(f"Update timeout for {self.__class__.__name__}: {retval}s => {timeout}s") self._timeout = timeout return retval @@ -1088,7 +1101,7 @@ def __init__( try: self._omc_loghandle = open(file=self._omc_logfile, mode="w+", encoding="utf-8") except OSError as ex: - raise OMCSessionException(f"Cannot open log file {self._omc_logfile}.") from ex + raise OMSessionException(f"Cannot open log file {self._omc_logfile}.") from ex # variables to store compiled re expressions use in self.sendExpression() self._re_log_entries: Optional[re.Pattern[str]] = None @@ -1103,7 +1116,7 @@ def __post_init__(self) -> None: """ port = self.get_port() if not isinstance(port, str): - raise OMCSessionException(f"Invalid content for port: {port}") + raise OMSessionException(f"Invalid content for port: {port}") # Create the ZeroMQ socket and connect to OMC server context = zmq.Context.instance() @@ -1118,7 +1131,7 @@ def __del__(self): if isinstance(self._omc_zmq, zmq.Socket): try: self.sendExpression(expr="quit()") - except OMCSessionException as exc: + except OMSessionException as exc: logger.warning(f"Exception on sending 'quit()' to OMC: {exc}! Continue nevertheless ...") finally: self._omc_zmq = None @@ -1157,7 +1170,7 @@ def _timeout_loop( if timeout is None: timeout = self._timeout if timeout <= 0: - raise OMCSessionException(f"Invalid timeout: {timeout}") + raise OMSessionException(f"Invalid timeout: {timeout}") timer = 0.0 yield True @@ -1206,7 +1219,7 @@ def omcpath(self, *path) -> OMPathABC: if isinstance(self, OMCSessionLocal): # noinspection PyArgumentList return OMCPath(*path) - raise OMCSessionException("OMCPath is supported for Python < 3.12 only if OMCSessionLocal is used!") + raise OMSessionException("OMCPath is supported for Python < 3.12 only if OMCSessionLocal is used!") return OMCPath(*path, session=self) def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC: @@ -1225,26 +1238,6 @@ def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC return self._tempdir(tempdir_base=tempdir_base) - @staticmethod - def _tempdir(tempdir_base: OMPathABC) -> OMPathABC: - names = [str(uuid.uuid4()) for _ in range(100)] - - tempdir: Optional[OMPathABC] = None - for name in names: - # create a unique temporary directory name - tempdir = tempdir_base / name - - if tempdir.exists(): - continue - - tempdir.mkdir(parents=True, exist_ok=False) - break - - if tempdir is None or not tempdir.is_dir(): - raise OMCSessionException("Cannot create a temporary directory!") - - return tempdir - def execute(self, command: str): warnings.warn( message="This function is depreciated and will be removed in future versions; " @@ -1259,12 +1252,12 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: """ Send an expression to the OMC server and return the result. - The complete error handling of the OMC result is done within this method using '"getMessagesStringInternal()'. - Caller should only check for OMCSessionException. + The complete error handling of the OMC result is done within this method using 'getMessagesStringInternal()'. + Caller should only check for OMSessionException. """ if self._omc_zmq is None: - raise OMCSessionException("No OMC running. Please create a new instance of OMCSession!") + raise OMSessionException("No OMC running. Please create a new instance of OMCSession!") logger.debug("sendExpression(expr='%r', parsed=%r)", str(expr), parsed) @@ -1279,11 +1272,11 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: # in the deletion process, the content is cleared. Thus, any access to a class attribute must be checked try: log_content = self.get_log() - except OMCSessionException: + except OMSessionException: log_content = 'log not available' logger.error(f"OMC did not start. Log-file says:\n{log_content}") - raise OMCSessionException(f"No connection with OMC (timeout={self._timeout:.2f}s).") + raise OMSessionException(f"No connection with OMC (timeout={self._timeout:.2f}s).") if expr == "quit()": self._omc_zmq.close() @@ -1293,7 +1286,7 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: result = self._omc_zmq.recv_string() if result.startswith('Error occurred building AST'): - raise OMCSessionException(f"OMC error: {result}") + raise OMSessionException(f"OMC error: {result}") if expr == "getErrorString()": # no error handling if 'getErrorString()' is called @@ -1377,8 +1370,8 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: msg_long_list.append(msg_long) if has_error: msg_long_str = '\n'.join(f"{idx:02d}: {msg}" for idx, msg in enumerate(msg_long_list)) - raise OMCSessionException(f"OMC error occurred for 'sendExpression(expr={expr}, parsed={parsed}):\n" - f"{msg_long_str}") + raise OMSessionException(f"OMC error occurred for 'sendExpression(expr={expr}, parsed={parsed}):\n" + f"{msg_long_str}") if not parsed: return result @@ -1390,14 +1383,14 @@ def sendExpression(self, expr: str, parsed: bool = True) -> Any: try: return om_parser_basic(result) except (TypeError, UnboundLocalError) as ex2: - raise OMCSessionException("Cannot parse OMC result") from ex2 + raise OMSessionException("Cannot parse OMC result") from ex2 def get_port(self) -> Optional[str]: """ Get the port to connect to the OMC session. """ if not isinstance(self._omc_port, str): - raise OMCSessionException(f"Invalid port to connect to OMC process: {self._omc_port}") + raise OMSessionException(f"Invalid port to connect to OMC process: {self._omc_port}") return self._omc_port def get_log(self) -> str: @@ -1405,7 +1398,7 @@ def get_log(self) -> str: Get the log file content of the OMC session. """ if self._omc_loghandle is None: - raise OMCSessionException("Log file not available!") + raise OMSessionException("Log file not available!") self._omc_loghandle.seek(0) log = self._omc_loghandle.read() @@ -1476,7 +1469,7 @@ def _omc_home_get(omhome: Optional[str | os.PathLike] = None) -> pathlib.Path: 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") + raise OMSessionException("Cannot find OpenModelica executable, please install from openmodelica.org") def _omc_process_get(self) -> subprocess.Popen: my_env = os.environ.copy() @@ -1510,8 +1503,8 @@ def _omc_port_get(self) -> str: break else: logger.error(f"OMC server did not start. Log-file says:\n{self.get_log()}") - raise OMCSessionException(f"OMC Server did not start (timeout={self._timeout:.2f}s, " - f"logfile={repr(self._omc_logfile)}).") + raise OMSessionException(f"OMC Server did not start (timeout={self._timeout:.2f}s, " + f"logfile={repr(self._omc_logfile)}).") 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 '?'}") @@ -1541,9 +1534,11 @@ def __init__( if omc_process is None: omc_process = OMCSessionLocal(omhome=omhome, timeout=timeout) elif not isinstance(omc_process, OMCSessionABC): - raise OMCSessionException("Invalid definition of the OMC process!") + raise OMSessionException("Invalid definition of the OMC process!") self.omc_process = omc_process + super().__init__(timeout=timeout) + def __del__(self): if hasattr(self, 'omc_process'): del self.omc_process @@ -1576,7 +1571,7 @@ def sendExpression(self, command: str, parsed: bool = True) -> Any: Send an expression to the OMC server and return the result. The complete error handling of the OMC result is done within this method using '"getMessagesStringInternal()'. - Caller should only check for OMCSessionException. + Caller should only check for OMSessionException. """ return self.omc_process.sendExpression(expr=command, parsed=parsed) @@ -1625,7 +1620,7 @@ def __init__( # connect to the running omc instance using ZMQ self._omc_port = self._omc_port_get(docker_cid=self._docker_container_id) if port is not None and not self._omc_port.endswith(f":{port}"): - raise OMCSessionException(f"Port mismatch: {self._omc_port} is not using the defined port {port}!") + raise OMSessionException(f"Port mismatch: {self._omc_port} is not using the defined port {port}!") self._cmd_prefix = self.model_execution_prefix() @@ -1643,13 +1638,13 @@ def _docker_process_get(self, docker_cid: str) -> Optional[DockerPopen]: try: docker_process = DockerPopen(int(columns[1])) except psutil.NoSuchProcess as ex: - raise OMCSessionException(f"Could not find PID {docker_top} - " - "is this a docker instance spawned without --pid=host?") from ex + raise OMSessionException(f"Could not find PID {docker_top} - " + "is this a docker instance spawned without --pid=host?") from ex if docker_process is not None: break else: logger.error(f"Docker did not start. Log-file says:\n{self.get_log()}") - raise OMCSessionException(f"Docker based OMC Server did not start (timeout={self._timeout:.2f}s).") + raise OMSessionException(f"Docker based OMC Server did not start (timeout={self._timeout:.2f}s).") return docker_process @@ -1680,7 +1675,7 @@ def _omc_port_get( port = None if not isinstance(docker_cid, str): - raise OMCSessionException(f"Invalid docker container ID: {docker_cid}") + raise OMSessionException(f"Invalid docker container ID: {docker_cid}") # See if the omc server is running loop = self._timeout_loop(timestep=0.1) @@ -1699,8 +1694,8 @@ def _omc_port_get( break else: logger.error(f"Docker did not start. Log-file says:\n{self.get_log()}") - raise OMCSessionException(f"Docker based OMC Server did not start (timeout={self._timeout:.2f}s, " - f"logfile={repr(self._omc_logfile)}).") + raise OMSessionException(f"Docker based OMC Server did not start (timeout={self._timeout:.2f}s, " + f"logfile={repr(self._omc_logfile)}).") logger.info(f"Docker based OMC Server is up and running at port {port}") @@ -1714,7 +1709,7 @@ def get_server_address(self) -> Optional[str]: output = subprocess.check_output(["docker", "inspect", self._docker_container_id]).decode().strip() address = json.loads(output)[0]["NetworkSettings"]["IPAddress"] if not isinstance(address, str): - raise OMCSessionException(f"Invalid docker server address: {address}!") + raise OMSessionException(f"Invalid docker server address: {address}!") return address return None @@ -1724,7 +1719,7 @@ def get_docker_container_id(self) -> str: Get the Docker container ID of the Docker container with the OMC server. """ if not isinstance(self._docker_container_id, str): - raise OMCSessionException(f"Invalid docker container ID: {self._docker_container_id}!") + raise OMSessionException(f"Invalid docker container ID: {self._docker_container_id}!") return self._docker_container_id @@ -1801,8 +1796,8 @@ def _docker_omc_cmd( if sys.platform == "win32": extra_flags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"] if not self._omc_port: - raise OMCSessionException("Docker on Windows requires knowing which port to connect to - " - "please set the interactivePort argument") + raise OMSessionException("Docker on Windows requires knowing which port to connect to - " + "please set the interactivePort argument") port: Optional[int] = None if isinstance(omc_port, str): @@ -1812,8 +1807,8 @@ def _docker_omc_cmd( if sys.platform == "win32": if not isinstance(port, int): - raise OMCSessionException("OMC on Windows needs the interactive port - " - f"missing or invalid value: {repr(omc_port)}!") + raise OMSessionException("OMC on Windows needs the interactive port - " + f"missing or invalid value: {repr(omc_port)}!") docker_network_str = ["-p", f"127.0.0.1:{port}:{port}"] elif self._docker_network == "host" or self._docker_network is None: docker_network_str = ["--network=host"] @@ -1821,8 +1816,8 @@ def _docker_omc_cmd( docker_network_str = [] extra_flags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"] else: - raise OMCSessionException(f'dockerNetwork was set to {self._docker_network}, ' - 'but only \"host\" or \"separate\" is allowed') + raise OMSessionException(f'dockerNetwork was set to {self._docker_network}, ' + 'but only \"host\" or \"separate\" is allowed') if isinstance(port, int): extra_flags = extra_flags + [f"--interactivePort={port}"] @@ -1849,7 +1844,7 @@ def _docker_omc_start( ) -> Tuple[subprocess.Popen, DockerPopen, str]: if not isinstance(docker_image, str): - raise OMCSessionException("A docker image name must be provided!") + raise OMSessionException("A docker image name must be provided!") my_env = os.environ.copy() @@ -1870,7 +1865,7 @@ def _docker_omc_start( 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}") + raise OMSessionException(f"Invalid content for docker container ID file path: {docker_cid_file}") # the provided value for docker_cid is not used docker_cid = None @@ -1885,14 +1880,14 @@ def _docker_omc_start( break if docker_cid is None: - raise OMCSessionException(f"Docker did not start (timeout={self._timeout:.2f}s might be too short " - "especially if you did not docker pull the image before this command). " - f"Log-file says:\n{self.get_log()}") + raise OMSessionException(f"Docker did not start (timeout={self._timeout:.2f}s might be too short " + "especially if you did not docker pull the image before this command). " + f"Log-file says:\n{self.get_log()}") docker_process = self._docker_process_get(docker_cid=docker_cid) if docker_process is None: logger.error(f"Docker did not start. Log-file says:\n{self.get_log()}") - raise OMCSessionException(f"Docker top did not contain omc process {self._random_string}.") + raise OMSessionException(f"Docker top did not contain omc process {self._random_string}.") return omc_process, docker_process, docker_cid @@ -1942,10 +1937,10 @@ def _docker_omc_cmd( if sys.platform == "win32": extra_flags = ["-d=zmqDangerousAcceptConnectionsFromAnywhere"] if not isinstance(omc_port, int): - raise OMCSessionException("Docker on Windows requires knowing which port to connect to - " - "Please set the interactivePort argument. Furthermore, 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.") + raise OMSessionException("Docker on Windows requires knowing which port to connect to - " + "Please set the interactivePort argument. Furthermore, 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(omc_port, int): extra_flags = extra_flags + [f"--interactivePort={omc_port}"] @@ -1969,7 +1964,7 @@ def _docker_omc_start( ) -> Tuple[subprocess.Popen, DockerPopen, str]: if not isinstance(docker_cid, str): - raise OMCSessionException("A docker container ID must be provided!") + raise OMSessionException("A docker container ID must be provided!") my_env = os.environ.copy() @@ -1991,8 +1986,8 @@ def _docker_omc_start( 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"/ {docker_cid}. Log-file says:\n{self.get_log()}") + raise OMSessionException(f"Docker top did not contain omc process {self._random_string} " + f"/ {docker_cid}. Log-file says:\n{self.get_log()}") return omc_process, docker_process, docker_cid @@ -2076,8 +2071,8 @@ def _omc_port_get(self) -> str: break else: logger.error(f"WSL based OMC server did not start. Log-file says:\n{self.get_log()}") - raise OMCSessionException(f"WSL based OMC Server did not start (timeout={self._timeout:2f}s, " - f"logfile={repr(self._omc_logfile)}).") + raise OMSessionException(f"WSL based OMC Server did not start (timeout={self._timeout:2f}s, " + f"logfile={repr(self._omc_logfile)}).") logger.info(f"WSL based OMC Server is up and running at ZMQ port {port} " f"pid={self._omc_process.pid if isinstance(self._omc_process, subprocess.Popen) else '?'}") @@ -2085,16 +2080,16 @@ def _omc_port_get(self) -> str: return port -class OMSessionRunner(OMSessionABC): +class OMSessionRunnerABC(OMSessionABC, metaclass=abc.ABCMeta): """ Implementation based on OMSessionABC without any use of an OMC server. """ def __init__( self, + ompath_runner: Type[OMPathRunnerABC], timeout: Optional[float] = None, version: str = "1.27.0", - ompath_runner: Type[OMPathRunnerABC] = OMPathRunnerLocal, cmd_prefix: Optional[list[str]] = None, model_execution_local: bool = True, ) -> None: @@ -2102,15 +2097,34 @@ def __init__( self._version = version if not issubclass(ompath_runner, OMPathRunnerABC): - raise OMCSessionException(f"Invalid OMPathRunner class: {type(ompath_runner)}!") + raise OMSessionException(f"Invalid OMPathRunner class: {type(ompath_runner)}!") self._ompath_runner = ompath_runner self.model_execution_local = model_execution_local if cmd_prefix is not None: self._cmd_prefix = cmd_prefix - # TODO: some checking?! - # if ompath_runner == Type[OMPathRunnerBash]: + +class OMSessionRunner(OMSessionRunnerABC): + """ + Implementation based on OMSessionABC without any use of an OMC server. + """ + + def __init__( + self, + ompath_runner: Type[OMPathRunnerABC] = OMPathRunnerLocal, + timeout: float = 10.0, + version: str = "1.27.0", + cmd_prefix: Optional[list[str]] = None, + model_execution_local: bool = True, + ) -> None: + super().__init__( + ompath_runner=ompath_runner, + timeout=timeout, + version=version, + cmd_prefix=cmd_prefix, + model_execution_local=model_execution_local, + ) def __post_init__(self) -> None: """ @@ -2153,7 +2167,7 @@ def omcpath_tempdir(self, tempdir_base: Optional[OMPathABC] = None) -> OMPathABC return self._tempdir(tempdir_base=tempdir_base) def sendExpression(self, expr: str, parsed: bool = True) -> Any: - raise OMCSessionException(f"{self.__class__.__name__} does not uses an OMC server!") + raise OMSessionException(f"{self.__class__.__name__} does not uses an OMC server!") DummyPopen = DockerPopen diff --git a/OMPython/__init__.py b/OMPython/__init__.py index 22c88137..96f5fb7c 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -33,11 +33,10 @@ OMSessionABC, OMSessionRunner, - OMCSessionABC, - ModelExecutionData, ModelExecutionException, + OMCSessionABC, OMCSessionCmd, OMCSessionDocker, OMCSessionDockerContainer, @@ -81,10 +80,9 @@ 'OMSessionABC', 'OMSessionRunner', - 'OMCSessionABC', - 'doe_get_solutions', + 'OMCSessionABC', 'OMCSessionCmd', 'OMCSessionDocker', 'OMCSessionDockerContainer', From 3da0c6bcecdc300c11785a2f0d01a1129068b121 Mon Sep 17 00:00:00 2001 From: syntron Date: Fri, 13 Feb 2026 21:29:02 +0100 Subject: [PATCH 3/3] [ModelExecution*] move classes into model_execution.py --- OMPython/ModelicaSystem.py | 253 +------------------------- OMPython/OMCSession.py | 86 --------- OMPython/__init__.py | 12 +- OMPython/model_execution.py | 350 ++++++++++++++++++++++++++++++++++++ 4 files changed, 361 insertions(+), 340 deletions(-) create mode 100644 OMPython/model_execution.py diff --git a/OMPython/ModelicaSystem.py b/OMPython/ModelicaSystem.py index 3159bb31..b1942e9a 100644 --- a/OMPython/ModelicaSystem.py +++ b/OMPython/ModelicaSystem.py @@ -21,10 +21,12 @@ import numpy as np -from OMPython.OMCSession import ( +from OMPython.model_execution import ( + ModelExecutionCmd, ModelExecutionData, ModelExecutionException, - +) +from OMPython.OMCSession import ( OMSessionException, OMCSessionLocal, @@ -95,253 +97,6 @@ def __getitem__(self, index: int): return {0: self.A, 1: self.B, 2: self.C, 3: self.D}[index] -class ModelExecutionCmd: - """ - All information about a compiled model executable. This should include data about all structured parameters, i.e. - parameters which need a recompilation of the model. All non-structured parameters can be easily changed without - the need for recompilation. - """ - - def __init__( - self, - runpath: os.PathLike, - cmd_prefix: list[str], - cmd_local: bool = False, - cmd_windows: bool = False, - timeout: float = 10.0, - model_name: Optional[str] = None, - ) -> None: - if model_name is None: - raise ModelExecutionException("Missing model name!") - - self._cmd_local = cmd_local - self._cmd_windows = cmd_windows - self._cmd_prefix = cmd_prefix - self._runpath = pathlib.PurePosixPath(runpath) - self._model_name = model_name - self._timeout = timeout - - # dictionaries of command line arguments for the model executable - self._args: dict[str, str | None] = {} - # 'override' argument needs special handling, as it is a dict on its own saved as dict elements following the - # structure: 'key' => 'key=value' - self._arg_override: dict[str, str] = {} - - def arg_set( - self, - key: str, - val: Optional[str | dict[str, Any] | numbers.Number] = None, - ) -> None: - """ - Set one argument for the executable model. - - Args: - key: identifier / argument name to be used for the call of the model executable. - val: value for the given key; None for no value and for key == 'override' a dictionary can be used which - indicates variables to override - """ - - def override2str( - orkey: str, - orval: str | bool | numbers.Number, - ) -> str: - """ - Convert a value for 'override' to a string taking into account differences between Modelica and Python. - """ - # check oval for any string representations of numbers (or bool) and convert these to Python representations - if isinstance(orval, str): - try: - val_evaluated = ast.literal_eval(orval) - if isinstance(val_evaluated, (numbers.Number, bool)): - orval = val_evaluated - except (ValueError, SyntaxError): - pass - - if isinstance(orval, str): - val_str = orval.strip() - elif isinstance(orval, bool): - val_str = 'true' if orval else 'false' - elif isinstance(orval, numbers.Number): - val_str = str(orval) - else: - raise ModelExecutionException(f"Invalid value for override key {orkey}: {type(orval)}") - - return f"{orkey}={val_str}" - - if not isinstance(key, str): - raise ModelExecutionException(f"Invalid argument key: {repr(key)} (type: {type(key)})") - key = key.strip() - - if isinstance(val, dict): - if key != 'override': - raise ModelExecutionException("Dictionary input only possible for key 'override'!") - - for okey, oval in val.items(): - if not isinstance(okey, str): - raise ModelExecutionException("Invalid key for argument 'override': " - f"{repr(okey)} (type: {type(okey)})") - - if not isinstance(oval, (str, bool, numbers.Number, type(None))): - raise ModelExecutionException(f"Invalid input for 'override'.{repr(okey)}: " - f"{repr(oval)} (type: {type(oval)})") - - if okey in self._arg_override: - if oval is None: - logger.info(f"Remove model executable override argument: {repr(self._arg_override[okey])}") - del self._arg_override[okey] - continue - - logger.info(f"Update model executable override argument: {repr(okey)} = {repr(oval)} " - f"(was: {repr(self._arg_override[okey])})") - - if oval is not None: - self._arg_override[okey] = override2str(orkey=okey, orval=oval) - - argval = ','.join(sorted(self._arg_override.values())) - elif val is None: - argval = None - elif isinstance(val, str): - argval = val.strip() - elif isinstance(val, numbers.Number): - argval = str(val) - else: - raise ModelExecutionException(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})") - - if key in self._args: - logger.warning(f"Override model executable argument: {repr(key)} = {repr(argval)} " - f"(was: {repr(self._args[key])})") - self._args[key] = argval - - def arg_get(self, key: str) -> Optional[str | dict[str, str | bool | numbers.Number]]: - """ - Return the value for the given key - """ - if key in self._args: - return self._args[key] - - return None - - def args_set( - self, - args: dict[str, Optional[str | dict[str, Any] | numbers.Number]], - ) -> None: - """ - Define arguments for the model executable. - """ - for arg in args: - self.arg_set(key=arg, val=args[arg]) - - def get_cmd_args(self) -> list[str]: - """ - Get a list with the command arguments for the model executable. - """ - - cmdl = [] - for key in sorted(self._args): - if self._args[key] is None: - cmdl.append(f"-{key}") - else: - cmdl.append(f"-{key}={self._args[key]}") - - return cmdl - - def definition(self) -> ModelExecutionData: - """ - Define all needed data to run the model executable. The data is stored in an OMCSessionRunData object. - """ - # ensure that a result filename is provided - result_file = self.arg_get('r') - if not isinstance(result_file, str): - result_file = (self._runpath / f"{self._model_name}.mat").as_posix() - - # as this is the local implementation, pathlib.Path can be used - cmd_path = self._runpath - - cmd_library_path = None - if self._cmd_local and self._cmd_windows: - cmd_library_path = "" - - # set the process environment from the generated .bat file in windows which should have all the dependencies - # for this pathlib.PurePosixPath() must be converted to a pathlib.Path() object, i.e. WindowsPath - path_bat = pathlib.Path(cmd_path) / f"{self._model_name}.bat" - if not path_bat.is_file(): - raise ModelExecutionException("Batch file (*.bat) does not exist " + str(path_bat)) - - content = path_bat.read_text(encoding='utf-8') - for line in content.splitlines(): - match = re.match(pattern=r"^SET PATH=([^%]*)", string=line, flags=re.IGNORECASE) - if match: - cmd_library_path = match.group(1).strip(';') # Remove any trailing semicolons - my_env = os.environ.copy() - my_env["PATH"] = cmd_library_path + os.pathsep + my_env["PATH"] - - cmd_model_executable = cmd_path / f"{self._model_name}.exe" - else: - # for Linux the paths to the needed libraries should be included in the executable (using rpath) - cmd_model_executable = cmd_path / self._model_name - - # define local(!) working directory - cmd_cwd_local = None - if self._cmd_local: - cmd_cwd_local = cmd_path.as_posix() - - omc_run_data = ModelExecutionData( - cmd_path=cmd_path.as_posix(), - cmd_model_name=self._model_name, - cmd_args=self.get_cmd_args(), - cmd_result_file=result_file, - cmd_prefix=self._cmd_prefix, - cmd_library_path=cmd_library_path, - cmd_model_executable=cmd_model_executable.as_posix(), - cmd_cwd_local=cmd_cwd_local, - cmd_timeout=self._timeout, - ) - - return omc_run_data - - @staticmethod - def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]: - """ - Parse a simflag definition; this is deprecated! - - The return data can be used as input for self.args_set(). - """ - warnings.warn( - message="The argument 'simflags' is depreciated and will be removed in future versions; " - "please use 'simargs' instead", - category=DeprecationWarning, - stacklevel=2, - ) - - simargs: dict[str, Optional[str | dict[str, Any] | numbers.Number]] = {} - - args = [s for s in simflags.split(' ') if s] - for arg in args: - if arg[0] != '-': - raise ModelExecutionException(f"Invalid simulation flag: {arg}") - arg = arg[1:] - parts = arg.split('=') - if len(parts) == 1: - simargs[parts[0]] = None - elif parts[0] == 'override': - override = '='.join(parts[1:]) - - override_dict = {} - for item in override.split(','): - kv = item.split('=') - if not 0 < len(kv) < 3: - raise ModelExecutionException(f"Invalid value for '-override': {override}") - if kv[0]: - try: - override_dict[kv[0]] = kv[1] - except (KeyError, IndexError) as ex: - raise ModelExecutionException(f"Invalid value for '-override': {override}") from ex - - simargs[parts[0]] = override_dict - - return simargs - - class ModelicaSystemABC(metaclass=abc.ABCMeta): """ Base class to simulate a Modelica models. diff --git a/OMPython/OMCSession.py b/OMPython/OMCSession.py index fcd39c2e..c9a6ca1b 100644 --- a/OMPython/OMCSession.py +++ b/OMPython/OMCSession.py @@ -6,7 +6,6 @@ from __future__ import annotations import abc -import dataclasses import io import json import logging @@ -812,91 +811,6 @@ def size(self) -> int: OMPathRunnerBash = _OMPathRunnerBash -class ModelExecutionException(Exception): - """ - Exception which is raised by ModelException* classes. - """ - - -@dataclasses.dataclass -class ModelExecutionData: - """ - Data class to store the command line data for running a model executable in the OMC environment. - - All data should be defined for the environment, where OMC is running (local, docker or WSL) - - To use this as a definition of an OMC simulation run, it has to be processed within - OMCProcess*.self_update(). This defines the attribute cmd_model_executable. - """ - # cmd_path is the expected working directory - cmd_path: str - cmd_model_name: str - # command prefix data (as list of strings); needed for docker or WSL - cmd_prefix: list[str] - # cmd_model_executable is build out of cmd_path and cmd_model_name; this is mainly needed on Windows (add *.exe) - cmd_model_executable: str - # command line arguments for the model executable - cmd_args: list[str] - # result file with the simulation output - cmd_result_file: str - # command timeout - cmd_timeout: float - - # additional library search path; this is mainly needed if OMCProcessLocal is run on Windows - cmd_library_path: Optional[str] = None - # working directory to be used on the *local* system - cmd_cwd_local: Optional[str] = None - - def get_cmd(self) -> list[str]: - """ - Get the command line to run the model executable in the environment defined by the OMCProcess definition. - """ - - cmdl = self.cmd_prefix - cmdl += [self.cmd_model_executable] - cmdl += self.cmd_args - - return cmdl - - def run(self) -> int: - """ - Run the model execution defined in this class. - """ - - my_env = os.environ.copy() - if isinstance(self.cmd_library_path, str): - my_env["PATH"] = self.cmd_library_path + os.pathsep + my_env["PATH"] - - cmdl = self.get_cmd() - - logger.debug("Run OM command %s in %s (timeout=%2fs)", repr(cmdl), self.cmd_path, self.cmd_timeout) - try: - cmdres = subprocess.run( - cmdl, - capture_output=True, - text=True, - env=my_env, - cwd=self.cmd_cwd_local, - timeout=self.cmd_timeout, - check=True, - ) - stdout = cmdres.stdout.strip() - stderr = cmdres.stderr.strip() - returncode = cmdres.returncode - - logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout) - - if stderr: - raise ModelExecutionException(f"Error running model executable {repr(cmdl)}: {stderr}") - except subprocess.TimeoutExpired as ex: - raise ModelExecutionException("OMPython timeout running model executable " - f"(timeout={self.cmd_timeout:.2f}s){repr(cmdl)}: {ex}") from ex - except subprocess.CalledProcessError as ex: - raise ModelExecutionException(f"Error running model executable {repr(cmdl)}: {ex}") from ex - - return returncode - - class PostInitCaller(type): """ Metaclass definition to define a new function __post_init__() which is called after all __init__() functions where diff --git a/OMPython/__init__.py b/OMPython/__init__.py index 96f5fb7c..3401585d 100644 --- a/OMPython/__init__.py +++ b/OMPython/__init__.py @@ -11,11 +11,16 @@ """ +from OMPython.model_execution import ( + ModelExecutionCmd, + ModelExecutionData, + ModelExecutionException, +) + from OMPython.ModelicaSystem import ( LinearizationResult, ModelicaSystem, ModelicaSystemOMC, - ModelExecutionCmd, ModelicaSystemDoE, ModelicaDoEOMC, ModelicaSystemError, @@ -33,9 +38,6 @@ OMSessionABC, OMSessionRunner, - ModelExecutionData, - ModelExecutionException, - OMCSessionABC, OMCSessionCmd, OMCSessionDocker, @@ -60,13 +62,13 @@ __all__ = [ 'LinearizationResult', + 'ModelExecutionCmd', 'ModelExecutionData', 'ModelExecutionException', 'ModelicaSystem', 'ModelicaSystemOMC', 'ModelicaSystemCmd', - 'ModelExecutionCmd', 'ModelicaSystemDoE', 'ModelicaDoEOMC', 'ModelicaSystemError', diff --git a/OMPython/model_execution.py b/OMPython/model_execution.py new file mode 100644 index 00000000..9f8b26a2 --- /dev/null +++ b/OMPython/model_execution.py @@ -0,0 +1,350 @@ +# -*- coding: utf-8 -*- +""" +Definition of needed tools to execute a compiled (binary) OpenModelica model. +""" + +import ast +import dataclasses +import logging +import numbers +import os +import pathlib +import re +import subprocess +from typing import Any, Optional +import warnings + +# define logger using the current module name as ID +logger = logging.getLogger(__name__) + + +class ModelExecutionException(Exception): + """ + Exception which is raised by ModelException* classes. + """ + + +@dataclasses.dataclass +class ModelExecutionData: + """ + Data class to store the command line data for running a model executable in the OMC environment. + + All data should be defined for the environment, where OMC is running (local, docker or WSL) + + To use this as a definition of an OMC simulation run, it has to be processed within + OMCProcess*.self_update(). This defines the attribute cmd_model_executable. + """ + # cmd_path is the expected working directory + cmd_path: str + cmd_model_name: str + # command prefix data (as list of strings); needed for docker or WSL + cmd_prefix: list[str] + # cmd_model_executable is build out of cmd_path and cmd_model_name; this is mainly needed on Windows (add *.exe) + cmd_model_executable: str + # command line arguments for the model executable + cmd_args: list[str] + # result file with the simulation output + cmd_result_file: str + # command timeout + cmd_timeout: float + + # additional library search path; this is mainly needed if OMCProcessLocal is run on Windows + cmd_library_path: Optional[str] = None + # working directory to be used on the *local* system + cmd_cwd_local: Optional[str] = None + + def get_cmd(self) -> list[str]: + """ + Get the command line to run the model executable in the environment defined by the OMCProcess definition. + """ + + cmdl = self.cmd_prefix + cmdl += [self.cmd_model_executable] + cmdl += self.cmd_args + + return cmdl + + def run(self) -> int: + """ + Run the model execution defined in this class. + """ + + my_env = os.environ.copy() + if isinstance(self.cmd_library_path, str): + my_env["PATH"] = self.cmd_library_path + os.pathsep + my_env["PATH"] + + cmdl = self.get_cmd() + + logger.debug("Run OM command %s in %s (timeout=%2fs)", repr(cmdl), self.cmd_path, self.cmd_timeout) + try: + cmdres = subprocess.run( + cmdl, + capture_output=True, + text=True, + env=my_env, + cwd=self.cmd_cwd_local, + timeout=self.cmd_timeout, + check=True, + ) + stdout = cmdres.stdout.strip() + stderr = cmdres.stderr.strip() + returncode = cmdres.returncode + + logger.debug("OM output for command %s:\n%s", repr(cmdl), stdout) + + if stderr: + raise ModelExecutionException(f"Error running model executable {repr(cmdl)}: {stderr}") + except subprocess.TimeoutExpired as ex: + raise ModelExecutionException("OMPython timeout running model executable " + f"(timeout={self.cmd_timeout:.2f}s){repr(cmdl)}: {ex}") from ex + except subprocess.CalledProcessError as ex: + raise ModelExecutionException(f"Error running model executable {repr(cmdl)}: {ex}") from ex + + return returncode + + +class ModelExecutionCmd: + """ + All information about a compiled model executable. This should include data about all structured parameters, i.e. + parameters which need a recompilation of the model. All non-structured parameters can be easily changed without + the need for recompilation. + """ + + def __init__( + self, + runpath: os.PathLike, + cmd_prefix: list[str], + cmd_local: bool = False, + cmd_windows: bool = False, + timeout: float = 10.0, + model_name: Optional[str] = None, + ) -> None: + if model_name is None: + raise ModelExecutionException("Missing model name!") + + self._cmd_local = cmd_local + self._cmd_windows = cmd_windows + self._cmd_prefix = cmd_prefix + self._runpath = pathlib.PurePosixPath(runpath) + self._model_name = model_name + self._timeout = timeout + + # dictionaries of command line arguments for the model executable + self._args: dict[str, str | None] = {} + # 'override' argument needs special handling, as it is a dict on its own saved as dict elements following the + # structure: 'key' => 'key=value' + self._arg_override: dict[str, str] = {} + + def arg_set( + self, + key: str, + val: Optional[str | dict[str, Any] | numbers.Number] = None, + ) -> None: + """ + Set one argument for the executable model. + + Args: + key: identifier / argument name to be used for the call of the model executable. + val: value for the given key; None for no value and for key == 'override' a dictionary can be used which + indicates variables to override + """ + + def override2str( + orkey: str, + orval: str | bool | numbers.Number, + ) -> str: + """ + Convert a value for 'override' to a string taking into account differences between Modelica and Python. + """ + # check oval for any string representations of numbers (or bool) and convert these to Python representations + if isinstance(orval, str): + try: + val_evaluated = ast.literal_eval(orval) + if isinstance(val_evaluated, (numbers.Number, bool)): + orval = val_evaluated + except (ValueError, SyntaxError): + pass + + if isinstance(orval, str): + val_str = orval.strip() + elif isinstance(orval, bool): + val_str = 'true' if orval else 'false' + elif isinstance(orval, numbers.Number): + val_str = str(orval) + else: + raise ModelExecutionException(f"Invalid value for override key {orkey}: {type(orval)}") + + return f"{orkey}={val_str}" + + if not isinstance(key, str): + raise ModelExecutionException(f"Invalid argument key: {repr(key)} (type: {type(key)})") + key = key.strip() + + if isinstance(val, dict): + if key != 'override': + raise ModelExecutionException("Dictionary input only possible for key 'override'!") + + for okey, oval in val.items(): + if not isinstance(okey, str): + raise ModelExecutionException("Invalid key for argument 'override': " + f"{repr(okey)} (type: {type(okey)})") + + if not isinstance(oval, (str, bool, numbers.Number, type(None))): + raise ModelExecutionException(f"Invalid input for 'override'.{repr(okey)}: " + f"{repr(oval)} (type: {type(oval)})") + + if okey in self._arg_override: + if oval is None: + logger.info(f"Remove model executable override argument: {repr(self._arg_override[okey])}") + del self._arg_override[okey] + continue + + logger.info(f"Update model executable override argument: {repr(okey)} = {repr(oval)} " + f"(was: {repr(self._arg_override[okey])})") + + if oval is not None: + self._arg_override[okey] = override2str(orkey=okey, orval=oval) + + argval = ','.join(sorted(self._arg_override.values())) + elif val is None: + argval = None + elif isinstance(val, str): + argval = val.strip() + elif isinstance(val, numbers.Number): + argval = str(val) + else: + raise ModelExecutionException(f"Invalid argument value for {repr(key)}: {repr(val)} (type: {type(val)})") + + if key in self._args: + logger.warning(f"Override model executable argument: {repr(key)} = {repr(argval)} " + f"(was: {repr(self._args[key])})") + self._args[key] = argval + + def arg_get(self, key: str) -> Optional[str | dict[str, str | bool | numbers.Number]]: + """ + Return the value for the given key + """ + if key in self._args: + return self._args[key] + + return None + + def args_set( + self, + args: dict[str, Optional[str | dict[str, Any] | numbers.Number]], + ) -> None: + """ + Define arguments for the model executable. + """ + for arg in args: + self.arg_set(key=arg, val=args[arg]) + + def get_cmd_args(self) -> list[str]: + """ + Get a list with the command arguments for the model executable. + """ + + cmdl = [] + for key in sorted(self._args): + if self._args[key] is None: + cmdl.append(f"-{key}") + else: + cmdl.append(f"-{key}={self._args[key]}") + + return cmdl + + def definition(self) -> ModelExecutionData: + """ + Define all needed data to run the model executable. The data is stored in an OMCSessionRunData object. + """ + # ensure that a result filename is provided + result_file = self.arg_get('r') + if not isinstance(result_file, str): + result_file = (self._runpath / f"{self._model_name}.mat").as_posix() + + # as this is the local implementation, pathlib.Path can be used + cmd_path = self._runpath + + cmd_library_path = None + if self._cmd_local and self._cmd_windows: + cmd_library_path = "" + + # set the process environment from the generated .bat file in windows which should have all the dependencies + # for this pathlib.PurePosixPath() must be converted to a pathlib.Path() object, i.e. WindowsPath + path_bat = pathlib.Path(cmd_path) / f"{self._model_name}.bat" + if not path_bat.is_file(): + raise ModelExecutionException("Batch file (*.bat) does not exist " + str(path_bat)) + + content = path_bat.read_text(encoding='utf-8') + for line in content.splitlines(): + match = re.match(pattern=r"^SET PATH=([^%]*)", string=line, flags=re.IGNORECASE) + if match: + cmd_library_path = match.group(1).strip(';') # Remove any trailing semicolons + my_env = os.environ.copy() + my_env["PATH"] = cmd_library_path + os.pathsep + my_env["PATH"] + + cmd_model_executable = cmd_path / f"{self._model_name}.exe" + else: + # for Linux the paths to the needed libraries should be included in the executable (using rpath) + cmd_model_executable = cmd_path / self._model_name + + # define local(!) working directory + cmd_cwd_local = None + if self._cmd_local: + cmd_cwd_local = cmd_path.as_posix() + + omc_run_data = ModelExecutionData( + cmd_path=cmd_path.as_posix(), + cmd_model_name=self._model_name, + cmd_args=self.get_cmd_args(), + cmd_result_file=result_file, + cmd_prefix=self._cmd_prefix, + cmd_library_path=cmd_library_path, + cmd_model_executable=cmd_model_executable.as_posix(), + cmd_cwd_local=cmd_cwd_local, + cmd_timeout=self._timeout, + ) + + return omc_run_data + + @staticmethod + def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | numbers.Number]]: + """ + Parse a simflag definition; this is deprecated! + + The return data can be used as input for self.args_set(). + """ + warnings.warn( + message="The argument 'simflags' is depreciated and will be removed in future versions; " + "please use 'simargs' instead", + category=DeprecationWarning, + stacklevel=2, + ) + + simargs: dict[str, Optional[str | dict[str, Any] | numbers.Number]] = {} + + args = [s for s in simflags.split(' ') if s] + for arg in args: + if arg[0] != '-': + raise ModelExecutionException(f"Invalid simulation flag: {arg}") + arg = arg[1:] + parts = arg.split('=') + if len(parts) == 1: + simargs[parts[0]] = None + elif parts[0] == 'override': + override = '='.join(parts[1:]) + + override_dict = {} + for item in override.split(','): + kv = item.split('=') + if not 0 < len(kv) < 3: + raise ModelExecutionException(f"Invalid value for '-override': {override}") + if kv[0]: + try: + override_dict[kv[0]] = kv[1] + except (KeyError, IndexError) as ex: + raise ModelExecutionException(f"Invalid value for '-override': {override}") from ex + + simargs[parts[0]] = override_dict + + return simargs