Skip to content

Records

deap_er.records

ArchiveStats(num_elites, num_cells, coverage, qd_score) dataclass

Summary statistics for a MAP-Elites archive.

Attributes:

Name Type Description
num_elites int

Number of stored elites.
num_cells int

Capacity of the tessellation, or the current elite count when the archive has no fixed cell budget.
coverage float

num_elites / num_cells, or 0 when num_cells is 0.
qd_score float

Sum of the first weighted objective over elites. Requires single-objective fitness on every stored elite.

CaseExam(ranges=None, mask=None, *, length=None)

A case subset stored as ranges or a 1-D bool mask.

This is data, not a genome. The same shapes feed case_errors (series segments) and sel_lexicase(..., cases=) (catalog indices via :meth:as_cases).

Store exactly one of ranges or mask.

Parameters:

Name Type Description Default
ranges CaseRanges | None

Half-open (start, stop) pairs or a (n, 2) integer table.

 None
mask ndarray | None

One-dimensional bool array.

 None
length int | None

Optional series span. When set and different from the fitness-case count, :meth:as_cases returns segment indices and range mutation uses this bound.

 None

Raises:

Type Description
ValueError

If both or neither form is given, or mask is not a 1-D bool array.
Source code in deap_er/private/records/case_exam.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    ranges: CaseRanges | None = None,
    mask: numpy.ndarray | None = None,
    *,
    length: int | None = None,
) -> None:
    """Store exactly one of ``ranges`` or ``mask``.

    Args:
        ranges: Half-open ``(start, stop)`` pairs or a ``(n, 2)``
            integer table.
        mask: One-dimensional ``bool`` array.
        length: Optional series span. When set and different from
            the fitness-case count, :meth:`as_cases` returns segment
            indices and range mutation uses this bound.

    Raises:
        ValueError: If both or neither form is given, or ``mask`` is
            not a 1-D ``bool`` array.
    """
    if (ranges is None) == (mask is None):
        raise ValueError("CaseExam requires exactly one of ranges or mask")
    self._length = length
    self._ranges: list[tuple[int, int]] | None
    self._mask: numpy.ndarray | None
    if mask is not None:
        array = numpy.asarray(mask)
        if array.ndim != 1:
            raise ValueError("a boolean mask must be one-dimensional")
        if array.dtype != bool:
            raise ValueError("a boolean mask must have dtype bool")
        self._mask = array.copy()
        self._ranges = None
        return
    self._mask = None
    self._ranges = [tuple(pair) for pair in (ranges if ranges is not None else ())]

ranges property

Live range list, or None when the exam stores a mask.

mask property

Live boolean mask, or None when the exam stores ranges.

length property

Optional series span, or None when unset.

from_cases(cases, n_cases) classmethod

Build a catalog-index exam from selected case indices.

Parameters:

Name Type Description Default
cases Sequence[int]

Fitness-case indices to mark True.

 required
n_cases int

Length of the catalog mask.

 required

Returns:

Type Description
CaseExam

An exam whose mask has length n_cases.

Raises:

Type Description
IndexError

If an index is not a valid case index.
ValueError

If n_cases is negative.
Source code in deap_er/private/records/case_exam.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@classmethod
def from_cases(cls, cases: Sequence[int], n_cases: int) -> CaseExam:
    """Build a catalog-index exam from selected case indices.

    Args:
        cases: Fitness-case indices to mark ``True``.
        n_cases: Length of the catalog mask.

    Returns:
        An exam whose mask has length ``n_cases``.

    Raises:
        IndexError: If an index is not a valid case index.
        ValueError: If ``n_cases`` is negative.
    """
    if n_cases < 0:
        raise ValueError("n_cases must be non-negative")
    mask = numpy.zeros(n_cases, dtype=bool)
    for idx in cases:
        mask[_case_index(idx, n_cases)] = True
    return cls(mask=mask)

as_ranges(length)

Return validated [start, stop) intervals against length.

Parameters:

Name Type Description Default
length int

Exclusive upper bound for endpoints, or the mask length.

 required

Returns:

Type Description
list[tuple[int, int]]

Half-open intervals in stored order.

Raises:

Type Description
ValueError

If stored bounds or the mask do not match length.
Source code in deap_er/private/records/case_exam.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def as_ranges(self, length: int) -> list[tuple[int, int]]:
    """Return validated ``[start, stop)`` intervals against ``length``.

    Args:
        length: Exclusive upper bound for endpoints, or the mask length.

    Returns:
        Half-open intervals in stored order.

    Raises:
        ValueError: If stored bounds or the mask do not match ``length``.
    """
    if self._mask is not None:
        return ranges_from_mask(self._mask, length)
    return normalize_case_ranges(self._ranges or [], length)

as_mask(length)

Return a boolean mask of length.

Parameters:

Name Type Description Default
length int

Length of the painted mask, or the stored mask length.

 required

Returns:

Type Description
ndarray

A 1-D bool copy (stored masks) or a painted range mask.

Raises:

Type Description
ValueError

If stored bounds or the mask do not match length.
Source code in deap_er/private/records/case_exam.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def as_mask(self, length: int) -> numpy.ndarray:
    """Return a boolean mask of ``length``.

    Args:
        length: Length of the painted mask, or the stored mask length.

    Returns:
        A 1-D ``bool`` copy (stored masks) or a painted range mask.

    Raises:
        ValueError: If stored bounds or the mask do not match ``length``.
    """
    if self._mask is not None:
        if self._mask.shape[0] != length:
            raise ValueError("a boolean mask must match the series length")
        return self._mask.copy()
    return mask_from_ranges(self._ranges or [], length)

as_cases(n_cases)

Interpret the exam as catalog indices or series segments.

A stored mask must have length n_cases. Catalog ranges (every stop <= n_cases) expand to those indices. Series ranges — any stop > n_cases, or an explicit length different from n_cases — return 0 .. n_segments-1.

Parameters:

Name Type Description Default
n_cases int

Number of fitness cases in the current pack.

 required

Returns:

Type Description
list[int]

Distinct case or segment indices in first-occurrence order.

Raises:

Type Description
ValueError

If stored catalog bounds or the mask do not match n_cases.
Source code in deap_er/private/records/case_exam.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def as_cases(self, n_cases: int) -> list[int]:
    """Interpret the exam as catalog indices or series segments.

    A stored mask must have length ``n_cases``. Catalog ranges
    (every ``stop <= n_cases``) expand to those indices. Series
    ranges — any ``stop > n_cases``, or an explicit ``length``
    different from ``n_cases`` — return ``0 .. n_segments-1``.

    Args:
        n_cases: Number of fitness cases in the current pack.

    Returns:
        Distinct case or segment indices in first-occurrence order.

    Raises:
        ValueError: If stored catalog bounds or the mask do not
            match ``n_cases``.
    """
    if self._mask is not None:
        if self._mask.shape[0] != n_cases:
            raise ValueError("a boolean mask must match the series length")
        return [int(idx) for idx in numpy.flatnonzero(self._mask)]
    if self._is_series(n_cases):
        return list(range(len(self._ranges or [])))
    seen: set[int] = set()
    chosen: list[int] = []
    for start, stop in normalize_case_ranges(self._ranges or [], n_cases):
        for idx in range(start, stop):
            if idx not in seen:
                seen.add(idx)
                chosen.append(idx)
    return chosen

mutation_bound(n_cases)

Exclusive endpoint bound for range mutation.

Parameters:

Name Type Description Default
n_cases int

Fitness-case count from the current pack.

 required

Returns:

Type Description
int

The series span when this exam is a series, otherwise
int

n_cases.
Source code in deap_er/private/records/case_exam.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def mutation_bound(self, n_cases: int) -> int:
    """Exclusive endpoint bound for range mutation.

    Args:
        n_cases: Fitness-case count from the current pack.

    Returns:
        The series span when this exam is a series, otherwise
        ``n_cases``.
    """
    if self._length is not None:
        return self._length
    if self._ranges and any(int(stop) > n_cases for _, stop in self._ranges):
        return max(int(stop) for _, stop in self._ranges)
    if self._mask is not None:
        return int(self._mask.shape[0])
    return n_cases

copy()

Return a copy of the stored ranges or mask.

Returns:

Type Description
CaseExam

