- Extract effects as fully decoupled plugins in engine/effects/ - Add EffectConfig, EffectContext dataclasses and EffectPlugin protocol - Add EffectRegistry for plugin discovery and management - Add EffectChain for ordered pipeline execution - Move built-in effects to effects_plugins/ directory - Add interactive effects config picker during startup - Add NTFY command handler for /effects commands - Add tests for effects system (24 new tests) - Update AGENTS.md with effects plugin documentation - Add conventional commits section to AGENTS.md chore: add coverage.xml to .gitignore
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
from engine.effects.registry import get_registry
|
|
|
|
|
|
def _get_effect_chain():
|
|
from engine.layers import get_effect_chain as _chain
|
|
|
|
return _chain()
|
|
|
|
|
|
def handle_effects_command(cmd: str) -> str:
|
|
"""Handle /effects command from NTFY message.
|
|
|
|
Commands:
|
|
/effects list - list all effects and their status
|
|
/effects <name> on - enable an effect
|
|
/effects <name> off - disable an effect
|
|
/effects <name> intensity <0.0-1.0> - set intensity
|
|
/effects reorder <name1>,<name2>,... - reorder pipeline
|
|
"""
|
|
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 len(parts) < 3:
|
|
return "Usage: /effects <name> on|off|intensity <value>"
|
|
|
|
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"
|
|
|
|
if action == "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"
|
|
|
|
return f"Unknown action: {action}"
|
|
|
|
|
|
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 <name> on|off")
|
|
lines.append(" /effects <name> intensity <0.0-1.0>")
|
|
lines.append(" /effects reorder name1,name2,...")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|