connectivity#

Connect to Ansys Fluent and transfer files and cases between sessions.

This package provides the gRPC connection to a running Fluent process (FluentConnection), file transfer strategies for standalone, containerized, and remote (PyPIM) Fluent deployments, and helpers for transferring case/data files between solver sessions.

Classes:

`ContainerFileTransferStrategy`([image_name, ...])	Provides a file transfer service based on the gRPC client and gRPC server.
`FileTransferStrategy`(args, *kwargs)	Provides the file transfer strategy.
`FluentConnection`([ip, port, password, ...])	Encapsulates a Fluent connection.
`PimFileTransferService`([pim_instance])	Provides a file transfer service based on PyPIM and the `simple_upload_server()` method.
`RemoteFileTransferStrategy`([server_ip, ...])
`StandaloneFileTransferStrategy`([server_cwd])	Provides the local file transfer strategy can be used for Fluent launched in the standalone mode.

Exceptions:

PortNotProvided()

Raised when port is not provided.

Functions:

transfer_case(source_instance, solvers, ...)

Transfer case between instances.

class ansys.fluent.core.connectivity.ContainerFileTransferStrategy(image_name=None, image_tag=None, port=None, mount_target=None, mount_source=None, certs_dir=None)#

Bases: FileTransferStrategy

Provides a file transfer service based on the gRPC client and gRPC server.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import ContainerFileTransferStrategy
>>> case_file_name = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")
>>> solver_session = pyfluent.launch_fluent(file_transfer_service=ContainerFileTransferStrategy())
>>> solver_session.upload(file_name=case_file_name, remote_file_name="elbow.cas.h5")
>>> solver_session.settings.file.read_case(file_name="elbow.cas.h5")
>>> solver_session.settings.file.write_case(file_name="write_elbow.cas.h5")
>>> solver_session.download(file_name="write_elbow.cas.h5", local_directory="<local_directory_path>")

Methods:

`__init__`([image_name, image_tag, port, ...])	Provides the gRPC-based remote file transfer strategy.
`download`(file_name[, local_directory])	Download a file from the server.
`file_exists_on_remote`(file_name)	Check if remote file exists.
`upload`(file_name[, remote_file_name])	Upload a file to the server.

__init__(image_name=None, image_tag=None, port=None, mount_target=None, mount_source=None, certs_dir=None)#

Provides the gRPC-based remote file transfer strategy.

Parameters:

image_name: str: Name of the image.
image_tagstr, optional: Tag of the image.
port: int, optional: Port for the file transfer service to use.
mount_target: str | Path, optional: Path inside the container where mount_source will be mounted to.
mount_source: str | Path, optional: Existing path in the host operating system that will be mounted to mount_target.
certs_dirstr, optional: Directory containing the TLS certificates. The default is None.

download(file_name, local_directory=None)#

Download a file from the server.

Parameters:

file_namelist[str] | str: File name.
local_directorystr, optional: Local directory. The default is None.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import ContainerFileTransferStrategy
>>> case_file_name = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")
>>> solver_session = pyfluent.launch_fluent(file_transfer_service=ContainerFileTransferStrategy())
>>> solver_session.settings.file.write_case(file_name="write_elbow.cas.h5")
>>> solver_session.download(file_name="write_elbow.cas.h5", local_directory="<local_directory_path>")

file_exists_on_remote(file_name)#

Check if remote file exists.

Parameters:

file_name: str: File name.

Returns:

Whether file exists.

upload(file_name, remote_file_name=None)#

Upload a file to the server.

Parameters:

file_namelist[str] | str: File name.
remote_file_namestr, optional: Remote file name. The default is None.

Raises:

FileNotFoundError: If a file does not exist.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import ContainerFileTransferStrategy
>>> case_file_name = examples.download_file("mixing_elbow.cas.h5", "pyfluent/mixing_elbow")
>>> solver_session = pyfluent.launch_fluent(file_transfer_service=ContainerFileTransferStrategy())
>>> solver_session.upload(file_name=case_file_name, remote_file_name="elbow.cas.h5")
>>> solver_session.settings.file.read_case(file_name="elbow.cas.h5")

class ansys.fluent.core.connectivity.FileTransferStrategy(*args, **kwargs)#

Bases: Protocol

Provides the file transfer strategy.

Methods:

`__init__`(args, *kwargs)
`download`(file_name[, local_directory])	Download a file from the server.
`upload`(file_name[, remote_file_name])	Upload a file to the server.

__init__(*args, **kwargs)#

download(file_name, local_directory=None)#

Download a file from the server.

Parameters:

file_namestr: File name.
local_directorystr, optional: Local directory. The default is None.

upload(file_name, remote_file_name=None)#