A new exam with the same subset.
Source code in deap_er/private/records/case_exam.py 
202
203
204
205
206
207
208
209
210
def copy(self) -> CaseExam:
    """Return a copy of the stored ranges or mask.

    Returns:
        A new exam with the same subset.
    """
    if self._mask is not None:
        return CaseExam(mask=self._mask, length=self._length)
    return CaseExam(ranges=list(self._ranges or []), length=self._length)

assign(other)

Replace this exam's storage with a copy of other.

Parameters:

Name Type Description Default
other CaseExam

Exam whose ranges or mask become this exam's data.

 required
Source code in deap_er/private/records/case_exam.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
def assign(self, other: CaseExam) -> None:
    """Replace this exam's storage with a copy of ``other``.

    Args:
        other: Exam whose ranges or mask become this exam's data.
    """
    self._length = other._length
    if other._mask is not None:
        self._mask = other._mask.copy()
        self._ranges = None
        return
    self._mask = None
    self._ranges = list(other._ranges or [])

CaseExamPool(exams, *, held_out=None, min_cases=1)

Cheap second population of :class:CaseExam subsets.

Stores exams and an optional caller-marked held-out exam. Scoring, mutation, and lexicase feed stay on the operators.

Create a pool of exams.

Parameters:

Name Type Description Default
exams Sequence[CaseExam]

Initial case subsets.

 required
held_out CaseExam | None

Optional exam injected on empty or all-solved collapse.

 None
min_cases int

Minimum catalog size after a guard repair.

 1

Raises:

Type Description
ValueError

If min_cases is less than 1.
Source code in deap_er/private/records/case_exam_pool.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    exams: Sequence[CaseExam],
    *,
    held_out: CaseExam | None = None,
    min_cases: int = 1,
) -> None:
    """Create a pool of exams.

    Args:
        exams: Initial case subsets.
        held_out: Optional exam injected on empty or all-solved collapse.
        min_cases: Minimum catalog size after a guard repair.

    Raises:
        ValueError: If ``min_cases`` is less than 1.
    """
    if min_cases < 1:
        raise ValueError("min_cases must be at least 1")
    self._exams = list(exams)
    self.held_out = held_out
    self.min_cases = min_cases
    self.last_good: CaseExam | None = None

exams property

Live list of exams in this pool.

__len__()

Return the number of stored exams.

Source code in deap_er/private/records/case_exam_pool.py 
59
60
61
def __len__(self) -> int:
    """Return the number of stored exams."""
    return len(self._exams)

__iter__()

Iterate over stored exams.

Source code in deap_er/private/records/case_exam_pool.py 
63
64
65
def __iter__(self) -> Iterator[CaseExam]:
    """Iterate over stored exams."""
    return iter(self._exams)

__getitem__(index)

Return the exam at index.

Source code in deap_er/private/records/case_exam_pool.py 
67
68
69
def __getitem__(self, index: int) -> CaseExam:
    """Return the exam at ``index``."""
    return self._exams[index]

CvtArchive(centroids)

MAP-Elites archive that assigns descriptors to CVT centroids.

Each individual is stored in the Voronoi cell of the nearest centroid. Fitness stays on ind.fitness; the caller supplies the behavior descriptor. fitness must be single-objective.

Parameters:

Name Type Description Default
centroids Sequence[Sequence[float]] | ndarray

(k, dims) array of cell centers in descriptor space.

 required

See the class docstring.

Source code in deap_er/private/records/cvt_archive.py 
50
51
52
53
54
55
56
def __init__(self, centroids: Sequence[Sequence[float]] | numpy.ndarray) -> None:
    """See the class docstring."""
    self._centroids = parse_centroids(centroids)
    self._cells: dict[int, Individual] = {}
    self._tree = (
        KDTree(self._centroids) if self._centroids.shape[0] >= KDTREE_MIN_CENTROIDS else None
    )

centroids property

Copy of the (k, dims) centroid array.

dimensions property

Number of behavior dimensions.

stats property

Coverage and quality-diversity score of the archive.

num_cells is the number of centroids. qd_score is the sum of fitness.wvalues[0] over elites.

from_samples(samples, k, *, n_iter=20) classmethod

Build an archive from k-means centroids of samples.

Parameters:

Name Type Description Default
samples Sequence[Sequence[float]] | ndarray

Behavior descriptors with shape (n, dims).

 required
k int

Number of centroids.

 required
n_iter int

Independent k-means runs passed to :func:cvt_centroids.

 20

Returns:

Type Description
CvtArchive

An empty archive whose cells are the computed centroids.
Source code in deap_er/private/records/cvt_archive.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@classmethod
def from_samples(
    cls,
    samples: Sequence[Sequence[float]] | numpy.ndarray,
    k: int,
    *,
    n_iter: int = 20,
) -> CvtArchive:
    """Build an archive from k-means centroids of ``samples``.

    Args:
        samples: Behavior descriptors with shape ``(n, dims)``.
        k: Number of centroids.
        n_iter: Independent k-means runs passed to
            :func:`cvt_centroids`.

    Returns:
        An empty archive whose cells are the computed centroids.
    """
    return cls(cvt_centroids(samples, k, n_iter=n_iter))

nearest_centroid(descriptor)

Return the index of the centroid nearest to descriptor.

Archives with fewer than 512 centroids break ties by lowest index. Larger archives follow scipy.spatial.KDTree.query order.

Parameters:

Name Type Description Default
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
int

Centroid index in 0 .. k-1.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions.
Source code in deap_er/private/records/cvt_archive.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def nearest_centroid(self, descriptor: Sequence[float]) -> int:
    """Return the index of the centroid nearest to ``descriptor``.

    Archives with fewer than 512 centroids break ties by
    lowest index. Larger archives follow
    ``scipy.spatial.KDTree.query`` order.

    Args:
        descriptor: Continuous behavior coordinates.

    Returns:
        Centroid index in ``0 .. k-1``.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``.
    """
    if len(descriptor) != self.dimensions:
        raise ValueError(
            f"descriptor length {len(descriptor)} does not match {self.dimensions} dimensions"
        )
    return self._centroid_index(descriptor)

add(individual, descriptor)

Insert individual when it improves its Voronoi cell.

Parameters:

Name Type Description Default
individual Any

Candidate with a valid fitness attribute.

 required
descriptor Sequence[float] | ndarray

Continuous behavior coordinates.

 required

Returns:

Type Description
bool

True when the archive stores individual.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions, or fitness is not single-objective.
Source code in deap_er/private/records/cvt_archive.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def add(self, individual: Any, descriptor: Sequence[float] | numpy.ndarray) -> bool:
    """Insert ``individual`` when it improves its Voronoi cell.

    Args:
        individual: Candidate with a valid fitness attribute.
        descriptor: Continuous behavior coordinates.

    Returns:
        True when the archive stores ``individual``.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``, or ``fitness`` is not single-objective.
    """
    if not check_archive_add(individual, descriptor, self.dimensions, "CvtArchive"):
        return False
    cell = self._centroid_index(descriptor)
    return replace_cell_if_better(self._cells, cell, individual)

elite_at(descriptor)

Return the elite in the cell for descriptor.

Parameters:

Name Type Description Default
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
Individual | None

The stored elite, or None when the cell is empty or
Individual | None

descriptor is non-finite.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions.
Source code in deap_er/private/records/cvt_archive.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def elite_at(self, descriptor: Sequence[float]) -> Individual | None:
    """Return the elite in the cell for ``descriptor``.

    Args:
        descriptor: Continuous behavior coordinates.

    Returns:
        The stored elite, or None when the cell is empty or
        ``descriptor`` is non-finite.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``.
    """
    return elite_at_descriptor(self._cells, descriptor, self.dimensions, self._centroid_index)

get(index)

Return the elite stored at centroid index.

Parameters:

Name Type Description Default
index int

Centroid index in 0 .. k-1.

 required

Returns:

Type Description
Individual | None

The stored elite, or None when the cell is empty.
Source code in deap_er/private/records/cvt_archive.py 
173
174
175
176
177
178
179
180
181
182
def get(self, index: int) -> Individual | None:
    """Return the elite stored at centroid ``index``.

    Args:
        index: Centroid index in ``0 .. k-1``.

    Returns:
        The stored elite, or None when the cell is empty.
    """
    return self._cells.get(index)

random_elites(n, *, replace=True)

Sample elites uniformly from filled cells.

Parameters:

Name Type Description Default
n int

Number of elites to return.

 required
replace bool

Sample with replacement when True.

 True

Returns:

Type Description
list[Individual]

Stored elites from distinct or repeated cells.

Raises:

Type Description
IndexError

