Spiking Neural Network (SNN)

graph TD
    A[Presynaptic Neuron] -->|Spike| B[Synapse]
    B -->|Weighted Signal| C[Postsynaptic Neuron]
    C -->|Membrane Potential| D{Threshold?}
    D -->|Yes| E[Fire Spike]
    D -->|No| C
    E --> F[Refractory Period]
    F --> C

# Leaky Integrate-and-Fire neuron implementation
class LIFNeuron:
    def __init__(self, threshold=1.0, decay=0.9, refractory_period=5):
        self.threshold = threshold
        self.decay = decay
        self.refractory_period = refractory_period
        self.membrane_potential = 0.0
        self.refractory_counter = 0
        self.spike = False

    def update(self, input_current):
        """Update neuron state"""
        # Check refractory period
        if self.refractory_counter > 0:
            self.refractory_counter -= 1
            self.spike = False
            return False

        # Update membrane potential
        self.membrane_potential = self.membrane_potential * self.decay + input_current

        # Check for spike
        if self.membrane_potential >= self.threshold:
            self.spike = True
            self.membrane_potential = 0.0
            self.refractory_counter = self.refractory_period
            return True
        else:
            self.spike = False
            return False

    def reset(self):
        """Reset neuron state"""
        self.membrane_potential = 0.0
        self.refractory_counter = 0
        self.spike = False

# Izhikevich neuron model implementation
class IzhikevichNeuron:
    def __init__(self, a=0.02, b=0.2, c=-65, d=8):
        self.a = a  # Time scale of recovery variable
        self.b = b  # Sensitivity of recovery variable
        self.c = c  # Reset value of membrane potential
        self.d = d  # Reset value of recovery variable
        self.v = -65.0  # Membrane potential
        self.u = b * self.v  # Recovery variable
        self.spike = False

    def update(self, input_current):
        """Update neuron state"""
        # Update membrane potential and recovery variable
        self.v += 0.5 * (0.04 * self.v**2 + 5 * self.v + 140 - self.u + input_current)
        self.v += 0.5 * (0.04 * self.v**2 + 5 * self.v + 140 - self.u + input_current)
        self.u += self.a * (self.b * self.v - self.u)

        # Check for spike
        if self.v >= 30.0:
            self.spike = True
            self.v = self.c
            self.u += self.d
            return True
        else:
            self.spike = False
            return False

    def reset(self):
        """Reset neuron state"""
        self.v = -65.0
        self.u = self.b * self.v
        self.spike = False

# STDP learning rule implementation
class STDP:
    def __init__(self, learning_rate=0.01, a_plus=0.1, a_minus=0.1, tau_plus=20, tau_minus=20):
        self.learning_rate = learning_rate
        self.a_plus = a_plus  # LTP amplitude
        self.a_minus = a_minus  # LTD amplitude
        self.tau_plus = tau_plus  # LTP time constant
        self.tau_minus = tau_minus  # LTD time constant
        self.pre_trace = 0.0
        self.post_trace = 0.0

    def update(self, weight, pre_spike, post_spike, delta_t):
        """Update weight based on spike timing"""
        # Update traces
        self.pre_trace = self.pre_trace * np.exp(-1/self.tau_plus) + pre_spike
        self.post_trace = self.post_trace * np.exp(-1/self.tau_minus) + post_spike

        # Calculate weight update
        if pre_spike and post_spike:
            if delta_t > 0:  # Pre before post
                dw = self.a_plus * np.exp(-delta_t/self.tau_plus)
            else:  # Post before pre
                dw = -self.a_minus * np.exp(delta_t/self.tau_minus)
            weight += self.learning_rate * dw

        return np.clip(weight, 0, 1)  # Keep weight between 0 and 1

