Merging Analysis alpha digits with main

requirements.txt

@@ -1,3 +1,7 @@
 numpy
 scipy
-numba
+numba
+matplotlib
+pandas
+tqdm
+torch

setup.py

@@ -0,0 +1,12 @@
+from setuptools import setup, find_packages
+
+setup(
+    name='generative_model',
+    version='0.1',
+    author='Yedidia AGNIMO & C. Yann Éric CHOHO',
+    author_email='yedidia.agnimo@ensae.fr // chohoyanneric.choho@ensae.fr',
+    description='Implement basics deep learning architecture for generative models.',
+    packages=find_packages(where='src'),
+    package_dir={'': 'src'},
+    python_requires='>=3.10',
+)

src/principal_dbn_alpha.py

@@ -0,0 +1,125 @@
+"""
+Module: dbn.py
+Module providing implementation of Deep Belief Network (DBN).
+"""
+
+from typing import List
+
+import numpy as np
+from principal_rbm_alpha import RBM
+
+
+class DBN:
+    """
+    Implementation of a Deep Belief Network (DBN).
+
+    Attributes:
+    - n_visible (int): Number of visible units.
+    - hidden_layer_sizes (List[int]): List of sizes for each hidden layer.
+    - rbms (List[RBM]): List of Restricted Boltzmann Machines (RBMs) forming the DBN.
+    - rng (numpy.random.Generator): Random number generator for sampling.
+    """
+
+    def __init__(
+            self,
+            n_visible: int,
+            hidden_layer_sizes: List[int],
+            random_state=None
+    ):
+        """
+        Initialize the Deep Belief Network.
+
+        Parameters:
+        - n_visible (int): Number of visible units.
+        - hidden_layer_sizes (List[int]): List of sizes for each hidden layer.
+        - random_state: Random seed for reproducibility.
+        """
+        self.n_visible = n_visible
+        self.hidden_layer_sizes = hidden_layer_sizes
+        self.rbms: List[RBM] = []
+        self.rng = np.random.default_rng(random_state)
+
+        # Initialize the first RBM
+        first_rbm = RBM(
+            n_visible=n_visible,
+            n_hidden=hidden_layer_sizes[0],
+            random_state=random_state,
+        )
+        self.rbms.append(first_rbm)
+
+        # Initialize RBMs for subsequent hidden layers
+        for i, size in enumerate(hidden_layer_sizes[1:], start=1):
+            rbm = RBM(
+                n_visible=hidden_layer_sizes[i - 1],
+                n_hidden=size,
+                random_state=random_state,
+            )
+            self.rbms.append(rbm)
+
+    def __getitem__(self, key):
+        return self.rbms[key]
+
+    def __repr__(self):
+        """
+        Return a string representation of the DBN object.
+        """
+        rbm_reprs = [repr(rbm) for rbm in self.rbms]
+        join_rbm_reprs = ",\n ".join(rbm_reprs)
+        return f"DBN([\n {join_rbm_reprs}\n])"
+
+    def train(
+        self,
+        data: np.ndarray,
+        learning_rate: float = 0.1,
+        n_epochs: int = 10,
+        batch_size: int = 10,
+        print_each: int = 10,
+    ) -> "DBN":
+        """
+        Train the DBN using Greedy layer-wise procedure.
+
+        Parameters:
+        - data (numpy.ndarray): Input data, shape (n_samples, n_visible).
+        - learning_rate (float): Learning rate for gradient descent. Default is 0.1.
+        - n_epochs (int): Number of training epochs. Default is 10.
+        - batch_size (int): Size of mini-batches. Default is 10.
+        - print_each: Print reconstruction error each `print_each` epochs.
+
+        Returns:
+        - DBN: Trained DBN instance.
+        """
+        input_data = data
+        for rbm in self.rbms:
+            rbm.train(
+                input_data,
+                learning_rate=learning_rate,
+                n_epochs=n_epochs,
+                batch_size=batch_size,
+                print_each=print_each,
+            )
+            # Update input data for the next RBM
+            input_data = rbm.input_output(input_data)
+
+        return self
+
+    def generate_image(self, n_samples: int = 1, n_gibbs_steps: int = 1) -> np.ndarray:
+        """
+        Generate samples from the DBN using Gibbs sampling.
+
+        Parameters:
+        - n_samples (int): Number of samples to generate. Default is 1.
+        - n_gibbs_steps (int): Number of Gibbs sampling steps. Default is 100.
+
+        Returns:
+        - numpy.ndarray: Generated samples, shape (n_samples, n_visible).
+        """
+        # samples = np.zeros((n_samples, self.n_visible))
+
+        # Generate samples using the first RBM in the DBN
+        samples = self.rbms[-1].generate_image(n_samples, n_gibbs_steps)
+        for rbm in reversed(self.rbms[:-1]):
+            # Sample from the conditional probability of layer l-1 given layer l: p(h_{s-1}|h_{s}).
+            h_probs = rbm.output_input(samples)
+            h = self.rng.binomial(1, p=h_probs)
+            samples = h
+        return samples

src/principal_rbm_alpha.py

@@ -1,4 +1,15 @@
 """Principal RBM alpha.
+#TODO: control for verbosity (add 'verbose' arg / think about where to progression bar with `tqdm`) 
+#TODO: add a representation "__repr__" to the class. Look like `RBM(n_visible, n_hidden, rng)`.
+#TODO: move `sigmoid`, and function related to data into others modules (utils, load_data).
+# HACK: optimize code, accelerate matrix computation with numba, parallelized when possible.
+#TODO: check relevance of using the RBM's RNG for generation phase (look inside the gibbs sampling).
+# If a seed has been define, the gibbs sampling step will return the same sample for each it will
+# sample the same h from the binomial -> #WARNING there might be something wrong with the function
+# --------------------------- Other Tags (Example usage) ---------------------.
+# FIXME: Example: This function is returning incorrect results for negative input values.
+# BUG: Example: Division by zero error occurs in certain cases.
+# HACK: Example: This code temporarily fixes the issue, but needs a proper solution.
 """
 
 import os
