@File : init.py @Time : 2024/06/07 14:06:26 @Author : Alejandro Marrero @Version : 1.0 @Contact : amarrerd@ull.edu.es @License : (C)Copyright 2024, Alejandro Marrero @Desc : None

DescriptorFn

Bases: Protocol

Defines the Protocol that all descriptable functions must follow

Source code in digneapy/core/_descriptors.py
25
26
27
28
29
30
31
32
33
34
35
class DescriptorFn(Protocol):
    """Defines the Protocol that all descriptable functions must follow"""

    def __call__(
        self,
        population: np.ndarray | Sequence[Instance],
        scores: Optional[np.ndarray],
        domain: Optional[Domain],
        *args,
        **kwargs,
    ) -> np.ndarray: ...

DescriptorPipeline

Pipeline to transform descriptors with several models

Source code in digneapy/core/_descriptors.py
 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
class DescriptorPipeline:
    """Pipeline to transform descriptors with several models"""

    def __init__(self, key: DescriptorKey, *transformers: Transformer):
        if not isinstance(key, str) or key not in descriptors_registry:
            raise KeyError(
                f"Unknown descriptor key {key}. Registered keys are: {descriptors_registry.keys()}"
            )
        if any(not isinstance(t, Transformer) for t in transformers):
            raise TypeError(
                f"All transformers must implement the Transformer Protocol. Got: {transformers}"
            )
        self._key = key
        self._transformers = tuple(transformers)

    def __call__(
        self,
        population: np.ndarray | Sequence[Instance],
        scores: Optional[np.ndarray] = None,
        domain: Optional[Domain] = None,
        *args,
        **kwargs,
    ) -> np.ndarray:
        descriptors = descriptors_registry[self._key](
            population, scores, domain, *args, **kwargs
        )
        for transfomer in self._transformers:
            descriptors = transfomer(descriptors)
        return descriptors

    def __repr__(self) -> str:
        steps = [self._key] + [
            getattr(t, "__name__", repr(t)) for t in self._transformers
        ]
        return f"DescriptorPipeline({' -> '.join(steps)})"

Domain

Bases: ABC

Domain is a class that defines the domain of the problem.

The domain is defined by its dimension and the bounds of each variable.

Source code in digneapy/core/_domain.py
 23
 24
 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
class Domain(ABC):
    """Domain is a class that defines the domain of the problem.

    The domain is defined by its dimension and the bounds of each variable.
    """

    def __init__(
        self,
        dimension: np.uint32 | int,
        bounds: Sequence[Tuple],
        dtype=np.float64,
        domain_name: str = "Domain",
        features_names: Optional[Sequence[str]] = [],
        seed: Optional[int | np.random.SeedSequence] = None,
        *args,
        **kwargs,
    ):

        try:
            self._dimension = int(dimension)
            if self._dimension <= 0:
                raise ValueError(
                    f"Cannot create a Domain({domain_name}) with negative "
                    f"or equal to zero dimensions. Got {dimension}."
                )
        except (TypeError, ValueError) as exc:
            raise ValueError("invalid dimension in Domain.") from exc

        if len(bounds) != self._dimension:
            raise ValueError(
                f"bounds mismatch in Domain({domain_name}). "
                f"They were expected {self._dimension} bounds but got {len(bounds)}"
            )
        else:
            ranges = list(zip(*bounds))
            self._lbs = np.asarray(ranges[0], dtype=dtype)
            self._ubs = np.asarray(ranges[1], dtype=dtype)

        self.__name__ = domain_name
        self._bounds = bounds
        self._dtype = dtype
        self.features_names = features_names

        self._seed = seed
        self._rng = np.random.default_rng(seed)

    @abstractmethod
    def generate_instances(
        self, n: np.uint32 | int = np.uint32(1)
    ) -> Sequence[Instance]:
        """Generates N instances for the domain.

        Args:
            n (int, optional): Number of instances to generate. Defaults to 1.

        Returns:
            List[Instance]: A list of Instance objects created from the raw numpy generation
        """
        raise NotImplementedError(
            "generate_n_instances is not implemented in Domain class."
        )

    @abstractmethod
    def generate_problems_from_instances(
        self, instances: Sequence[Instance] | np.ndarray
    ) -> Sequence[Problem]:
        msg = "generate_problems_from_instances is not implemented in Domain class."
        raise NotImplementedError(msg)

    @abstractmethod
    def extract_features(
        self, instances: Sequence[Instance] | np.ndarray
    ) -> np.ndarray:
        """Extract the features of the instances based on the domain

        Args:
            instance (Instance): Instance to extract the features from

        Returns:
            Tuple: Values of each feature
        """
        msg = "extract_features is not implemented in Domain class."
        raise NotImplementedError(msg)

    @abstractmethod
    def extract_features_as_dict(
        self, instances: Sequence[Instance] | np.ndarray
    ) -> Sequence[Dict]:
        """Creates a dictionary with the features of the instance.

        The key are the names of each feature and the values are
        the values extracted from instance.

        Args:
            instance (Instance): Instance to extract the features from

        Returns:
            Mapping[str, float]: Dictionary with the names/values of each feature
        """
        msg = "extract_features_as_dict is not implemented in Domain class."
        raise NotImplementedError(msg)

    @property
    def bounds(self):
        return self._bounds

    @property
    def lbs(self) -> np.ndarray:
        return self._lbs

    @property
    def ubs(self) -> np.ndarray:
        return self._ubs

    def get_bounds_at(self, i: int) -> tuple:
        if i < 0 or i > len(self._bounds):
            raise ValueError(
                f"Index {i} out-of-range. The bounds are 0-{len(self._bounds)} "
            )
        return (self._lbs[i], self._ubs[i])

    @property
    def dimension(self):
        return self._dimension

    def __len__(self):
        return self._dimension

extract_features(instances) abstractmethod

Extract the features of the instances based on the domain

Parameters:
  • instance (Instance) –

    Instance to extract the features from

Returns:
  • Tuple( ndarray ) –

    Values of each feature

Source code in digneapy/core/_domain.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@abstractmethod
def extract_features(
    self, instances: Sequence[Instance] | np.ndarray
) -> np.ndarray:
    """Extract the features of the instances based on the domain

    Args:
        instance (Instance): Instance to extract the features from

    Returns:
        Tuple: Values of each feature
    """
    msg = "extract_features is not implemented in Domain class."
    raise NotImplementedError(msg)

extract_features_as_dict(instances) abstractmethod

Creates a dictionary with the features of the instance.

The key are the names of each feature and the values are the values extracted from instance.

Parameters:
  • instance (Instance) –

    Instance to extract the features from

Returns:
  • Sequence[Dict]

    Mapping[str, float]: Dictionary with the names/values of each feature

Source code in digneapy/core/_domain.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@abstractmethod
def extract_features_as_dict(
    self, instances: Sequence[Instance] | np.ndarray
) -> Sequence[Dict]:
    """Creates a dictionary with the features of the instance.

    The key are the names of each feature and the values are
    the values extracted from instance.

    Args:
        instance (Instance): Instance to extract the features from

    Returns:
        Mapping[str, float]: Dictionary with the names/values of each feature
    """
    msg = "extract_features_as_dict is not implemented in Domain class."
    raise NotImplementedError(msg)

generate_instances(n=np.uint32(1)) abstractmethod

Generates N instances for the domain.

Parameters:
  • n (int, default: uint32(1) ) –

    Number of instances to generate. Defaults to 1.

Returns:
  • Sequence[Instance]

    List[Instance]: A list of Instance objects created from the raw numpy generation

Source code in digneapy/core/_domain.py
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@abstractmethod
def generate_instances(
    self, n: np.uint32 | int = np.uint32(1)
) -> Sequence[Instance]:
    """Generates N instances for the domain.

    Args:
        n (int, optional): Number of instances to generate. Defaults to 1.

    Returns:
        List[Instance]: A list of Instance objects created from the raw numpy generation
    """
    raise NotImplementedError(
        "generate_n_instances is not implemented in Domain class."
    )

