openscm_calibration.store.base

Base implementation of store of results

Classes:

Name	Description
OptResStore	Store for results during optimisation
SupportsListLikeHandling	Class that supports handling a list-like

OptResStore

Bases: Generic[DataContainer]

Store for results during optimisation

Methods:

Name	Description
append_result_cost_x	Append result, cost and x from a successful run to the results
from_n_runs	Initialise based on expected number of runs
from_n_runs_manager	Initialise based on expected number of runs for use in parallel work
get_available_index	Get an available index to write into
get_costs_labelled_xsamples_res	Get costs, x_samples and res from runs
get_costs_xsamples_res	Get costs, x_samples and res from runs
note_failed_run	Note that a run failed
set_result_cost_x	Set result, cost and x at a given index

Attributes:

Name	Type	Description
add_iteration_to_res	Callable[[DataContainer, int], DataContainer]	Add iteration information to the results of the run
available_indices	MutableSequence[int]	Indices available to be written into
costs	MutableSequence[None | float]	Costs of runs
params	tuple[str]	Names of the parameters being stored in x_samples
res	MutableSequence[None | DataContainer]	Results of runs
x_samples	MutableSequence[None | NDArray[number[Any]]]	x vectors sampled

Source code in src/openscm_calibration/store/base.py

@define
class OptResStore(Generic[DataContainer]):
    """
    Store for results during optimisation
    """

    res: MutableSequence[None | DataContainer] = field(validator=[_all_none_to_start])
    """Results of runs"""

    costs: MutableSequence[None | float] = field(
        validator=[_all_none_to_start, _same_length_as_res]
    )
    """Costs of runs"""

    x_samples: MutableSequence[None | np.typing.NDArray[np.number[Any]]] = field(
        validator=[_all_none_to_start, _same_length_as_res]
    )
    """x vectors sampled"""

    params: tuple[str]
    """Names of the parameters being stored in `x_samples`"""

    available_indices: MutableSequence[int] = field(
        validator=[_same_length_as_res, _contains_indices_in_res]
    )
    """Indices available to be written into"""

    add_iteration_to_res: Callable[[DataContainer, int], DataContainer]
    """
    Add iteration information to the results of the run

    This has to be done at run time, because we don't know the iteration in advance
    (the iteration information comes from `self.get_available_index`).
    """

    @classmethod
    def from_n_runs(
        cls,
        n_runs: int,
        params: tuple[str],
        add_iteration_to_res: Callable[[DataContainer, int], DataContainer],
    ) -> OptResStore[Any]:
        """
        Initialise based on expected number of runs

        Parameters
        ----------
        n_runs
            Expected number of runs

        params
            Names of the parameters that are being sampled

        add_iteration_to_res
            Function that adds iteration information to the result of a run.

            For further information, see the docstring of `self`.

        Returns
        -------
        :
            Initialised store
        """
        # Reverse so that using pop counts up
        available_indices = list(range(n_runs))[::-1]
        return cls(
            res=[None] * n_runs,
            costs=[None] * n_runs,
            x_samples=[None] * n_runs,
            params=params,
            available_indices=available_indices,
            add_iteration_to_res=add_iteration_to_res,
        )

    @classmethod
    def from_n_runs_manager(
        cls,
        n_runs: int,
        manager: SupportsListLikeHandling,
        params: tuple[str],
        add_iteration_to_res: Callable[[DataContainer, int], DataContainer],
    ) -> OptResStore[Any]:
        """
        Initialise based on expected number of runs for use in parallel work

        Parameters
        ----------
        n_runs
            Expected number of runs

        manager
            Manager of lists (e.g. [`multiprocess.managers.SyncManager`][])

        params
            Names of the parameters that are being sampled

        add_iteration_to_res
            Function that adds iteration information to the result of a run.

            For further information, see the docstring of `self`.

        Returns
        -------
        :
            Initialised store
        """
        # Reverse so that using pop counts up
        available_indices = list(range(n_runs))[::-1]

        return cls(
            res=manager.list([None] * n_runs),
            costs=manager.list([None] * n_runs),
            x_samples=manager.list([None] * n_runs),
            params=params,
            available_indices=manager.list(available_indices),
            add_iteration_to_res=add_iteration_to_res,
        )

    def get_available_index(self) -> int:
        """
        Get an available index to write into

        Returns
        -------
        :
            Available index. This index is now no longer considered available.
        """
        return self.available_indices.pop()

    def set_result_cost_x(
        self,
        res: None | DataContainer,
        cost: float,
        x: np.typing.NDArray[np.number[Any]],
        idx: int,
    ) -> None:
        """
        Set result, cost and x at a given index

        Parameters
        ----------
        res
            Result to append (use `None` for a failed run)

        cost
            Cost associated with the run

        x
            Parameter array associated with the run

        idx
            Index in `self.costs`, `self.x_samples` and `self.res` to write into
        """
        len_x = len(x)
        len_params = len(self.params)
        if len_x != len_params:
            raise MismatchLengthError(
                "x",
                length=len_x,
                expected_name="self.params",
                expected_length=len_params,
            )

        self.costs[idx] = cost
        self.x_samples[idx] = x
        self.res[idx] = res

    def append_result_cost_x(
        self,
        res: DataContainer,
        cost: float,
        x: np.typing.NDArray[np.number[Any]],
    ) -> None:
        """
        Append result, cost and x from a successful run to the results

        Parameters
        ----------
        res
            Result to append (use `None` for a failed run)

        cost
            Cost associated with the run

        x
            Parameter array associated with the run
        """
        iteration = self.get_available_index()
        # res_keep = res.copy()
        # res_keep["it"] = iteration
        res_keep = self.add_iteration_to_res(res, iteration)

        self.set_result_cost_x(
            res=res_keep,
            cost=cost,
            x=x,
            idx=iteration,
        )

    def note_failed_run(
        self,
        cost: float,
        x: np.typing.NDArray[np.number[Any]],
    ) -> None:
        """
        Note that a run failed

        Typically, `cost` will be `np.inf`.

        The cost and x parameters are appended to the results, as well as an
        indicator that the run was a failure in `self.res`.

        Parameters
        ----------
        cost
            Cost associated with the run

        x
            Parameter array associated with the run
        """
        iteration = self.get_available_index()
        self.set_result_cost_x(
            res=None,
            cost=cost,
            x=x,
            idx=iteration,
        )

    def get_costs_xsamples_res(
        self,
    ) -> tuple[
        tuple[float, ...],
        tuple[np.typing.NDArray[np.number[Any]], ...],
        tuple[DataContainer, ...],
    ]:
        """
        Get costs, x_samples and res from runs

        Returns
        -------
        :
            Costs, x_samples and res from all runs which were attempted
            (i.e. we include failed runs here)
        """
        # There may be a better algorithm for this, PRs welcome :)
        if all(x is None for x in self.x_samples):
            return ((), (), ())

        tmp = tuple(
            zip(
                *[
                    (
                        self.costs[i],
                        x,
                        self.res[i],
                    )
                    for i, x in enumerate(self.x_samples)
                    # x is only None if no run was attempted yet
                    if x is not None
                ]
            )
        )

        # Help out type hinting
        costs: tuple[float, ...] = tmp[0]
        xs_out: tuple[np.typing.NDArray[np.number[Any]], ...] = tmp[1]
        ress: tuple[DataContainer, ...] = tmp[2]

        out = (costs, xs_out, ress)

        return out

    def get_costs_labelled_xsamples_res(
        self,
    ) -> tuple[
        tuple[float, ...],
        dict[str, np.typing.NDArray[np.number[Any]]],
        tuple[DataContainer, ...],
    ]:
        """
        Get costs, x_samples and res from runs

        Returns
        -------
        :
            Costs, x_samples and res from all runs which were attempted
            (i.e. we include failed runs here)
        """
        unlabelled = self.get_costs_xsamples_res()
        if not any(unlabelled):
            return (
                unlabelled[0],
                {p: np.array([]) for p in self.params},
                unlabelled[2],
            )

        x_samples_stacked = np.vstack(unlabelled[1])
        xs_labelled = {p: x_samples_stacked[:, i] for i, p in enumerate(self.params)}

        out = (unlabelled[0], xs_labelled, unlabelled[2])

        return out