Upload a file to the server.

Parameters:

file_namestr: File name.
remote_file_namestr, optional: Remote file name. The default is None.

class ansys.fluent.core.connectivity.FluentConnection(ip=None, port=None, password=None, address=None, channel=None, allow_remote_host=False, certificates_folder=None, insecure_mode=False, cleanup_on_exit=True, remote_instance=None, file_transfer_service=None, slurm_job_id=None, inside_container=None, container=None, compose_config=None)#

Bases: object

Encapsulates a Fluent connection.

Methods

exit()

Close the Fluent connection and exit Fluent.

Methods:

`__init__`([ip, port, password, address, ...])	Initialize a Session.
`create_grpc_service`(service, *args)	Create a gRPC service.
`exit`([timeout, timeout_force, wait])	Close the Fluent connection and exit Fluent.
`force_exit`()	Immediately terminates the Fluent client, losing unsaved progress and data.
`register_finalizer_cb`(cb[, at_start])	Register a callback to run with the finalizer.
`wait_process_finished`([wait])	Returns `True` if local Fluent processes have finished, `False` if they are still running when wait limit (default 100 seconds) is reached.

Attributes:

health_check

Provides access to Health Check service.

__init__(ip=None, port=None, password=None, address=None, channel=None, allow_remote_host=False, certificates_folder=None, insecure_mode=False, cleanup_on_exit=True, remote_instance=None, file_transfer_service=None, slurm_job_id=None, inside_container=None, container=None, compose_config=None)#

Initialize a Session.

Parameters:

ipstr, optional: IP address to connect to existing Fluent instance. Used only when channel is None. Defaults to "127.0.0.1" and can also be set by the environment variable PYFLUENT_FLUENT_IP=<ip>.
portint, optional: Port to connect to existing Fluent instance. Used only when channel is None. Defaults value can be set by the environment variable PYFLUENT_FLUENT_PORT=<port>.
passwordstr, optional: Password to connect to existing Fluent instance.
addressstr, optional: Address to connect to existing Fluent instance.
channelgrpc.Channel, optional: Grpc channel to use to connect to existing Fluent instance. ip and port arguments will be ignored when channel is specified.
cleanup_on_exitbool, optional: When True, the connected Fluent session will be shut down when PyFluent is exited or exit() is called on the session instance, by default True.
remote_instanceansys.platform.instancemanagement.Instance: The corresponding remote instance when Fluent is launched through PyPIM. This instance will be deleted when calling Session.exit().
file_transfer_serviceoptional: File transfer service for uploading files to and downloading files from the server.
slurm_job_id: bool, optional: Job ID of a Fluent session running within a Slurm environment.
inside_container: bool, optional: Whether the Fluent session that is being connected to is running inside a Docker container.
container: ContainerT, optional: The container instance if the Fluent session is running inside a container.
compose_config: ComposeConfig, optional: Configuration for Docker Compose or Podman Compose.
allow_remote_hostbool, optional: Whether to allow connecting to a remote Fluent instance.
certificates_folderstr, optional: Path to the folder containing TLS certificates for Fluent’s gRPC server.
insecure_modebool, optional: If True, Fluent’s gRPC server will be connected in insecure mode without TLS. This mode is not recommended. For more details on the implications and usage of insecure mode, refer to the Fluent documentation.

Raises:

PortNotProvided: If port is not provided.

create_grpc_service(service, *args)#

Create a gRPC service.

Parameters:

serviceAny: service class
argsAny, optional: additional arguments, by default empty

Returns:

Any: service object

exit(timeout=None, timeout_force=True, wait=False)#

Close the Fluent connection and exit Fluent.

Parameters:

timeoutfloat, optional: Time in seconds before considering that the exit request has timed out. If omitted or specified as None, then the request will not time out and will lock up the interpreter while waiting for a response. Will return earlier if request succeeds earlier.
timeout_forcebool, optional: If not specified, defaults to True. If True, attempts to terminate the Fluent process if exit request reached timeout. If no timeout is set, this option is ignored. Executes force_exit() or force_exit_container(), depending on how Fluent was launched.
waitfloat, int or bool, optional: Specifies whether to wait for local Fluent processes to finish completely before proceeding. If omitted or specified as False, will proceed as usual without waiting for the Fluent processes to finish. Can be set to True which will wait for up to 100 seconds, or set to a float or int value to specify the wait limit. If wait limit is reached, will forcefully terminate the Fluent process. If set to wait, will return as soon as processes completely finish. Does not work for remote Fluent processes.

Notes

Can also set the PYFLUENT_TIMEOUT_FORCE_EXIT environment variable to specify the number of seconds and alter the default timeout value. Setting this env var to a non-number value, such as OFF, will return this function to default behavior. Note that the environment variable will be ignored if timeout is specified when calling this function.

