refactor: Deprecate 'task' from enhanced meshing workflow. (#3241)

* fix: Compound child creation and naming.

* refactor: Deprecate 'task' from enhanced meshing workflow.

* Update deprecation warning.

* Update Deprecation message.

* Use APIName instead of helpString whereever avaialble.

* Update logic to clean flag after usage.

* Add test.

* Update logic.

* Update tests/test_new_meshing_workflow.py

* Update src/ansys/fluent/core/workflow.py

Author: prmukherj

src/ansys/fluent/core/workflow.py

@@ -379,7 +379,9 @@ def python_name(self) -> str:
 
     def _set_python_name(self):
         this_command = self._command()
-        self._python_name = camel_to_snake_case(this_command.get_attr("helpString"))
+        self._python_name = camel_to_snake_case(
+            this_command.get_attr("APIName") or this_command.get_attr("helpString")
+        )
         self._cache_data(this_command)
 
     def _cache_data(self, command):
@@ -429,7 +431,10 @@ def __getattr__(self, attr):
             return ArgumentWrapper(self, attr)
         except Exception as ex:
             logger.debug(str(ex))
-        return self._task_objects.get(attr, None)
+        result = self._task_objects.get(attr, None)
+        if result:
+            return result
+        return super().__getattribute__(attr)
 
     def __setattr__(self, attr, value):
         logger.debug(f"BaseTask.__setattr__({attr}, {value})")
@@ -490,18 +495,15 @@ def update_child_tasks(self, setup_type_changed: bool):
         """Update child tasks."""
         self._task.UpdateChildTasks(SetupTypeChanged=setup_type_changed)
 
-    def insert_compound_child_task(self):
-        """Insert a compound child task."""
-        return self._task.InsertCompoundChildTask()
-
     def _get_next_python_task_names(self) -> list[str]:
         self._python_task_names_map = {}
         for command_name in self._task.GetNextPossibleTasks():
+            comm_obj = getattr(
+                self._command_source._command_source, command_name
+            ).create_instance()
             self._python_task_names_map[
                 camel_to_snake_case(
-                    getattr(self._command_source._command_source, command_name)
-                    .create_instance()
-                    .get_attr("helpString")
+                    comm_obj.get_attr("APIName") or comm_obj.get_attr("helpString")
                 )
             ] = command_name
         return list(self._python_task_names_map.keys())
@@ -1163,6 +1165,10 @@ def _add_child(self, state: Optional[dict] = None) -> None:
         state.update({"add_child": "yes"})
         self.arguments.update_dict(state)
 
+    def insert_compound_child_task(self):
+        """Insert a compound child task."""
+        return self.add_child_and_update()
+
     def add_child_and_update(self, state=None, defer_update=None):
         """Add a child to this CompoundTask and update.
 
@@ -1183,18 +1189,20 @@ def add_child_and_update(self, state=None, defer_update=None):
             )
         self._command_source._compound_child = True
         self._command_source._parent_of_compound_child = py_name
-        if self._fluent_version >= FluentVersion.v241:
-            if defer_update is None:
-                defer_update = False
-            self._task.AddChildAndUpdate(DeferUpdate=defer_update)
-        else:
-            if defer_update is not None:
-                warnings.warn(
-                    " The 'defer_update()' method is supported in Fluent 2024 R1 and later.",
-                    PyFluentUserWarning,
-                )
-            self._task.AddChildAndUpdate()
-        self._command_source._compound_child = False
+        try:
+            if self._fluent_version >= FluentVersion.v241:
+                if defer_update is None:
+                    defer_update = False
+                self._task.AddChildAndUpdate(DeferUpdate=defer_update)
+            else:
+                if defer_update is not None:
+                    warnings.warn(
+                        "The 'defer_update()' method is supported in Fluent 2024 R1 and later.",
+                        PyFluentUserWarning,
+                    )
+                self._task.AddChildAndUpdate()
+        finally:
+            self._command_source._compound_child = False
         return self.last_child()
 
     def last_child(self) -> BaseTask:
@@ -1324,6 +1332,29 @@ def task(self, name: str) -> BaseTask:
 
         The wrapper adds extra functionality.
 
+        Parameters
+        ----------
+        name : str
+            Task name - the display name, not the internal ID.
+        Returns
+        -------
+        BaseTask
+            wrapped task object.
+        """
+        py_name = self.tasks()[
+            [repr(task) for task in self.tasks()].index(repr(self._task(name)))
+        ].python_name()
+        warnings.warn(
+            f"'task' is deprecated -> Use '{py_name}' instead.",
+            PyFluentDeprecationWarning,
+        )
+        return self._task(name)
+
+    def _task(self, name: str) -> BaseTask:
+        """Get a TaskObject by name, in a ``BaseTask`` wrapper.
+
+        The wrapper adds extra functionality.
+
         Parameters
         ----------
         name : str
@@ -1393,7 +1424,7 @@ def inactive_tasks() -> list:
     def __getattr__(self, attr):
         """Delegate attribute lookup to the wrapped workflow object."""
         if attr in self._repeated_task_python_name_display_text_map:
-            return self.task(self._repeated_task_python_name_display_text_map[attr])
+            return self._task(self._repeated_task_python_name_display_text_map[attr])
         _task_object = self._task_objects.get(attr)
         if _task_object:
             return _task_object
@@ -1453,7 +1484,7 @@ def _workflow_and_task_list_state(self) -> Tuple[dict, dict]:
     def _task_by_id_impl(self, task_id, workflow_state):
         task_key = "TaskObject:" + task_id
         task_state = workflow_state[task_key]
-        return self.task(task_state["_name_"])
+        return self._task(task_state["_name_"])
 
     def _task_by_id(self, task_id):
         workflow_state = self._workflow_state()
@@ -1536,7 +1567,9 @@ def _populate_first_tasks_python_name_command_id_map(self):
                 if isinstance(command_obj, PyCommand):
                     command_obj_instance = command_obj.create_instance()
                     if not command_obj_instance.get_attr("requiredInputs"):
-                        help_str = command_obj_instance.get_attr("helpString")
+                        help_str = command_obj_instance.get_attr(
+                            "APIName"
+                        ) or command_obj_instance.get_attr("helpString")
                         if help_str:
                             self._initial_task_python_names_map[help_str] = command
                     del command_obj_instance

tests/test_new_meshing_workflow.py

@@ -1703,3 +1703,28 @@ def test_accessors_for_argument_sub_items(new_meshing_session):
         msg.value.args[0]
         == "'PyTextualCommandArgumentsSubItem' object has no attribute 'min'"
     )
+
+
+@pytest.mark.skip("https://github.com/ansys/pyfluent/issues/3263")
+@pytest.mark.codegen_required
+@pytest.mark.fluent_version(">=25.1")
+def test_scenario_with_common_python_names_from_fdl(new_meshing_session):
+    meshing = new_meshing_session
+
+    fault_tolerant = meshing.fault_tolerant()
+
+    # Check if all task names are unique.
+    assert len(fault_tolerant.task_names()) == len(set(fault_tolerant.task_names()))
+
+    # APIName from fdl file
+    assert "create_volume_mesh" in fault_tolerant.task_names()
+    assert "generate_volume_mesh" in fault_tolerant.task_names()
+    assert "generate_surface_mesh" in fault_tolerant.task_names()
+
+    watertight = meshing.watertight()
+    # Check if all task names are unique.
+    assert len(watertight.task_names()) == len(set(watertight.task_names()))
+
+    two_dimensional = meshing.two_dimensional_meshing()
+    # Check if all task names are unique.
+    assert len(two_dimensional.task_names()) == len(set(two_dimensional.task_names()))