| 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 |
| |
| |
| 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 |
|
|