Launcher: implement gRPC transport option handling

doc/source/changelog/116.added.md

@@ -0,0 +1 @@
+Launcher: implement gRPC transport option handling

doc/source/user_guide/launcher/plugin_creation.rst

@@ -43,9 +43,9 @@ there's quite a lot going on in this code, descriptions of each part are provide
     from typing import Optional
     import subprocess
 
-    from ansys.tools.local_product_launcher.interface import LauncherProtocol, ServerType
-    from ansys.tools.local_product_launcher.helpers.ports import find_free_ports
-    from ansys.tools.local_product_launcher.helpers.grpc import check_grpc_health
+    from ansys.tools.common.launcher.interface import LauncherProtocol, ServerType
+    from ansys.tools.common.launcher.helpers.ports import find_free_ports
+    from ansys.tools.common.launcher.helpers.grpc import check_grpc_health
 
 
     class DirectLauncher(LauncherProtocol[LauncherConfig]):
@@ -190,15 +190,15 @@ You define the entrypoint in your package's build configuration. The exact synta
 
         .. code:: toml
 
-            [project.entry-points."ansys.tools.local_product_launcher.launcher"]
+            [project.entry-points."ansys.tools.common.launcher"]
             "ACP.direct" = "<your.module.name>:DirectLauncher"
 
         In a ``setup.cfg`` file:
 
         .. code:: cfg
 
             [options.entry_points]
-            ansys.tools.local_product_launcher.launcher =
+            ansys.tools.common.launcher =
                 ACP.direct = <your.module.name>:DirectLauncher
 
         In a ``setup.py`` file:
