Spaces:
Paused
Paused
| """Conservative topology self-organization for Brain-5D. | |
| The engine is optional and disabled by default. It uses the public manipulator | |
| instead of mutating core dictionaries directly. STDP/reward learning stays in | |
| ``src.learning``; this engine only changes topology and performs slow structural | |
| adaptation. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Any | |
| from src.core.spatial_index import ( | |
| Coord5D, | |
| iter_neighbour_coords, | |
| pack_coords, | |
| unpack_coords, | |
| ) | |
| from src.manipulation.manipulator import Brain5DManipulator | |
| class SelfOrganizationParameters: | |
| enabled: bool = False | |
| interval_ticks: int = 100 | |
| pruning_enabled: bool = False | |
| pruning_weight_threshold: float = 0.005 | |
| pruning_min_age_ticks: int = 1000 | |
| sprouting_enabled: bool = False | |
| sprouting_max_out_degree: int = 12 | |
| sprouting_radius: float = 2.0 | |
| sprouting_weight: float = 0.05 | |
| sprouting_delay: int = 1 | |
| neurogenesis_enabled: bool = False | |
| neurogenesis_spike_delta_threshold: int = 50 | |
| neurogenesis_radius: float = 1.0 | |
| neurogenesis_max_per_cycle: int = 1 | |
| max_neurons: int = 0 # 0 = unlimited | |
| def from_config(cls, config: dict[str, Any]) -> SelfOrganizationParameters: | |
| c = config.get("self_organization", {}) | |
| params = cls( | |
| **{k: c[k] for k in cls.__dataclass_fields__ if k in c} | |
| ) # pylint: disable=no-member | |
| # Also support top-level max_neurons for backward compatibility | |
| if params.max_neurons == 0 and "max_neurons" in config: | |
| object.__setattr__(params, "max_neurons", int(config["max_neurons"])) | |
| return params | |
| def validate(self) -> None: | |
| if self.interval_ticks < 1: | |
| raise ValueError("self_organization.interval_ticks must be >= 1") | |
| if self.sprouting_max_out_degree < 0: | |
| raise ValueError("sprouting_max_out_degree must be >= 0") | |
| if self.sprouting_delay < 1: | |
| raise ValueError("sprouting_delay must be >= 1") | |
| if self.neurogenesis_max_per_cycle < 0: | |
| raise ValueError("neurogenesis_max_per_cycle must be >= 0") | |
| class SelfOrganizationStats: | |
| cycles: int | |
| pruned_synapses: int | |
| created_synapses: int | |
| created_neurons: int | |
| class SelfOrganizationEngine: | |
| """Slow structural adaptation layer attached through a post-step hook.""" | |
| def __init__( | |
| self, network: Any, manipulator: Brain5DManipulator, config: dict[str, Any] | |
| ): | |
| self.network = network | |
| self.manipulator = manipulator | |
| self.params = SelfOrganizationParameters.from_config(config) | |
| self.params.validate() | |
| self._attached = False | |
| self._cycles = 0 | |
| self._pruned = 0 | |
| self._created_synapses = 0 | |
| self._created_neurons = 0 | |
| self._last_spike_counter: dict[int, int] = {} | |
| def stats(self) -> SelfOrganizationStats: | |
| return SelfOrganizationStats( | |
| self._cycles, self._pruned, self._created_synapses, self._created_neurons | |
| ) | |
| def attach(self) -> None: | |
| if not self._attached: | |
| self.network.add_post_step_hook(self.update) | |
| self._attached = True | |
| def detach(self) -> None: | |
| if self._attached: | |
| self.network.remove_post_step_hook(self.update) | |
| self._attached = False | |
| def update(self, step_result: Any) -> None: | |
| if not self.params.enabled: | |
| return | |
| tick = int(step_result.tick) | |
| if (tick + 1) % self.params.interval_ticks != 0: | |
| return | |
| self.run_cycle(tick) | |
| def run_cycle(self, tick: int | None = None) -> SelfOrganizationStats: | |
| tick = int(self.network.current_tick if tick is None else tick) | |
| if self.params.pruning_enabled: | |
| self._run_pruning(tick) | |
| if self.params.sprouting_enabled: | |
| self._run_sprouting(tick) | |
| if self.params.neurogenesis_enabled: | |
| self._run_neurogenesis(tick) | |
| self._cycles += 1 | |
| return self.stats | |
| def _run_pruning(self, tick: int) -> None: | |
| to_remove: list[tuple[int, int]] = [] | |
| for source_id, synapses in tuple(self.network.synapses.items()): | |
| for syn in tuple(synapses): | |
| meta = self.manipulator.synapse_metadata.get((source_id, syn.target_id)) | |
| created_tick = 0 if meta is None else int(meta.created_tick) | |
| age = tick - created_tick | |
| if ( | |
| age >= self.params.pruning_min_age_ticks | |
| and abs(float(syn.weight)) < self.params.pruning_weight_threshold | |
| ): | |
| to_remove.append((source_id, syn.target_id)) | |
| for source_id, target_id in to_remove: | |
| self.manipulator.delete_synapse(source_id, target_id) | |
| self._pruned += 1 | |
| def _run_sprouting(self, _tick: int) -> None: | |
| for source_id in tuple(self.network.neurons): | |
| outgoing = self.network.synapses.get(source_id, ()) | |
| if len(outgoing) >= self.params.sprouting_max_out_degree: | |
| continue | |
| connected = {s.target_id for s in outgoing} | |
| coord = unpack_coords(source_id) | |
| target_id = None | |
| for ncoord in iter_neighbour_coords( | |
| coord, self.network.dimensions, self.params.sprouting_radius | |
| ): | |
| candidate = pack_coords(*ncoord) | |
| if ( | |
| candidate == source_id | |
| or candidate not in self.network.neurons | |
| or candidate in connected | |
| ): | |
| continue | |
| target_id = candidate | |
| break | |
| if target_id is None: | |
| continue | |
| self.manipulator.create_synapse( | |
| source_id, | |
| target_id, | |
| self.params.sprouting_weight, | |
| min(self.params.sprouting_delay, int(self.network.max_delay)), | |
| ) | |
| self._created_synapses += 1 | |
| def _run_neurogenesis(self, _tick: int) -> None: | |
| if ( | |
| self.params.max_neurons | |
| and len(self.network.neurons) >= self.params.max_neurons | |
| ): | |
| return | |
| created = 0 | |
| ranked = sorted( | |
| self.network.neurons.items(), | |
| key=lambda item: int(item[1].spike_counter) | |
| - self._last_spike_counter.get(item[0], 0), | |
| reverse=True, | |
| ) | |
| for parent_id, neuron in ranked: | |
| delta = int(neuron.spike_counter) - self._last_spike_counter.get( | |
| parent_id, 0 | |
| ) | |
| self._last_spike_counter[parent_id] = int(neuron.spike_counter) | |
| if delta < self.params.neurogenesis_spike_delta_threshold: | |
| continue | |
| free_coord = self._find_free_coord(parent_id) | |
| if free_coord is None: | |
| continue | |
| child_id = self.manipulator.create_neuron(free_coord) | |
| self.manipulator.set_neuron( | |
| child_id, | |
| a=neuron.a, | |
| b=neuron.b, | |
| c=neuron.c, | |
| d=neuron.d, | |
| v=neuron.c, | |
| u=neuron.b * neuron.c, | |
| energy=min(1.0, max(0.0, float(neuron.energy))), | |
| ) | |
| self.manipulator.create_synapse( | |
| parent_id, child_id, self.params.sprouting_weight, 1 | |
| ) | |
| self._created_neurons += 1 | |
| self._created_synapses += 1 | |
| created += 1 | |
| if created >= self.params.neurogenesis_max_per_cycle: | |
| break | |
| def _find_free_coord(self, neuron_id: int) -> Coord5D | None: | |
| coord = unpack_coords(neuron_id) | |
| for candidate in iter_neighbour_coords( | |
| coord, self.network.dimensions, self.params.neurogenesis_radius | |
| ): | |
| nid = pack_coords(*candidate) | |
| if nid not in self.network.neurons: | |
| return candidate | |
| return None | |