# -*- coding: utf-8 -*- """ Expression Condition - Evaluiert Python-Ausdrücke mit App-Zugriff. """ from typing import Any, TYPE_CHECKING from trixy_core.scheduler.condition.base import Condition from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError if TYPE_CHECKING: from trixy_core.application import TrixyApplication class ExpressionCondition(Condition): """ Evaluiert einen Python-Ausdruck mit Zugriff auf App-Komponenten. Verfügbare Variablen im Ausdruck: - app: TrixyApplication Instanz - satellites: SatelliteManager - events: EventManager - plugins: PluginManager - config: ConfigManager - scheduler: Scheduler - context: Aktueller Condition-Kontext Beispiele: ```python # Satellite-Anzahl prüfen ExpressionCondition("satellites.count > 0") # Musik-Status prüfen ExpressionCondition("app.music_player.state == 'stopped'") # Kombinierte Bedingung ExpressionCondition( "satellites.count > 0 and events.listener_count('wakeword') > 0" ) # Mit Kontext-Variable ExpressionCondition("context.get('temperature', 0) > 25") # Satellite in bestimmtem Raum verbunden ExpressionCondition("satellites.get_by_room('kitchen') is not None") # Plugin aktiviert ExpressionCondition("plugins.is_enabled('spotify')") ``` """ def __init__( self, expression: str, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: expression: Python-Ausdruck der zu bool evaluiert wird name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name or f"expr:{expression[:30]}", invert) self._expression = expression @property def expression(self) -> str: """Python-Ausdruck.""" return self._expression def _build_eval_context(self, context: dict) -> dict[str, Any]: """ Erstellt den Evaluierungs-Kontext mit App-Zugriff. Args: context: Condition-Kontext Returns: Dictionary mit allen verfügbaren Variablen """ eval_ctx: dict[str, Any] = { # Basis-Kontext "context": context, "ctx": context, # Python builtins (eingeschränkt) "len": len, "str": str, "int": int, "float": float, "bool": bool, "list": list, "dict": dict, "set": set, "tuple": tuple, "min": min, "max": max, "sum": sum, "abs": abs, "round": round, "any": any, "all": all, "sorted": sorted, "reversed": reversed, "enumerate": enumerate, "zip": zip, "range": range, "isinstance": isinstance, "hasattr": hasattr, "getattr": getattr, "None": None, "True": True, "False": False, } # App aus Kontext holen app = context.get("app") or context.get("application") if app: eval_ctx["app"] = app # Manager-Shortcuts if hasattr(app, "satellite_manager"): eval_ctx["satellites"] = app.satellite_manager eval_ctx["satellite_manager"] = app.satellite_manager if hasattr(app, "event_manager"): eval_ctx["events"] = app.event_manager eval_ctx["event_manager"] = app.event_manager if hasattr(app, "plugin_manager"): eval_ctx["plugins"] = app.plugin_manager eval_ctx["plugin_manager"] = app.plugin_manager if hasattr(app, "config_manager"): eval_ctx["config"] = app.config_manager eval_ctx["config_manager"] = app.config_manager if hasattr(app, "scheduler"): eval_ctx["scheduler"] = app.scheduler # Weitere App-Attribute if hasattr(app, "music_player"): eval_ctx["music"] = app.music_player eval_ctx["music_player"] = app.music_player if hasattr(app, "conversation_manager"): eval_ctx["conversations"] = app.conversation_manager if hasattr(app, "wakeword_service"): eval_ctx["wakeword"] = app.wakeword_service # Direkte Manager aus Kontext (falls ohne App) for key in ["satellite_manager", "event_manager", "plugin_manager", "config_manager", "scheduler"]: if key in context and key not in eval_ctx: eval_ctx[key] = context[key] # Kurznamen short_name = key.replace("_manager", "s").replace("_", "") if short_name not in eval_ctx: eval_ctx[short_name] = context[key] return eval_ctx def _evaluate(self, context: dict) -> bool: """Evaluiert den Ausdruck.""" try: eval_ctx = self._build_eval_context(context) # Ausdruck evaluieren result = eval(self._expression, {"__builtins__": {}}, eval_ctx) return bool(result) except Exception as e: from trixy_core.utils.debug import perror perror(f"Expression-Fehler '{self._expression}': {e}") return False def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "expression": self._expression, } @classmethod def from_dict(cls, data: dict) -> "ExpressionCondition": """Erstellt Condition aus Dictionary.""" return cls( expression=data["expression"], name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Ausdruck" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="expression", label="Python-Ausdruck", field_type=FieldType.TEXTAREA, required=True, lines=3, help_text="Python-Ausdruck der True/False zurueckgibt", placeholder="satellites.count > 0", ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] expr = data.get("expression", "").strip() if not expr: errors.append(FormValidationError("expression", "Python-Ausdruck ist erforderlich")) return errors def __repr__(self) -> str: return f"ExpressionCondition({self._expression!r})" class LambdaCondition(Condition): """ Condition mit Lambda-Funktion. Für komplexere Logik die nicht in einen Einzeiler passt, aber trotzdem inline definiert werden soll. Beispiel: ```python LambdaCondition( lambda ctx: ( ctx.get("satellites", {}).get("count", 0) > 0 and ctx.get("last_activity") is not None ) ) ``` """ def __init__( self, func: callable, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: func: Lambda/Funktion die context → bool mapped name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name or "lambda", invert) self._func = func def _evaluate(self, context: dict) -> bool: """Ruft die Funktion auf.""" try: result = self._func(context) return bool(result) except Exception as e: from trixy_core.utils.debug import perror perror(f"Lambda-Fehler: {e}") return False @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Lambda-Bedingung" def to_dict(self) -> dict: """Konvertiert zu Dictionary (ohne Funktion).""" return { **self._base_dict(), # Lambda kann nicht serialisiert werden "func": "", } # Shortcut-Funktionen für häufige Expressions def satellites_connected(min_count: int = 1) -> ExpressionCondition: """Condition: Mindestens X Satellites verbunden.""" return ExpressionCondition( f"satellites.count >= {min_count}", name=f"satellites_min_{min_count}", ) def satellite_in_room(room: str) -> ExpressionCondition: """Condition: Satellite in bestimmtem Raum verbunden.""" return ExpressionCondition( f"satellites.get_by_room('{room}') is not None", name=f"satellite_in_{room}", ) def plugin_enabled(plugin_name: str) -> ExpressionCondition: """Condition: Plugin ist aktiviert.""" return ExpressionCondition( f"plugins.is_enabled('{plugin_name}')", name=f"plugin_{plugin_name}_enabled", ) def music_playing() -> ExpressionCondition: """Condition: Musik wird abgespielt.""" return ExpressionCondition( "music_player is not None and music_player.is_playing", name="music_playing", ) def music_stopped() -> ExpressionCondition: """Condition: Musik ist gestoppt.""" return ExpressionCondition( "music_player is None or not music_player.is_playing", name="music_stopped", ) def conversation_active() -> ExpressionCondition: """Condition: Eine Konversation ist aktiv.""" return ExpressionCondition( "conversations is not None and conversations.has_active_session()", name="conversation_active", ) def no_conversation() -> ExpressionCondition: """Condition: Keine aktive Konversation.""" return ExpressionCondition( "conversations is None or not conversations.has_active_session()", name="no_conversation", )