Update launchers typing to be more specific

Author: Gobot1234

src/ansys/fluent/core/launcher/container_launcher.py

@@ -39,7 +39,7 @@
 import logging
 import os
 import time
-from typing import Any
+from typing import Any, TypedDict, Unpack
 
 from ansys.fluent.core.fluent_connection import FluentConnection
 from ansys.fluent.core.launcher.fluent_container import (
@@ -64,6 +64,33 @@
 from ansys.fluent.core.session import _parse_server_info_file
 from ansys.fluent.core.utils.fluent_version import FluentVersion
 
+
+class ContainerArgsWithoutDryRun(TypedDict, total=False):
+    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):
+    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 +116,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.
@@ -184,31 +192,28 @@ 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}"
 
         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,

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, Unpack, final, overload
+
+from typing_extensions import Required
 
 import ansys.fluent.core as pyfluent
 from ansys.fluent.core._types import PathType
@@ -58,6 +60,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 deprecate_arguments
 from ansys.fluent.core.utils.fluent_version import FluentVersion
@@ -127,6 +130,94 @@ def _version_to_dimension(old_arg_val):
         return None
 
 
+class LaunchFluentArgs(TypedDict, total=False):
+    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):
+    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]: ...
+
 def _custom_converter_gui(kwargs):
     old_val = kwargs.pop("show_gui", None)
     kwargs["ui_mode"] = _show_gui_to_ui_mode(old_val, **kwargs)
@@ -152,14 +243,15 @@ def _custom_converter_dimension(kwargs):
     converter=_custom_converter_dimension,
 )
 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,
@@ -172,18 +264,26 @@ def launch_fluent(
     case_file_name: "PathType | None" = None,
     case_data_file_name: "PathType | 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: "PathType | None" = None,
     fluent_path: "PathType | 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.
 
@@ -204,8 +304,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``
@@ -270,7 +369,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
@@ -372,6 +471,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,
@@ -380,7 +480,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