# Feedforward SNN implementation
class FeedforwardSNN:
    def __init__(self, input_size, hidden_size, output_size):
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.output_size = output_size

        # Initialize neurons
        self.hidden_neurons = [LIFNeuron() for _ in range(hidden_size)]
        self.output_neurons = [LIFNeuron() for _ in range(output_size)]

        # Initialize weights
        self.input_weights = np.random.rand(input_size, hidden_size) * 0.2
        self.hidden_weights = np.random.rand(hidden_size, output_size) * 0.2

        # Initialize STDP
        self.stdp = STDP()

    def forward(self, input_spikes, timesteps=20):
        """Forward pass through the network"""
        # Initialize spike records
        hidden_spikes = np.zeros((timesteps, self.hidden_size))
        output_spikes = np.zeros((timesteps, self.output_size))

        for t in range(timesteps):
            # Input to hidden layer
            hidden_input = np.dot(input_spikes[t], self.input_weights)
            for i, neuron in enumerate(self.hidden_neurons):
                hidden_spikes[t, i] = neuron.update(hidden_input[i])

            # Hidden to output layer
            output_input = np.dot(hidden_spikes[t], self.hidden_weights)
            for i, neuron in enumerate(self.output_neurons):
                output_spikes[t, i] = neuron.update(output_input[i])

        return output_spikes

    def train(self, input_spikes, target_spikes, epochs=10):
        """Train the network using STDP"""
        for epoch in range(epochs):
            # Forward pass
            hidden_spikes = np.zeros((input_spikes.shape[0], self.hidden_size))
            output_spikes = np.zeros((input_spikes.shape[0], self.output_size))

            for t in range(input_spikes.shape[0]):
                # Input to hidden layer
                hidden_input = np.dot(input_spikes[t], self.input_weights)
                for i, neuron in enumerate(self.hidden_neurons):
                    hidden_spikes[t, i] = neuron.update(hidden_input[i])

                # Hidden to output layer
                output_input = np.dot(hidden_spikes[t], self.hidden_weights)
                for i, neuron in enumerate(self.output_neurons):
                    output_spikes[t, i] = neuron.update(output_input[i])

                    # Apply STDP
                    for j in range(self.hidden_size):
                        if hidden_spikes[t, j]:
                            delta_t = t - np.argmax(hidden_spikes[:, j])
                            self.hidden_weights[j, i] = self.stdp.update(
                                self.hidden_weights[j, i],
                                hidden_spikes[t, j],
                                output_spikes[t, i],
                                delta_t
                            )

            # Reset neurons
            for neuron in self.hidden_neurons + self.output_neurons:
                neuron.reset()

# Recurrent SNN implementation
class RecurrentSNN:
    def __init__(self, input_size, hidden_size, output_size):
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.output_size = output_size

        # Initialize neurons
        self.hidden_neurons = [LIFNeuron() for _ in range(hidden_size)]
        self.output_neurons = [LIFNeuron() for _ in range(output_size)]

        # Initialize weights
        self.input_weights = np.random.rand(input_size, hidden_size) * 0.2
        self.hidden_weights = np.random.rand(hidden_size, hidden_size) * 0.1
        self.output_weights = np.random.rand(hidden_size, output_size) * 0.2

    def forward(self, input_spikes, timesteps=20):
        """Forward pass through the network"""
        # Initialize spike records
        hidden_spikes = np.zeros((timesteps, self.hidden_size))
        output_spikes = np.zeros((timesteps, self.output_size))

        for t in range(timesteps):
            # Input to hidden layer
            hidden_input = np.dot(input_spikes[t], self.input_weights)

            # Recurrent connections
            if t > 0:
                hidden_input += np.dot(hidden_spikes[t-1], self.hidden_weights)

            # Update hidden neurons
            for i, neuron in enumerate(self.hidden_neurons):
                hidden_spikes[t, i] = neuron.update(hidden_input[i])

            # Hidden to output layer
            output_input = np.dot(hidden_spikes[t], self.output_weights)
            for i, neuron in enumerate(self.output_neurons):
                output_spikes[t, i] = neuron.update(output_input[i])

        return output_spikes

    def reset(self):
        """Reset all neurons"""
        for neuron in self.hidden_neurons + self.output_neurons:
            neuron.reset()