If the archive is empty.
ValueError

If n is negative, or replace is False and n exceeds the number of elites.
Source code in deap_er/private/records/cvt_archive.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def random_elites(self, n: int, *, replace: bool = True) -> list[Individual]:
    """Sample elites uniformly from filled cells.

    Args:
        n: Number of elites to return.
        replace: Sample with replacement when True.

    Returns:
        Stored elites from distinct or repeated cells.

    Raises:
        IndexError: If the archive is empty.
        ValueError: If ``n`` is negative, or ``replace`` is False and
            ``n`` exceeds the number of elites.
    """
    return sample_random_elites(
        list(self._cells.values()),
        n,
        replace=replace,
        empty_message="random_elites from empty CvtArchive",
    )

clear()

Remove every stored elite.

Source code in deap_er/private/records/cvt_archive.py 
206
207
208
def clear(self) -> None:
    """Remove every stored elite."""
    self._cells.clear()

__len__()

Return the number of filled cells.

Source code in deap_er/private/records/cvt_archive.py 
210
211
212
def __len__(self) -> int:
    """Return the number of filled cells."""
    return len(self._cells)

__contains__(index)

Return whether centroid index holds an elite.

Source code in deap_er/private/records/cvt_archive.py 
214
215
216
def __contains__(self, index: int) -> bool:
    """Return whether centroid ``index`` holds an elite."""
    return index in self._cells

__iter__()

Iterate over stored elites.

Source code in deap_er/private/records/cvt_archive.py 
218
219
220
def __iter__(self) -> Iterator[Individual]:
    """Iterate over stored elites."""
    return iter(self._cells.values())

GridArchive(ranges, bins)

MAP-Elites grid archive indexed by a behavior descriptor.

The caller supplies a continuous behavior descriptor for each individual. The archive bins descriptors into a uniform grid and keeps the best individual per cell according to fitness.

fitness must be single-objective (one weight). Multi-objective fitness types are rejected by :meth:add. stats.qd_score sums the first weighted objective (:attr:~deap_er.base.Fitness.wvalues element zero) across filled cells.

Parameters:

Name Type Description Default
ranges Sequence[tuple[float, float]]

(low, high) bounds per behavior dimension.

 required
bins Sequence[int] | int

Resolution per dimension, or one integer for every dimension.

 required

See the class docstring.

Source code in deap_er/private/records/grid_archive.py 
54
55
56
57
58
59
60
61
def __init__(
    self,
    ranges: Sequence[tuple[float, float]],
    bins: Sequence[int] | int,
) -> None:
    """See the class docstring."""
    self._ranges, self._bins, self._num_cells = parse_grid_config(ranges, bins)
    self._cells: dict[tuple[int, ...], Individual] = {}

dimensions property

Number of behavior dimensions.

bins property

Resolution of the grid along each behavior dimension.

ranges property

(low, high) bounds per behavior dimension.

stats property

Coverage and quality-diversity score of the archive.

qd_score is the sum of fitness.wvalues[0] over elites. It is a MAP-Elites-style scalar quality total, not a sum across multiple objectives.

descriptor_to_index(descriptor)

Map a behavior descriptor to its grid cell.

Coordinates outside ranges are clipped before binning.

Parameters:

Name Type Description Default
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
tuple[int, ...]

Integer grid index per dimension.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions.
Source code in deap_er/private/records/grid_archive.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def descriptor_to_index(self, descriptor: Sequence[float]) -> tuple[int, ...]:
    """Map a behavior descriptor to its grid cell.

    Coordinates outside ``ranges`` are clipped before binning.

    Args:
        descriptor: Continuous behavior coordinates.

    Returns:
        Integer grid index per dimension.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``.
    """
    return descriptor_to_index(descriptor, self._ranges, self._bins)

index_to_descriptor_center(index)

Return the center of a grid cell in behavior space.

Parameters:

Name Type Description Default
index tuple[int, ...]

Integer grid index per dimension.

 required

Returns:

Type Description
tuple[float, ...]

Center coordinate per behavior dimension.

Raises:

Type Description
ValueError

If index length or any coordinate is out of range.
Source code in deap_er/private/records/grid_archive.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def index_to_descriptor_center(self, index: tuple[int, ...]) -> tuple[float, ...]:
    """Return the center of a grid cell in behavior space.

    Args:
        index: Integer grid index per dimension.

    Returns:
        Center coordinate per behavior dimension.

    Raises:
        ValueError: If ``index`` length or any coordinate is out of
            range.
    """
    return index_to_descriptor_center(index, self._ranges, self._bins)

add(individual, descriptor)

Insert individual when it improves its behavior cell.

Parameters:

Name Type Description Default
individual Any

Candidate with a valid fitness attribute.

 required
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
bool

True when the archive stores individual.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions, or fitness is not single-objective.
Source code in deap_er/private/records/grid_archive.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def add(self, individual: Any, descriptor: Sequence[float]) -> bool:
    """Insert ``individual`` when it improves its behavior cell.

    Args:
        individual: Candidate with a valid fitness attribute.
        descriptor: Continuous behavior coordinates.

    Returns:
        True when the archive stores ``individual``.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``, or ``fitness`` is not single-objective.
    """
    if not check_archive_add(individual, descriptor, self.dimensions, "GridArchive"):
        return False
    cell = self.descriptor_to_index(descriptor)
    return replace_cell_if_better(self._cells, cell, individual)

elite_at(descriptor)

Return the elite in the cell for descriptor.

Parameters:

Name Type Description Default
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
Individual | None

The stored elite, or None when the cell is empty or
Individual | None

descriptor is non-finite.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions.
Source code in deap_er/private/records/grid_archive.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def elite_at(self, descriptor: Sequence[float]) -> Individual | None:
    """Return the elite in the cell for ``descriptor``.

    Args:
        descriptor: Continuous behavior coordinates.

    Returns:
        The stored elite, or None when the cell is empty or
        ``descriptor`` is non-finite.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``.
    """
    return elite_at_descriptor(
        self._cells, descriptor, self.dimensions, self.descriptor_to_index
    )

get(index)

Return the elite stored at index.

Parameters:

Name Type Description Default
index tuple[int, ...]

Integer grid index per dimension.

 required

Returns:

Type Description
Individual | None

The stored elite, or None when the cell is empty.
Source code in deap_er/private/records/grid_archive.py 
157
158
159
160
161
162
163
164
165
166
def get(self, index: tuple[int, ...]) -> Individual | None:
    """Return the elite stored at ``index``.

    Args:
        index: Integer grid index per dimension.

    Returns:
        The stored elite, or None when the cell is empty.
    """
    return self._cells.get(index)

random_elites(n, *, replace=True)

Sample elites uniformly from filled cells.

Parameters:

Name Type Description Default
n int

Number of elites to return.

 required
replace bool

Sample with replacement when True.

 True

Returns:

Type Description
list[Individual]

Stored elites from distinct or repeated cells.

Raises:

Type Description
IndexError

If the archive is empty.
ValueError

If n is negative, or replace is False and n exceeds the number of elites.
Source code in deap_er/private/records/grid_archive.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def random_elites(self, n: int, *, replace: bool = True) -> list[Individual]:
    """Sample elites uniformly from filled cells.

    Args:
        n: Number of elites to return.
        replace: Sample with replacement when True.

    Returns:
        Stored elites from distinct or repeated cells.

    Raises:
        IndexError: If the archive is empty.
        ValueError: If ``n`` is negative, or ``replace`` is False and
            ``n`` exceeds the number of elites.
    """
    return sample_random_elites(
        list(self._cells.values()),
        n,
        replace=replace,
        empty_message="random_elites from empty GridArchive",
    )

clear()

Remove every stored elite.

Source code in deap_er/private/records/grid_archive.py 
190
191
192
def clear(self) -> None:
    """Remove every stored elite."""
    self._cells.clear()

__len__()

Return the number of filled cells.

Source code in deap_er/private/records/grid_archive.py 
194
195
196
def __len__(self) -> int:
    """Return the number of filled cells."""
    return len(self._cells)

__contains__(index)

Return whether index holds an elite.

Source code in deap_er/private/records/grid_archive.py 
198
199
200
def __contains__(self, index: tuple[int, ...]) -> bool:
    """Return whether ``index`` holds an elite."""
    return index in self._cells

__iter__()

Iterate over stored elites.

Source code in deap_er/private/records/grid_archive.py 
202
203
204
def __iter__(self) -> Iterator[Individual]:
    """Iterate over stored elites."""
    return iter(self._cells.values())

