Spaces:
Paused
Paused
| """Deterministic network impulse probes and observable response signatures.""" | |
| from __future__ import annotations | |
| from collections.abc import Callable, Mapping | |
| from dataclasses import dataclass | |
| from typing import Any, Protocol | |
| class ImpulseRuntime(Protocol): | |
| """Minimal runtime boundary required by :class:`NetworkImpulseProbe`.""" | |
| def inject_current_batch(self, currents: Mapping[int, float]) -> None: ... | |
| def step(self) -> Mapping[str, Any]: ... | |
| class NetworkResponseSignature: | |
| """Measured response to one controlled input impulse.""" | |
| first_response_latency: int | None | |
| last_response_latency: int | None | |
| activated_neurons: int | |
| total_spikes: int | |
| peak_spike_rate: float | |
| propagation_depth: int | |
| recurrent_events: int | |
| return_latency: int | None | |
| spike_sequence: tuple[tuple[int, int], ...] = () | |
| network_state_digest_before: str | None = None | |
| network_state_digest_after: str | None = None | |
| ticks_executed: int = 0 | |
| delivered_synaptic_events: int = 0 | |
| synaptic_activity_ticks: int = 0 | |
| max_synaptic_current_targets: int = 0 | |
| total_synapses: int = 0 | |
| stopped_on_quiescence: bool = False | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "first_response_latency": self.first_response_latency, | |
| "last_response_latency": self.last_response_latency, | |
| "activated_neurons": self.activated_neurons, | |
| "total_spikes": self.total_spikes, | |
| "peak_spike_rate": self.peak_spike_rate, | |
| "propagation_depth": self.propagation_depth, | |
| "recurrent_events": self.recurrent_events, | |
| "return_latency": self.return_latency, | |
| "spike_sequence": [ | |
| {"tick": tick, "neuron_id": neuron_id} | |
| for tick, neuron_id in self.spike_sequence | |
| ], | |
| "network_state_digest_before": self.network_state_digest_before, | |
| "network_state_digest_after": self.network_state_digest_after, | |
| "ticks_executed": self.ticks_executed, | |
| "delivered_synaptic_events": self.delivered_synaptic_events, | |
| "synaptic_activity_ticks": self.synaptic_activity_ticks, | |
| "max_synaptic_current_targets": self.max_synaptic_current_targets, | |
| "total_synapses": self.total_synapses, | |
| "stopped_on_quiescence": self.stopped_on_quiescence, | |
| } | |
| class NetworkImpulseProbe: | |
| """Inject one bounded impulse and summarize only observed runtime output. | |
| ``max_ticks`` is the requested observation window. By default the complete | |
| window is executed even when the runtime reports temporary quiescence. A | |
| caller that explicitly wants an early-stop probe can pass a smaller | |
| ``min_ticks`` value. | |
| """ | |
| def __init__( | |
| self, | |
| *, | |
| source_neuron: int, | |
| current: float = 1.0, | |
| max_ticks: int = 100, | |
| min_ticks: int | None = None, | |
| state_digest: Callable[[], str] | None = None, | |
| ) -> None: | |
| if max_ticks < 1: | |
| raise ValueError("max_ticks must be positive") | |
| effective_min_ticks = max_ticks if min_ticks is None else min_ticks | |
| if effective_min_ticks < 0: | |
| raise ValueError("min_ticks must not be negative") | |
| if effective_min_ticks > max_ticks: | |
| raise ValueError("min_ticks must not exceed max_ticks") | |
| self.source_neuron = source_neuron | |
| self.current = current | |
| self.max_ticks = max_ticks | |
| self.min_ticks = effective_min_ticks | |
| self.state_digest = state_digest | |
| def run(self, runtime: ImpulseRuntime) -> NetworkResponseSignature: | |
| """Run until quiescence after ``min_ticks`` or until ``max_ticks``.""" | |
| before = self.state_digest() if self.state_digest else None | |
| runtime.inject_current_batch({self.source_neuron: self.current}) | |
| response_ticks: list[int] = [] | |
| response_neurons: set[int] = set() | |
| total_spikes = 0 | |
| peak_rate = 0.0 | |
| recurrent_events = 0 | |
| return_latency: int | None = None | |
| spike_sequence: list[tuple[int, int]] = [] | |
| ticks_executed = 0 | |
| delivered_synaptic_events = 0 | |
| synaptic_activity_ticks = 0 | |
| max_synaptic_current_targets = 0 | |
| total_synapses = 0 | |
| stopped_on_quiescence = False | |
| for tick in range(self.max_ticks): | |
| result = runtime.step() | |
| ticks_executed = tick + 1 | |
| raw_ids = result.get("spike_ids") | |
| if raw_ids is None: | |
| raw_ids = result.get("output_spike_ids") | |
| if raw_ids is None: | |
| raw_ids = () | |
| spike_ids = tuple(int(value) for value in raw_ids) | |
| output_ids = tuple( | |
| int(value) for value in result.get("output_spike_ids", spike_ids) | |
| ) | |
| delivered = int(result.get("delivered_events", 0) or 0) | |
| synaptic_targets = int(result.get("synaptic_current_targets", 0) or 0) | |
| delivered_synaptic_events += delivered | |
| if delivered > 0 or synaptic_targets > 0: | |
| synaptic_activity_ticks += 1 | |
| max_synaptic_current_targets = max( | |
| max_synaptic_current_targets, synaptic_targets | |
| ) | |
| total_synapses = max( | |
| total_synapses, int(result.get("total_synapses", 0) or 0) | |
| ) | |
| if spike_ids: | |
| spike_sequence.extend((tick, neuron_id) for neuron_id in spike_ids) | |
| response_neurons.update(spike_ids) | |
| total_spikes += len(spike_ids) | |
| peak_rate = max(peak_rate, float(len(spike_ids))) | |
| if ( | |
| self.source_neuron in spike_ids | |
| and return_latency is None | |
| and tick > 0 | |
| ): | |
| return_latency = tick | |
| if tick > 0 and self.source_neuron in spike_ids: | |
| recurrent_events += 1 | |
| if output_ids: | |
| response_ticks.append(tick) | |
| if result.get("quiescent", False) and ticks_executed >= self.min_ticks: | |
| stopped_on_quiescence = ticks_executed < self.max_ticks | |
| break | |
| after = self.state_digest() if self.state_digest else None | |
| return NetworkResponseSignature( | |
| first_response_latency=response_ticks[0] if response_ticks else None, | |
| last_response_latency=response_ticks[-1] if response_ticks else None, | |
| activated_neurons=len(response_neurons), | |
| total_spikes=total_spikes, | |
| peak_spike_rate=peak_rate, | |
| propagation_depth=( | |
| (response_ticks[-1] - response_ticks[0] + 1) if response_ticks else 0 | |
| ), | |
| recurrent_events=recurrent_events, | |
| return_latency=return_latency, | |
| spike_sequence=tuple(spike_sequence), | |
| network_state_digest_before=before, | |
| network_state_digest_after=after, | |
| ticks_executed=ticks_executed, | |
| delivered_synaptic_events=delivered_synaptic_events, | |
| synaptic_activity_ticks=synaptic_activity_ticks, | |
| max_synaptic_current_targets=max_synaptic_current_targets, | |
| total_synapses=total_synapses, | |
| stopped_on_quiescence=stopped_on_quiescence, | |
| ) | |