"""
This file is part of CLIMADA.
Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS.
CLIMADA is free software: you can redistribute it and/or modify it under the
terms of the GNU General Public License as published by the Free
Software Foundation, version 3.
CLIMADA 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 General Public License for more details.
You should have received a copy of the GNU General Public License along
with CLIMADA. If not, see <https://www.gnu.org/licenses/>.
---
Calibration Base Classes and Interfaces
"""
from abc import ABC, abstractmethod
from dataclasses import InitVar, dataclass, field
from numbers import Number
from pathlib import Path
from typing import Any, Callable, Dict, Mapping, Optional, Tuple, Union
import h5py
import numpy as np
import pandas as pd
import seaborn as sns
from scipy.optimize import Bounds, LinearConstraint, NonlinearConstraint
from climada.engine import Impact, ImpactCalc
from climada.entity import Exposures, ImpactFuncSet
from climada.hazard import Hazard
ConstraintType = Union[LinearConstraint, NonlinearConstraint, Mapping]
[docs]
@dataclass
class Output:
"""Generic output of a calibration task
Attributes
----------
params : Mapping (str, Number)
The optimal parameters
target : Number
The target function value for the optimal parameters
"""
params: Mapping[str, Number]
target: Number
[docs]
def to_hdf5(self, filepath: Union[Path, str], mode: str = "x"):
"""Write the output into an H5 file
This stores the data as attributes because we only store single numbers, not
arrays
Parameters
----------
filepath : Path or str
The filepath to store the data.
mode : str (optional)
The mode for opening the file. Defaults to ``x`` (Create file, fail if
exists).
"""
with h5py.File(filepath, mode=mode) as file:
# Store target
grp = file.create_group("base")
grp.attrs["target"] = self.target
# Store params
grp_params = grp.create_group("params")
for p_name, p_val in self.params.items():
grp_params.attrs[p_name] = p_val
[docs]
@classmethod
def from_hdf5(cls, filepath: Union[Path, str]):
"""Create an output object from an H5 file"""
with h5py.File(filepath) as file:
target = file["base"].attrs["target"]
params = dict(file["base"]["params"].attrs.items())
return cls(params=params, target=target)
[docs]
@dataclass
class OutputEvaluator:
"""Evaluate the output of a calibration task
Parameters
----------
input : Input
The input object for the optimization task.
output : Output
The output object returned by the optimization task.
Attributes
----------
impf_set : climada.entity.ImpactFuncSet
The impact function set built from the optimized parameters
impact : climada.engine.Impact
An impact object calculated using the optimal :py:attr:`impf_set`
"""
input: Input
output: Output
def __post_init__(self):
"""Compute the impact for the optimal parameters"""
self.impf_set = self.input.impact_func_creator(**self.output.params)
self.impact = ImpactCalc(
exposures=self.input.exposure,
impfset=self.impf_set,
hazard=self.input.hazard,
).impact(assign_centroids=True, save_mat=True)
self._impact_label = f"Impact [{self.input.exposure.value_unit}]"
[docs]
def plot_at_event(
self,
data_transf: Callable[[pd.DataFrame], pd.DataFrame] = lambda x: x,
**plot_kwargs,
):
"""Create a bar plot comparing estimated model output and data per event.
Every row of the :py:attr:`Input.data` is considered an event.
The data to be plotted can be transformed with a generic function
``data_transf``.
Parameters
----------
data_transf : Callable (pd.DataFrame -> pd.DataFrame), optional
A function that transforms the data to plot before plotting.
It receives a dataframe whose rows represent events and whose columns
represent the modelled impact and the calibration data, respectively.
By default, the data is not transformed.
plot_kwargs
Keyword arguments passed to the ``DataFrame.plot.bar`` method.
Returns
-------
ax : matplotlib.axes.Axes
The plot axis returned by ``DataFrame.plot.bar``
Note
----
This plot does *not* include the ignored impact, see :py:attr:`Input.data`.
"""
data, impact = self.input.impact_to_aligned_df(self.impact)
values = pd.concat(
[impact.sum(axis="columns"), data.sum(axis="columns")],
axis=1,
).rename(columns={0: "Model", 1: "Data"})
# Transform data before plotting
values = data_transf(values)
# Now plot
ylabel = plot_kwargs.pop("ylabel", self._impact_label)
return values.plot.bar(ylabel=ylabel, **plot_kwargs)
[docs]
def plot_at_region(
self,
data_transf: Callable[[pd.DataFrame], pd.DataFrame] = lambda x: x,
**plot_kwargs,
):
"""Create a bar plot comparing estimated model output and data per event
Every column of the :py:attr:`Input.data` is considered a region.
The data to be plotted can be transformed with a generic function
``data_transf``.
Parameters
----------
data_transf : Callable (pd.DataFrame -> pd.DataFrame), optional
A function that transforms the data to plot before plotting.
It receives a dataframe whose rows represent regions and whose columns
represent the modelled impact and the calibration data, respectively.
By default, the data is not transformed.
plot_kwargs
Keyword arguments passed to the ``DataFrame.plot.bar`` method.
Returns
-------
ax : matplotlib.axes.Axes
The plot axis returned by ``DataFrame.plot.bar``.
Note
----
This plot does *not* include the ignored impact, see :py:attr:`Input.data`.
"""
data, impact = self.input.impact_to_aligned_df(self.impact)
values = pd.concat(
[impact.sum(axis="index"), data.sum(axis="index")],
axis=1,
).rename(columns={0: "Model", 1: "Data"})
# Transform data before plotting
values = data_transf(values)
# Now plot
ylabel = plot_kwargs.pop("ylabel", self._impact_label)
return values.plot.bar(ylabel=ylabel, **plot_kwargs)
[docs]
def plot_event_region_heatmap(
self,
data_transf: Callable[[pd.DataFrame], pd.DataFrame] = lambda x: x,
**plot_kwargs,
):
"""Plot a heatmap comparing all events per all regions
Every column of the :py:attr:`Input.data` is considered a region, and every
row is considered an event.
The data to be plotted can be transformed with a generic function
``data_transf``.
Parameters
----------
data_transf : Callable (pd.DataFrame -> pd.DataFrame), optional
A function that transforms the data to plot before plotting.
It receives a dataframe whose rows represent events and whose columns
represent the regions, respectively.
By default, the data is not transformed.
plot_kwargs
Keyword arguments passed to the ``DataFrame.plot.bar`` method.
Returns
-------
ax : matplotlib.axes.Axes
The plot axis returned by ``DataFrame.plot.bar``.
"""
# Data preparation
data, impact = self.input.impact_to_aligned_df(self.impact)
values = (impact + 1) / (data + 1) # Avoid division by zero
values = values.transform(np.log10)
# Transform data
values = data_transf(values)
# Default plot settings
annot = plot_kwargs.pop("annot", True)
vmax = plot_kwargs.pop("vmax", 3)
vmin = plot_kwargs.pop("vmin", -vmax)
center = plot_kwargs.pop("center", 0)
fmt = plot_kwargs.pop("fmt", ".1f")
cmap = plot_kwargs.pop("cmap", "RdBu_r")
cbar_kws = plot_kwargs.pop(
"cbar_kws", {"label": r"Model Error $\log_{10}(\mathrm{Impact})$"}
)
return sns.heatmap(
values,
annot=annot,
vmin=vmin,
vmax=vmax,
center=center,
fmt=fmt,
cmap=cmap,
cbar_kws=cbar_kws,
**plot_kwargs,
)
[docs]
@dataclass
class Optimizer(ABC):
"""Abstract base class (interface) for an optimization
This defines the interface for optimizers in CLIMADA. New optimizers can be created
by deriving from this class and overriding at least the :py:meth:`run` method.
Attributes
----------
input : Input
The input object for the optimization task. See :py:class:`Input`.
"""
input: Input
[docs]
def _target_func(self, data: pd.DataFrame, predicted: pd.DataFrame) -> Number:
"""Target function for the optimizer
The default version of this function simply returns the value of the cost
function evaluated on the arguments.
Parameters
----------
data : pandas.DataFrame
The reference data used for calibration. By default, this is
:py:attr:`Input.data`.
predicted : pandas.DataFrame
The impact predicted by the data calibration after it has been transformed
into a dataframe by :py:attr:`Input.impact_to_dataframe`.
Returns
-------
The value of the target function for the optimizer.
"""
return self.input.cost_func(data, predicted)
[docs]
def _kwargs_to_impact_func_creator(self, *_, **kwargs) -> Dict[str, Any]:
"""Define how the parameters to :py:meth:`_opt_func` must be transformed
Optimizers may implement different ways of representing the parameters (e.g.,
key-value pairs, arrays, etc.). Depending on this representation, the parameters
must be transformed to match the syntax of the impact function generator used,
see :py:attr:`Input.impact_func_creator`.
In this default version, the method simply returns its keyword arguments as
mapping. Override this method if the optimizer used *does not* represent
parameters as key-value pairs.
Parameters
----------
kwargs
The parameters as key-value pairs.
Returns
-------
The parameters as key-value pairs.
"""
return kwargs
[docs]
def _opt_func(self, *args, **kwargs) -> Number:
"""The optimization function iterated by the optimizer
This function takes arbitrary arguments from the optimizer, generates a new set
of impact functions from it, computes the impact, and finally calculates the
target function value and returns it.
Parameters
----------
args, kwargs
Arbitrary arguments from the optimizer, including parameters
Returns
-------
Target function value for the given arguments
"""
# Create the impact function set from a new parameter estimate
params = self._kwargs_to_impact_func_creator(*args, **kwargs)
impf_set = self.input.impact_func_creator(**params)
# Compute the impact
impact = ImpactCalc(
exposures=self.input.exposure,
impfset=impf_set,
hazard=self.input.hazard,
).impact(**self.input.impact_calc_kwds)
# Transform to DataFrame, align, and compute target function
data_aligned, impact_df_aligned = self.input.impact_to_aligned_df(
impact, fillna=0
)
return self._target_func(data_aligned, impact_df_aligned)
[docs]
@abstractmethod
def run(self, **opt_kwargs) -> Output:
"""Execute the optimization"""