# Enhanced STDP implementation
class EnhancedSTDP:
    def __init__(self, learning_rate=0.01, a_plus=0.1, a_minus=0.1,
                 tau_plus=20, tau_minus=20, w_min=0, w_max=1):
        self.learning_rate = learning_rate
        self.a_plus = a_plus
        self.a_minus = a_minus
        self.tau_plus = tau_plus
        self.tau_minus = tau_minus
        self.w_min = w_min
        self.w_max = w_max
        self.pre_traces = {}
        self.post_traces = {}

    def update(self, weight, pre_neuron, post_neuron, t):
        """Update weight based on spike timing"""
        # Initialize traces if not present
        if pre_neuron not in self.pre_traces:
            self.pre_traces[pre_neuron] = 0.0
        if post_neuron not in self.post_traces:
            self.post_traces[post_neuron] = 0.0

        # Update traces
        self.pre_traces[pre_neuron] = self.pre_traces[pre_neuron] * np.exp(-1/self.tau_plus)
        self.post_traces[post_neuron] = self.post_traces[post_neuron] * np.exp(-1/self.tau_minus)

        # Calculate weight update
        if pre_neuron.spike and post_neuron.spike:
            delta_t = t - pre_neuron.last_spike_time
            if delta_t > 0:  # Pre before post
                dw = self.a_plus * np.exp(-delta_t/self.tau_plus)
            else:  # Post before pre
                dw = -self.a_minus * np.exp(delta_t/self.tau_minus)
            weight += self.learning_rate * dw

        # Update traces with current spikes
        if pre_neuron.spike:
            self.pre_traces[pre_neuron] += 1.0
            pre_neuron.last_spike_time = t
        if post_neuron.spike:
            self.post_traces[post_neuron] += 1.0

        return np.clip(weight, self.w_min, self.w_max)

# ANN to SNN conversion
def ann_to_snn(ann_model, timesteps=20, v_threshold=1.0):
    """Convert a trained ANN to SNN"""
    # Create SNN with same architecture
    snn = FeedforwardSNN(
        input_size=ann_model.layers[0].input_shape[1],
        hidden_size=ann_model.layers[1].units,
        output_size=ann_model.layers[-1].units
    )

    # Transfer weights (normalized)
    snn.input_weights = ann_model.layers[0].get_weights()[0] / np.max(ann_model.layers[0].get_weights()[0])
    snn.hidden_weights = ann_model.layers[1].get_weights()[0] / np.max(ann_model.layers[1].get_weights()[0])

    # Set threshold
    for neuron in snn.hidden_neurons + snn.output_neurons:
        neuron.threshold = v_threshold

    return snn

# Example usage
# ann = create_ann_model()
# snn = ann_to_snn(ann, timesteps=20)

# BPTT for SNNs implementation
def bptt_snn(snn, input_spikes, target_spikes, learning_rate=0.01):
    """Backpropagation Through Time for SNNs"""
    timesteps = input_spikes.shape[0]

    # Forward pass
    hidden_spikes = np.zeros((timesteps, snn.hidden_size))
    output_spikes = np.zeros((timesteps, snn.output_size))

    for t in range(timesteps):
        # Input to hidden layer
        hidden_input = np.dot(input_spikes[t], snn.input_weights)
        for i, neuron in enumerate(snn.hidden_neurons):
            hidden_spikes[t, i] = neuron.update(hidden_input[i])

        # Hidden to output layer
        output_input = np.dot(hidden_spikes[t], snn.hidden_weights)
        for i, neuron in enumerate(snn.output_neurons):
            output_spikes[t, i] = neuron.update(output_input[i])

    # Backward pass
    d_hidden_weights = np.zeros_like(snn.hidden_weights)
    d_input_weights = np.zeros_like(snn.input_weights)

    # Output error
    output_error = output_spikes - target_spikes

    for t in range(timesteps-1, -1, -1):
        # Output layer gradient
        d_output = output_error[t]

        # Hidden to output weights
        d_hidden_weights += np.outer(hidden_spikes[t], d_output)

        # Hidden layer error
        d_hidden = np.dot(d_output, snn.hidden_weights.T)

        # Input to hidden weights
        d_input_weights += np.outer(input_spikes[t], d_hidden)

    # Update weights
    snn.hidden_weights -= learning_rate * d_hidden_weights / timesteps
    snn.input_weights -= learning_rate * d_input_weights / timesteps

    # Reset neurons
    snn.reset()

    return np.mean(np.abs(output_error))