@@ -169,7 +180,7 @@ def _reconstruction_error(
         """
         return np.round(np.power(output_img - input_img, 2).mean(), 3)
 
-    def entree_sortie(self, data: np.ndarray) -> np.ndarray:
+    def input_output(self, data: np.ndarray) -> np.ndarray:
         """
         Compute hidden units given visible units.
 
@@ -181,7 +192,7 @@ def entree_sortie(self, data: np.ndarray) -> np.ndarray:
         """
         return self._sigmoid(data @ self.W + self.b)
 
-    def sortie_entree(self, data_h: np.ndarray) -> np.ndarray:
+    def output_input(self, data_h: np.ndarray) -> np.ndarray:
         """
         Compute visible units given hidden units.
 
@@ -218,9 +229,9 @@ def train(
             self.rng.shuffle(data)
             for i in tqdm(range(0, n_samples, batch_size), desc=f"Epoch {epoch}"):
                 batch = data[i: i + batch_size]
-                pos_h_probs = self.entree_sortie(batch)
-                pos_v_probs = self.sortie_entree(pos_h_probs)
-                neg_h_probs = self.entree_sortie(pos_v_probs)
+                pos_h_probs = self.input_output(batch)
+                pos_v_probs = self.output_input(pos_h_probs)
+                neg_h_probs = self.input_output(pos_v_probs)
 
                 # Update weights and biases
                 self.W += (
@@ -242,7 +253,7 @@ def train(
 
         return self
 
-    def generer_image(self, n_samples: int = 1, n_gibbs_steps: int = 1) -> np.ndarray:
+    def generate_image(self, n_samples: int=1, n_gibbs_steps: int=1) -> np.ndarray:
         """
         Generate samples from the RBM using Gibbs sampling.
 
@@ -260,8 +271,8 @@ def generer_image(self, n_samples: int = 1, n_gibbs_steps: int = 1) -> np.ndarra
             1, self.rng.random(), size=n_samples * self.n_visible
         ).reshape((n_samples, self.n_visible))
         for i in range(n_samples):
+            h_probs = self._sigmoid(V[i] @ self.W + self.b)
             for _ in range(n_gibbs_steps):
-                h_probs = self._sigmoid(V[i] @ self.W + self.b)
                 h = self.rng.binomial(1, h_probs)
                 v_probs = self._sigmoid(h @ self.W.T + self.a)
                 v = self.rng.binomial(1, v_probs)

src/tests/test_rbm.py

@@ -0,0 +1,62 @@
+import os
+import numpy as np
+import scipy.io
+import unittest
+from rbm import RBM, _load_data, _map_character_to_index, read_alpha_digit
+
+DATA_FOLDER = "../data/"
+ALPHA_DIGIT_PATH = os.path.join(DATA_FOLDER, "binaryalphadigs.mat")
+
+
+class TestRBM(unittest.TestCase):
+    def setUp(self):
+        # Load alpha_digit data for testing
+        self.data = read_alpha_digit(file_path=ALPHA_DIGIT_PATH, character='A')
+        self.n_samples = self.data.shape[0]
+        self.n_visible = self.data.shape[1]
+        self.n_hidden = 100
+        self.rbm = RBM(n_visible=self.n_visible, n_hidden=self.n_hidden)
+
+    def test__sigmoid(self):
+        # Test _sigmoid method with positive and negative values
+        x = np.array([1, -1, 0])
+        sigmoid_x = self.rbm._sigmoid(x)
+        self.assertTrue(np.allclose(sigmoid_x, [0.73105858, 0.26894142, 0.5]))
+
+    def test__reconstruction_error(self):
+        # Test _reconstruction_error method with arrays containing zeros
+        input_img = np.zeros_like(self.data)
+        output_img = np.zeros_like(self.data)
+        error = self.rbm._reconstruction_error(input_img, output_img)
+        self.assertEqual(error, 0.0)
+
+    def test_entree_sortie(self):
+        # Test entree_sortie method with a small input array
+        input_data = np.ones((2, self.n_visible))
+        output = self.rbm.entree_sortie(input_data)
+        self.assertEqual(output.shape, (2, self.n_hidden))
+
+    def test_sortie_entree(self):
+        # Test sortie_entree method with a small input array
+        input_data = np.ones((2, self.n_hidden))
+        output = self.rbm.sortie_entree(input_data)
+        self.assertEqual(output.shape, (2, self.n_visible))
+
+    def test_train(self):
+        # Test train method
+        trained_rbm = self.rbm.train(
+            self.data[:100], learning_rate=0.1, n_epochs=1, batch_size=10
+        )
+        self.assertTrue(hasattr(trained_rbm, 'W'))
+        self.assertTrue(hasattr(trained_rbm, 'a'))
+        self.assertTrue(hasattr(trained_rbm, 'b'))
+
+    def test_generer_image(self):
+        # Test generer_image method
+        samples = self.rbm.generer_image(n_samples=2, n_gibbs_steps=10)
+        self.assertEqual(samples.shape, (2, self.n_visible))
+        self.assertTrue(np.all((samples == 0) | (samples == 1)))
+
+
+if __name__ == "__main__":
+    unittest.main()