Instance

Source code in digneapy/core/_instance.py
 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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
class Instance:
    __slots__ = (
        "_variables",
        "_fitness",
        "_performance_bias",
        "_novelty",
        "_descriptor",
        "_portfolio_scores",
        "_descriptor_dim",
        "_portfolio_dim",
        "_dtype",
    )

    def __init__(
        self,
        variables: Sequence | np.ndarray,
        fitness: float | np.float64 = np.float64(0.0),
        performance_bias: float | np.float64 = np.float64(0.0),
        novelty: float | np.float64 = np.float64(0.0),
        descriptor: Optional[Sequence[float | np.float64]] = None,
        portfolio_scores: Optional[Sequence[float | np.float64]] = None,
        dtype=np.uint32,
    ):
        """Creates an instance of a Instance for QD algorithms.


        This class is used to represent a solution in a QD algorithm. It contains the
        variables, fitness, performance bias, novelty, descriptor and portfolio scores
        of the instance.

        The variables, descriptor and portfolio scores are stored as a numpy array.
        The fitness, performance and novelty are stored as np.float64.

        Args:
            variables (Sequence): Variables or genome of the instance.
            fitness (float, optional): Fitness of the instance. Defaults to 0.0.
            performance_bias (float, optional): Performance score. Defaults to 0.0.
            novelty (float, optional): Novelty score. Defaults to 0.0.
            descriptor (Optional[Sequence[float | np.float64], optional): Tuple with the descriptor information of the instance. Defaults to None.
            portfolio_scores (Optional[Sequence[float | np.float64], optional): Scores of the solvers in the portfolio. Defaults to None.

        Raises:
            ValueError: If fitness, performance_bias or novelty are not convertible to float.
        """
        self._dtype = dtype
        try:
            self._fitness = float(fitness)
            self._performance_bias = float(performance_bias)
            self._novelty = float(novelty)
        except (TypeError, ValueError) as exception:
            exception.add_note(f"fitness: {fitness}")
            exception.add_note(f"performance_bias: {performance_bias}")
            exception.add_note(f"novelty: {novelty}")
            raise TypeError("Wrong parameters for Instance.") from exception
        # Todo: Consider fix the dimensions of the descriptor and portfolio
        # if type(descriptor_dim) is not int or descriptor_dim <= 0:
        #     raise ValueError(
        #         f"descriptor_dim must be a positive integer. Got {descriptor_dim}."
        #     )
        # else:
        #     self._descriptor_dim = int(descriptor_dim)

        # if type(portfolio_dim) is not int or portfolio_dim <= 0:
        #     raise ValueError(
        #         f"portfolio_dim must be a positive integer. Got {portfolio_dim}."
        #     )
        # else:
        #     self._portfolio_dim = int(portfolio_dim)
        if variables is None or len(variables) == 0:
            raise ValueError(f"variables has to be a valid sequence. Got: {variables}")
        else:
            self._variables = np.asarray(variables, dtype=self._dtype, copy=True)

        if descriptor is not None and len(descriptor) == 0:
            raise ValueError(
                f"descriptors must be either None or a sequence with at least one value. Got: {descriptor}"
            )
        else:
            if descriptor is None:
                self._descriptor = np.empty(0, dtype=np.float64)
            else:
                self._descriptor = np.asarray(descriptor, dtype=np.float64, copy=True)

        if portfolio_scores is not None and len(portfolio_scores) == 0:
            raise ValueError(
                f"portfolio_scores must be either None or a sequence with at least one value. Got: {portfolio_scores}"
            )
        else:
            if portfolio_scores is None:
                self._portfolio_scores = np.empty((0), dtype=np.float64)
            else:
                self._portfolio_scores = np.asarray(portfolio_scores, dtype=np.float64)

    @property
    def dtype(self):
        return self._dtype

    def clone(self) -> Self:
        """Create a clone of the current instance.

        This avoids Python-level list/tuple conversions by copying the underlying
        NumPy arrays directly.

        Returns:
            Self: Instance object
        """
        new_instance = object.__new__(type(self))
        new_instance._dtype = self._dtype
        new_instance._fitness = self._fitness
        new_instance._performance_bias = self._performance_bias
        new_instance._novelty = self._novelty
        new_instance._variables = self._variables.copy()
        new_instance._descriptor = self._descriptor.copy()
        new_instance._portfolio_scores = self._portfolio_scores.copy()
        return new_instance

    def clone_with(self, **overrides):
        """Clones an Instance with overriden attributes

        Returns:
            Instance
        """
        new_object = self.clone()
        for key, value in overrides.items():
            setattr(new_object, key, value)
        return new_object

    @property
    def variables(self):
        return self._variables

    @variables.setter
    def variables(self, new_variables: np.ndarray):
        if new_variables is None or len(new_variables) == 0:
            raise ValueError(
                f"new_variables cannot be None nor be empty. Got: {new_variables}."
            )

        elif len(new_variables) != len(self._variables):
            raise ValueError(
                "Updating the variables of an Instance object with a different number of values. "
                f"Instance have {len(self._variables)} variables "
                f"and the new_variables sequence have {len(new_variables)}"
            )
        else:
            self._variables = np.asarray(new_variables)

    @property
    def performance_bias(self) -> float | np.float64:
        return self._performance_bias

    @performance_bias.setter
    def performance_bias(self, performance: float | np.float64):
        try:
            self._performance_bias = float(performance)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"performance_bias value is not a float. Got: {performance}."
            ) from exc

    @property
    def novelty(self) -> float | np.float64:
        return self._novelty

    @novelty.setter
    def novelty(self, nov_score: float | np.float64):
        try:
            self._novelty = float(nov_score)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"nov_score value is not a float. Got: {nov_score}."
            ) from exc

    @property
    def fitness(self) -> float | np.float64:
        return self._fitness

    @fitness.setter
    def fitness(self, new_fitness: float | np.float64):
        try:
            self._fitness = float(new_fitness)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"new_fitness value is not a float. Got: {new_fitness}."
            ) from exc

    @property
    def descriptor(self) -> np.ndarray:
        return self._descriptor

    @descriptor.setter
    def descriptor(self, descriptor: Sequence[float | np.float64] | np.ndarray):
        self._descriptor = np.asarray(descriptor)

    @property
    def portfolio_scores(self):
        return self._portfolio_scores

    @portfolio_scores.setter
    def portfolio_scores(self, scores: Sequence[float | np.float64] | np.ndarray):
        self._portfolio_scores = np.asarray(scores)

    def __repr__(self):
        return self.__str__()

    def __str__(self):
        import reprlib

        descriptor = reprlib.repr(self.descriptor)
        performance = reprlib.repr(self.portfolio_scores)
        performance = performance[performance.find("(") : performance.rfind(")") + 1]
        return (
            f"Instance:\n"
            f"   - fitness = {self.fitness}\n"
            f"   - performance_bias = {self.performance_bias}\n"
            f"   - novelty = {self.novelty}\n"
            f"   - descriptor = {descriptor}\n"
            f"   - portfolio scores = {performance}\n"
        )

    def __iter__(self):
        return iter(self._variables)

    def __len__(self):
        return len(self._variables)

    def __getitem__(self, key: int | slice) -> Sequence | np.ndarray:
        """Accessor to variables of the Instance

        Args:
            key (index | slice): index or slice to access a subset of variables

        Returns:
            Sequence or np.ndarray: If accessed with a slice the subset of variables
                are returned. Otherwise, it returns a single scalar at variables[key].
        """
        if not isinstance(key, (int, slice)):
            raise TypeError(
                f"Instance cannot be indexed with type: {type(key)}. Use slice or int."
            )
        if isinstance(key, slice):
            return self._variables[key]
        else:
            index = operator.index(key)
            return self.variables[index]

    def __setitem__(self, key: int | slice, value):
        """Setter to variables of the Instance

        Args:
            key (int | slice): index or slice to access a subset of variables
            value: Value to set in the variables

        """
        if not isinstance(key, (int, np.integer, np.unsignedinteger, slice)):
            raise TypeError(
                f"Instance cannot be update via __setitem__ with type: {type(key)}. Use slice or int."
            )

        if isinstance(key, slice):
            # Compute how many positions this slice actually targets
            start, stop, step = key.indices(len(self.variables))
            target_count = len(range(start, stop, step))
            # Accept any sequence; reject bare scalars for slice assignment
            try:
                expected_len = len(value)
            except TypeError:
                raise TypeError(
                    f"[Instance] slice assignment requires a sequence, got scalar {value!r}. "
                    f"Expected {target_count} value(s)."
                )

            if expected_len != target_count:
                raise ValueError(
                    f"[Instance] slice targets {target_count} element(s) but value has "
                    f"{expected_len} element(s)."
                )
        else:
            index = operator.index(key)
            try:
                if len(value) != 1:
                    raise ValueError(
                        f"[Instance] index {index!r} targets 1 element but value has "
                        f"{len(value)} element(s). Use a slice to assign multiple values."
                    )
            except TypeError:
                pass  # len() failed which means that we have a scalar value

        self._variables[key] = value

    def __eq__(self, other: Self) -> bool:
        """Compares two Instances based on their variables.

        Args:
            other (Self): Another instance to compare.

        Returns:
            bool: Returns True if the two instances have the same number of variables,
            and all of them are equal. Returns False otherwise.
        """
        if not isinstance(other, Instance):
            raise TypeError(
                f"Other of type {other.__class__.__name__} can not be compared with an Instance."
            )

        else:
            try:
                return all(a == b for a, b in zip(self, other, strict=True))
            except ValueError:
                return False

    def __gt__(self, other: Self) -> bool | np.bool:
        """Compares two Instances based on their fitness.

        Args:
            other (Self): Another instance to compare.

        Returns:
            bool: Returns True if the self has a greater fitness
            than the other. Returns False otherwise.
        """
        if not isinstance(other, Instance):
            raise TypeError(
                f"Other of type {other.__class__.__name__} can not be compared with an Instance."
            )
        else:
            return self.fitness > other.fitness

    def __ge__(self, other: Self) -> bool | np.bool:
        """Compares two Instances based on their fitness.

        Args:
            other (Self): Another instance to compare.

        Returns:
            bool: Returns True if the self has a greater or equal fitness
            than the other. Returns False otherwise.
        """
        if not isinstance(other, Instance):
            raise TypeError(
                f"Other of type {other.__class__.__name__} can not be compared with an Instance."
            )
        else:
            return self.fitness >= other.fitness

    def to_dict(
        self,
        variables_names: Optional[Sequence[str]] = None,
        descriptor_names: Optional[Sequence[str]] = None,
        portfolio_names: Optional[Sequence[str]] = None,
    ) -> dict:
        """Convert the instance to a dictionary.

        The keys are the names of the attributes and the values are the values of the attributes.

        Args:
            variables_names   (Optional[Sequence[str]], optional): Names of the variables in the dictionary, otherwise v_i. Defaults to None.
            descriptor_names: (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.
            portfolio_names   (Optional[Sequence[str]], optional): Name of the solvers, otherwise solver_i. Defaults to None.

        Returns:
            dict: Dictionary with the attributes of the instance as keys and the values of the attributes as values.
        """
        _instance_data = {}

        descriptor_names = _validate_column_names(
            "descriptor", descriptor_names, len(self._descriptor), fallback_keyword="d"
        )
        _instance_data = {
            **{key: value for key, value in zip(descriptor_names, self._descriptor)}
        }

        variables_names = _validate_column_names(
            "variables_names", variables_names, len(self), fallback_keyword="v"
        )
        _instance_data["variables"] = {
            key: value for key, value in zip(variables_names, self._variables)
        }

        portfolio_names = _validate_column_names(
            "portfolio_names",
            portfolio_names,
            len(self.portfolio_scores),
            fallback_keyword="alg",
        )
        _instance_data["portfolio_scores"] = {
            key: value for key, value in zip(portfolio_names, self._portfolio_scores)
        }

        _instance_data = {
            "target": portfolio_names[0],
            "fitness": self._fitness,
            "novelty": self._novelty,
            "performance_bias": self._performance_bias,
            **_instance_data,
        }
        return _instance_data

    def to_json(self) -> str:
        """Convert the instance to a JSON string.

        The keys are the names of the attributes and the values are the values of the attributes.

        Returns:
            str: JSON string with the attributes of the instance as keys and the values of the attributes as values.
        """
        import json

        # Todo: Need to change dtypes because np is not JSON serializable
        return json.dumps(self.to_dict(), sort_keys=False, indent=2)

    def to_df(
        self,
        variables_names: Optional[Sequence[str]] = None,
        descriptor_names: Optional[Sequence[str]] = None,
        portfolio_names: Optional[Sequence[str]] = None,
    ) -> pl.DataFrame:
        """Creates a Polars DataFrame from the instance.

        Args:
            variables_names (Optional[Sequence[str]], optional): Names of the variables in the dictionary, otherwise v_i. Defaults to None.
            descriptor_names: (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.
            portfolio_names (Optional[Sequence[str]], optional): Name of the solvers, otherwise solver_i. Defaults to None.

        Returns:
            DataFrame: Polars DataFrame with the attributes of the instance as keys and the values of the attributes as values.
        """
        _flatten_data = {}
        for key, value in self.to_dict(
            variables_names=variables_names,
            descriptor_names=descriptor_names,
            portfolio_names=portfolio_names,
        ).items():
            if isinstance(value, dict):  # Flatten nested dicts
                for sub_key, sub_value in value.items():
                    _flatten_data[sub_key] = sub_value
            else:
                _flatten_data[key] = value
        return pl.DataFrame(_flatten_data)

