File size: 1,851 Bytes
d868fac 691f45a d868fac 691f45a d868fac 691f45a d868fac 691f45a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | 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
|