Merge branch 'master' into improve_exception

Author: adeas31

OMPython/ModelicaSystem.py

@@ -38,6 +38,7 @@
 import logging
 import numbers
 import os
+import pathlib
 import queue
 import textwrap
 import threading
@@ -322,6 +323,10 @@ def parse_simflags(simflags: str) -> dict[str, Optional[str | dict[str, Any] | n
 
 
 class ModelicaSystem:
+    """
+    Class to simulate a Modelica model using OpenModelica via OMCSessionZMQ.
+    """
+
     def __init__(
             self,
             commandLineOptions: Optional[list[str]] = None,
@@ -446,18 +451,30 @@ def model(
         # set variables
         self._model_name = name  # Model class name
         self._libraries = libraries  # may be needed if model is derived from other model
-        if file is not None:
-            file_name = self._session.omcpath(file).resolve()
-        else:
-            file_name = None
-        self._file_name = file_name  # Model file/package name
         self._variable_filter = variable_filter
 
-        if self._file_name is not None and not self._file_name.is_file():  # if file does not exist
-            raise IOError(f"{self._file_name} does not exist!")
-
         if self._libraries:
             self._loadLibrary(libraries=self._libraries)
+
+        self._file_name = None
+        if file is not None:
+            file_path = pathlib.Path(file)
+            # special handling for OMCProcessLocal - consider a relative path
+            if isinstance(self._session.omc_process, OMCProcessLocal) and not file_path.is_absolute():
+                file_path = pathlib.Path.cwd() / file_path
+            if not file_path.is_file():
+                raise IOError(f"Model file {file_path} does not exist!")
+
+            self._file_name = self.getWorkDirectory() / file_path.name
+            if (isinstance(self._session.omc_process, OMCProcessLocal)
+                    and file_path.as_posix() == self._file_name.as_posix()):
+                pass
+            elif self._file_name.is_file():
+                raise IOError(f"Simulation model file {self._file_name} exist - not overwriting!")
+            else:
+                content = file_path.read_text(encoding='utf-8')
+                self._file_name.write_text(content)
+
         if self._file_name is not None:
             self._loadFile(fileName=self._file_name)
 
@@ -1685,7 +1702,7 @@ def convertFmu2Mo(
             raise ModelicaSystemError(f"Missing FMU file: {fmu_path.as_posix()}")
 
         filename = self._requestApi(apiName='importFMU', entity=fmu_path.as_posix())
-        filepath = self._work_dir / filename
+        filepath = self.getWorkDirectory() / filename
 
         # report proper error message
         if not filepath.is_file():
@@ -2037,7 +2054,7 @@ def prepare(self) -> int:
 
                 pk_value = pc_structure[idx_structure]
                 if isinstance(pk_value, str):
-                    pk_value_str = pk_value.replace('"', '\\"')
+                    pk_value_str = self.session().escape_str(pk_value)
                     expression = f"setParameterValue({self._model_name}, {pk_structure}, \"{pk_value_str}\")"
                 elif isinstance(pk_value, bool):
                     pk_value_bool_str = "true" if pk_value else "false"

OMPython/OMCSession.py

@@ -336,7 +336,7 @@ def write_text(self, data: str, encoding=None, errors=None, newline=None):
         if not isinstance(data, str):
             raise TypeError(f"data must be str, not {data.__class__.__name__}")
 
-        data_omc = data.replace('"', '\\"')
+        data_omc = self._session.escape_str(data)
         self._session.sendExpression(f'writeFile("{self.as_posix()}", "{data_omc}", false);')
 
         return len(data)
@@ -576,6 +576,13 @@ def __del__(self):
 
         self.omc_zmq = None
 
+    @staticmethod
+    def escape_str(value: str) -> str:
+        """
+        Escape a string such that it can be used as string within OMC expressions, i.e. escape all double quotes.
+        """
+        return value.replace("\\", "\\\\").replace('"', '\\"')
+
     def omcpath(self, *path) -> OMCPath:
         """
         Create an OMCPath object based on the given path segments and the current OMC session.

tests/test_ModelicaSystem.py

@@ -51,10 +51,11 @@ def worker():
 
 def test_setParameters():
     omc = OMPython.OMCSessionZMQ()
-    model_path = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels/"
+    model_path_str = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels"
+    model_path = omc.omcpath(model_path_str)
     mod = OMPython.ModelicaSystem()
     mod.model(
-        file=model_path + "BouncingBall.mo",
+        file=model_path / "BouncingBall.mo",
         name="BouncingBall",
     )
 
@@ -85,10 +86,11 @@ def test_setParameters():
 
 def test_setSimulationOptions():
     omc = OMPython.OMCSessionZMQ()
-    model_path = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels/"
+    model_path_str = omc.sendExpression("getInstallationDirectoryPath()") + "/share/doc/omc/testmodels"
+    model_path = omc.omcpath(model_path_str)
     mod = OMPython.ModelicaSystem()
     mod.model(
-        file=model_path + "BouncingBall.mo",
+        file=model_path / "BouncingBall.mo",
         name="BouncingBall",
     )
 
@@ -112,7 +114,6 @@ def test_setSimulationOptions():
     assert d["tolerance"] == "1.2e-08"
 
 
-@pytest.mark.skip("will fail / fix available")
 def test_relative_path(model_firstorder):
     cwd = pathlib.Path.cwd()
     (fd, name) = tempfile.mkstemp(prefix='tmpOMPython.tests', dir=cwd, text=True)
@@ -152,30 +153,25 @@ def test_customBuildDirectory(tmp_path, model_firstorder):
 
 @skip_on_windows
 @skip_python_older_312
-def test_getSolutions_docker(model_firstorder_content):
+def test_getSolutions_docker(model_firstorder):
     omcp = OMPython.OMCProcessDocker(docker="openmodelica/openmodelica:v1.25.0-minimal")
     omc = OMPython.OMCSessionZMQ(omc_process=omcp)
 
-    modelpath = omc.omcpath_tempdir() / 'M.mo'
-    modelpath.write_text(model_firstorder_content)
-
-    file_path = pathlib.Path(modelpath)
     mod = OMPython.ModelicaSystem(
         omc_process=omc.omc_process,
     )
     mod.model(
         name="M",
-        file=file_path,
+        file=model_firstorder.as_posix(),
     )
 
     _run_getSolutions(mod)
 
 
 def test_getSolutions(model_firstorder):
-    filePath = model_firstorder.as_posix()
     mod = OMPython.ModelicaSystem()
     mod.model(
-        file=filePath,
+        file=model_firstorder.as_posix(),
         name="M",
     )
 

tests/test_ModelicaSystemDoE.py

@@ -71,15 +71,13 @@ def test_ModelicaSystemDoE_docker(tmp_path, model_doe, param_doe):
     omc = OMPython.OMCSessionZMQ(omc_process=omcp)
     assert omc.sendExpression("getVersion()") == "OpenModelica 1.25.0"
 
-    modelpath = omc.omcpath_tempdir() / 'M.mo'
-    modelpath.write_text(model_doe.read_text())
-
+    modelpath = omc.omcpath_tempdir()
     doe_mod = OMPython.ModelicaSystemDoE(
-        fileName=modelpath.as_posix(),
+        fileName=model_doe.as_posix(),
         modelName="M",
         parameters=param_doe,
         omc_process=omcp,
-        resultpath=modelpath.parent,
+        resultpath=modelpath,
         simargs={"override": {'stopTime': 1.0}},
     )
 

tests/test_OMCPath.py

@@ -76,3 +76,17 @@ def _run_OMCPath_checks(om: OMPython.OMCSessionZMQ):
     assert p3.parent.is_dir()
     p3.unlink()
     assert p3.is_file() is False
+
+
+def test_OMCPath_write_file(tmpdir):
+    om = OMPython.OMCSessionZMQ()
+
+    data = "abc # \\t # \" # \\n # xyz"
+
+    p1 = om.omcpath_tempdir()
+    p2 = p1 / 'test.txt'
+    p2.write_text(data=data)
+
+    assert data == p2.read_text()
+
+    del om