import asyncio import json import os try: import decky except ImportError: # Allow running tests without the decky module (tests mock the path) decky = None def parse_mangohud_log(log_content: str) -> dict: """Parse a MangoHud log file's content and return FPS stats. Returns: {fpsAvg, fpsLow, fpsHigh, fpsOnePercentLow, tdpWatts, error?} """ lines = log_content.strip().split('\n') header_idx = None fps_col = 0 frametime_col = None gpu_power_col = None for i, line in enumerate(lines): stripped = line.strip() if not stripped or stripped.startswith('#'): continue if 'fps' in stripped.lower() and ',' in stripped: columns = [c.strip().lower() for c in stripped.split(',')] if 'fps' in columns: fps_col = columns.index('fps') if 'frametime' in columns: frametime_col = columns.index('frametime') if 'gpu_power' in columns: gpu_power_col = columns.index('gpu_power') header_idx = i break if header_idx is None: return {"error": "Could not find FPS column in log header"} fps_values = [] frametime_values = [] gpu_power_values = [] for line in lines[header_idx + 1:]: stripped = line.strip() if not stripped or stripped.startswith('#'): continue if stripped.startswith(('97%', 'AVG', '1%', '0.1%', '5%')): continue parts = [p.strip() for p in stripped.split(',')] try: fps = float(parts[fps_col]) fps_values.append(fps) if frametime_col is not None and frametime_col < len(parts): frametime_values.append(float(parts[frametime_col])) if gpu_power_col is not None and gpu_power_col < len(parts): gpu_power_values.append(float(parts[gpu_power_col])) except (ValueError, IndexError): continue if not fps_values: return {"error": "No FPS data found in log"} fps_avg = round(sum(fps_values) / len(fps_values), 1) fps_low = round(min(fps_values), 1) fps_high = round(max(fps_values), 1) # 1% low: average the slowest 1% of frame times (largest frametimes), # then convert to FPS. Falls back to the lowest FPS percentile if no # frametime data is available. if frametime_values: sorted_ft = sorted(frametime_values) one_percent_count = max(1, int(len(sorted_ft) * 0.01)) worst_ft = sorted_ft[-one_percent_count:] avg_worst_ft = sum(worst_ft) / len(worst_ft) fps_one_percent_low = round(1000.0 / avg_worst_ft, 1) if avg_worst_ft > 0 else None else: sorted_fps = sorted(fps_values) one_percent_idx = max(0, int(len(sorted_fps) * 0.01)) fps_one_percent_low = round(sorted_fps[one_percent_idx], 1) tdp_watts = None if gpu_power_values: tdp_watts = round(sum(gpu_power_values) / len(gpu_power_values), 1) return { "fpsAvg": fps_avg, "fpsLow": fps_low, "fpsHigh": fps_high, "fpsOnePercentLow": fps_one_percent_low, "tdpWatts": tdp_watts, } class Plugin: async def _main(self): if decky: decky.logger.info(f"DeckyVault plugin loaded: {decky.DECKY_PLUGIN_NAME}") self._settings_path = self._get_settings_path() self._settings = self._read_settings() async def _unload(self): if decky: decky.logger.info("DeckyVault plugin unloading") async def _uninstall(self): if decky: decky.logger.info("DeckyVault plugin uninstalled") def _get_settings_path(self): if decky: return os.path.join(decky.DECKY_PLUGIN_SETTINGS_DIR, "settings.json") return os.path.join(os.path.expanduser("~"), ".deckyvault-test", "settings.json") def _read_settings(self): """Read settings from JSON file. Returns empty dict if file missing.""" if os.path.exists(self._settings_path): try: with open(self._settings_path, 'r') as f: return json.load(f) except (json.JSONDecodeError, IOError): return {} return {} def _write_settings(self, settings): """Write settings to JSON file, creating directory if needed.""" os.makedirs(os.path.dirname(self._settings_path), exist_ok=True) with open(self._settings_path, 'w') as f: json.dump(settings, f, indent=2) async def get_settings(self) -> dict: """RPC: Return all plugin settings.""" return self._settings async def set_setting(self, key: str, value) -> dict: """RPC: Set a single setting and persist. Returns updated settings.""" self._settings[key] = value self._write_settings(self._settings) return self._settings async def check_mangohud(self) -> dict: """RPC: Check if MangoHud is installed. Returns {installed: bool, path: str, version: str}.""" import subprocess try: result = subprocess.run( ["which", "mangohud"], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: mangohud_path = result.stdout.strip() # Get version version_result = subprocess.run( ["mangohud", "--version"], capture_output=True, text=True, timeout=5 ) version = version_result.stdout.strip() if version_result.returncode == 0 else "unknown" return {"installed": True, "path": mangohud_path, "version": version} else: return {"installed": False, "path": "", "version": ""} except Exception as e: return {"installed": False, "path": "", "version": "", "error": str(e)} async def write_mangohud_config(self) -> dict: """RPC: Write the MangoHud logging config to ~/.config/MangoHud/MangoHud.conf. Returns {success: bool, path: str, error: str?}.""" try: home = os.path.expanduser("~") config_dir = os.path.join(home, ".config", "MangoHud") config_path = os.path.join(config_dir, "MangoHud.conf") os.makedirs(config_dir, exist_ok=True) # MangoHud config that enables logging with the metrics we need. # output_folder is required for logging to work. # We log to /tmp so the plugin can read it after the session. config_content = """\ # DeckyVault MangoHud logging config output_folder=/tmp output_file=deckyvault-mangohud.log log_duration=0 fps frame_timing cpu_power gpu_power cpu_temp gpu_temp benchmark_percentiles=97,AVG,1,0.1 """ with open(config_path, 'w') as f: f.write(config_content) return {"success": True, "path": config_path} except Exception as e: return {"success": False, "path": "", "error": str(e)} async def get_mangohud_config(self) -> dict: """RPC: Read the current MangoHud config. Returns {exists: bool, content: str, path: str}.""" home = os.path.expanduser("~") config_path = os.path.join(home, ".config", "MangoHud", "MangoHud.conf") if os.path.exists(config_path): with open(config_path, 'r') as f: return {"exists": True, "content": f.read(), "path": config_path} return {"exists": False, "content": "", "path": config_path} async def read_and_parse_mangohud_log(self, log_path: str = "/tmp/deckyvault-mangohud.log") -> dict: """RPC: Read the MangoHud log file and return parsed FPS stats. Returns parsed stats dict or {error: str}.""" if not os.path.exists(log_path): return {"error": f"MangoHud log not found at {log_path}. Make sure MangoHud is enabled and logging."} try: with open(log_path, 'r') as f: content = f.read() if not content.strip(): return {"error": "MangoHud log is empty. Recording may have been too short."} return parse_mangohud_log(content) except Exception as e: return {"error": f"Failed to read log: {str(e)}"} async def clear_mangohud_log(self, log_path: str = "/tmp/deckyvault-mangohud.log") -> dict: """RPC: Delete the MangoHud log file so the next recording starts fresh.""" try: if os.path.exists(log_path): os.remove(log_path) return {"success": True} except Exception as e: return {"success": False, "error": str(e)}