Source code for gemseo.disciplines.concatenater

# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
# Contributors:
#    INITIAL AUTHORS - API and implementation and/or documentation
#        :author: Jean-Christophe Giret
#    OTHER AUTHORS   - MACROSCOPIC CHANGES
"""The concatenation of several input variables into a single one."""

from __future__ import annotations

from itertools import accumulate
from typing import TYPE_CHECKING

from numpy import concatenate
from scipy.sparse import csr_array

from gemseo.core.discipline import MDODiscipline

if TYPE_CHECKING:
    from collections.abc import Sequence


[docs] class Concatenater(MDODiscipline): """Concatenate input variables into a single output variable. These input variables can be scaled before concatenation. Examples: >>> from gemseo import create_discipline >>> sellar_system_disc = create_discipline("SellarSystem") >>> constraint_names = ["c1", "c2"] >>> output_name = ["c"] >>> concatenation_disc = create_discipline( ... "Concatenater", constraint_names, output_name ... ) >>> disciplines = [sellar_system_disc, concatenation_disc] >>> chain = create_discipline("MDOChain", disciplines=disciplines) >>> print(chain.execute()) >>> print(chain.linearize(compute_all_jacobians=True)) """ def __init__( self, input_variables: Sequence[str], output_variable: str, input_coefficients: dict[str, float] | None = None, ) -> None: """ Args: input_variables: The input variables to concatenate. output_variable: The output variable name. input_coefficients: The coefficients related to the different input variables. """ # noqa: D205 D212 D415 super().__init__() self.input_grammar.update_from_names(input_variables) self.output_grammar.update_from_names([output_variable]) self.__coefficients = dict.fromkeys(input_variables, 1.0) if input_coefficients: self.__coefficients.update(input_coefficients) self.__output_name = output_variable def _run(self) -> None: """Run the discipline.""" input_data = self.get_input_data() self.local_data[self.__output_name] = concatenate([ self.__coefficients[input_name] * input_data[input_name] for input_name in self.get_input_data_names() ]) def _compute_jacobian( self, inputs: Sequence[str] | None = None, outputs: Sequence[str] | None = None, ) -> None: """Compute the jacobian matrix. Args: inputs: The linearization should be performed with respect to inputs list. If ``None``, linearization should be performed wrt all inputs (Default value = None) outputs: The linearization should be performed on outputs list. If ``None``, linearization should be performed on all outputs (Default value = None) """ self._init_jacobian(inputs, outputs, init_type=self.InitJacobianType.SPARSE) input_names = self.get_input_data_names() input_sizes = [input_.size for input_ in self.get_all_inputs()] total_size = sum(input_sizes) # Instead of manually accumulating, we use the accumulate() iterator. jac = self.jac[self.__output_name] for name, size, start in zip( input_names, input_sizes, accumulate(input_sizes, initial=0) ): jac[name] = csr_array( ( [self.__coefficients[name]] * size, (range(start, start + size), range(size)), ), shape=(total_size, size), )