Spaces:
Sleeping
Sleeping
File size: 8,092 Bytes
3d46076 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | import copy
import hashlib
import numpy as np
from typing import Dict, Any, Tuple, List, Optional
from src.connectome.types import ConnectomeGraph
class StructuralMutator:
"""
Applies measurable structural mutations to the biological connectome:
- Weight perturbation (strengthening / weakening)
- Synaptic pruning (removing low-efficacy connections)
- Synaptic rewiring (forming new proximate connections)
- Population growth (adding new interneurons with biological coordinates)
All mutations produce an audit record and are rollback-capable.
"""
def __init__(self, seed: int = 42):
self.seed = seed
self.rng = np.random.RandomState(seed)
def mutate_weights(self, graph: ConnectomeGraph, rate: float = 0.05, scale: float = 0.1) -> Dict[str, Any]:
mask = self.rng.rand(len(graph.weights)) < rate
count = int(np.sum(mask))
if count > 0:
deltas = self.rng.normal(0.0, scale, count).astype(np.float32)
graph.weights[mask] = np.clip(graph.weights[mask] + deltas, 0.01, 1.0)
graph.graph_hash = graph.compute_graph_hash()
return {
"type": "weight_perturbation",
"synapses_modified": count,
"rate": rate,
"scale": scale
}
def prune_synapses(self, graph: ConnectomeGraph, threshold: float = 0.03) -> Dict[str, Any]:
new_row_offsets = [0]
new_col_indices = []
new_weights = []
pruned = 0
for i in range(graph.num_neurons):
start = graph.row_offsets[i]
end = graph.row_offsets[i + 1]
for k in range(start, end):
if graph.weights[k] >= threshold:
new_col_indices.append(graph.col_indices[k])
new_weights.append(graph.weights[k])
else:
pruned += 1
new_row_offsets.append(len(new_col_indices))
graph.row_offsets = np.array(new_row_offsets, dtype=np.int32)
graph.col_indices = np.array(new_col_indices, dtype=np.int32)
graph.weights = np.array(new_weights, dtype=np.float32)
graph.validate_invariants()
graph.graph_hash = graph.compute_graph_hash()
return {
"type": "synaptic_pruning",
"pruned_synapses": pruned,
"threshold": threshold,
"remaining_synapses": len(graph.weights)
}
def rewire_synapses(self, graph: ConnectomeGraph, num_new: int = 15, max_distance: float = 6000.0) -> Dict[str, Any]:
# CSR CONVENTION v3: rows store INCOMING edges. Creating src->tgt means
# appending src to row tgt's source list.
N = graph.num_neurons
added = 0
adj = {i: list(zip(graph.col_indices[graph.row_offsets[i]:graph.row_offsets[i+1]],
graph.weights[graph.row_offsets[i]:graph.row_offsets[i+1]]))
for i in range(N)}
attempts = 0
while added < num_new and attempts < num_new * 10:
attempts += 1
src = self.rng.randint(0, N)
tgt = self.rng.randint(0, N)
if src == tgt:
continue
existing = [s for s, _ in adj[tgt]]
if src in existing:
continue
dist = np.linalg.norm(graph.coordinates[src] - graph.coordinates[tgt])
if dist < max_distance:
w = float(self.rng.uniform(0.05, 0.25))
adj[tgt].append((src, w))
added += 1
new_row_offsets = [0]
new_col_indices = []
new_weights = []
for i in range(N):
adj[i].sort(key=lambda x: x[0])
for t, w in adj[i]:
new_col_indices.append(t)
new_weights.append(w)
new_row_offsets.append(len(new_col_indices))
graph.row_offsets = np.array(new_row_offsets, dtype=np.int32)
graph.col_indices = np.array(new_col_indices, dtype=np.int32)
graph.weights = np.array(new_weights, dtype=np.float32)
graph.validate_invariants()
graph.graph_hash = graph.compute_graph_hash()
return {
"type": "synaptic_rewiring",
"new_synapses_added": added,
"total_synapses": len(graph.weights)
}
def grow_population(self, graph: ConnectomeGraph, new_neurons_count: int = 4) -> Dict[str, Any]:
"""
Adds new interneurons placed within the male CNS volume and connects them to local neighbors.
"""
mean_coord = graph.coordinates.mean(axis=0)
std_coord = graph.coordinates.std(axis=0) * 0.5
new_coords = self.rng.normal(mean_coord, std_coord, (new_neurons_count, 3)).astype(np.float32)
last_id = int(np.max(graph.neuron_ids))
new_ids = np.array([last_id + 1 + i for i in range(new_neurons_count)], dtype=np.int64)
new_tbars = np.full(new_neurons_count, 50, dtype=np.int32)
new_sides = ["M" for _ in range(new_neurons_count)]
old_N = graph.num_neurons
updated_ids = np.concatenate([graph.neuron_ids, new_ids])
updated_coords = np.vstack([graph.coordinates, new_coords])
updated_tbars = np.concatenate([graph.tbars, new_tbars])
updated_sides = graph.sides + new_sides
adj = {i: list(zip(graph.col_indices[graph.row_offsets[i]:graph.row_offsets[i+1]],
graph.weights[graph.row_offsets[i]:graph.row_offsets[i+1]]))
for i in range(old_N)}
# Connect new neurons to closest existing neurons
for n_i in range(new_neurons_count):
curr_idx = old_N + n_i
adj[curr_idx] = []
dists = np.linalg.norm(graph.coordinates - new_coords[n_i], axis=1)
nearest = np.argsort(dists)[:4]
for near in nearest:
w = float(self.rng.uniform(0.1, 0.3))
adj[curr_idx].append((int(near), w))
adj[int(near)].append((curr_idx, w))
new_row_offsets = [0]
new_col_indices = []
new_weights = []
total_N = old_N + new_neurons_count
for i in range(total_N):
adj[i].sort(key=lambda x: x[0])
for t, w in adj[i]:
new_col_indices.append(t)
new_weights.append(w)
new_row_offsets.append(len(new_col_indices))
graph.neuron_ids = updated_ids
graph.coordinates = updated_coords
graph.tbars = updated_tbars
graph.sides = updated_sides
graph.row_offsets = np.array(new_row_offsets, dtype=np.int32)
graph.col_indices = np.array(new_col_indices, dtype=np.int32)
graph.weights = np.array(new_weights, dtype=np.float32)
graph.validate_invariants()
graph.graph_hash = graph.compute_graph_hash()
return {
"type": "population_growth",
"neurons_added": new_neurons_count,
"total_neurons": total_N,
"total_synapses": len(graph.weights)
}
def clone_graph(self, graph: ConnectomeGraph) -> ConnectomeGraph:
# P9: preserve graph subclass (mode/provenance), population registry,
# and metadata so candidates remain the same kind of graph as the parent.
cls = type(graph)
clone = cls(
neuron_ids=graph.neuron_ids.copy(),
coordinates=graph.coordinates.copy(),
tbars=graph.tbars.copy(),
sides=list(graph.sides),
row_offsets=graph.row_offsets.copy(),
col_indices=graph.col_indices.copy(),
weights=graph.weights.copy(),
populations=graph.populations,
provenance_metadata=dict(graph.provenance_metadata),
)
# Recompute: never propagate a possibly stale hash.
clone.graph_hash = clone.compute_graph_hash()
return clone
|