aeromaps.core.process

High-level AeroMAPS process orchestration.

This module defines the main process class that orchestrates parameter initialization, model instantiation, GEMSEO configuration, generic energy carrier handling, and data export for the AeroMAPS framework.

AeroMAPSProcess

AeroMAPSProcess(configuration_file=None, custom_models=None, optimisation=False)

Bases: object

High-level AeroMAPS process driver.

This class configures parameters, instantiates discipline models, builds GMESEO objects, handles generic energy carrier pathways, and manages input and output data structures for AeroMAPS studies.

Parameters:

Name	Type	Description	Default
configuration_file		Path to a configuration JSON file overriding default settings.	None
models		Dictionary of model instances to be used in the process.	required
optimisation		Whether to configure GEMSEO for optimisation instead of a pure MDA chain.	False

Attributes:

Name	Type	Description
configuration_file		Path of the active configuration JSON file.
models		Dictionary of discipline and auxiliary models used in the process.
parameters		Central parameter container used by all models and disciplines.
disciplines		List of wrapped discipline objects used by GEMSEO or the MDA chain.
data		Dictionary storing structured inputs and outputs, including scalar, string, vector, climate, and LCA results.
json		Dictionary reserved for JSON-compatible representations of results.
mda_chain		GEMSEO MDAChain instance used when running pure MDA analyses.
scenario		GEMSEO scenario instance for conventional MDO.
scenario_adapted		GEMSEO scenario of scenario instance for the bilevel optimization problem.
gemseo_settings		Dictionary containing all GEMSEO-related configuration options.
fleet		Fleet instance when the bottom-up fleet model is activated, else None.
fleet_model		FleetModel instance wrapping the fleet when the bottom-up model is used.
energy_resources_data		Parsed configuration data for generic energy resources.
energy_processes_data		Parsed configuration data for generic energy processes.
energy_carriers_data		Parsed configuration data for aviation energy carrier pathways.
pathways_manager		EnergyCarrierManager instance describing available energy pathways.
climate_historical_data		Historical climate dataset used by climate-related models.

Initialize an AeroMAPSProcess instance.

This method loads configuration settings, initializes parameters, deep-copies the provided models dictionary when needed, and performs the common setup. It then configures either an MDA chain or an optimization scenario depending on the specified mode.

Parameters:

Name	Type	Description	Default
configuration_file		Path to a configuration YAML file overriding default settings.	None
custom_models		Dictionary of additional model instances to be merged with the standard models loaded from the configuration file's models.standards list. If None, only the standard models are used.	None
optimisation		Whether to configure GEMSEO for optimization instead of a pure MDA chain.	False

Source code in aeromaps/core/process.py

def __init__(
    self,
    configuration_file=None,
    custom_models=None,
    optimisation=False,
):
    """Initialize an AeroMAPSProcess instance.

    This method loads configuration settings, initializes parameters,
    deep-copies the provided models dictionary when needed, and
    performs the common setup. It then configures either an MDA chain
    or an optimization scenario depending on the specified mode.

    Parameters
    ----------
    configuration_file
        Path to a configuration YAML file overriding default
        settings.
    custom_models
        Dictionary of additional model instances to be merged with
        the standard models loaded from the configuration file's
        `models.standards` list. If None, only the standard models
        are used.
    optimisation
        Whether to configure GEMSEO for optimization instead of a
        pure MDA chain.
    """
    self.configuration_file = (
        os.path.abspath(os.fspath(configuration_file))
        if configuration_file is not None
        else None
    )
    self._initialize_configuration()

    # Load standard models from config
    standard_models = self._load_models_from_config()

    # Merge with user-provided models (user models override/extend standard models)
    if custom_models is not None:
        standard_models.update(custom_models)

    models = standard_models

    # Recopy models to avoid shared state between instances.
    # For specific models that would be too heavy to deepcopy, set attribute `deepcopy_at_init` to False.
    # E.g., models that load large datasets that are read-only (c.f. LCA model).
    self.models = {
        k: deepcopy(v) if getattr(v, "deepcopy_at_init", True) else v for k, v in models.items()
    }

    # Initialize inputs
    self._initialize_inputs()

    self.common_setup()
    if not optimisation:
        self.setup_mda()
    else:
        self.setup_optimisation()

common_setup

common_setup()

Perform common setup steps independent of analysis type.

This method initializes the disciplines list, the main data container, and JSON storage, and computes index structures and climate data. It also stores the flag indicating whether to add example aircraft and subcategories to the fleet.

Warning

This method should be called only if end year was modified, otherwise it is called in init.

Source code in aeromaps/core/process.py

