def optimise(args: argparse.Namespace | None = None) -> None:
"""
Run optimisation.
Parameters
----------
args : argparse.Namespace | None
Command line arguments for modifying configuration.
"""
_set_logging(log_level=args.log_level)
logger.debug(f"\n{pformat(vars(args))}\n")
_config = _parse_configuration(args)
_set_logging(log_level=_config["log_level"])
processing_function = partial(
layopt.trussopt,
# Placeholder, # filter_levels
width=_config["width"],
height=_config["height"],
stress_tensile=_config["stress_tensile"],
stress_compressive=_config["stress_compressive"],
joint_cost=_config["joint_cost"],
loaded_points=_config["loaded_points"],
load_direction=_config["load_direction"],
load_large=_config["load_large"],
load_small=_config["load_small"],
max_length=_config["max_length"],
support_points=_config["support_points"],
# filter_levels=_config["filter_levels"],
primal_method=_config["primal_method"],
problem_name=_config["problem_name"],
# save_to_csv=_config["save_to_csv"],
notes=_config["notes"],
)
_config["filter_levels"] = [0.001, 0.01, 0.1]
# Run processing in parallel
with Pool(processes=_config["cores"]) as pool:
all_results = defaultdict()
with tqdm(
total=len(_config["filter_levels"]),
desc=f"Solving optimisation for {len(_config['filter_levels'])}, results are under {_config['output_dir']}.",
) as pbar:
for _, _, result, filter_level in pool.imap_unordered(
processing_function, _config["filter_levels"]
):
if result is not None:
all_results[filter_level] = result
pbar.update()
logger.info("Optimisation complete.📈📉✅")
# Aggregate results into a single data frame and write output
all_results_df = pd.concat(all_results.values())
all_results_df = all_results_df.sort_values(["filter_level"])
all_results_df.to_csv(_config["output_dir"] / _config["csv_filename"], index=False)
io.write_config(_config)
logger.info(f"Results saved to {_config['output_dir'] / _config['csv_filename']}")
completion_message(_config=_config)