Run arbitrary Python scripts on HPS#

This example shows how to run arbitrary Python scripts. It uses the uv package to generate the required environments on the fly.

The example sets up a project that plots sin(x) using NumPy and Matplotlib and then saves the figure to a file.

The metadata header present in the eval.py script, which defines the dependencies, enables uv to take care of the environment setup:

# /// script
# requires-python = "==3.12"
# dependencies = [
#     "numpy",
#     "matplotlib"
# ]
# ///

For more information, see Inline script metadata in the Python Packaging User Guide and Running a script with dependencies in the uv documentation.

Prerequisites#

For the example to run, uv must be installed and registered on the scaler/evaluator. For installation instructions, see Installing uv in the uv documentation.

Once uv is installed, the package must be registered in the scaler/evaluator with the following properties:

Property	Value
Name	uv
Version	0.7.19
Installation Path	/path/to/uv
Executable	/path/to/uv/bin/uv

Be sure to adjust the version to the one you have installed.

Define a custom cache directory#

The preceding steps set up uv with the cache located in its default location in the user home directory (~/.cache/uv). Depending on your situation, you might prefer a different cache location, such as a shared directory accessible to all evaluators. To define a custom uv cache directory, add the following environment variable to the uv package registration in the scaler/evaluator:

Environment Variable	Value
UV_CACHE_DIR	/path/to/custom/uv/cache/dir

Create offline air-gapped setups#

If internet is not available, you can create offline air-gapped setups for uv using one of these options:

Pre-populate the uv cache with all desired dependencies.
Provide a local Python package index and set uv to use it. For more information, see Package indexes in the uv documentation. This index can then sit in a shared location, with node-local caching applied.
Use pre-generated virtual environments. For more information, see uv venv in the uv documentation.

To turn off network access, you can either set the UV_OFFLINE environment variable or use the --offline flag with many uv commands.

Run the example#

To run the example, execute the project_setup.py script:

uv run project_setup.py

This command sets up a project with a number of jobs. Each job generates a plot.png file.

Options#

The example supports the following command line arguments:

Flag	Example	Description
`-U`, `--url`	`--url=https://localhost:8443/hps`	URL of the target HPS instance
`-u`, `--username`	`--username=repuser`	Username to log into HPS
`-p`, `--password`	`--password=topSecret`	Password to log into HPS
`-j`, `--num-jobs`	`--num-jobs=10`	Number of jobs to generate

Files#

Descriptions follow of the relevant example files.

The project creation script, project_setup.py, handles all communication with the HPS instance, defines the project, and generates the jobs.

project_setup.py#

# /// script
# requires-python = "==3.10"
# dependencies = [
#     "ansys-hps-client",
#     "packaging"
# ]
# ///

"""Python with uv example."""

import argparse
import logging
import os

from ansys.hps.client import Client, HPSError
from ansys.hps.client.jms import (
    File,
    JmsApi,
    Job,
    JobDefinition,
    Project,
    ProjectApi,
    ResourceRequirements,
    Software,
    SuccessCriteria,
    TaskDefinition,
)

log = logging.getLogger(__name__)


def create_project(client, num_jobs):
    log.debug("=== Create Project")
    jms_api = JmsApi(client)
    proj = jms_api.create_project(
        Project(
            name=f"Python UV example - {num_jobs} jobs",
            priority=1,
            active=True,
        ),
        replace=True,
    )
    project_api = ProjectApi(client, proj.id)

    log.debug("=== Define Files")
    cwd = os.path.dirname(__file__)
    # Input Files
    files = [
        File(
            name="eval",
            evaluation_path="eval.py",
            type="text/plain",
            src=os.path.join(cwd, "eval.py"),
        ),
        File(
            name="exec_script",
            evaluation_path="exec_script.py",
            type="text/plain",
            src=os.path.join(cwd, "exec_script.py"),
        ),
        File(
            name="plot",
            evaluation_path="plot.png",
            type="image/png",
            collect=True,
        ),
    ]
    files = project_api.create_files(files)
    file_ids = {f.name: f.id for f in files}

    log.debug("=== Define Task")
    task_def = TaskDefinition(
        name="plotting",
        software_requirements=[Software(name="uv")],
        resource_requirements=ResourceRequirements(
            num_cores=0.5,
            memory=100 * 1024 * 1024,  # 100 MB
            disk_space=10 * 1024 * 1024,  # 10 MB
        ),
        execution_level=0,
        max_execution_time=500.0,
        use_execution_script=True,
        execution_script_id=file_ids["exec_script"],
        execution_command="%executable% run %file:eval%",
        input_file_ids=[file_ids["eval"]],
        output_file_ids=[file_ids["plot"]],
        success_criteria=SuccessCriteria(
            return_code=0,
            require_all_output_files=True,
        ),
    )
    task_defs = project_api.create_task_definitions([task_def])

    print("== Define Job")
    job_def = JobDefinition(
        name="JobDefinition.1", active=True, task_definition_ids=[task_defs[0].id]
    )
    job_def = project_api.create_job_definitions([job_def])[0]
    log.debug(f"== Create {num_jobs} Jobs")
    jobs = []
    for i in range(num_jobs):
        jobs.append(Job(name=f"Job.{i}", eval_status="pending", job_definition_id=job_def.id))
    project_api.create_jobs(jobs)
    log.info(f"Created project '{proj.name}', ID='{proj.id}'")
    return proj


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-j", "--num-jobs", type=int, default=10)
    parser.add_argument("-U", "--url", default="https://localhost:8443/hps")
    parser.add_argument("-u", "--username", default="repuser")
    parser.add_argument("-p", "--password", default="repuser")
    args = parser.parse_args()

    logger = logging.getLogger()
    logging.basicConfig(format="%(message)s", level=logging.DEBUG)

    try:
        log.info("Connect to HPC Platform Services")
        client = Client(url=args.url, username=args.username, password=args.password)
        log.info(f"HPS URL: {client.url}")
        proj = create_project(
            client=client,
            num_jobs=args.num_jobs,
        )

    except HPSError as e:
        log.error(str(e))