def common_setup(self):
    """Perform common setup steps independent of analysis type.

    This method initializes the disciplines list, the main data
    container, and JSON storage, and computes index structures and
    climate data. It also stores the flag indicating whether to add
    example aircraft and subcategories to the fleet.

    Warning
    ---------
    This method should be called only if end year was modified, otherwise it is called in __init__.

    """
    self.disciplines = []
    self.data = {}
    self.json = {}
    self._initialize_data()

setup_mda

setup_mda()

Configure the process for a standalone MDA chain.

This method initializes generic energy inputs and disciplines, then builds a GEMSEO MDAChain with default convergence settings for multidisciplinary analysis execution of AeroMAPS.

Warning

This method should be called only if end year was modified, otherwise it is called in init.

Source code in aeromaps/core/process.py

def setup_mda(self):
    """Configure the process for a standalone MDA chain.

    This method initializes generic energy inputs and disciplines,
    then builds a GEMSEO MDAChain with default convergence settings
    for multidisciplinary analysis execution of AeroMAPS.

    Warning
    ---------
    This method should be called only if end year was modified, otherwise it is called in __init__.
    """
    # Initialize energy carriers
    self._initialize_generic_energy()
    # Initialize climate model
    self._initialize_climate_model()
    # Initialize LCA model
    self._initialize_lca_model()
    # Initialize disciplines
    self._initialize_disciplines()

    self.mda_chain = MDAChain(
        disciplines=self.disciplines,
        tolerance=1e-5,
        initialize_defaults=True,
        inner_mda_name="MDAGaussSeidel",
        log_convergence=True,
    )

setup_optimisation

setup_optimisation()

Configure the process for GEMSEO-based optimization.

This method initializes the internal GEMSEO settings dictionary so that optimization scenarios can be defined and executed later.

Source code in aeromaps/core/process.py

def setup_optimisation(self):
    """Configure the process for GEMSEO-based optimization.

    This method initializes the internal GEMSEO settings dictionary
    so that optimization scenarios can be defined and executed later.
    """
    self._initialize_gemseo_settings()

create_gemseo_scenario

create_gemseo_scenario()

Build a single-level GEMSEO MDO scenario.

This method initializes generic energy inputs and disciplines, and then creates a GEMSEO scenario using the current gemseo_settings for objective, design space, scenario type, and formulation.

Source code in aeromaps/core/process.py

def create_gemseo_scenario(self):
    """Build a single-level GEMSEO MDO scenario.

    This method initializes generic energy inputs and disciplines,
    and then creates a GEMSEO scenario using the current
    ``gemseo_settings`` for objective, design space, scenario type,
    and formulation.
    """
    self._initialize_generic_energy()
    self._initialize_climate_model()
    self._initialize_lca_model()
    self._initialize_disciplines()

    self.scenario = create_scenario(
        disciplines=self.disciplines,
        objective_name=self.gemseo_settings["objective_name"],
        design_space=self.gemseo_settings["design_space"],
        scenario_type=self.gemseo_settings["scenario_type"],
        formulation_name=self.gemseo_settings["formulation"],
        main_mda_settings={
            "inner_mda_name": "MDAGaussSeidel",
            "max_mda_iter": 12,
            "initialize_defaults": True,
            "tolerance": 1e-4,
        },
        # grammar_type=self.gemseo_settings["grammar_type"],
        # input_data=self.input_data,
    )

create_gemseo_bilevel

create_gemseo_bilevel()

Build a GEMSEO bilevel optimization formulation.

This method wraps an inner GEMSEO scenario in an MDOScenarioAdapter and creates an outer scenario that optimizes over the adapter. If the inner scenario is not yet defined, it is created using the current gemseo_settings.

Source code in aeromaps/core/process.py

def create_gemseo_bilevel(self):
    """Build a GEMSEO bilevel optimization formulation.

    This method wraps an inner GEMSEO scenario in an
    ``MDOScenarioAdapter`` and creates an outer scenario that
    optimizes over the adapter. If the inner scenario is not yet
    defined, it is created using the current ``gemseo_settings``.
    """
    # if no scenario is created raise an error create_gemseo_scenario needs to be called first
    if self.scenario is None:
        logging.warning(
            f"Inner scenario of the bilevel formulation was not fully defined. Creating it with the following settings:"
            f"Arguments used: disciplines={self.disciplines}, "
            f"objective_name={self.gemseo_settings['objective_name']}, "
            f"design_space={self.gemseo_settings['design_space']}, "
            f"scenario_type={self.gemseo_settings['scenario_type']}, "
            f"formulation_name={self.gemseo_settings['formulation']}"
        )
        self.create_gemseo_scenario()

    self.scenario.set_algorithm(self.gemseo_settings["algorithm_inner"])

    # dv_names = self.scenario.formulation.design_variables.keys()
    self.adapter = MDOScenarioAdapter(
        # TODO make generic --> ?
        self.scenario,
        input_names=self.gemseo_settings["doe_input_names"],
        output_names=self.gemseo_settings["doe_output_names"],
        reset_x0_before_opt=True,
        set_x0_before_opt=False,
    )

    self.scenario_adapted = create_scenario(
        self.adapter,
        formulation_name=self.gemseo_settings["formulation"],
        objective_name=self.gemseo_settings["objective_name_outer"],
        design_space=self.gemseo_settings["design_space_outer"],
        scenario_type="MDO",
    )

