Update app.py

app.py

@@ -2,18 +2,312 @@ import gradio as gr
 import numpy as np
 import matplotlib.pyplot as plt
 
-# Include your SNN code here, for brevity I'll only show the Gradio interface part.
+def relu(x):
+    return np.maximum(0, x)
+
+def relu_derivative(x):
+    return (x > 0).astype(float)
+
+def tanh(x):
+    return np.tanh(x)
+
+def tanh_derivative(x):
+    return 1 - np.tanh(x)**2
+
+class EveOptimizer:
+    def __init__(self, params, learning_rate=0.001, beta1=0.9, beta2=0.999, epsilon=1e-8):
+        self.params = params
+        self.lr = learning_rate
+        self.beta1 = beta1
+        self.beta2 = beta2
+        self.epsilon = epsilon
+        self.t = 0
+        self.m = [np.zeros_like(p) for p in params]
+        self.v = [np.zeros_like(p) for p in params]
+        self.fractal_memory = [np.zeros_like(p) for p in params]
+
+    def step(self, grads):
+        self.t += 1
+        for i, (param, grad) in enumerate(zip(self.params, grads)):
+            self.m[i] = self.beta1 * self.m[i] + (1 - self.beta1) * grad
+            self.v[i] = self.beta2 * self.v[i] + (1 - self.beta2) * (grad ** 2)
+
+            m_hat = self.m[i] / (1 - self.beta1 ** self.t)
+            v_hat = self.v[i] / (1 - self.beta2 ** self.t)
+
+            fractal_factor = self.fractal_adjustment(param, grad)
+            self.fractal_memory[i] = 0.9 * self.fractal_memory[i] + 0.1 * fractal_factor
+
+            param -= self.lr * m_hat / (np.sqrt(v_hat) + self.epsilon) * self.fractal_memory[i]
+
+    def fractal_adjustment(self, param, grad):
+        c = np.mean(grad) + 1j * np.std(param)
+        z = 0
+        for _ in range(10):
+            z = z**2 + c
+            if abs(z) > 2:
+                break
+        return 1 / (1 + abs(z))
+
+class BatchNormalization:
+    def __init__(self, input_shape):
+        self.gamma = np.ones(input_shape)
+        self.beta = np.zeros(input_shape)
+        self.epsilon = 1e-5
+        self.moving_mean = np.zeros(input_shape)
+        self.moving_var = np.ones(input_shape)
+
+    def forward(self, x, training=True):
+        if training:
+            mean = np.mean(x, axis=0)
+            var = np.var(x, axis=0)
+            self.moving_mean = 0.99 * self.moving_mean + 0.01 * mean
+            self.moving_var = 0.99 * self.moving_var + 0.01 * var
+        else:
+            mean = self.moving_mean
+            var = self.moving_var
+
+        x_norm = (x - mean) / np.sqrt(var + self.epsilon)
+        out = self.gamma * x_norm + self.beta
+        if training:
+            self.cache = (x, x_norm, mean, var)
+        return out
+
+    def backward(self, dout):
+        x, x_norm, mean, var = self.cache
+        m = x.shape[0]
+
+        dx_norm = dout * self.gamma
+        dvar = np.sum(dx_norm * (x - mean) * -0.5 * (var + self.epsilon)**(-1.5), axis=0)
+        dmean = np.sum(dx_norm * -1 / np.sqrt(var + self.epsilon), axis=0) + dvar * np.mean(-2 * (x - mean), axis=0)
+
+        dx = dx_norm / np.sqrt(var + self.epsilon) + dvar * 2 * (x - mean) / m + dmean / m
+        dgamma = np.sum(dout * x_norm, axis=0)
+        dbeta = np.sum(dout, axis=0)
+
+        return dx, dgamma, dbeta
+
+class Reward:
+    def __init__(self):
+        self.lowest_avg_batch_loss = float('inf')
+        self.lowest_max_batch_loss = float('inf')
+        self.best_weights = None
+
+    def update(self, avg_batch_loss, max_batch_loss, network):
+        improved = False
+        if avg_batch_loss < self.lowest_avg_batch_loss:
+            self.lowest_avg_batch_loss = avg_batch_loss
+            improved = True
+        if max_batch_loss < self.lowest_max_batch_loss:
+            self.lowest_max_batch_loss = max_batch_loss
+            improved = True
+        if improved:
+            self.best_weights = self.get_network_weights(network)
+
+    def get_network_weights(self, network):
+        weights = []
+        for layer in network.layers:
+            layer_weights = []
+            for agent in layer.agents:
+                agent_weights = {
+                    'weights': agent.weights.copy(),
+                    'bias': agent.bias.copy(),
+                    'bn_gamma': agent.bn.gamma.copy(),
+                    'bn_beta': agent.bn.beta.copy()
+                }
+                layer_weights.append(agent_weights)
+            weights.append(layer_weights)
+        return weights
+
+    def apply_best_weights(self, network):
+        if self.best_weights is not None:
+            for layer, layer_weights in zip(network.layers, self.best_weights):
+                for agent, agent_weights in zip(layer.agents, layer_weights):
+                    agent.weights = agent_weights['weights'].copy()
+                    agent.bias = agent_weights['bias'].copy()
+                    agent.bn.gamma = agent_weights['bn_gamma'].copy()
+                    agent.bn.beta = agent_weights['bn_beta'].copy()
+
+class Agent:
+    def __init__(self, id, input_size, output_size, fractal_method):
+        self.id = id
+        self.weights = np.random.randn(input_size, output_size) * np.sqrt(2. / input_size)
+        self.bias = np.zeros((1, output_size))
+        self.fractal_method = fractal_method
+        self.bn = BatchNormalization((output_size,))
+
+        self.optimizer = EveOptimizer([self.weights, self.bias, self.bn.gamma, self.bn.beta])
+
+    def forward(self, x, training=True):
+        self.last_input = x
+        z = np.dot(x, self.weights) + self.bias
+        z_bn = self.bn.forward(z, training)
+        self.last_output = relu(z_bn)
+        return self.last_output
+
+    def backward(self, error, l2_lambda=1e-5):
+        delta = error * relu_derivative(self.last_output)
+        delta, dgamma, dbeta = self.bn.backward(delta)
+
+        dw = np.dot(self.last_input.T, delta) + l2_lambda * self.weights
+        db = np.sum(delta, axis=0, keepdims=True)
+
+        self.optimizer.step([dw, db, dgamma, dbeta])
+
+        return np.dot(delta, self.weights.T)
+
+    def apply_fractal(self, x):
+        return self.fractal_method(x)
+
+class Swarm:
+    def __init__(self, num_agents, input_size, output_size, fractal_method):
+        self.agents = [Agent(i, input_size, output_size, fractal_method) for i in range(num_agents)]
+
+    def forward(self, x, training=True):
+        results = [agent.forward(x, training) for agent in self.agents]
+        return np.mean(results, axis=0)
+
+    def backward(self, error, l2_lambda):
+        errors = [agent.backward(error, l2_lambda) for agent in self.agents]
+        return np.mean(errors, axis=0)
+
+    def apply_fractal(self, x):
+        results = [agent.apply_fractal(x) for agent in self.agents]
+        return np.mean(results, axis=0)
+
+class SwarmNeuralNetwork:
+    def __init__(self, layer_sizes, fractal_methods):
+        self.layers = []
+        for i in range(len(layer_sizes) - 2):
+            self.layers.append(Swarm(num_agents=3,
+                                     input_size=layer_sizes[i],
+                                     output_size=layer_sizes[i+1],
+                                     fractal_method=fractal_methods[i]))
+        self.output_layer = Swarm(num_agents=1,
+                                  input_size=layer_sizes[-2],
+                                  output_size=layer_sizes[-1],
+                                  fractal_method=fractal_methods[-1])
+        self.reward = Reward()
+
+    def forward(self, x, training=True):
+        self.layer_outputs = [x]
+        for layer in self.layers:
+            x = layer.forward(x, training)
+            self.layer_outputs.append(x)
+        self.final_output = tanh(self.output_layer.forward(x, training))
+        return self.final_output
+
+    def backward(self, error, l2_lambda=1e-5):
+        error = error * tanh_derivative(self.final_output)
+        error = self.output_layer.backward(error, l2_lambda)
+        for i in reversed(range(len(self.layers))):
+            error = self.layers[i].backward(error, l2_lambda)
+
+    def train(self, X, y, epochs, batch_size=32, l2_lambda=1e-5, patience=50):
+        best_mse = float('inf')
+        patience_counter = 0
+
+        for epoch in range(epochs):
+            indices = np.arange(len(X))
+            np.random.shuffle(indices)
+
+            self.reward.apply_best_weights(self)
+
+            epoch_losses = []
+            for start_idx in range(0, len(X) - batch_size + 1, batch_size):
+                batch_indices = indices[start_idx:start_idx+batch_size]
+                X_batch = X[batch_indices]
+                y_batch = y[batch_indices]
+
+                output = self.forward(X_batch)
+                error = y_batch - output
+
+                error = np.clip(error, -1, 1)
+
+                self.backward(error, l2_lambda)
+
+                epoch_losses.append(np.mean(np.square(error)))
+
+            avg_batch_loss = np.mean(epoch_losses)
+            max_batch_loss = np.max(epoch_losses)
+            self.reward.update(avg_batch_loss, max_batch_loss, self)
+
+            mse = np.mean(np.square(y - self.forward(X, training=False)))
+
+            if epoch % 100 == 0:
+                print(f"Epoch {epoch}, MSE: {mse:.6f}, Avg Batch Loss: {avg_batch_loss:.6f}, Min Batch Loss: {np.min(epoch_losses):.6f}, Max Batch Loss: {max_batch_loss:.6f}")
+
+            if mse < best_mse:
+                best_mse = mse
+                patience_counter = 0
+            else:
+                patience_counter += 1
+
+            if patience_counter >= patience:
+                print(f"Early stopping at epoch {epoch}")
+                break
+
+        return best_mse
+
+    def apply_fractals(self, x):
+        fractal_outputs = []
+        for i, layer in enumerate(self.layers):
+            x = self.layer_outputs[i+1]
+            fractal_output = layer.apply_fractal(x)
+            fractal_outputs.append(fractal_output)
+        return fractal_outputs
+
+def sierpinski_fractal(input_data):
+    t = np.linspace(0, 2 * np.pi, input_data.shape[0])
+    x = np.mean(input_data) * np.cos(t)
+    y = np.mean(input_data) * np.sin(t)
+    return x, y
+
+def mandelbrot_fractal(input_data, max_iter=10):
+    output = np.zeros(input_data.shape[0])
+    for i in range(input_data.shape[0]):
+        c = input_data[i, 0] + 0.1j * np.std(input_data)
+        z = 0
+        for n in range(max_iter):
+            if abs(z) > 2:
+                output[i] = n
+                break
+            z = z*z + c
+        else:
+            output[i] = max_iter
+    return output
+
+def julia_fractal(input_data, max_iter=10):
+    output = np.zeros(input_data.shape[0])
+    c = -0.8 + 0.156j
+    for i in range(input_data.shape[0]):
+        z = input_data[i, 0] + 0.1j * np.std(input_data)
+        for n in range(max_iter):
+            if abs(z) > 2:
+                output[i] = n
+                break
+            z = z*z + c
+        else:
+            output[i] = max_iter
+    return output
+
+def run_snn(epochs, batch_size, l2_lambda, patience):
+    np.random.seed(42)
+
+    X = np.linspace(0, 10, 1000).reshape(-1, 1)
+    y = np.sin(X).reshape(-1, 1)
+
+    X = (X - X.min()) / (X.max() - X.min())
+    y = (y - y.min()) / (y.max() - y.min())
 
