feat: Update server based enhanced meshing workflow.

doc/changelog.d/4779.added.md

@@ -0,0 +1 @@
+Update server based enhanced meshing workflow.

doc/source/user_guide/meshing/new_meshing_workflows.rst

@@ -519,7 +519,7 @@ Load workflow
     from ansys.fluent.core import examples
 
     saved_workflow_path = examples.download_file(
-        "sample_watertight_workflow.wft", "pyfluent/meshing_workflow"
+        "sample_watertight_workflow.wft", "pyfluent/meshing_workflows"
     )
     meshing_session = pyfluent.launch_fluent(
         mode=pyfluent.FluentMode.MESHING, precision=pyfluent.Precision.DOUBLE, processor_count=2

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

@@ -152,9 +152,32 @@ def __init__(
             fluent_version=fluent_version,
             initialize=initialize,
         )
+        self._parent_workflow = workflow
         self._part_management = part_management
         self._pm_file_management = pm_file_management
 
+    @property
+    def parts(self) -> PyMenuGeneric | None:
+        """Access part-management in fault-tolerant mode.
+
+        Returns
+        -------
+        PyMenuGeneric | None
+            Part-management.
+        """
+        return self._parent_workflow.parts
+
+    @property
+    def parts_files(self):
+        """Access the part-management file-management object in fault-tolerant mode.
+
+        Returns
+        -------
+        PyMenuGeneric | None
+            File management object in the part management object.
+        """
+        return self._parent_workflow.parts_files
+
     @property
     def part_management(self) -> PyMenuGeneric | None:
         """Access part-management in fault-tolerant mode.
@@ -164,6 +187,7 @@ def part_management(self) -> PyMenuGeneric | None:
         PyMenuGeneric | None
             Part-management.
         """
+        # TODO: Remove this after migrating to the new workflow
         return self._part_management
 
     @property
@@ -175,6 +199,7 @@ def pm_file_management(self):
         PyMenuGeneric | None
             File management object in the part management object.
         """
+        # TODO: Remove this after migrating to the new workflow
         return self._pm_file_management
 
 

src/ansys/fluent/core/workflow_new.py

@@ -49,7 +49,7 @@
 import re
 from typing import ValuesView
 
-from ansys.fluent.core.services.datamodel_se import PyMenu
+from ansys.fluent.core.services.datamodel_se import PyCommand, PyMenu
 from ansys.fluent.core.utils.fluent_version import FluentVersion
 
 
@@ -171,9 +171,11 @@ def command_name_to_task_name(meshing_root, command_name: str) -> str:
     # TODO: This is a fix only for 26R1 as the server lacks the mechanism to return mapped values
     #  for '<Task Object>.get_next_possible_tasks()'.
     command_instance = getattr(meshing_root, command_name).create_instance()
-    return command_instance.get_attr("APIName") or command_instance.get_attr(
+    retval = command_instance.get_attr("APIName") or command_instance.get_attr(
         "helpString"
     )
+    del command_instance
+    return retval
 
 
 class Workflow:
@@ -509,8 +511,67 @@ def delete_tasks(self, list_of_tasks: list[TaskObject]):
 
         self._workflow.general.delete_tasks(list_of_tasks=items_to_be_deleted)
 
+    @property
+    def insertable_tasks(self):
+        """Tasks that can be inserted on a blank workflow."""
+        return self._FirstTask(self)
+
+    class _FirstTask:
+        def __init__(self, workflow):
+            """Initialize an ``_FirstTask`` instance."""
+            self._workflow = workflow
+            self._insertable_tasks = []
+            self._initial_task_map = {}
+
+            for command in dir(workflow._command_source):
+                if command in ["SwitchToSolution", "set_state", "setState"]:
+                    continue
+                command_obj = getattr(workflow._command_source, command)
+                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(
+                            "APIName"
+                        ) or command_obj_instance.get_attr("helpString")
+                        if help_str:
+                            self._initial_task_map[help_str] = command
+                    del command_obj_instance
+
+            if self._workflow._workflow.general.workflow.task_list() == []:
+                for item in self._initial_task_map:
+                    insertable_task = type("Insert", (self._Insert,), {})(
+                        self._workflow,
+                        item,
+                        self._initial_task_map,
+                    )
+                    setattr(self, item, insertable_task)
+                    self._insertable_tasks.append(insertable_task)
+
+        def __call__(self):
+            return self._insertable_tasks
+
+        class _Insert:
+            def __init__(self, workflow, name, task_map):
+                """Initialize an ``_Insert`` instance."""
+                self._workflow = workflow
+                self._name = name
+                self._task_map = task_map
+
+            def insert(self):
+                """Insert a task in the workflow."""
+                return self._workflow.general.insert_new_task(
+                    command_name=self._task_map[self._name]
+                )
+
+            def __repr__(self):
+                return f"<Insertable '{self._name}' task>"
+
     def __getattr__(self, item):
         """Enable attribute-style access to tasks."""
+        if item in ["parts", "parts_files"]:
+            raise AttributeError(
+                f"'{item}' is only supported in Fault-tolerant Meshing workflows."
+            )
         if item not in self._task_dict:
             self.tasks()
         if item in self._task_dict:

tests/test_server_meshing_workflow.py

@@ -1660,3 +1660,47 @@ def test_new_watertight_workflow_using_traversal(
     assert new_meshing_session_wo_exit.is_active() is False
     solver.exit()
     assert solver.is_active() is False
+
+
+@pytest.mark.codegen_required
+@pytest.mark.fluent_version(">=26.1")
+def test_created_workflow(new_meshing_session, use_server_meshing_workflow):
+    meshing = new_meshing_session
+    created_workflow = meshing.create_workflow()
+
+    assert sorted([repr(x) for x in created_workflow.insertable_tasks()]) == sorted(
+        [
+            "<Insertable 'create_group' task>",
+            "<Insertable 'import_geometry' task>",
+            "<Insertable 'load_cad_geometry' task>",
+            "<Insertable 'import_cad_and_part_management' task>",
+            "<Insertable 'custom_journal_task' task>",
+        ]
+    )
+
+    created_workflow.insertable_tasks()[1].insert()
+
+    assert created_workflow.insertable_tasks() == []
+
+    assert "<Insertable 'add_local_sizing' task>" in [
+        repr(x) for x in created_workflow.import_geometry.insertable_tasks()
+    ]
+    created_workflow.import_geometry.insertable_tasks.add_local_sizing.insert()
+    assert "<Insertable 'add_local_sizing' task>" not in [
+        repr(x) for x in created_workflow.import_geometry.insertable_tasks()
+    ]
+    assert sorted(created_workflow.task_names()) == sorted(
+        ["import_geometry", "add_local_sizing_wtm"]
+    )
+
+
+@pytest.mark.codegen_required
+@pytest.mark.fluent_version(">=26.1")
+def test_loaded_workflow(new_meshing_session, use_server_meshing_workflow):
+    meshing = new_meshing_session
+    saved_workflow_path = examples.download_file(
+        "sample_watertight_workflow.wft", "pyfluent/meshing_workflows"
+    )
+    loaded_workflow = meshing.load_workflow(file_path=saved_workflow_path)
+    assert "set_up_rotational_periodic_boundaries" in loaded_workflow.task_names()
+    assert "import_boi_geometry" in loaded_workflow.task_names()