@@ -210,7 +210,7 @@ You define the entrypoint in your package's build configuration. The exact synta
             setup(
                 # ...,
                 entry_points={
-                    "ansys.tools.local_product_launcher.launcher": [
+                    "ansys.tools.common.launcher": [
                         "ACP.direct = <your.module.name>:DirectLauncher"
                     ]
                 }
@@ -225,7 +225,7 @@ You define the entrypoint in your package's build configuration. The exact synta
 
         .. code:: toml
 
-            [project.entry-points."ansys.tools.local_product_launcher.launcher"]
+            [project.entry-points."ansys.tools.common.launcher"]
             "ACP.direct" = "<your.module.name>:DirectLauncher"
 
         For more information, see the `Entry points sections <https://flit.pypa.io/en/stable/pyproject_toml.html#pyproject-project-entrypoints>`_ in the Flit documentation.
@@ -236,12 +236,12 @@ You define the entrypoint in your package's build configuration. The exact synta
 
         .. code:: toml
 
-            [tool.poetry.plugins."ansys.tools.local_product_launcher.launcher"]
+            [tool.poetry.plugins."ansys.tools.common.launcher"]
             "ACP.direct" = "<your.module.name>:DirectLauncher"
 
         For more information, see the `plugins <https://python-poetry.org/docs/pyproject#plugins>`_ in the Poetry documentation.
 
-In all cases, ``ansys.tools.local_product_launcher.launcher`` is an identifier specifying that the entrypoint defines a local product launcher plugin. It must be kept the same.
+In all cases, ``ansys.tools.common.launcher`` is an identifier specifying that the entrypoint defines a local product launcher plugin. It must be kept the same.
 
 The entrypoint itself has two parts:
 
@@ -275,7 +275,7 @@ the ``binary_path``:
     from typing import Union
 
     from ansys.tools.path import get_available_ansys_installations
-    from ansys.tools.local_product_launcher.interface import METADATA_KEY_DOC
+    from ansys.tools.common.launcher.interface import METADATA_KEY_DOC
 
 
     def get_default_binary_path() -> str:
@@ -316,7 +316,7 @@ For example, to make ``DirectLauncher`` the fallback for ACP, add this entry poi
 
 .. code:: toml
 
-    [project.entry-points."ansys.tools.local_product_launcher.launcher"]
+    [project.entry-points."ansys.tools.common.launcher"]
     "ACP.__fallback__" = "<your.module.name>:DirectLauncher"
 
 The fallback launch mode is used with its default configuration. This means that the configuration class must have default values for all its fields.
@@ -333,7 +333,7 @@ to ``True`` in the ``metadata`` dictionary:
 
     import dataclasses
 
-    from ansys.tools.local_product_launcher.interface import METADATA_KEY_NOPROMPT
+    from ansys.tools.common.launcher.interface import METADATA_KEY_NOPROMPT
 
 
     @dataclasses.dataclass

pyproject.toml

@@ -48,7 +48,7 @@ urls.Documentation = "https://ansys.tools.docs.pyansys.com/"
 urls.Issues = "https://github.com/ansys/ansys-tools-common/issues"
 urls.Releases = "https://github.com/ansys/ansys-tools-common/releases/"
 urls.Source = "https://github.com/ansys/ansys-tools-common"
-scripts.ansys-launcher = "ansys.tools.local_product_launcher._cli:cli"
+scripts.ansys-launcher = "ansys.tools.common.launcher._cli:cli"
 scripts.save-ansys-path = "ansys.tools.common.path.save:cli"
 
 [dependency-groups]

src/ansys/tools/common/launcher/__init__.py

@@ -22,13 +22,14 @@
 
 """Local product launcher."""
 
-from . import config, helpers, interface, product_instance
+from . import config, grpc_transport, helpers, interface, product_instance
 from .launch import launch_product
 
 __all__ = [
-    "interface",
-    "helpers",
     "config",
-    "product_instance",
+    "grpc_transport",
+    "helpers",
+    "interface",
     "launch_product",
+    "product_instance",
 ]

src/ansys/tools/common/launcher/_cli.py

@@ -253,12 +253,15 @@ def show_config() -> None:
                         try:
                             config = get_config_for(product_name=product_name, launch_mode=launch_mode)
                             click.echo("        No configuration is set (uses defaults).")
-                        except KeyError:
+                        except (KeyError, RuntimeError):
                             click.echo("        No configuration is set (no defaults available).")
                             continue
-                    config = get_config_for(product_name=product_name, launch_mode=launch_mode)
-                    for field in dataclasses.fields(config):
-                        click.echo(f"        {field.name}: {getattr(config, field.name)}")
+                    try:
+                        config = get_config_for(product_name=product_name, launch_mode=launch_mode)
+                        for field in dataclasses.fields(config):
+                            click.echo(f"        {field.name}: {getattr(config, field.name)}")
+                    except TypeError:
+                        click.echo("        No configuration is set (invalid configuration).")
             except KeyError:
                 click.echo("    No configuration is set.")
             click.echo("")

src/ansys/tools/common/launcher/_plugins.py

@@ -30,7 +30,8 @@
 
 from .interface import FALLBACK_LAUNCH_MODE_NAME, DataclassProtocol, LauncherProtocol
 
-LAUNCHER_ENTRY_POINT = "ansys.tools.local_product_launcher.launcher"
+LAUNCHER_ENTRY_POINT = "ansys.tools.common.launcher"
+DEPRECATED_LAUNCHER_ENTRY_POINT = "ansys.tools.local_product_launcher.launcher"
 
 
 def get_launcher(*, product_name: str, launch_mode: str) -> type[LauncherProtocol[DataclassProtocol]]:
@@ -110,6 +111,6 @@ def get_fallback_launcher(product_name: str) -> type[LauncherProtocol[DataclassP
 def _get_entry_points() -> tuple[importlib.metadata.EntryPoint, ...]:
     """Get all Local Product Launcher plugin entrypoints for launchers."""
     try:
-        return entry_points(group=LAUNCHER_ENTRY_POINT)  # type: ignore
+        return entry_points(group=LAUNCHER_ENTRY_POINT) + entry_points(group=DEPRECATED_LAUNCHER_ENTRY_POINT)  # type: ignore
     except KeyError:
         return tuple()

src/ansys/tools/common/launcher/grpc_transport.py

@@ -0,0 +1,179 @@
+# Copyright (C) 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.
+
+"""Defines options for connecting to a gRPC server."""
+
+from abc import ABC, abstractmethod
+from dataclasses import asdict, dataclass
+import enum
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar
+
+import grpc
+
+from .. import cyberchannel
+
+__all__ = [
+    "TransportMode",
+    "UDSOptions",
+    "WNUAOptions",
+    "MTLSOptions",
+    "InsecureOptions",
+    "TransportOptionsType",
+]
+
+# For Python 3.10 and below, emulate the behavior of StrEnum by
+# inheriting from str and enum.Enum.
+# Note that this does *not* work on Python 3.11+, since the default
+# Enum format method has changed and will not return the value of
+# the enum member.
+# When type checking, always use the Python 3.10 workaround, otherwise
+# the StrEnum resolves as 'Any'.
+if TYPE_CHECKING:  # pragma: no cover
+
+    class StrEnum(str, enum.Enum):
+        """String enum."""
+
+else:
+    try:
+        from enum import StrEnum
+    except ImportError:
+        import enum
+
+        class StrEnum(str, enum.Enum):
+            """String enum."""
+
+            pass
+
+
+class TransportMode(StrEnum):
+    """Enumeration of transport modes supported by the FileTransfer Tool."""
+
+    UDS = "uds"
+    WNUA = "wnua"
+    MTLS = "mtls"
+    INSECURE = "insecure"
+
+
+class TransportOptionsBase(ABC):
+    """Base class for transport options."""
+
+    _MODE: ClassVar[TransportMode]
+
+    @property
+    def mode(self) -> TransportMode:
+        """Transport mode."""
+        return self._MODE
+
+    def create_channel(self, **extra_kwargs: Any) -> grpc.Channel:
+        """Create a gRPC channel using the transport options.
+
+        Parameters
+        ----------
+        extra_kwargs :
+            Extra keyword arguments to pass to the channel creation function.
+
+        Returns
+        -------
+        :
+            gRPC channel created using the transport options.
+        """
+        return cyberchannel.create_channel(**self._to_cyberchannel_kwargs(), **extra_kwargs)
+
+    @abstractmethod
+    def _to_cyberchannel_kwargs(self) -> dict[str, Any]:
+        """Convert transport options to cyberchannel keyword arguments.
+
+        Returns
+        -------
+        :
+            Dictionary of keyword arguments for cyberchannel.
+        """
+        pass
+
+
+@dataclass(kw_only=True)
+class UDSOptions(TransportOptionsBase):
+    """Options for UDS transport mode."""
+
+    _MODE = TransportMode.UDS
+
+    uds_service: str
+    uds_dir: str | Path | None = None
+    uds_id: str | None = None
+
+    def _to_cyberchannel_kwargs(self) -> dict[str, Any]:
+        return asdict(self) | {"transport_mode": self.mode.value}
+
+
+@dataclass(kw_only=True)
+class WNUAOptions(TransportOptionsBase):
+    """Options for WNUA transport mode."""
+
+    _MODE = TransportMode.WNUA
+
+    port: int
+
+    def _to_cyberchannel_kwargs(self) -> dict[str, Any]:
+        return asdict(self) | {"transport_mode": self.mode.value, "host": "localhost"}
+
+
+@dataclass(kw_only=True)
+class MTLSOptions(TransportOptionsBase):
+    """Options for mTLS transport mode."""
+
+    _MODE = TransportMode.MTLS
+
+    certs_dir: str | Path | None = None
+    host: str = "localhost"
+    port: int
+    allow_remote_host: bool = False
+
+    def _to_cyberchannel_kwargs(self) -> dict[str, Any]:
+        if not self.allow_remote_host:
+            if self.host not in ("localhost", "127.0.0.1"):
+                raise ValueError(f"Remote host '{self.host}' specified without setting 'allow_remote_host=True'.")
+        res = asdict(self)
+        res.pop("allow_remote_host", None)
+        return res | {"transport_mode": self.mode.value}
+
+
+@dataclass(kw_only=True)
+class InsecureOptions(TransportOptionsBase):
+    """Options for insecure transport mode."""
+
+    _MODE = TransportMode.INSECURE
+
+    host: str = "localhost"
+    port: int
+    allow_remote_host: bool = False
+
+    def _to_cyberchannel_kwargs(self) -> dict[str, Any]:
+        if not self.allow_remote_host:
+            if self.host not in ("localhost", "127.0.0.1"):
+                raise ValueError(f"Remote host '{self.host}' specified without setting 'allow_remote_host=True'.")
+        res = asdict(self)
+        res.pop("allow_remote_host", None)
+        return res | {"transport_mode": self.mode.value}
+
+
+TransportOptionsType = UDSOptions | WNUAOptions | MTLSOptions | InsecureOptions

src/ansys/tools/common/launcher/interface.py

@@ -26,10 +26,13 @@
 class and register it.
 """
 
+from collections.abc import Mapping
 from enum import Enum, auto
 from types import MappingProxyType
 from typing import Any, ClassVar, Protocol, TypeVar, runtime_checkable
 
+from .grpc_transport import TransportOptionsType
+
 __all__ = [
     "DataclassProtocol",
     "FALLBACK_LAUNCH_MODE_NAME",
@@ -109,7 +112,7 @@ class LauncherProtocol(Protocol[LAUNCHER_CONFIG_T]):
         must be an instance of ``CONFIG_MODEL``.
     """
 
-    CONFIG_MODEL: ClassVar[type[LAUNCHER_CONFIG_T]]
+    CONFIG_MODEL: type[LAUNCHER_CONFIG_T]
     """Defines the configuration options for the launcher.
 
     The configuration options that this launcher accepts, specified
@@ -118,7 +121,7 @@ class LauncherProtocol(Protocol[LAUNCHER_CONFIG_T]):
     used in the configuration CLI, if available.
     """
 
-    SERVER_SPEC: ClassVar[dict[str, ServerType]] = MappingProxyType({})
+    SERVER_SPEC: ClassVar[Mapping[str, ServerType]] = MappingProxyType({})
     """Defines the server types that are started.
 
     Examples
@@ -191,3 +194,20 @@ def urls(self) -> dict[str, str]:
         >>> launcher.urls
         {"MAIN": "http://127.0.0.1:8080", "FILE_TRANSFER": "grpc://127.0.0.1:50051"}
         """
+        if any(val == ServerType.GENERIC for val in self.SERVER_SPEC.values()):
+            raise NotImplementedError("LauncherProtocol.urls must be implemented if any generic servers are started.")
+        return {}
+
+    @property
+    def transport_options(self) -> dict[str, TransportOptionsType]:
+        """Dictionary of transport options for the gRPC servers started.
+
+        The keys of the returned dictionary must correspond to the keys
+        defined in the :attr:`.LauncherProtocol.SERVER_SPEC` attribute
+        which have the value :attr:`ServerType.GRPC`.
+        """
+        if any(val == ServerType.GRPC for val in self.SERVER_SPEC.values()):
+            raise NotImplementedError(
+                "LauncherProtocol.transport_options must be implemented if any gRPC servers are started."
+            )
+        return {}