The script eval.py, which is evaluated on HPS, contains the code to plot a sine and then save the figure.

eval.py#

# /// script
# requires-python = "==3.12"
# dependencies = [
#     "numpy",
#     "matplotlib"
# ]
# ///


import matplotlib.pyplot as plt
import numpy as np

if __name__ == "__main__":
    # Generate plot
    ts = np.linspace(0.0, 10.0, 100)
    ys = np.sin(ts)

    fig, ax = plt.subplots()
    ax.plot(ts, ys)
    ax.set_xlabel("Time [s]")
    ax.set_ylabel("Displacement [cm]")
    plt.savefig("plot.png", dpi=200)

    # Uncomment to enable venv cleanup in exec script, see execution script for details
    # import json
    # import sys
    # with open("output_parameters.json", "w") as out_file:
    #    json.dump({"exe": sys.executable}, out_file, indent=4)

The execution script, exec_script.py, uses uv to run the evaluation script.

exec_script.py#

"""Simple execution script for Python with uv.

Command formed: uv run <script_file>
"""

import json
import os
import shutil

from ansys.rep.common.logging import log
from ansys.rep.evaluator.task_manager import ApplicationExecution


class PythonExecution(ApplicationExecution):
    def execute(self):
        log.info("Start uv execution script")

        # Identify files
        script_file = next((f for f in self.context.input_files if f["name"] == "eval"), None)
        assert script_file, "Python script file missing"
        output_filename = "output_parameters.json"

        # Identify applications
        app_uv = next((a for a in self.context.software if a["name"] == "uv"), None)
        assert app_uv, "Cannot find app uv"

        # Add " around exe if needed for Windows
        exes = {"uv": app_uv["executable"]}
        for k, v in exes.items():
            if " " in v and not v.startswith('"'):
                exes[k] = f'"{v}"'  # noqa

        # Pass env vars correctly
        env = dict(os.environ)
        env.update(self.context.environment)

        ## Run evaluation script
        cmd = f"{exes['uv']} run {script_file['path']}"
        self.run_and_capture_output(cmd, shell=True, env=env)

        # Read eval.py output parameters
        output_parameters = {}
        try:
            log.debug(f"Loading output parameters from {output_filename}")
            with open(output_filename) as out_file:
                output_parameters = json.load(out_file)
            self.context.parameter_values.update(output_parameters)
            log.debug(f"Loaded output parameters: {output_parameters}")
        except Exception as ex:
            log.info("No output parameters found.")
            log.debug(f"Failed to read output_parameters from file: {ex}")

        # If exe path is in out params, delete the venv folder to avoid runaway uv venv cache
        # See https://github.com/astral-sh/uv/issues/13431
        if "exe" in output_parameters.keys():
            try:
                venv_cache = os.path.abspath(os.path.join(output_parameters["exe"], "..", ".."))
                if os.path.exists(venv_cache):
                    log.debug(f"Cleaning venv cache at {venv_cache}...")
                    shutil.rmtree(venv_cache)
                else:
                    log.debug(f"Venv cache path {venv_cache} does not exist.")
            except Exception as ex:
                log.debug(f"Couldn't clean venv cache at {venv_cache}: {ex}")

        log.info("End Python execution script")

Download the ZIP file for the uv example and use a tool such as 7-Zip to extract the files.

Download ZIP file