feat: Load saved workflows. (#2718)

* feat: Load saved workflows.

* Update docstring.

* feat: Create new workflows (#2722)

* feat: Create new workflows

* Fix doc.

* fix behavior.

* Fix docs.

* Clean-up.

* Update behavior.

* Fix test.

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

@@ -240,3 +240,67 @@ class WorkflowMode(Enum):
     FAULT_TOLERANT_MESHING_MODE = FaultTolerantMeshingWorkflow
     TWO_DIMENSIONAL_MESHING_MODE = TwoDimensionalMeshingWorkflow
     TOPOLOGY_BASED_MESHING_MODE = TopologyBasedMeshingWorkflow
+
+
+class LoadWorkflow(Workflow):
+    """Provides a specialization of the workflow wrapper for a loaded workflow."""
+
+    def __init__(
+        self,
+        workflow: PyMenuGeneric,
+        meshing: PyMenuGeneric,
+        file_path: str,
+        fluent_version: FluentVersion,
+    ) -> None:
+        """Initialize LoadWorkflow.
+
+        Parameters
+        ----------
+        workflow : PyMenuGeneric
+            Underlying workflow object.
+        meshing : PyMenuGeneric
+            Meshing object.
+        file_path: str
+            Path to the saved workflow.
+        fluent_version: FluentVersion
+            Version of Fluent in this session.
+        """
+        super().__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)
+
+
+class CreateWorkflow(Workflow):
+    """Provides a specialization of the workflow wrapper for newly created workflow."""
+
+    def __init__(
+        self,
+        workflow: PyMenuGeneric,
+        meshing: PyMenuGeneric,
+        fluent_version: FluentVersion,
+    ) -> None:
+        """Initialize CreateWorkflow.
+
+        Parameters
+        ----------
+        workflow : PyMenuGeneric
+            Underlying workflow object.
+        meshing : PyMenuGeneric
+            Meshing object.
+        fluent_version: FluentVersion
+            Version of Fluent in this session.
+        """
+        super().__init__(
+            workflow=workflow, command_source=meshing, fluent_version=fluent_version
+        )
+        self._meshing = meshing
+
+    def create(self) -> None:
+        """Create a workflow."""
+        self._create_workflow()

src/ansys/fluent/core/session_base_meshing.py

@@ -4,7 +4,11 @@
 import logging
 
 from ansys.fluent.core.fluent_connection import FluentConnection
-from ansys.fluent.core.meshing.meshing_workflow import WorkflowMode
+from ansys.fluent.core.meshing.meshing_workflow import (
+    CreateWorkflow,
+    LoadWorkflow,
+    WorkflowMode,
+)
 from ansys.fluent.core.services.datamodel_se import PyMenuGeneric
 from ansys.fluent.core.services.datamodel_tui import TUIMenu
 from ansys.fluent.core.session_shared import _CODEGEN_MSG_DATAMODEL, _CODEGEN_MSG_TUI
@@ -48,6 +52,8 @@ def __init__(
         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
@@ -199,6 +205,28 @@ def topology_based_meshing_workflow(self):
             )
         return self._tb_workflow
 
+    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(
+                self._workflow_se,
+                self.meshing,
+                file_path,
+                self.get_fluent_version(),
+            )
+        return self._loaded_workflow
+
+    @property
+    def create_workflow(self):
+        """Datamodel root of workflow exposed in object-oriented manner."""
+        if not self._created_workflow:
+            self._created_workflow = CreateWorkflow(
+                self._workflow_se,
+                self.meshing,
+                self.get_fluent_version(),
+            )
+        return self._created_workflow
+
     @property
     def PartManagement(self):
         """Datamdoel root of PartManagement."""

src/ansys/fluent/core/session_pure_meshing.py

@@ -111,6 +111,16 @@ def two_dimensional_meshing(self):
         self._base_meshing.two_dimensional_meshing_workflow.reinitialize()
         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 new meshing workflow."""
+        self._base_meshing.create_workflow.create()
+        return self._base_meshing.create_workflow
+
     def topology_based(self):
         """Get a new topology-based meshing workflow.
 

src/ansys/fluent/core/session_pure_meshing.pyi

@@ -20,6 +20,8 @@ class PureMeshing:
     def fault_tolerant(self): ...
     def two_dimensional_meshing(self): ...
     def topology_based(self): ...
+    def load_workflow(self, file_path: str): ...
+    def create_workflow(self): ...
     @property
     def PartManagement(self) -> partmanagement_root: ...
     @property

src/ansys/fluent/core/workflow.py

@@ -12,6 +12,7 @@
 from ansys.fluent.core.data_model_cache import DataModelCache
 from ansys.fluent.core.services.datamodel_se import (
     PyCallableStateObject,
+    PyCommand,
     PyMenuGeneric,
     PySingletonCommandArgumentsSubItem,
 )
@@ -567,7 +568,7 @@ def __init__(self, base_task, name):
 
             def insert(self):
                 """Insert a task in the workflow."""
-                self._base_task.insert_next_task(task_name=self._name)
+                return self._base_task.insert_next_task(task_name=self._name)
 
             def __repr__(self):
                 return f"<Insertable '{self._name}' task>"
@@ -1292,6 +1293,7 @@ def __init__(
         self._help_string_display_id_map = {}
         self._help_string_display_text_map = {}
         self._repeated_task_help_string_display_text_map = {}
+        self._initial_task_python_names_map = {}
         self._unwanted_attrs = {
             "reset_workflow",
             "initialize_workflow",
@@ -1455,11 +1457,82 @@ def _attr_from_wrapped_workflow(self, attr):
         except AttributeError:
             pass
 
-    def _new_workflow(self, name: str, dynamic_interface: bool = True):
+    def _activate_dynamic_interface(self, dynamic_interface: bool):
         self._dynamic_interface = dynamic_interface
-        self._workflow.InitializeWorkflow(WorkflowType=name)
         self._initialize_methods(dynamic_interface=dynamic_interface)
 
+    def _new_workflow(self, name: str, dynamic_interface: bool = True):
+        self._workflow.InitializeWorkflow(WorkflowType=name)
+        self._activate_dynamic_interface(dynamic_interface=dynamic_interface)
+
+    def _load_workflow(self, file_path: str, dynamic_interface: bool = True):
+        self._workflow.LoadWorkflow(FilePath=file_path)
+        self._activate_dynamic_interface(dynamic_interface=dynamic_interface)
+
+    def get_initial_task_list_while_creating_new_workflow(self):
+        """Get list of independent tasks that can be inserted at initial level while
+        creating a new workflow."""
+        self._get_first_tasks_help_string_command_id_map()
+        return list(self._initial_task_python_names_map)
+
+    def _create_workflow(self, dynamic_interface: bool = True):
+        self._workflow.CreateNewWorkflow()
+        self._activate_dynamic_interface(dynamic_interface=dynamic_interface)
+
+    @property
+    def first_tasks(self):
+        """Tasks that can be inserted on a blank workflow."""
+        return self._FirstTask(self)
+
+    class _FirstTask:
+        def __init__(self, workflow):
+            """Initialize _FirstTask."""
+            self._workflow = workflow
+            self._insertable_tasks = []
+            if len(self._workflow.get_available_task_names()) == 0:
+                for (
+                    item
+                ) in self._workflow.get_initial_task_list_while_creating_new_workflow():
+                    insertable_task = type("Insert", (self._Insert,), {})(
+                        self._workflow, item
+                    )
+                    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):
+                """Initialize an ``_Insert`` instance."""
+                self._workflow = workflow
+                self._name = name
+
+            def insert(self):
+                """Insert a task in the workflow."""
+                return self._workflow._workflow.InsertNewTask(
+                    CommandName=self._workflow._initial_task_python_names_map[
+                        self._name
+                    ]
+                )
+
+            def __repr__(self):
+                return f"<Insertable '{self._name}' task>"
+
+    def _get_first_tasks_help_string_command_id_map(self):
+        if not self._initial_task_python_names_map:
+            for command in dir(self._command_source):
+                if command in ["SwitchToSolution", "set_state"]:
+                    continue
+                command_obj = getattr(self._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("helpString")
+                        if help_str:
+                            self._initial_task_python_names_map[help_str] = command
+                    del command_obj_instance
+
     def _initialize_methods(self, dynamic_interface: bool):
         _init_task_accessors(self)
         if dynamic_interface:

tests/test_new_meshing_workflow.py

@@ -628,7 +628,9 @@ def test_snake_case_attrs_in_new_meshing_workflow(new_mesh_session):
         "mixing_elbow.pmdb", "pyfluent/mixing_elbow"
     )
     watertight = new_mesh_session.watertight()
-    _assert_snake_case_attrs(dir(watertight))
+    dir_watertight = dir(watertight)
+    dir_watertight.remove("_FirstTask")
+    _assert_snake_case_attrs(dir_watertight)
     dir_watertight_import_geometry = dir(watertight.import_geometry)
     dir_watertight_import_geometry.remove("_NextTask")
     _assert_snake_case_attrs(dir_watertight_import_geometry)
@@ -1399,3 +1401,53 @@ def test_object_oriented_task_inserting_in_workflows(new_mesh_session):
     time.sleep(1)
     assert watertight.import_boi_geometry.arguments()
     assert watertight.import_boi_geometry_1.arguments()
+
+
+@pytest.mark.codegen_required
+@pytest.mark.fluent_version(">=24.1")
+def test_loaded_workflow(new_mesh_session):
+    meshing = new_mesh_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.get_available_task_names()
+    )
+    time.sleep(2.5)
+    assert "import_boi_geometry" in loaded_workflow.get_available_task_names()
+    assert loaded_workflow.import_boi_geometry_1.arguments()
+
+
+@pytest.mark.codegen_required
+@pytest.mark.fluent_version(">=24.1")
+def test_created_workflow(new_mesh_session):
+    meshing = new_mesh_session
+    created_workflow = meshing.create_workflow()
+
+    assert sorted([repr(x) for x in created_workflow.first_tasks()]) == sorted(
+        [
+            "<Insertable 'import_geometry' task>",
+            "<Insertable 'load_cad_geometry' task>",
+            "<Insertable 'import_cad_and_part_management' task>",
+            "<Insertable 'custom_journal_task' task>",
+        ]
+    )
+
+    created_workflow.first_tasks.import_geometry.insert()
+
+    assert created_workflow.first_tasks() == []
+
+    time.sleep(2.5)
+
+    assert "<Insertable 'add_local_sizing' task>" in [
+        repr(x) for x in created_workflow.import_geometry.next_tasks()
+    ]
+    created_workflow.import_geometry.next_tasks.add_local_sizing.insert()
+    assert "<Insertable 'add_local_sizing' task>" not in [
+        repr(x) for x in created_workflow.import_geometry.next_tasks()
+    ]
+    assert sorted(created_workflow.get_available_task_names()) == sorted(
+        ["import_geometry", "add_local_sizing"]
+    )