workflow#

Workflow module that wraps and extends the core functionality.

This module provides a high-level, Pythonic interface for working with Ansys Fluent workflows. It wraps the underlying datamodel service layer to provide intuitive navigation, task management, and workflow operations.

The main classes are:

Workflow: Top-level workflow container that manages tasks and provides navigation between them.
TaskObject: Individual task wrapper that provides access to task properties, arguments, execution, and navigation to sibling/child tasks.

Notes#

This module is designed for Fluent 26R1 and later versions. Some features may not be available in earlier versions.

The workflow system provides both imperative and declarative approaches to building simulation workflows, with automatic dependency management and validation.

Classes:

`TaskObject`(task_object, base_name, workflow, ...)	Wrapper for individual workflow task objects.
`Workflow`(workflow, command_source, ...)	High-level workflow container that manages tasks and provides navigation.

Functions:

`build_specific_interface`(task_object)	Build a dynamic interface type that exposes task-specific commands/properties while delegating back to the task_object.
`command_name_to_task_name`(meshing_root, ...)	Convert a command name to its corresponding task display name.
`is_compound_child`(task_type)	Check if a task type represents a compound child task.
`make_task_wrapper`(task_obj, name, workflow, ...)	Wraps TaskObjects.

class ansys.fluent.core.workflow_new.TaskObject(task_object, base_name, workflow, parent, meshing_root)#

Bases: object

Wrapper for individual workflow task objects.

TaskObject provides a high-level interface for interacting with individual tasks in a workflow. It exposes task properties, arguments, execution methods, and navigation capabilities.

Methods:

`__init__`(task_object, base_name, workflow, ...)	Initialize a TaskObject wrapper.
`children`()	Get ordered list of direct child tasks.
`delete`()	Deletes the task item on which it is called.
`first_child`()	Get the first child task of this task.
`has_next`()	Check if there is a next sibling task.
`has_parent`()	Check if this task has a parent container.
`has_previous`()	Check if there is a previous sibling task.
`last_child`()	Get the last child task of this task.
`mark_as_updated`()	Mark tasks in workflow as updated.
`next`()	Returns the next sibling task item.
`parent`()	Get the parent container of this task.
`previous`()	Returns the previous sibling task item.

Attributes:

insertable_tasks

Get interface for inserting tasks after this one.

__init__(task_object, base_name, workflow, parent, meshing_root)#

Initialize a TaskObject wrapper.

Parameters:

task_objectPyMenu: The underlying datamodel task object.
base_namestr: Python-friendly base name for the task.
workflowPyMenu: Reference to the parent workflow datamodel.
parentUnion[Workflow, TaskObject]: Parent container (Workflow or parent TaskObject).

Notes

This constructor is called internally by make_task_wrapper(). Users should not instantiate TaskObject directly.

children()#

Get ordered list of direct child tasks.

Returns:

List[TaskObject]: Ordered list of child task wrappers, or empty list if no children.

delete()#: Deletes the task item on which it is called.

first_child()#

Get the first child task of this task.

Returns:

TaskObject or None: The first child task, or None if no children exist.

Examples

>>> parent = '<workflow>'.describe_geometry
>>> first = parent.first_child()
>>> if first:
...     print(f"First child: {first.name()}")

Navigate through children:

>>> current = parent.first_child()
>>> while current:
...     print(current.name())
...     if current.has_next():
...         current = current.next()
...     else:
...         break

has_next()#

Check if there is a next sibling task.

Determines whether this task has a sibling task that follows it in the workflow sequence at the same level.

Returns:

bool: True if a next sibling exists, False if this is the last task.

has_parent()#

Check if this task has a parent container.

Returns:

bool: True if task has a parent (Workflow or TaskObject), False otherwise.

has_previous()#

Check if there is a previous sibling task.

Determines whether this task has a sibling task that precedes it in the workflow sequence at the same level.

Returns:

bool: True if a previous sibling exists, False if this is the first task.

property insertable_tasks#

Get interface for inserting tasks after this one.

Returns a dynamic object that exposes all valid task types that can be inserted after the current task. Each insertable task is accessible as an attribute with an insert() method.

Returns:

_NextTask: Object with attributes for each insertable task type.

Examples

Basic usage:

>>> task = '<workflow>'.import_geometry
>>>
>>> # See what's available
>>> available = task.insertable_tasks()
>>> for insertable in available:
...     print(insertable)
<Insertable 'import_boi_geometry' task>
<Insertable 'set_up_rotational_periodic_boundaries' task>
<Insertable 'create_local_refinement_regions' task>
<Insertable 'custom_journal_task' task>

Insert specific task:

>>> # Insert by accessing as attribute
>>> task.insertable_tasks.import_boi_geometry.insert()

Access specific task after insertion:

>>> # Access task as attribute
>>> '<workflow>'.import_boi_geometry

last_child()#

Get the last child task of this task.

Returns:

TaskObject or None: The last child task, or None if no children exist.

Examples

>>> parent = '<workflow>'.describe_geometry
>>> last = parent.last_child()
>>> if last:
...     print(f"Last child: {last.name()}")

