diff --git a/pyproject.toml b/pyproject.toml
index 432c1222bb4a..9e3cdfaf95ff 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -33,6 +33,7 @@ dependencies = [
"pandas>=1.1.0,<3.0.0",
"pyansys-tools-report>=0.8.1",
"pyyaml>=6.0",
+ "typing-extensions>=4.12"
]
dynamic = ["version"]
@@ -233,3 +234,8 @@ skips = [
"B604",
"B607",
]
+
+[tool.basedpyright]
+reportUnknownMemberType = false
+reportExplicitAny = false
+reportPrivateUsage = false
\ No newline at end of file
diff --git a/src/ansys/fluent/core/examples/downloads.py b/src/ansys/fluent/core/examples/downloads.py
index f41527e71ccd..b17c88f63e3e 100644
--- a/src/ansys/fluent/core/examples/downloads.py
+++ b/src/ansys/fluent/core/examples/downloads.py
@@ -27,6 +27,7 @@
from pathlib import Path
import re
import shutil
+from typing import TypeAlias
import zipfile
import ansys.fluent.core as pyfluent
@@ -71,10 +72,13 @@ def _get_file_url(file_name: str, directory: str | None = None) -> str:
return f"https://github.com/ansys/example-data/raw/main/{file_name}"
+PathType: TypeAlias = "os.PathLike[str | bytes] | str | bytes"
+
+
def _retrieve_file(
url: str,
file_name: str,
- save_path: str | None = None,
+ save_path: PathType | None = None,
return_without_path: bool | None = False,
) -> str:
"""Download specified file from specified URL."""
@@ -121,7 +125,7 @@ def _retrieve_file(
def download_file(
file_name: str,
directory: str | None = None,
- save_path: str | None = None,
+ save_path: PathType | None = None,
return_without_path: bool | None = None,
) -> str:
"""Download specified example file from the Ansys example data repository.
diff --git a/src/ansys/fluent/core/launcher/container_launcher.py b/src/ansys/fluent/core/launcher/container_launcher.py
index ad24c691c5a4..127966071183 100644
--- a/src/ansys/fluent/core/launcher/container_launcher.py
+++ b/src/ansys/fluent/core/launcher/container_launcher.py
@@ -35,11 +35,12 @@
>>> container_solver_session = container_solver_launcher()
"""
-import inspect
import logging
import os
import time
-from typing import Any
+from typing import Any, TypedDict
+
+from typing_extensions import Unpack
from ansys.fluent.core.fluent_connection import FluentConnection
from ansys.fluent.core.launcher.fluent_container import (
@@ -64,6 +65,37 @@
from ansys.fluent.core.session import _parse_server_info_file
from ansys.fluent.core.utils.fluent_version import FluentVersion
+
+class ContainerArgsWithoutDryRun(
+ TypedDict, total=False
+): # pylint: disable=missing-class-docstring
+ ui_mode: UIMode | str | None
+ graphics_driver: (
+ FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
+ )
+ product_version: FluentVersion | str | float | int | None
+ dimension: Dimension | int | None
+ precision: Precision | str | None
+ processor_count: int | None
+ start_timeout: int
+ additional_arguments: str
+ container_dict: dict[str, Any] | None
+ cleanup_on_exit: bool
+ start_transcript: bool
+ py: bool | None
+ gpu: bool | None
+ start_watchdog: bool | None
+ file_transfer_service: Any | None
+ use_docker_compose: bool | None
+ use_podman_compose: bool | None
+
+
+class ContainerArgs(
+ ContainerArgsWithoutDryRun, total=False
+): # pylint: disable=missing-class-docstring
+ dry_run: bool
+
+
_THIS_DIR = os.path.dirname(__file__)
_OPTIONS_FILE = os.path.join(_THIS_DIR, "fluent_launcher_options.json")
logger = logging.getLogger("pyfluent.launcher")
@@ -89,27 +121,8 @@ class DockerLauncher:
def __init__(
self,
- mode: FluentMode | str | None = None,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- container_dict: dict | None = None,
- dry_run: bool = False,
- cleanup_on_exit: bool = True,
- start_transcript: bool = True,
- py: bool | None = None,
- gpu: bool | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
- use_docker_compose: bool | None = None,
- use_podman_compose: bool | None = None,
+ mode: FluentMode | str,
+ **kwargs: Unpack[ContainerArgs],
):
"""
Launch a Fluent session in container mode.
@@ -183,20 +196,17 @@ def __init__(
In job scheduler environments (e.g., SLURM, LSF, PBS), resources and compute nodes are allocated,
and core counts are queried from these environments before being passed to Fluent.
"""
- locals_ = locals().copy()
- argvals = {
- arg: locals_.get(arg)
- for arg in inspect.getargvalues(inspect.currentframe()).args
- }
- self.argvals, self.new_session = _get_argvals_and_session(argvals)
- if self.argvals["start_timeout"] is None:
+ self.argvals, self.new_session = _get_argvals_and_session(
+ {**kwargs, mode: mode}
+ )
+ if self.argvals.get("start_timeout") is None:
self.argvals["start_timeout"] = 60
- self.file_transfer_service = file_transfer_service
+ self.file_transfer_service = kwargs.get("file_transfer_service")
if self.argvals["mode"] == FluentMode.SOLVER_ICING:
self.argvals["fluent_icing"] = True
- if self.argvals["container_dict"] is None:
+ if self.argvals.get("container_dict") is None:
self.argvals["container_dict"] = {}
- if self.argvals["product_version"]:
+ if "product_version" in self.argvals:
self.argvals["container_dict"][
"image_tag"
] = f"v{FluentVersion(self.argvals['product_version']).value}"
@@ -204,10 +214,12 @@ def __init__(
self._args = _build_fluent_launch_args_string(**self.argvals).split()
if FluentMode.is_meshing(self.argvals["mode"]):
self._args.append(" -meshing")
+
+ use_docker_compose = kwargs.get("use_docker_compose")
+ use_podman_compose = kwargs.get("use_podman_compose")
self._compose_config = ComposeConfig(use_docker_compose, use_podman_compose)
def __call__(self):
-
if self.argvals["dry_run"]:
config_dict, *_ = configure_container_dict(
self._args,
diff --git a/src/ansys/fluent/core/launcher/launcher.py b/src/ansys/fluent/core/launcher/launcher.py
index e7b7c8754ddf..6d4acd67c5a5 100644
--- a/src/ansys/fluent/core/launcher/launcher.py
+++ b/src/ansys/fluent/core/launcher/launcher.py
@@ -29,7 +29,9 @@
import inspect
import logging
import os
-from typing import Any, Dict
+from typing import Any, Literal, TypedDict, overload
+
+from typing_extensions import Required, Unpack
import ansys.fluent.core as pyfluent
from ansys.fluent.core.fluent_connection import FluentConnection
@@ -57,6 +59,7 @@
from ansys.fluent.core.session_meshing import Meshing
from ansys.fluent.core.session_pure_meshing import PureMeshing
from ansys.fluent.core.session_solver import Solver
+from ansys.fluent.core.session_solver_aero import SolverAero
from ansys.fluent.core.session_solver_icing import SolverIcing
from ansys.fluent.core.utils.deprecate import all_deprecators
from ansys.fluent.core.utils.fluent_version import FluentVersion
@@ -66,7 +69,7 @@
logger = logging.getLogger("pyfluent.launcher")
-def create_launcher(fluent_launch_mode: LaunchMode = None, **kwargs):
+def create_launcher(fluent_launch_mode: LaunchMode, **kwargs):
"""Use the factory function to create a launcher for supported launch modes.
Parameters
@@ -82,7 +85,7 @@ def create_launcher(fluent_launch_mode: LaunchMode = None, **kwargs):
Session launcher.
Raises
------
- DisallowedValuesError
+ ValueError
If an unknown Fluent launch mode is passed.
"""
if fluent_launch_mode == LaunchMode.STANDALONE:
@@ -93,6 +96,7 @@ def create_launcher(fluent_launch_mode: LaunchMode = None, **kwargs):
return PIMLauncher(**kwargs)
elif fluent_launch_mode == LaunchMode.SLURM:
return SlurmLauncher(**kwargs)
+ raise ValueError(f"launch mode invalid: {fluent_launch_mode!r}")
def _show_gui_to_ui_mode(old_arg_val, **kwds):
@@ -125,6 +129,111 @@ def _version_to_dimension(old_arg_val):
return None
+class LaunchFluentArgs(
+ TypedDict, total=False
+): # pylint: disable=missing-class-docstring
+ product_version: FluentVersion | str | float | int | None
+ dimension: Dimension | int
+ precision: Precision | str
+ processor_count: int | None
+ journal_file_names: None | str | list[str]
+ start_timeout: int
+ additional_arguments: str
+ env: dict[str, Any] | None
+ start_container: bool | None
+ container_dict: dict[str, Any] | None
+ cleanup_on_exit: bool
+ start_transcript: bool
+ ui_mode: UIMode | str | None
+ graphics_driver: (
+ FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
+ )
+ case_file_name: str | None
+ case_data_file_name: str | None
+ lightweight_mode: bool | None
+ py: bool | None
+ gpu: bool | list[int] | None
+ cwd: str | None
+ fluent_path: str | None
+ topy: str | list | None
+ start_watchdog: bool | None
+ file_transfer_service: Any | None
+ use_docker_compose: bool
+ use_podman_compose: bool
+
+
+class SlurmSchedulerOptions(
+ TypedDict, total=False
+): # pylint: disable=missing-class-docstring
+ scheduler: Required[Literal["slurm"]]
+ scheduler_headnode: str
+ scheduler_queue: str
+ scheduler_account: str
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ mode: Literal[FluentMode.MESHING, "meshing"],
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> Meshing: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ mode: Literal[FluentMode.PURE_MESHING, "pure_meshing"],
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> PureMeshing: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ mode: Literal[FluentMode.SOLVER, "solver"] = FluentMode.SOLVER,
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> Solver: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ mode: Literal[FluentMode.SOLVER_ICING, "solver_icing"],
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> SolverIcing: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ mode: Literal[FluentMode.SOLVER_AERO, "solver_aero"] = ...,
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> SolverAero: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[False] = False,
+ scheduler_options: SlurmSchedulerOptions,
+ mode: FluentMode | str = FluentMode.SOLVER,
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> SlurmFuture: ...
+
+
+@overload
+def launch_fluent(
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+) -> dict[str, Any]: ...
+
+
# pylint: disable=unused-argument
@all_deprecators(
deprecate_arg_mappings=[
@@ -145,14 +254,15 @@ def _version_to_dimension(old_arg_val):
warn_message="",
)
def launch_fluent(
+ *,
product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
+ dimension: Dimension | int = Dimension.THREE,
+ precision: Precision | str = Precision.DOUBLE,
processor_count: int | None = None,
journal_file_names: None | str | list[str] = None,
- start_timeout: int = None,
+ start_timeout: int | None = None,
additional_arguments: str = "",
- env: Dict[str, Any] | None = None,
+ env: dict[str, Any] | None = None,
start_container: bool | None = None,
container_dict: dict | None = None,
dry_run: bool = False,
@@ -165,18 +275,26 @@ def launch_fluent(
case_file_name: str | None = None,
case_data_file_name: str | None = None,
lightweight_mode: bool | None = None,
- mode: FluentMode | str | None = None,
+ mode: FluentMode | str = FluentMode.SOLVER,
py: bool | None = None,
gpu: bool | list[int] | None = None,
cwd: str | None = None,
fluent_path: str | None = None,
topy: str | list | None = None,
start_watchdog: bool | None = None,
- scheduler_options: dict | None = None,
+ scheduler_options: SlurmSchedulerOptions | None = None,
file_transfer_service: Any | None = None,
- use_docker_compose: bool | None = None,
- use_podman_compose: bool | None = None,
-) -> Meshing | PureMeshing | Solver | SolverIcing | SlurmFuture | dict:
+ use_docker_compose: bool = False,
+ use_podman_compose: bool = False,
+) -> (
+ Meshing
+ | PureMeshing
+ | Solver
+ | SolverIcing
+ | SolverAero
+ | SlurmFuture
+ | dict[Any, Any]
+):
"""Launch Fluent locally in server mode or connect to a running Fluent server
instance.
@@ -197,8 +315,7 @@ def launch_fluent(
in which case ``Dimension.THREE`` is used. Options are either the values of the
``Dimension`` enum (``Dimension.TWO`` or ``Dimension.THREE``) or any of ``2`` and ``3``.
precision : Precision or str, optional
- Floating point precision. The default is ``None``, in which case ``Precision.DOUBLE``
- is used. Options are either the values of the ``Precision`` enum (``Precision.SINGLE``
+ Floating point precision. Options are either the values of the ``Precision`` enum (``Precision.SINGLE``
or ``Precision.DOUBLE``) or any of ``"double"`` and ``"single"``.
processor_count : int, optional
Number of processors. The default is ``None``, in which case ``1``
@@ -267,7 +384,7 @@ def launch_fluent(
This parameter is used only when ``case_file_name`` is provided. The default is ``False``.
mode : FluentMode or str or None, optional
Launch mode of Fluent to point to a specific session type. Can be a
- ``FluentMode`` enum member or a string. The default value is ``None``.
+ ``FluentMode`` enum member or a string. The default value is ``SOLVER``.
Valid string options include ``"meshing"``, ``"pure-meshing"``, and
``"solver"``.
py : bool, optional
@@ -369,6 +486,7 @@ def _mode_to_launcher_type(fluent_launch_mode: LaunchMode):
def connect_to_fluent(
+ *,
ip: str | None = None,
port: int | None = None,
cleanup_on_exit: bool = False,
@@ -377,7 +495,7 @@ def connect_to_fluent(
password: str | None = None,
start_watchdog: bool | None = None,
file_transfer_service: Any | None = None,
-) -> Meshing | PureMeshing | Solver | SolverIcing:
+) -> Meshing | PureMeshing | Solver | SolverIcing | SolverAero:
"""Connect to an existing Fluent server instance.
Parameters
diff --git a/src/ansys/fluent/core/launcher/pim_launcher.py b/src/ansys/fluent/core/launcher/pim_launcher.py
index dd3788d7a7c4..5315f9fe8e6e 100644
--- a/src/ansys/fluent/core/launcher/pim_launcher.py
+++ b/src/ansys/fluent/core/launcher/pim_launcher.py
@@ -35,10 +35,11 @@
>>> pim_solver_session = pim_solver_launcher()
"""
-import inspect
import logging
import os
-from typing import Any, Dict
+from typing import Any, Dict, TypedDict
+
+from typing_extensions import Unpack
from ansys.fluent.core.fluent_connection import FluentConnection, _get_max_c_int_limit
from ansys.fluent.core.launcher.launch_options import (
@@ -58,6 +59,37 @@
from ansys.fluent.core.utils.fluent_version import FluentVersion
import ansys.platform.instancemanagement as pypim
+
+class PIMArgsWithoutDryRun(
+ TypedDict, total=False
+): # pylint: disable=missing-class-docstring
+ ui_mode: UIMode | str | None
+ graphics_driver: (
+ FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
+ )
+ product_version: FluentVersion | str | float | int | None
+ dimension: Dimension | int | None
+ precision: Precision | str | None
+ processor_count: int | None
+ start_timeout: int
+ additional_arguments: str
+ cleanup_on_exit: bool
+ start_transcript: bool
+ gpu: bool | None
+ start_watchdog: bool | None
+ file_transfer_service: Any | None
+
+
+class PIMArgs(
+ PIMArgsWithoutDryRun, total=False
+): # pylint: disable=missing-class-docstring
+ dry_run: bool
+
+
+class PIMArgsWithMode(PIMArgs, total=False): # pylint: disable=missing-class-docstring
+ mode: FluentMode | str | None
+
+
_THIS_DIR = os.path.dirname(__file__)
_OPTIONS_FILE = os.path.join(_THIS_DIR, "fluent_launcher_options.json")
logger = logging.getLogger("pyfluent.launcher")
@@ -68,23 +100,7 @@ class PIMLauncher:
def __init__(
self,
- mode: FluentMode | str | None = None,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- cleanup_on_exit: bool = True,
- dry_run: bool | None = None,
- start_transcript: bool = True,
- gpu: bool | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
+ **kwargs: Unpack[PIMArgsWithMode],
):
"""
Launch a Fluent session in `PIM `_ mode.
@@ -148,6 +164,9 @@ def __init__(
In job scheduler environments (e.g., SLURM, LSF, PBS), resources and compute nodes are allocated,
and core counts are queried from these environments before being passed to Fluent.
"""
+ additional_arguments = kwargs.get("additional_arguments", "")
+ start_watchdog = kwargs.get("start_watchdog")
+ file_transfer_service = kwargs.get("file_transfer_service")
if additional_arguments:
logger.warning(
@@ -160,14 +179,9 @@ def __init__(
"'start_watchdog' argument for 'launch_fluent()' method is not supported "
"when starting a remote Fluent PyPIM client."
)
- locals_ = locals().copy()
- argvals = {
- arg: locals_.get(arg)
- for arg in inspect.getargvalues(inspect.currentframe()).args
- }
- self.argvals, self.new_session = _get_argvals_and_session(argvals)
+ self.argvals, self.new_session = _get_argvals_and_session(kwargs)
self.file_transfer_service = file_transfer_service
- if self.argvals["start_timeout"] is None:
+ if self.argvals.get("start_timeout") is None:
self.argvals["start_timeout"] = 60
def __call__(self):
diff --git a/src/ansys/fluent/core/launcher/slurm_launcher.py b/src/ansys/fluent/core/launcher/slurm_launcher.py
index 75ae43556bf7..6f0443d3a567 100644
--- a/src/ansys/fluent/core/launcher/slurm_launcher.py
+++ b/src/ansys/fluent/core/launcher/slurm_launcher.py
@@ -68,7 +68,9 @@
import shutil
import subprocess
import time
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, Generic
+
+from typing_extensions import TypeVar
from ansys.fluent.core.exceptions import InvalidArgument
from ansys.fluent.core.launcher.launch_options import (
@@ -161,7 +163,14 @@ def cancel(job_id: int) -> None:
subprocess.run(["scancel", f"{job_id}"])
-class SlurmFuture:
+SessionT = TypeVar(
+ "SessionT",
+ bound="Meshing | PureMeshing | Solver | SolverIcing",
+ default="Meshing | PureMeshing | Solver | SolverIcing",
+)
+
+
+class SlurmFuture(Generic[SessionT]):
"""Encapsulates asynchronous launch of Fluent within a Slurm environment.
The interface is similar to Python's
@@ -221,9 +230,7 @@ def done(self) -> bool:
finished running, otherwise ``False``."""
return self._get_state() in ["", "CANCELLED", "COMPLETED"]
- def result(
- self, timeout: int = None
- ) -> Meshing | PureMeshing | Solver | SolverIcing:
+ def result(self, timeout: int | None = None) -> SessionT:
"""Return the session instance corresponding to the Fluent launch. If Fluent
hasn't yet launched, then this method will wait up to timeout seconds. If Fluent
hasn't launched in timeout seconds, then a TimeoutError will be raised. If
@@ -246,7 +253,7 @@ def result(
"""
return self._future.result(timeout)
- def exception(self, timeout: int = None) -> Exception:
+ def exception(self, timeout: int | None = None) -> Exception:
"""Return the exception raised by the Fluent launch. If Fluent hasn't yet
launched, then this method will wait up to timeout seconds. If Fluent hasn't
launched in timeout seconds, then a TimeoutError will be raised. If timeout is
diff --git a/src/ansys/fluent/core/launcher/standalone_launcher.py b/src/ansys/fluent/core/launcher/standalone_launcher.py
index 10e2fd4573e7..7730853bb9e8 100644
--- a/src/ansys/fluent/core/launcher/standalone_launcher.py
+++ b/src/ansys/fluent/core/launcher/standalone_launcher.py
@@ -35,12 +35,13 @@
>>> standalone_solver_session = standalone_solver_launcher()
"""
-import inspect
import logging
import os
from pathlib import Path
import subprocess
-from typing import Any, Dict
+from typing import Any, TypedDict
+
+from typing_extensions import Unpack
from ansys.fluent.core.launcher.error_handler import (
LaunchFluentError,
@@ -56,6 +57,7 @@
_get_argvals_and_session,
_get_standalone_launch_fluent_version,
)
+from ansys.fluent.core.launcher.launcher import LaunchFluentArgs
from ansys.fluent.core.launcher.launcher_utils import (
_await_fluent_launch,
_build_journal_argument,
@@ -69,8 +71,45 @@
_get_server_info_file_names,
)
import ansys.fluent.core.launcher.watchdog as watchdog
+from ansys.fluent.core.session import BaseSession
from ansys.fluent.core.utils.fluent_version import FluentVersion
+
+class StandaloneArgsWithoutDryRun(
+ TypedDict, total=False
+): # pylint: disable=missing-class-docstring
+ product_version: FluentVersion | str | float | int | None
+ dimension: Dimension | int
+ precision: Precision | str
+ processor_count: int | None
+ journal_file_names: None | str | list[str]
+ start_timeout: int
+ additional_arguments: str
+ env: dict[str, Any] | None
+ cleanup_on_exit: bool
+ start_transcript: bool
+ ui_mode: UIMode | str | None
+ graphics_driver: (
+ FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
+ )
+ case_file_name: str | None
+ case_data_file_name: str | None
+ lightweight_mode: bool | None
+ py: bool | None
+ gpu: bool | list[int] | None
+ cwd: str | None
+ fluent_path: str | None
+ topy: str | list[Any] | None
+ start_watchdog: bool | None
+ file_transfer_service: Any | None
+
+
+class StandaloneArgs(
+ StandaloneArgsWithoutDryRun, total=False
+): # pylint: disable=missing-class-docstring
+ dry_run: bool | None
+
+
logger = logging.getLogger("pyfluent.launcher")
@@ -79,32 +118,10 @@ class StandaloneLauncher:
def __init__(
self,
- mode: FluentMode | str | None = None,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- journal_file_names: None | str | list[str] = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- env: Dict[str, Any] | None = None,
- cleanup_on_exit: bool = True,
+ *,
+ mode: FluentMode,
dry_run: bool = False,
- start_transcript: bool = True,
- case_file_name: str | None = None,
- case_data_file_name: str | None = None,
- lightweight_mode: bool | None = None,
- py: bool | None = None,
- gpu: bool | None = None,
- cwd: str | None = None,
- fluent_path: str | None = None,
- topy: str | list | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
+ **kwargs: Unpack[LaunchFluentArgs],
):
"""
Launch a Fluent session in standalone mode.
@@ -186,19 +203,14 @@ def __init__(
"""
import ansys.fluent.core as pyfluent
- locals_ = locals().copy()
- argvals = {
- arg: locals_.get(arg)
- for arg in inspect.getargvalues(inspect.currentframe()).args
- }
- self.argvals, self.new_session = _get_argvals_and_session(argvals)
- self.file_transfer_service = file_transfer_service
+ self.argvals, self.new_session = _get_argvals_and_session(kwargs)
+ self.file_transfer_service = kwargs.get("file_transfer_service")
if pyfluent.config.show_fluent_gui:
- ui_mode = UIMode.GUI
- self.argvals["ui_mode"] = UIMode(ui_mode)
- if self.argvals["start_timeout"] is None:
+ kwargs["ui_mode"] = UIMode.GUI
+ self.argvals["ui_mode"] = UIMode(kwargs.get("ui_mode"))
+ if self.argvals.get("start_timeout") is None:
self.argvals["start_timeout"] = 60
- if self.argvals["lightweight_mode"] is None:
+ if self.argvals.get("lightweight_mode") is None:
self.argvals["lightweight_mode"] = False
fluent_version = _get_standalone_launch_fluent_version(self.argvals)
if fluent_version:
@@ -207,7 +219,7 @@ def __init__(
if (
fluent_version
and fluent_version >= FluentVersion.v251
- and self.argvals["py"] is None
+ and self.argvals.get("py") is None
):
self.argvals["py"] = True
@@ -225,12 +237,12 @@ def __init__(
self._sifile_last_mtime = Path(self._server_info_file_name).stat().st_mtime
self._kwargs = _get_subprocess_kwargs_for_fluent(
- self.argvals["env"], self.argvals
+ self.argvals.get("env", {}), self.argvals
)
- if self.argvals["cwd"]:
- self._kwargs.update(cwd=self.argvals["cwd"])
+ if "cwd" in self.argvals:
+ self._kwargs.update(cwd=self.argvals.get("cwd"))
self._launch_string += _build_journal_argument(
- self.argvals["topy"], self.argvals["journal_file_names"]
+ self.argvals.get("topy", []), self.argvals.get("journal_file_names")
)
if is_windows():
@@ -243,14 +255,14 @@ def __init__(
# Using 'start.exe' is better; otherwise Fluent is more susceptible to bad termination attempts.
self._launch_cmd = 'start "" ' + self._launch_string
else:
- if self.argvals["ui_mode"] not in [UIMode.GUI, UIMode.HIDDEN_GUI]:
+ if self.argvals.get("ui_mode") not in [UIMode.GUI, UIMode.HIDDEN_GUI]:
# Using nohup to hide Fluent output from the current terminal
self._launch_cmd = "nohup " + self._launch_string + " &"
else:
self._launch_cmd = self._launch_string
- def __call__(self):
- if self.argvals["dry_run"]:
+ def __call__(self) -> tuple[str, str] | BaseSession:
+ if self.argvals.get("dry_run"):
print(f"Fluent launch string: {self._launch_string}")
return self._launch_string, self._server_info_file_name
try:
@@ -261,7 +273,7 @@ def __call__(self):
try:
_await_fluent_launch(
self._server_info_file_name,
- self.argvals["start_timeout"],
+ self.argvals.get("start_timeout", 60),
self._sifile_last_mtime,
)
except TimeoutError as ex:
@@ -275,7 +287,7 @@ def __call__(self):
process = subprocess.Popen(launch_cmd, **self._kwargs)
_await_fluent_launch(
self._server_info_file_name,
- self.argvals["start_timeout"],
+ self.argvals.get("start_timeout", 60),
self._sifile_last_mtime,
)
else:
@@ -284,36 +296,36 @@ def __call__(self):
session = self.new_session._create_from_server_info_file(
server_info_file_name=self._server_info_file_name,
file_transfer_service=self.file_transfer_service,
- cleanup_on_exit=self.argvals["cleanup_on_exit"],
- start_transcript=self.argvals["start_transcript"],
+ cleanup_on_exit=self.argvals.get("cleanup_on_exit"),
+ start_transcript=self.argvals.get("start_transcript"),
launcher_args=self.argvals,
inside_container=False,
)
session._process = process
start_watchdog = _confirm_watchdog_start(
- self.argvals["start_watchdog"],
- self.argvals["cleanup_on_exit"],
+ self.argvals.get("start_watchdog"),
+ self.argvals.get("cleanup_on_exit"),
session._fluent_connection,
)
if start_watchdog:
logger.info("Launching Watchdog for local Fluent client...")
ip, port, password = _get_server_info(self._server_info_file_name)
watchdog.launch(os.getpid(), port, password, ip)
- if self.argvals["case_file_name"]:
- if FluentMode.is_meshing(self.argvals["mode"]):
- session.tui.file.read_case(self.argvals["case_file_name"])
- elif self.argvals["lightweight_mode"]:
- session.read_case_lightweight(self.argvals["case_file_name"])
+ if self.argvals.get("case_file_name"):
+ if FluentMode.is_meshing(self.argvals.get("mode")):
+ session.tui.file.read_case(self.argvals.get("case_file_name"))
+ elif self.argvals.get("lightweight_mode"):
+ session.read_case_lightweight(self.argvals.get("case_file_name"))
else:
session.file.read(
file_type="case",
- file_name=self.argvals["case_file_name"],
+ file_name=self.argvals.get("case_file_name"),
)
- if self.argvals["case_data_file_name"]:
- if not FluentMode.is_meshing(self.argvals["mode"]):
+ if self.argvals.get("case_data_file_name"):
+ if not FluentMode.is_meshing(self.argvals.get("mode")):
session.file.read(
file_type="case-data",
- file_name=self.argvals["case_data_file_name"],
+ file_name=self.argvals.get("case_data_file_name"),
)
else:
raise RuntimeError(
diff --git a/src/ansys/fluent/core/parametric.py b/src/ansys/fluent/core/parametric.py
index fb5c9ae19e79..a13f38837745 100644
--- a/src/ansys/fluent/core/parametric.py
+++ b/src/ansys/fluent/core/parametric.py
@@ -60,7 +60,7 @@
def convert_design_point_parameter_units(
- value: Dict[str, float | int | str]
+ value: Dict[str, float | int | str],
) -> Dict[str, float | int]:
"""Convert design point parameter units."""
diff --git a/src/ansys/fluent/core/session_meshing.py b/src/ansys/fluent/core/session_meshing.py
index 901d5a149a15..c2989d150b4b 100644
--- a/src/ansys/fluent/core/session_meshing.py
+++ b/src/ansys/fluent/core/session_meshing.py
@@ -95,49 +95,12 @@ def switch_to_solver(self) -> Any:
return solver_session
def __getattribute__(self, item: str):
- if super(Meshing, self).__getattribute__(
- "_fluent_connection"
- ) is None and item not in [
+ if super().__getattribute__("_fluent_connection") is None and item not in [
"is_active",
"_fluent_connection",
]:
raise AttributeError(
- f"'{__class__.__name__}' object has no attribute '{item}'"
+ f"'{self.__class__.__name__}' object has no attribute '{item}'"
)
- return super(Meshing, self).__getattribute__(item)
-
- @property
- def tui(self):
- """Meshing TUI root."""
- return super(Meshing, self).tui
-
- @property
- def meshing(self):
- """Meshing datamodel root."""
- return super(Meshing, self).meshing
-
- @property
- def meshing_utilities(self):
- """Meshing utilities datamodel root."""
- return super(Meshing, self).meshing_utilities
-
- @property
- def workflow(self):
- """Workflow datamodel root."""
- return super(Meshing, self).workflow
-
- @property
- def PartManagement(self):
- """Part management datamodel root."""
- return super(Meshing, self).PartManagement
-
- @property
- def PMFileManagement(self):
- """Part management file management datamodel root."""
- return super(Meshing, self).PMFileManagement
-
- @property
- def preferences(self):
- """Preferences datamodel root."""
- return super(Meshing, self).preferences
+ return super().__getattribute__(item)
diff --git a/src/ansys/fluent/core/session_meshing.pyi b/src/ansys/fluent/core/session_meshing.pyi
deleted file mode 100644
index 048bdf335c5d..000000000000
--- a/src/ansys/fluent/core/session_meshing.pyi
+++ /dev/null
@@ -1,53 +0,0 @@
-# Copyright (C) 2021 - 2025 ANSYS, Inc. and/or its affiliates.
-# SPDX-License-Identifier: MIT
-#
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-
-from ansys.fluent.core.generated.datamodel_252.meshing import Root as meshing_root
-from ansys.fluent.core.generated.datamodel_252.meshing_utilities import (
- Root as meshing_utilities_root,
-)
-from ansys.fluent.core.generated.datamodel_252.part_management import (
- Root as partmanagement_root,
-)
-from ansys.fluent.core.generated.datamodel_252.pm_file_management import (
- Root as pmfilemanagement_root,
-)
-from ansys.fluent.core.generated.datamodel_252.preferences import (
- Root as preferences_root,
-)
-from ansys.fluent.core.generated.datamodel_252.workflow import Root as workflow_root
-from ansys.fluent.core.generated.meshing.tui_252 import main_menu
-
-class Meshing:
- @property
- def tui(self) -> main_menu: ...
- @property
- def meshing(self) -> meshing_root: ...
- @property
- def meshing_utilities(self) -> meshing_utilities_root: ...
- @property
- def workflow(self) -> workflow_root: ...
- @property
- def PartManagement(self) -> partmanagement_root: ...
- @property
- def PMFileManagement(self) -> pmfilemanagement_root: ...
- @property
- def preferences(self) -> preferences_root: ...
diff --git a/src/ansys/fluent/core/session_pure_meshing.py b/src/ansys/fluent/core/session_pure_meshing.py
index fd4c9b9e249e..60f104f802ba 100644
--- a/src/ansys/fluent/core/session_pure_meshing.py
+++ b/src/ansys/fluent/core/session_pure_meshing.py
@@ -23,7 +23,7 @@
"""Module containing class encapsulating Fluent connection."""
import functools
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Dict, cast
import ansys.fluent.core as pyfluent
from ansys.fluent.core.data_model_cache import DataModelCache, NameKey
@@ -37,6 +37,23 @@
from ansys.fluent.core.utils.data_transfer import transfer_case
from ansys.fluent.core.utils.fluent_version import FluentVersion
+if TYPE_CHECKING:
+ from ansys.fluent.core.generated.datamodel_252.meshing import Root as meshing_root
+ from ansys.fluent.core.generated.datamodel_252.meshing_utilities import (
+ Root as meshing_utilities_root,
+ )
+ from ansys.fluent.core.generated.datamodel_252.part_management import (
+ Root as partmanagement_root,
+ )
+ from ansys.fluent.core.generated.datamodel_252.pm_file_management import (
+ Root as pmfilemanagement_root,
+ )
+ from ansys.fluent.core.generated.datamodel_252.preferences import (
+ Root as preferences_root,
+ )
+ from ansys.fluent.core.generated.datamodel_252.workflow import Root as workflow_root
+ from ansys.fluent.core.generated.meshing.tui_252 import main_menu
+
class PureMeshing(BaseSession):
"""Encapsulates a Fluent meshing session.
@@ -125,26 +142,27 @@ def __init__(
self._fluent_connection.register_finalizer_cb(stream.stop)
@property
- def tui(self):
+ def tui(self) -> "main_menu":
"""Instance of ``main_menu`` on which Fluent's SolverTUI methods can be
executed."""
- return self._base_meshing.tui
+ return cast("main_menu", self._base_meshing.tui)
@property
- def meshing(self):
+ def meshing(self) -> "meshing_root":
"""Datamodel root of meshing."""
- return self._base_meshing.meshing
+ return cast("meshing_root", self._base_meshing.meshing)
@property
- def meshing_utilities(self):
+ def meshing_utilities(self) -> "meshing_utilities_root | None":
"""Datamodel root of meshing_utilities."""
if self.get_fluent_version() >= FluentVersion.v242:
- return self._base_meshing.meshing_utilities
+ return cast("meshing_utilities_root", self._base_meshing.meshing_utilities)
+ return None
@property
- def workflow(self):
+ def workflow(self) -> "workflow_root":
"""Datamodel root of workflow."""
- return self._base_meshing.workflow
+ return cast("workflow_root", self._base_meshing.workflow)
def watertight(self):
"""Get a new watertight workflow."""
@@ -184,19 +202,19 @@ def topology_based(self):
return self._base_meshing.topology_based_meshing_workflow()
@property
- def PartManagement(self):
+ def PartManagement(self) -> "partmanagement_root":
"""Datamodel root of PartManagement."""
- return self._base_meshing.PartManagement
+ return cast("partmanagement_root", self._base_meshing.PartManagement)
@property
- def PMFileManagement(self):
+ def PMFileManagement(self) -> "pmfilemanagement_root":
"""Datamodel root of PMFileManagement."""
- return self._base_meshing.PMFileManagement
+ return cast("pmfilemanagement_root", self._base_meshing.PMFileManagement)
@property
- def preferences(self):
+ def preferences(self) -> "preferences_root":
"""Datamodel root of preferences."""
- return self._base_meshing.preferences
+ return cast("preferences_root", self._base_meshing.preferences)
def transfer_mesh_to_solvers(
self,
diff --git a/src/ansys/fluent/core/session_pure_meshing.pyi b/src/ansys/fluent/core/session_pure_meshing.pyi
deleted file mode 100644
index d4bfea5052c7..000000000000
--- a/src/ansys/fluent/core/session_pure_meshing.pyi
+++ /dev/null
@@ -1,69 +0,0 @@
-# Copyright (C) 2021 - 2025 ANSYS, Inc. and/or its affiliates.
-# SPDX-License-Identifier: MIT
-#
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-
-from ansys.fluent.core.generated.datamodel_252.meshing import Root as meshing_root
-from ansys.fluent.core.generated.datamodel_252.meshing_utilities import (
- Root as meshing_utilities_root,
-)
-from ansys.fluent.core.generated.datamodel_252.part_management import (
- Root as partmanagement_root,
-)
-from ansys.fluent.core.generated.datamodel_252.pm_file_management import (
- Root as pmfilemanagement_root,
-)
-from ansys.fluent.core.generated.datamodel_252.preferences import (
- Root as preferences_root,
-)
-from ansys.fluent.core.generated.datamodel_252.workflow import Root as workflow_root
-from ansys.fluent.core.generated.meshing.tui_252 import main_menu
-
-class PureMeshing:
- @property
- def tui(self) -> main_menu: ...
- @property
- def meshing(self) -> meshing_root: ...
- @property
- def meshing_utilities(self) -> meshing_utilities_root: ...
- @property
- def workflow(self) -> workflow_root: ...
- def watertight(self): ...
- 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
- def PMFileManagement(self) -> pmfilemanagement_root: ...
- @property
- def preferences(self) -> preferences_root: ...
- def transfer_mesh_to_solvers(
- self,
- solvers,
- file_type: str = ...,
- file_name_stem: str = ...,
- num_files_to_try: int = ...,
- clean_up_mesh_file: bool = ...,
- overwrite_previous: bool = ...,
- ): ...
- def enable_beta_features(self): ...
diff --git a/src/ansys/fluent/core/session_solver.py b/src/ansys/fluent/core/session_solver.py
index 364684d0c7ae..db2db698fcc3 100644
--- a/src/ansys/fluent/core/session_solver.py
+++ b/src/ansys/fluent/core/session_solver.py
@@ -24,7 +24,7 @@
import logging
import threading
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Dict, cast
import warnings
import weakref
@@ -60,6 +60,14 @@
)
from ansys.fluent.core.workflow import ClassicWorkflow
+if TYPE_CHECKING:
+ from ansys.fluent.core.generated.datamodel_252.preferences import (
+ Root as preferences_root,
+ )
+ import ansys.fluent.core.generated.solver.settings_252 as settings_root
+ from ansys.fluent.core.generated.solver.tui_252 import main_menu
+
+
tui_logger = logging.getLogger("pyfluent.tui")
datamodel_logger = logging.getLogger("pyfluent.datamodel")
@@ -181,7 +189,7 @@ def _solution_variable_data(self) -> SolutionVariableData:
)
@property
- def settings(self):
+ def settings(self) -> "settings_root.root":
"""Settings root handle."""
if self._settings is None:
#: Root settings object.
@@ -192,7 +200,7 @@ def settings(self):
file_transfer_service=self._file_transfer_service,
scheme_eval=self.scheme.eval,
)
- return self._settings
+ return cast("settings_root.root", self._settings)
@property
def svar_data(self):
@@ -246,16 +254,16 @@ def _version(self):
return self._fluent_version
@property
- def tui(self):
+ def tui(self) -> "main_menu":
"""Instance of ``main_menu`` on which Fluent's SolverTUI methods can be
executed."""
if self._tui is None:
self._tui = _make_tui_module(self, "solver")
- return self._tui
+ return cast("main_menu", self._tui)
@property
- def workflow(self):
+ def workflow(self) -> ClassicWorkflow:
"""Datamodel root for workflow."""
if not self._workflow:
self._workflow = ClassicWorkflow(
@@ -277,18 +285,18 @@ def _interrupt(cls, command):
command._root.solution.run_calculation.interrupt()
@property
- def system_coupling(self):
+ def system_coupling(self) -> SystemCoupling:
"""System coupling object."""
if self._system_coupling is None:
self._system_coupling = SystemCoupling(self)
return self._system_coupling
@property
- def preferences(self):
+ def preferences(self) -> "preferences_root":
"""Datamodel root of preferences."""
if self._preferences is None:
self._preferences = _make_datamodel_module(self, "preferences")
- return self._preferences
+ return cast("preferences_root", self._preferences)
def _start_bg_session_and_sync(self, launcher_args):
"""Start a background session and sync it with the current session."""
diff --git a/src/ansys/fluent/core/session_solver.pyi b/src/ansys/fluent/core/session_solver.pyi
deleted file mode 100644
index a834a1fd1611..000000000000
--- a/src/ansys/fluent/core/session_solver.pyi
+++ /dev/null
@@ -1,47 +0,0 @@
-# Copyright (C) 2021 - 2025 ANSYS, Inc. and/or its affiliates.
-# SPDX-License-Identifier: MIT
-#
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-
-from ansys.fluent.core.generated.datamodel_252.preferences import (
- Root as preferences_root,
-)
-from ansys.fluent.core.generated.datamodel_252.workflow import Root as workflow_root
-import ansys.fluent.core.generated.solver.settings_252 as settings_root
-from ansys.fluent.core.generated.solver.tui_252 import main_menu
-from ansys.fluent.core.system_coupling import SystemCoupling
-
-class Solver:
- @property
- def version(self): ...
- @property
- def tui(self) -> main_menu: ...
- @property
- def workflow(self) -> workflow_root: ...
- @property
- def system_coupling(self) -> SystemCoupling: ...
- @property
- def preferences(self) -> preferences_root: ...
- def read_case_lightweight(self, file_name: str): ...
- def read_case(self, file_name: str): ...
- def write_case(self, file_name: str): ...
- @property
- def settings(self) -> settings_root.root: ...
- def enable_beta_features(self): ...
diff --git a/src/ansys/fluent/core/session_utilities.py b/src/ansys/fluent/core/session_utilities.py
index e3a16c654ee0..cb286b0c328f 100644
--- a/src/ansys/fluent/core/session_utilities.py
+++ b/src/ansys/fluent/core/session_utilities.py
@@ -22,21 +22,34 @@
"""Session utilities."""
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Literal, overload
-import ansys.fluent.core as pyfluent
-from ansys.fluent.core.launcher.container_launcher import DockerLauncher
+from typing_extensions import Unpack
+
+from ansys.fluent.core import (
+ session_meshing,
+ session_pure_meshing,
+ session_solver,
+ session_solver_aero,
+ session_solver_icing,
+)
+from ansys.fluent.core.launcher.container_launcher import (
+ ContainerArgsWithoutDryRun,
+ DockerLauncher,
+)
from ansys.fluent.core.launcher.launch_options import (
- Dimension,
- FluentLinuxGraphicsDriver,
FluentMode,
- FluentWindowsGraphicsDriver,
- Precision,
- UIMode,
)
-from ansys.fluent.core.launcher.pim_launcher import PIMLauncher
-from ansys.fluent.core.launcher.standalone_launcher import StandaloneLauncher
-from ansys.fluent.core.utils.fluent_version import FluentVersion
+from ansys.fluent.core.launcher.launcher import LaunchFluentArgs, connect_to_fluent
+from ansys.fluent.core.launcher.pim_launcher import (
+ PIMArgsWithoutDryRun,
+ PIMLauncher,
+)
+from ansys.fluent.core.launcher.standalone_launcher import (
+ StandaloneArgsWithoutDryRun,
+ StandaloneLauncher,
+)
+from ansys.fluent.core.session import BaseSession
class SessionBase:
@@ -47,7 +60,7 @@ class SessionBase:
or `from_pim` functions to create a session.
"""
- _session_mode = {
+ _session_mode: dict[str, FluentMode] = {
"Meshing": FluentMode.MESHING,
"PureMeshing": FluentMode.PURE_MESHING,
"PrePost": FluentMode.PRE_POST,
@@ -56,35 +69,31 @@ class SessionBase:
"SolverIcing": FluentMode.SOLVER_ICING,
}
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[StandaloneArgsWithoutDryRun],
+ ) -> BaseSession: ...
+
+ @overload
@classmethod
def from_install(
cls,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- journal_file_names: None | str | list[str] = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- env: Dict[str, Any] = {}, # noqa: B006
- cleanup_on_exit: bool = True,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[StandaloneArgsWithoutDryRun],
+ ) -> tuple[str, str]: ...
+
+ @classmethod
+ def from_install( # pylint: disable=missing-param-doc
+ cls,
+ *,
dry_run: bool = False,
- start_transcript: bool = True,
- case_file_name: str | None = None,
- case_data_file_name: str | None = None,
- lightweight_mode: bool | None = None,
- py: bool | None = None,
- gpu: bool | None = None,
- cwd: str | None = None,
- fluent_path: str | None = None,
- topy: str | list | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
- ):
+ **kwargs: Unpack[StandaloneArgsWithoutDryRun],
+ ) -> BaseSession | tuple[str, str]:
"""
Launch a Fluent session in standalone mode.
@@ -161,34 +170,36 @@ def from_install(
In job scheduler environments (e.g., SLURM, LSF, PBS), resources and compute nodes are allocated,
and core counts are queried from these environments before being passed to Fluent.
"""
- mode = cls._session_mode[cls.__name__]
- argvals = locals().copy()
- argvals.pop("cls", None) # Remove the class reference from the arguments
- launcher = StandaloneLauncher(**argvals)
+ launcher = StandaloneLauncher(
+ **kwargs, dry_run=dry_run, mode=cls._session_mode[cls.__name__]
+ )
return launcher()
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> BaseSession: ...
+
+ @overload
@classmethod
def from_container(
cls,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- container_dict: dict | None = None,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @classmethod
+ def from_container( # pylint: disable=missing-param-doc
+ cls,
+ *,
dry_run: bool = False,
- cleanup_on_exit: bool = True,
- start_transcript: bool = True,
- py: bool | None = None,
- gpu: bool | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
- ):
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> BaseSession | dict[str, Any]:
"""
Launch a Fluent session in container mode.
@@ -255,32 +266,36 @@ def from_container(
In job scheduler environments (e.g., SLURM, LSF, PBS), resources and compute nodes are allocated,
and core counts are queried from these environments before being passed to Fluent.
"""
- mode = cls._session_mode[cls.__name__]
- argvals = locals().copy()
- argvals.pop("cls", None)
- launcher = DockerLauncher(**argvals)
+ launcher = DockerLauncher(
+ **kwargs, dry_run=dry_run, mode=cls._session_mode[cls.__name__]
+ )
return launcher()
+ @overload
@classmethod
def from_pim(
cls,
- ui_mode: UIMode | str | None = None,
- graphics_driver: (
- FluentWindowsGraphicsDriver | FluentLinuxGraphicsDriver | str | None
- ) = None,
- product_version: FluentVersion | str | float | int | None = None,
- dimension: Dimension | int | None = None,
- precision: Precision | str | None = None,
- processor_count: int | None = None,
- start_timeout: int = 60,
- additional_arguments: str = "",
- cleanup_on_exit: bool = True,
- dry_run: bool | None = None,
- start_transcript: bool = True,
- gpu: bool | None = None,
- start_watchdog: bool | None = None,
- file_transfer_service: Any | None = None,
- ):
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> BaseSession: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @classmethod
+ def from_pim( # pylint: disable=missing-param-doc
+ cls,
+ *,
+ dry_run: bool = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> BaseSession | dict[str, Any]:
"""
Launch a Fluent session in `PIM `_ mode.
@@ -342,10 +357,10 @@ def from_pim(
In job scheduler environments (e.g., SLURM, LSF, PBS), resources and compute nodes are allocated,
and core counts are queried from these environments before being passed to Fluent.
"""
- mode = cls._session_mode[cls.__name__]
- argvals = locals().copy()
- argvals.pop("cls", None)
- launcher = PIMLauncher(**argvals)
+ kwargs_with_mode = dict(kwargs)
+ kwargs_with_mode["mode"] = cls._session_mode[cls.__name__]
+ kwargs_with_mode["dry_run"] = dry_run
+ launcher = PIMLauncher(**kwargs_with_mode)
return launcher()
@classmethod
@@ -382,7 +397,7 @@ def from_connection(
TypeError
If the session type does not match the expected session type.
"""
- session = pyfluent.connect_to_fluent(
+ session = connect_to_fluent(
ip=ip,
port=port,
server_info_file_name=server_info_file_name,
@@ -403,34 +418,358 @@ def from_connection(
class Meshing(SessionBase):
"""Encapsulates a Fluent server for meshing session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_meshing.Meshing: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_meshing.Meshing: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_meshing.Meshing: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
class PureMeshing(SessionBase):
"""Encapsulates a Fluent server for pure meshing session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_pure_meshing.PureMeshing: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_pure_meshing.PureMeshing: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_pure_meshing.PureMeshing: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
class PrePost(SessionBase):
"""Encapsulates a Fluent server for pre-post session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
class Solver(SessionBase):
"""Encapsulates a Fluent server for solver session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_solver.Solver: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
class SolverAero(SessionBase):
"""Encapsulates a Fluent server for solver aero session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_solver_aero.SolverAero: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_solver_aero.SolverAero: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_solver_aero.SolverAero: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
class SolverIcing(SessionBase):
"""Encapsulates a Fluent server for solver icing session connection."""
- pass
+ if TYPE_CHECKING:
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> session_solver_icing.SolverIcing: ...
+
+ @overload
+ @classmethod
+ def from_install(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[LaunchFluentArgs],
+ ) -> tuple[str, str]: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> session_solver_icing.SolverIcing: ...
+
+ @overload
+ @classmethod
+ def from_container(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[ContainerArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[False] = False,
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> session_solver_icing.SolverIcing: ...
+
+ @overload
+ @classmethod
+ def from_pim(
+ cls,
+ *,
+ dry_run: Literal[True],
+ **kwargs: Unpack[PIMArgsWithoutDryRun],
+ ) -> dict[str, Any]: ...