__eq__(other)

Compares two Instances based on their variables.

Parameters:
  • other (Self) –

    Another instance to compare.

Returns:
  • bool( bool ) –

    Returns True if the two instances have the same number of variables,

  • bool

    and all of them are equal. Returns False otherwise.

Source code in digneapy/core/_instance.py
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def __eq__(self, other: Self) -> bool:
    """Compares two Instances based on their variables.

    Args:
        other (Self): Another instance to compare.

    Returns:
        bool: Returns True if the two instances have the same number of variables,
        and all of them are equal. Returns False otherwise.
    """
    if not isinstance(other, Instance):
        raise TypeError(
            f"Other of type {other.__class__.__name__} can not be compared with an Instance."
        )

    else:
        try:
            return all(a == b for a, b in zip(self, other, strict=True))
        except ValueError:
            return False

__ge__(other)

Compares two Instances based on their fitness.

Parameters:
  • other (Self) –

    Another instance to compare.

Returns:
  • bool( bool | bool ) –

    Returns True if the self has a greater or equal fitness

  • bool | bool

    than the other. Returns False otherwise.

Source code in digneapy/core/_instance.py
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def __ge__(self, other: Self) -> bool | np.bool:
    """Compares two Instances based on their fitness.

    Args:
        other (Self): Another instance to compare.

    Returns:
        bool: Returns True if the self has a greater or equal fitness
        than the other. Returns False otherwise.
    """
    if not isinstance(other, Instance):
        raise TypeError(
            f"Other of type {other.__class__.__name__} can not be compared with an Instance."
        )
    else:
        return self.fitness >= other.fitness

__getitem__(key)

Accessor to variables of the Instance

Parameters:
  • key (index | slice) –

    index or slice to access a subset of variables

Returns:
  • Sequence | ndarray

    Sequence or np.ndarray: If accessed with a slice the subset of variables are returned. Otherwise, it returns a single scalar at variables[key].