HallOfFame(maxsize, similar=eq)

Bases: BaseRecordStorage

Archive of the best individuals seen during evolution.

Members stay sorted by fitness so the first item is the best individual seen so far, according to the fitness weights.

Parameters:

Name Type Description Default
maxsize int

Maximum number of individuals to keep.

 required
similar Callable[..., Any]

Equality test used to skip duplicates. Defaults to operator.eq.

 eq

See the class docstring.

Source code in deap_er/private/records/hall_of_fame.py 
108
109
110
111
112
def __init__(self, maxsize: int, similar: Callable[..., Any] = eq) -> None:
    """See the class docstring."""
    self.maxsize = maxsize
    self.similar = similar
    super().__init__()

update(population)

Update the archive from population.

Better individuals replace the worst members. The archive stays at most maxsize and skips individuals already present according to similar. Individuals without a comparable fitness (missing, invalid, or non-finite) are ignored.

Parameters:

Name Type Description Default
population Sequence[Any]

Individuals that may have a fitness attribute.

 required
Source code in deap_er/private/records/hall_of_fame.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def update(self, population: Sequence[Any]) -> None:
    """Update the archive from ``population``.

    Better individuals replace the worst members. The archive stays
    at most ``maxsize`` and skips individuals already present
    according to ``similar``. Individuals without a comparable
    fitness (missing, invalid, or non-finite) are ignored.

    Args:
        population: Individuals that may have a fitness attribute.
    """
    if self.maxsize == 0:
        return
    for ind in population:
        self._update_one(ind)

to_json()

Serialize maxsize and archive members to JSON.

Source code in deap_er/private/records/hall_of_fame.py 
166
167
168
def to_json(self) -> str:
    """Serialize ``maxsize`` and archive members to JSON."""
    return hall_of_fame_to_json(self)

from_json(text, ind_cls=None) classmethod

Rebuild a hall of fame from :meth:to_json output.

Source code in deap_er/private/records/hall_of_fame.py 
170
171
172
173
@classmethod
def from_json(cls, text: str, ind_cls: type[Any] | None = None) -> HallOfFame:
    """Rebuild a hall of fame from :meth:`to_json` output."""
    return hall_of_fame_from_json(text, ind_cls, cls)

ParetoFront(similar=eq)

Bases: BaseRecordStorage

Archive of every non-dominated individual seen during evolution.

The front is unbounded: every unique non-dominated individual is kept.

Parameters:

Name Type Description Default
similar Callable[..., Any]

Equality test used to skip duplicates. Defaults to operator.eq.

 eq

See the class docstring.

Source code in deap_er/private/records/hall_of_fame.py 
186
187
188
189
def __init__(self, similar: Callable[..., Any] = eq) -> None:
    """See the class docstring."""
    self.similar = similar
    super().__init__()

update(population)

Add non-dominated individuals from population.

Members dominated by a new individual are removed. Similar individuals with equal fitness are not added again. Individuals without a comparable fitness (missing, invalid, or non-finite) are ignored.

Parameters:

Name Type Description Default
population Sequence[Any]

Individuals that may have a fitness attribute.

 required
Source code in deap_er/private/records/hall_of_fame.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def update(self, population: Sequence[Any]) -> None:
    """Add non-dominated individuals from ``population``.

    Members dominated by a new individual are removed. Similar
    individuals with equal fitness are not added again.
    Individuals without a comparable fitness (missing, invalid,
    or non-finite) are ignored.

    Args:
        population: Individuals that may have a fitness attribute.
    """
    for ind in population:
        if not has_comparable_fitness(ind):
            continue
        is_dominated, has_twin, to_remove = self._front_verdict(ind)
        for i in reversed(to_remove):
            self.remove(i)
        if not is_dominated and not has_twin:
            self.insert(ind)

History()

Genealogy of individuals produced during evolution.

Call update on the initial population and after each variation, or wrap variation operators with decorator.

Create an empty genealogy.

Source code in deap_er/private/records/history.py 
30
31
32
33
34
def __init__(self) -> None:
    """Create an empty genealogy."""
    self.genealogy_index = 0
    self.genealogy_history = {}
    self.genealogy_tree = {}

decorator property

Decorator that records a variation operator's returned individuals.

update(individuals)

Record individuals in the genealogy.

Call this on the initial population and after each variation. Individuals that already have history_index become the parents of the newly recorded entries; otherwise the entries are roots.

Parameters:

Name Type Description Default
individuals list[Individual]

Individuals to add to the genealogy.

 required
Source code in deap_er/private/records/history.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def update(self, individuals: list[Individual]) -> None:
    """Record ``individuals`` in the genealogy.

    Call this on the initial population and after each variation.
    Individuals that already have ``history_index`` become the
    parents of the newly recorded entries; otherwise the entries
    are roots.

    Args:
        individuals: Individuals to add to the genealogy.
    """
    parent_indices = tuple(
        ind.history_index for ind in individuals if hasattr(ind, "history_index")
    )

    for ind in individuals:
        self.genealogy_index += 1
        ind.history_index = self.genealogy_index
        self.genealogy_history[self.genealogy_index] = deepcopy(ind)
        self.genealogy_tree[self.genealogy_index] = parent_indices

get_genealogy(individual, max_depth=float('inf'))

Return the ancestor graph of an individual.

The individual must have a history_index set by update. The graph includes parents up to max_depth variation steps. The default max_depth walks back to the start of the evolution.

Parameters:

Name Type Description Default
individual Individual

Individual at the root of the genealogy tree.

 required
max_depth float

Maximum number of variation steps to walk.

 float('inf')

Returns:

Type Description
dict[int, Any]

Mapping of individual index to a tuple of parent indices.

Raises:

Type Description
AttributeError

If the individual has no history_index.
Source code in deap_er/private/records/history.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def get_genealogy(
    self, individual: Individual, max_depth: float = float("inf")
) -> dict[int, Any]:
    """Return the ancestor graph of an individual.

    The individual must have a ``history_index`` set by ``update``.
    The graph includes parents up to ``max_depth`` variation steps.
    The default ``max_depth`` walks back to the start of the
    evolution.

    Args:
        individual: Individual at the root of the genealogy tree.
        max_depth: Maximum number of variation steps to walk.

    Returns:
        Mapping of individual index to a tuple of parent indices.

    Raises:
        AttributeError: If the individual has no ``history_index``.
    """

    def _recursive(index: int, depth: int) -> None:
        if index not in self.genealogy_tree:
            return
        depth += 1
        if depth > max_depth:
            return
        parent_indices = self.genealogy_tree[index]
        gtree[index] = parent_indices
        for ind in parent_indices:
            if ind not in visited:
                _recursive(ind, depth)
            visited.add(ind)

    if hasattr(individual, "history_index"):
        visited = set()
        gtree = {}
        _recursive(individual.history_index, 0)
        return gtree
    else:
        raise AttributeError("The individual must have the 'history_index' attribute.")

Logbook()

Bases: list[dict[str, Any]]

Chronological evolution records as a list of dictionaries.

Retrieve columns with select. Nested dictionaries passed to record become named chapters. Set header to control column order when printing.

Create an empty logbook.

Source code in deap_er/private/records/logbook.py 
30
31
32
33
34
35
36
37
38
def __init__(self) -> None:
    """Create an empty logbook."""
    self.chapters = defaultdict(Logbook)
    self.buff_index: int = 0
    self.log_header: bool = True
    self.columns_len: list[int] = []
    self.header: list[str] = []
    self.header_streamed: bool = False
    super().__init__()

stream property

Formatted text of entries recorded since the last stream read.

record(**data)

Append one chronological entry.

Nested dict values are recorded into named chapters. Remaining keys form the entry on this logbook. Non-dict keys are also copied into each chapter.

Parameters:

Name Type Description Default
**data Any

Fields for the new entry. Dict values become chapters.

 {}
Source code in deap_er/private/records/logbook.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def record(self, **data: Any) -> None:
    """Append one chronological entry.

    Nested dict values are recorded into named chapters. Remaining
    keys form the entry on this logbook. Non-dict keys are also
    copied into each chapter.

    Args:
        **data: Fields for the new entry. Dict values become chapters.
    """
    apply_to_all = {k: v for k, v in data.items() if not isinstance(v, dict)}
    for key, value in list(data.items()):
        if isinstance(value, dict):
            chapter_infos = value.copy()
            chapter_infos.update(apply_to_all)
            self.chapters[key].record(**chapter_infos)
            del data[key]
    self.append(data)