add_iteration_to_res instance-attribute

add_iteration_to_res: Callable[
    [DataContainer, int], DataContainer
]

Add iteration information to the results of the run

This has to be done at run time, because we don't know the iteration in advance (the iteration information comes from self.get_available_index).

available_indices class-attribute instance-attribute

available_indices: MutableSequence[int] = field(
    validator=[
        _same_length_as_res,
        _contains_indices_in_res,
    ]
)

Indices available to be written into

costs class-attribute instance-attribute

costs: MutableSequence[None | float] = field(
    validator=[_all_none_to_start, _same_length_as_res]
)

Costs of runs

params instance-attribute

params: tuple[str]

Names of the parameters being stored in x_samples

res class-attribute instance-attribute

res: MutableSequence[None | DataContainer] = field(
    validator=[_all_none_to_start]
)

Results of runs

x_samples class-attribute instance-attribute

x_samples: MutableSequence[None | NDArray[number[Any]]] = (
    field(
        validator=[_all_none_to_start, _same_length_as_res]
    )
)

x vectors sampled

append_result_cost_x

append_result_cost_x(
    res: DataContainer, cost: float, x: NDArray[number[Any]]
) -> None

Append result, cost and x from a successful run to the results

Parameters:

Name	Type	Description	Default
res	DataContainer	Result to append (use None for a failed run)	required
cost	float	Cost associated with the run	required
x	NDArray[number[Any]]	Parameter array associated with the run	required