-def run_snn(X, epochs, batch_size, l2_lambda, patience):
-    # Your SNN initialization and training code here
     snn = SwarmNeuralNetwork(layer_sizes=[1, 32, 16, 8, 1],
                              fractal_methods=[sierpinski_fractal, mandelbrot_fractal, julia_fractal, julia_fractal])
+
     snn.train(X, y, epochs=epochs, batch_size=batch_size, l2_lambda=l2_lambda, patience=patience)
+
     y_pred = snn.forward(X, training=False)
     fractal_outputs = snn.apply_fractals(X)
-    return y_pred, fractal_outputs
 
-def plot_results(X, y, y_pred, fractal_outputs):
     fig, axs = plt.subplots(2, 2, figsize=(15, 10))
     
     axs[0, 0].plot(X, y, label='True')
@@ -31,26 +325,21 @@ def plot_results(X, y, y_pred, fractal_outputs):
     axs[1, 1].plot(X, fractal_outputs[2])
     axs[1, 1].set_title('Julia Fractal Output')
 
+    plt.tight_layout()
     return fig
 
-def main_interface(epochs, batch_size, l2_lambda, patience):
-    X = np.linspace(0, 10, 1000).reshape(-1, 1)
-    y = np.sin(X).reshape(-1, 1)
-    X = (X - X.min()) / (X.max() - X.min())
-    y = (y - y.min()) / (y.max() - y.min())
-    
-    y_pred, fractal_outputs = run_snn(X, epochs, batch_size, l2_lambda, patience)
-    fig = plot_results(X, y, y_pred, fractal_outputs)
-    
-    return fig
+with gr.Blocks() as demo:
+    epochs = gr.Slider(1, 10000, value=5000, label="Epochs")
+    batch_size = gr.Slider(1, 100, value=32, label="Batch Size")
+    l2_lambda = gr.Slider(0.0001, 0.1, value=0.00001, label="L2 Lambda")
+    patience = gr.Slider(1, 1000, value=50, label="Patience")
+
+    plot = gr.Plot()
+
+    def update_plot(epochs, batch_size, l2_lambda, patience):
+        return run_snn(epochs, batch_size, l2_lambda, patience)
+
+    btn = gr.Button("Run SNN")
+    btn.click(update_plot, inputs=[epochs, batch_size, l2_lambda, patience], outputs=plot)
 
-gr.Interface(
-    fn=main_interface,
-    inputs=[
-        gr.inputs.Slider(1, 10000, default=5000, label="Epochs"),
-        gr.inputs.Slider(1, 100, default=32, label="Batch Size"),
-        gr.inputs.Slider(0.0001, 0.1, default=0.00001, label="L2 Lambda"),
-        gr.inputs.Slider(1, 1000, default=50, label="Patience")
-    ],
-    outputs=gr.outputs.Plot()
-).launch()
+demo.launch()