25
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
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
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
119
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175 | class UnstructuredArchive(Archive):
"""Unstructured Archive for Novelty Search algorithms.
The archive stores new instances if their novelty or distance
is larger than the :attr:novelty_threshold.
"""
def __init__(
self,
k: np.uint32 | int,
novelty_threshold: float,
instances: Optional[Sequence[Instance]] = None,
):
"""Creates an instance of a UnstructuredArchive for Quality Diversity algorithms
Args:
k (np.uint32 | int): Number of neighbours to calculate the diversity of the instances.
threshold (float): Minimum value of novelty to include an Instance into the archive.
instances (Optional[Sequence[Instance]], optional): Instances to initialise the archive. Defaults to None.
Raises:
ValueError: K is not valid. K must be a positive integer.
ValueError: Threshold is not valid. Threshold must be a positive float.
"""
if not isinstance(k, (int, np.integer, np.unsignedinteger)) or k <= 0:
raise ValueError(
f"UnstructuredArchive expects k to be a positive integer. Got {k}"
)
if type(novelty_threshold) not in (float, int) or novelty_threshold < 0.0:
raise ValueError(
f"UnstructuredArchive expects a floating point threshold >= 0. Got {novelty_threshold}"
)
super().__init__(instances)
self._k = int(k)
self._novelty_threshold = float(novelty_threshold)
@property
def k(self) -> int:
"""Number of neighbours to calculate the novelty
Returns:
int
"""
return self._k
@property
def novelty_threshold(self) -> float:
"""Novelty Threshold
Returns:
float: Novelty threshold of the archive
"""
return self._novelty_threshold
def __str__(self):
return f"UnstructuredArchive(k={self._k},threshold={self._novelty_threshold},data=(|{len(self)}|))"
def __call__(self, descriptors: np.ndarray) -> np.ndarray:
"""Computes the Novelty Search of the instances
It uses the descriptors of the instances to calculate their novelty
with respect to the archive and the current batch of instances.
It uses the Euclidean distance to compute the sparseness.
If the archive is empty, it returns a NumPy array with the minimum threshold.
Args:
descriptors (np.ndarray): Numpy array with the descriptors of the instances
Returns:
np.ndarray: novelty scores (s) of the instances descriptors
"""
return self.compute_novelty(descriptors)
def compute_novelty(self, descriptors: np.ndarray) -> np.ndarray:
"""Computes the Novelty Search of the instances
It uses the descriptors of the instances to calculate their novelty
with respect to the archive and the current batch of instances.
It uses the Euclidean distance to compute the sparseness.
If the archive is empty, it returns a NumPy array with the minimum threshold.
Args:
descriptors (np.ndarray): Numpy array with the descriptors of the instances
Returns:
np.ndarray: novelty scores (s) of the instances descriptors
"""
num_instances = len(descriptors)
num_archive = len(self)
if num_archive == 0:
return np.full(num_instances, fill_value=self._novelty_threshold)
else:
effective_k = min((num_archive + num_instances) - 1, self._k)
distances = cdist(
descriptors, np.vstack([descriptors, self._storage[Keys.descriptors]])
)
np.fill_diagonal(distances, np.inf)
knn = np.partition(distances, effective_k, axis=1)[:, : effective_k + 1]
novelty = np.mean(knn, axis=1)
return novelty
def extend(
self,
instances: Sequence[Instance],
novelty_scores: Optional[np.ndarray] = None,
descriptors: Optional[np.ndarray] = None,
*args,
**kwargs,
):
"""Extends the archive with the new batch of instances.
Args:
instances (Sequence[Instance]): Instances to insert in the archive.
novelty_scores (Optional[np.ndarray], optional): Novelty scores of the instances.
If not given, the novelyt_scores are calculated using compute_novelty. Defaults to None.
descriptors (Optional[np.ndarray], optional): Descriptors of the instances to calculate their novelty.
If not given, they are extracted from the Instance objects inside the instances collection. Defaults to None.
Raises:
TypeError: If any object inside the instances parameter is not a object of the Instance class.
ValueError: If there is a mismatch in the shapes (lens) of the instances, novelty_scores and descriptors.
"""
if check_valid_instance_batch(instances=instances):
if descriptors is None:
descriptors = np.asarray([
instance.descriptor for instance in instances
])
if novelty_scores is None:
novelty_scores = self.compute_novelty(descriptors=descriptors)
if check_valid_shapes(instances, novelty_scores, descriptors):
novel_enough_mask = novelty_scores >= self._novelty_threshold
valid_instances = np.where(novel_enough_mask)[0].astype(np.int32)
for idx in valid_instances:
self._storage[Keys.instances].append(instances[idx].clone())
self._storage[Keys.descriptors].append(descriptors[idx])
else:
raise ValueError(
"Shape mismatch between the instances, novelty_scores and descriptors."
f"instances have {len(instances)} instances, "
f"novelty_scores contains {len(novelty_scores)} and "
f"descriptors contains {len(descriptors)}."
)
else:
raise TypeError(
"All objects inside the instances sequence must be object of the Instance class."
)
|