Two-bar truss#
This example shows how to create an HPS project solving a two-bar truss problem with Python.
For the original problem description, see R.L. Fox, Optimization Methods in Engineering Design, Addison Wesley, 1971. Also see Optimization Methods for Engineering Design.
You can download the ZIP file
for the two-bar truss
example and use a tool such as 7-Zip to extract the files.
Here is the project_setup.py
script for this example:
"""
Python two-bar truss example.
"""
import argparse
import logging
import os
import random
from ansys.hps.client import Client, HPSError
from ansys.hps.client.jms import (
File,
FitnessDefinition,
FloatParameterDefinition,
JmsApi,
Job,
JobDefinition,
ParameterMapping,
Project,
ProjectApi,
ResourceRequirements,
Software,
TaskDefinition,
)
log = logging.getLogger(__name__)
def main(client, num_jobs, use_exec_script) -> Project:
"""
Create project solving a two-bar truss problem with Python.
For the original problem description, see R.L. Fox, Optimization Methods in Engineering Design, Addison Wesley, 1971.
Also see `Optimization Methods for Engineering Design <https://apmonitor.com/me575/uploads/Main/optimization_book.pdf>`_.
""" # noqa
log.debug("=== Project")
proj = Project(name="Two-bar Truss Problem", priority=1, active=True)
jms_api = JmsApi(client)
proj = jms_api.create_project(proj, replace=True)
project_api = ProjectApi(client, proj.id)
log.debug("=== Files")
cwd = os.path.dirname(__file__)
files = [
File(
name="inp",
evaluation_path="input_parameters.json",
type="text/plain",
src=os.path.join(cwd, "input_parameters.json"),
),
File(
name="script",
evaluation_path="evaluate.py",
type="text/plain",
src=os.path.join(cwd, "evaluate.py"),
),
File(
name="result",
evaluation_path="output_parameters.json",
type="text/plain",
collect=True,
),
]
if use_exec_script:
# Define and upload an exemplary exec script to run Python
files.append(
File(
name="exec_python",
evaluation_path="exec_python.py",
type="application/x-python-code",
src=os.path.join(cwd, "exec_python.py"),
)
)
files = project_api.create_files(files)
file_ids = {f.name: f.id for f in files}
log.debug("=== Job Definition with simulation workflow and parameters")
job_def = JobDefinition(
name="JobDefinition.1",
active=True,
)
# Input params
input_params = [
FloatParameterDefinition(
name="height", lower_limit=10, upper_limit=100.0, default=30, units="in"
),
FloatParameterDefinition(
name="diameter", lower_limit=0.2, upper_limit=5, default=3, units="in"
),
FloatParameterDefinition(
name="thickness", lower_limit=0.03, upper_limit=0.6, default=0.15, units="in"
),
FloatParameterDefinition(
name="separation_distance", lower_limit=40, upper_limit=150, default=60, units="in"
),
FloatParameterDefinition(
name="young_modulus", lower_limit=1e6, upper_limit=1e8, default=3e7, units="lbs in^-2"
),
FloatParameterDefinition(
name="density", lower_limit=0.1, upper_limit=0.6, default=0.3, units="lbs in^-2"
),
FloatParameterDefinition(
name="load", lower_limit=1e1, upper_limit=1e5, default=66e3, units="lbs"
),
]
input_params = project_api.create_parameter_definitions(input_params)
mappings = [
ParameterMapping(
key_string='"H"',
tokenizer=":",
parameter_definition_id=input_params[0].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"d"',
tokenizer=":",
parameter_definition_id=input_params[1].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"t"',
tokenizer=":",
parameter_definition_id=input_params[2].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"B"',
tokenizer=":",
parameter_definition_id=input_params[3].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"E"',
tokenizer=":",
parameter_definition_id=input_params[4].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"rho"',
tokenizer=":",
parameter_definition_id=input_params[5].id,
file_id=file_ids["inp"],
),
ParameterMapping(
key_string='"P"',
tokenizer=":",
parameter_definition_id=input_params[6].id,
file_id=file_ids["inp"],
),
]
output_params = [
FloatParameterDefinition(name="weight", units="lbs"),
FloatParameterDefinition(name="stress", units="ksi"),
FloatParameterDefinition(name="buckling_stress", units="ksi"),
FloatParameterDefinition(name="deflection", units="in"),
]
output_params = project_api.create_parameter_definitions(output_params)
mappings.extend(
[
ParameterMapping(
key_string='"weight"',
tokenizer=":",
parameter_definition_id=output_params[0].id,
file_id=file_ids["result"],
),
ParameterMapping(
key_string='"stress"',
tokenizer=":",
parameter_definition_id=output_params[1].id,
file_id=file_ids["result"],
),
ParameterMapping(
key_string='"buckling_stress"',
tokenizer=":",
parameter_definition_id=output_params[2].id,
file_id=file_ids["result"],
),
ParameterMapping(
key_string='"deflection"',
tokenizer=":",
parameter_definition_id=output_params[3].id,
file_id=file_ids["result"],
),
]
)
mappings = project_api.create_parameter_mappings(mappings)
job_def.parameter_definition_ids = [o.id for o in input_params + output_params]
job_def.parameter_mapping_ids = [o.id for o in mappings]
task_def = TaskDefinition(
name="python_evaluation",
software_requirements=[Software(name="Python", version="3.10")],
execution_command="%executable% %file:script% %file:inp%",
resource_requirements=ResourceRequirements(
num_cores=0.5,
),
execution_level=0,
max_execution_time=30.0,
input_file_ids=[file_ids["script"], file_ids["inp"]],
output_file_ids=[file_ids["result"]],
)
if use_exec_script:
task_def.use_execution_script = True
task_def.execution_script_id = file_ids["exec_python"]
task_def = project_api.create_task_definitions([task_def])[0]
job_def.task_definition_ids = [task_def.id]
# Fitness definition
fd = FitnessDefinition(error_fitness=10.0)
fd.add_fitness_term(
name="weight",
type="design_objective",
weighting_factor=1.0,
expression="map_design_objective( values['weight'], 35, 20)",
)
fd.add_fitness_term(
name="max_stress",
type="limit_constraint",
weighting_factor=1.0,
expression="map_limit_constraint( values['stress'], 85, 10 )",
)
fd.add_fitness_term(
name="max_deflection",
type="limit_constraint",
weighting_factor=1.0,
expression="map_limit_constraint( values['deflection'], 0.25, 0.05 )",
)
job_def.fitness_definition = fd
# Create job_definition in project
job_def = project_api.create_job_definitions([job_def])[0]
params = project_api.get_parameter_definitions(job_def.parameter_definition_ids)
log.debug("=== Jobs")
jobs = []
for i in range(num_jobs):
values = {
p.name: p.lower_limit + random.random() * (p.upper_limit - p.lower_limit)
for p in params
if p.mode == "input"
}
jobs.append(
Job(name=f"Job.{i}", values=values, eval_status="pending", job_definition_id=job_def.id)
)
jobs = 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("-U", "--url", default="https://127.0.0.1:8443/hps")
parser.add_argument("-u", "--username", default="repuser")
parser.add_argument("-p", "--password", default="repuser")
parser.add_argument("-n", "--num-jobs", type=int, default=50)
parser.add_argument("-es", "--use-exec-script", default=False, action="store_true")
args = parser.parse_args()
logger = logging.getLogger()
logging.basicConfig(format="[%(asctime)s | %(levelname)s] %(message)s", level=logging.DEBUG)
client = Client(url=args.url, username=args.username, password=args.password)
try:
main(client, num_jobs=args.num_jobs, use_exec_script=args.use_exec_script)
except HPSError as e:
log.error(str(e))