aeromaps.core.gemseo

A discipline interfacing a Python function.

CustomDataConverter

Bases: SimpleGrammarDataConverter

get_value_size classmethod

get_value_size(name, value)

Return the size of a data value.

The size is typically what is returned by ndarray.size or len(list). The size of a number is 1.

Args: name: The data name. value: The data value to get the size from.

Returns: The size.

Source code in aeromaps/core/gemseo.py

@classmethod
def get_value_size(cls, name: str, value: ValueType) -> int:
    """Return the size of a data value.

    The size is typically what is returned by ``ndarray.size`` or ``len(list)``.
    The size of a number is 1.


    Args:
        name: The data name.
        value: The data value to get the size from.

    Returns:
        The size.
    """
    if isinstance(value, cls._NON_ARRAY_TYPES):
        return 1
    elif isinstance(value, (list, tuple)):
        return len(value)
    return cast("NumberArray", value).size

AeroMAPSAutoModelWrapper

AeroMAPSAutoModelWrapper(model)

Bases: AutoPyDiscipline

Wraps the AeroMAPSModel class into a discipline. Inputs and outputs are automatically declared from the model's compute() function signature.

Source code in aeromaps/core/gemseo.py

def __init__(self, model):
    self.model: AeroMAPSModel = model

    self.default_grammar_type = Discipline.GrammarType.SIMPLE

    super(AeroMAPSAutoModelWrapper, self).__init__(
        py_func=self.model.compute,
    )
    # self.io.data_processor = AutoDiscDataProcessor()

    self.name = model.__class__.__name__

    self.update_defaults()

AeroMAPSCustomModelWrapper

AeroMAPSCustomModelWrapper(model)

Bases: Discipline

Wraps the AeroMAPSModel class into a discipline. Inputs and outputs are declared through the attributes 'input_names' and 'output_names' of the model.

Source code in aeromaps/core/gemseo.py

def __init__(self, model):
    super().__init__()

    # Whether to skip data type validation (at your own risk)
    if getattr(model, "_skip_data_type_validation", False):
        # self.input_grammar = SimplerGrammar("InputGrammar")
        # self.output_grammar = SimplerGrammar("OutputGrammar")
        self.input_grammar._validate = lambda data, msg: True
        self.output_grammar._validate = lambda data, msg: True

    # Set input and output grammars from model attributes
    if isinstance(model.input_names, dict):
        self.input_grammar.update_from_data(model.input_names)
    else:  # assume list of names
        self.input_grammar.update_from_names(model.input_names)

    if isinstance(model.output_names, dict):
        self.output_grammar.update_from_data(model.output_names)
    else:  # assume list of names
        self.output_grammar.update_from_names(model.output_names)

    # Set the model
    self.model: AeroMAPSModel = model
    self.name = model.__class__.__name__

    # Initialize default input data
    self.update_defaults()

disable_gemseo_execution_statistics

disable_gemseo_execution_statistics()

Disable GEMSEO's execution statistics shared memory.

GEMSEO's ExecutionStatistics creates semaphores for each discipline via multiprocessing.Value(). With many disciplines (20+ regions × 50+ models), this exhausts macOS semaphore limits (kern.sysv.shmmni=32).

This function patches ExecutionStatistics to use regular Python attributes instead of shared memory, avoiding semaphore creation.

Safe to call multiple times (only patches once).

TODO: Investigate possibilities with gemseo.configure ?

Source code in aeromaps/core/gemseo.py

def disable_gemseo_execution_statistics():
    """Disable GEMSEO's execution statistics shared memory.

    GEMSEO's ExecutionStatistics creates semaphores for each discipline via
    multiprocessing.Value(). With many disciplines (20+ regions × 50+ models),
    this exhausts macOS semaphore limits (kern.sysv.shmmni=32).

    This function patches ExecutionStatistics to use regular Python attributes
    instead of shared memory, avoiding semaphore creation.

    Safe to call multiple times (only patches once).
    # TODO: Investigate possibilities with gemseo.configure ?
    """
    global _EXECUTION_STATISTICS_PATCHED

    if _EXECUTION_STATISTICS_PATCHED:
        return

    try:
        from gemseo.core.execution_statistics import ExecutionStatistics

        def _patched_init(self, *args, **kwargs):
            """Skip shared memory initialization to avoid semaphore exhaustion."""
            # Initialize as regular attributes instead of shared memory
            self._ExecutionStatistics__duration = 0.0
            self._ExecutionStatistics__n_executions = 0
            self._ExecutionStatistics__n_linearizations = 0
            self._ExecutionStatistics__n_calls_to_jacobian = 0
            self._ExecutionStatistics__execution_time = {}

        ExecutionStatistics._init_shared_memory_attrs_before = _patched_init
        _EXECUTION_STATISTICS_PATCHED = True
        LOGGER.debug("Patched GEMSEO ExecutionStatistics to use non-shared state")
    except Exception as patch_err:
        LOGGER.warning(f"Could not patch GEMSEO ExecutionStatistics: {patch_err}")

apply_namespace_to_discipline

apply_namespace_to_discipline(discipline, namespace)

Apply a namespace prefix to all inputs and outputs of a GEMSEO discipline.