compute

compute()

Run the configured analysis or optimization.

This method prepares input data, then executes either a bilevel optimization, a single-level GEMSEO scenario, or an MDA chain depending on the current configuration. After execution, it updates the internal data structures with model outputs.

Source code in aeromaps/core/process.py

def compute(self):
    """Run the configured analysis or optimization.

    This method prepares input data, then executes either a bilevel
    optimization, a single-level GEMSEO scenario, or an MDA chain
    depending on the current configuration. After execution, it
    updates the internal data structures with model outputs.
    """
    input_data = self._pre_compute()
    if hasattr(self, "scenario") and self.scenario:
        if hasattr(self, "scenario_adapted") and self.scenario_adapted:
            print("Running bi-level MDO")
            # self.scenario.default_inputs.update(self.scenario.options)
            self.scenario_adapted.execute(self.gemseo_settings["algorithm_outer"])
        else:
            print("Running MDO")
            self.scenario.execute(self.gemseo_settings["algorithm"])
    else:
        if not hasattr(self, "mda_chain") or self.mda_chain is None:
            raise ValueError("MDA chain not created. Please call setup_mda() first.")
        else:
            print("Running MDA")
            self.mda_chain.execute(input_data=input_data)

    self._update_data_from_model()

get_dataframes

get_dataframes()

Return all main DataFrames as a dictionary, generated on demand.

This method generates and returns a dictionary of key DataFrames representing inputs, outputs, and climate-related quantities in a tabular form suitable for inspection or export.

Returns:

Type	Description
dataframes	Dictionary mapping DataFrame names to pandas DataFrame instances for data information, inputs, and outputs.

Source code in aeromaps/core/process.py

def get_dataframes(self):
    """Return all main DataFrames as a dictionary, generated on demand.

    This method generates and returns a dictionary of key DataFrames
    representing inputs, outputs, and climate-related quantities in a
    tabular form suitable for inspection or export.

    Returns
    -------
    dataframes
        Dictionary mapping DataFrame names to pandas DataFrame
        instances for data information, inputs, and outputs.
    """
    return {
        "data_information": self._get_data_information_df(),
        "vector_inputs": self._get_vector_inputs_df(),
        "float_inputs": self._get_float_inputs_df(),
        "str_inputs": self._get_str_inputs_df(),
        "vector_outputs": self._get_vector_outputs_df(),
        "float_outputs": self._get_float_outputs_df(),
        "climate_outputs": self._get_climate_outputs_df(),
        # Add more if needed
    }

get_json

get_json()

Return the model outputs as a JSON-serializable dictionary.

Returns:

Type	Description
json_data	Dictionary containing JSON-compatible inputs and outputs.

Source code in aeromaps/core/process.py

def get_json(self):
    """
    Return the model outputs as a JSON-serializable dictionary.

    Returns
    -------
    json_data
        Dictionary containing JSON-compatible inputs and outputs.
    """
    return self._data_to_json()

write_json

write_json(file_name=None)

Write model inputs and outputs to a JSON file.

This method builds the JSON-compatible data and writes it to disk, using either the provided file name or the path defined in the configuration.

Parameters:

Name	Type	Description	Default
file_name		Path to the output JSON file. If None, the path from the configuration is used.	None

Source code in aeromaps/core/process.py

def write_json(self, file_name=None):
    """Write model inputs and outputs to a JSON file.

    This method builds the JSON-compatible data and writes it to
    disk, using either the provided file name or the path defined in
    the configuration.

    Parameters
    ----------
    file_name
        Path to the output JSON file. If None, the path from the
        configuration is used.
    """
    if file_name is None:
        file_name = self._resolve_config_path(
            "data", "outputs", "json_outputs_file",
            default_filename="outputs.json"
        )

    # Ensure the directory exists
    os.makedirs(os.path.dirname(file_name), exist_ok=True)

    # Retrieve the data from the model
    json_data = self.get_json()

    with open(file_name, "w", encoding="utf-8") as f:
        dump(json_data, f, ensure_ascii=False, indent=4)

write_excel

write_excel(file_name=None)

Write main result tables to an Excel workbook.

This method exports data information, inputs, and outputs into separate sheets of a single Excel file.

Parameters:

Name	Type	Description	Default
file_name		Path to the output Excel file. If None, the path from the configuration is used.	None

Source code in aeromaps/core/process.py