Examples

>>> import ansys.fluent.core as pyfluent
>>> session = pyfluent.launch_fluent()
>>> session.exit()

force_exit()#

Immediately terminates the Fluent client, losing unsaved progress and data.

Raises:

subprocess.CalledProcessError: If the cleanup script fails to execute.

Notes

If the Fluent session is responsive, prefer using exit() instead.

Examples

>>> import ansys.fluent.core as pyfluent
>>> session = pyfluent.launch_fluent()
>>> session.force_exit()

property health_check#: Provides access to Health Check service.

Deprecated since version 0.32: No longer required at this level.

register_finalizer_cb(cb, at_start=False)#: Register a callback to run with the finalizer.

wait_process_finished(wait=100)#

Returns True if local Fluent processes have finished, False if they are still running when wait limit (default 100 seconds) is reached. Immediately cancels and returns None if wait is set to False.

Parameters:

waitfloat, int or bool, optional: How long to wait for processes to finish before returning, by default 100 seconds. Can also be set to True, which will result in waiting indefinitely.

Raises:

UnsupportedRemoteFluentInstance: If current Fluent instance is running remotely.
WaitTypeError: If wait is specified improperly.

class ansys.fluent.core.connectivity.PimFileTransferService(pim_instance=None)#

Bases: object

Provides a file transfer service based on PyPIM and the simple_upload_server() method.

Attributes:

pim_instance: PIM instance: Instance of PIM which supports upload server services.
file_service: Client instance: Instance of Client which supports upload and download methods.

Methods

upload(	file_name, remote_file_name ) Upload a file to the server.
download(	file_name, local_directory ) Download a file from the server.

Methods:

`__init__`([pim_instance])	Initialize PimFileTransferService.
`download`(file_name[, local_directory])	Download a file from the server.
`download_file`(file_name[, local_directory])	Download a file from the server supported by PyPIM<https://pypim.docs.pyansys.com/version/stable/>.
`file_exists_on_remote`(file_name)	Check if remote file exists.
`is_configured`()	Check pypim configuration.
`upload`(file_name[, remote_file_name])	Upload a file to the server.
`upload_file`(file_name[, remote_file_name])	Upload a file to the server supported by PyPIM<https://pypim.docs.pyansys.com/version/stable/>.

__init__(pim_instance=None)#

Initialize PimFileTransferService.

Parameters:

pim_instance: Any, optional: PIM instance.

download(file_name, local_directory=None)#

Download a file from the server.

Parameters:

file_namelist[str] | str: File name.
local_directorystr, optional: Local directory. The default is the current working directory.

download_file(file_name, local_directory=None)#

Download a file from the server supported by PyPIM<https://pypim.docs.pyansys.com/version/stable/>.

Parameters:

file_namestr: File name.
local_directorystr, optional: local directory, by default None

Raises:

FileNotFoundError: If the remote file does not exist.
PyPIMConfigurationError: If PyPIM is not configured.

file_exists_on_remote(file_name)#

Check if remote file exists.

Parameters:

file_name: str: File name.

Returns:

Whether file exists.

is_configured()#: Check pypim configuration.

upload(file_name, remote_file_name=None)#

Upload a file to the server.

Parameters:

file_namelist[str] | str: File name.
remote_file_namestr, optional: Remote file name. The default is None.

Raises:

FileNotFoundError: If a file does not exist.

upload_file(file_name, remote_file_name=None)#

Upload a file to the server supported by PyPIM<https://pypim.docs.pyansys.com/version/stable/>.

Parameters:

file_namestr: File name.
remote_file_namestr, optional: Remote file name. The default is None.

Raises:

FileNotFoundError: If the file does not exist.
PyPIMConfigurationError: If PyPIM is not configured.

exception ansys.fluent.core.connectivity.PortNotProvided#

Bases: ValueError

Raised when port is not provided.

Methods:

__init__()

Initialize PortNotProvided.

__init__()#: Initialize PortNotProvided.

class ansys.fluent.core.connectivity.RemoteFileTransferStrategy(server_ip=None, server_port=None, certs_dir=None)#

Bases: FileTransferStrategy

Provides a file transfer service based on the gRPC client and gRPC server.

Methods:

`__init__`([server_ip, server_port, certs_dir])	Provides the gRPC-based remote file transfer strategy.
`download`(file_name[, local_directory])	Download a file from the server.
`upload`(file_name[, remote_file_name])	Upload a file to the server.

__init__(server_ip=None, server_port=None, certs_dir=None)#

Provides the gRPC-based remote file transfer strategy.

Parameters:

