Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions config/performance.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
performance:
target_fps: 60
asset_cache_limit_mb: 200
scene_cache_mode: "on_demand"
diagnostics_enabled: true
unload_idle_scenes: true

profiles:
balanced:
target_fps: 60
cache_limit: 200
diagnostics: true
low_power:
target_fps: 30
cache_limit: 100
diagnostics: false
max_quality:
target_fps: 120
cache_limit: 400
diagnostics: false
115 changes: 115 additions & 0 deletions engine/performance_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
from __future__ import annotations

import logging
import os
import sys
from typing import List, Dict, Optional

try: # pragma: no cover - allow running tests without pygame
import pygame
except Exception: # pragma: no cover - allow running tests without pygame
pygame = None

try:
import yaml
except Exception: # pragma: no cover - allow running tests without PyYAML
yaml = None

from .asset_manager import AssetManager

logger = logging.getLogger(__name__)


class PerformanceManager:
"""Runtime performance tuning and diagnostics."""

def __init__(self, config_path: str = "config/performance.yaml") -> None:
self.config: Dict = self._load_config(config_path)
perf = self.config.get("performance", {})
self.target_fps: int = int(perf.get("target_fps", 60))
self.clock = pygame.time.Clock() if pygame else None
self.asset_cache = AssetManager()
self.resource_log: List[str] = []
self.frame_times: List[float] = []
self.diagnostics_enabled: bool = bool(perf.get("diagnostics_enabled", False))
self.max_cache_size_mb: int = int(perf.get("asset_cache_limit_mb", 200))
self._profiles: Dict[str, Dict] = self.config.get("profiles", {})

# ------------------------------------------------------------------
# Config helpers
# ------------------------------------------------------------------
def _load_config(self, path: str) -> Dict:
if not yaml or not os.path.exists(path):
return {}
with open(path, "r", encoding="utf-8") as fh:
return yaml.safe_load(fh) or {}

# ------------------------------------------------------------------
# Framerate helpers
# ------------------------------------------------------------------
def set_target_fps(self, fps: int) -> None:
self.target_fps = int(fps)

def track_frame(self) -> None:
if not self.clock:
return
ms = self.clock.tick(self.target_fps)
self.frame_times.append(ms / 1000.0)
if len(self.frame_times) > 60:
self.frame_times.pop(0)

def enforce_frame_delay(self) -> None:
if self.clock:
self.clock.tick(self.target_fps)

# ------------------------------------------------------------------
# Diagnostics & cache helpers
# ------------------------------------------------------------------
def log_resource_usage(self) -> None:
if not self.diagnostics_enabled:
return
current = self.cache_size_mb()
msg = f"Cache size: {current:.2f} MB"
self.resource_log.append(msg)
logger.debug(msg)

def cache_size_mb(self) -> float:
size = 0
for coll in (self.asset_cache.image_cache, self.asset_cache.sound_cache, self.asset_cache.music_cache):
for obj in coll.values():
try:
size += sys.getsizeof(obj)
except Exception:
pass
return size / (1024 * 1024)

def clear_unused_assets(self) -> None:
self.asset_cache.clear_cache()

def trim_cache(self) -> None:
if self.cache_size_mb() > self.max_cache_size_mb:
self.clear_unused_assets()

def get_diagnostics(self) -> Dict:
avg_fps = 0.0
if self.frame_times:
avg_fps = 1.0 / (sum(self.frame_times) / len(self.frame_times))
return {
"average_fps": avg_fps,
"cache_size_mb": self.cache_size_mb(),
"frames_tracked": len(self.frame_times),
}

def flush_scene_cache(self, scene_id: str) -> None:
self.clear_unused_assets()

def apply_performance_profile(self, profile_name: str) -> None:
profile = self._profiles.get(profile_name, {})
if not profile:
return
if "target_fps" in profile:
self.target_fps = int(profile["target_fps"])
if "cache_limit" in profile:
self.max_cache_size_mb = int(profile["cache_limit"])
if "diagnostics" in profile:
self.diagnostics_enabled = bool(profile["diagnostics"])
Loading