All checks were successful
Build Simulation and Test / Run All Tests (push) Successful in 2m53s
93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
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,
|
|
} |