File size: 2,055 Bytes
691f45a 3025bb3 691f45a 3025bb3 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 59 60 61 62 63 64 65 | 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 ImageType
from nodes.vue_nodes import VueNodeRenderable, VueNodeData
ImageDataNodeType = NodeType(
kind=NodeKind.IMAGE_DATA,
display_name="Image",
inputs=[],
outputs=[PortSchema(name="image", dtype=ImageType, direction=PortDirection.OUTPUT)],
)
@dataclass(slots=True)
class ImageDataNode(NodeInstance):
"""Node that exposes a constant image 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 "image" output port. Here we simply ensure the port
exists and treat the current value as the result.
"""
port = self.outputs.get("image") 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 ImageDataNodeRenderable(VueNodeRenderable[ImageDataNode]):
def to_vue_node_data(self, node: ImageDataNode) -> VueNodeData:
data = VueNodeData()
data.fields.append(
{
"name": "image",
"kind": "image_upload",
"label": "Image",
"placeholder": "Upload or paste image...",
}
)
port = node.outputs.get("image") if node.outputs is not None else None
val = ""
if port is not None and port.value is not None:
if port.schema.dtype.encode:
val = port.schema.dtype.encode(port.value)
else:
val = str(port.value)
data.values["image"] = val
data.executable = False
return data
|