Source code for gemseo.core.data_processor

# 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 - initial API and implementation and/or initial
#                         documentation
#        :author: Francois Gallard
#    OTHER AUTHORS   - MACROSCOPIC CHANGES
"""
Data conversion between discipline data check and _run()
********************************************************
"""
from __future__ import annotations

import logging

from numpy import array
from numpy import complex128

LOGGER = logging.getLogger(__name__)


[docs]class DataProcessor: """Abstract class for pre and post processing data of MDODisciplines. Executes a pre processing of input data after they are checked by MDODiscipline.check_data, and before the _run method of the discipline is called. Similarly, the post processing method is executed after the _run method and before the output data is checked by |g| This is usefull to cast the data types, since the |g| kernel only handles numpy arrays, and interfaces to disciplinary tools, workflow engines or environment can require different types """
[docs] def pre_process_data(self, data): """Executes a pre processing of input data after they are checked by MDODiscipline.check_data, and before the _run method of the discipline is called. Args: data: The input data to process. Returns: The processed input data. """ raise NotImplementedError()
[docs] def post_process_data(self, data): """Executes a post processing of discipline output data after the _run method of the discipline, before they are checked by MDODiscipline.check_output_data, Args: data: The output data to process. Returns: The processed output data. """ raise NotImplementedError()
[docs]class FloatDataProcessor(DataProcessor): """A data preprocessor that converts all gemseo scalar input data to floats, and converts all discipline output data to numpy arrays."""
[docs] def pre_process_data(self, data): """Executes a pre processing of input data after they are checked by MDODiscipline.check_data, and before the _run method of the discipline is called. Args: data: The input data to process. Returns: The processed input data. """ processed_data = data.copy() for key, val in data.items(): if len(val) == 1: processed_data[key] = float(val[0]) else: processed_data[key] = [float(val_i) for val_i in val] return processed_data
[docs] def post_process_data(self, data): """Executes a post processing of discipline output data after the _run method of the discipline, before they are checked by MDODiscipline.check_output_data, Args: data: The output data to process. Returns: The processed output data. """ processed_data = data.copy() for key, val in data.items(): if not hasattr(val, "__len__"): processed_data[key] = array([val]) else: processed_data[key] = array(val) return processed_data
[docs]class ComplexDataProcessor(DataProcessor): """A data preprocessor that converts all gemseo complex arrays input data to floats arrays, and converts all discipline output data to numpy complex arrays."""
[docs] def pre_process_data(self, data): """Executes a pre processing of input data after they are checked by MDODiscipline.check_data, and before the _run method of the discipline is called. Args: data: The input data to process. Returns: The processed input data. """ processed_data = data.copy() for key, val in data.items(): processed_data[key] = array(val.real) return processed_data
[docs] def post_process_data(self, data): """Executes a post processing of discipline output data after the _run method of the discipline, before they are checked by MDODiscipline.check_output_data, Args: data: The output data to process. Returns: The processed output data. """ processed_data = data.copy() for key, val in data.items(): processed_data[key] = array(val, dtype=complex128) return processed_data
[docs]class NameMapping(DataProcessor): """A data preprocessor that maps process level data names to local discipline data names.""" def __init__(self, mapping): """ Args: mapping: A mapping structure of the form ``{global_name: local_name}`` where ``global_name`` must be consistent with the grammar of the discipline. The local name is the data provided to the :meth:`.MDODiscipline._run` method. """ super().__init__() self.mapping = mapping self.reverse_mapping = { local_key: global_key for global_key, local_key in mapping.items() }
[docs] def pre_process_data(self, data): """Executes a pre processing of input data after they are checked by MDODiscipline.check_data, and before the _run method of the discipline is called. Args: data: The input data to process. Returns: The processed input data. """ mapping = self.mapping return {mapping[global_key]: value for global_key, value in data.items()}
[docs] def post_process_data(self, data): """Executes a post processing of discipline output data after the _run method of the discipline, before they are checked by MDODiscipline.check_output_data, Args: data: The output data to process. Returns: The processed output data. """ reverse_mapping = self.reverse_mapping return {reverse_mapping[local_key]: value for local_key, value in data.items()}