mark_as_updated()#: Mark tasks in workflow as updated.

next()#: Returns the next sibling task item.

parent()#

Get the parent container of this task.

Returns:

Union[Workflow, TaskObject]: The parent container. Can be: - Workflow instance for top-level tasks - TaskObject instance for nested child tasks

previous()#: Returns the previous sibling task item.

class ansys.fluent.core.workflow_new.Workflow(workflow, command_source, fluent_version)#

Bases: object

High-level workflow container that manages tasks and provides navigation.

The Workflow class wraps the underlying datamodel workflow object and provides a Pythonic interface for:

Discovering and accessing tasks
Creating, loading, and saving workflows
Navigating task hierarchies
Managing task lifecycles (creation/deletion)

Classes:

FirstTask(workflow)

Helper exposing insertable tasks for an empty workflow.

Methods:

`__init__`(workflow, command_source, ...)	Initialize Workflow.
`children`()	Get the top-level tasks in the workflow in display order.
`delete_tasks`(list_of_tasks)	Delete multiple tasks from the workflow.
`first_child`()	Get the first top-level task in the workflow.
`last_child`()	Get the last top-level task in the workflow.
`load_state`(list_of_roots)	Load the state of the workflow.
`save_workflow`(file_path)	Save the current workflow to a file.
`task_names`()	Get Python-friendly names for all available tasks.
`tasks`()	Get the complete list of tasks in the workflow.

Attributes:

insertable_tasks

Tasks that can be inserted into an empty workflow.

class FirstTask(workflow)#

Bases: object

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.

Methods:

__init__(workflow)

Initialize a FirstTask instance.

__init__(workflow)#: Initialize a FirstTask instance.

__init__(workflow, command_source, fluent_version)#: Initialize Workflow.

children()#

Get the top-level tasks in the workflow in display order.

Returns an ordered list of the workflow’s main tasks (those directly under the workflow root, not nested child tasks). The order reflects the execution sequence in the workflow.

Returns:

List[TaskObject]: Ordered list of top-level task wrappers.

delete_tasks(list_of_tasks)#

Delete multiple tasks from the workflow.

Removes the specified tasks from the workflow. Tasks are identified by TaskObject instances.

Parameters:

list_of_tasks: list[TaskObject]: List of task objects to delete.

Raises:

TypeError: If list contains items that are neither TaskObject nor str.

first_child()#

Get the first top-level task in the workflow.

Returns:

TaskObject or None: The first task in the workflow, or None if the workflow is empty.

Notes

Returns None for empty workflows. Always check before accessing properties.

Examples

>>> first = '<workflow>'.first_child()
>>> if first:
...     print(f"Starting task: {first.name()}")
...     first()  # Execute it

>>> # Navigate from first to last
>>> current = '<workflow>'.first_child()
>>> while current and current.has_next():
...     print(current.name())
...     current()  # Execute it
...     current = current.next()

property insertable_tasks: 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).

last_child()#

Get the last top-level task in the workflow.

Returns:

TaskObject or None: The last task in the workflow, or None if the workflow is empty.

Examples

>>> last = '<workflow>'.last_child()
>>> if last:
...     print(f"Final task: {last.name()}")
...     last()  # Execute it

>>> # Execute workflow in reverse
>>> current = '<workflow>'.last_child()
>>> while current and current.has_previous():
...     print(current.name())
...     current()  # Execute it
...     current = current.previous()

load_state(list_of_roots)#: Load the state of the workflow.

save_workflow(file_path)#: Save the current workflow to a file.

task_names()#

Get Python-friendly names for all available tasks.

Returns the list of task names as they would be accessed via Python attribute syntax (e.g., “import_geometry” for “Import Geometry”).

tasks()#

Get the complete list of tasks in the workflow.

This method builds and returns a comprehensive list of all task objects currently present in the workflow, including:

Top-level tasks
Compound child tasks (tasks with multiple instances)
Dynamically created tasks

The method rebuilds its internal task cache on each call to ensure freshness, though this can be expensive for large workflows.

ansys.fluent.core.workflow_new.build_specific_interface(task_object)#: Build a dynamic interface type that exposes task-specific commands/properties while delegating back to the task_object.

ansys.fluent.core.workflow_new.command_name_to_task_name(meshing_root, command_name)#

Convert a command name to its corresponding task display name.

This function maps internal command names (used by the Fluent core) to user-facing task names.

Parameters:

meshing_rootPyMenu: The root meshing datamodel object.
command_namestr: Internal command name (e.g., “ImportGeometry”).

Returns:

str: User-facing task name (e.g., “import_geometry”).

Notes

This is a workaround for Fluent 26R1.

ansys.fluent.core.workflow_new.is_compound_child(task_type)#

Check if a task type represents a compound child task. This encapsulates a string comparison to avoid repetition.

Parameters:

task_typestr: The task type string to check.

Returns:

bool: True if the task type is “Compound Child”, False otherwise.

ansys.fluent.core.workflow_new.make_task_wrapper(task_obj, name, workflow, parent, meshing_root)#: Wraps TaskObjects.