select(*names)

Return recorded values for one or more field names.

A missing name yields None in that column. One name returns a flat list; several names return a list of lists.

Parameters:

Name Type Description Default
*names str

Field names to retrieve.

 ()

Returns:

Type Description
list[Any]

Values for the requested names, in chronological order.
Source code in deap_er/private/records/logbook.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def select(self, *names: str) -> list[Any]:
    """Return recorded values for one or more field names.

    A missing name yields ``None`` in that column. One name
    returns a flat list; several names return a list of lists.

    Args:
        *names: Field names to retrieve.

    Returns:
        Values for the requested names, in chronological order.
    """
    if len(names) == 1:
        return [entry.get(names[0], None) for entry in self]
    return [[entry.get(name, None) for entry in self] for name in names]

pop(index=0)

Remove and return the entry at index.

The stream cursor is moved back when the removed entry has already been streamed. The chapter row that shares gen is removed from every chapter. A row without gen is paired by index when the chapter is the same length.

Parameters:

Name Type Description Default
index SupportsIndex

Position of the entry to remove.

 0

Returns:

Type Description
dict[str, Any]

The removed entry.
Source code in deap_er/private/records/logbook.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@override
def pop(self, index: SupportsIndex = 0) -> dict[str, Any]:
    """Remove and return the entry at ``index``.

    The stream cursor is moved back when the removed entry has
    already been streamed. The chapter row that shares ``gen``
    is removed from every chapter. A row without ``gen`` is
    paired by index when the chapter is the same length.

    Args:
        index: Position of the entry to remove.

    Returns:
        The removed entry.
    """
    idx = int(index)
    if idx < 0:
        idx += len(self)
    if 0 <= idx < len(self):
        generation = self[idx].get("gen")
        for chapter in self.chapters.values():
            if not chapter:
                continue
            match = self.chapter_index_for_generation(chapter, generation, idx)
            if match is not None:
                chapter.pop(match)
    if idx < self.buff_index:
        self.buff_index -= 1
    return super().pop(idx)

chapter_index_for_generation(chapter, generation, parent_index)

Return the chapter row that shares generation.

When several rows share a generation, the match is the occurrence that lines up with parent_index. When generation is missing and the chapter is the same length as this logbook, the match is positional.

Parameters:

Name Type Description Default
chapter Logbook

Nested logbook to search.

 required
generation Any

Generation value from the parent entry.

 required
parent_index int

Parent row being paired.

 required

Returns:

Type Description
int | None

Matching chapter index, or None.
Source code in deap_er/private/records/logbook.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def chapter_index_for_generation(
    self, chapter: "Logbook", generation: Any, parent_index: int
) -> int | None:
    """Return the chapter row that shares ``generation``.

    When several rows share a generation, the match is the
    occurrence that lines up with ``parent_index``. When
    ``generation`` is missing and the chapter is the same
    length as this logbook, the match is positional.

    Args:
        chapter: Nested logbook to search.
        generation: Generation value from the parent entry.
        parent_index: Parent row being paired.

    Returns:
        Matching chapter index, or None.
    """
    if generation is None:
        if 0 <= parent_index < len(chapter) == len(self):
            return parent_index
        return None
    remaining = sum(1 for entry in self[parent_index:] if entry.get("gen") == generation)
    matches = [i for i, entry in enumerate(chapter) if entry.get("gen") == generation]
    if remaining == 0 or len(matches) < remaining:
        return None
    return matches[-remaining]

__delitem__(key)

Delete an entry and the same index from every chapter.

Source code in deap_er/private/records/logbook.py 
152
153
154
155
156
157
158
@override
def __delitem__(self, key: SupportsIndex | slice, /) -> None:
    """Delete an entry and the same index from every chapter."""
    if isinstance(key, slice):
        self._delete_slice(key)
    else:
        self.pop(key)

clear()

Remove every entry and the matching chapter rows.

Uses the same chapter pairing and stream-cursor rules as del logbook[:].

Source code in deap_er/private/records/logbook.py 
160
161
162
163
164
165
166
167
@override
def clear(self) -> None:
    """Remove every entry and the matching chapter rows.

    Uses the same chapter pairing and stream-cursor rules as
    ``del logbook[:]``.
    """
    del self[:]

__txt__(start_index)

Format rows from start_index as aligned column strings.

Parameters:

Name Type Description Default
start_index int

First entry to include.

 required

Returns:

Type Description
list[str]

One formatted line per row, including a header when
list[str]

start_index is 0 and log_header is True.
Source code in deap_er/private/records/logbook.py 
169
170
171
172
173
174
175
176
177
178
179
def __txt__(self, start_index: int) -> list[str]:
    """Format rows from ``start_index`` as aligned column strings.

    Args:
        start_index: First entry to include.

    Returns:
        One formatted line per row, including a header when
        ``start_index`` is 0 and ``log_header`` is True.
    """
    return format_txt(self, start_index)

__str__()

Return the logbook as an aligned text table.

Source code in deap_er/private/records/logbook.py 
181
182
183
184
@override
def __str__(self) -> str:
    """Return the logbook as an aligned text table."""
    return "\n".join(self.__txt__(0))

to_json()

Serialize entries, chapters, and the header to JSON.

NumPy scalars become Python numbers. Other non-JSON values become strings.

Returns:

Type Description
str

A JSON document.
Source code in deap_er/private/records/logbook.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def to_json(self) -> str:
    """Serialize entries, chapters, and the header to JSON.

    NumPy scalars become Python numbers. Other non-JSON values
    become strings.

    Returns:
        A JSON document.
    """
    payload = {
        "header": self.header,
        "entries": [_json_ready(entry) for entry in self],
        "chapters": {
            name: json.loads(chapter.to_json()) for name, chapter in self.chapters.items()
        },
    }
    return json.dumps(payload)

from_json(text) classmethod

Rebuild a logbook from :meth:to_json output.

Parameters:

Name Type Description Default
text str

JSON document produced by :meth:to_json.

 required

Returns:

Type Description
Logbook

A logbook with restored entries, chapters, and header.
Source code in deap_er/private/records/logbook.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@classmethod
def from_json(cls, text: str) -> "Logbook":
    """Rebuild a logbook from :meth:`to_json` output.

    Args:
        text: JSON document produced by :meth:`to_json`.

    Returns:
        A logbook with restored entries, chapters, and header.
    """
    data = json.loads(text)
    book = cls()
    book.header = list(data.get("header", []))
    book.extend(data.get("entries", []))
    for name, chapter in data.get("chapters", {}).items():
        book.chapters[name] = cls.from_json(json.dumps(chapter))
    return book

PolicyObservation(solve_bits, unsolved_count, train_score, held_out_score, archive_coverage, qd_score, nevals, rows_seen, promoted_library_size, fitness_invalid, last_action_rejected) dataclass

Fixed Push policy observation record.

This is not a genome and not a domain metric. It is the only typed layout a private policy may read: summaries from case errors, exams, archives, promoted-library size, and eval budget counters. Raw column packs and matrix[t] never appear here.

Attributes:

Name Type Description
solve_bits tuple[int, ...]

Per-case 0/1 solve flags for the observed individual.
unsolved_count int

Number of cases in solve_bits that are not solved.
train_score float

Sum of train-exam difficulty scores from :func:~deap_er.tools.score_case_exams.
held_out_score float | None

Held-out exam difficulty, or None when no held-out exam is marked.
archive_coverage float

MAP-Elites coverage from :class:~deap_er.records.ArchiveStats.
qd_score float

MAP-Elites quality-diversity total from :class:~deap_er.records.ArchiveStats.
nevals int

Evaluations consumed this generation or step.
rows_seen int

Rows in the evaluation matrix seen so far.
promoted_library_size int

Count of promoted primitive names.
fitness_invalid bool

True when the observed individual lacks valid fitness.
last_action_rejected bool

True when the last policy action was rejected by a guard.

as_tuple()

Return fields in fixed schema order for Push or linear policies.

Source code in deap_er/private/records/policy_observation.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def as_tuple(
    self,
) -> tuple[
    tuple[int, ...],
    int,
    float,
    float | None,
    float,
    float,
    int,
    int,
    int,
    bool,
    bool,
]:
    """Return fields in fixed schema order for Push or linear policies."""
    return (
        self.solve_bits,
        self.unsolved_count,
        self.train_score,
        self.held_out_score,
        self.archive_coverage,
        self.qd_score,
        self.nevals,
        self.rows_seen,
        self.promoted_library_size,
        self.fitness_invalid,
        self.last_action_rejected,
    )

