Launcher plugin creation#

This page explains how to create a plugin for the Local Product Launcher. The plugin in the example launches Ansys Composite PrepPost (ACP) as a subprocess.

The Local Product Launcher defines the interface that a plugin must satisfy in the interface module.

Note

To simplify the example, the plugin business logic is kept minimal.

Create configuration#

To start, you must create the user-definable configuration for the launcher. Because ACP should be run as a subprocess, the path to the server binary must be defined.

This configuration is defined as a dataclass:

from dataclasses import dataclass

@dataclass
class DirectLauncherConfig:
    binary_path: str

The configuration class defines a single binary_path option of type str.

Define launcher#

Next, you must define the launcher itself. The full launcher code follows. Because there’s quite a lot going on in this code, descriptions of each part are provided.

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

class DirectLauncher(LauncherProtocol[LauncherConfig]):
    CONFIG_MODEL = DirectLauncherConfig
    SERVER_SPEC = {"main": ServerType.GRPC}

    def __init__(self, *, config: DirectLaunchConfig):
        self._config = config
        self._url: str
        self._process: subprocess.Popen[str]

    def start(self) -> None:
        port = find_free_ports()[0]
        self._url = f"localhost:{port}"
        self._process = subprocess.Popen(
            [
                self._config.binary_path,
                f"--server-address=0.0.0.0:{port}",
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            text=True,
        )

    def stop(self, *, timeout: Optional[float]=None) -> None:
        self._process.terminate()
        try:
            self._process.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            self._process.kill()
            self._process.wait()

    def check(self, timeout: Optional[float] = None) -> bool:
        channel = grpc.insecure_channel(self.urls[ServerKey.MAIN])
        return check_grpc_health(channel=channel, timeout=timeout)

    @property
    def urls(self) -> dict[str, str]:
        return {"main": self._url}

The launcher class inherits from LauncherProtocol[LauncherConfig]. This isn’t a requirement, but it means a type checker like mypy can verify that the LauncherProtocol interface is fulfilled.

Next, setting CONFIG_MODEL = DirectLauncherConfig connects the launcher to the configuration class.

The subsequent line, SERVER_SPEC = {"main": ServerType.GRPC}, defines which kind of servers the product starts. Here, there’s only a single server, which is accessible via gRPC. The keys in this dictionary can be chosen arbitrarily, but they should be consistent across the launcher implementation. Ideally, you use the key to convey some meaning. For example, "main" could refer to the main interface to your product and file_transfer could refer to an additional service for file upload and download.

The __init__ method must accept exactly one keyword-only argument, config, which contains the configuration instance. In this example, the configuration is stored in the _config attribute. For the _url and _process attributes, only the type is declared for the benefits of the type checker

def __init__(self, *, config: DirectLaunchConfig):
    self._config = config
    self._url: str
    self._process: subprocess.Popen[str]

The core of the launcher implementation is in the start() and stop() methods:

def start(self) -> None:
    port = find_free_ports()[0]
    self._url = f"localhost:{port}"
    self._process = subprocess.Popen(
        [
            self._config.binary_path,
            f"--server-address=0.0.0.0:{port}",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        text=True,
    )

This start() method selects an available port using the find_free_ports() function. It then starts the server as a subprocess. Note that here, the server output is simply discarded. In a real launcher, the option to redirect it (for example to a file) should be added. The _url attribute keeps track of the URL and port that the server should be accessible on.

The start() method terminates the subprocess:

def stop(self, *, timeout: Optional[float]=None) -> None:
    self._process.terminate()
    try:
        self._process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        self._process.kill()
        self._process.wait()

If your product is prone to ignoring SIGTERM, you might want to add a timeout to the .wait() method and retry with the .kill() method instead of the .terminate() method.

Next, you must provide a way to verify that the product has successfully launched. This is implemented in the check. Because the server implements gRPC health checking, the check_grpc_health() helper can be used for this purpose:

def check(self, timeout: Optional[float] = None) -> bool:
    channel = grpc.insecure_channel(self.urls["main"])
    return check_grpc_health(channel=channel, timeout=timeout)

Finally, the _url attribute stored in the start() method must be made available in the urls property:

@property
def urls(self) -> dict[str, str]:
    return {"main": self._url}

Note that the return value for the urls property should adhere to the schema defined in SERVER_SPEC.

Register entrypoint#

Having defined all the necessary components for a Local Product Launcher plugin, you can now register the plugin, which makes it available. You do this through the Python entrypoints mechanism.

You define the entrypoint in your package’s build configuration. The exact syntax depends on which packaging tool you use:

Setuptools can accept its configuration in one of three ways. Choose the one that applies to your project:

In a pyproject.toml file:

[project.entry-points."ansys.tools.local_product_launcher.launcher"]
"ACP.direct" = "<your.module.name>:DirectLauncher"

In a setup.cfg file:

[options.entry_points]
ansys.tools.local_product_launcher.launcher =
    ACP.direct = <your.module.name>:DirectLauncher

In a setup.py file:

from setuptools import setup

setup(
    # ...,
    entry_points = {
        'ansys.tools.local_product_launcher.launcher': [
            'ACP.direct = <your.module.name>:DirectLauncher'
        ]
    }
)

For more information, see the setuptools documentation.

In a pyproject.toml file:

[project.entry-points."ansys.tools.local_product_launcher.launcher"]
"ACP.direct" = "<your.module.name>:DirectLauncher"

For more information, see the flit documentation.

In a pyproject.toml file:

[tool.poetry.plugins."ansys.tools.local_product_launcher.launcher"]
"ACP.direct" = "<your.module.name>:DirectLauncher"

For more information, see 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.

The entrypoint itself has two parts:

  • The entrypoint name ACP.direct consists of two parts: ACP is the product name, and direct is the launch mode identifier. The name must be of this format and contain exactly one dot . separating the two parts.

  • The entrypoint value <your.module.name>:DirectLauncher defines where the launcher implementation is located. In other words, it must load the launcher class:

    from <your.module.name> import DirectLauncher

For the entrypoints to update, you must re-install your package (even if it was installed with pip install -e).

Add command-line default and description#

With the three preceding parts, you’ve successfully created a Local Product Launcher plugin.

You can now improve the usability of the command line by adding a default and description to the configuration class.

To do so, edit the DirectLaunchConfig class, using the dataclasses.field() function to enrich the binary_path:

  • The default value is specified as the default argument.

  • The description is given in the metadata dictionary, using the special key METADATA_KEY_DOC.

import os
import dataclasses
from typing import Union

from ansys.tools.path import get_available_ansys_installations
from ansys.tools.local_product_launcher.interface import METADATA_KEY_DOC


def get_default_binary_path() -> str:
    try:
        installations = get_available_ansys_installations()
        ans_root = installations[max(installations)]
        binary_path = os.path.join(ans_root, "ACP", "acp_grpcserver")
        if os.name == "nt":
            binary_path += ".exe"
        return binary_path
    except (RuntimeError, FileNotFoundError):
        return ""


@dataclasses.dataclass
class DirectLaunchConfig:

    binary_path: str = dataclasses.field(
        default=get_default_binary_path(),
        metadata={
            METADATA_KEY_DOC: "Path to the ACP gRPC server executable."
        },
    )

For the default value, use the get_available_ansys_installations helper to find the Ansys installation directory.

Now, when running ansys-launcher configure ACP direct, users can see and accept the default value if they want.

Note

If the default value is None, it is converted to the string default for the command-line interface. This allows implementing more complicated default behaviors that may not be expressible when the command-line interface is run.

Add a fallback launch mode#

If you want to provide a fallback launch mode that can be used without any configuration, you can add an entrypoint with the special name <product>.__fallback__.

For example, if you wanted the DirectLauncher to be the fallback for ACP, you could add this entrypoint:

[project.entry-points."ansys.tools.local_product_launcher.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.

Hide advanced options#

If your launcher plugin has advanced options, you can skip prompting the user for them by default. This is done by setting the special key METADATA_KEY_NOPROMPT to True in the metadata dictionary:

import dataclasses

from ansys.tools.local_product_launcher.interface import METADATA_KEY_NOPROMPT


@dataclasses.dataclass
class DirectLaunchConfig:
    <...>
    environment_variables: dict[str, str] = field(
        default={},
        metadata={
            METADATA_KEY_DOC: "Extra environment variables to define when launching the server.",
            METADATA_KEY_NOPROMPT: True
        }
    )