Spaces:
Paused
Paused
File size: 5,163 Bytes
5e0b58b | 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 | """Measure RuntimeController pacing without changing simulation ``dt`` semantics."""
from __future__ import annotations
import argparse
import hashlib
import json
import platform
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from src.controller.runtime import RuntimeController
@dataclass(slots=True)
class _StepResult:
spikes_this_tick: int = 0
class _BenchmarkNetwork:
"""Minimal deterministic network used only to measure controller pacing."""
def __init__(self) -> None:
self.current_tick = 0
self.neuron_count = 1
self.synapse_count = 0
self.queued_event_count = 0
def step(self) -> _StepResult:
self.current_tick += 1
return _StepResult()
def run_deterministic_batch(
target_hz: float | None,
*,
ticks: int,
dt_ms: float = 1.0,
) -> dict[str, Any]:
"""Run identical synchronous simulation work under one pacing setting."""
if ticks <= 0:
raise ValueError("ticks must be positive")
network = _BenchmarkNetwork()
telemetry = RuntimeController(network, target_hz=target_hz).run_ticks(ticks)
state = json.dumps(
{"tick": network.current_tick, "dt_ms": dt_ms},
sort_keys=True,
separators=(",", ":"),
)
return {
"target_hz": target_hz,
"ticks": telemetry.completed_ticks,
"spikes": telemetry.spikes_this_batch,
"state_digest": hashlib.sha256(state.encode("utf-8")).hexdigest(),
"dt_ms": dt_ms,
}
def run_pacing_case(
target_hz: float | None,
*,
duration_seconds: float = 0.2,
batch_size: int = 1,
dt_ms: float = 1.0,
) -> dict[str, Any]:
"""Run one bounded pacing case and return a machine-readable measurement."""
if duration_seconds <= 0.0:
raise ValueError("duration_seconds must be positive")
if target_hz is not None and target_hz <= 0.0:
raise ValueError("target_hz must be positive or None")
network = _BenchmarkNetwork()
controller = RuntimeController(
network,
batch_size=batch_size,
target_hz=target_hz,
telemetry_interval_ticks=1,
)
controller.start()
warmup_deadline = time.perf_counter() + max(0.1, duration_seconds)
while controller.telemetry.completed_ticks == 0:
if time.perf_counter() >= warmup_deadline:
raise RuntimeError("RuntimeController did not complete a warmup tick")
time.sleep(0.001)
warmup_ticks = controller.telemetry.completed_ticks
measurement_duration = duration_seconds
if target_hz is not None:
measurement_duration = max(
measurement_duration, (3.0 * batch_size) / target_hz
)
started = time.perf_counter()
time.sleep(measurement_duration)
controller.stop()
elapsed = max(time.perf_counter() - started, 1e-9)
telemetry = controller.telemetry
completed_ticks = max(0, telemetry.completed_ticks - warmup_ticks)
achieved_hz = completed_ticks / elapsed
return {
"target_hz": target_hz,
"achieved_hz": achieved_hz,
"realtime_ratio": (
None if target_hz is None else achieved_hz / target_hz
),
"duration_seconds": elapsed,
"measurement_duration_seconds": measurement_duration,
"warmup_ticks": warmup_ticks,
"ticks": completed_ticks,
"dt_ms": dt_ms,
"tick_cost_ms": telemetry.tick_latency_ms,
"tick_profile_ms": dict(telemetry.tick_profile),
"profile_phase_contract": sorted(telemetry.tick_profile or {}),
"controller_runtime_mode": telemetry.runtime_mode,
"simulation_tick": network.current_tick,
}
def build_report(
*,
duration_seconds: float = 0.2,
targets: tuple[float | None, ...] = (1.0, 10.0, None),
dt_ms: float = 1.0,
) -> dict[str, Any]:
"""Build a bounded pacing report for low-rate, targeted and unlimited modes."""
return {
"schema_version": 1,
"benchmark": "runtime_pacing",
"python": platform.python_version(),
"platform": platform.platform(),
"duration_seconds_requested": duration_seconds,
"dt_ms": dt_ms,
"cases": [
run_pacing_case(target, duration_seconds=duration_seconds, dt_ms=dt_ms)
for target in targets
],
"scientific_claim": False,
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--duration", type=float, default=0.2)
parser.add_argument("--dt-ms", type=float, default=1.0)
parser.add_argument("--output", type=Path)
args = parser.parse_args()
report = build_report(duration_seconds=args.duration, dt_ms=args.dt_ms)
payload = json.dumps(report, indent=2)
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(payload + "\n", encoding="utf-8")
print(payload)
return 0
if __name__ == "__main__":
raise SystemExit(main()) |