fix: extend unsubscribing cb for load and create wfs. (#2869)

* fix: extend unsubscribing cb for load and create wfs.

* Update test.

* Refactor workflow structure.

* Fix issue with call-back clearing.

* Exception handling in '_refresh_task_accessors' to run in PyConsole.

* Update test for reinitialize.

* Revert test for reinitialize.

* Added a TODO note.

src/ansys/fluent/core/meshing/meshing_workflow.py

@@ -69,18 +69,16 @@ def __init__(
         self._meshing = meshing
         self._name = name
         self._identifier = identifier
+        self._unsubscribe_root_affected_callback()
+        self._new_workflow(name=self._name)
 
     def reinitialize(self) -> None:
         """Reinitialize the same workflow."""
-        self._new_workflow(name=self._name, reinitialize=True)
-
-    def initialize(self) -> None:
-        """Initialize a workflow."""
         self._new_workflow(name=self._name)
 
     def __getattribute__(self, item: str):
         if (
-            item not in ["reinitialize", "initialize"]
+            item not in ["reinitialize"]
             and not item.startswith("_")
             and not getattr(self._meshing.GlobalSettings, self._identifier)()
         ):
@@ -273,11 +271,8 @@ def __init__(
             workflow=workflow, command_source=meshing, fluent_version=fluent_version
         )
         self._meshing = meshing
-        self._file_path = file_path
-
-    def load(self) -> None:
-        """Load a workflow."""
-        self._load_workflow(file_path=self._file_path)
+        self._unsubscribe_root_affected_callback()
+        self._load_workflow(file_path=file_path)
 
 
 class CreateWorkflow(Workflow):
@@ -305,7 +300,5 @@ def __init__(
             workflow=workflow, command_source=meshing, fluent_version=fluent_version
         )
         self._meshing = meshing
-
-    def create(self) -> None:
-        """Create a workflow."""
+        self._unsubscribe_root_affected_callback()
         self._create_workflow()

src/ansys/fluent/core/session_base_meshing.py

@@ -50,12 +50,6 @@ def __init__(
         self._fluent_version = fluent_version
         self._meshing_utilities = None
         self._old_workflow = None
-        self._wt_workflow = None
-        self._ft_workflow = None
-        self._2dm_workflow = None
-        self._tb_workflow = None
-        self._loaded_workflow = None
-        self._created_workflow = None
         self._part_management = None
         self._pm_file_management = None
         self._preferences = None
@@ -135,70 +129,58 @@ def workflow(self):
     @property
     def watertight_workflow(self):
         """Datamodel root of workflow exposed in object-oriented manner."""
-        if not self._wt_workflow:
-            self._wt_workflow = WorkflowMode.WATERTIGHT_MESHING_MODE.value(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                self.get_fluent_version(),
-            )
-        return self._wt_workflow
+        return WorkflowMode.WATERTIGHT_MESHING_MODE.value(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            self.get_fluent_version(),
+        )
 
     @property
     def fault_tolerant_workflow(self):
         """Datamodel root of workflow exposed in object-oriented manner."""
-        if not self._ft_workflow:
-            self._ft_workflow = WorkflowMode.FAULT_TOLERANT_MESHING_MODE.value(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                self.PartManagement,
-                self.PMFileManagement,
-                self.get_fluent_version(),
-            )
-        return self._ft_workflow
+        return WorkflowMode.FAULT_TOLERANT_MESHING_MODE.value(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            self.PartManagement,
+            self.PMFileManagement,
+            self.get_fluent_version(),
+        )
 
     @property
     def two_dimensional_meshing_workflow(self):
         """Data model root of the workflow exposed in an object-oriented manner."""
-        if not self._2dm_workflow:
-            self._2dm_workflow = WorkflowMode.TWO_DIMENSIONAL_MESHING_MODE.value(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                self.get_fluent_version(),
-            )
-        return self._2dm_workflow
+        return WorkflowMode.TWO_DIMENSIONAL_MESHING_MODE.value(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            self.get_fluent_version(),
+        )
 
     @property
     def topology_based_meshing_workflow(self):
         """Datamodel root of workflow exposed in object-oriented manner."""
-        if not self._tb_workflow:
-            self._tb_workflow = WorkflowMode.TOPOLOGY_BASED_MESHING_MODE.value(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                self.get_fluent_version(),
-            )
-        return self._tb_workflow
+        return WorkflowMode.TOPOLOGY_BASED_MESHING_MODE.value(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            self.get_fluent_version(),
+        )
 
     def load_workflow(self, file_path: str):
         """Datamodel root of workflow exposed in object-oriented manner."""
-        if not self._loaded_workflow:
-            self._loaded_workflow = LoadWorkflow(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                file_path,
-                self.get_fluent_version(),
-            )
-        return self._loaded_workflow
+        return LoadWorkflow(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            file_path,
+            self.get_fluent_version(),
+        )
 
     @property
     def create_workflow(self):
         """Datamodel root of the workflow exposed in an object-oriented manner."""
-        if not self._created_workflow:
-            self._created_workflow = CreateWorkflow(
-                _make_datamodel_module(self, "workflow"),
-                self.meshing,
-                self.get_fluent_version(),
-            )
-        return self._created_workflow
+        return CreateWorkflow(
+            _make_datamodel_module(self, "workflow"),
+            self.meshing,
+            self.get_fluent_version(),
+        )
 
     @property
     def PartManagement(self):

src/ansys/fluent/core/session_pure_meshing.py

@@ -117,27 +117,22 @@ def workflow(self):
 
     def watertight(self):
         """Get a new watertight workflow."""
-        self._base_meshing.watertight_workflow.initialize()
         return self._base_meshing.watertight_workflow
 
     def fault_tolerant(self):
         """Get a new fault-tolerant workflow."""
-        self._base_meshing.fault_tolerant_workflow.initialize()
         return self._base_meshing.fault_tolerant_workflow
 
     def two_dimensional_meshing(self):
         """Get a new 2D meshing workflow."""
-        self._base_meshing.two_dimensional_meshing_workflow.initialize()
         return self._base_meshing.two_dimensional_meshing_workflow
 
     def load_workflow(self, file_path: str):
         """Load a saved workflow."""
-        self._base_meshing.load_workflow(file_path=file_path).load()
         return self._base_meshing.load_workflow(file_path=file_path)
 
     def create_workflow(self):
         """Create a meshing workflow."""
-        self._base_meshing.create_workflow.create()
         return self._base_meshing.create_workflow
 
     def topology_based(self):

src/ansys/fluent/core/workflow.py

@@ -1466,14 +1466,13 @@ def _activate_dynamic_interface(self, dynamic_interface: bool):
         self._dynamic_interface = dynamic_interface
         self._initialize_methods(dynamic_interface=dynamic_interface)
 
-    def _new_workflow(
-        self, name: str, dynamic_interface: bool = True, reinitialize: bool = False
-    ):
-        if not reinitialize:
-            # if the same workflow is not being reinitialized, unsubscribe the root affected callback
-            if self._workflow.service in self._root_affected_cb_by_server:
-                self._root_affected_cb_by_server[self._workflow.service].unsubscribe()
-                self._root_affected_cb_by_server.pop(self._workflow.service)
+    def _unsubscribe_root_affected_callback(self):
+        # if the same workflow is not being reinitialized, unsubscribe the root affected callback
+        if self._workflow.service in self._root_affected_cb_by_server:
+            self._root_affected_cb_by_server[self._workflow.service].unsubscribe()
+            self._root_affected_cb_by_server.pop(self._workflow.service)
+
+    def _new_workflow(self, name: str, dynamic_interface: bool = True):
         self._workflow.InitializeWorkflow(WorkflowType=name)
         self._activate_dynamic_interface(dynamic_interface=dynamic_interface)
 
@@ -1558,7 +1557,12 @@ def refresh_after_sleep(_):
                     logger.debug("Already _refreshing, ...")
                 self._refreshing = True
                 logger.debug("Call _refresh_task_accessors")
-                _refresh_task_accessors(self)
+                try:
+                    _refresh_task_accessors(self)
+                except Exception:
+                    # Is there a more specific Exception derived class
+                    # for which we know it is correct to pass?
+                    pass
                 self._refresh_count += 1
                 self._refreshing = False
 

tests/test_new_meshing_workflow.py

@@ -1497,7 +1497,15 @@ def test_independent_meshing_sessions_without_dm_caching(
 
 
 @pytest.mark.codegen_required
-@pytest.mark.fluent_version(">=24.1")
+@pytest.mark.fluent_version(">=24.2")
 def test_switching_workflow_interface(new_mesh_session):
-    watertight = new_mesh_session.watertight()
-    fault_tolerant = new_mesh_session.fault_tolerant()
+    wt1 = new_mesh_session.watertight()
+    ft = new_mesh_session.fault_tolerant()
+    tw = new_mesh_session.two_dimensional_meshing()
+    cw = new_mesh_session.create_workflow()
+    saved_workflow_path = examples.download_file(
+        "sample_watertight_workflow.wft", "pyfluent/meshing_workflows"
+    )
+    lw = new_mesh_session.load_workflow(file_path=saved_workflow_path)
+    wt2 = new_mesh_session.watertight()
+    del wt1, ft, tw, cw, lw, wt2