@File : _neural_networks.py @Time : 2024/09/12 14:21:29 @Author : Alejandro Marrero @Version : 1.0 @Contact : amarrerd@ull.edu.es @License : (C)Copyright 2024, Alejandro Marrero @Desc : None

NNEncoder

Bases: Transformer

Source code in digneapy/transformers/neural.py
 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
class NNEncoder(Transformer):
    def __init__(
        self,
        name: str,
        input_shape: Sequence[int],
        shape: Sequence[int],
        activations: Sequence[Optional[str]],
        evaluation_metric: Optional[Callable] = None,
        loss_fn: Optional[Callable] = None,
        optimizer: Optional[keras.Optimizer] = keras.optimizers.Nadam(),
        scale: bool = True,
    ):
        """Neural Network used to transform a space into another. This class uses a Keras backend.

        Args:
            name (str): Name of the model to be saved with. Expected a .keras extension.
            shape (Tuple[int]): Tuple with the number of cells per layer.
            activations (Tuple[str]): Activation functions for each layer.
        Raises:
            ValueError: Raises if any attribute is not valid.
        """
        if len(activations) != len(shape):
            msg = f"Expected {len(shape)} activation functions but only got {len(activations)}"
            raise ValueError(msg)
        if not name.endswith(".keras"):
            name = name + ".keras"

        super().__init__(name)
        self.scaler = StandardScaler()
        self.input_shape = input_shape
        self._shape = shape
        self._activations = activations
        self._eval_metric = evaluation_metric
        self._loss_fn = loss_fn
        self._optimizer = optimizer

        self._model = keras.models.Sequential()
        self._model.add(keras.layers.InputLayer(shape=input_shape))
        for d, act in zip(shape, activations):
            self._model.add(keras.layers.Dense(d, activation=act))
        self._model.compile(
            loss=self._loss_fn, optimizer=optimizer, metrics=[self._eval_metric]
        )
        self._expected_shapes = [v.shape for v in self._model.trainable_variables]
        self._expected_sizes = [np.prod(s) for s in self._expected_shapes]
        self._size = np.sum(self._expected_sizes)

    def __str__(self) -> str:
        tokens = []
        self._model.summary(print_fn=lambda x: tokens.append(x))
        return "\n".join(tokens)

    def __repr__(self) -> str:
        return self.__str__()

    def save(self, filename: Optional[str] = None):
        if filename is not None:
            self._model.save(filename)
        else:
            self._model.save(self._name)

    def update_weights(self, weights: Sequence[float]):
        if len(weights) != self._size:
            msg = f"Error in the amount of weights in NN.update_weigths. Expected {self._size} and got {len(weights)}"
            raise ValueError(msg)

        new_weights = [None] * len(self._expected_shapes)
        idx = 0
        i = 0
        for shape, size in zip(self._expected_shapes, self._expected_sizes):
            new_weights[i] = np.reshape(weights[idx : idx + size], shape)
            idx += size
            i += 1

        self._model.set_weights(new_weights)
        return True

    def predict(self, x: npt.NDArray, batch_size: int = 1024) -> np.ndarray:
        if x is None or len(x) == 0:
            msg = "x cannot be None in KerasNN predict"
            raise RuntimeError(msg)
        if isinstance(x, list):
            x = np.vstack(x)
        elif x.ndim == 1:
            x.reshape(1, -1)
        x_scaled = self.scaler.fit_transform(x)
        return self._model.predict(x_scaled, batch_size=batch_size, verbose=0)

    def __call__(self, x: npt.NDArray) -> np.ndarray:
        return self.predict(x)

__init__(name, input_shape, shape, activations, evaluation_metric=None, loss_fn=None, optimizer=keras.optimizers.Nadam(), scale=True)

Neural Network used to transform a space into another. This class uses a Keras backend.

Parameters:
  • name (str) –

    Name of the model to be saved with. Expected a .keras extension.

  • shape (Tuple[int]) –

    Tuple with the number of cells per layer.

  • activations (Tuple[str]) –

    Activation functions for each layer.

Raises: ValueError: Raises if any attribute is not valid.

Source code in digneapy/transformers/neural.py
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
def __init__(
    self,
    name: str,
    input_shape: Sequence[int],
    shape: Sequence[int],
    activations: Sequence[Optional[str]],
    evaluation_metric: Optional[Callable] = None,
    loss_fn: Optional[Callable] = None,
    optimizer: Optional[keras.Optimizer] = keras.optimizers.Nadam(),
    scale: bool = True,
):
    """Neural Network used to transform a space into another. This class uses a Keras backend.

    Args:
        name (str): Name of the model to be saved with. Expected a .keras extension.
        shape (Tuple[int]): Tuple with the number of cells per layer.
        activations (Tuple[str]): Activation functions for each layer.
    Raises:
        ValueError: Raises if any attribute is not valid.
    """
    if len(activations) != len(shape):
        msg = f"Expected {len(shape)} activation functions but only got {len(activations)}"
        raise ValueError(msg)
    if not name.endswith(".keras"):
        name = name + ".keras"

    super().__init__(name)
    self.scaler = StandardScaler()
    self.input_shape = input_shape
    self._shape = shape
    self._activations = activations
    self._eval_metric = evaluation_metric
    self._loss_fn = loss_fn
    self._optimizer = optimizer

    self._model = keras.models.Sequential()
    self._model.add(keras.layers.InputLayer(shape=input_shape))
    for d, act in zip(shape, activations):
        self._model.add(keras.layers.Dense(d, activation=act))
    self._model.compile(
        loss=self._loss_fn, optimizer=optimizer, metrics=[self._eval_metric]
    )
    self._expected_shapes = [v.shape for v in self._model.trainable_variables]
    self._expected_sizes = [np.prod(s) for s in self._expected_shapes]
    self._size = np.sum(self._expected_sizes)