def write_excel(self, file_name=None):
    """Write main result tables to an Excel workbook.

    This method exports data information, inputs, and outputs into
    separate sheets of a single Excel file.

    Parameters
    ----------
    file_name
        Path to the output Excel file. If None, the path from the
        configuration is used.
    """
    if file_name is None:
        file_name = self._resolve_config_path(
            "data", "outputs", "excel_outputs_file",
            default_filename="data.xlsx"
        )
    with pd.ExcelWriter(file_name) as writer:
        self._get_data_information_df().to_excel(writer, sheet_name="Data Information")
        self._get_vector_inputs_df().to_excel(writer, sheet_name="Vector Inputs")
        self._get_float_inputs_df().to_excel(writer, sheet_name="Float Inputs")
        self._get_str_inputs_df().to_excel(writer, sheet_name="String Inputs")
        self._get_vector_outputs_df().to_excel(writer, sheet_name="Vector Outputs")
        self._get_float_outputs_df().to_excel(writer, sheet_name="Float Outputs")
        self._get_climate_outputs_df().to_excel(writer, sheet_name="Climate Outputs")

generate_n2

generate_n2()

Generate an N2 diagram for the current disciplines.

This method calls GEMSEO to create an N2 plot describing the coupling structure between the configured disciplines.

Source code in aeromaps/core/process.py

def generate_n2(self):
    """Generate an N2 diagram for the current disciplines.

    This method calls GEMSEO to create an N2 plot describing the
    coupling structure between the configured disciplines.
    """
    generate_n2_plot(self.disciplines)

list_available_plots

list_available_plots()

List the names of supported plots.

Returns:

Type	Description
plot_names	List of strings identifying available plot functions.

Source code in aeromaps/core/process.py

def list_available_plots(self):
    """List the names of supported plots.

    Returns
    -------
    plot_names
        List of strings identifying available plot functions.
    """
    return list(available_plots.keys())

list_float_inputs

list_float_inputs()

Return the current scalar input values.

Returns:

Type	Description
float_inputs	Dictionary of scalar input names and their values.

Source code in aeromaps/core/process.py

def list_float_inputs(self):
    """Return the current scalar input values.

    Returns
    -------
    float_inputs
        Dictionary of scalar input names and their values.
    """
    return self.data["float_inputs"]

list_str_inputs

list_str_inputs()

Return the current string input values.

Returns:

Type	Description
str_inputs	Dictionary of string input names and their values.

Source code in aeromaps/core/process.py

def list_str_inputs(self):
    """Return the current string input values.

    Returns
    -------
    str_inputs
        Dictionary of string input names and their values.
    """
    return self.data["str_inputs"]

plot

plot(name, save=False, size_inches=None, remove_title=False)

Generate a predefined AeroMAPS plot.

Depending on the plot name, this method uses either generic or fleet-specific plotting functions and optionally saves the figure to a PDF file.

Parameters:

Name	Type	Description	Default
name		Identifier of the plot to generate, possible to obtain from list_available_plots().	required
save		Whether to save the generated plot as a PDF file.	False
size_inches		Optional figure size in inches as a tuple or list.	None
remove_title		Whether to remove the plot title before saving.	False

Returns:

Type	Description
fig	Object holding the created plot, as returned by the plot function.

Source code in aeromaps/core/process.py

def plot(self, name, save=False, size_inches=None, remove_title=False):
    """Generate a predefined AeroMAPS plot.

    Depending on the plot name, this method uses either generic or
    fleet-specific plotting functions and optionally saves the figure
    to a PDF file.

    Parameters
    ----------
    name
        Identifier of the plot to generate, possible to obtain from list_available_plots().
    save
        Whether to save the generated plot as a PDF file.
    size_inches
        Optional figure size in inches as a tuple or list.
    remove_title
        Whether to remove the plot title before saving.

    Returns
    -------
    fig
        Object holding the created plot, as returned by the plot
        function.
    """
    if name in available_plots_fleet:
        try:
            # todo: if we pass the process to the plot, fleet_model is no longer needed as an argument.
            fig = available_plots_fleet[name](self, self.fleet_model)
            if save:
                if size_inches is not None:
                    fig.fig.set_size_inches(size_inches)
                if remove_title:
                    fig.fig.gca().set_title("")
                fig.fig.savefig(f"{name}.pdf", bbox_inches="tight")
        except AttributeError as e:
            raise NameError(
                f"Plot {name} requires using bottom up fleet model. Original error: {e}"
            )
    elif name in available_plots:
        fig = available_plots[name](self)
        if save:
            if size_inches is not None:
                fig.fig.set_size_inches(size_inches)
            if remove_title:
                fig.fig.gca().set_title("")
            fig.fig.savefig(f"{name}.pdf", bbox_inches="tight")
    else:
        raise NameError(
            f"Plot {name} is not available. List of available plots: {list(available_plots.keys()), list(available_plots_fleet.keys())}"
        )
    return fig