import os
import time
import copy
from datetime import date, datetime
import yaml
import matplotlib.pyplot as plt
from matplotlib.lines import Line2D
from matplotlib.legend import Legend
import niceplots
from mpi4py import MPI
from mdss.utils.helpers import load_yaml_input, load_csv_data, make_dir, print_msg, MachineType, YAMLInputType
from mdss.src.main_helper import execute, submit_job_on_hpc
from mdss.resources.misc_defaults import def_plot_options
from mdss.resources.yaml_config import ref_plot_options, check_input_yaml
comm = MPI.COMM_WORLD
[docs]
class simulation():
"""
Executes aero(structural) simulations using the `Top` class defined in [`aerostruct.py`](aerostruct.py).
This class sets up and runs aerodynamic and/or aerostructural simulations based on input parameters provided via a YAML configuration file. It validates the input, manages directories, and handles outputs, including summary files. The simulations are run using subprocesses.
Inputs
----------
- **yaml_input** : str
Path to the YAML file or raw YAML string containing simulation configuration and information.
"""
def __init__(self, yaml_input: str):
check_input_yaml(yaml_input) # Validate the input yaml file
msg = f"YAML file validation is successful"
print_msg(msg, None, comm)
self.sim_info, self.yaml_input_type = load_yaml_input(yaml_input, comm)
self.out_dir = os.path.abspath(self.sim_info['out_dir'])
if self.yaml_input_type == YAMLInputType.FILE:
self.info_file = yaml_input
elif self.yaml_input_type == YAMLInputType.STRING:
self.info_file = os.path.join(self.out_dir, "input.yaml")
with open(self.info_file, 'w') as f:
yaml.dump(self.sim_info, f, sort_keys=False)
self.machine_type = MachineType.from_string(self.sim_info['machine_type']) # Convert string to enum
# Additional options
self.final_out_file = os.path.join(self.out_dir, "overall_sim_info.yaml") # Set the overall simulation info file name.
self.subprocess_flag = True # To toggle opting subprocess.
self.record_subprocess = False # To toggle to record subprocess output.
if self.machine_type == MachineType.HPC:
self.wait_for_job = False # To toggle to wait for the job to finish.
# Create the output directory if it doesn't exist
make_dir(self.out_dir, comm)
################################################################################
# Code for user to run simulations
################################################################################
[docs]
def run(self):
"""
Executes the simulation on either a local machine or an HPC.
This method checks the simulation settings from the input YAML file. Based on the machine_type, it either runs the simulation locally or generates an HPC job script for execution.
Notes
-----
- For local execution, it directly calls `run_problem()`.
- For HPC execution, it creates a Python file and a job script, then submits the job.
"""
sim_info_copy = copy.deepcopy(self.sim_info)
if self.machine_type == MachineType.LOCAL: # Running on a local machine
execute(self)
elif self.machine_type == MachineType.HPC: # Running on a HPC currently supports Great Lakes.
job_id = submit_job_on_hpc(sim_info_copy, self.info_file, self.wait_for_job, comm) # Submit job script
################################################################################
# Code for Post Processing
################################################################################
[docs]
class post_process:
"""
Performs post-processing operations for simulation results.
This class provides functionality to visualize and compare aerodynamic performance data
such as Lift Coefficient (C<sub>L</sub>) and Drag Coefficient (C<sub>D</sub>) against Angle of Attack (Alpha),
based on the simulation configuration provided via a YAML file.
Inputs
------
- **out_dir**: str
Path to the output directory. The output directory should contain the final out file from the simulation.
"""
def __init__(self, out_dir: str, plot_options: dict={}):
self.out_dir = os.path.abspath(out_dir)
self.final_out_file = os.path.join(self.out_dir, "overall_sim_info.yaml") # Setting the overall simulation info file.
try:
self.sim_out_info,_ = load_yaml_input(self.final_out_file, comm)
except:
msg = f"{self.final_out_file} does not exist. Make sure it is the right output directory."
print_msg(msg, None, comm)
raise FileNotFoundError("")
# Additional Options
plot_options = def_plot_options
plot_options.update(plot_options)
self.plot_options = ref_plot_options.model_validate(plot_options)
[docs]
def gen_case_plots(self):
"""
Generates plots comparing experimental data with simulation results for each case and hierarchy.
This method loops through all hierarchies, cases, and scenarios in the simulation output,
and generates side-by-side plots of C<sub>L</sub> and C<sub>D</sub> versus Angle of Attack (Alpha) for each case.
Each scenario is plotted using a distinct marker, and each mesh refinement level is plotted using a different color.
Experimental data, if provided, is overlaid for validation.
Outputs
--------
- *PNG File*:
A comparison plot showing C<sub>L</sub> and C<sub>D</sub> vs Alpha for all scenarios and refinement levels of a case.
- *PNG File*:
A comparison plot showing C_L and C<sub>D</sub> vs Alpha for all scenarios and refinement levels of a case.
The file is saved in the scenario output directory for each case using the case name.
Notes
------
- Experimental data is optional. If not provided, only simulation data is plotted.
- Markers distinguish scenarios; colors distinguish mesh refinement levels.
- A shared legend is placed outside the figure to indicate scenario markers.
- Axis spines are formatted using `niceplots.adjust_spines()` and figures are saved at high resolution (400 dpi).
- Figures are titled using the case name and saved using `niceplots.save_figs()`.
"""
sim_out_info = copy.deepcopy(self.sim_out_info)
for hierarchy, hierarchy_info in enumerate(sim_out_info['hierarchies']): # loop for Hierarchy level
for case, case_info in enumerate(hierarchy_info['cases']): # loop for cases in hierarchy
scenario_legend_entries = []
fig, axs = self._create_fig(case_info["name"].replace("_", " ").upper()) # Create Figure
colors = self.plot_options.colors
if not colors: # Checks if the list is empty
colors = niceplots.get_colors_list()
colors = self.plot_options.colors
if not colors: # Checks if the list is empty
colors = niceplots.get_colors_list()
for scenario, scenario_info in enumerate(case_info['scenarios']): # loop for scenarios that may present
scenario_out_dir = scenario_info['sim_info']['scenario_out_dir']
plot_args = {
'label': scenario_info['name'].replace("_", " ").upper(),
'color': colors[scenario],
}
# To generate plots comparing the refinement levels
scenario_legend_entry = self._add_scenario_level_plots(axs, scenario_info['name'], scenario_info.get('exp_data', None), case_info['mesh_files'], scenario_out_dir, **plot_args)
scenario_legend_entries.append(scenario_legend_entry)
################################# End of Scenario loop ########################################
self._set_legends(fig, axs, scenario_legend_entries)
fig_name = os.path.join(os.path.dirname(scenario_out_dir), case_info['name'])
niceplots.save_figs(fig, fig_name, ["png"], format_kwargs={"png": {"dpi": 400}}, bbox_inches="tight")
[docs]
def custom_compare(self, custom_compare_info: dict, plt_name: str):
"""
Generates a combined plot comparing specific scenarios across hierarchies and cases.
This method creates a figure with two subplots: one for C<sub>L</sub> vs Alpha and another for C<sub>D</sub> vs Alpha.
This method creates a figure with two subplots: one for C<sub>L</sub> vs Alpha and another for C<sub>D</sub> vs Alpha.
It overlays selected scenarios (across different cases and hierarchies) and creates a shared legend
to highlight which scenario each marker represents.
Inputs
-------
- **custom_compare_info**: dict
A dictionary defining the scenarios to be compared.
The structure of the dictionary should be:
.. code:: python
{
"hierarchy_name": {
"case_name": {
"scenarios": ["scenario_name_1", "scenario_name_2"],
"mesh_files": ["mesh_level_1", "mesh_level_2"] # Optional
}
}
}
- *hierarchy_name*: str
Name of the hierarchy the scenario belongs to.
- *case_name*: str
Name of the case within the hierarchy.
- *scenarios*: list[str]
List of scenario names to be plotted.
- *mesh_files*: list[str], optional
List of mesh refinement levels to include for that scenario. If not specified, defaults to all mesh files under the case.
- **plt_name**: str
Name used for the plot title and the saved file name (PNG format).
Outputs
--------
- **PNG File**:
A side-by-side comparison plot showing C<sub>L</sub> and C<sub>D</sub> vs Alpha for all selected scenarios.
The plot is saved in the output directory specified during initialization.
Notes
------
- Each scenario is plotted using a consistent color, with markers indicating refinement levels.
- Experimental data is included when available.
- A shared legend (outside the plot) shows scenario identifiers and their corresponding markers.
"""
sim_out_info = copy.deepcopy(self.sim_out_info)
fig, axs = self._create_fig(plt_name.replace("_", " ").upper()) # Create Figure
scenario_legend_entries = []
found_scenarios = False
count = 0 # To get marker style
colors = self.plot_options.colors
if not colors: # Checks if the list is empty
colors = niceplots.get_colors_list()
scenarios_list = []
for hierarchy, hierarchy_info in custom_compare_info.items():
for case, case_info in hierarchy_info.items():
for scenario in case_info['scenarios']:
scenario_info = {'hierarchy':hierarchy, 'case': case, 'scenario': scenario}
if 'mesh_files' in case_info.keys():
scenario_info['mesh_files'] = case_info['mesh_files']
scenarios_list.append(scenario_info)
for s in scenarios_list:
for hierarchy_info in sim_out_info['hierarchies']:
if hierarchy_info['name'] != s['hierarchy']:
continue
for case_info in hierarchy_info['cases']:
if case_info['name'] != s['case']:
continue
for scenario_info in case_info['scenarios']:
if scenario_info['name'] != s['scenario']:
continue
found_scenarios = True
mesh_files = s.get('mesh_files', case_info['mesh_files'])
scenario_out_dir = scenario_info['sim_info'].get('scenario_out_dir', '.')
label = f"{case_info['name']} - {scenario_info['name']}"
plot_args = {
'label': label.replace("_", " ").upper(),
'color': colors[count]
}
scenario_legend_entry = self._add_scenario_level_plots(axs, scenario_info['name'], scenario_info.get('exp_data', None), mesh_files, scenario_out_dir, **plot_args)
scenario_legend_entries.append(scenario_legend_entry)
count+=1
if not found_scenarios:
return ValueError("None of the scenarios are found")
self._set_legends(fig, axs, scenario_legend_entries)
fig_name = os.path.join(self.out_dir, plt_name)
niceplots.save_figs(fig, fig_name, ["png"], format_kwargs={"png": {"dpi": 400}}, bbox_inches="tight")
def _add_plot_from_csv(self, axs, csv_file:str, **kwargs):
"""
Adds a plot of Angle of Attack vs Lift and Drag Coefficients from a CSV file.
This method expects two subplots: one for C<sub>L</sub> (Lift Coefficient) vs Alpha, and one for C<sub>D</sub> (Drag Coefficient) vs Alpha.
The CSV must contain the columns: 'Alpha', 'CL', and 'CD'.
Inputs
-------
- **axs**: list[matplotlib.axes._subplots.AxesSubplot]
A list of two matplotlib axes. axs[0] is used for plotting C<sub>L</sub> vs Alpha, and axs[1] for C<sub>D</sub> vs Alpha.
- **csv_file**: str
Path to the CSV file containing simulation or experimental data. The file must have 'Alpha', 'CL', and 'CD' columns.
- ****kwargs**:
Optional keyword arguments to customize the plot appearance.
- *label* : str
Label for the plotted line (used in legends). Default is None.
- *color* : str
Color of the plotted line. Default is 'black'.
- *linestyle* : str
Line style for the plotted line. Default is '--'.
- *marker* : str
Marker style for the data points. Default is 's'.
Outputs
--------
- **Adds plot lines to the existing subplots**:
- axs[0] will have a line for C<sub>L</sub> vs Alpha.
- axs[1] will have a line for C<sub>D</sub> vs Alpha.
Notes
------
- If the CSV file cannot be read or is missing required columns, a warning is printed and the plot is skipped.
"""
label = kwargs.get('label', None)
color = kwargs.get('color', 'black')
linestyle = kwargs.get('linestyle', '--')
marker = kwargs.get('marker', 's')
sim_data = load_csv_data(csv_file, comm)
if sim_data is not None:
for ax, y_key in zip(axs, ['CL', 'CD']):
ax.plot(
sim_data['Alpha'], sim_data[y_key],
label=label,
color=color,
linestyle=linestyle,
marker=marker
)
else:
msg = f"{csv_file} is not readable.\nContinuing to plot without '{label}' data."
print_msg(msg, 'warning', comm)
def _add_scenario_level_plots(self, axs, scenario_name, exp_data, mesh_files, scenario_out_dir, **kwargs):
"""
Adds plots for a specific scenario (experimental + simulation) to the existing subplots.
This method:
- Plots experimental data for the scenario if a valid CSV path is provided.
- Loops over mesh refinement levels and plots ADflow results from each mesh file.
- Creates a `Line2D` entry for the scenario to be used in an external legend.
Inputs
-------
- **axs**: list[matplotlib.axes._subplots.AxesSubplot]
A list of two matplotlib axes. axs[0] is for C<sub>L</sub> vs Alpha, and axs[1] is for C<sub>D</sub> vs Alpha.
- **scenario_name**: str
Name of the scenario, used for labeling and legend entry.
- **exp_data**: str or None
Path to the experimental data CSV file. If None, no experimental data is plotted.
- **mesh_files**: list[str]
List of mesh refinement levels to be plotted (e.g., ['coarse', 'medium', 'fine']).
- **scenario_out_dir**: str
Path to the scenario's output directory, where refinement-level folders are located.
- ****kwargs**:
Optional styling arguments passed to `_add_plot_from_csv()`:
- *label* : str
Label for the scenario used in the external legend. Defaults to a cleaned version of `scenario_name`.
- *color* : str
Base color for the scenario legend marker. Defaults to 'black'.
- *linestyle* : str
Line style for the plots. Will be set to '--' for experimental data, and '-' for simulation data.
- *marker* : str
Marker style for the scenario legend entry. Defaults to 's'.
- *markersize* : int
Size of the legend marker. Defaults to 10.
Outputs
--------
- **scenario_legend_entry**: matplotlib.lines.Line2D
A legend entry representing the scenario (based on marker and label) to be added to the external legend.
Notes
------
- Experimental data will only be plotted if the provided `exp_data` file is valid.
- Simulation results are expected to be located in `${scenario_out_dir}/${mesh_file}/ADflow_output.csv`.
"""
scenario_label = scenario_name.replace("_", " ")
label = kwargs.get('label', scenario_label)
color = kwargs.get('color', 'black')
linestyle = kwargs.get('linestyle', '-')
marker = kwargs.get('marker', 's')
markersize = kwargs.get('markersize', 8)
if exp_data: # Add plots experimental data to the plot
exp_args = {
'label': f"{label} - Experimental",
'color': color,
'linestyle': '',
'marker': 'D',
'markersize': markersize + 4,
}
self._add_plot_from_csv(axs, exp_data, **exp_args)
for ii, mesh_file in enumerate(mesh_files): # Loop for refinement levels
refinement_level_dir = os.path.join(scenario_out_dir, f"{mesh_file}")
ADflow_out_file = os.path.join(refinement_level_dir, "ADflow_output.csv")
# Update kwargs
plot_args = {
'label': f"{label} - {mesh_file}",
'color': color,
'linestyle': '-',
'marker': self._get_marker_style(ii),
'markersize': markersize,
}
self._add_plot_from_csv(axs, ADflow_out_file, **plot_args) # To add simulation data to the plots
scenario_legend_entry = Line2D([0], [0], marker=marker, color=color, linestyle='', markersize=markersize, label=label) # Create a legend entry for the scenario
return scenario_legend_entry
def _create_fig(self, title, niceplots_style=None):
"""
Creates a matplotlib figure with subplots for C<sub>L</sub> and C<sub>D</sub> vs Alpha.
This method initializes the figure layout and applies consistent niceplots styling.
Inputs
-------
- **title**: str
Title to be shown at the top of the figure.
- **niceplots_style**: str or None
Optional name of the niceplots style to apply. If None, uses `self.niceplots_style`.
Outputs
--------
- **fig**: matplotlib.figure.Figure
The created figure object.
- **axs**: list[matplotlib.axes._subplots.AxesSubplot]
A list of two subplots for plotting C<sub>L</sub> and C<sub>D</sub> vs Alpha.
Notes
------
- Subplots are pre-configured with axis titles, labels, and grids.
"""
if niceplots_style is None:
niceplots_style = self.plot_options.niceplots_style
figsize = self.plot_options.figsize
plt.style.use(niceplots.get_style(niceplots_style))
fig, axs = plt.subplots(1, 2, figsize=(14, 6), layout="constrained")
fig.suptitle(title)
ylabels = ['$C_L$', '$C_D$']
for ax, ylabel in zip(axs, ylabels):
ax.set_xlabel('Alpha (deg)')
ax.set_ylabel(ylabel)
ax.grid(True)
return fig, axs
def _set_legends(self, fig, axs, scenario_legend_entries):
mesh_handles, mesh_labels = axs[0].get_legend_handles_labels()
# Create the legends
# scenario_legend = Legend(fig, handles=scenario_legend_entries,
# labels=[h.get_label() for h in scenario_legend_entries],
# loc='center left',
# bbox_to_anchor=(1.0, 0.25),
# title='Scenarios',
# frameon=True,
# fontsize=10,
# labelspacing=0.3)
mesh_legend = Legend(fig, handles=mesh_handles,
labels=mesh_labels,
loc='center left',
bbox_to_anchor=(1.0, 0.75),
title='Meshes',
frameon=True,
fontsize=10,
labelspacing=0.3)
#fig.add_artist(scenario_legend)
fig.add_artist(mesh_legend)
niceplots.adjust_spines(axs[0])
niceplots.adjust_spines(axs[1])
#fig.tight_layout(rect=[0, 0, 0.95, 1])
def _get_marker_style(self, idx):
"""
Function to loop though the marker styles listed here.
Add more if needed.
Inputs
-------
- **idx**: int
Index of the current loop
Outputs
--------
- **Marker Style**: str
Marker style for the current index
"""
markers = ['s', 'o', '^', 'v','X', 'P', '.', 'H', 'p', '*', 'h', '+', 'x']
return markers[idx % len(markers)]