Fix numpy.reshape bug and add tests

Author: ahmedfgad

docs/md/pygad.md

@@ -863,13 +863,13 @@ import functools
 import operator
 
 def img2chromosome(img_arr):
-    return numpy.reshape(a=img_arr, newshape=(functools.reduce(operator.mul, img_arr.shape)))
+    return numpy.reshape(img_arr, (functools.reduce(operator.mul, img_arr.shape)))
 
 def chromosome2img(vector, shape):
     if len(vector) != functools.reduce(operator.mul, shape):
         raise ValueError(f"A vector of length {len(vector)} into an array of shape {shape}.")
 
-    return numpy.reshape(a=vector, newshape=shape)
+    return numpy.reshape(vector, shape)
  ```
 
 ### Create an Instance of the `pygad.GA` Class

docs/source/pygad.rst

@@ -1727,13 +1727,13 @@ its code is listed below.
    import operator
 
    def img2chromosome(img_arr):
-       return numpy.reshape(a=img_arr, newshape=(functools.reduce(operator.mul, img_arr.shape)))
+       return numpy.reshape(img_arr, (functools.reduce(operator.mul, img_arr.shape)))
 
    def chromosome2img(vector, shape):
        if len(vector) != functools.reduce(operator.mul, shape):
            raise ValueError(f"A vector of length {len(vector)} into an array of shape {shape}.")
 
-       return numpy.reshape(a=vector, newshape=shape)
+       return numpy.reshape(vector, shape)
 
 .. _create-an-instance-of-the-pygadga-class-2:
 

pygad/cnn/cnn.py

@@ -126,7 +126,7 @@ def layers_weights_as_matrix(model, vector_weights):
 
             weights_vector=vector_weights[start:start + layer_weights_size]
     #        matrix = pygad.nn.DenseLayer.to_array(vector=weights_vector, shape=layer_weights_shape)
-            matrix = numpy.reshape(weights_vector, newshape=(layer_weights_shape))
+            matrix = numpy.reshape(weights_vector, (layer_weights_shape))
             network_weights.append(matrix)
 
             start = start + layer_weights_size
@@ -163,11 +163,11 @@ def layers_weights_as_vector(model, initial=True):
         if type(layer) in [Conv2D, Dense]:
             # If the 'initial' parameter is True, append the initial weights. Otherwise, append the trained weights.
             if initial == True:
-                vector = numpy.reshape(layer.initial_weights, newshape=(layer.initial_weights.size))
+                vector = numpy.reshape(layer.initial_weights, (layer.initial_weights.size))
     #            vector = pygad.nn.DenseLayer.to_vector(matrix=layer.initial_weights)
                 network_weights.extend(vector)
             elif initial == False:
-                vector = numpy.reshape(layer.trained_weights, newshape=(layer.trained_weights.size))
+                vector = numpy.reshape(layer.trained_weights, (layer.trained_weights.size))
     #            vector = pygad.nn.DenseLayer.to_vector(array=layer.trained_weights)
                 network_weights.extend(vector)
             else:

pygad/kerasga/kerasga.py

@@ -23,7 +23,7 @@ def model_weights_as_vector(model):
         if layer.trainable:
             layer_weights = layer.get_weights()
             for l_weights in layer_weights:
-                vector = numpy.reshape(l_weights, newshape=(l_weights.size))
+                vector = numpy.reshape(l_weights, (l_weights.size))
                 weights_vector.extend(vector)
 
     return numpy.array(weights_vector)
@@ -57,7 +57,7 @@ def model_weights_as_matrix(model, weights_vector):
                 layer_weights_size = l_weights.size
 
                 layer_weights_vector = weights_vector[start:start + layer_weights_size]
-                layer_weights_matrix = numpy.reshape(layer_weights_vector, newshape=(layer_weights_shape))
+                layer_weights_matrix = numpy.reshape(layer_weights_vector, (layer_weights_shape))
                 weights_matrix.append(layer_weights_matrix)
 
                 start = start + layer_weights_size

pygad/nn/nn.py

@@ -56,11 +56,11 @@ def layers_weights_as_vector(last_layer, initial=True):
     while "previous_layer" in layer.__init__.__code__.co_varnames:
         # If the 'initial' parameter is True, append the initial weights. Otherwise, append the trained weights.
         if initial == True:
-            vector = numpy.reshape(layer.initial_weights, newshape=(layer.initial_weights.size))
+            vector = numpy.reshape(layer.initial_weights, (layer.initial_weights.size))
 #            vector = DenseLayer.to_vector(matrix=layer.initial_weights)
             network_weights.extend(vector)
         elif initial == False:
-            vector = numpy.reshape(layer.trained_weights, newshape=(layer.trained_weights.size))
+            vector = numpy.reshape(layer.trained_weights, (layer.trained_weights.size))
 #            vector = DenseLayer.to_vector(array=layer.trained_weights)
             network_weights.extend(vector)
         else:
@@ -98,7 +98,7 @@ def layers_weights_as_matrix(last_layer, vector_weights):
 
         weights_vector=vector_weights[start:start + layer_weights_size]
 #        matrix = DenseLayer.to_array(vector=weights_vector, shape=layer_weights_shape)
-        matrix = numpy.reshape(weights_vector, newshape=(layer_weights_shape))
+        matrix = numpy.reshape(weights_vector, (layer_weights_shape))
         network_weights.append(matrix)
 
         start = start + layer_weights_size
@@ -338,7 +338,7 @@ def to_vector(array):
     """
     if not (type(array) is numpy.ndarray):
         raise TypeError(f"An input of type numpy.ndarray is expected but an input of type {type(array)} found.")
-    return numpy.reshape(array, newshape=(array.size))
+    return numpy.reshape(array, (array.size))
 
 def to_array(vector, shape):
     """
@@ -357,7 +357,7 @@ def to_array(vector, shape):
         raise ValueError(f"A 1D NumPy array is expected but an array of {vector.ndim} dimensions found.")
     if vector.size != functools.reduce(lambda x,y:x*y, shape, 1): # (operator.mul == lambda x,y:x*y
         raise ValueError(f"Mismatch between the vector length and the array shape. A vector of length {vector.size} cannot be converted into a array of shape ({shape}).")
-    return numpy.reshape(vector, newshape=shape)
+    return numpy.reshape(vector, shape)
 
 class InputLayer:
     """

pygad/torchga/torchga.py

@@ -10,7 +10,7 @@ def model_weights_as_vector(model):
         # cpu() is called for making shore the data is moved from GPU to cpu
         # numpy() is called for converting the tensor into a NumPy array.
         curr_weights = curr_weights.cpu().detach().numpy()
-        vector = numpy.reshape(curr_weights, newshape=(curr_weights.size))
+        vector = numpy.reshape(curr_weights, (curr_weights.size))
         weights_vector.extend(vector)
 
     return numpy.array(weights_vector)
@@ -28,7 +28,7 @@ def model_weights_as_dict(model, weights_vector):
         layer_weights_size = w_matrix.size
 
         layer_weights_vector = weights_vector[start:start + layer_weights_size]
-        layer_weights_matrix = numpy.reshape(layer_weights_vector, newshape=(layer_weights_shape))
+        layer_weights_matrix = numpy.reshape(layer_weights_vector, (layer_weights_shape))
         weights_dict[key] = torch.from_numpy(layer_weights_matrix)
 
         start = start + layer_weights_size

tests/test_gann.py

@@ -0,0 +1,75 @@
+import pygad.gann
+import pygad.nn
+import numpy
+
+def test_gann_regression():
+    """Test GANN for a simple regression problem."""
+    # Data
+    data_inputs = numpy.array([[0.02, 0.1, 0.15],
+                               [0.7, 0.6, 0.8],
+                               [1.5, 1.2, 1.7],
+                               [3.2, 2.9, 3.1]])
+    data_outputs = numpy.array([0.1, 0.6, 1.3, 2.5])
+
+    # GANN architecture
+    num_inputs = data_inputs.shape[1]
+    num_classes = 1 # Regression
+
+    gann_instance = pygad.gann.GANN(num_solutions=10,
+                                     num_neurons_input=num_inputs,
+                                     num_neurons_hidden_layers=[5],
+                                     num_neurons_output=num_classes,
+                                     hidden_activations="relu",
+                                     output_activation="None")
+
+    # The number of genes is the total number of weights in the network.
+    # We can get it by converting the weights of any network in the population into a vector.
+    num_genes = len(pygad.nn.layers_weights_as_vector(last_layer=gann_instance.population_networks[0]))
+
+    def fitness_func(ga_instance, solution, solution_idx):
+        # Update the weights of the network associated with the current solution.
+        # GANN.update_population_trained_weights expects weights for ALL solutions.
+        # To avoid updating all, we can update just the one we need.
+        
+        # However, for simplicity and to test GANN's intended flow:
+        population_matrices = pygad.gann.population_as_matrices(num_networks=ga_instance.sol_per_pop,
+                                                               population_vectors=ga_instance.population)
+        gann_instance.update_population_trained_weights(population_trained_weights=population_matrices)
+
+        predictions = pygad.nn.predict(last_layer=gann_instance.population_networks[solution_idx],
+                                       data_inputs=data_inputs)
+        
+        # Mean Absolute Error
+        abs_error = numpy.mean(numpy.abs(predictions.flatten() - data_outputs)) + 0.00000001
+        fitness = 1.0 / abs_error
+        return fitness
+
+    ga_instance = pygad.GA(num_generations=5,
+                           num_parents_mating=4,
+                           fitness_func=fitness_func,
+                           sol_per_pop=10,
+                           num_genes=num_genes,
+                           random_seed=42,
+                           suppress_warnings=True)
+
+    ga_instance.run()
+    assert ga_instance.run_completed
+    print("test_gann_regression passed.")
+
+def test_nn_direct_usage():
+    """Test pygad.nn layers directly."""
+    input_layer = pygad.nn.InputLayer(num_inputs=3)
+    dense_layer = pygad.nn.DenseLayer(num_neurons=2, previous_layer=input_layer, activation_function="relu")
+    output_layer = pygad.nn.DenseLayer(num_neurons=1, previous_layer=dense_layer, activation_function="sigmoid")
+    
+    data_inputs = numpy.array([[0.1, 0.2, 0.3]])
+    predictions = pygad.nn.predict(last_layer=output_layer, data_inputs=data_inputs)
+    
+    assert predictions.shape == (1, 1)
+    assert 0 <= predictions[0, 0] <= 1
+    print("test_nn_direct_usage passed.")
+
+if __name__ == "__main__":
+    test_gann_regression()
+    test_nn_direct_usage()
+    print("\nAll GANN/NN tests passed!")

tests/test_operators.py

@@ -0,0 +1,77 @@
+import pygad
+import numpy
+import random
+
+# Global constants for testing
+num_generations = 5
+num_parents_mating = 4
+sol_per_pop = 10
+num_genes = 10
+random_seed = 42
+
+def fitness_func(ga_instance, solution, solution_idx):
+    return numpy.sum(solution)
+
+def fitness_func_multi(ga_instance, solution, solution_idx):
+    return [numpy.sum(solution), numpy.sum(solution**2)]
+
+def run_ga_with_params(parent_selection_type='sss', crossover_type='single_point', mutation_type='random', multi_objective=False):
+    if multi_objective:
+        f = fitness_func_multi
+    else:
+        f = fitness_func
+        
+    ga_instance = pygad.GA(num_generations=num_generations,
+                           num_parents_mating=num_parents_mating,
+                           fitness_func=f,
+                           sol_per_pop=sol_per_pop,
+                           num_genes=num_genes,
+                           parent_selection_type=parent_selection_type,
+                           crossover_type=crossover_type,
+                           mutation_type=mutation_type,
+                           random_seed=random_seed,
+                           suppress_warnings=True)
+    ga_instance.run()
+    return ga_instance
+
+def test_selection_operators():
+    operators = ['sss', 'rws', 'sus', 'rank', 'random', 'tournament']
+    for op in operators:
+        ga = run_ga_with_params(parent_selection_type=op)
+        # Verify parents were selected
+        assert ga.last_generation_parents.shape == (num_parents_mating, num_genes)
+        print(f"Selection operator '{op}' passed.")
+
+def test_crossover_operators():
+    operators = ['single_point', 'two_points', 'uniform', 'scattered']
+    for op in operators:
+        ga = run_ga_with_params(crossover_type=op)
+        # Verify population shape
+        assert ga.population.shape == (sol_per_pop, num_genes)
+        print(f"Crossover operator '{op}' passed.")
+
+def test_mutation_operators():
+    operators = ['random', 'swap', 'inversion', 'scramble']
+    for op in operators:
+        ga = run_ga_with_params(mutation_type=op)
+        # Verify population shape
+        assert ga.population.shape == (sol_per_pop, num_genes)
+        print(f"Mutation operator '{op}' passed.")
+
+def test_multi_objective_selection():
+    # NSGA-II is usually used for multi-objective
+    ga = run_ga_with_params(parent_selection_type='nsga2', multi_objective=True)
+    assert ga.last_generation_parents.shape == (num_parents_mating, num_genes)
+    print("Multi-objective selection (nsga2) passed.")
+    
+    # Tournament NSGA-II
+    ga = run_ga_with_params(parent_selection_type='tournament_nsga2', multi_objective=True)
+    assert ga.last_generation_parents.shape == (num_parents_mating, num_genes)
+    print("Multi-objective selection (tournament_nsga2) passed.")
+
+if __name__ == "__main__":
+    test_selection_operators()
+    test_crossover_operators()
+    test_mutation_operators()
+    test_multi_objective_selection()
+    print("\nAll operator tests passed!")

tests/test_parallel.py

@@ -0,0 +1,73 @@
+import pygad
+import numpy
+import time
+
+# Global constants for testing
+num_generations = 5
+num_parents_mating = 4
+sol_per_pop = 10
+num_genes = 3
+random_seed = 42
+
+def fitness_func(ga_instance, solution, solution_idx):
+    # Simulate some work
+    # time.sleep(0.01)
+    return numpy.sum(solution**2)
+
+def fitness_func_batch(ga_instance, solutions, indices):
+    return [numpy.sum(s**2) for s in solutions]
+
+def test_parallel_thread():
+    """Test parallel_processing with 'thread' mode."""
+    ga_instance = pygad.GA(num_generations=num_generations,
+                           num_parents_mating=num_parents_mating,
+                           fitness_func=fitness_func,
+                           sol_per_pop=sol_per_pop,
+                           num_genes=num_genes,
+                           parallel_processing=["thread", 2],
+                           random_seed=random_seed,
+                           suppress_warnings=True
+                           )
+    ga_instance.run()
+    assert ga_instance.run_completed
+    print("test_parallel_thread passed.")
+
+def test_parallel_process():
+    """Test parallel_processing with 'process' mode."""
+    # Note: 'process' mode might be tricky in some environments (e.g. Windows without if __name__ == '__main__':)
+    # But for a CI environment it should be tested.
+    ga_instance = pygad.GA(num_generations=num_generations,
+                           num_parents_mating=num_parents_mating,
+                           fitness_func=fitness_func,
+                           sol_per_pop=sol_per_pop,
+                           num_genes=num_genes,
+                           parallel_processing=["process", 2],
+                           random_seed=random_seed,
+                           suppress_warnings=True
+                           )
+    ga_instance.run()
+    assert ga_instance.run_completed
+    print("test_parallel_process passed.")
+
+def test_parallel_thread_batch():
+    """Test parallel_processing with 'thread' mode and batch fitness."""
+    ga_instance = pygad.GA(num_generations=num_generations,
+                           num_parents_mating=num_parents_mating,
+                           fitness_func=fitness_func_batch,
+                           sol_per_pop=sol_per_pop,
+                           num_genes=num_genes,
+                           parallel_processing=["thread", 2],
+                           fitness_batch_size=2,
+                           random_seed=random_seed,
+                           suppress_warnings=True
+                           )
+    ga_instance.run()
+    assert ga_instance.run_completed
+    print("test_parallel_thread_batch passed.")
+
+if __name__ == "__main__":
+    # For 'process' mode to work on Windows/macOS, we need this guard
+    test_parallel_thread()
+    test_parallel_process()
+    test_parallel_thread_batch()
+    print("\nAll parallel tests passed!")