diff --git a/doc/changelog.d/4779.added.md b/doc/changelog.d/4779.added.md new file mode 100644 index 00000000000..5fa2e442961 --- /dev/null +++ b/doc/changelog.d/4779.added.md @@ -0,0 +1 @@ +Update server based enhanced meshing workflow. diff --git a/doc/source/user_guide/meshing/new_meshing_workflows.rst b/doc/source/user_guide/meshing/new_meshing_workflows.rst index ba7d2af7d30..65c6a8d815d 100644 --- a/doc/source/user_guide/meshing/new_meshing_workflows.rst +++ b/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 diff --git a/src/ansys/fluent/core/meshing/meshing_workflow_new.py b/src/ansys/fluent/core/meshing/meshing_workflow_new.py index bd2fcf31c42..60599b4f1ab 100644 --- a/src/ansys/fluent/core/meshing/meshing_workflow_new.py +++ b/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 diff --git a/src/ansys/fluent/core/workflow_new.py b/src/ansys/fluent/core/workflow_new.py index 69b3bf6119b..102de7f0a88 100644 --- a/src/ansys/fluent/core/workflow_new.py +++ b/src/ansys/fluent/core/workflow_new.py @@ -168,8 +168,9 @@ def command_name_to_task_name(meshing_root, command_name: str) -> str: ----- This is a workaround for Fluent 26R1. """ - # TODO: This is a fix only for 26R1 as the server lacks the mechanism to return mapped values - # for '.get_next_possible_tasks()'. + # TODO: This fix is applicable till the server lacks the mechanism to return mapped values + # for '.get_next_possible_tasks()' and + # for '.get_new_insertable_tasks()'. command_instance = getattr(meshing_root, command_name).create_instance() return command_instance.get_attr("APIName") or command_instance.get_attr( "helpString" @@ -509,8 +510,100 @@ 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) -> FirstTask: + """Tasks that can be inserted into an empty workflow. + + Returns a helper that exposes the set of valid starting tasks for a blank + workflow as attributes. Each attribute is an object with an `insert()` + method that inserts that task into the workflow. + + Notes + ----- + - This helper only populates insertable tasks when the workflow is empty. + - Task names are exposed using Python-friendly identifiers (snake_case). + """ + return self.FirstTask(self) + + class FirstTask: + """Helper exposing insertable tasks for an empty workflow. + + This container dynamically creates attributes for each command that the + server allows as the first task in a new workflow. + + Access an attribute and call `.insert()` to insert that task. + """ + + def __init__(self, workflow): + """Initialize a ``FirstTask`` instance.""" + self._workflow = workflow + self._insertable_tasks: list = [] + # Map: server command name -> python-friendly task name + self._initial_task_map: dict[str, str] = {} + + # Query server for commands that can start a new workflow. + # Older Fluent versions don’t provide this API; use a fallback list. + try: + initial_tasks = self._workflow.general.get_insertable_tasks() + except AttributeError: + # For Fluent Version 26R1 or before. + initial_tasks = ["ImportGeometry", "PartManagement", "RunCustomJournal"] + for command in initial_tasks: + self._initial_task_map[command] = command_name_to_task_name( + self._workflow._command_source, command + ) + # Only expose these attributes when the workflow is empty. + if self._workflow._workflow.general.workflow.task_list() == []: + for command_name, python_name in self._initial_task_map.items(): + # Build a lightweight proxy object with an insert() method. + insertable_task = type("Insert", (self._Insert,), {})( + self._workflow, + command_name, + self._initial_task_map, + ) + # Expose as attribute: e.g., .insertable_tasks.import_geometry.insert() + setattr(self, python_name, insertable_task) + self._insertable_tasks.append(insertable_task) + + def __call__(self) -> list: + """Return all insertable task proxies as a list.""" + return self._insertable_tasks + + class _Insert: + """Represents a single insertable starting task. + + Provides the `insert()` method to add this task to the workflow. + """ + + def __init__(self, workflow, name, task_map): + """Initialize an _Insert instance. + + Parameters + ---------- + workflow : Workflow + Target workflow. + name : str + Server command name (e.g., "ImportGeometry"). + task_map : dict[str, str] + Mapping from server command name -> python-friendly task name. + """ + self._workflow = workflow + self._name = name + self._task_map = task_map + + def insert(self) -> None: + """Insert this task into the workflow as the first task.""" + self._workflow.general.insert_new_task(command_name=self._name) + + def __repr__(self) -> str: + return f"" + 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: diff --git a/tests/test_server_meshing_workflow.py b/tests/test_server_meshing_workflow.py index 5cc0a4e8a2a..318d6995e9a 100644 --- a/tests/test_server_meshing_workflow.py +++ b/tests/test_server_meshing_workflow.py @@ -1660,3 +1660,45 @@ 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( + [ + "", + "", + "", + ] + ) + + created_workflow.insertable_tasks()[0].insert() + + assert created_workflow.insertable_tasks() == [] + + assert "" in [ + repr(x) for x in created_workflow.import_geometry.insertable_tasks() + ] + created_workflow.import_geometry.insertable_tasks.add_local_sizing.insert() + assert "" 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()