Spaces:
Paused
Paused
Download src/controller/runtime.py from ThomasHeisig/Brain-5D-Space: direct link, hf CLI and curl.
- Browser
- Download file 34.6 kB
-
https://huggingface.co/spaces/ThomasHeisig/Brain-5D-Space/resolve/main/src/controller/runtime.py
- Command line
-
hf download hf://spaces/ThomasHeisig/Brain-5D-Space/src/controller/runtime.py
-
curl -L -o runtime.py https://huggingface.co/spaces/ThomasHeisig/Brain-5D-Space/resolve/main/src/controller/runtime.py
34.6 kB
| """Thread-safe runtime controller for interactive Brain-5D operation. | |
| The controller owns *when* ticks run, not the network implementation itself. | |
| It is intentionally dependency-light and works against a small typed Protocol. | |
| This provides a clean separation between: | |
| - The simulation core (network, neurons, synapses) | |
| - The control layer (start, pause, stop, step, run_ticks) | |
| - The presentation layer (dashboard, telemetry) | |
| The controller runs in a separate daemon thread and provides thread-safe | |
| access to the simulation state via locks and events. | |
| Example: | |
| >>> from src.controller import RuntimeController | |
| >>> controller = RuntimeController(network) | |
| >>> controller.start() | |
| >>> time.sleep(1) | |
| >>> controller.pause() | |
| >>> controller.run_ticks(10) | |
| >>> controller.stop() | |
| Integration with dashboard: | |
| >>> from src.dashboard.operator_bridge import OperatorBridge | |
| >>> bridge = OperatorBridge(controller=controller) | |
| >>> serve_dashboard(host="127.0.0.1", port=8765, structural_bridge=bridge) | |
| """ | |
| from __future__ import annotations | |
| import threading | |
| import time | |
| from collections.abc import Callable | |
| from dataclasses import dataclass, replace | |
| from enum import Enum | |
| from typing import Any, Protocol, cast | |
| PROFILE_PHASES = ( | |
| "network_step", | |
| "tick_segment", | |
| "post_tick_hooks", | |
| "learning", | |
| "homeostasis", | |
| "structural", | |
| "embodiment", | |
| "neural_symbiosis_msba", | |
| "dashboard_telemetry", | |
| "storage", | |
| ) | |
| # ============================================================================ | |
| # Protocols (Minimal contracts for loose coupling) | |
| # ============================================================================ | |
| class StepResultLike(Protocol): | |
| """Minimal result contract required by the controller.""" | |
| def spikes_this_tick(self) -> int: ... | |
| class RuntimeNetworkLike(Protocol): | |
| """Read/write boundary used by RuntimeController. | |
| Properties are used instead of mutable Protocol attributes so concrete | |
| dict/list implementations do not fail Pyright because of invariance. | |
| """ | |
| def current_tick(self) -> int: ... | |
| def synapse_count(self) -> int: ... | |
| def queued_event_count(self) -> int: ... | |
| def neuron_count(self) -> int: ... | |
| def step(self) -> StepResultLike: ... | |
| class HomeostasisLike(Protocol): | |
| """Minimal homeostasis contract required by the controller.""" | |
| def enabled(self) -> bool: ... | |
| def update(self, step_result: StepResultLike) -> None: ... | |
| # ============================================================================ | |
| # Enums | |
| # ============================================================================ | |
| class ControllerState(str, Enum): | |
| """Possible states of the runtime controller.""" | |
| IDLE = "idle" | |
| RUNNING = "running" | |
| PAUSED = "paused" | |
| STOPPED = "stopped" | |
| ERROR = "error" | |
| def is_active(self) -> bool: | |
| """Return True if the controller is in a running or paused state.""" | |
| return self in {ControllerState.RUNNING, ControllerState.PAUSED} | |
| def is_terminated(self) -> bool: | |
| """Return True if the controller is stopped or in error state.""" | |
| return self in {ControllerState.STOPPED, ControllerState.ERROR} | |
| class ControllerCommand(str, Enum): | |
| """Commands that can be sent to the runtime controller.""" | |
| START = "start" | |
| PAUSE = "pause" | |
| RESUME = "resume" | |
| STOP = "stop" | |
| STEP = "step" | |
| RUN_TICKS = "run_ticks" | |
| SNAPSHOT = "snapshot" | |
| def from_string(cls, value: str) -> ControllerCommand | None: | |
| """Convert a string to a ControllerCommand, or return None.""" | |
| try: | |
| return cls(value) | |
| except ValueError: | |
| return None | |
| # ============================================================================ | |
| # Telemetry | |
| # ============================================================================ | |
| class RuntimeTelemetry: | |
| """Snapshot of runtime telemetry data. | |
| Attributes: | |
| tick: Current simulation tick. | |
| ticks_per_second: Achieved tick rate (ticks per second). | |
| batch_duration_ms: Time taken for the last batch in milliseconds. | |
| spikes_this_batch: Number of spikes in the last batch. | |
| neurons: Total number of neurons in the network. | |
| synapses: Total number of synapses in the network. | |
| queue_depth: Number of queued spike events. | |
| controller_state: Current state of the controller. | |
| requested_ticks: Total ticks requested by manual commands. | |
| completed_ticks: Total ticks completed. | |
| last_error: Last error message (if any). | |
| """ | |
| tick: int | |
| ticks_per_second: float | |
| batch_duration_ms: float | |
| spikes_this_batch: int | |
| neurons: int | |
| synapses: int | |
| queue_depth: int | |
| controller_state: ControllerState | |
| requested_ticks: int | |
| completed_ticks: int | |
| last_error: str | None = None | |
| target_hz: float | None = None | |
| simulation_speed_ratio: float = 0.0 | |
| tick_latency_ms: float = 0.0 | |
| jitter_ms: float = 0.0 | |
| compute_saturation: float = 0.0 | |
| runtime_mode: str = "MAX" | |
| tick_profile: dict[str, float] | None = None | |
| max_possible_hz: float | None = None | |
| def to_dict(self) -> dict[str, Any]: | |
| """Convert to dictionary for JSON serialization.""" | |
| return { | |
| "tick": self.tick, | |
| "ticks_per_second": self.ticks_per_second, | |
| "batch_duration_ms": self.batch_duration_ms, | |
| "spikes_this_batch": self.spikes_this_batch, | |
| "neurons": self.neurons, | |
| "synapses": self.synapses, | |
| "queue_depth": self.queue_depth, | |
| "controller_state": self.controller_state.value, | |
| "requested_ticks": self.requested_ticks, | |
| "completed_ticks": self.completed_ticks, | |
| "last_error": self.last_error, | |
| "target_hz": self.target_hz, | |
| "simulation_speed_ratio": self.simulation_speed_ratio, | |
| "tick_latency_ms": self.tick_latency_ms, | |
| "jitter_ms": self.jitter_ms, | |
| "compute_saturation": self.compute_saturation, | |
| "runtime_mode": self.runtime_mode, | |
| "tick_profile": self.tick_profile, | |
| "max_possible_hz": self.max_possible_hz, | |
| } | |
| def to_json(self) -> dict[str, Any]: | |
| """Alias for to_dict() for DashboardControlService compatibility.""" | |
| return self.to_dict() | |
| # ============================================================================ | |
| # Callback Types | |
| # ============================================================================ | |
| SnapshotCallback = Callable[[], None] | |
| """Callback for snapshot requests.""" | |
| PostTickHook = Callable[[int, Any], None] | |
| """Callback after each tick, receives current tick number and the StepResult. | |
| The second argument is the StepResult-like object returned by network.step(). | |
| Hooks should accept ``**kwargs`` for forward compatibility. | |
| """ | |
| PreTickHook = Callable[[int], None] | |
| """Callback before each tick, receives current tick number before stepping.""" | |
| ErrorCallback = Callable[[Exception], None] | |
| """Callback when an error occurs in the runtime loop.""" | |
| # ============================================================================ | |
| # Runtime Controller | |
| # ============================================================================ | |
| class RuntimeController: | |
| """Own the simulation clock and expose safe operator commands. | |
| This controller provides thread-safe control over the Brain-5D simulation: | |
| - Start/stop continuous execution in a daemon thread | |
| - Pause/resume execution | |
| - Execute finite batches of ticks synchronously | |
| - Request snapshots at safe boundaries | |
| - Register post-tick hooks | |
| The controller is designed to be used with the dashboard's OperatorBridge | |
| for interactive control. | |
| Thread-safety: | |
| All public methods are thread-safe. The controller uses an RLock | |
| for state access and threading.Event for pause/stop signaling. | |
| Example: | |
| >>> controller = RuntimeController(network, homeostasis) | |
| >>> controller.add_hook(lambda tick: print(f"Tick {tick}")) | |
| >>> controller.start() | |
| >>> time.sleep(2) | |
| >>> controller.run_ticks(100) # runs synchronously | |
| >>> controller.pause() | |
| >>> controller.resume() | |
| >>> controller.stop() | |
| """ | |
| def __init__( | |
| self, | |
| network: RuntimeNetworkLike, | |
| homeostasis: HomeostasisLike | None = None, | |
| *, | |
| batch_size: int = 10, | |
| loop_delay_ms: float = 0.0, | |
| target_hz: float | None = None, | |
| telemetry_interval_ticks: int = 10, | |
| snapshot_callback: SnapshotCallback | None = None, | |
| max_manual_ticks: int = 100_000, | |
| ) -> None: | |
| """Initialize the runtime controller. | |
| Args: | |
| network: The network instance (must implement RuntimeNetworkLike). | |
| homeostasis: Optional homeostasis engine for rate regulation. | |
| batch_size: Number of ticks per batch in continuous mode. | |
| loop_delay_ms: Delay between batches in continuous mode. | |
| telemetry_interval_ticks: How often to update telemetry. | |
| snapshot_callback: Callback for snapshot requests. | |
| max_manual_ticks: Maximum ticks allowed in a single run_ticks call. | |
| Raises: | |
| ValueError: If batch_size, loop_delay_ms, telemetry_interval_ticks, | |
| or max_manual_ticks have invalid values. | |
| """ | |
| if batch_size <= 0: | |
| raise ValueError(f"batch_size must be > 0, got {batch_size}") | |
| if telemetry_interval_ticks <= 0: | |
| raise ValueError( | |
| f"telemetry_interval_ticks must be > 0, got {telemetry_interval_ticks}" | |
| ) | |
| if loop_delay_ms < 0: | |
| raise ValueError(f"loop_delay_ms must be >= 0, got {loop_delay_ms}") | |
| if target_hz is not None and target_hz <= 0: | |
| raise ValueError(f"target_hz must be > 0 or None, got {target_hz}") | |
| if max_manual_ticks <= 0: | |
| raise ValueError(f"max_manual_ticks must be > 0, got {max_manual_ticks}") | |
| self.network: RuntimeNetworkLike = network | |
| self.homeostasis: HomeostasisLike | None = homeostasis | |
| self._batch_size: int = batch_size | |
| self._loop_delay_ms: float = loop_delay_ms | |
| self._target_hz: float | None = target_hz | |
| self._telemetry_interval_ticks: int = telemetry_interval_ticks | |
| self._snapshot_callback: SnapshotCallback | None = snapshot_callback | |
| self._max_manual_ticks: int = max_manual_ticks | |
| self._state: ControllerState = ControllerState.IDLE | |
| self._lock: threading.RLock = threading.RLock() | |
| self._stop_event: threading.Event = threading.Event() | |
| self._pause_event: threading.Event = threading.Event() | |
| self._thread: threading.Thread | None = None | |
| self._hooks: list[PostTickHook] = [] | |
| self._pre_hooks: list[PreTickHook] = [] | |
| self._error_callbacks: list[ErrorCallback] = [] | |
| self._snapshot_requested: bool = False | |
| self._requested_ticks: int = 0 | |
| self._completed_ticks: int = 0 | |
| self._last_tick_latency_ms: float = 0.0 | |
| self._tick_latency_samples: list[float] = [] | |
| self._phase_totals_ms: dict[str, float] = { | |
| phase: 0.0 for phase in PROFILE_PHASES | |
| } | |
| self._telemetry: RuntimeTelemetry = self._make_telemetry(0.0, 0.0, 0) | |
| # ======================================================================== | |
| # Properties | |
| # ======================================================================== | |
| def state(self) -> ControllerState: | |
| """Get the current controller state (thread-safe).""" | |
| with self._lock: | |
| return self._state | |
| def telemetry(self) -> RuntimeTelemetry: | |
| """Get the latest telemetry snapshot (thread-safe).""" | |
| with self._lock: | |
| return self._telemetry | |
| def is_running(self) -> bool: | |
| """Check if the controller is currently running (thread-safe).""" | |
| with self._lock: | |
| return self._state == ControllerState.RUNNING | |
| def is_paused(self) -> bool: | |
| """Check if the controller is currently paused (thread-safe).""" | |
| with self._lock: | |
| return self._state == ControllerState.PAUSED | |
| def is_idle(self) -> bool: | |
| """Check if the controller is idle (thread-safe).""" | |
| with self._lock: | |
| return self._state == ControllerState.IDLE | |
| # ======================================================================== | |
| # Hook Management | |
| # ======================================================================== | |
| def add_hook(self, hook: PostTickHook) -> None: | |
| """Register a hook that runs after each tick. | |
| Args: | |
| hook: Callback receiving the current tick number and StepResult. | |
| """ | |
| with self._lock: | |
| if hook not in self._hooks: | |
| self._hooks.append(hook) | |
| def record_phase(self, phase: str, elapsed_ms: float) -> None: | |
| """Record an externally measured subsystem phase for runtime telemetry.""" | |
| if phase not in PROFILE_PHASES: | |
| raise ValueError(f"unknown runtime profile phase: {phase}") | |
| if elapsed_ms < 0.0: | |
| raise ValueError("elapsed_ms must be non-negative") | |
| with self._lock: | |
| self._phase_totals_ms[phase] = ( | |
| self._phase_totals_ms.get(phase, 0.0) + float(elapsed_ms) | |
| ) | |
| def remove_hook(self, hook: PostTickHook) -> bool: | |
| """Remove a previously registered hook. | |
| Returns: | |
| True if the hook was removed, False if not found. | |
| """ | |
| with self._lock: | |
| try: | |
| self._hooks.remove(hook) | |
| return True | |
| except ValueError: | |
| return False | |
| def clear_hooks(self) -> None: | |
| """Remove all registered hooks.""" | |
| with self._lock: | |
| self._hooks.clear() | |
| def add_pre_hook(self, hook: PreTickHook) -> None: | |
| """Register a hook that runs before each tick. | |
| Args: | |
| hook: Callback receiving the current tick number before stepping. | |
| """ | |
| with self._lock: | |
| if hook not in self._pre_hooks: | |
| self._pre_hooks.append(hook) | |
| def remove_pre_hook(self, hook: PreTickHook) -> bool: | |
| """Remove a previously registered pre-tick hook. | |
| Returns: | |
| True if the hook was removed, False if not found. | |
| """ | |
| with self._lock: | |
| try: | |
| self._pre_hooks.remove(hook) | |
| return True | |
| except ValueError: | |
| return False | |
| def add_error_callback(self, callback: ErrorCallback) -> None: | |
| """Register a callback for runtime errors. | |
| Args: | |
| callback: Function receiving the exception. | |
| """ | |
| with self._lock: | |
| if callback not in self._error_callbacks: | |
| self._error_callbacks.append(callback) | |
| # ======================================================================== | |
| # Control Commands | |
| # ======================================================================== | |
| def start(self) -> RuntimeTelemetry: | |
| """Start continuous execution in a daemon thread. | |
| If the controller is already running, this is a no-op. | |
| If the controller is paused, it resumes execution. | |
| If the controller is stopped, it creates a new thread. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| if self._state == ControllerState.RUNNING: | |
| return self._telemetry | |
| if self._state == ControllerState.PAUSED: | |
| self._state = ControllerState.RUNNING | |
| self._pause_event.clear() | |
| return self._telemetry | |
| if self._thread is not None and self._thread.is_alive(): | |
| self._state = ControllerState.RUNNING | |
| self._pause_event.clear() | |
| return self._telemetry | |
| # Start new thread | |
| self._stop_event.clear() | |
| self._pause_event.clear() | |
| self._state = ControllerState.RUNNING | |
| self._thread = threading.Thread( | |
| target=self._run_loop, | |
| name="brain5d-runtime", | |
| daemon=True, | |
| ) | |
| self._thread.start() | |
| return self._telemetry | |
| def pause(self) -> RuntimeTelemetry: | |
| """Pause continuous execution. | |
| The controller remains in a paused state until resume() is called. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| if self._state == ControllerState.RUNNING: | |
| self._state = ControllerState.PAUSED | |
| self._pause_event.set() | |
| return self._telemetry | |
| def resume(self) -> RuntimeTelemetry: | |
| """Resume continuous execution after pause. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| if self._state == ControllerState.PAUSED: | |
| self._state = ControllerState.RUNNING | |
| self._pause_event.clear() | |
| return self._telemetry | |
| def stop(self) -> RuntimeTelemetry: | |
| """Stop continuous execution gracefully. | |
| The controller thread will exit after completing the current batch. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| self._stop_event.set() | |
| self._pause_event.set() | |
| self._state = ControllerState.STOPPED | |
| return self._telemetry | |
| def step_once(self) -> RuntimeTelemetry: | |
| """Execute exactly one tick synchronously. | |
| This method is only available when the controller is not running | |
| continuously. | |
| Returns: | |
| Updated telemetry after the step. | |
| Raises: | |
| RuntimeError: If the controller is currently running. | |
| """ | |
| if self.state == ControllerState.RUNNING: | |
| raise RuntimeError("step_once is unavailable while running") | |
| return self.run_ticks(1) | |
| def single_step(self) -> RuntimeTelemetry: | |
| """Alias for step_once (for compatibility with OperatorBridge).""" | |
| return self.step_once() | |
| def run_ticks(self, count: int) -> RuntimeTelemetry: | |
| """Execute a finite batch of ticks synchronously. | |
| This method is only available when the controller is not running | |
| continuously. | |
| Args: | |
| count: Number of ticks to execute (1 - max_manual_ticks). | |
| Returns: | |
| Updated telemetry after the batch. | |
| Raises: | |
| TypeError: If count is not an integer. | |
| ValueError: If count is out of range. | |
| RuntimeError: If the controller is currently running. | |
| """ | |
| if isinstance(count, bool): | |
| raise TypeError("count must be an integer") | |
| if not 0 < count <= self._max_manual_ticks: | |
| raise ValueError(f"count must be in [1, {self._max_manual_ticks}]") | |
| if self.state == ControllerState.RUNNING: | |
| raise RuntimeError("run_ticks is unavailable while running") | |
| with self._lock: | |
| self._requested_ticks += count | |
| started = time.perf_counter() | |
| spikes = self._execute_ticks(count) | |
| elapsed_ms = (time.perf_counter() - started) * 1000.0 | |
| with self._lock: | |
| self._completed_ticks += count | |
| self._telemetry = self._make_telemetry(elapsed_ms, elapsed_ms, spikes) | |
| return self._telemetry | |
| def run_loop(self, count: int | None = None) -> RuntimeTelemetry: | |
| """Run a finite batch or start continuous execution. | |
| This is a convenience method that delegates to either run_ticks | |
| (if count is provided) or start() (if count is None). | |
| Args: | |
| count: Optional number of ticks to run. If provided, runs | |
| synchronously via run_ticks. If None, starts continuous | |
| execution. | |
| Returns: | |
| Current telemetry. | |
| """ | |
| if count is not None: | |
| return self.run_ticks(count) | |
| self.start() | |
| return self.telemetry | |
| def request_snapshot(self) -> RuntimeTelemetry: | |
| """Request a snapshot at the next safe controller boundary. | |
| If the controller is not running, the snapshot is taken immediately. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| self._snapshot_requested = True | |
| # If not running, flush immediately | |
| if self.state != ControllerState.RUNNING: | |
| self._flush_snapshot_request() | |
| return self.telemetry | |
| # ======================================================================== | |
| # Internal Methods | |
| # ======================================================================== | |
| def _run_loop(self) -> None: | |
| """Main loop for the daemon thread.""" | |
| last_tick = self.network.current_tick | |
| try: | |
| while not self._stop_event.is_set(): | |
| # Check pause | |
| if self._pause_event.is_set(): | |
| time.sleep(0.05) | |
| continue | |
| # Execute batch | |
| started = time.perf_counter() | |
| spikes = self._execute_ticks(self._batch_size) | |
| elapsed_ms = (time.perf_counter() - started) * 1000.0 | |
| # Update telemetry | |
| with self._lock: | |
| self._completed_ticks += self._batch_size | |
| tick_delta = self.network.current_tick - last_tick | |
| if tick_delta >= self._telemetry_interval_ticks: | |
| self._telemetry = self._make_telemetry( | |
| elapsed_ms, | |
| elapsed_ms, | |
| spikes, | |
| ) | |
| last_tick = self.network.current_tick | |
| # Flush snapshot request | |
| self._flush_snapshot_request() | |
| # Apply an optional target clock without changing SNN dt. | |
| target_delay_ms = 0.0 | |
| if self._target_hz is not None: | |
| target_batch_ms = self._batch_size * 1000.0 / self._target_hz | |
| target_delay_ms = max(0.0, target_batch_ms - elapsed_ms) | |
| delay_ms = max(self._loop_delay_ms, target_delay_ms) | |
| if delay_ms: | |
| time.sleep(delay_ms / 1000.0) | |
| except Exception as exc: | |
| # Handle errors and propagate to callbacks | |
| with self._lock: | |
| self._state = ControllerState.ERROR | |
| old = self._telemetry | |
| self._telemetry = RuntimeTelemetry( | |
| tick=old.tick, | |
| ticks_per_second=old.ticks_per_second, | |
| batch_duration_ms=old.batch_duration_ms, | |
| spikes_this_batch=old.spikes_this_batch, | |
| neurons=old.neurons, | |
| synapses=old.synapses, | |
| queue_depth=old.queue_depth, | |
| controller_state=ControllerState.ERROR, | |
| requested_ticks=old.requested_ticks, | |
| completed_ticks=old.completed_ticks, | |
| last_error=str(exc), | |
| target_hz=old.target_hz, | |
| simulation_speed_ratio=old.simulation_speed_ratio, | |
| tick_latency_ms=old.tick_latency_ms, | |
| jitter_ms=old.jitter_ms, | |
| compute_saturation=old.compute_saturation, | |
| runtime_mode=old.runtime_mode, | |
| tick_profile=old.tick_profile, | |
| max_possible_hz=old.max_possible_hz, | |
| ) | |
| # Notify error callbacks | |
| with self._lock: | |
| callbacks = tuple(self._error_callbacks) | |
| for callback in callbacks: | |
| try: | |
| callback(exc) | |
| except Exception: | |
| pass | |
| return | |
| with self._lock: | |
| self._state = ControllerState.STOPPED | |
| def _execute_ticks(self, count: int) -> int: | |
| """Execute a number of ticks and return total spikes.""" | |
| spikes_total = 0 | |
| self._phase_totals_ms = {phase: 0.0 for phase in PROFILE_PHASES} | |
| with self._lock: | |
| pre_hooks = tuple(self._pre_hooks) | |
| hooks = tuple(self._hooks) | |
| batched_results: tuple[StepResultLike, ...] | None = None | |
| if not pre_hooks and count > 1: | |
| batch_step = getattr(self.network, "step_batch", None) | |
| if callable(batch_step): | |
| network_started = time.perf_counter() | |
| typed_batch_step = cast( | |
| Callable[[int], tuple[StepResultLike, ...]], batch_step | |
| ) | |
| batched_results = typed_batch_step(count) | |
| network_elapsed_ms = (time.perf_counter() - network_started) * 1000.0 | |
| reported_core_ms = sum( | |
| float(getattr(result, "core_step_ms", 0.0)) | |
| for result in batched_results | |
| ) | |
| self._phase_totals_ms["network_step"] = ( | |
| reported_core_ms if reported_core_ms > 0 else network_elapsed_ms | |
| ) | |
| results = batched_results if batched_results is not None else (None,) * count | |
| for batched_result in results: | |
| tick_started = time.perf_counter() | |
| # Check stop signal | |
| if self._stop_event.is_set() and self.state == ControllerState.RUNNING: | |
| break | |
| # Run pre-tick hooks (e.g. stimulus) | |
| for hook in pre_hooks: | |
| phase_started = time.perf_counter() | |
| try: | |
| hook(self.network.current_tick) | |
| except Exception: | |
| pass # Hook errors are isolated | |
| self._record_phase("pre_tick_hooks", phase_started) | |
| # Execute one tick | |
| if batched_result is None: | |
| network_started = time.perf_counter() | |
| result = self.network.step() | |
| self._record_phase("network_step", network_started) | |
| else: | |
| result = batched_result | |
| self._phase_totals_ms["network_step"] = self._phase_totals_ms.get( | |
| "network_step", 0.0 | |
| ) + float(getattr(result, "core_step_ms", 0.0)) | |
| spikes_total += result.spikes_this_tick | |
| self._record_phase("tick_segment", tick_started) | |
| # Update homeostasis | |
| if self.homeostasis is not None and self.homeostasis.enabled: | |
| phase_started = time.perf_counter() | |
| self.homeostasis.update(result) | |
| self._record_phase("homeostasis", phase_started) | |
| # Run post-tick hooks | |
| post_started = time.perf_counter() | |
| for i in range(len(hooks)): | |
| try: | |
| hooks[i](self.network.current_tick, result) | |
| except Exception: | |
| pass # Hook errors are isolated | |
| self._record_phase("post_tick_hooks", post_started) | |
| tick_latency_ms = (time.perf_counter() - tick_started) * 1000.0 | |
| self._last_tick_latency_ms = tick_latency_ms | |
| self._tick_latency_samples.append(tick_latency_ms) | |
| if len(self._tick_latency_samples) > 1000: | |
| self._tick_latency_samples.pop(0) | |
| return spikes_total | |
| def _record_phase(self, phase: str, started: float) -> None: | |
| """Accumulate coarse runtime phase timings for the latest profile.""" | |
| elapsed_ms = (time.perf_counter() - started) * 1000.0 | |
| self._phase_totals_ms[phase] = ( | |
| self._phase_totals_ms.get(phase, 0.0) + elapsed_ms | |
| ) | |
| def _flush_snapshot_request(self) -> None: | |
| """Execute the snapshot callback if requested.""" | |
| with self._lock: | |
| requested = self._snapshot_requested | |
| self._snapshot_requested = False | |
| if requested and self._snapshot_callback is not None: | |
| try: | |
| self._snapshot_callback() | |
| except Exception: | |
| pass # Callback errors are isolated | |
| def _make_telemetry( | |
| self, | |
| batch_duration_ms: float, | |
| elapsed_ms: float, | |
| spikes: int, | |
| ) -> RuntimeTelemetry: | |
| """Create a telemetry snapshot.""" | |
| tps: float = 0.0 | |
| if elapsed_ms > 0: | |
| tps = self._batch_size * 1000.0 / elapsed_ms | |
| latency = self._last_tick_latency_ms | |
| jitter = 0.0 | |
| if len(self._tick_latency_samples) > 1: | |
| mean = sum(self._tick_latency_samples) / len(self._tick_latency_samples) | |
| jitter = ( | |
| sum((sample - mean) ** 2 for sample in self._tick_latency_samples) | |
| / len(self._tick_latency_samples) | |
| ) ** 0.5 | |
| target = self._target_hz | |
| ratio = tps / 1000.0 if target is None else tps / (1.0 / 0.001) | |
| saturation = ( | |
| 0.0 if target is None or target == 0 else min(1.0, target / max(tps, 0.001)) | |
| ) | |
| mode = ( | |
| "MAX" | |
| if target is None | |
| else ("COMPUTE LIMITED" if tps < target * 0.98 else "TARGETED") | |
| ) | |
| max_possible_hz = 1000.0 / latency if latency > 0 else None | |
| return RuntimeTelemetry( | |
| tick=self.network.current_tick, | |
| ticks_per_second=tps, | |
| batch_duration_ms=batch_duration_ms, | |
| spikes_this_batch=spikes, | |
| neurons=self.network.neuron_count, | |
| synapses=self.network.synapse_count, | |
| queue_depth=self.network.queued_event_count, | |
| controller_state=self._state, | |
| requested_ticks=self._requested_ticks, | |
| completed_ticks=self._completed_ticks, | |
| target_hz=target, | |
| simulation_speed_ratio=ratio, | |
| tick_latency_ms=latency, | |
| jitter_ms=jitter, | |
| compute_saturation=saturation, | |
| runtime_mode=mode, | |
| tick_profile=dict(self._phase_totals_ms), | |
| max_possible_hz=max_possible_hz, | |
| ) | |
| # ======================================================================== | |
| # Snapshot (for OperatorBridge compatibility) | |
| # ======================================================================== | |
| def snapshot(self) -> RuntimeTelemetry: | |
| """Return the current telemetry snapshot (for OperatorBridge). | |
| Returns: | |
| Current telemetry data. | |
| """ | |
| return self.telemetry | |
| # ======================================================================== | |
| # DashboardControlService Compatibility | |
| # ======================================================================== | |
| def step(self, ticks: int = 1) -> RuntimeTelemetry: | |
| """Execute a finite batch of ticks synchronously (alias for run_ticks). | |
| This method provides compatibility with DashboardControlService which | |
| expects a ``step(ticks)`` signature. | |
| Args: | |
| ticks: Number of ticks to execute. | |
| Returns: | |
| Updated telemetry after the batch. | |
| Raises: | |
| RuntimeError: If the controller is currently running continuously. | |
| """ | |
| return self.run_ticks(ticks) | |
| def run(self, *, loop_size: int | None = None) -> RuntimeTelemetry: # noqa: ARG001 | |
| """Start continuous execution (alias for start). | |
| This method provides compatibility with DashboardControlService which | |
| expects a ``run()`` signature. | |
| Args: | |
| loop_size: Ignored in this implementation; continuous execution | |
| uses the batch_size configured at construction time. | |
| Returns: | |
| Current telemetry. | |
| """ | |
| self.start() | |
| return self.telemetry | |
| def configure(self, **kwargs: Any) -> RuntimeTelemetry: | |
| """Configure runtime parameters (for OperatorBridge). | |
| Args: | |
| loop_size: Override batch_size. | |
| delay_ms: Override loop_delay_ms. | |
| Returns: | |
| Current telemetry snapshot. | |
| """ | |
| with self._lock: | |
| if "loop_size" in kwargs: | |
| loop_size = kwargs["loop_size"] | |
| if loop_size is not None: | |
| if not isinstance(loop_size, int) or loop_size <= 0: | |
| raise ValueError( | |
| f"loop_size must be a positive int, got {loop_size}" | |
| ) | |
| self._batch_size = loop_size | |
| if "delay_ms" in kwargs: | |
| delay_ms = kwargs["delay_ms"] | |
| if delay_ms is not None: | |
| if not isinstance(delay_ms, (int, float)) or delay_ms < 0: | |
| raise ValueError(f"delay_ms must be >= 0, got {delay_ms}") | |
| self._loop_delay_ms = float(delay_ms) | |
| if "target_hz" in kwargs: | |
| target_hz = kwargs["target_hz"] | |
| if target_hz is not None and ( | |
| not isinstance(target_hz, (int, float)) or target_hz <= 0 | |
| ): | |
| raise ValueError("target_hz must be > 0 or None") | |
| self._target_hz = float(target_hz) if target_hz is not None else None | |
| self._telemetry = replace( | |
| self._telemetry, | |
| target_hz=self._target_hz, | |
| runtime_mode="MAX" if self._target_hz is None else "TARGETED", | |
| ) | |
| return self._telemetry | |
| # ============================================================================ | |
| # Module Exports | |
| # ============================================================================ | |
| __all__ = [ | |
| # Protocols | |
| "StepResultLike", | |
| "RuntimeNetworkLike", | |
| "HomeostasisLike", | |
| # Enums | |
| "ControllerState", | |
| "ControllerCommand", | |
| # Telemetry | |
| "RuntimeTelemetry", | |
| # Callbacks | |
| "SnapshotCallback", | |
| "PostTickHook", | |
| "PreTickHook", | |
| "ErrorCallback", | |
| # Main class | |
| "RuntimeController", | |
| ] | |