Source code in digneapy/core/_instance.py
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def __getitem__(self, key: int | slice) -> Sequence | np.ndarray:
    """Accessor to variables of the Instance

    Args:
        key (index | slice): index or slice to access a subset of variables

    Returns:
        Sequence or np.ndarray: If accessed with a slice the subset of variables
            are returned. Otherwise, it returns a single scalar at variables[key].
    """
    if not isinstance(key, (int, slice)):
        raise TypeError(
            f"Instance cannot be indexed with type: {type(key)}. Use slice or int."
        )
    if isinstance(key, slice):
        return self._variables[key]
    else:
        index = operator.index(key)
        return self.variables[index]

__gt__(other)

Compares two Instances based on their fitness.

Parameters:
  • other (Self) –

    Another instance to compare.

Returns:
  • bool( bool | bool ) –

    Returns True if the self has a greater fitness

  • bool | bool

    than the other. Returns False otherwise.

Source code in digneapy/core/_instance.py
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def __gt__(self, other: Self) -> bool | np.bool:
    """Compares two Instances based on their fitness.

    Args:
        other (Self): Another instance to compare.

    Returns:
        bool: Returns True if the self has a greater fitness
        than the other. Returns False otherwise.
    """
    if not isinstance(other, Instance):
        raise TypeError(
            f"Other of type {other.__class__.__name__} can not be compared with an Instance."
        )
    else:
        return self.fitness > other.fitness

__init__(variables, fitness=np.float64(0.0), performance_bias=np.float64(0.0), novelty=np.float64(0.0), descriptor=None, portfolio_scores=None, dtype=np.uint32)

Creates an instance of a Instance for QD algorithms.

This class is used to represent a solution in a QD algorithm. It contains the variables, fitness, performance bias, novelty, descriptor and portfolio scores of the instance.

The variables, descriptor and portfolio scores are stored as a numpy array. The fitness, performance and novelty are stored as np.float64.

