entity/configs/node/loop_timer.py - main - githubmirror/ChatDev

entity/configs/node/loop_timer.py
133 строки4 KB
laansdole
feat: move duration as enum option
10 фев 2026, 12:23
10 фев 2026, 12:237807e5a
"""Configuration for loop timer guard nodes."""

from dataclasses import dataclass
from typing import Mapping, Any, Optional

from entity.configs.base import (
    BaseConfig,
    ConfigError,
    ConfigFieldSpec,
    EnumOption,
    require_mapping,
    extend_path,
    optional_str,
)


@dataclass
class LoopTimerConfig(BaseConfig):
    """Configuration schema for the loop timer node type."""

    max_duration: float = 60.0
    duration_unit: str = "seconds"
    reset_on_emit: bool = True
    message: Optional[str] = None
    passthrough: bool = False

    @classmethod
    def from_dict(
        cls, data: Mapping[str, Any] | None, *, path: str
    ) -> "LoopTimerConfig":
        mapping = require_mapping(data or {}, path)
        max_duration_raw = mapping.get("max_duration", 60.0)
        try:
            max_duration = float(max_duration_raw)
        except (TypeError, ValueError) as exc:  # pragma: no cover - defensive
            raise ConfigError(
                "max_duration must be a number",
                extend_path(path, "max_duration"),
            ) from exc

        if max_duration <= 0:
            raise ConfigError(
                "max_duration must be > 0", extend_path(path, "max_duration")
            )

        duration_unit = str(mapping.get("duration_unit", "seconds"))
        valid_units = ["seconds", "minutes", "hours"]
        if duration_unit not in valid_units:
            raise ConfigError(
                f"duration_unit must be one of: {', '.join(valid_units)}",
                extend_path(path, "duration_unit"),
            )

        reset_on_emit = bool(mapping.get("reset_on_emit", True))
        message = optional_str(mapping, "message", path)
        passthrough = bool(mapping.get("passthrough", False))

        return cls(
            max_duration=max_duration,
            duration_unit=duration_unit,
            reset_on_emit=reset_on_emit,
            message=message,
            passthrough=passthrough,
            path=path,
        )

    def validate(self) -> None:
        if self.max_duration <= 0:
            raise ConfigError(
                "max_duration must be > 0", extend_path(self.path, "max_duration")
            )

        valid_units = ["seconds", "minutes", "hours"]
        if self.duration_unit not in valid_units:
            raise ConfigError(
                f"duration_unit must be one of: {', '.join(valid_units)}",
                extend_path(self.path, "duration_unit"),
            )

    FIELD_SPECS = {
        "max_duration": ConfigFieldSpec(
            name="max_duration",
            display_name="Maximum Duration",
            type_hint="float",
            required=True,
            default=60.0,
            description="How long the loop can run before this node emits an output.",
        ),
        "duration_unit": ConfigFieldSpec(
            name="duration_unit",
            display_name="Duration Unit",
            type_hint="str",
            required=True,
            default="seconds",
            description="Unit of time for max_duration: 'seconds', 'minutes', or 'hours'.",
            enum=["seconds", "minutes", "hours"],
            enum_options=[
                EnumOption(
                    value="seconds", label="Seconds", description="Time in seconds"
                ),
                EnumOption(
                    value="minutes", label="Minutes", description="Time in minutes"
                ),
                EnumOption(value="hours", label="Hours", description="Time in hours"),
            ],
        ),
        "reset_on_emit": ConfigFieldSpec(
            name="reset_on_emit",
            display_name="Reset After Emit",
            type_hint="bool",
            required=False,
            default=True,
            description="Whether to reset the internal timer after reaching the limit.",
            advance=True,
        ),
        "message": ConfigFieldSpec(
            name="message",
            display_name="Release Message",
            type_hint="text",
            required=False,
            description="Optional text sent downstream once the time limit is reached.",
            advance=True,
        ),
        "passthrough": ConfigFieldSpec(
            name="passthrough",
            display_name="Passthrough Mode",
            type_hint="bool",
            required=False,
            default=False,
            description="If true, after emitting the limit message, all subsequent inputs pass through unchanged.",
            advance=True,
        ),
    }