# Neuromorphic computing with SNNs
class NeuromorphicSystem:
    def __init__(self, input_size, hidden_size, output_size):
        self.snn = RecurrentSNN(input_size, hidden_size, output_size)
        self.stdp = EnhancedSTDP()

    def process_event(self, event, t):
        """Process a single event"""
        # Convert event to spike
        input_spike = np.zeros(self.snn.input_size)
        input_spike[event['sensor']] = 1.0

        # Forward pass
        hidden_input = np.dot(input_spike, self.snn.input_weights)
        hidden_spikes = np.zeros(self.snn.hidden_size)

        for i, neuron in enumerate(self.snn.hidden_neurons):
            hidden_spikes[i] = neuron.update(hidden_input[i])

        # Apply STDP for learning
        for i, h_neuron in enumerate(self.snn.hidden_neurons):
            if h_neuron.spike:
                for j, o_neuron in enumerate(self.snn.output_neurons):
                    if o_neuron.spike:
                        self.snn.hidden_weights[i, j] = self.stdp.update(
                            self.snn.hidden_weights[i, j],
                            h_neuron,
                            o_neuron,
                            t
                        )

        return hidden_spikes

    def reset(self):
        """Reset the system"""
        self.snn.reset()

# Event-based vision with SNNs
class EventCameraSNN:
    def __init__(self, width, height, n_classes):
        self.width = width
        self.height = height
        self.n_classes = n_classes

        # Create SNN for event processing
        self.snn = FeedforwardSNN(
            input_size=width * height,
            hidden_size=256,
            output_size=n_classes
        )

        # Create event buffer
        self.event_buffer = []

    def process_event(self, event):
        """Process a single event from event camera"""
        # Add event to buffer
        self.event_buffer.append(event)

        # Convert events to spike train
        spike_train = np.zeros((20, self.width * self.height))
        for i, e in enumerate(self.event_buffer[-20:]):
            idx = e['y'] * self.width + e['x']
            spike_train[i, idx] = 1.0 if e['polarity'] else -1.0

        # Process with SNN
        output = self.snn.forward(spike_train)

        return np.sum(output, axis=0)  # Return accumulated spikes

    def classify(self, events):
        """Classify a sequence of events"""
        for event in events:
            self.process_event(event)

        # Get final classification
        output = self.process_event({'x': 0, 'y': 0, 'polarity': 0})  # Dummy event
        return np.argmax(output)

# Edge AI with SNNs
class EdgeAISNN:
    def __init__(self, input_size, hidden_size, output_size):
        self.snn = FeedforwardSNN(input_size, hidden_size, output_size)
        self.threshold = 0.8  # Decision threshold

    def process_sensor_data(self, sensor_data, timesteps=10):
        """Process sensor data on edge device"""
        # Convert sensor data to spike train
        spike_train = np.zeros((timesteps, self.snn.input_size))
        for t in range(timesteps):
            for i in range(self.snn.input_size):
                # Simple encoding: spike if value exceeds threshold
                if sensor_data[t, i] > 0.5:
                    spike_train[t, i] = 1.0

        # Process with SNN
        output = self.snn.forward(spike_train)

        # Make decision
        decision = np.sum(output, axis=0) / timesteps
        return decision > self.threshold

    def low_power_mode(self):
        """Switch to low power mode"""
        # Reduce timesteps for lower power consumption
        self.timesteps = 5
        # Increase neuron thresholds
        for neuron in self.snn.hidden_neurons + self.snn.output_neurons:
            neuron.threshold *= 1.5