Source code in src/openscm_calibration/store/base.py

def append_result_cost_x(
    self,
    res: DataContainer,
    cost: float,
    x: np.typing.NDArray[np.number[Any]],
) -> None:
    """
    Append result, cost and x from a successful run to the results

    Parameters
    ----------
    res
        Result to append (use `None` for a failed run)

    cost
        Cost associated with the run

    x
        Parameter array associated with the run
    """
    iteration = self.get_available_index()
    # res_keep = res.copy()
    # res_keep["it"] = iteration
    res_keep = self.add_iteration_to_res(res, iteration)

    self.set_result_cost_x(
        res=res_keep,
        cost=cost,
        x=x,
        idx=iteration,
    )

from_n_runs classmethod

from_n_runs(
    n_runs: int,
    params: tuple[str],
    add_iteration_to_res: Callable[
        [DataContainer, int], DataContainer
    ],
) -> OptResStore[Any]

Initialise based on expected number of runs

Parameters:

Name	Type	Description	Default
n_runs	int	Expected number of runs	required
params	tuple[str]	Names of the parameters that are being sampled	required
add_iteration_to_res	Callable[[DataContainer, int], DataContainer]	Function that adds iteration information to the result of a run. For further information, see the docstring of self.	required

Returns:

Type	Description
OptResStore[Any]	Initialised store

Source code in src/openscm_calibration/store/base.py

@classmethod
def from_n_runs(
    cls,
    n_runs: int,
    params: tuple[str],
    add_iteration_to_res: Callable[[DataContainer, int], DataContainer],
) -> OptResStore[Any]:
    """
    Initialise based on expected number of runs

    Parameters
    ----------
    n_runs
        Expected number of runs

    params
        Names of the parameters that are being sampled

    add_iteration_to_res
        Function that adds iteration information to the result of a run.

        For further information, see the docstring of `self`.

    Returns
    -------
    :
        Initialised store
    """
    # Reverse so that using pop counts up
    available_indices = list(range(n_runs))[::-1]
    return cls(
        res=[None] * n_runs,
        costs=[None] * n_runs,
        x_samples=[None] * n_runs,
        params=params,
        available_indices=available_indices,
        add_iteration_to_res=add_iteration_to_res,
    )

from_n_runs_manager classmethod

from_n_runs_manager(
    n_runs: int,
    manager: SupportsListLikeHandling,
    params: tuple[str],
    add_iteration_to_res: Callable[
        [DataContainer, int], DataContainer
    ],
) -> OptResStore[Any]

