add Flood engine for 2d and couple models (#77)

* add floodEngine for 2d and couple models

US100380: Run 2D models from MIKE+Py

* add test data and test codes for flood engine

Task100891: Write notebook example to run 2d simulation and coupling simulation.

* change argument names

change camel case to snake case

* update flood engine test marker

mikeplus/datatableaccess.py

@@ -12,6 +12,7 @@
 from DHI.Amelia.Infrastructure.Interface.UtilityHelper import GeoAPIHelper
 from DHI.Amelia.DataModule.Interface.Services import IMuGeomTable
 from DHI.Amelia.DataModule.Services.DataTables import AmlUndoRedoManager
+from DHI.Amelia.DataModule.Services.ImportExportPfsFile import ImportExportPfsFile
 
 from .dotnet import as_dotnet_list
 from .dotnet import from_dotnet_datetime
@@ -77,6 +78,7 @@ def open_database(self):
         datatables.SetEumAppUnitSystem(data_source.UnitSystemOption)
         datatables.OnResetContainer(None, None)
         datatables.UndoRedoManager = AmlUndoRedoManager()
+        datatables.ImportExportPfsFile = ImportExportPfsFile()
         self._datatables = datatables
 
     def close_database(self):

mikeplus/engines/__init__.py

@@ -13,9 +13,11 @@
 from .engine1d import Engine1D  # noqa: E402
 from .epanet import EPANET  # noqa: E402
 from .swmm import SWMM  # noqa: E402
+from .flood_engine import FloodEngine  # noqa: E402
 
 __all__ = [
     "Engine1D",
     "EPANET",
     "SWMM",
+    "FloodEngine",
 ]

mikeplus/engines/flood_engine.py

@@ -0,0 +1,129 @@
+import os.path
+import time
+from pathlib import Path
+from DHI.Amelia.Tools.EngineTool import EngineTool
+from DHI.Amelia.DataModule.Interface.Services import IMsmProjectTable
+from DHI.Amelia.DomainServices.Interface.SharedEntity import DhiEngineSimpleLauncher
+from DHI.Amelia.GlobalUtility.DataType import MUSimulationOption
+from System.Collections.Generic import List
+
+
+class FloodEngine:
+    """The FloodEngine class can run 1D/2D/FLOOD simulation, print log files, and get the result files path"""
+
+    def __init__(self, data_tables):
+        self._data_tables = data_tables
+        self._result_files = None
+
+    def run(self, sim_muid=None, verbose=False):
+        """Run 1D/2D/Flood simulation
+
+        Parameters
+        ----------
+        simMuid : string, optional
+            simulation muid, it will use the current active simulation muid if simMuid is None, by default None.
+        verbose : bool, optional
+            print log file or not, by default False.
+
+        Examples
+        --------
+        >>>data_access = DataTableAccess(muppOrSqlite)
+        >>>data_access.open_database()
+        >>>engine = FloodEngine(data_access.datatables)
+        >>>engine.run()
+        >>>data_access.close_database()
+        """
+        if sim_muid is None:
+            sim_muid = self._get_active_muid()
+            if sim_muid is None:
+                raise ValueError("Simulation id can't be none.")
+        if verbose:
+            print("Simulation id is " + sim_muid)
+
+        engine_tool = EngineTool()
+        engine_tool.DataTables = self._data_tables
+        msg = List[str]()
+        launcher = DhiEngineSimpleLauncher()
+        success, launcher, msg = engine_tool.RunEngine_CS(launcher, msg, sim_muid)
+        if success and launcher is not None:
+            launcher.Start()
+        elif msg is not None and verbose:
+            joined_msg = "\n".Join(msg)
+            print(joined_msg)
+
+        if self._result_files is None:
+            self._result_files = self._get_result_files(sim_muid)
+
+        engine_start = False
+        while not launcher.IsEngineRunning:
+            time.sleep(1)
+        engine_start = True
+
+        while engine_start and not launcher.IsEngineExit:
+            time.sleep(1)
+
+        if verbose:
+            for log_file in self._get_log_files(sim_muid, launcher.SimulationOption):
+                self._print_log(log_file)
+
+    def _get_active_muid(self):
+        muid = self._data_tables["msm_Project"].GetMuidsWhere("ActiveProject=1")
+        if muid is None and muid.Count == 0:
+            return None
+        return muid[0]
+
+    def _get_result_files(self, sim_muid):
+        project = self._data_tables["msm_Project"]
+        prj = IMsmProjectTable(project)
+        res_files_dictionary = prj.GetResultFilePath(sim_muid)
+        res_files = []
+        for item in res_files_dictionary:
+            res_files.append(os.path.abspath(item.Value))
+        return res_files
+
+    def _get_log_files(self, sim_muid, sim_option):
+        db_mupp_file = Path(self._data_tables.DataSource.BaseFullPath)
+        dir = db_mupp_file.parent
+        file_name = db_mupp_file.stem
+        prefix = Path(dir) / f"{file_name}_{sim_muid}"
+        log_files = []
+        if sim_option == MUSimulationOption.CS_MIKE_1D:
+            log_files.append(f"{prefix}.log")
+        elif sim_option == MUSimulationOption.CS_MIKE_21FM:
+            log_files.append(f"{prefix}_m21fm.log")
+        elif (
+            sim_option == MUSimulationOption.CS_MIKE_COUPLING
+            or sim_option == MUSimulationOption.CS_MIKE_COUPLING_21FMModelLink
+        ):
+            log_files.append(f"{prefix}_m1d.log")
+            log_files.append(f"{prefix}_m21fm.log")
+            log_files.append(f"{prefix}_mf.log")
+
+        for file in log_files:
+            if not os.path.exists(file):
+                log_files.remove(file)
+        return log_files
+
+    def _print_log(self, log_file):
+        if os.path.exists(log_file):
+            with open(log_file) as f:
+                lines = f.readlines()
+                for line in lines:
+                    print(line)
+            return True
+        else:
+            return False
+
+    @property
+    def result_files(self):
+        """Get the current simulation result files path
+
+        Returns
+        -------
+        string list
+            The result files path of current simulation
+        """
+        if self._result_files is None:
+            sim_muid = self._get_active_muid()
+            self._result_files = self._get_result_files(sim_muid)
+        return self._result_files

notebooks/EngineAccessor.ipynb

@@ -75,6 +75,24 @@
     "data_access.close_database()\n",
     "os.chdir(current_dir)"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "a775728e",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\"\"\"To run flood engine\"\"\"\n",
+    "from mikeplus import DataTableAccess\n",
+    "from mikeplus.engines.floodEngine import FloodEngine\n",
+    "\n",
+    "data_access = DataTableAccess(\"../tests/testdata/Db/2D Blue Beach/100y_combined.sqlite\")\n",
+    "data_access.open_database()\n",
+    "engine = FloodEngine(data_access.datatables)\n",
+    "engine.run()\n",
+    "data_access.close_database()\n"
+   ]
   }
  ],
  "metadata": {

tests/test_engine.py

@@ -4,6 +4,7 @@
 from mikeplus.engines.engine1d import Engine1D
 from mikeplus.engines.epanet import EPANET
 from mikeplus.engines.swmm import SWMM
+from mikeplus.engines.flood_engine import FloodEngine
 
 
 @pytest.mark.slow(reason="Test run slow because of the license check.")
@@ -58,3 +59,23 @@ def test_swmm_engine():
     data_access.close_database()
     os.chdir(current_dir)
     assert os.path.exists(result_file)
+
+
+@pytest.mark.license_required
+def test_flood_engine():
+    dbFile = os.path.join(
+        "tests", "testdata", "Db", "2D Blue Beach", "100y_combined.sqlite"
+    )
+    data_access = DataTableAccess(dbFile)
+    data_access.open_database()
+    engine = FloodEngine(data_access.datatables)
+    result_files = engine.result_files
+    if result_files is not None:
+        for file in result_files:
+            if os.path.exists(file):
+                os.remove(file)
+    engine.run()
+    data_access.close_database()
+    assert result_files is not None
+    for file in result_files:
+        assert os.path.exists(result_files)