nodes/text_data.py

from __future__ import annotations

from dataclasses import dataclass

from dataflow.enums import NodeKind, PortDirection
from dataflow.nodes_base import NodeType, NodeInstance
from dataflow.ports import PortSchema
from nodes.data_types import TextType
from nodes.vue_nodes import VueNodeRenderable, VueNodeData

TextDataNodeType = NodeType(
    kind=NodeKind.TEXT_DATA,
    display_name="Text",
    inputs=[],
    outputs=[PortSchema(name="text", dtype=TextType, direction=PortDirection.OUTPUT)],
)


@dataclass(slots=True)
class TextDataNode(NodeInstance):
    """Node that exposes a constant text value via its output port.

    The text is stored on the "text" output port and edited through the UI.
    """

    async def process(self) -> None:
        """Data nodes do not compute anything, they just expose their current value.

        The UI writes into the "text" output port. Here we simply ensure the port
        exists and treat the current value as the result.
        """
        port = self.outputs.get("text") if self.outputs is not None else None
        if port is None:
            return
        # nothing to compute, but this method must exist so graph.execute
        # can safely walk upstream without raising
        return


class TextDataNodeRenderable(VueNodeRenderable[TextDataNode]):
    def to_vue_node_data(self, node: TextDataNode) -> VueNodeData:
        data = VueNodeData()

        data.fields.append(
            {
                "name": "text",
                "kind": "textarea",
                "label": "Text",
                "placeholder": "Enter text...",
            }
        )

        port = node.outputs.get("text") if node.outputs is not None else None
        data.values["text"] = "" if port is None or port.value is None else str(port.value)

        data.executable = False
        return data