This function creates a deep copy of the discipline and applies the namespace to isolate regional I/O variables. This is essential for multi-regional scenarios where multiple instances of the same discipline must coexist without variable conflicts.

Parameters:

Name	Type	Description	Default
discipline	Discipline	The GEMSEO discipline to namespace. Will be deep-copied to avoid modifying the original.	required
namespace	str	The namespace prefix to apply (e.g., "FR", "DE"). Variables will be renamed from "var_name" to "{namespace}:var_name".	required

Returns:

Type	Description
Discipline	A new discipline instance with namespaced inputs and outputs.

Notes

The deep copy is necessary because add_namespace_to_input/output() modifies the discipline in place. Without it, we would modify the original disciplines stored in the process objects.

Examples:

>>> from aeromaps.core.gemseo import apply_namespace_to_discipline
>>> namespaced_disc = apply_namespace_to_discipline(discipline, "FR")
>>> # Now inputs/outputs are prefixed: "co2_emissions" -> "FR:co2_emissions"

Source code in aeromaps/core/gemseo.py

def apply_namespace_to_discipline(discipline: Discipline, namespace: str) -> Discipline:
    """Apply a namespace prefix to all inputs and outputs of a GEMSEO discipline.

    This function creates a deep copy of the discipline and applies the namespace
    to isolate regional I/O variables. This is essential for multi-regional scenarios
    where multiple instances of the same discipline must coexist without variable conflicts.

    Parameters
    ----------
    discipline
        The GEMSEO discipline to namespace. Will be deep-copied to avoid
        modifying the original.
    namespace
        The namespace prefix to apply (e.g., "FR", "DE"). Variables will be
        renamed from "var_name" to "{namespace}:var_name".

    Returns
    -------
    Discipline
        A new discipline instance with namespaced inputs and outputs.

    Notes
    -----
    The deep copy is necessary because `add_namespace_to_input/output()` modifies
    the discipline in place. Without it, we would modify the original disciplines
    stored in the process objects.

    Examples
    --------
    >>> from aeromaps.core.gemseo import apply_namespace_to_discipline
    >>> namespaced_disc = apply_namespace_to_discipline(discipline, "FR")
    >>> # Now inputs/outputs are prefixed: "co2_emissions" -> "FR:co2_emissions"
    """
    from copy import deepcopy

    ns_disc = deepcopy(discipline)

    # Collect all variable names first for consistency
    input_names = list(ns_disc.input_grammar.names)
    output_names = list(ns_disc.output_grammar.names)

    # Apply namespace to all inputs and outputs
    for name in input_names:
        ns_disc.add_namespace_to_input(name, namespace)
    for name in output_names:
        ns_disc.add_namespace_to_output(name, namespace)

    # Update discipline name to include region identifier
    ns_disc.name = f"{namespace}_{ns_disc.name}"

    return ns_disc

apply_namespace_to_disciplines

apply_namespace_to_disciplines(disciplines, namespace)

Apply a namespace to a list of disciplines.

Convenience function to namespace multiple disciplines at once.

Parameters:

Name	Type	Description	Default
disciplines	list[Discipline]	List of GEMSEO disciplines to namespace.	required
namespace	str	The namespace prefix to apply.	required

Returns:

Type	Description
list[Discipline]	List of new discipline instances with namespaced I/O.

Source code in aeromaps/core/gemseo.py

def apply_namespace_to_disciplines(
    disciplines: list[Discipline], namespace: str
) -> list[Discipline]:
    """Apply a namespace to a list of disciplines.

    Convenience function to namespace multiple disciplines at once.

    Parameters
    ----------
    disciplines
        List of GEMSEO disciplines to namespace.
    namespace
        The namespace prefix to apply.

    Returns
    -------
    list[Discipline]
        List of new discipline instances with namespaced I/O.
    """
    return [apply_namespace_to_discipline(d, namespace) for d in disciplines]

build_namespaced_inputs

build_namespaced_inputs(parameters, namespace)

Build a namespaced input dictionary from an AeroMAPS parameters object.

This function converts all parameters into a dictionary with namespaced keys, suitable for execution of a multi-regional MDAChain.

Parameters:

Name	Type	Description	Default
parameters		AeroMAPS Parameters object containing model inputs.	required
namespace	str	The namespace prefix to apply to all parameter names.	required

Returns:

Type	Description
dict	Dictionary with namespaced keys mapping to parameter values. E.g., {"FR:rpk_init": , "FR:energy_consumption_init": , ...}

Source code in aeromaps/core/gemseo.py

def build_namespaced_inputs(parameters, namespace: str) -> dict:
    """Build a namespaced input dictionary from an AeroMAPS parameters object.

    This function converts all parameters into a dictionary with namespaced keys,
    suitable for execution of a multi-regional MDAChain.

    Parameters
    ----------
    parameters
        AeroMAPS Parameters object containing model inputs.
    namespace
        The namespace prefix to apply to all parameter names.

    Returns
    -------
    dict
        Dictionary with namespaced keys mapping to parameter values.
        E.g., {"FR:rpk_init": <value>, "FR:energy_consumption_init": <value>, ...}
    """
    input_data = {}
    params_dict = parameters.to_dict()

    for key, value in params_dict.items():
        namespaced_key = f"{namespace}:{key}"
        input_data[namespaced_key] = value

    return input_data