Initialise based on expected number of runs for use in parallel work

Parameters:

Name	Type	Description	Default
n_runs	int	Expected number of runs	required
manager	SupportsListLikeHandling	Manager of lists (e.g. [multiprocess.managers.SyncManager][multiprocess.managers.SyncManager])	required
params	tuple[str]	Names of the parameters that are being sampled	required
add_iteration_to_res	Callable[[DataContainer, int], DataContainer]	Function that adds iteration information to the result of a run. For further information, see the docstring of self.	required

Returns:

Type	Description
OptResStore[Any]	Initialised store

Source code in src/openscm_calibration/store/base.py

@classmethod
def from_n_runs_manager(
    cls,
    n_runs: int,
    manager: SupportsListLikeHandling,
    params: tuple[str],
    add_iteration_to_res: Callable[[DataContainer, int], DataContainer],
) -> OptResStore[Any]:
    """
    Initialise based on expected number of runs for use in parallel work

    Parameters
    ----------
    n_runs
        Expected number of runs

    manager
        Manager of lists (e.g. [`multiprocess.managers.SyncManager`][])

    params
        Names of the parameters that are being sampled

    add_iteration_to_res
        Function that adds iteration information to the result of a run.

        For further information, see the docstring of `self`.

    Returns
    -------
    :
        Initialised store
    """
    # Reverse so that using pop counts up
    available_indices = list(range(n_runs))[::-1]

    return cls(
        res=manager.list([None] * n_runs),
        costs=manager.list([None] * n_runs),
        x_samples=manager.list([None] * n_runs),
        params=params,
        available_indices=manager.list(available_indices),
        add_iteration_to_res=add_iteration_to_res,
    )

get_available_index

get_available_index() -> int

Get an available index to write into

Returns:

Type	Description
int	Available index. This index is now no longer considered available.

Source code in src/openscm_calibration/store/base.py

def get_available_index(self) -> int:
    """
    Get an available index to write into

    Returns
    -------
    :
        Available index. This index is now no longer considered available.
    """
    return self.available_indices.pop()

get_costs_labelled_xsamples_res

get_costs_labelled_xsamples_res() -> tuple[
    tuple[float, ...],
    dict[str, NDArray[number[Any]]],
    tuple[DataContainer, ...],
]

Get costs, x_samples and res from runs

Returns:

Type	Description
tuple[tuple[float, ...], dict[str, NDArray[number[Any]]], tuple[DataContainer, ...]]	Costs, x_samples and res from all runs which were attempted (i.e. we include failed runs here)

Source code in src/openscm_calibration/store/base.py

def get_costs_labelled_xsamples_res(
    self,
) -> tuple[
    tuple[float, ...],
    dict[str, np.typing.NDArray[np.number[Any]]],
    tuple[DataContainer, ...],
]:
    """
    Get costs, x_samples and res from runs

    Returns
    -------
    :
        Costs, x_samples and res from all runs which were attempted
        (i.e. we include failed runs here)
    """
    unlabelled = self.get_costs_xsamples_res()
    if not any(unlabelled):
        return (
            unlabelled[0],
            {p: np.array([]) for p in self.params},
            unlabelled[2],
        )

    x_samples_stacked = np.vstack(unlabelled[1])
    xs_labelled = {p: x_samples_stacked[:, i] for i, p in enumerate(self.params)}

    out = (unlabelled[0], xs_labelled, unlabelled[2])

    return out

get_costs_xsamples_res

get_costs_xsamples_res() -> tuple[
    tuple[float, ...],
    tuple[NDArray[number[Any]], ...],
    tuple[DataContainer, ...],
]

Get costs, x_samples and res from runs

Returns:

Type	Description
tuple[tuple[float, ...], tuple[NDArray[number[Any]], ...], tuple[DataContainer, ...]]	Costs, x_samples and res from all runs which were attempted (i.e. we include failed runs here)

Source code in src/openscm_calibration/store/base.py

