from engine.effects.performance import get_monitor from engine.effects.registry import get_registry _effect_chain_ref = None def _get_effect_chain(): global _effect_chain_ref if _effect_chain_ref is not None: return _effect_chain_ref try: from engine.legacy.layers import get_effect_chain as _chain return _chain() except Exception: return None def set_effect_chain_ref(chain) -> None: global _effect_chain_ref _effect_chain_ref = chain def handle_effects_command(cmd: str) -> str: """Handle /effects command from NTFY message. Commands: /effects list - list all effects and their status /effects on - enable an effect /effects off - disable an effect /effects intensity <0.0-1.0> - set intensity /effects reorder ,,... - reorder pipeline /effects stats - show performance statistics """ parts = cmd.strip().split() if not parts or parts[0] != "/effects": return "Unknown command" registry = get_registry() chain = _get_effect_chain() if len(parts) == 1 or parts[1] == "list": result = ["Effects:"] for name, plugin in registry.list_all().items(): status = "ON" if plugin.config.enabled else "OFF" intensity = plugin.config.intensity result.append(f" {name}: {status} (intensity={intensity})") if chain: result.append(f"Order: {chain.get_order()}") return "\n".join(result) if parts[1] == "stats": return _format_stats() if parts[1] == "reorder" and len(parts) >= 3: new_order = parts[2].split(",") if chain and chain.reorder(new_order): return f"Reordered pipeline: {new_order}" return "Failed to reorder pipeline" if len(parts) < 3: return "Usage: /effects on|off|intensity " effect_name = parts[1] action = parts[2] if effect_name not in registry.list_all(): return f"Unknown effect: {effect_name}" if action == "on": registry.enable(effect_name) return f"Enabled: {effect_name}" if action == "off": registry.disable(effect_name) return f"Disabled: {effect_name}" if action == "intensity" and len(parts) >= 4: try: value = float(parts[3]) if not 0.0 <= value <= 1.0: return "Intensity must be between 0.0 and 1.0" plugin = registry.get(effect_name) if plugin: plugin.config.intensity = value return f"Set {effect_name} intensity to {value}" except ValueError: return "Invalid intensity value" return f"Unknown action: {action}" def _format_stats() -> str: monitor = get_monitor() stats = monitor.get_stats() if "error" in stats: return stats["error"] lines = ["Performance Stats:"] pipeline = stats["pipeline"] lines.append( f" Pipeline: avg={pipeline['avg_ms']:.2f}ms min={pipeline['min_ms']:.2f}ms max={pipeline['max_ms']:.2f}ms (over {stats['frame_count']} frames)" ) if stats["effects"]: lines.append(" Per-effect (avg ms):") for name, effect_stats in stats["effects"].items(): lines.append( f" {name}: avg={effect_stats['avg_ms']:.2f}ms min={effect_stats['min_ms']:.2f}ms max={effect_stats['max_ms']:.2f}ms" ) return "\n".join(lines) def show_effects_menu() -> str: """Generate effects menu text for display.""" registry = get_registry() chain = _get_effect_chain() lines = [ "\033[1;38;5;231m=== EFFECTS MENU ===\033[0m", "", "Effects:", ] for name, plugin in registry.list_all().items(): status = "ON" if plugin.config.enabled else "OFF" intensity = plugin.config.intensity lines.append(f" [{status:3}] {name}: intensity={intensity:.2f}") if chain: lines.append("") lines.append(f"Pipeline order: {' -> '.join(chain.get_order())}") lines.append("") lines.append("Controls:") lines.append(" /effects on|off") lines.append(" /effects intensity <0.0-1.0>") lines.append(" /effects reorder name1,name2,...") lines.append("") return "\n".join(lines)