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