SemanticSurrogate(*, metric='euclidean')

Last-generation semantic store for nearest and linear lookup.

update replaces the stored pack and scalar targets. It does not write ind.fitness. predict is a stand-in for last-generation semantics, not a learned quality-diversity model.

Parameters:

Name Type Description Default
metric SemanticMetric

Default finite-mask distance for nearest lookup.

 'euclidean'

See the class docstring.

Source code in deap_er/private/records/semantic_surrogate.py 
36
37
38
39
40
41
def __init__(self, *, metric: SemanticMetric = "euclidean") -> None:
    """See the class docstring."""
    self._metric: SemanticMetric = metric
    self._matrix: numpy.ndarray | None = None
    self._values: numpy.ndarray | None = None
    self._valid: numpy.ndarray | None = None

metric property

Default nearest-neighbor metric.

update(matrix, values, *, valid=None, individuals=None, trust_matrix=False)

Replace the stored last-generation pack.

Parameters:

Name Type Description Default
matrix ndarray | Sequence[Sequence[float]]

Semantic pack of shape (n_individuals, n_rows).

 required
values ndarray | Sequence[float]

Scalar target per row, typically fitness.wvalues[0].

 required
valid ndarray | None

Optional per-row warmup mask stored with the pack.

 None
individuals Sequence[Any] | None

Optional population used by trust_matrix.

 None
trust_matrix bool

When True, accept matrix on shape alone.

 False

Raises:

Type Description
ValueError

If values length does not match the pack, or the pack does not match individuals.
Source code in deap_er/private/records/semantic_surrogate.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def update(
    self,
    matrix: numpy.ndarray | Sequence[Sequence[float]],
    values: numpy.ndarray | Sequence[float],
    *,
    valid: numpy.ndarray | None = None,
    individuals: Sequence[Any] | None = None,
    trust_matrix: bool = False,
) -> None:
    """Replace the stored last-generation pack.

    Args:
        matrix: Semantic pack of shape ``(n_individuals, n_rows)``.
        values: Scalar target per row, typically ``fitness.wvalues[0]``.
        valid: Optional per-row warmup mask stored with the pack.
        individuals: Optional population used by ``trust_matrix``.
        trust_matrix: When ``True``, accept ``matrix`` on shape alone.

    Raises:
        ValueError: If ``values`` length does not match the pack, or
            the pack does not match ``individuals``.
    """
    if individuals is None:
        packed = as_semantic_matrix(matrix)
    else:
        packed = validate_semantic_matrix(matrix, individuals, trust_matrix=trust_matrix)
    stored = numpy.asarray(values, dtype=numpy.float64)
    if stored.ndim != 1 or stored.shape[0] != packed.shape[0]:
        raise ValueError("values must be a one-dimensional vector matching n_individuals")
    if valid is not None:
        sample_valid = numpy.asarray(valid, dtype=bool)
        if sample_valid.ndim != 1 or sample_valid.shape[0] != packed.shape[1]:
            raise ValueError("valid must be a one-dimensional mask matching the series length")
        self._valid = sample_valid.copy()
    else:
        self._valid = None
    self._matrix = packed.copy()
    self._values = stored.copy()

nearest(query, k=1, *, metric=None, valid=None)

Return stored-row indices nearest to query.

Parameters:

Name Type Description Default
query ndarray | Sequence[float]

Semantic row of length n_rows.

 required
k int

Maximum number of neighbors to return.

 1
metric SemanticMetric | None

Distance used for this call. Defaults to the constructor metric.

 None
valid ndarray | None

Warmup mask. Defaults to the mask from update.

 None

Returns:

Type Description
ndarray

Neighbor indices in increasing distance order.

Raises:

Type Description
ValueError

If the store is empty.
Source code in deap_er/private/records/semantic_surrogate.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def nearest(
    self,
    query: numpy.ndarray | Sequence[float],
    k: int = 1,
    *,
    metric: SemanticMetric | None = None,
    valid: numpy.ndarray | None = None,
) -> numpy.ndarray:
    """Return stored-row indices nearest to ``query``.

    Args:
        query: Semantic row of length ``n_rows``.
        k: Maximum number of neighbors to return.
        metric: Distance used for this call. Defaults to the
            constructor metric.
        valid: Warmup mask. Defaults to the mask from ``update``.

    Returns:
        Neighbor indices in increasing distance order.

    Raises:
        ValueError: If the store is empty.
    """
    packed, _values = self._require_store()
    mask = self._valid if valid is None else valid
    return semantic_nearest(
        query,
        packed,
        k=k,
        metric=self._metric if metric is None else metric,
        valid=mask,
    )

predict(query, *, kind='nearest', k=1, metric=None, valid=None)

Predict a scalar from last-generation semantics.

nearest returns the stored value of the nearest row, or the mean of k neighbors. linear fits least squares on finite stored rows. Fallback to nearest happens only when the design is empty or rank < 1, not when rank < min(shape). Underdetermined packs (more columns than rows) keep the minimum-norm solution.

Parameters:

Name Type Description Default
query ndarray | Sequence[float]

Semantic row of length n_rows.

 required
kind SurrogateKind

nearest or linear.

 'nearest'
k int

Neighbor count for nearest (and the linear fallback).

 1
metric SemanticMetric | None

Distance used for nearest lookup.

 None
valid ndarray | None

Warmup mask. Defaults to the mask from update.

 None

Returns:

Type Description
float

Predicted scalar, or nan when no finite neighbor exists.

Raises:

Type Description
ValueError

If the store is empty or kind is unknown.
Source code in deap_er/private/records/semantic_surrogate.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def predict(
    self,
    query: numpy.ndarray | Sequence[float],
    *,
    kind: SurrogateKind = "nearest",
    k: int = 1,
    metric: SemanticMetric | None = None,
    valid: numpy.ndarray | None = None,
) -> float:
    """Predict a scalar from last-generation semantics.

    ``nearest`` returns the stored value of the nearest row, or the
    mean of ``k`` neighbors. ``linear`` fits least squares on finite
    stored rows. Fallback to nearest happens only when the design is
    empty or ``rank < 1``, not when ``rank < min(shape)``.
    Underdetermined packs (more columns than rows) keep the
    minimum-norm solution.

    Args:
        query: Semantic row of length ``n_rows``.
        kind: ``nearest`` or ``linear``.
        k: Neighbor count for ``nearest`` (and the linear fallback).
        metric: Distance used for nearest lookup.
        valid: Warmup mask. Defaults to the mask from ``update``.

    Returns:
        Predicted scalar, or ``nan`` when no finite neighbor exists.

    Raises:
        ValueError: If the store is empty or ``kind`` is unknown.
    """
    if kind == "nearest":
        return self._predict_nearest(query, k=k, metric=metric, valid=valid)
    if kind == "linear":
        return self._predict_linear(query, k=k, metric=metric, valid=valid)
    raise ValueError(f"kind must be 'nearest' or 'linear', got {kind!r}")

MultiStatistics

Bases: dict[str, Any]

Compile several named Statistics objects in one call.

Construct with keyword arguments that map a chapter name to a Statistics instance, for example MultiStatistics(fitness=stats_fit, size=stats_size). register forwards the same function to every chapter unless chapters names a subset.

fields property

Sorted names of the contained Statistics objects.

register(name, func, *args, chapters=None, **kwargs)

Register func on contained Statistics objects.

Parameters:

Name Type Description Default
name str

Key used for this statistic in each chapter record.

 required
func Callable[..., Any]

Function applied to each chapter's key values.

 required
*args Any

Positional arguments bound into func.

 ()
chapters str | Iterable[str] | None

Chapter name or names to update. None registers on every chapter.

 None
**kwargs Any

Keyword arguments bound into func.

 {}
Source code in deap_er/private/records/statistics.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def register(
    self,
    name: str,
    func: Callable[..., Any],
    *args: Any,
    chapters: str | Iterable[str] | None = None,
    **kwargs: Any,
) -> None:
    """Register ``func`` on contained ``Statistics`` objects.

    Args:
        name: Key used for this statistic in each chapter record.
        func: Function applied to each chapter's key values.
        *args: Positional arguments bound into ``func``.
        chapters: Chapter name or names to update. ``None``
            registers on every chapter.
        **kwargs: Keyword arguments bound into ``func``.
    """
    if chapters is None:
        targets = self.values()
    else:
        names = (chapters,) if isinstance(chapters, str) else tuple(chapters)
        targets = [self[chapter] for chapter in names]
    for stats in targets:
        stats.register(name, func, *args, **kwargs)