Parameters:
  • variables (Sequence) –

    Variables or genome of the instance.

  • fitness (float, default: float64(0.0) ) –

    Fitness of the instance. Defaults to 0.0.

  • performance_bias (float, default: float64(0.0) ) –

    Performance score. Defaults to 0.0.

  • novelty (float, default: float64(0.0) ) –

    Novelty score. Defaults to 0.0.

  • descriptor (Optional[Sequence[float | np.float64], default: None ) –

    Tuple with the descriptor information of the instance. Defaults to None.

  • portfolio_scores (Optional[Sequence[float | np.float64], default: None ) –

    Scores of the solvers in the portfolio. Defaults to None.

Raises:
  • ValueError

    If fitness, performance_bias or novelty are not convertible to float.

Source code in digneapy/core/_instance.py
 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
def __init__(
    self,
    variables: Sequence | np.ndarray,
    fitness: float | np.float64 = np.float64(0.0),
    performance_bias: float | np.float64 = np.float64(0.0),
    novelty: float | np.float64 = np.float64(0.0),
    descriptor: Optional[Sequence[float | np.float64]] = None,
    portfolio_scores: Optional[Sequence[float | np.float64]] = None,
    dtype=np.uint32,
):
    """Creates an instance of a Instance for QD algorithms.


    This class is used to represent a solution in a QD algorithm. It contains the
    variables, fitness, performance bias, novelty, descriptor and portfolio scores
    of the instance.

    The variables, descriptor and portfolio scores are stored as a numpy array.
    The fitness, performance and novelty are stored as np.float64.

    Args:
        variables (Sequence): Variables or genome of the instance.
        fitness (float, optional): Fitness of the instance. Defaults to 0.0.
        performance_bias (float, optional): Performance score. Defaults to 0.0.
        novelty (float, optional): Novelty score. Defaults to 0.0.
        descriptor (Optional[Sequence[float | np.float64], optional): Tuple with the descriptor information of the instance. Defaults to None.
        portfolio_scores (Optional[Sequence[float | np.float64], optional): Scores of the solvers in the portfolio. Defaults to None.

    Raises:
        ValueError: If fitness, performance_bias or novelty are not convertible to float.
    """
    self._dtype = dtype
    try:
        self._fitness = float(fitness)
        self._performance_bias = float(performance_bias)
        self._novelty = float(novelty)
    except (TypeError, ValueError) as exception:
        exception.add_note(f"fitness: {fitness}")
        exception.add_note(f"performance_bias: {performance_bias}")
        exception.add_note(f"novelty: {novelty}")
        raise TypeError("Wrong parameters for Instance.") from exception
    # Todo: Consider fix the dimensions of the descriptor and portfolio
    # if type(descriptor_dim) is not int or descriptor_dim <= 0:
    #     raise ValueError(
    #         f"descriptor_dim must be a positive integer. Got {descriptor_dim}."
    #     )
    # else:
    #     self._descriptor_dim = int(descriptor_dim)

    # if type(portfolio_dim) is not int or portfolio_dim <= 0:
    #     raise ValueError(
    #         f"portfolio_dim must be a positive integer. Got {portfolio_dim}."
    #     )
    # else:
    #     self._portfolio_dim = int(portfolio_dim)
    if variables is None or len(variables) == 0:
        raise ValueError(f"variables has to be a valid sequence. Got: {variables}")
    else:
        self._variables = np.asarray(variables, dtype=self._dtype, copy=True)

    if descriptor is not None and len(descriptor) == 0:
        raise ValueError(
            f"descriptors must be either None or a sequence with at least one value. Got: {descriptor}"
        )
    else:
        if descriptor is None:
            self._descriptor = np.empty(0, dtype=np.float64)
        else:
            self._descriptor = np.asarray(descriptor, dtype=np.float64, copy=True)

    if portfolio_scores is not None and len(portfolio_scores) == 0:
        raise ValueError(
            f"portfolio_scores must be either None or a sequence with at least one value. Got: {portfolio_scores}"
        )
    else:
        if portfolio_scores is None:
            self._portfolio_scores = np.empty((0), dtype=np.float64)
        else:
            self._portfolio_scores = np.asarray(portfolio_scores, dtype=np.float64)

__setitem__(key, value)

Setter to variables of the Instance

Parameters:
  • key (int | slice) –

    index or slice to access a subset of variables

  • value

    Value to set in the variables

Source code in digneapy/core/_instance.py
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def __setitem__(self, key: int | slice, value):
    """Setter to variables of the Instance

    Args:
        key (int | slice): index or slice to access a subset of variables
        value: Value to set in the variables

    """
    if not isinstance(key, (int, np.integer, np.unsignedinteger, slice)):
        raise TypeError(
            f"Instance cannot be update via __setitem__ with type: {type(key)}. Use slice or int."
        )

    if isinstance(key, slice):
        # Compute how many positions this slice actually targets
        start, stop, step = key.indices(len(self.variables))
        target_count = len(range(start, stop, step))
        # Accept any sequence; reject bare scalars for slice assignment
        try:
            expected_len = len(value)
        except TypeError:
            raise TypeError(
                f"[Instance] slice assignment requires a sequence, got scalar {value!r}. "
                f"Expected {target_count} value(s)."
            )

        if expected_len != target_count:
            raise ValueError(
                f"[Instance] slice targets {target_count} element(s) but value has "
                f"{expected_len} element(s)."
            )
    else:
        index = operator.index(key)
        try:
            if len(value) != 1:
                raise ValueError(
                    f"[Instance] index {index!r} targets 1 element but value has "
                    f"{len(value)} element(s). Use a slice to assign multiple values."
                )
        except TypeError:
            pass  # len() failed which means that we have a scalar value

    self._variables[key] = value

clone()

Create a clone of the current instance.

This avoids Python-level list/tuple conversions by copying the underlying NumPy arrays directly.

Returns:
  • Self( Self ) –

    Instance object

Source code in digneapy/core/_instance.py
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def clone(self) -> Self:
    """Create a clone of the current instance.

    This avoids Python-level list/tuple conversions by copying the underlying
    NumPy arrays directly.

    Returns:
        Self: Instance object
    """
    new_instance = object.__new__(type(self))
    new_instance._dtype = self._dtype
    new_instance._fitness = self._fitness
    new_instance._performance_bias = self._performance_bias
    new_instance._novelty = self._novelty
    new_instance._variables = self._variables.copy()
    new_instance._descriptor = self._descriptor.copy()
    new_instance._portfolio_scores = self._portfolio_scores.copy()
    return new_instance

clone_with(**overrides)

Clones an Instance with overriden attributes

Returns:
  • Instance

Source code in digneapy/core/_instance.py
173
174
175
176
177
178
179
180
181
182
def clone_with(self, **overrides):
    """Clones an Instance with overriden attributes

    Returns:
        Instance
    """
    new_object = self.clone()
    for key, value in overrides.items():
        setattr(new_object, key, value)
    return new_object

to_df(variables_names=None, descriptor_names=None, portfolio_names=None)

Creates a Polars DataFrame from the instance.

Parameters:
  • variables_names (Optional[Sequence[str]], default: None ) –

    Names of the variables in the dictionary, otherwise v_i. Defaults to None.

  • descriptor_names (Optional[Sequence[str]], default: None ) –

    (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.

  • portfolio_names (Optional[Sequence[str]], default: None ) –

    Name of the solvers, otherwise solver_i. Defaults to None.

Returns:
  • DataFrame( DataFrame ) –

    Polars DataFrame with the attributes of the instance as keys and the values of the attributes as values.

Source code in digneapy/core/_instance.py
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
def to_df(
    self,
    variables_names: Optional[Sequence[str]] = None,
    descriptor_names: Optional[Sequence[str]] = None,
    portfolio_names: Optional[Sequence[str]] = None,
) -> pl.DataFrame:
    """Creates a Polars DataFrame from the instance.

    Args:
        variables_names (Optional[Sequence[str]], optional): Names of the variables in the dictionary, otherwise v_i. Defaults to None.
        descriptor_names: (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.
        portfolio_names (Optional[Sequence[str]], optional): Name of the solvers, otherwise solver_i. Defaults to None.

    Returns:
        DataFrame: Polars DataFrame with the attributes of the instance as keys and the values of the attributes as values.
    """
    _flatten_data = {}
    for key, value in self.to_dict(
        variables_names=variables_names,
        descriptor_names=descriptor_names,
        portfolio_names=portfolio_names,
    ).items():
        if isinstance(value, dict):  # Flatten nested dicts
            for sub_key, sub_value in value.items():
                _flatten_data[sub_key] = sub_value
        else:
            _flatten_data[key] = value
    return pl.DataFrame(_flatten_data)

to_dict(variables_names=None, descriptor_names=None, portfolio_names=None)

Convert the instance to a dictionary.

The keys are the names of the attributes and the values are the values of the attributes.

Parameters:
  • variables_names (Optional[Sequence[str]], default: None ) –

    Names of the variables in the dictionary, otherwise v_i. Defaults to None.

  • descriptor_names (Optional[Sequence[str]], default: None ) –

    (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.

  • portfolio_names (Optional[Sequence[str]], default: None ) –

    Name of the solvers, otherwise solver_i. Defaults to None.

Returns:
  • dict( dict ) –

    Dictionary with the attributes of the instance as keys and the values of the attributes as values.

Source code in digneapy/core/_instance.py
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def to_dict(
    self,
    variables_names: Optional[Sequence[str]] = None,
    descriptor_names: Optional[Sequence[str]] = None,
    portfolio_names: Optional[Sequence[str]] = None,
) -> dict:
    """Convert the instance to a dictionary.

    The keys are the names of the attributes and the values are the values of the attributes.

    Args:
        variables_names   (Optional[Sequence[str]], optional): Names of the variables in the dictionary, otherwise v_i. Defaults to None.
        descriptor_names: (Optional[Sequence[str]], optional): Names of the components of the descriptor, otherwisde di. Default to None.
        portfolio_names   (Optional[Sequence[str]], optional): Name of the solvers, otherwise solver_i. Defaults to None.

    Returns:
        dict: Dictionary with the attributes of the instance as keys and the values of the attributes as values.
    """
    _instance_data = {}

    descriptor_names = _validate_column_names(
        "descriptor", descriptor_names, len(self._descriptor), fallback_keyword="d"
    )
    _instance_data = {
        **{key: value for key, value in zip(descriptor_names, self._descriptor)}
    }

    variables_names = _validate_column_names(
        "variables_names", variables_names, len(self), fallback_keyword="v"
    )
    _instance_data["variables"] = {
        key: value for key, value in zip(variables_names, self._variables)
    }

    portfolio_names = _validate_column_names(
        "portfolio_names",
        portfolio_names,
        len(self.portfolio_scores),
        fallback_keyword="alg",
    )
    _instance_data["portfolio_scores"] = {
        key: value for key, value in zip(portfolio_names, self._portfolio_scores)
    }

    _instance_data = {
        "target": portfolio_names[0],
        "fitness": self._fitness,
        "novelty": self._novelty,
        "performance_bias": self._performance_bias,
        **_instance_data,
    }
    return _instance_data

to_json()

Convert the instance to a JSON string.

The keys are the names of the attributes and the values are the values of the attributes.

Returns:
  • str( str ) –

    JSON string with the attributes of the instance as keys and the values of the attributes as values.

Source code in digneapy/core/_instance.py
455
456
457
458
459
460
461
462
463
464
465
466
def to_json(self) -> str:
    """Convert the instance to a JSON string.

    The keys are the names of the attributes and the values are the values of the attributes.

    Returns:
        str: JSON string with the attributes of the instance as keys and the values of the attributes as values.
    """
    import json

    # Todo: Need to change dtypes because np is not JSON serializable
    return json.dumps(self.to_dict(), sort_keys=False, indent=2)

Logbook

Bases: Logbook

Extends Deap.tools.Logbook to include additionaly features such as to_df.

Source code in digneapy/core/_metrics.py
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
class Logbook(tools.Logbook):
    """Extends Deap.tools.Logbook to include additionaly features such as to_df."""

    def __init__(self):
        super().__init__()
        self._statistics = Statistics()
        self.header = "generation", "novelty", "performance bias", "fitness"
        self._chapters_headers = (
            "min",
            "mean",
            "std",
            "max",
            "qd_score",
        )
        self.chapters["novelty"].header = self._chapters_headers[:-1]
        self.chapters["performance_bias"].header = self._chapters_headers
        self.chapters["fitness"].header = self._chapters_headers

    def update(
        self,
        generation: int,
        instances: Sequence[Instance],
        feedback: bool = False,
    ):
        if generation < 0:
            raise ValueError(
                f"generation value {generation} must be greater than zero in Logbook update."
            )
        # Checks for None, empty and not isinstance() delegated to Statistics
        self.record(generation=generation, **self._statistics(instances))
        if feedback:  # pragma: no cover
            print(self.stream)

    def to_df(self) -> pl.DataFrame:
        fitness = pl.from_dicts(self.chapters["fitness"], schema=self._chapters_headers)
        novelty = pl.from_dicts(self.chapters["novelty"], schema=self._chapters_headers)
        performance = pl.from_dicts(
            self.chapters["performance_bias"], schema=self._chapters_headers
        )
        generations = pl.Series("generation", self.select("generation"))

        df = (
            fitness
            .rename({h: f"{h}_fitness" for h in self._chapters_headers})
            .with_columns(generations)
            .join(
                novelty.rename({
                    h: f"{h}_novelty" for h in self._chapters_headers
                }).with_columns(generations),
                on="generation",
            )
            .join(
                performance.rename({
                    h: f"{h}_performance_bias" for h in self._chapters_headers
                }).with_columns(generations),
                on="generation",
            )
        )

        return df.select(["generation", *[c for c in df.columns if c != "generation"]])

Problem

Bases: ABC

Source code in digneapy/core/_problem.py
 23
 24
 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
class Problem(ABC):
    def __init__(
        self,
        dimension: np.uint32 | int,
        bounds: Sequence[Tuple],
        name: str = "Problem",
        dtype=np.float64,
        seed: Optional[int | np.random.SeedSequence] = None,
        *args,
        **kwargs,
    ):
        """Creates a new Problem object.

        The problem is defined by its dimension and the bounds of each variable.

        Args:
            dimension (int): Number of variables in the problem
            bounds (Sequence[tuple]): Bounds of each variable in the problem
            name (str, optional): Name of the problem for printing and logging purposes. Defaults to "Problem".
            dtype (_type_, optional): Type of the variables. Defaults to np.float64.
            seed (int, optional): Seed for the random number generation (_rng). Defaults to None.
        """
        try:
            self._dimension = int(dimension)
            if dimension <= 0:
                raise ValueError
        except Exception as te:
            te.add_note(f"Dimension must be a postive integer. Got: {dimension}")
            raise te

        self.__name__ = name
        self._dimension = dimension
        self._bounds = bounds
        self._dtype = dtype
        if len(self._bounds) != 0:
            ranges = list(zip(*bounds))
            self._lbs = np.asarray(ranges[0], dtype=dtype)
            self._ubs = np.asarray(ranges[1], dtype=dtype)
        self._seed = seed
        self._rng = np.random.default_rng(seed)

    @property
    def dimension(self):
        return self._dimension

    def __len__(self):
        return self._dimension

    @property
    def bounds(self):
        return self._bounds

    @property
    def lbs(self):
        return self._lbs

    @property
    def ubs(self):
        return self._ubs

    def get_bounds_at(self, i: int) -> tuple:
        if i < 0 or i > len(self._bounds):
            raise ValueError(
                f"Index {i} out-of-range. The bounds are 0-{len(self._bounds)} "
            )
        return (self._lbs[i], self._ubs[i])

    @abstractmethod
    def create_solution(self) -> Solution | np.ndarray:
        """Creates a random solution to the problem.
        This method can be used to initialise the solutions
        for any algorithm
        """
        msg = "create_solution method not implemented in Problem"
        raise NotImplementedError(msg)

    @abstractmethod
    def __array__(self, dtype: Any = None, copy: Optional[bool] = None) -> np.ndarray:
        msg = "__array__ method not implemented in Problem"
        raise NotImplementedError(msg)

    @abstractmethod
    def evaluate(
        self, individual: Sequence | Solution | np.ndarray
    ) -> Tuple[float, ...]:
        """Evaluates the candidate individual with the information of the problem

        Args:
            individual (Sequence | Solution | np.ndarray): Individual to evaluate

        Returns:
            Tuple[float]: fitness
        """
        msg = "evaluate method not implemented in Problem"
        raise NotImplementedError(msg)

    @abstractmethod
    def __call__(
        self, individual: Sequence | Solution | np.ndarray
    ) -> Tuple[float, ...]:
        msg = "__call__ method not implemented in Problem"
        raise NotImplementedError(msg)

    @abstractmethod
    def to_instance(self) -> Instance:
        """Creates an instance from the information of the problem.
        This method is used in the generators to create instances to evolve
        """
        msg = "to_instance method not implemented in Problem"
        raise NotImplementedError(msg)

    @abstractmethod
    def to_file(self, filename: str):
        msg = "to_file method not implemented in Problem"
        raise NotImplementedError(msg)

    @classmethod
    def from_file(cls, filename: str):
        msg = "from_file method not implemented in Problem"
        raise NotImplementedError(msg)

__init__(dimension, bounds, name='Problem', dtype=np.float64, seed=None, *args, **kwargs)

Creates a new Problem object.

The problem is defined by its dimension and the bounds of each variable.

Parameters:
  • dimension (int) –

    Number of variables in the problem

  • bounds (Sequence[tuple]) –

    Bounds of each variable in the problem

  • name (str, default: 'Problem' ) –

    Name of the problem for printing and logging purposes. Defaults to "Problem".

  • dtype (_type_, default: float64 ) –

    Type of the variables. Defaults to np.float64.

  • seed (int, default: None ) –

    Seed for the random number generation (_rng). Defaults to None.

Source code in digneapy/core/_problem.py
24
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
def __init__(
    self,
    dimension: np.uint32 | int,
    bounds: Sequence[Tuple],
    name: str = "Problem",
    dtype=np.float64,
    seed: Optional[int | np.random.SeedSequence] = None,
    *args,
    **kwargs,
):
    """Creates a new Problem object.

    The problem is defined by its dimension and the bounds of each variable.

    Args:
        dimension (int): Number of variables in the problem
        bounds (Sequence[tuple]): Bounds of each variable in the problem
        name (str, optional): Name of the problem for printing and logging purposes. Defaults to "Problem".
        dtype (_type_, optional): Type of the variables. Defaults to np.float64.
        seed (int, optional): Seed for the random number generation (_rng). Defaults to None.
    """
    try:
        self._dimension = int(dimension)
        if dimension <= 0:
            raise ValueError
    except Exception as te:
        te.add_note(f"Dimension must be a postive integer. Got: {dimension}")
        raise te

    self.__name__ = name
    self._dimension = dimension
    self._bounds = bounds
    self._dtype = dtype
    if len(self._bounds) != 0:
        ranges = list(zip(*bounds))
        self._lbs = np.asarray(ranges[0], dtype=dtype)
        self._ubs = np.asarray(ranges[1], dtype=dtype)
    self._seed = seed
    self._rng = np.random.default_rng(seed)

create_solution() abstractmethod

Creates a random solution to the problem. This method can be used to initialise the solutions for any algorithm

Source code in digneapy/core/_problem.py
90
91
92
93
94
95
96
97
@abstractmethod
def create_solution(self) -> Solution | np.ndarray:
    """Creates a random solution to the problem.
    This method can be used to initialise the solutions
    for any algorithm
    """
    msg = "create_solution method not implemented in Problem"
    raise NotImplementedError(msg)

evaluate(individual) abstractmethod

Evaluates the candidate individual with the information of the problem

Parameters:
  • individual (Sequence | Solution | ndarray) –

    Individual to evaluate

Returns:
  • Tuple[float, ...]

    Tuple[float]: fitness

Source code in digneapy/core/_problem.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@abstractmethod
def evaluate(
    self, individual: Sequence | Solution | np.ndarray
) -> Tuple[float, ...]:
    """Evaluates the candidate individual with the information of the problem

    Args:
        individual (Sequence | Solution | np.ndarray): Individual to evaluate

    Returns:
        Tuple[float]: fitness
    """
    msg = "evaluate method not implemented in Problem"
    raise NotImplementedError(msg)

to_instance() abstractmethod

Creates an instance from the information of the problem. This method is used in the generators to create instances to evolve

Source code in digneapy/core/_problem.py
126
127
128
129
130
131
132
@abstractmethod
def to_instance(self) -> Instance:
    """Creates an instance from the information of the problem.
    This method is used in the generators to create instances to evolve
    """
    msg = "to_instance method not implemented in Problem"
    raise NotImplementedError(msg)

Solution

Class representing a solution in a genetic algorithm.

It contains the variables, objectives, constraints, and fitness of the solution.

Source code in digneapy/core/_solution.py
 20
 21
 22
 23
 24
 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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class Solution:
    """
    Class representing a solution in a genetic algorithm.

    It contains the variables, objectives, constraints, and fitness of the solution.
    """

    __slots__ = (
        "_variables",
        "_objectives",
        "_constraints",
        "_fitness",
        "_dtype",
        "_otype",
    )

    def __init__(
        self,
        variables: Optional[Iterable] = [],
        objectives: Optional[Iterable] = [],
        constraints: Optional[Iterable] = [],
        fitness: float | np.float64 = 0.0,
        dtype=np.uint32,
        otype=np.float64,
    ):
        """Creates a new solution object.

        The variables is a numpy array of the solution's genes.
        The objectives and constraints are numpy arrays of the solution's objectives and constraints.
        The fitness is a float representing the solution's fitness value.

        Args:
            variables (Optional[Iterable], optional): Tuple or any other iterable with the variables/variables. Defaults to None.
            objectives (Optional[Iterable], optional): Tuple or any other iterable with the objectives values. Defaults to None.
            constraints (Optional[Iterable], optional): Tuple or any other iterable with the constraint values. Defaults to None.
            fitness (float, optional): Fitness of the solution. Defaults to 0.0.
        """

        self._otype = otype
        self._dtype = dtype
        self._variables = np.asarray(variables, dtype=self.dtype)
        self._objectives = np.asarray(objectives, dtype=self._otype)
        self._constraints = np.asarray(constraints, dtype=self._otype)
        try:
            if fitness is None:
                self._fitness = np.float64(0)
            else:
                self._fitness = np.float64(fitness)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                "Fitness must be convertible to float in Solution"
            ) from exc

    @property
    def dtype(self):
        return self._dtype

    @property
    def variables(self):
        return self._variables

    @variables.setter
    def variables(self, new_variables: np.ndarray):
        if len(new_variables) != len(self._variables):
            raise ValueError(
                "Updating the variables of an Solution object with a different number of values. "
                f"Solution have {len(self._variables)} "
                f"variables and the new_variables sequence have {len(new_variables)}"
            )
        self._variables = np.asarray(new_variables)

    @property
    def fitness(self) -> np.float64 | float:
        return self._fitness

    @fitness.setter
    def fitness(self, f: np.float64 | float):
        try:
            self._fitness = float(f)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"The fitness value {f} is not a float in fitness setter of class Solution."
            ) from exc
        finally:
            self._fitness = np.float64(self._fitness)

    @property
    def objectives(self):
        return self._objectives

    @objectives.setter
    def objectives(self, new_objectives: np.ndarray):
        if len(new_objectives) != len(self._objectives):
            raise ValueError(
                "Updating the objectives of an Solution object with a different number of values. "
                f"Solution have {len(self._objectives)} "
                f"objectives and the new_objectives sequence have {len(new_objectives)}"
            )
        self._objectives = np.asarray(new_objectives)

    @property
    def constraints(self):
        return self._constraints

    @constraints.setter
    def constraints(self, new_constraints: np.ndarray):
        if len(new_constraints) != len(self._constraints):
            raise ValueError(
                "Updating the constraints of an Solution object with a different number of values. "
                f"Solution have {len(self._constraints)} "
                f"constraints and the new_constraints sequence have {len(new_constraints)}"
            )
        self._constraints = np.asarray(new_constraints)

    def clone(self) -> Self:
        """Returns a deep copy of the solution. It is more efficient than using the copy module.

        Returns:
            Self: Solution object
        """
        return type(self)(
            variables=list(self._variables),
            objectives=list(self._objectives),
            constraints=list(self._constraints),
            fitness=self._fitness,
            otype=self._otype,
        )

    def clone_with(self, **overrides):
        """Clones an Instance with overriden attributes

        Returns:
            Instance
        """
        new_object = self.clone()
        for key, value in overrides.items():
            setattr(new_object, key, value)
        return new_object

    def __str__(self) -> str:
        return f"Solution(dim={len(self.variables)},f={self.fitness},objs={self.objectives},const={self.constraints})"

    def __repr__(self) -> str:
        return f"Solution<dim={len(self.variables)},f={self.fitness},objs={self.objectives},const={self.constraints}>"

    def __len__(self) -> int:
        return len(self.variables)

    def __iter__(self):
        return iter(self.variables)

    def __eq__(self, other: Self) -> bool:
        if not isinstance(other, Solution):
            raise TypeError(
                "Other of type {other.__class__.__name__} can not be compared with a Solution."
            )
        else:
            try:
                return all(a == b for a, b in zip(self, other, strict=True))
            except ValueError:
                return False

    def __gt__(self, other: Self) -> np.bool:
        if not isinstance(other, Solution):
            raise TypeError(
                f"Other of type {other.__class__.__name__} can not be compared with a Solution."
            )

        return self.fitness > other.fitness

    def __getitem__(self, key: int | slice) -> Sequence | np.ndarray:
        """Accessor to variables of the Solution

        Args:
            key (int | slice): index or slice to access a subset of variables

        Returns:
            Sequence or np.ndarray: If accessed with a slice the subset of variables
                are returned. Otherwise, it returns a single scalar at variables[key].
        """
        if not isinstance(key, (int, slice)):
            raise TypeError(
                f"Solution cannot be indexed with type: {type(key)}. Use slice or int."
            )
        if isinstance(key, slice):
            return self._variables[key]
        else:
            index = operator.index(key)
            return self.variables[index]

    def __setitem__(self, key: int | slice, value):
        """Setter to variables of the Solution

        Args:
            key (int | slice): index or slice to access a subset of variables
            value: Value to set in the variables

        """
        if not isinstance(key, (int, np.integer, np.unsignedinteger, slice)):
            raise TypeError(
                f"Solution cannot be update via __setitem__ with type: {type(key)}. Use slice or int."
            )

        if isinstance(key, slice):
            # Compute how many positions this slice actually targets
            start, stop, step = key.indices(len(self.variables))
            target_count = len(range(start, stop, step))
            # Accept any sequence; reject bare scalars for slice assignment
            try:
                expected_len = len(value)
            except TypeError:
                raise TypeError(
                    f"[Solution] slice assignment requires a sequence, got scalar {value!r}. "
                    f"Expected {target_count} value(s)."
                )

            if expected_len != target_count:
                raise ValueError(
                    f"[Solution] slice targets {target_count} element(s) but value has "
                    f"{expected_len} element(s)."
                )
        else:
            index = operator.index(key)
            try:
                if len(value) != 1:
                    raise ValueError(
                        f"[Solution] index {index!r} targets 1 element but value has "
                        f"{len(value)} element(s). Use a slice to assign multiple values."
                    )
            except TypeError:
                pass  # len() failed which means that we have a scalar value

        self._variables[key] = value

__getitem__(key)

Accessor to variables of the Solution

Parameters:
  • key (int | slice) –

    index or slice to access a subset of variables

Returns:
  • Sequence | ndarray

    Sequence or np.ndarray: If accessed with a slice the subset of variables are returned. Otherwise, it returns a single scalar at variables[key].

Source code in digneapy/core/_solution.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def __getitem__(self, key: int | slice) -> Sequence | np.ndarray:
    """Accessor to variables of the Solution

    Args:
        key (int | slice): index or slice to access a subset of variables

    Returns:
        Sequence or np.ndarray: If accessed with a slice the subset of variables
            are returned. Otherwise, it returns a single scalar at variables[key].
    """
    if not isinstance(key, (int, slice)):
        raise TypeError(
            f"Solution cannot be indexed with type: {type(key)}. Use slice or int."
        )
    if isinstance(key, slice):
        return self._variables[key]
    else:
        index = operator.index(key)
        return self.variables[index]

__init__(variables=[], objectives=[], constraints=[], fitness=0.0, dtype=np.uint32, otype=np.float64)

Creates a new solution object.

The variables is a numpy array of the solution's genes. The objectives and constraints are numpy arrays of the solution's objectives and constraints. The fitness is a float representing the solution's fitness value.

Parameters:
  • variables (Optional[Iterable], default: [] ) –

    Tuple or any other iterable with the variables/variables. Defaults to None.

  • objectives (Optional[Iterable], default: [] ) –

    Tuple or any other iterable with the objectives values. Defaults to None.

  • constraints (Optional[Iterable], default: [] ) –

    Tuple or any other iterable with the constraint values. Defaults to None.

  • fitness (float, default: 0.0 ) –

    Fitness of the solution. Defaults to 0.0.

Source code in digneapy/core/_solution.py
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
def __init__(
    self,
    variables: Optional[Iterable] = [],
    objectives: Optional[Iterable] = [],
    constraints: Optional[Iterable] = [],
    fitness: float | np.float64 = 0.0,
    dtype=np.uint32,
    otype=np.float64,
):
    """Creates a new solution object.

    The variables is a numpy array of the solution's genes.
    The objectives and constraints are numpy arrays of the solution's objectives and constraints.
    The fitness is a float representing the solution's fitness value.

    Args:
        variables (Optional[Iterable], optional): Tuple or any other iterable with the variables/variables. Defaults to None.
        objectives (Optional[Iterable], optional): Tuple or any other iterable with the objectives values. Defaults to None.
        constraints (Optional[Iterable], optional): Tuple or any other iterable with the constraint values. Defaults to None.
        fitness (float, optional): Fitness of the solution. Defaults to 0.0.
    """

    self._otype = otype
    self._dtype = dtype
    self._variables = np.asarray(variables, dtype=self.dtype)
    self._objectives = np.asarray(objectives, dtype=self._otype)
    self._constraints = np.asarray(constraints, dtype=self._otype)
    try:
        if fitness is None:
            self._fitness = np.float64(0)
        else:
            self._fitness = np.float64(fitness)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            "Fitness must be convertible to float in Solution"
        ) from exc

__setitem__(key, value)

Setter to variables of the Solution

Parameters:
  • key (int | slice) –

    index or slice to access a subset of variables

  • value

    Value to set in the variables

Source code in digneapy/core/_solution.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def __setitem__(self, key: int | slice, value):
    """Setter to variables of the Solution

    Args:
        key (int | slice): index or slice to access a subset of variables
        value: Value to set in the variables

    """
    if not isinstance(key, (int, np.integer, np.unsignedinteger, slice)):
        raise TypeError(
            f"Solution cannot be update via __setitem__ with type: {type(key)}. Use slice or int."
        )

    if isinstance(key, slice):
        # Compute how many positions this slice actually targets
        start, stop, step = key.indices(len(self.variables))
        target_count = len(range(start, stop, step))
        # Accept any sequence; reject bare scalars for slice assignment
        try:
            expected_len = len(value)
        except TypeError:
            raise TypeError(
                f"[Solution] slice assignment requires a sequence, got scalar {value!r}. "
                f"Expected {target_count} value(s)."
            )

        if expected_len != target_count:
            raise ValueError(
                f"[Solution] slice targets {target_count} element(s) but value has "
                f"{expected_len} element(s)."
            )
    else:
        index = operator.index(key)
        try:
            if len(value) != 1:
                raise ValueError(
                    f"[Solution] index {index!r} targets 1 element but value has "
                    f"{len(value)} element(s). Use a slice to assign multiple values."
                )
        except TypeError:
            pass  # len() failed which means that we have a scalar value

    self._variables[key] = value

clone()

Returns a deep copy of the solution. It is more efficient than using the copy module.

Returns:
  • Self( Self ) –

    Solution object

Source code in digneapy/core/_solution.py
134
135
136
137
138
139
140
141
142
143
144
145
146
def clone(self) -> Self:
    """Returns a deep copy of the solution. It is more efficient than using the copy module.

    Returns:
        Self: Solution object
    """
    return type(self)(
        variables=list(self._variables),
        objectives=list(self._objectives),
        constraints=list(self._constraints),
        fitness=self._fitness,
        otype=self._otype,
    )

clone_with(**overrides)

Clones an Instance with overriden attributes

Returns:
  • Instance

Source code in digneapy/core/_solution.py
148
149
150
151
152
153
154
155
156
157
def clone_with(self, **overrides):
    """Clones an Instance with overriden attributes

    Returns:
        Instance
    """
    new_object = self.clone()
    for key, value in overrides.items():
        setattr(new_object, key, value)
    return new_object

Solver

Bases: Protocol

Protocol that defines any Solver in digneapy.

A Solver, is any callable type that receives a Problem (P) as argument and returns a list of Solution objects.

Source code in digneapy/core/_solver.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@runtime_checkable
class Solver(Protocol):
    """Protocol that defines any Solver in digneapy.

    A Solver, is any callable type that receives a Problem (P)
    as argument and returns a list of Solution objects.
    """

    @abstractmethod
    def __call__(self, problem: Problem, *args, **kwargs) -> list[Solution]:
        """Solves a optimisation problem

        Args:
            problem (Problem): Any optimisation problem or callablle that receives a Sequence and returns a Tuple[float].

        Raises:
            NotImplementedError: Must be implemented by subclasses

        Returns:
            List[Solution]: Returns a sequence of olutions
        """
        msg = "__call__ method not implemented in Solver"
        raise NotImplementedError(msg)

__call__(problem, *args, **kwargs) abstractmethod

Solves a optimisation problem

Parameters:
  • problem (Problem) –

    Any optimisation problem or callablle that receives a Sequence and returns a Tuple[float].

Raises:
  • NotImplementedError

    Must be implemented by subclasses

Returns:
  • list[Solution]

    List[Solution]: Returns a sequence of olutions

Source code in digneapy/core/_solver.py
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@abstractmethod
def __call__(self, problem: Problem, *args, **kwargs) -> list[Solution]:
    """Solves a optimisation problem

    Args:
        problem (Problem): Any optimisation problem or callablle that receives a Sequence and returns a Tuple[float].

    Raises:
        NotImplementedError: Must be implemented by subclasses

    Returns:
        List[Solution]: Returns a sequence of olutions
    """
    msg = "__call__ method not implemented in Solver"
    raise NotImplementedError(msg)

Statistics

Source code in digneapy/core/_metrics.py
 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
class Statistics:
    def __init__(self):
        self._stats_novelty = tools.Statistics(key=attrgetter("novelty"))
        self._stats_perf_bias = tools.Statistics(key=attrgetter("performance_bias"))
        self._stats_fitness = tools.Statistics(key=attrgetter("fitness"))

        self._stats = tools.MultiStatistics(
            novelty=self._stats_novelty,
            performance_bias=self._stats_perf_bias,
            fitness=self._stats_fitness,
        )
        self._stats.register("mean", np.mean)
        self._stats.register("std", np.std)
        self._stats.register("min", np.min)
        self._stats.register("max", np.max)
        self._stats.register("qd_score", np.sum)

    def __call__(self, instances: Sequence[Instance]) -> Mapping:
        """Compiles the statistics of the population or Archive.

        Args:
            instances (Sequence[Instance] | Archive): Instances to extract metrics.

        Raises:
            ValueError: If instances or archive are empty.
            TypeError: If any object in instances/archive are not an Instance or subclass of Instance.

        Returns:
            Mapping: Dict with the metrics of the instances
        """
        try:
            if len(instances) == 0:
                raise ValueError("instances is empty in Statistics.")

            # ignore np.inf values which can occur in early steps
            with np.errstate(invalid="ignore"):
                record = self._stats.compile(instances)
                return record

        except Exception as exc:
            raise RuntimeError(
                f"Unexpected error in Statistics: {exc} \nfor instances = {instances}"
            ) from exc

__call__(instances)

Compiles the statistics of the population or Archive.

Parameters:
  • instances (Sequence[Instance] | Archive) –

    Instances to extract metrics.

Raises:
  • ValueError

    If instances or archive are empty.

  • TypeError

    If any object in instances/archive are not an Instance or subclass of Instance.

Returns:
  • Mapping( Mapping ) –

    Dict with the metrics of the instances

Source code in digneapy/core/_metrics.py
 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
def __call__(self, instances: Sequence[Instance]) -> Mapping:
    """Compiles the statistics of the population or Archive.

    Args:
        instances (Sequence[Instance] | Archive): Instances to extract metrics.

    Raises:
        ValueError: If instances or archive are empty.
        TypeError: If any object in instances/archive are not an Instance or subclass of Instance.

    Returns:
        Mapping: Dict with the metrics of the instances
    """
    try:
        if len(instances) == 0:
            raise ValueError("instances is empty in Statistics.")

        # ignore np.inf values which can occur in early steps
        with np.errstate(invalid="ignore"):
            record = self._stats.compile(instances)
            return record

    except Exception as exc:
        raise RuntimeError(
            f"Unexpected error in Statistics: {exc} \nfor instances = {instances}"
        ) from exc