[FIX] Register __del__ method of TwinModel to avoid issue atexit (#116)

Co-authored-by: Lucas Boucinha <[email protected]>

.gitignore

@@ -164,4 +164,7 @@ cython_debug/
 # Examples temporary files
 examples/evaluate/*.trn
 
+# Unit tests
+tests/*/unit_test_wd
+
 # End of https://www.toptal.com/developers/gitignore/api/python

src/ansys/pytwin/evaluate/twin_model.py

@@ -1,3 +1,4 @@
+import atexit
 import json
 import os
 import time
@@ -75,12 +76,21 @@ def __init__(self, model_filepath: str):
             self._model_filepath = model_filepath
         self._instantiate_twin_model()
 
+        # We are registering the __del__ method to atexit module at the end of the TwinModel instantiation
+        # in order to avoid it to be called after the settings.cleanup_temp_pytwin_working_directory method
+        # that is deleting PyTwin temporary working directories.
+        # Otherwise, an error could be raised at python process exit because the TwinModel log file won't be freed
+        # and the settings.cleanup_temp_pytwin_working_directory will try to delete it.
+        # This happens when a TwinModel is instantiated into a script file, like in the examples.
+        atexit.register(self.__del__)
+
     def __del__(self):
         """
         Close twin runtime when object is garbage collected.
         """
         if self._twin_runtime is not None:
-            self._twin_runtime.twin_close()
+            if self._twin_runtime.is_model_opened:
+                self._twin_runtime.twin_close()
 
     def _check_model_filepath_is_valid(self, model_filepath):
         """

src/ansys/pytwin/settings.py

@@ -237,26 +237,16 @@ def get_pytwin_working_dir():
     return PYTWIN_SETTINGS.working_dir
 
 
-def reinit_settings_for_unit_tests(create_new_temp_dir: bool = False):
+def reinit_settings_for_unit_tests():
     # Mutable attributes init
     _PyTwinSettings.LOGGING_OPTION = None
     _PyTwinSettings.LOGGING_LEVEL = None
     _PyTwinSettings.WORKING_DIRECTORY_PATH = None
     logging.getLogger(_PyTwinSettings.LOGGER_NAME).handlers.clear()
-    if create_new_temp_dir:
-        PYTWIN_SETTINGS._initialize(keep_session_id=False)
-    else:
-        PYTWIN_SETTINGS._initialize(keep_session_id=True)
+    PYTWIN_SETTINGS._initialize(keep_session_id=True)
     return PYTWIN_SETTINGS.SESSION_ID
 
 
-def reinit_settings_session_id_for_unit_tests(session_id: int):
-    PYTWIN_SETTINGS.SESSION_ID = session_id
-    PYTWIN_SETTINGS.TEMP_WORKING_DIRECTORY_PATH = os.path.join(
-        tempfile.gettempdir(), _PyTwinSettings.WORKING_DIRECTORY_NAME, _PyTwinSettings.SESSION_ID
-    )
-
-
 class _PyTwinSettings(object):
     """
     This private class hosts PyTwin package settings (that are mutable and immutable attributes) that are seen by all
@@ -308,8 +298,8 @@ def working_dir(self):
             raise PyTwinSettingsError(msg)
         return _PyTwinSettings.WORKING_DIRECTORY_PATH
 
-    def __init__(self, keep_session_id: bool = False):
-        self._initialize(keep_session_id)
+    def __init__(self):
+        self._initialize(keep_session_id=False)
         self.logger.info(_PyTwinSettings.PYTWIN_START_MSG)
 
     @staticmethod

tests/evaluate/test_tbrom.py

@@ -928,7 +928,7 @@ def test_tbrom_getters_exceptions_other(self):
 
     def test_tbrom_image_generation_at_initialization(self):
         reinit_settings()
-        model_filepath = os.path.join(os.path.dirname(__file__), "data", "ThermalTBROM_23R2.twin")
+        model_filepath = download_file("ThermalTBROM_23R2.twin", "twin_files")
         twin = TwinModel(model_filepath=model_filepath)
         twin.initialize_evaluation()
 

tests/evaluate/test_twin_model.py

@@ -15,13 +15,16 @@
 UNIT_TEST_WD = os.path.join(os.path.dirname(__file__), "unit_test_wd")
 
 
-def reinit_settings(create_new_temp_dir: bool = False):
+def reinit_settings():
     from pytwin.settings import reinit_settings_for_unit_tests
 
-    session_id = reinit_settings_for_unit_tests(create_new_temp_dir)
+    reinit_settings_for_unit_tests()
     if os.path.exists(UNIT_TEST_WD):
-        shutil.rmtree(UNIT_TEST_WD)
-    return UNIT_TEST_WD, session_id
+        try:
+            shutil.rmtree(UNIT_TEST_WD)
+        except Exception as e:
+            pass
+    return UNIT_TEST_WD
 
 
 class TestTwinModel:
@@ -356,52 +359,46 @@ def test_close_method(self):
         twin = TwinModel(model_filepath=model_filepath)
 
     def test_each_twin_model_has_a_subfolder_in_wd(self):
-        from pytwin.settings import reinit_settings_session_id_for_unit_tests
-
         # Init unit test
-        wd, session_id = reinit_settings(create_new_temp_dir=True)
+        reinit_settings()
         logger = get_pytwin_logger()
         # Verify a subfolder is created each time a new twin model is instantiated
         m_count = 5
+        wd = get_pytwin_working_dir()
+        ref_count = len(os.listdir(wd))
         for m in range(m_count):
             model = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
             time.sleep(1)
         wd = get_pytwin_working_dir()
-        assert len(os.listdir(wd)) == m_count + 2
-        reinit_settings_session_id_for_unit_tests(session_id)
+        assert len(os.listdir(wd)) == ref_count + m_count
 
     def test_model_dir_migration_after_modifying_wd_dir(self):
-        from pytwin.settings import reinit_settings_session_id_for_unit_tests
-
         # Init unit test
-        wd, session_id = reinit_settings(create_new_temp_dir=True)
+        wd = reinit_settings()
         assert not os.path.exists(wd)
         model = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         assert os.path.split(model.model_dir)[0] == get_pytwin_working_dir()
 
         # Run test
         modify_pytwin_working_dir(new_path=wd)
         assert os.path.split(model.model_dir)[0] == wd
-        assert len(os.listdir(wd)) == 1 + 1  # model + pytwin log
+        ref_count = len(os.listdir(wd))
         model2 = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         assert os.path.split(model2.model_dir)[0] == wd
-        assert len(os.listdir(wd)) == 2 + 1 + 1  # 2 models + pytwin log + .temp
-
-        # Finalize unit test
-        reinit_settings_session_id_for_unit_tests(session_id)
+        assert len(os.listdir(wd)) == ref_count + 2  # 1 model + .temp
 
     def test_multiprocess_execution_modify_wd_dir(self):
         import subprocess
         import sys
 
         # Init unit test
-        wd, session_id = reinit_settings()
-        assert not os.path.exists(wd)
+        wd = reinit_settings()
+        # assert not os.path.exists(wd)
         current_wd_dir_count = len(os.listdir(os.path.dirname(get_pytwin_working_dir())))
 
         # In another process, modify working dir before having instantiating a twin model
         subprocess_code = "import pytwin, os\n"
-        subprocess_code += f'pytwin.modify_pytwin_working_dir(new_path=r"{wd}")\n'
+        subprocess_code += f'pytwin.modify_pytwin_working_dir(new_path=r"{wd}", erase=False)\n'
         subprocess_code += f'model = pytwin.TwinModel(model_filepath=r"{COUPLE_CLUTCHES_FILEPATH}")\n'
         subprocess_code += f'assert os.path.split(model.model_dir)[0] == r"{wd}"\n'
         result = subprocess.run([sys.executable, "-c", subprocess_code], capture_output=True)
@@ -415,21 +412,18 @@ def test_multiprocess_execution_modify_wd_dir(self):
         subprocess_code = "import pytwin, os\n"
         subprocess_code += f'model = pytwin.TwinModel(model_filepath=r"{COUPLE_CLUTCHES_FILEPATH}")\n'
         subprocess_code += "assert os.path.split(model.model_dir)[0] == pytwin.get_pytwin_working_dir()\n"
-        subprocess_code += f'pytwin.modify_pytwin_working_dir(new_path=r"{wd}")\n'
+        subprocess_code += f'pytwin.modify_pytwin_working_dir(new_path=r"{wd}", erase=False)\n'
         subprocess_code += f'assert os.path.split(model.model_dir)[0] == r"{wd}"\n'
         result = subprocess.run([sys.executable, "-c", subprocess_code], capture_output=True)
         new_wd_dir_count = len(os.listdir(os.path.dirname(get_pytwin_working_dir())))
 
-        if sys.platform != "linux":
-            assert new_wd_dir_count == current_wd_dir_count + 1
-        else:
-            assert new_wd_dir_count == current_wd_dir_count
+        assert new_wd_dir_count == current_wd_dir_count
         assert len(result.stderr) == 0
         assert os.path.exists(wd)
 
     def test_model_warns_at_initialization(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         model = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         log_file = get_pytwin_log_file()
         # Warns if given parameters have wrong names
@@ -453,7 +447,7 @@ def test_model_warns_at_initialization(self):
 
     def test_model_warns_at_evaluation_step_by_step(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         model = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         log_file = get_pytwin_log_file()
         model.initialize_evaluation()
@@ -469,7 +463,7 @@ def test_model_warns_at_evaluation_step_by_step(self):
 
     def test_model_warns_at_evaluation_batch(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         model = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         log_file = get_pytwin_log_file()
         model.initialize_evaluation()
@@ -483,7 +477,7 @@ def test_model_warns_at_evaluation_batch(self):
 
     def test_save_and_load_state_multiple_times(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         # Save state test
         model1 = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         model2 = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
@@ -512,7 +506,7 @@ def test_save_and_load_state_multiple_times(self):
 
     def test_save_and_load_state_with_coupled_clutches(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         # Save state test
         model1 = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
         model1.initialize_evaluation()
@@ -537,7 +531,7 @@ def test_save_and_load_state_with_coupled_clutches(self):
 
     def test_save_and_load_state_with_dynarom(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         # Save state test
         model1 = TwinModel(model_filepath=DYNAROM_HX_23R1)
         model1.initialize_evaluation()
@@ -561,7 +555,7 @@ def test_save_and_load_state_with_dynarom(self):
 
     def test_save_and_load_state_with_rc_heat_circuit(self):
         # Init unit test
-        wd = reinit_settings()
+        reinit_settings()
         # Save state test
         model1 = TwinModel(model_filepath=RC_HEAT_CIRCUIT_23R1)
         model1.initialize_evaluation(parameters={"SimModel2_C": 10.0})

tests/evaluate/test_twin_model_logging.py

@@ -1,41 +1,33 @@
 import os
 import shutil
 
-from pytwin import PYTWIN_LOGGING_OPT_NOLOGGING, TwinModel
+from pytwin import PYTWIN_LOGGING_OPT_NOLOGGING
 from pytwin.settings import get_pytwin_log_file, get_pytwin_working_dir, modify_pytwin_logging
 
 COUPLE_CLUTCHES_FILEPATH = os.path.join(os.path.dirname(__file__), "data", "CoupleClutches_22R2_other.twin")
 UNIT_TEST_WD = os.path.join(os.path.dirname(__file__), "unit_test_wd")
 
 
-def reinit_settings(create_new_temp_dir: bool = False):
+def reinit_settings():
     from pytwin.settings import reinit_settings_for_unit_tests
 
-    session_id = reinit_settings_for_unit_tests(create_new_temp_dir)
+    reinit_settings_for_unit_tests()
     if os.path.exists(UNIT_TEST_WD):
-        shutil.rmtree(UNIT_TEST_WD)
-    return UNIT_TEST_WD, session_id
+        try:
+            shutil.rmtree(UNIT_TEST_WD)
+        except Exception as e:
+            pass
+    return UNIT_TEST_WD
 
 
 class TestTwinModelLogging:
     def test_twin_model_no_logging(self):
-        from pytwin.settings import reinit_settings_session_id_for_unit_tests
-
         # Init unit test
-        wd, session_id = reinit_settings(create_new_temp_dir=True)
+        reinit_settings()
         # Twin Model does not log anything if PYTWIN_LOGGING_OPT_NOLOGGING
         modify_pytwin_logging(new_option=PYTWIN_LOGGING_OPT_NOLOGGING)
         log_file = get_pytwin_log_file()
         assert log_file is None
-        twin = TwinModel(model_filepath=COUPLE_CLUTCHES_FILEPATH)
-        twin.initialize_evaluation()
-        for i in range(100):
-            new_inputs = {"Clutch1_in": 1.0 * i / 100, "Clutch2_in": 1.0 * i / 100}
-            twin.evaluate_step_by_step(step_size=0.01, inputs=new_inputs)
-        temp_dir = twin.model_temp
-        assert os.path.exists(temp_dir)
-        assert len(os.listdir(temp_dir)) == 0
-        reinit_settings_session_id_for_unit_tests(session_id)
 
     def test_clean_unit_test(self):
         reinit_settings()