def get_costs_xsamples_res(
    self,
) -> tuple[
    tuple[float, ...],
    tuple[np.typing.NDArray[np.number[Any]], ...],
    tuple[DataContainer, ...],
]:
    """
    Get costs, x_samples and res from runs

    Returns
    -------
    :
        Costs, x_samples and res from all runs which were attempted
        (i.e. we include failed runs here)
    """
    # There may be a better algorithm for this, PRs welcome :)
    if all(x is None for x in self.x_samples):
        return ((), (), ())

    tmp = tuple(
        zip(
            *[
                (
                    self.costs[i],
                    x,
                    self.res[i],
                )
                for i, x in enumerate(self.x_samples)
                # x is only None if no run was attempted yet
                if x is not None
            ]
        )
    )

    # Help out type hinting
    costs: tuple[float, ...] = tmp[0]
    xs_out: tuple[np.typing.NDArray[np.number[Any]], ...] = tmp[1]
    ress: tuple[DataContainer, ...] = tmp[2]

    out = (costs, xs_out, ress)

    return out

note_failed_run

note_failed_run(
    cost: float, x: NDArray[number[Any]]
) -> None

Note that a run failed

Typically, cost will be np.inf.

The cost and x parameters are appended to the results, as well as an indicator that the run was a failure in self.res.

Parameters:

Name	Type	Description	Default
cost	float	Cost associated with the run	required
x	NDArray[number[Any]]	Parameter array associated with the run	required

Source code in src/openscm_calibration/store/base.py

def note_failed_run(
    self,
    cost: float,
    x: np.typing.NDArray[np.number[Any]],
) -> None:
    """
    Note that a run failed

    Typically, `cost` will be `np.inf`.

    The cost and x parameters are appended to the results, as well as an
    indicator that the run was a failure in `self.res`.

    Parameters
    ----------
    cost
        Cost associated with the run

    x
        Parameter array associated with the run
    """
    iteration = self.get_available_index()
    self.set_result_cost_x(
        res=None,
        cost=cost,
        x=x,
        idx=iteration,
    )

set_result_cost_x

set_result_cost_x(
    res: None | DataContainer,
    cost: float,
    x: NDArray[number[Any]],
    idx: int,
) -> None

Set result, cost and x at a given index

Parameters:

Name	Type	Description	Default
res	None | DataContainer	Result to append (use None for a failed run)	required
cost	float	Cost associated with the run	required
x	NDArray[number[Any]]	Parameter array associated with the run	required
idx	int	Index in self.costs, self.x_samples and self.res to write into	required

Source code in src/openscm_calibration/store/base.py

def set_result_cost_x(
    self,
    res: None | DataContainer,
    cost: float,
    x: np.typing.NDArray[np.number[Any]],
    idx: int,
) -> None:
    """
    Set result, cost and x at a given index

    Parameters
    ----------
    res
        Result to append (use `None` for a failed run)

    cost
        Cost associated with the run

    x
        Parameter array associated with the run

    idx
        Index in `self.costs`, `self.x_samples` and `self.res` to write into
    """
    len_x = len(x)
    len_params = len(self.params)
    if len_x != len_params:
        raise MismatchLengthError(
            "x",
            length=len_x,
            expected_name="self.params",
            expected_length=len_params,
        )

    self.costs[idx] = cost
    self.x_samples[idx] = x
    self.res[idx] = res

SupportsListLikeHandling

Bases: Protocol

Class that supports handling a list-like

Methods:

Name	Description
list	Get a new object that behaves like a MutableSequence

Source code in src/openscm_calibration/store/base.py

class SupportsListLikeHandling(Protocol):
    """
    Class that supports handling a list-like
    """

    def list(self, list_to_handle: MutableSequence[Any]) -> MutableSequence[Any]:
        """
        Get a new object that behaves like a [`MutableSequence`][collections.abc.MutableSequence]
        """  # noqa: E501

list

list(
    list_to_handle: MutableSequence[Any],
) -> MutableSequence[Any]

Get a new object that behaves like a MutableSequence

Source code in src/openscm_calibration/store/base.py

def list(self, list_to_handle: MutableSequence[Any]) -> MutableSequence[Any]:
    """
    Get a new object that behaves like a [`MutableSequence`][collections.abc.MutableSequence]
    """  # noqa: E501