"""NEAT Genome implementation. This module implements the core NEAT genome structure and operations. Each genome represents a neural network with nodes (neurons) and connections (synapses). The genome can be mutated to evolve the network structure and weights over time. """ from dataclasses import dataclass import jax.numpy as jnp import jax.random as jrandom from typing import Dict, List, Tuple, Optional import time import random import numpy as np @dataclass class NodeGene: """Node gene containing activation function and type. Attributes: node_id: Unique identifier for this node node_type: Type of node ('input', 'hidden', 'recurrent', or 'output') activation: Activation function ('tanh', 'relu', 'sigmoid', or 'linear') """ node_id: int node_type: str # 'input', 'hidden', 'recurrent', or 'output' activation: str # 'tanh', 'relu', 'sigmoid', or 'linear' @dataclass class ConnectionGene: """Connection gene containing connection properties. Attributes: source: ID of source node target: ID of target node weight: Connection weight enabled: Whether connection is enabled innovation: Unique innovation number for this connection """ source: int target: int weight: float enabled: bool = True innovation: int = 0 class Genome: """NEAT Genome implementation. A genome represents a neural network as a collection of node and connection genes. The network topology can be modified through mutation operations. Attributes: input_size: Number of input nodes output_size: Number of output nodes node_genes: Dictionary mapping node IDs to NodeGene objects connection_genes: List of ConnectionGene objects key: Random key for reproducible randomness innovation_number: Counter for assigning unique innovation numbers """ def __init__(self, input_size: int, output_size: int): """Initialize genome with specified number of inputs and outputs. Args: input_size: Number of input nodes output_size: Number of output nodes (must be 3 for volleyball) """ self.input_size = input_size self.output_size = output_size self.node_genes: Dict[int, NodeGene] = {} self.connection_genes: List[ConnectionGene] = [] # Initialize random key timestamp = int(time.time() * 1000) self.key = jrandom.PRNGKey(hash((input_size, output_size, timestamp)) % (2**32)) # Counter for assigning unique innovation numbers self.innovation_number = 0 # Initialize minimal network structure self._init_minimal() def _init_minimal(self): """Initialize minimal feed-forward network structure. Network structure: - Input nodes [0-7]: Game state inputs - Hidden layer 1 [8-15]: First processing layer (8 nodes) - Hidden layer 2 [16-23]: Second processing layer (8 nodes) - Output nodes [24-26]: Action outputs (left, right, jump) Using larger initial weights for faster learning: - Input->Hidden1: N(0, 2.0) for strong initial responses - Hidden1->Hidden2: N(0, 2.0) for feature processing - Hidden2->Output: N(0, 4.0) for decisive actions """ # Create input nodes (0-7) for i in range(8): # Only 8 inputs used self.node_genes[i] = NodeGene( node_id=i, node_type='input', activation='linear' # Input nodes are always linear ) # Create first hidden layer (8-15) hidden1_size = 8 hidden1_start = 8 # Right after inputs for i in range(hidden1_size): node_id = hidden1_start + i self.node_genes[node_id] = NodeGene( node_id=node_id, node_type='hidden', activation='relu' # ReLU for faster learning ) # Connect all inputs to this hidden node for input_id in range(8): weight = float(jrandom.normal(self.key) * 2.0) self.connection_genes.append(ConnectionGene( source=input_id, target=node_id, weight=weight, enabled=True, innovation=self.innovation_number )) self.innovation_number += 1 # Create second hidden layer (16-23) hidden2_size = 8 hidden2_start = hidden1_start + hidden1_size for i in range(hidden2_size): node_id = hidden2_start + i self.node_genes[node_id] = NodeGene( node_id=node_id, node_type='hidden', activation='relu' # ReLU for faster learning ) # Connect all hidden1 nodes to this hidden2 node for h1_id in range(hidden1_start, hidden1_start + hidden1_size): weight = float(jrandom.normal(self.key) * 2.0) self.connection_genes.append(ConnectionGene( source=h1_id, target=node_id, weight=weight, enabled=True, innovation=self.innovation_number )) self.innovation_number += 1 # Create output nodes (24-26) output_start = hidden2_start + hidden2_size for i in range(self.output_size): node_id = output_start + i self.node_genes[node_id] = NodeGene( node_id=node_id, node_type='output', activation='tanh' # tanh for [-1,1] outputs ) # Connect all hidden2 nodes to this output for h2_id in range(hidden2_start, hidden2_start + hidden2_size): weight = float(jrandom.normal(self.key) * 4.0) # Larger weights for outputs self.connection_genes.append(ConnectionGene( source=h2_id, target=node_id, weight=weight, enabled=True, innovation=self.innovation_number )) self.innovation_number += 1 def mutate(self, config: Dict): """Mutate the genome by modifying weights and network structure. Args: config: Dictionary containing mutation parameters: - weight_mutation_rate: Probability of mutating each weight - weight_mutation_power: Standard deviation for weight mutations - add_node_rate: Probability of adding a new node - add_connection_rate: Probability of adding a new connection """ # Mutate connection weights for conn in self.connection_genes: if jrandom.uniform(self.key) < config['weight_mutation_rate']: # Get new random key self.key, subkey = jrandom.split(self.key) # Add random value from normal distribution conn.weight += float(jrandom.normal(subkey) * config['weight_mutation_power']) # Add new nodes (disabled for now since we're using fixed topology) if config['add_node_rate'] > 0: if jrandom.uniform(self.key) < config['add_node_rate']: self._add_node() # Add new connections (disabled for now) if config['add_connection_rate'] > 0: if jrandom.uniform(self.key) < config['add_connection_rate']: self._add_connection() def _add_node(self): """Add a new node by splitting an existing connection.""" if not self.connection_genes: return # Choose a random connection to split conn_to_split = np.random.choice(self.connection_genes) conn_to_split.enabled = False # Create new node new_node_id = max(self.node_genes.keys()) + 1 self.node_genes[new_node_id] = NodeGene( node_id=new_node_id, node_type='hidden', activation='relu' ) # Create two new connections self.connection_genes.extend([ ConnectionGene( source=conn_to_split.source, target=new_node_id, weight=1.0, enabled=True, innovation=self.innovation_number ), ConnectionGene( source=new_node_id, target=conn_to_split.target, weight=conn_to_split.weight, enabled=True, innovation=self.innovation_number + 1 ) ]) self.innovation_number += 2 def _add_connection(self): """Add a new connection between two unconnected nodes.""" # Get list of all possible connections existing_connections = {(c.source, c.target) for c in self.connection_genes} possible_connections = [] for source in self.node_genes: for target in self.node_genes: # Skip if connection already exists if (source, target) in existing_connections: continue # Skip if would create cycle (except recurrent) if self.node_genes[source].node_type != 'recurrent' and \ self.would_create_cycle(source, target): continue possible_connections.append((source, target)) if possible_connections: # Choose random connection source, target = random.choice(possible_connections) # Create new connection weight = float(jrandom.normal(self.key) * 1.0) self.connection_genes.append(ConnectionGene( source=source, target=target, weight=weight, enabled=True, innovation=self.innovation_number )) self.innovation_number += 1 def would_create_cycle(self, source: int, target: int) -> bool: """Check if adding connection would create cycle in network. Args: source: Source node ID target: Target node ID Returns: True if connection would create cycle, False otherwise """ # Skip cycle detection for recurrent connections if self.node_genes[source].node_type == 'recurrent' or \ self.node_genes[target].node_type == 'recurrent': return False # Do depth-first search from target to see if we can reach source visited = set() def dfs(node: int) -> bool: if node == source: return True if node in visited: return False visited.add(node) for conn in self.connection_genes: if conn.source == node and conn.enabled: if dfs(conn.target): return True return False return dfs(target) def add_node_between(self, source: int, target: int): """Add a new node between two nodes, splitting an existing connection. Args: source: Source node ID target: Target node ID """ # Find and disable the existing connection for conn in self.connection_genes: if conn.source == source and conn.target == target and conn.enabled: conn.enabled = False # Create new node new_id = max(self.node_genes.keys()) + 1 self.node_genes[new_id] = NodeGene( node_id=new_id, node_type='hidden', activation='relu' ) # Create two new connections self.connection_genes.extend([ ConnectionGene( source=source, target=new_id, weight=1.0, enabled=True, innovation=self.innovation_number ), ConnectionGene( source=new_id, target=target, weight=conn.weight, enabled=True, innovation=self.innovation_number + 1 ) ]) self.innovation_number += 2 break def add_connection(self, source: int, target: int, weight: Optional[float] = None) -> bool: """Add a new connection between two nodes. Args: source: Source node ID target: Target node ID weight: Optional connection weight. If None, a random weight is generated. Returns: True if connection was added, False if invalid or already exists """ # Check if connection already exists if any(c.source == source and c.target == target for c in self.connection_genes): return False # Validate nodes exist if source not in self.node_genes or target not in self.node_genes: return False # Ensure feed-forward (no cycles) if source >= target: # Simple way to ensure feed-forward return False # Generate random weight if not provided if weight is None: weight = float(jrandom.normal(self.key) * 1.0) # Add new connection self.connection_genes.append(ConnectionGene( source=source, target=target, weight=weight, enabled=True, innovation=self.innovation_number )) self.innovation_number += 1 return True def crossover(self, other: 'Genome', key: jnp.ndarray) -> 'Genome': """Perform crossover between two genomes. Args: other: Other parent genome key: JAX PRNG key Returns: Child genome """ # Create child genome child = Genome(self.input_size, self.output_size) # Inherit node genes for node_id in self.node_genes: if node_id in other.node_genes: # Inherit randomly from either parent if jrandom.uniform(key) < 0.5: child.node_genes[node_id] = self.node_genes[node_id] else: child.node_genes[node_id] = other.node_genes[node_id] else: # Inherit from fitter parent child.node_genes[node_id] = self.node_genes[node_id] # Inherit connection genes for conn in self.connection_genes: if conn.innovation in [c.innovation for c in other.connection_genes]: # Inherit randomly from either parent other_conn = next(c for c in other.connection_genes if c.innovation == conn.innovation) if jrandom.uniform(key) < 0.5: child.connection_genes.append(ConnectionGene( source=conn.source, target=conn.target, weight=conn.weight, enabled=conn.enabled, innovation=conn.innovation )) else: child.connection_genes.append(ConnectionGene( source=other_conn.source, target=other_conn.target, weight=other_conn.weight, enabled=other_conn.enabled, innovation=other_conn.innovation )) else: # Inherit from fitter parent child.connection_genes.append(ConnectionGene( source=conn.source, target=conn.target, weight=conn.weight, enabled=conn.enabled, innovation=conn.innovation )) return child def clone(self) -> 'Genome': """Create a copy of this genome. Returns: Copy of genome """ clone = Genome(self.input_size, self.output_size) clone.node_genes = self.node_genes.copy() clone.connection_genes = [ConnectionGene(**conn.__dict__) for conn in self.connection_genes] return clone @property def n_nodes(self) -> int: """Get total number of nodes in the genome.""" return len(self.node_genes)