# 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, Matthias De Lozzo
# OTHER AUTHORS - MACROSCOPIC CHANGES
"""Caching module to store only one entry."""
from __future__ import annotations
import logging
from copy import deepcopy
from typing import Generator
from typing import Iterable
from gemseo.core.cache import AbstractCache
from gemseo.core.cache import CacheEntry
from gemseo.core.cache import Data
from gemseo.core.cache import JacobianData
from gemseo.utils.data_conversion import deepcopy_dict_of_arrays
from gemseo.utils.testing import compare_dict_of_arrays
LOGGER = logging.getLogger(__name__)
[docs]class SimpleCache(AbstractCache):
"""Dictionary-based cache storing a unique entry.
When caching an input data different from this entry, this entry is replaced by a
new one initialized with this input data.
"""
def __init__(
self,
tolerance: float = 0.0,
name: str | None = None,
) -> None:
super().__init__(tolerance, name)
self.__input_data_for_outputs = {}
self.__output_data = {}
self.__input_data_for_jacobian = {}
self.__jacobian_data = {}
self.__last_input_data = {}
self.__penultimate_input_data = {}
[docs] def clear(self) -> None:
super().clear()
self.__input_data_for_outputs = {}
self.__output_data = {}
self.__input_data_for_jacobian = {}
self.__jacobian_data = {}
self.__last_input_data = {}
self.__penultimate_input_data = {}
def __iter__(self) -> Generator[CacheEntry]:
if self.__penultimate_input_data:
yield self.__penultimate_input_data
yield self.last_entry
def __len__(self) -> int:
return bool(self.__penultimate_input_data) + bool(self.__last_input_data)
def __create_input_cache(
self,
input_data: Data,
output_names: Iterable[str] | None = None,
) -> Data:
"""Create the input data.
Args:
input_data: The data containing the input data to cache.
output_names: The names of the outputs to cache.
If ``None``, consider all the outputs.
Returns:
A copy of the input data.
"""
if output_names is None:
cached_input_data = deepcopy_dict_of_arrays(input_data)
else:
cached_input_data = {k: v for k, v in input_data.items()}
# If also an output, keep a copy of the original input value
for output_name in output_names:
input_value = input_data.get(output_name)
if input_value is not None:
cached_input_data[output_name] = deepcopy(input_value)
if not self.__is_cached(self.__last_input_data, cached_input_data):
self.__penultimate_input_data = self.__last_input_data
self.__last_input_data = cached_input_data
return cached_input_data
def __is_cached(
self,
cached_input_data: Data,
input_data: Data,
) -> bool:
"""Check if an input data is cached.
Args:
cached_input_data: The cached input data.
input_data: The input data to be verified.
Returns:
Whether the input data is cached.
"""
if not cached_input_data:
return False
if self.tolerance == 0.0:
if compare_dict_of_arrays(input_data, cached_input_data):
return True
else:
if compare_dict_of_arrays(input_data, cached_input_data, self.tolerance):
return True
return False
[docs] def cache_outputs(
self,
input_data: Data,
output_data: Data,
) -> None:
self.__input_data_for_outputs = self.__create_input_cache(input_data)
self.__output_data = deepcopy_dict_of_arrays(output_data)
def __getitem__(
self,
input_data: Data,
) -> CacheEntry:
output_data, jacobian_data = {}, {}
if self.__is_cached(self.__input_data_for_outputs, input_data):
output_data = self.__output_data
if self.__is_cached(self.__input_data_for_jacobian, input_data):
jacobian_data = self.__jacobian_data
return CacheEntry(input_data, output_data, jacobian_data)
[docs] def cache_jacobian(
self,
input_data: Data,
jacobian_data: JacobianData,
) -> None:
self.__input_data_for_jacobian = self.__create_input_cache(input_data)
self.__jacobian_data = jacobian_data
def __retrieve_entry(
self,
cached_input_data: Data,
) -> CacheEntry:
"""Return the cache entry corresponding to a cached input data.
Args:
cached_input_data: The cached input data.
Returns:
The cache entry corresponding to this cached input data.
"""
input_data = {}
output_data = {}
jacobian_data = {}
if self.__is_cached(self.__input_data_for_outputs, cached_input_data):
input_data = self.__input_data_for_outputs
output_data = self.__output_data
if self.__is_cached(self.__input_data_for_jacobian, cached_input_data):
input_data = self.__input_data_for_jacobian
jacobian_data = self.__jacobian_data
return CacheEntry(input_data, output_data, jacobian_data)
@property
def penultimate_entry(self) -> CacheEntry:
"""The penultimate cache entry."""
return self.__retrieve_entry(self.__penultimate_input_data)
@property
def last_entry(self) -> CacheEntry:
return self.__retrieve_entry(self.__last_input_data)
