Note
Go to the bottom of this page to download the ZIP file for the MAPDL motorbike frame example.
MAPDL motorbike frame: Project query#
This example builds upon the example_mapdl_motorbike_frame example. It shows how to query resources of an existing project and how to download output files.
Here is the project_setup.py file for this project:
project_query.py#
"""
Example to query resources from a project.
- Query values from evaluated jobs, computing some simple statistics on parameter values.
- Download files from the project
"""
import argparse
import logging
import os
from statistics import mean, stdev
from ansys.hps.client import Client, HPSError
from ansys.hps.client.jms import JmsApi, ProjectApi
log = logging.getLogger(__name__)
def print_value_stats(values, title):
"""Print value stats."""
log.info(f"{title}")
# log.info(f"Num : {len(values)}")
log.info(f"Mean : {mean(values)} Stdev: {stdev(values)}")
log.info(f"(Min, Max): ({min(values)}, {max(values)})")
def query_stats(client, project_name):
"""Query statistics."""
log.info("=== Query values and compoute statistics ===")
jms_api = JmsApi(client)
log.info("=== Project")
project = jms_api.get_project_by_name(name=project_name)
log.info(f"ID: {project.id}")
log.info(f"Created on: {project.creation_time}")
project_api = ProjectApi(client, project.id)
log.info("=== Query jobs")
jobs = project_api.get_jobs(eval_status="evaluated", fields=["id", "values", "elapsed_time"])
if not jobs:
log.info("No evaluated jobs found to compute statistics from, exiting")
return
log.info(f"Statistics across {len(jobs)} jobs")
values = [job.values["mapdl_elapsed_time_obtain_license"] for job in jobs]
print_value_stats(
values, "=== License checkoput (parameter: mapdl_elapsed_time_obtain_license)"
)
values = [job.values["mapdl_elapsed_time"] for job in jobs]
print_value_stats(values, "=== Elapsed time MAPDL (mapdl_elapsed_time)")
values = [job.elapsed_time for job in jobs]
print_value_stats(values, "=== Elapsed time HPS (elapsed_time)")
log.info("=== Query tasks")
tasks = project_api.get_tasks(
eval_status="evaluated", fields=["id", "prolog_time", "running_time", "finished_time"]
)
log.info("Statistics across tasks")
values = [(t.running_time - t.prolog_time).total_seconds() for t in tasks]
print_value_stats(values, "=== Prolog time (running_time - prolog_time")
values = [(t.finished_time - t.running_time).total_seconds() for t in tasks]
print_value_stats(values, "=== Running time (finished_time - running_time")
def download_files(client, project_name):
"""Download files."""
out_path = os.path.join(os.path.dirname(__file__), "downloads")
log.info(f"Downloading files to {out_path}")
jms_api = JmsApi(client)
project = jms_api.get_project_by_name(name=project_name)
# Todo: Fix needed in the backend:
# Currently only the get_project() called with id returns all fields of project.
project = jms_api.get_project(id=project.id)
log.info(f"Project id: {project.id}")
project_api = ProjectApi(client, project.id)
jobs = project_api.get_jobs(eval_status="evaluated", fields=["id", "values", "elapsed_time"])
log.info(f"# evaluated jobs: {len(jobs)}")
num = min(len(jobs), 10)
log.info(
f"=== Example 1: Downloading output files of {num} jobs using ProjectApi.download_file()"
)
for job in jobs[0:num]:
log.info(f"Job {job.id}")
for task in project_api.get_tasks(job_id=job.id):
log.info(f"Task {task.id}")
files = project_api.get_files(id=task.output_file_ids)
for f in files:
fpath = os.path.join(out_path, f"task_{task.id}")
log.info(f"Download output file {f.evaluation_path} to {fpath}")
project_api.download_file(file=f, target_path=fpath)
num = 10
files = project_api.get_files(type="image/jpeg", limit=num, content=True)
num = len(jobs)
log.info(f"=== Example 2: Global download of {num} jpeg images in project using content=True")
for f in files:
if f.content:
fpath = os.path.join(out_path, f"task_{f.id}_{f.evaluation_path}")
log.info(f"Download image file to {fpath}")
with open(fpath, "wb") as bf:
bf.write(f.content)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("-n", "--name", type=str, default="Mapdl Motorbike Frame")
parser.add_argument("-j", "--num-jobs", type=int, default=500)
parser.add_argument("-U", "--url", default="https://127.0.0.1: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}")
query_stats(client=client, project_name=args.name)
download_files(client=client, project_name=args.name)
except HPSError as e:
log.error(str(e))
Download the ZIP file for the MAPDL motorbike frame example and use a tool such as 7-Zip to extract the files.