Here is an example of such a component, just to give you a headstart (just use it as inspiration, don't reuse it!):
```vue_flow.py
# vueflow_canvas.py
from __future__ import annotations
import asyncio
from typing import Any, Dict, Iterable, List, Optional, Union
from nicegui.element import Element
class VueFlowCanvas(Element, component="vueflow_canvas.vue"):
"""
NiceGUI wrapper for Vue Flow.
Features:
- set graph from Python
- add, update, remove nodes and edges
- request current graph from Python
- get events in Python
- define simple custom nodes using HTML content
- register external Vue components by global name for custom nodes
"""
def __init__(
self,
*,
height: int = 600,
show_controls: bool = True,
fit_on_init: bool = True,
snap_to_grid: bool = True,
connection_mode: str = "loose", # "loose" or "strict"
allow_user_edit: bool = True,
initial_nodes: Optional[Iterable[Dict[str, Any]]] = None,
initial_edges: Optional[Iterable[Dict[str, Any]]] = None,
core_version: str = "latest",
extras_version: str = "latest",
) -> None:
super().__init__()
self._props.update(
height=height,
showControls=show_controls,
fitOnInit=fit_on_init,
snapToGrid=snap_to_grid,
connectionMode=connection_mode,
allowUserEdit=allow_user_edit,
initialNodes=list(initial_nodes or []),
initialEdges=list(initial_edges or []),
coreVersion=core_version,
extrasVersion=extras_version,
)
# graph bulk ops
def set_graph(self, nodes: Iterable[Dict[str, Any]], edges: Iterable[Dict[str, Any]]) -> None:
self.run_method("setGraph", list(nodes), list(edges))
def clear(self) -> None:
self.run_method("clearGraph")
# node ops
def add_node(self, node: Dict[str, Any]) -> None:
self.run_method("addNode", dict(node))
def update_node(self, node_id: str, patch: Dict[str, Any]) -> None:
self.run_method("updateNode", str(node_id), dict(patch))
def remove_node(self, node_id: str) -> None:
self.run_method("removeNode", str(node_id))
# edge ops
def add_edge(self, edge: Dict[str, Any]) -> None:
self.run_method("addEdge", dict(edge))
def remove_edge(self, edge_id: str) -> None:
self.run_method("removeEdge", str(edge_id))
# view ops
def fit_view(self) -> None:
self.run_method("fitView")
def zoom_in(self) -> None:
self.run_method("zoomIn")
def zoom_out(self) -> None:
self.run_method("zoomOut")
# import and export
def import_graph(self, jsonish: Union[str, Dict[str, Any]]) -> None:
self.run_method("importGraph", jsonish)
def export_graph(self) -> None:
"""Triggers a 'graph' event you can listen to with .on('graph', handler)."""
self.run_method("exportGraph")
def request_graph(self) -> None:
"""Same as export_graph but named for request semantics."""
self.run_method("requestGraph")
async def get_graph(self) -> Dict[str, List[Dict[str, Any]]]:
"""
Awaitable helper that returns the current graph as a dict with 'nodes' and 'edges'.
"""
loop = asyncio.get_running_loop()
fut: asyncio.Future = loop.create_future()
def once(e):
if not fut.done():
fut.set_result(e.args if isinstance(e.args, dict) else e.args[0] if e.args else {"nodes": [], "edges": []})
self.off("graph", once)
self.on("graph", once)
self.run_method("requestGraph")
result = await fut
if not isinstance(result, dict):
return {"nodes": [], "edges": []}
return result
# custom node helpers
def register_html_node_type(self, name: str) -> None:
"""
Registers a node type that renders node.data["html"] inside the node.
Use it by creating nodes with type=name and data={"html": "Hi"}.
"""
self.run_method("registerHtmlNodeType", str(name))
def register_external_node_type(self, name: str, global_var: str) -> None:
"""
Registers a node type backed by a Vue component available at window[global_var].
This is useful if you add a script that defines window.MyCoolNode = {...}.
"""
self.run_method("registerExternalNodeType", str(name), str(global_var))
```
```vue_flow.vue
nodes: {{ statNodes }} edges: {{ statEdges }}
```
We would like to use https://vueflow.dev/ inside our nicegui based application to create node based environments. For that, we need a nicegui component that allows us to interact with vueflow. The general idea is the following:
- We need a vue-flow-canvas that is displayed on the website (by default full width and full height by default), the component should also reflect the nicegui style (bright or dark mode), and feel like a native nicegui component (follow their rules)
- The events of the graph view canvas should be sent back to python to be able to interact with it (for example when nodes are moved, connections are made, nodes are added and so on)
- Also, it should be possible to programmatically create nodes and connections
- It should be possible to define new nodes types as new nicegui (vue) components. there should already be a base node which also supports events, connections and so on. It should be possible to define connection data types (text, image and so on) (or multi / single connectables) for a node (inputs and outputs) and of course the gui (as html). if possible, it would be also great that such a type can also include sub-elements from nicegui (like a textbox or and image maybe later, just as example). Such a node-type has a "process" event which is called when the graph is executing this node. Please merge my description of the requirements with the suppport vueflow or nicegui already provides
Please use python 3.12 standard to programm it, make a good oop architecture (one file per class), use dataclasses, typehints and also docstrings (:param: style). Write good code and reusable / extendable code. Always use enums instead of strings if possible (types and so on).
As an example how it should work, I'll add you some pseudo code how it could look like:
```python
@dataclass
class DataType(Generics[T]):
name: str
type: Type[T]
@dataclass
class DataPort(ABC, Generics[DT]):
name: str
type: Type[DT]
value: DT.type | None = None
@dataclass
class InputDataPort(DataPort[DT])
accepts: int = 10 # (how many inputs are allowed) maybe rename this
@dataclass
class OutputDataPort(DataPort[DT])
state: DataPortState.Dirty
TextDataType = DataType("Text", str)
ImageDataType = DataType("Image", np.ndarray) # uses base64 string to exchange image data
@dataclass
class NodeBase(Generics)
name: str
auto_process: bool = False
@abstractmethod
def process()
pass
# post_init: create a list of all inputs and one of all outputs by analyzing the fields and check InputDataPort / OutputDataPort types
@dataclass
class TextDataNode(NodeBase):
text: OutputDataPort("Text", TextDataType)
@dataclass
class TextToImageNode(NodeBase):
text: InputDataPort("Prompt", TextDataType, 5)
image: OuputDataPort("Image", ImageDataType)
@abstractmethod
def process():
print(f"mocking generation process with {text.value}")
image.value = np.zeros((100, 100))
@dataclass
class Connection:
start: OutputDataPort
end: InputDataPort
class DataGraph:
nodes: list[NodeBase] = []
connections: list[Connection] = []
def execute(node: NodeBase | None):
# go through all node inputs, check which are connected and maybe dirty, recurse upwards the graph
# re-process this node
# send the data downstream to all nodes which auto_process = true
# maybe optomize how this work. it would be good to be able to set if a processing node should automatically run or not
pass
# ui relevant
@dataclass
class NodeElement(Generics[NodeBase])
pass
class TextDataNodeElement(NodeElement[TextDataNode], component="text_data_node_element.vue"):
# render the text data node elemnt
pass
class TextToImageNodeElement(NodeElement[TextToImageNode], component="text_to_image_node_element.vue"):
# render the text-to-image node elemnt
pass
```
```python
graph = DataGraph()
# example gui
@ui.page('/')
def page():
canvas = DataFlowCanvas(graph, elements=[TextDataNodeElement, TextToImageNodeElement])
canvas.on_new_connection = ...
ui.run()
```