PyFluentBackend#
- class ansys.fluent.mcp.solve.backends.pyfluent.PyFluentBackend(*, label: str = 'PyFluent (solve)')#
Bases:
ansys.fluent.mcp.common.backend.BackendPersistent PyFluent Solver session.
Overview#
Connect to the configured backend or service. |
|
Close resources for the PyFluentBackend object. |
|
Return whether connected. |
|
Close backend resources from synchronous code. |
|
Discover every named-object collection. |
|
Return instance names for a single named-object collection using |
|
Return the state. |
|
Return the active status. |
|
Return the allowed values. |
|
Batched per-node settings-attribute fetch. |
|
Recursively fetch attributes for all children of |
|
Cheap pre-flight: |
|
Return |
|
Introspect a Command’s keyword-argument signature via PyFluent. |
|
Describe the field shape of a fresh child of a NamedObject collection. |
|
Enumerate solver field/variable names available for reports & post. |
|
Return the targeted context. |
|
Execute Python code through the backend runtime. |
|
Return |
|
Probe live mesh element totals via Fluent Scheme. |
|
Return mesh quality information from the backend. |
|
Run the backend mesh check operation. |
|
Best-effort summary of where the solver currently is. |
|
Validate code. |
|
Capture a screenshot from the backend runtime. |
Import detail#
from ansys.fluent.mcp.solve.backends.pyfluent import PyFluentBackend
Attribute detail#
- PyFluentBackend.kind = 'pyfluent'#
- PyFluentBackend.label = 'PyFluent (solve)'#
Method detail#
- async PyFluentBackend.connect(*, ip: str | None = None, port: int | None = None, password: str | None = None, server_info_file: str | None = None, precision: str = 'double', processor_count: int = 1, ui_mode: str = 'gui', product_version: str | None = None, dimension: int | str | None = None, mode: str | None = None, gpu: bool | list[int] | None = None, journal_file_names: str | list[str] | None = None, case_file_name: str | None = None, case_data_file_name: str | None = None, cwd: str | None = None, fluent_path: str | None = None, env: dict[str, Any] | None = None, graphics_driver: str | None = None, scheduler_options: dict[str, Any] | None = None, start_timeout: int | None = None, cleanup_on_exit: bool | None = None, additional_arguments: str | None = None, **_: Any) ansys.fluent.mcp.common.models.ConnectResult#
Connect to the configured backend or service.
- Parameters:
- ip
Optional[str] IP address of the Fluent server to connect to.
- port
Optional[int] Port number of the Fluent server to connect to.
- password
Optional[str] Password used when connecting to a Fluent server.
- server_info_file
Optional[str] Server-info file used to connect to an existing Fluent session.
- precision
str Solver precision requested for the Fluent session.
- processor_count
int Number of processors requested for the Fluent session.
- ui_mode
str Fluent UI mode requested for launch.
- product_version
Optional[str] Fluent product version requested for launch.
- dimension
Optional[int|str] Solver dimension requested for the Fluent session.
- mode
Optional[str] Execution or launch mode requested by the caller.
- gpu
Optional[bool |list[int]] GPU option forwarded to the Fluent launch API.
- journal_file_names
Optional[str|list[str]] Journal files passed to Fluent during startup.
- case_file_name
Optional[str] Case file path passed to Fluent at startup.
- case_data_file_name
Optional[str] Case-data file path passed to Fluent at startup.
- cwd
Optional[str] Working directory used when launching Fluent.
- fluent_path
Optional[str] Path for the fluent.
- env
Optional[dict[str,Any]] Environment mapping to read instead of the process environment.
- graphics_driver
Optional[str] Graphics driver requested for the Fluent session.
- scheduler_options
Optional[dict[str,Any]] Scheduler options forwarded to the Fluent launch API.
- start_timeout
Optional[int] Timeout in seconds used while starting Fluent.
- cleanup_on_exit
Optional[bool] Whether Fluent resources should be cleaned up when the process exits.
- additional_arguments
Optional[str] Additional launch arguments forwarded to Fluent.
- _
Any Ignored compatibility options accepted by the backend interface.
- ip
- Returns:
ConnectResultConnectResult produced by the operation.
- async PyFluentBackend.disconnect() None#
Close resources for the PyFluentBackend object.
- Returns:
NoneThe function completes through its side effects.
- PyFluentBackend.is_connected() bool#
Return whether connected.
- Returns:
- bool
Boolean result of the operation.
- PyFluentBackend.close_sync() None#
Close backend resources from synchronous code.
- Returns:
NoneThe function completes through its side effects.
- async PyFluentBackend.list_named_objects() dict[str, Any]#
Discover every named-object collection.
Strategy:
Try the Scheme
api-get-named-object-names(one call, covers every collection Fluent knows about).Fall back to a generic depth-limited tree walk.
No hardcoded path list.
- async PyFluentBackend.get_named_object_names(collection_path: str) list[str]#
Return instance names for a single named-object collection using
get_object_names().A single cheap Scheme call that does not need to read the full collection state.
Falls back to the full
list_named_objects()snapshot if the collection node does not exposeget_object_names.
- async PyFluentBackend.get_active_status(paths: list[str]) dict[str, bool]#
Return the active status.
- async PyFluentBackend.get_allowed_values(paths: list[str]) dict[str, list[Any]]#
Return the allowed values.
- async PyFluentBackend.get_node_attrs(paths: list[str], attrs: list[str]) dict[str, dict[str, Any]]#
Batched per-node settings-attribute fetch.
For each
pathresolves the settings node and asks Fluent for ALL requestedattrsin one Scheme RPC via PyFluent’snode.get_attrs([...]). Returns{path: {attr: value}}.attrsuse the Scheme spelling ("active?","read-only?","user-creatable?","allowed-values","min","max","default","units-quantity","file-purpose"). Attrs the node does not expose are simply omitted from its inner dict — Fluent never raises for missing attrs in a batch request.Unresolvable paths and nodes that do not implement
get_attrsmap to{}. EmptypathsraisesInvalidArgumentsError; emptyattrsraises the same.This is the foundation for enrichment passes such as
describe_named_object_templatethat would otherwise issue one RPC per attribute.
- async PyFluentBackend.get_node_attrs_bulk(parent_path: str, attrs: list[str]) dict[str, dict[str, Any]]#
Recursively fetch attributes for all children of
parent_path.Calls
node.get_attrs(attrs, True), a single Scheme RPC — returning{relative_child_path: {attr: value}}for every descendant that exposes those attributes. This replaces N per-fieldget_node_attrsround-trips when the caller needs metadata for a whole subtree (such as validating every leaf in aset_namedvalue dict before apply).Falls back to
{}gracefully when the node is unreachable or does not support the recursive form.
- async PyFluentBackend.probe_path(paths: list[str]) dict[str, dict[str, Any]]#
Cheap pre-flight:
{path: {exists, is_active, is_user_creatable, kind}}.Designed to be called once before a batch of mutating writes so the executor can skip steps that would fail anyway (path missing, settings node currently inactive, NamedObject family that does not accept
create). Each path costs ONE Scheme RPC via the leaf’s batchedget_attrs([active?, user-creatable?])plus a local class-name lookup.kindis one of"missing","command","query","named_object","group","leaf". Unknown / unusual nodes default to"leaf".
- async PyFluentBackend.mesh_adjacency_probe(cellzones: list[str], *, bc_filter: tuple[str, Ellipsis] | None = None) dict[str, list[str]]#
Return
{cellzone -> [adjacent_face_zone_names]}.Read-only. Implementation walks every BC family that exposes
adjacent_cell_zone(wall, velocity_inlet, pressure_outlet, mass_flow_inlet, …) and inverts the mapping. Coupled-wallshadow_face_zoneentries are added to BOTH sides so the cellzone-↔-cellzone neighbour query (set-intersection on shared face names) finds CHT solid-fluid pairs.IMPORTANT — PyFluent Query objects: both
adjacent_cell_zoneandshadow_face_zoneare PyFluent Query objects (not Parameters). They are never included inget_state()output and must be read by calling them:instance.adjacent_cell_zone(). The implementation enumerates instance names via a batchedget_state()(cheap), then calls the Query per-instance.bc_filter(optional) restricts the walk to the listed BC families (e.g.("wall",)to answer “walls adjacent to X”;("velocity_inlet","pressure_inlet")for inlets). Unknown families are silently dropped. The coupled-wall shadow pass runs only when"wall"is in scope (or the filter is None).Limitation: interior face zones are NOT enumerated — they are gated INACTIVE on
boundary_conditions.interior[*]and themesh.adjacencyCommand’s argument-binding side-channel is broken on Fluent ≥ 27.1 (cellzones.set_stateraisesattr value not defined). Callers that need interior-face coverage must supplement withrun_code.
- async PyFluentBackend.get_command_arguments(path: str) dict[str, Any] | None#
Introspect a Command’s keyword-argument signature via PyFluent.
- async PyFluentBackend.describe_named_object_template(path: str) dict[str, Any] | None#
Describe the field shape of a fresh child of a NamedObject collection.
Uses PyFluent’s static settings classes (
child_object_type) so we never have to create-and-delete a transient object on the live session. Returns a flat{field_name: {type_hint, allowed_values?, child_kind}}map plus the create-command signature when present.
- async PyFluentBackend.list_fields(*, scope: str = 'any') dict[str, Any] | None#
Enumerate solver field/variable names available for reports & post.
scopeis a hint passed to Fluent’s field-info API:"any"(default),"surface","cell","node". Returns{"fields": [...], "scope": scope}orNonewhen unavailable.
- async PyFluentBackend.get_targeted_context(*, paths_to_check: list[str], named_object_types: list[str] | None = None, instance_state_fetch: list[str] | None = None) dict[str, Any]#
Return the targeted context.
- Parameters:
- Returns:
-
async PyFluentBackend.run_code(code: str, *, namespace: dict[str, Any] | None = None, filename: str = '
' ) ansys.fluent.mcp.common.models.RunCodeResult# Execute Python code through the backend runtime.
- async PyFluentBackend.get_help(path: str) dict[str, Any]#
Return
{path, doc, kind, child_names, allowed_values}.
- async PyFluentBackend.mesh_counts() dict[str, int | None]#
Probe live mesh element totals via Fluent Scheme.
Returns
{"cell_count", "face_count", "node_count"}; each value isintwhen the Scheme call resolved orNonewhen unavailable (no mesh loaded, partition pending, Scheme call failed). Never raises — surfaces failures as an all-Nonepayload so callers can branch onis Nonewithout try/except.Why Scheme: Fluent does not surface mesh-element totals on the settings tree. The canonical accessor is
(inquire-grids)— it returns a list whose LAST element is the global grid summary in the shape(.cells faces nodes ...) cadr/caddr/cadddrextract the three counts. This matches the pattern Fluent itself uses internally (e.g. inwb-is-data-file-compatible?for cross-file count checks) and is stable across versions back to 19.0.rpgetvar 'tinfo/n-cellsdoes NOT exist as an rpvar.
- async PyFluentBackend.mesh_quality() dict[str, float | None]#
Return mesh quality information from the backend.
- async PyFluentBackend.solver_status() dict[str, Any]#
Best-effort summary of where the solver currently is.
- async PyFluentBackend.validate_code(code: str) ansys.fluent.mcp.common.models.RunCodeResult#
Validate code.
- Parameters:
- code
str Python code or command text to execute or validate.
- code
- Returns:
RunCodeResultRunCodeResult produced by the operation.