`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,
    disable_execution_statistics=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	Description	Default
`configuration_file`	Path to a configuration JSON file overriding default settings.	`None`
`custom_models`	Dictionary of model instances to be used in the process.	`None`
`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	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`
`disable_execution_statistics`	Whether to disable GEMSEO's execution statistics shared memory. If False, statistics are enabled. Set to True to disable (useful when running many disciplines to avoid semaphore exhaustion).	`False`

Source code in aeromaps/core/process.py

def __init__(
    self,
    configuration_file=None,
    custom_models=None,
    optimisation=False,
    disable_execution_statistics=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.
    disable_execution_statistics
        Whether to disable GEMSEO's execution statistics shared memory.
        If False, statistics are enabled. Set to True to disable
        (useful when running many disciplines to avoid semaphore exhaustion).
    """
    # Initialize pathways_manager to None - will be populated if energy models are used
    self.pathways_manager = None

    # Initialize operations_manager to None - will be populated if generic operations models are used
    self.operations_manager = None

    # Initialize offsets_manager to None - will be populated if generic offsets models are used
    self.offsets_manager = None

    # How a non-converged MDA is reported at the end of compute(): "raise"
    # (default), "warn" or "ignore". An unconverged run is not a solution of the
    # coupled system and is indistinguishable from a converged one in the output
    # DataFrames, hence the default. Set to "warn" from a notebook to inspect a
    # scenario that does not converge yet.
    self.on_mda_failure = "raise"
    self._mda_context = ""

    custom_logger_config(logging.getLogger("gemseo.utils.source_parsing"))

    self.configuration_file = (
        os.path.abspath(os.fspath(configuration_file))
        if configuration_file is not None
        else None
    )

    # Handle execution statistics
    if disable_execution_statistics:
        from aeromaps.core.gemseo import disable_gemseo_execution_statistics

        disable_gemseo_execution_statistics()
        logging.info("Disabled GEMSEO execution statistics")

    self._initialize_configuration()

    # Store mode flags
    self._optimisation = optimisation

    # --- Standard initialization ---
    # 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()
    }

    self._initialize_inputs()

    # Common setup (disciplines list, data containers, etc.)
    self.common_setup()

    # --- Mode-specific setup ---
    if optimisation:
        self.setup_optimisation()
    else:
        self.setup_mda()

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__.

    """
    # Initialization order is load-bearing — do not reorder:
    #   1. _initialize_inputs()        — loads parameters.json + user JSON, then
    #                                    calls _format_input_vectors() to pad and
    #                                    index the *_init arrays from that JSON
    #   2. _initialize_markets()       — pushes market YAML values (growth rates,
    #                                    covid params, share defaults, …) into parameters
    #   3. _initialize_climate_model() / _initialize_lca_model() / _initialize_generic_energy()
    #   4. _initialize_vector_inputs() — loads AeroSCOPE-derived partitioning data;
    #                                    scalars override YAML defaults for market shares,
    #                                    vectors use .update() to fill only the historical
    #                                    slice of already-formatted Series

    # TODO : think about the execution order.
    #    Currently it is driven by what should be available for models and what writes paramaters last (e.g. partitionning after custom setups)

    # Initialize markets registry (must precede disciplines that consume it)
    self._initialize_markets()
    self._initialize_climate_model()
    self._initialize_lca_model()
    self._initialize_generic_energy()
    self._initialize_operations()
    self._initialize_offsets()
    self._initialize_vector_inputs()

    # Fail loudly on stale `_2019` config keys (renamed when prospection_start_year
    # became flexible). Runs after every user surface (JSON inputs, market YAML
    # leaves, vector/partitioning inputs) has been applied to self.parameters.
    self._check_renamed_inputs()

    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 is called in init. A change of the year bounds needs no new call: compute() rebuilds the MDA chain itself.

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 is called in __init__. A change of the year bounds needs no new call:
    compute() rebuilds the MDA chain itself.
    """
    #  Initialize conventional disciplines.
    # Here and not in common_setup because one need to create scenario
    # and declare constraints as models from the same place. TODO; clarify why though.
    self._initialize_disciplines()

    # TODO: expose these MDA settings (tolerance, max_mda_iter, inner_mda_name,
    # ...) as kwargs read from the configuration file instead of hardcoding them.
    # Tolerance must be tight enough to resolve the price-elastic demand loop
    # (doc_net_energy_per_rpk_mean <-> rpk). At 1e-5 the Gauss-Seidel solver
    # reports convergence while that coupling is still ~25% off in SAF-type
    # scenarios; max_mda_iter gives it room to reach the tighter tolerance.
    # Tuning these from a notebook is a minefield -- assign on the chain, not on
    # its inner MDAs, and see "Tuning an MDAChain" in aeromaps/core/gemseo.py.
    self.mda_chain = MDAChain(
        disciplines=self.disciplines,
        tolerance=1e-10,
        max_mda_iter=200,
        initialize_defaults=True,
        inner_mda_name="MDAGaussSeidel",
        log_convergence=True,
    )
    self._configure_mda_chain()

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_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()
    ran_mda = False
    if hasattr(self, "scenario") and self.scenario:
        if hasattr(self, "scenario_adapted") and self.scenario_adapted:
            if self.gemseo_settings.get("algorithm_outer") is None:
                raise ValueError(
                    "Cannot run bi-level MDO: 'algorithm_outer' is not set in gemseo_settings."
                )
            logging.info("Running bi-level MDO")
            # self.scenario.default_inputs.update(self.scenario.options)
            self.scenario_adapted.execute(self.gemseo_settings["algorithm_outer"])
        else:
            if self.gemseo_settings.get("algorithm") is None:
                raise ValueError("Cannot run MDO: 'algorithm' is not set in gemseo_settings.")
            logging.info("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:
            logging.info("Running MDA")
            self.mda_chain.execute(input_data=input_data)
            ran_mda = True

    self._update_data_from_model()

    # Checked after the outputs have been harvested, so that a failed run is still
    # inspectable by whoever catches the error.
    if ran_mda:
        check_mda_convergence(
            self.mda_chain, context=self._mda_context, on_failure=self.on_mda_failure
        )

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"
        )
    if file_name is None:
        raise ValueError("Cannot resolve output JSON file path. Check your configuration.")

    # Ensure the directory exists
    try:
        os.makedirs(os.path.dirname(file_name), exist_ok=True)
    except OSError as e:
        raise OSError(f"Cannot create output directory for '{file_name}': {e}") from e

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

    try:
        with open(file_name, "w", encoding="utf-8") as f:
            dump(json_data, f, ensure_ascii=False, indent=4)
    except OSError as e:
        raise OSError(f"Failed to write JSON output to '{file_name}': {e}") from e

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"
        )
    if file_name is None:
        raise ValueError("Cannot resolve output Excel file path. Check your configuration.")
    try:
        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")
            # self.lca_outputs_xarray.to_excel(writer, sheet_name="LCA Outputs")
    except OSError as e:
        raise OSError(f"Failed to write Excel output to '{file_name}': {e}") from e

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)

describe_models ¶

describe_models(
    include_agnostic=True,
    scope=None,
    domain=None,
    display=True,
)

Summarise every discipline wired into this process.

Surfaces, for each discipline, its domain (air traffic, fleet, energy, costs, climate, … — derived from the model class's module path under aeromaps.models, so new models classify automatically), its :attr:~aeromaps.models.base.AeroMAPSModel.MARKET_SCOPE (per-market / cross-market / aggregator / market-agnostic), its :attr:~aeromaps.models.base.AeroMAPSModel.MODEL_APPROACH (top-down / bottom-up, where applicable), its coupling role — loop (inside the MDA feedback cycle, re-run every iteration and therefore driving convergence cost) vs feed-fwd (run once), derived from the GEMSEO coupling structure — the market it is bound to, its region namespace (multi-regional runs) and its I/O counts. This is the topology + cost view that is otherwise only recoverable by reading the factory wiring and the N2 plot.

Parameters:

Name	Type	Description	Default
`include_agnostic`	`bool`	Include `market_agnostic` disciplines in the per-discipline table. True by default (all disciplines are shown, grouped by domain); pass False for the market-centric view that hides them. They are always counted in the summaries either way.	`True`
`scope`	`str or None`	If set (one of :data:`~aeromaps.models.base.MARKET_SCOPES`), restrict the table to that scope only.	`None`
`domain`	`str or None`	If set (one of the domains listed in the domain summary, e.g. `"impacts/costs"`), restrict the table to that domain only. Combines with `scope`.	`None`
`display`	`bool`	If True, print the summary; the string is returned in either case.	`True`

Returns:

Type	Description
`str`	The formatted multi-line summary.

Source code in aeromaps/core/process.py

def describe_models(self, include_agnostic=True, scope=None, domain=None, display=True):
    """Summarise every discipline wired into this process.

    Surfaces, for each discipline, its **domain** (air traffic, fleet, energy,
    costs, climate, … — derived from the model class's module path under
    ``aeromaps.models``, so new models classify automatically), its
    :attr:`~aeromaps.models.base.AeroMAPSModel.MARKET_SCOPE`
    (per-market / cross-market / aggregator / market-agnostic), its
    :attr:`~aeromaps.models.base.AeroMAPSModel.MODEL_APPROACH` (top-down /
    bottom-up, where applicable), its **coupling
    role** — ``loop`` (inside the MDA feedback cycle, re-run every iteration and
    therefore driving convergence cost) vs ``feed-fwd`` (run once), derived from
    the GEMSEO coupling structure — the market it is bound to, its region namespace
    (multi-regional runs) and its I/O counts. This is the topology + cost view that
    is otherwise only recoverable by reading the factory wiring and the N2 plot.

    Parameters
    ----------
    include_agnostic : bool
        Include ``market_agnostic`` disciplines in the per-discipline table.
        True by default (all disciplines are shown, grouped by domain); pass
        False for the market-centric view that hides them. They are always
        counted in the summaries either way.
    scope : str or None
        If set (one of :data:`~aeromaps.models.base.MARKET_SCOPES`), restrict
        the table to that scope only.
    domain : str or None
        If set (one of the domains listed in the domain summary, e.g.
        ``"impacts/costs"``), restrict the table to that domain only.
        Combines with ``scope``.
    display : bool
        If True, print the summary; the string is returned in either case.

    Returns
    -------
    str
        The formatted multi-line summary.
    """
    from aeromaps.models.base import MARKET_SCOPES

    _ORDER = ["per_market", "cross_market", "aggregator", "market_agnostic"]
    disciplines = list(getattr(self, "disciplines", []) or [])

    def _domain(model):
        # Domain = the model's home (sub)package under aeromaps.models, e.g.
        # "impacts/generic_energy_model" or "air_transport/air_traffic".
        # Derived from the class's module path — no per-model annotation, so
        # new models classify automatically. The trailing module file is
        # dropped to keep package-level granularity (root-level modules keep
        # their own name).
        parts = (type(model).__module__ or "?").split(".")
        if parts[:2] == ["aeromaps", "models"]:
            parts = parts[2:]
        pkg = parts[:-1] or parts
        return "/".join(pkg[:2]) if pkg else "?"

    # Which disciplines sit inside the MDA feedback loop (strongly coupled) — they
    # are re-executed on every Gauss-Seidel iteration and drive convergence cost;
    # the rest are feed-forward (run once). Best-effort: skip on any error/version.
    strong_ids, coupling_available = set(), False
    try:
        from gemseo.core.coupling_structure import CouplingStructure

        strong_ids = {
            id(d) for d in CouplingStructure(disciplines).strongly_coupled_disciplines
        }
        coupling_available = True
    except Exception:
        pass

    rows = []
    counts = {s: 0 for s in _ORDER}
    for discipline in disciplines:
        model = getattr(discipline, "model", None)
        if model is None:
            continue
        model_scope = getattr(model, "MARKET_SCOPE", "market_agnostic")
        counts[model_scope] = counts.get(model_scope, 0) + 1

        # Market binding: the market_id for per-market disciplines, the market
        # count for the spanning ones, "-" when the discipline ignores markets.
        if model_scope == "per_market":
            market = str(getattr(model, "market_id", "?"))
        elif model_scope in ("cross_market", "aggregator"):
            ids = getattr(model, "passenger_market_ids", None) or getattr(
                model, "freight_market_ids", None
            )
            market = f"(all: {len(ids)})" if ids else "(all)"
        else:
            market = "-"

        try:
            n_in = len(list(discipline.input_grammar.names))
            n_out = len(list(discipline.output_grammar.names))
        except Exception:
            n_in = n_out = "?"

        rows.append(
            {
                "domain": _domain(model),
                "scope": model_scope,
                "approach": getattr(model, "MODEL_APPROACH", None) or "—",
                "coupling": (
                    ("loop" if id(discipline) in strong_ids else "feed-fwd")
                    if coupling_available
                    else "?"
                ),
                "model": type(model).__name__,
                "instance": getattr(model, "name", None) or getattr(discipline, "name", "?"),
                "region": self._region_of(discipline) or "",
                "market": market,
                "n_in": n_in,
                "n_out": n_out,
            }
        )

    # Row filtering for the table (summaries always count everything).
    shown = rows
    if domain is not None:
        available_domains = sorted({r["domain"] for r in rows})
        if domain not in available_domains:
            raise ValueError(f"domain must be one of {available_domains}; got {domain!r}")
        shown = [r for r in shown if r["domain"] == domain]
    if scope is not None:
        if scope not in MARKET_SCOPES:
            raise ValueError(f"scope must be one of {sorted(MARKET_SCOPES)}; got {scope!r}")
        shown = [r for r in shown if r["scope"] == scope]
    elif not include_agnostic and domain is None:
        shown = [r for r in shown if r["scope"] != "market_agnostic"]

    rank = {s: i for i, s in enumerate(_ORDER)}
    shown.sort(
        key=lambda r: (r["domain"], rank.get(r["scope"], 99), r["market"], r["instance"])
    )

    has_region = any(r["region"] for r in rows)
    n_coupled = sum(1 for r in rows if r["coupling"] == "loop")

    domain_counts = {}
    for r in rows:
        domain_counts[r["domain"]] = domain_counts.get(r["domain"], 0) + 1

    lines = [f"{type(self).__name__}: {len(rows)} disciplines"]
    lines.append("")
    lines.append("Domain summary:")
    dom_width = max((len(d) for d in domain_counts), default=0)
    for d in sorted(domain_counts, key=lambda d: (-domain_counts[d], d)):
        lines.append(f"  {d:<{dom_width}}{domain_counts[d]:>5}")
    lines.append("")
    lines.append("Market scope summary:")
    for s in _ORDER:
        lines.append(f"  {s:<16}{counts.get(s, 0):>4}")
    lines.append(f"  {'-' * 20}")
    lines.append(f"  {'total':<16}{len(rows):>4}")
    if coupling_available:
        lines.append("")
        lines.append(
            f"  in MDA feedback loop: {n_coupled} / {len(rows)} disciplines "
            f"(re-run every iteration); {len(rows) - n_coupled} feed-forward"
        )
    lines.append("")

    hidden = len(rows) - len(shown)
    if shown:
        # (key, header, max_width) — column widths are fitted to the data up to
        # max_width; longer cells are truncated with an ellipsis to keep alignment.
        cols = [
            ("domain", "DOMAIN", 30),
            ("scope", "SCOPE", 15),
            ("approach", "APPROACH", 10),
        ]
        if coupling_available:
            cols.append(("coupling", "COUPLING", 9))
        cols += [("model", "MODEL", 40), ("instance", "INSTANCE", 30)]
        if has_region:
            cols.append(("region", "REGION", 12))
        cols += [("market", "MARKET", 14), ("n_in", "#IN", 4), ("n_out", "#OUT", 4)]

        def _cell(value, width):
            text = str(value)
            if len(text) > width:
                text = text[: width - 1] + "…"
            return f"{text:<{width}}  "

        widths = {
            key: min(maxw, max(len(header), *(len(str(r[key])) for r in shown)))
            for key, header, maxw in cols
        }
        header = "  " + "".join(_cell(title, widths[key]) for key, title, _ in cols)
        lines.append(header)
        lines.append("  " + "-" * (len(header) - 2))
        for r in shown:
            lines.append("  " + "".join(_cell(r[key], widths[key]) for key, _, _ in cols))
        if coupling_available:
            lines.append("")
            lines.append(
                "  COUPLING: 'loop' = inside the MDA feedback cycle (cost scales with "
                "iterations); 'feed-fwd' = executed once."
            )
    if hidden and scope is None and domain is None and not include_agnostic:
        lines.append("")
        lines.append(
            f"  ({hidden} market_agnostic disciplines hidden; "
            f"pass include_agnostic=True to show)"
        )

    output = "\n".join(lines)
    if display:
        print(output)
    return output

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(), *available_plots_fleet.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,
    fig=None,
    ax=None,
    legend=True,
    **kwargs,
)

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`
`fig`	`Figure`	Existing figure to draw into. If provided together with `ax`, no new figure/axes are created.	`None`
`ax`	`Axes`	Existing axes to draw into. Must be provided together with `fig`.	`None`
`legend`	`bool or str`	Controls the legend. `True` (default) keeps the legend as created by the plot. `False` hides it. A string value (e.g. `"upper right"`) moves the legend to the given location.	`True`

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,
    fig=None,
    ax=None,
    legend=True,
    **kwargs,
):
    """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.
    fig : matplotlib.figure.Figure, optional
        Existing figure to draw into. If provided together with ``ax``,
        no new figure/axes are created.
    ax : matplotlib.axes.Axes, optional
        Existing axes to draw into. Must be provided together with ``fig``.
    legend : bool or str, optional
        Controls the legend. ``True`` (default) keeps the legend as created
        by the plot. ``False`` hides it. A string value (e.g. ``"upper right"``)
        moves the legend to the given location.

    Returns
    -------
    fig
        Object holding the created plot, as returned by the plot
        function.
    """
    # Extra keywords reach the plot class, so options a plot exposes on its
    # constructor (mfsp_type, or the decomposition's anchors) can be set from
    # a live process. Matches ResultsView.plot, which already forwards them.
    plot_kwargs = dict(fig=fig, ax=ax, legend=legend, **kwargs)
    if name in available_plots_fleet:
        try:
            fig_obj = available_plots_fleet[name](self, **plot_kwargs)
            if save:
                if size_inches is not None:
                    fig_obj.fig.set_size_inches(size_inches)
                if remove_title:
                    fig_obj.fig.gca().set_title("")
                fig_obj.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_obj = available_plots[name](self, **plot_kwargs)
        if save:
            if size_inches is not None:
                fig_obj.fig.set_size_inches(size_inches)
            if remove_title:
                fig_obj.fig.gca().set_title("")
            fig_obj.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_obj

aeromaps.core.process¶

AeroMAPSProcess ¶

common_setup ¶

setup_mda ¶

setup_optimisation ¶

create_gemseo_scenario ¶

create_gemseo_bilevel ¶

compute ¶

get_dataframes ¶

get_json ¶

write_json ¶

write_excel ¶

generate_n2 ¶

describe_models ¶

list_available_plots ¶

list_float_inputs ¶

list_str_inputs ¶

plot ¶

`aeromaps.core.process`¶