neoLegacy/tools/performance-monitor/models.py
itsRevela be17d4028f feat: add performance monitor tool
DLL injector + Python GUI for real-time server metrics.
Hooks into tick loop, chunk generation, and player chunk map
to collect timing data over TCP, displayed in a tkinter dashboard.
Includes bot spawner for load testing.
2026-04-05 15:49:52 -05:00

123 lines
3.6 KiB
Python

"""
Data models for performance monitor snapshots.
All dataclasses are frozen (immutable) per project coding style.
Parsing functions validate at the boundary.
"""
from dataclasses import dataclass
@dataclass(frozen=True)
class LevelSnapshot:
dimension: int
level_tick_us: int
entity_tick_us: int
entity_tick_skipped: bool
tracker_tick_us: int
entity_count: int
global_entity_count: int
player_count: int
loaded_chunks: int
entities_to_remove: int
tile_entity_count: int
@dataclass(frozen=True)
class TickSnapshot:
tick: int
timestamp_ms: int
total_us: int
pool_reset_us: int
levels: tuple[LevelSnapshot, ...]
players_tick_us: int
connection_tick_us: int
console_tick_us: int
total_players: int
memory_used_mb: float
memory_total_mb: float
is_autosaving: bool
is_paused: bool
@property
def total_ms(self) -> float:
return self.total_us / 1000.0
@property
def tps(self) -> float:
"""Estimated TPS: 20 if tick fits in budget, lower if it overruns."""
if self.total_us <= 50000:
return 20.0
# Tick took longer than 50ms -- server can't keep up
return min(1_000_000 / self.total_us, 20.0)
@property
def total_entity_count(self) -> int:
return sum(lv.entity_count for lv in self.levels)
@dataclass(frozen=True)
class AutosaveSnapshot:
tick: int
timestamp_ms: int
total_us: int
players_us: int
levels_us: int
rules_us: int
flush_us: int
@property
def total_ms(self) -> float:
return self.total_us / 1000.0
def parse_level(data: dict) -> LevelSnapshot:
return LevelSnapshot(
dimension=data.get("dimension", 0),
level_tick_us=data.get("level_tick_us", 0),
entity_tick_us=data.get("entity_tick_us", 0),
entity_tick_skipped=data.get("entity_tick_skipped", False),
tracker_tick_us=data.get("tracker_tick_us", 0),
entity_count=data.get("entity_count", 0),
global_entity_count=data.get("global_entity_count", 0),
player_count=data.get("player_count", 0),
loaded_chunks=data.get("loaded_chunks", 0),
entities_to_remove=data.get("entities_to_remove", 0),
tile_entity_count=data.get("tile_entity_count", 0),
)
def parse_tick(data: dict) -> TickSnapshot:
phases = data.get("phases", {})
levels_raw = phases.get("levels", [])
return TickSnapshot(
tick=data.get("tick", 0),
timestamp_ms=data.get("timestamp_ms", 0),
total_us=data.get("total_us", 0),
pool_reset_us=phases.get("pool_reset_us", 0),
levels=tuple(parse_level(lv) for lv in levels_raw),
players_tick_us=phases.get("players_tick_us", 0),
connection_tick_us=phases.get("connection_tick_us", 0),
console_tick_us=phases.get("console_tick_us", 0),
total_players=data.get("server", {}).get("total_players", 0),
memory_used_mb=data.get("server", {}).get("memory_used_mb", 0.0),
memory_total_mb=data.get("server", {}).get("memory_total_mb", 0.0),
is_autosaving=data.get("server", {}).get("is_autosaving", False),
is_paused=data.get("server", {}).get("is_paused", False),
)
def parse_autosave(data: dict) -> AutosaveSnapshot:
breakdown = data.get("breakdown", {})
return AutosaveSnapshot(
tick=data.get("tick", 0),
timestamp_ms=data.get("timestamp_ms", 0),
total_us=data.get("total_us", 0),
players_us=breakdown.get("players_us", 0),
levels_us=breakdown.get("levels_us", 0),
rules_us=breakdown.get("rules_us", 0),
flush_us=breakdown.get("flush_us", 0),
)