import time import random import statistics import hashlib import pickle class HeadlessSimulationBenchmark: def __init__(self, setup_world, random_seed=42): """ :param setup_world: Callable that returns a World instance. :param random_seed: Seed for random number generation. """ self.setup_world = setup_world self.random_seed = random_seed self.world = None self.tps_history = [] self._running = False self.ticks_elapsed_time = None # Track time for designated ticks def set_random_seed(self, seed): self.random_seed = seed random.seed(seed) def start(self, ticks=100, max_seconds=None): self.set_random_seed(self.random_seed) self.world = self.setup_world(self.random_seed) self.tps_history.clear() self._running = True tick_count = 0 start_time = time.perf_counter() last_time = start_time # For precise tick timing tick_timing_start = None if ticks is not None: tick_timing_start = time.perf_counter() while self._running and (ticks is None or tick_count < ticks): self.world.tick_all() tick_count += 1 now = time.perf_counter() elapsed = now - last_time if elapsed > 0: self.tps_history.append(1.0 / elapsed) last_time = now if max_seconds and (now - start_time) > max_seconds: break if ticks is not None: tick_timing_end = time.perf_counter() self.ticks_elapsed_time = tick_timing_end - tick_timing_start else: self.ticks_elapsed_time = None self._running = False def stop(self): self._running = False def get_tps_history(self): return self.tps_history def get_tps_average(self): return statistics.mean(self.tps_history) if self.tps_history else 0.0 def get_tps_stddev(self): return statistics.stdev(self.tps_history) if len(self.tps_history) > 1 else 0.0 def get_simulation_hash(self): # Serialize the world state and hash it for determinism checks state = [] for obj in self.world.get_objects(): state.append(( type(obj).__name__, getattr(obj, "position", None), getattr(obj, "rotation", None), getattr(obj, "flags", None), getattr(obj, "interaction_radius", None), getattr(obj, "max_visual_width", None), )) state_bytes = pickle.dumps(state) return hashlib.sha256(state_bytes).hexdigest() def get_summary(self): return { "tps_avg": self.get_tps_average(), "tps_stddev": self.get_tps_stddev(), "ticks": len(self.tps_history), "simulation_hash": self.get_simulation_hash(), "ticks_elapsed_time": self.ticks_elapsed_time, }