compile(data)

Compile every contained Statistics object on data.

Parameters:

Name Type Description Default
data Iterable[Any]

Iterable of elements passed to each chapter.

 required

Returns:

Type Description
dict[str, Any]

Mapping of chapter name to that chapter's compiled record.
Source code in deap_er/private/records/statistics.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def compile(self, data: Iterable[Any]) -> dict[str, Any]:
    """Compile every contained ``Statistics`` object on ``data``.

    Args:
        data: Iterable of elements passed to each chapter.

    Returns:
        Mapping of chapter name to that chapter's compiled record.
    """
    materialized = list(data)
    record = {}
    for name, stats in self.items():
        record[name] = stats.compile(materialized)
    return record

Statistics(key=None)

Compile named statistics on a sequence of objects.

key selects the value scored on each element. The default key is the identity function. The key may return a sequence when the registered functions accept one, for example a multi-objective fitness passed to a NumPy statistic.

Parameters:

Name Type Description Default
key Callable[..., Any] | None

Extracts the value to score from each element. Defaults to the identity function.

 None

See the class docstring.

Source code in deap_er/private/records/statistics.py 
31
32
33
34
35
def __init__(self, key: Callable[..., Any] | None = None) -> None:
    """See the class docstring."""
    self.key = key if key else lambda obj: obj
    self.functions = {}
    self.fields = []

register(name, func, *args, **kwargs)

Register a statistic computed by compile.

Extra positional and keyword arguments are bound into func.

Parameters:

Name Type Description Default
name str

Key used for this statistic in the compiled record.

 required
func Callable[..., Any]

Function applied to the sequence of key values.

 required
*args Any

Positional arguments bound into func.

 ()
**kwargs Any

Keyword arguments bound into func.

 {}
Source code in deap_er/private/records/statistics.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
def register(self, name: str, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
    """Register a statistic computed by ``compile``.

    Extra positional and keyword arguments are bound into ``func``.

    Args:
        name: Key used for this statistic in the compiled record.
        func: Function applied to the sequence of key values.
        *args: Positional arguments bound into ``func``.
        **kwargs: Keyword arguments bound into ``func``.
    """
    self.functions[name] = partial(func, *args, **kwargs)
    self.fields.append(name)

compile(data)

Compute every registered statistic on data.

Parameters:

Name Type Description Default
data Iterable[Any]

Iterable of elements passed through key.

 required

Returns:

Type Description
dict[str, Any]

Mapping of registered names to computed values.
Source code in deap_er/private/records/statistics.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def compile(self, data: Iterable[Any]) -> dict[str, Any]:
    """Compute every registered statistic on ``data``.

    Args:
        data: Iterable of elements passed through ``key``.

    Returns:
        Mapping of registered names to computed values.
    """
    entry = {}
    values = tuple(self.key(elem) for elem in data)
    for key, func in self.functions.items():
        entry[key] = func(values)
    return entry

UnstructuredArchive(dimensions, min_distance, *, max_elites=None)

MAP-Elites archive that keeps elites by descriptor distance.

A candidate is added when it is at least min_distance from every stored elite and the archive is under capacity. Otherwise it replaces the nearest neighbor if it is strictly fitter. Scale descriptor axes yourself when units differ (for example turnover versus win rate).

Parameters:

Name Type Description Default
dimensions int

Length of each behavior descriptor.

 required
min_distance float

Euclidean threshold that opens a new niche.

 required
max_elites int | None

Optional cap. When full, a far candidate competes with the nearest elite instead of growing the archive.

 None

See the class docstring.

Source code in deap_er/private/records/unstructured_archive.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    dimensions: int,
    min_distance: float,
    *,
    max_elites: int | None = None,
) -> None:
    """See the class docstring."""
    if dimensions < 1:
        raise ValueError("dimensions must be at least 1")
    if not math.isfinite(min_distance) or min_distance <= 0.0:
        raise ValueError("min_distance must be a positive finite number")
    if max_elites is not None and max_elites < 1:
        raise ValueError("max_elites must be at least 1")
    self._dimensions = int(dimensions)
    self._min_distance = float(min_distance)
    self._max_elites = max_elites
    self._elites: list[Individual] = []
    self._descriptors = numpy.empty((0, self._dimensions), dtype=numpy.float64)

dimensions property

Number of behavior dimensions.

min_distance property

Euclidean threshold that opens a new niche.

max_elites property

Elite cap, or None when the archive may grow without bound.

descriptors property

Copy of stored descriptors with shape (n, dimensions).

stats property

Coverage and quality-diversity score of the archive.

num_cells is max_elites when a cap is set, otherwise the current elite count (so uncapped coverage is 1.0 when the archive is non-empty). qd_score is the sum of fitness.wvalues[0] over elites.

add(individual, descriptor)

Insert individual when it opens a niche or beats a neighbor.

Parameters:

Name Type Description Default
individual Any

Candidate with a valid fitness attribute.

 required
descriptor Sequence[float] | ndarray

Continuous behavior coordinates.

 required

Returns:

Type Description
bool

True when the archive stores individual.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions, or fitness is not single-objective.
Source code in deap_er/private/records/unstructured_archive.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def add(self, individual: Any, descriptor: Sequence[float] | numpy.ndarray) -> bool:
    """Insert ``individual`` when it opens a niche or beats a neighbor.

    Args:
        individual: Candidate with a valid fitness attribute.
        descriptor: Continuous behavior coordinates.

    Returns:
        True when the archive stores ``individual``.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``, or ``fitness`` is not single-objective.
    """
    if not check_archive_add(individual, descriptor, self.dimensions, "UnstructuredArchive"):
        return False
    query = numpy.array(descriptor, dtype=numpy.float64, copy=True)
    if not self._elites:
        self._elites.append(deepcopy(individual))
        self._descriptors = query.reshape(1, -1)
        return True
    nearest = nearest_index(self._descriptors, query)
    dist = float(numpy.linalg.norm(self._descriptors[nearest] - query))
    at_capacity = self._max_elites is not None and len(self._elites) >= self._max_elites
    if dist >= self._min_distance and not at_capacity:
        self._elites.append(deepcopy(individual))
        self._descriptors = numpy.vstack((self._descriptors, query))
        return True
    incumbent = self._elites[nearest]
    if individual.fitness <= incumbent.fitness:
        return False
    self._elites[nearest] = deepcopy(individual)
    self._descriptors[nearest] = query
    return True

elite_at(descriptor)

Return the nearest stored elite to descriptor.

Parameters:

Name Type Description Default
descriptor Sequence[float]

Continuous behavior coordinates.

 required

Returns:

Type Description
Individual | None

The nearest elite, or None when the archive is empty or
Individual | None

descriptor is non-finite.

Raises:

Type Description
ValueError

If descriptor length does not match dimensions.
Source code in deap_er/private/records/unstructured_archive.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def elite_at(self, descriptor: Sequence[float]) -> Individual | None:
    """Return the nearest stored elite to ``descriptor``.

    Args:
        descriptor: Continuous behavior coordinates.

    Returns:
        The nearest elite, or None when the archive is empty or
        ``descriptor`` is non-finite.

    Raises:
        ValueError: If ``descriptor`` length does not match
            ``dimensions``.
    """
    if len(descriptor) != self.dimensions:
        raise ValueError(
            f"descriptor length {len(descriptor)} does not match {self.dimensions} dimensions"
        )
    if not all(math.isfinite(float(value)) for value in descriptor):
        return None
    if not self._elites:
        return None
    return self._elites[nearest_index(self._descriptors, descriptor)]

random_elites(n, *, replace=True)

Sample elites uniformly from stored members.

Parameters:

Name Type Description Default
n int

Number of elites to return.

 required
replace bool

Sample with replacement when True.

 True

Returns:

Type Description
list[Individual]

Stored elites from distinct or repeated members.

Raises:

Type Description
IndexError

If the archive is empty.
ValueError