server_ip: str: IP address of the server.
server_portint: Port of the server.
certs_dirstr, optional: Directory containing the TLS certificates. The default is None.

download(file_name, local_directory=None)#

Download a file from the server.

Parameters:

file_namelist[str] | str: File name.
local_directorystr, optional: Local directory. The default is None.

upload(file_name, remote_file_name=None)#

Upload a file to the server.

Parameters:

file_namelist[str] | str: File name.
remote_file_namestr, optional: Remote file name. The default is None.

Raises:

FileNotFoundError: If a file does not exist.

class ansys.fluent.core.connectivity.StandaloneFileTransferStrategy(server_cwd=None)#

Bases: FileTransferStrategy

Provides the local file transfer strategy can be used for Fluent launched in the standalone mode.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import StandaloneFileTransferStrategy
>>> mesh_file_name = examples.download_file("mixing_elbow.msh.h5", "pyfluent/mixing_elbow")
>>> meshing_session = pyfluent.launch_fluent(mode=pyfluent.FluentMode.MESHING, file_transfer_service=StandaloneFileTransferStrategy())
>>> meshing_session.upload(file_name=mesh_file_name, remote_file_name="elbow.msh.h5")
>>> meshing_session.meshing.File.ReadMesh(FileName="elbow.msh.h5")
>>> meshing_session.meshing.File.WriteMesh(FileName="write_elbow.msh.h5")
>>> meshing_session.download(file_name="write_elbow.msh.h5", local_directory="<local_directory_path>")

Methods:

`__init__`([server_cwd])	Local File Transfer Service.
`download`(file_name[, local_directory])	Download a file from the server.
`file_exists_on_remote`(file_name)	Check if remote file exists.
`upload`(file_name[, remote_file_name])	Upload a file to the server.

__init__(server_cwd=None)#

Local File Transfer Service.

Parameters:

server_cwd: str, Optional: Current working directory of server/Fluent.

download(file_name, local_directory=None)#

Download a file from the server.

Parameters:

file_namelist[str] | str: File name.
local_directorystr, optional: Local directory. The default is None.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import StandaloneFileTransferStrategy
>>> mesh_file_name = examples.download_file("mixing_elbow.msh.h5", "pyfluent/mixing_elbow")
>>> meshing_session = pyfluent.launch_fluent(mode=pyfluent.FluentMode.MESHING, file_transfer_service=StandaloneFileTransferStrategy())
>>> meshing_session.meshing.File.WriteMesh(FileName="write_elbow.msh.h5")
>>> meshing_session.download(file_name="write_elbow.msh.h5", local_directory="<local_directory_path>")

file_exists_on_remote(file_name)#

Check if remote file exists.

Parameters:

file_name: str: File name.

Returns:

Whether file exists.

upload(file_name, remote_file_name=None)#

Upload a file to the server.

Parameters:

file_namelist[str] | str: File name.
remote_file_namestr, optional: Remote file name. The default is None.

Raises:

FileNotFoundError: If a file does not exist.

Examples

>>> import ansys.fluent.core as pyfluent
>>> from ansys.fluent.core import examples
>>> from ansys.fluent.core.connectivity.file_transfer_service import StandaloneFileTransferStrategy
>>> mesh_file_name = examples.download_file("mixing_elbow.msh.h5", "pyfluent/mixing_elbow")
>>> meshing_session = pyfluent.launch_fluent(mode=pyfluent.FluentMode.MESHING, file_transfer_service=StandaloneFileTransferStrategy())
>>> meshing_session.upload(file_name=mesh_file_name, remote_file_name="elbow.msh.h5")
>>> meshing_session.meshing.File.ReadMesh(FileName="elbow.msh.h5")

ansys.fluent.core.connectivity.transfer_case(source_instance, solvers, file_type, file_name_stem, num_files_to_try, clean_up_temp_file, overwrite_previous, workdir=None, container_workdir=None)#

Transfer case between instances.

Parameters:

source_instanceobject: Fluent instance (tested for meshing instance)
solversiterable: Sequence of solver instances
file_typestr: “case” or “mesh”
file_name_stemstr: Optional file name stem
num_files_to_tryint: Optional number of files to try to write, each with a different generated name. Defaults to 1
clean_up_temp_file: bool: Whether to remove the file at the end
overwrite_previous: bool: Whether to overwrite the file if it already exists
workdirstr, optional: Working directory that is accessible by the Fluent client as well as PyFluent.
container_workdirstr, optional: If using a Fluent container image, specifies the working directory that is accessible by the Fluent client inside the container, which should also be mounted to the container from the host system path specified in workdir.

Returns:

None

Raises:

MeshWriteError: If mesh cannot be written from source_instance.