If n is negative, or replace is False and n exceeds the number of elites.
Source code in deap_er/private/records/unstructured_archive.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def random_elites(self, n: int, *, replace: bool = True) -> list[Individual]:
    """Sample elites uniformly from stored members.

    Args:
        n: Number of elites to return.
        replace: Sample with replacement when True.

    Returns:
        Stored elites from distinct or repeated members.

    Raises:
        IndexError: If the archive is empty.
        ValueError: If ``n`` is negative, or ``replace`` is False and
            ``n`` exceeds the number of elites.
    """
    return sample_random_elites(
        self._elites,
        n,
        replace=replace,
        empty_message="random_elites from empty UnstructuredArchive",
    )

clear()

Remove every stored elite.

Source code in deap_er/private/records/unstructured_archive.py 
183
184
185
186
def clear(self) -> None:
    """Remove every stored elite."""
    self._elites.clear()
    self._descriptors = numpy.empty((0, self._dimensions), dtype=numpy.float64)

__len__()

Return the number of stored elites.

Source code in deap_er/private/records/unstructured_archive.py 
188
189
190
def __len__(self) -> int:
    """Return the number of stored elites."""
    return len(self._elites)

__iter__()

Iterate over stored elites.

Source code in deap_er/private/records/unstructured_archive.py 
192
193
194
def __iter__(self) -> Iterator[Individual]:
    """Iterate over stored elites."""
    return iter(self._elites)

coerce_case_exam(exam, n_cases=None)

Wrap ranges, a mask, or catalog indices as a :class:CaseExam.

Parameters:

Name Type Description Default
exam CaseExam | CaseRanges | Sequence[int]

An exam, a range table, a 1-D bool mask, or case indices.

 required
n_cases int | None

Required when exam is a sequence of catalog indices.

 None

Returns:

Type Description
CaseExam

exam if it is already a :class:CaseExam, otherwise a new exam.

Raises:

Type Description
ValueError

If case indices are given without n_cases.
Source code in deap_er/private/records/case_exam_pool.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def coerce_case_exam(
    exam: CaseExam | CaseRanges | Sequence[int],
    n_cases: int | None = None,
) -> CaseExam:
    """Wrap ranges, a mask, or catalog indices as a :class:`CaseExam`.

    Args:
        exam: An exam, a range table, a 1-D ``bool`` mask, or case indices.
        n_cases: Required when ``exam`` is a sequence of catalog indices.

    Returns:
        ``exam`` if it is already a :class:`CaseExam`, otherwise a new exam.

    Raises:
        ValueError: If case indices are given without ``n_cases``.
    """
    if isinstance(exam, CaseExam):
        return exam
    if isinstance(exam, numpy.ndarray):
        if exam.dtype == bool:
            return CaseExam(mask=exam)
        return CaseExam(ranges=exam)
    if not exam:
        return CaseExam(ranges=[])
    first = exam[0]
    if isinstance(first, tuple | list | numpy.ndarray) and len(first) == 2:
        return CaseExam(ranges=cast(CaseRanges, exam))
    if n_cases is None:
        raise ValueError("n_cases is required to coerce case indices")
    return CaseExam.from_cases(cast(Sequence[int], exam), n_cases)

cvt_centroids(samples, k, *, n_iter=20)

Compute k centroids by k-means on a behavior sample.

Seeding uses the process-wide tools.rng generator so checkpointed runs stay reproducible. Callers who already have centroids can pass them straight to :class:~deap_er.records.CvtArchive.

Parameters:

Name Type Description Default
samples Sequence[Sequence[float]] | ndarray

Behavior descriptors with shape (n, dims).

 required
k int

Number of centroids. Must be at least 1 and at most n.

 required
n_iter int

Independent k-means runs; the lowest-distortion result is kept.

 20

Returns:

Type Description
ndarray

Contiguous (k, dims) array of centroids.

Raises:

Type Description
ValueError

If samples is empty, not 2-D, or non-finite, or if k or n_iter is invalid.
Source code in deap_er/private/records/cvt_centroids.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def cvt_centroids(
    samples: Sequence[Sequence[float]] | numpy.ndarray,
    k: int,
    *,
    n_iter: int = 20,
) -> numpy.ndarray:
    """Compute ``k`` centroids by k-means on a behavior sample.

    Seeding uses the process-wide ``tools.rng`` generator so
    checkpointed runs stay reproducible. Callers who already have
    centroids can pass them straight to
    :class:`~deap_er.records.CvtArchive`.

    Args:
        samples: Behavior descriptors with shape ``(n, dims)``.
        k: Number of centroids. Must be at least 1 and at most ``n``.
        n_iter: Independent k-means runs; the lowest-distortion result
            is kept.

    Returns:
        Contiguous ``(k, dims)`` array of centroids.

    Raises:
        ValueError: If ``samples`` is empty, not 2-D, or non-finite,
            or if ``k`` or ``n_iter`` is invalid.
    """
    array = numpy.asarray(samples, dtype=numpy.float64)
    if array.ndim != 2 or array.shape[0] < 1 or array.shape[1] < 1:
        raise ValueError("samples must be a non-empty 2-D array")
    if not bool(numpy.isfinite(array).all()):
        raise ValueError("samples must be finite")
    k_value = int(k)
    if k_value < 1:
        raise ValueError("k must be at least 1")
    if k_value > array.shape[0]:
        raise ValueError("k cannot exceed the number of samples")
    if n_iter < 1:
        raise ValueError("n_iter must be at least 1")
    seed = int(rng.integers(0, 2**31))
    centroids, _distortion = kmeans(array, k_value, iter=n_iter, rng=seed)
    result = numpy.ascontiguousarray(centroids, dtype=numpy.float64)
    if result.shape != (k_value, array.shape[1]):
        raise ValueError(f"k-means returned {result.shape[0]} centroids, expected {k_value}")
    return result

policy_generalization_gap(train_score, held_out_score)

Build the generalization-gap Logbook chapter payload.

Train-exam quality is logged for comparison only. Policy fitness stays on held_out_score via :func:~deap_er.tools.policy_held_out_fitness.

Parameters:

Name Type Description Default
train_score float

Sum of train-exam difficulties from :func:~deap_er.tools.policy_exam_scores.

 required
held_out_score float | None

Held-out exam difficulty, or None when no held-out exam is marked.

 required

Returns:

Type Description
dict[str, float | None]

Chapter fields train, held_out, and gap where
dict[str, float | None]

gap is train - held_out when held-out is present.
Source code in deap_er/private/records/policy_generalization.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def policy_generalization_gap(
    train_score: float,
    held_out_score: float | None,
) -> dict[str, float | None]:
    """Build the generalization-gap Logbook chapter payload.

    Train-exam quality is logged for comparison only. Policy fitness
    stays on ``held_out_score`` via
    :func:`~deap_er.tools.policy_held_out_fitness`.

    Args:
        train_score: Sum of train-exam difficulties from
            :func:`~deap_er.tools.policy_exam_scores`.
        held_out_score: Held-out exam difficulty, or ``None`` when
            no held-out exam is marked.

    Returns:
        Chapter fields ``train``, ``held_out``, and ``gap`` where
        ``gap`` is ``train - held_out`` when held-out is present.
    """
    gap = None
    if held_out_score is not None:
        gap = float(train_score) - float(held_out_score)
    return {
        "train": float(train_score),
        "held_out": None if held_out_score is None else float(held_out_score),
        "gap": gap,
    }

record_policy_generalization_gap(logbook, *, gen, train_score, held_out_score, **extra)

Append one generation row with a generalization-gap chapter.

Parameters:

Name Type Description Default
logbook Logbook

Evolution logbook to update.

 required
gen int

Generation index shared with the parent row.

 required
train_score float

Train-exam difficulty sum for observation.

 required
held_out_score float | None

Held-out difficulty used for policy fitness.

 required
**extra Any

Additional parent-row fields such as nevals.

 {}
Source code in deap_er/private/records/policy_generalization.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def record_policy_generalization_gap(
    logbook: Logbook,
    *,
    gen: int,
    train_score: float,
    held_out_score: float | None,
    **extra: Any,
) -> None:
    """Append one generation row with a generalization-gap chapter.

    Args:
        logbook: Evolution logbook to update.
        gen: Generation index shared with the parent row.
        train_score: Train-exam difficulty sum for observation.
        held_out_score: Held-out difficulty used for policy fitness.
        **extra: Additional parent-row fields such as ``nevals``.
    """
    chapter = policy_generalization_gap(train_score, held_out_score)
    logbook.record(
        gen=gen,
        **extra,
        **{POLICY_GENERALIZATION_GAP_CHAPTER: chapter},
    )