import code
from contextlib import redirect_stdout, redirect_stderr
from ironflow.model import dtypes, NodeInputBP, NodeOutputBP
from ironflow.model.node import Node, PlaceholderWidgetsContainer
from ironflow.gui.boxes.node_interface.input_widgets import SliderControl
widgets = PlaceholderWidgetsContainer()
[docs]class NodeBase(Node):
version = "v0.1"
color = "#FFCA00"
[docs]class DualNodeBase(NodeBase):
"""For nodes that can be active and passive"""
version = "v0.1"
def __init__(self, params, active=True):
super().__init__(params)
self.active = active
if active:
self.actions["make passive"] = {"method": self.make_passive}
else:
self.actions["make active"] = {"method": self.make_active}
[docs] def make_passive(self):
del self.actions["make passive"]
self.delete_input(0)
self.delete_output(0)
self.active = False
self.actions["make active"] = {"method": self.make_active}
[docs] def make_active(self):
del self.actions["make active"]
self.create_input(type_="exec", insert=0)
self.create_output(type_="exec", insert=0)
self.active = True
self.actions["make passive"] = {"method": self.make_passive}
[docs] def get_state(self) -> dict:
return {"active": self.active}
[docs] def set_state(self, data: dict, version):
self.active = data["active"]
# -------------------------------------------
[docs]class Checkpoint_Node(NodeBase):
"""Provides a simple checkpoint to reroute your connections"""
title = "checkpoint"
version = "v0.1"
init_inputs = [
NodeInputBP(type_="data"),
]
init_outputs = [
NodeOutputBP(type_="data"),
]
style = "small"
def __init__(self, params):
super().__init__(params)
self.display_title = ""
self.active = False
# initial actions
self.actions["add output"] = {"method": self.add_output}
self.actions["remove output"] = {"0": {"method": self.remove_output, "data": 0}}
self.actions["make active"] = {"method": self.make_active}
"""State transitions"""
[docs] def clear_ports(self):
# remove all outputs
for i in range(len(self.outputs)):
self.delete_output(0)
# remove all inputs
for i in range(len(self.inputs)):
self.delete_input(0)
[docs] def make_active(self):
self.active = True
# rebuild inputs and outputs
self.clear_ports()
self.create_input(type_="exec")
self.create_output(type_="exec")
# update actions
del self.actions["make active"]
self.actions["make passive"] = {"method": self.make_passive}
self.actions["remove output"] = {"0": {"method": self.remove_output, "data": 0}}
[docs] def make_passive(self):
self.active = False
# rebuild inputs and outputs
self.clear_ports()
self.create_input(type_="data")
self.create_output(type_="data")
# update actions
del self.actions["make passive"]
self.actions["make active"] = {"method": self.make_active}
self.actions["remove output"] = {"0": {"method": self.remove_output, "data": 0}}
"""Actions"""
[docs] def add_output(self):
index = len(self.outputs)
if self.active:
self.create_output(type_="exec")
else:
self.create_output(type_="data")
self.actions["remove output"][str(index)] = {
"method": self.remove_output,
"data": index,
}
[docs] def remove_output(self, index):
self.delete_output(index)
del self.actions["remove output"][str(len(self.outputs))]
"""Behavior"""
[docs] def update_event(self, inp=-1):
if self.active and inp == 0:
for i in range(len(self.outputs)):
self.exec_output(i)
elif not self.active:
data = self.input(0)
for i in range(len(self.outputs)):
self.set_output_val(i, data)
"""State Reload"""
[docs] def get_state(self) -> dict:
return {
"active": self.active,
"num outputs": len(self.outputs),
}
[docs] def set_state(self, data: dict, version):
self.actions["remove output"] = {
{"method": self.remove_output, "data": i}
for i in range(data["num outputs"])
}
if data["active"]:
self.make_active()
[docs]class Print_Node(DualNodeBase):
title = "Print"
version = "v0.1"
init_inputs = [
NodeInputBP(type_="exec"),
NodeInputBP(dtype=dtypes.Data(size="m")),
]
init_outputs = [
NodeOutputBP(type_="exec"),
]
color = "#5d95de"
def __init__(self, params):
super().__init__(params, active=True)
[docs] def update_event(self, inp=-1):
if self.active and inp == 0:
self.val = self.input(1)
elif not self.active:
self.val = self.input(0)
import logging
[docs]class Log_Node(DualNodeBase):
title = "Log"
version = "v0.1"
init_inputs = [
NodeInputBP(type_="exec"),
NodeInputBP("msg", type_="data"),
]
init_outputs = [
NodeOutputBP(type_="exec"),
]
main_widget_class = widgets.LogNode_MainWidget
main_widget_pos = "below ports"
color = "#5d95de"
def __init__(self, params):
super().__init__(params, active=True)
self.logger = self.new_logger("Log Node")
self.targets = {
**self.script.logs_manager.default_loggers,
"own": self.logger,
}
self.target = "global"
[docs] def update_event(self, inp=-1):
if self.active and inp == 0:
i = 1
elif not self.active:
i = 0
else:
return
msg = self.input(i)
self.targets[self.target].log(logging.INFO, msg=msg)
[docs] def get_state(self) -> dict:
return {
**super().get_state(),
"target": self.target,
}
[docs] def set_state(self, data: dict, version):
super().set_state(data, version)
self.target = data["target"]
if self.session.gui and self.main_widget():
self.main_widget().set_target(self.target)
[docs]class Clock_Node(NodeBase):
title = "clock"
version = "v0.1"
init_inputs = [
NodeInputBP(dtype=dtypes.Float(default=0.1), label="delay"),
NodeInputBP(
dtype=dtypes.Integer(default=-1, bounds=(-1, 1000)), label="iterations"
),
]
init_outputs = [NodeOutputBP(type_="exec")]
color = "#5d95de"
main_widget_class = widgets.ClockNode_MainWidget
main_widget_pos = "below ports"
def __init__(self, params):
super().__init__(params)
self.actions["start"] = {"method": self.start}
self.actions["stop"] = {"method": self.stop}
if self.session.gui:
from qtpy.QtCore import QTimer
self.timer = QTimer(self)
self.timer.timeout.connect(self.timeouted)
self.iteration = 0
[docs] def timeouted(self):
self.exec_output(0)
self.iteration += 1
if -1 < self.input(1) <= self.iteration:
self.stop()
[docs] def start(self):
if self.session.gui:
self.timer.setInterval(self.input(0) * 1000)
self.timer.start()
else:
import time
for i in range(self.input(1)):
self.exec_output(0)
time.sleep(self.input(0))
[docs] def stop(self):
self.iteration = 0
if self.session.gui:
self.timer.stop()
[docs] def toggle(self):
# triggered from main widget
if self.session.gui:
if self.timer.isActive():
self.stop()
else:
self.start()
[docs] def update_event(self, inp=-1):
if self.session.gui:
self.timer.setInterval(self.input(0) * 1000)
[docs] def remove_event(self):
self.stop()
[docs]class Slider_Node(NodeBase):
title = "slider"
version = "v0.1"
init_inputs = [
NodeInputBP(dtype=dtypes.Integer(default=1), label="scl"),
NodeInputBP(dtype=dtypes.Boolean(default=False), label="round"),
]
init_outputs = [
NodeOutputBP(),
]
main_widget_class = widgets.SliderNode_MainWidget
main_widget_pos = "below ports"
input_widget = SliderControl
def __init__(self, params):
super().__init__(params)
self.val = 0
[docs] def place_event(self):
self.update()
[docs] def view_place_event(self):
# when running in gui mode, the value might come from the input widget
self.update()
[docs] def update_event(self, inp=-1):
v = self.input(0) * self.val
if self.input(1):
v = round(v)
self.set_output_val(0, v)
[docs] def get_state(self) -> dict:
return {
"val": self.val,
}
[docs] def set_state(self, data: dict, version):
self.val = data["val"]
class _DynamicPorts_Node(NodeBase):
version = "v0.1"
init_inputs = []
init_outputs = []
def __init__(self, params):
super().__init__(params)
self.actions["add input"] = {"method": self.add_inp}
self.actions["add output"] = {"method": self.add_out}
self.num_inputs = 0
self.num_outputs = 0
def add_inp(self):
self.create_input()
index = self.num_inputs
self.actions[f"remove input {index}"] = {
"method": self.remove_inp,
"data": index,
}
self.num_inputs += 1
def remove_inp(self, index):
self.delete_input(index)
self.num_inputs -= 1
del self.actions[f"remove input {self.num_inputs}"]
def add_out(self):
self.create_output()
index = self.num_outputs
self.actions[f"remove output {index}"] = {
"method": self.remove_out,
"data": index,
}
self.num_outputs += 1
def remove_out(self, index):
self.delete_output(index)
self.num_outputs -= 1
del self.actions[f"remove output {self.num_outputs}"]
def get_state(self) -> dict:
return {
"num inputs": self.num_inputs,
"num outputs": self.num_outputs,
}
def set_state(self, data: dict):
self.num_inputs = data["num inputs"]
self.num_outputs = data["num outputs"]
[docs]class Exec_Node(_DynamicPorts_Node):
title = "exec"
version = "v0.1"
main_widget_class = widgets.CodeNode_MainWidget
main_widget_pos = "between ports"
def __init__(self, params):
super().__init__(params)
self.code = None
[docs] def place_event(self):
pass
[docs] def update_event(self, inp=-1):
exec(self.code)
[docs] def get_state(self) -> dict:
return {
**super().get_state(),
"code": self.code,
}
[docs] def set_state(self, data: dict, version):
super().set_state(data, version)
self.code = data["code"]
[docs]class Eval_Node(NodeBase):
title = "eval"
version = "v0.1"
init_inputs = [
# NodeInputBP(),
]
init_outputs = [
NodeOutputBP(),
]
main_widget_class = widgets.EvalNode_MainWidget
main_widget_pos = "between ports"
def __init__(self, params):
super().__init__(params)
self.actions["add input"] = {"method": self.add_param_input}
self.number_param_inputs = 0
self.expression_code = None
[docs] def place_event(self):
if self.number_param_inputs == 0:
self.add_param_input()
[docs] def update_event(self, inp=-1):
inp = [self.input(i) for i in range(self.number_param_inputs)]
self.set_output_val(0, eval(self.expression_code))
[docs] def get_state(self) -> dict:
return {
"num param inputs": self.number_param_inputs,
"expression code": self.expression_code,
}
[docs] def set_state(self, data: dict, version):
self.number_param_inputs = data["num param inputs"]
self.expression_code = data["expression code"]
[docs]class Interpreter_Node(NodeBase):
"""Provides a python interpreter via a basic console with access to the
node's properties."""
title = "interpreter"
version = "v0.1"
init_inputs = []
init_outputs = []
main_widget_class = widgets.InterpreterConsole
# DEFAULT COMMANDS
[docs] def clear(self):
self.hist.clear()
self._hist_updated()
[docs] def reset(self):
self.interp = code.InteractiveInterpreter(locals=locals())
COMMANDS = {
"clear": clear,
"reset": reset,
}
def __init__(self, params):
super().__init__(params)
self.interp = None
self.hist: [str] = []
self.buffer: [str] = []
self.reset()
def _hist_updated(self):
if self.session.gui:
self.main_widget().interp_updated()
[docs] def write(self, line: str):
self.hist.append(line)
self._hist_updated()
[docs]class Storage_Node(NodeBase):
"""Sequentially stores all the data provided at the input in an array.
A COPY of the storage array is provided at the output"""
title = "store"
version = "v0.1"
init_inputs = [
NodeInputBP(),
]
init_outputs = [
NodeOutputBP(),
]
color = "#aadd55"
def __init__(self, params):
super().__init__(params)
self.storage = []
self.actions["clear"] = {"method": self.clear}
[docs] def clear(self):
self.storage.clear()
self.set_output_val(0, [])
[docs] def update_event(self, inp=-1):
self.storage.append(self.input(0))
self.set_output_val(0, self.storage.copy())
[docs] def get_state(self) -> dict:
return {
"data": self.storage,
}
[docs] def set_state(self, data: dict, version):
self.storage = data["data"]
import uuid
[docs]class LinkIN_Node(NodeBase):
"""You can use link OUT nodes to link them up to this node.
Whenever a link IN node receives data (or an execution signal),
if there is a linked OUT node, it will receive the data
and propagate it further."""
title = "link IN"
version = "v0.1"
init_inputs = [
NodeInputBP(),
]
init_outputs = [] # no outputs
# instances registration
INSTANCES = {} # {UUID: node}
def __init__(self, params):
super().__init__(params)
self.display_title = "link"
# register
self.ID: uuid.UUID = uuid.uuid4()
self.INSTANCES[str(self.ID)] = self
self.actions["add input"] = {"method": self.add_inp}
self.actions["remove inp"] = {}
self.actions["copy ID"] = {"method": self.copy_ID}
self.linked_node: LinkOUT_Node = None
[docs] def copy_ID(self):
from qtpy.QtWidgets import QApplication
QApplication.clipboard().setText(str(self.ID))
[docs] def add_inp(self):
index = len(self.inputs)
self.create_input()
self.actions["remove inp"][str(index)] = {
"method": self.rem_inp,
"data": index,
}
if self.linked_node is not None:
self.linked_node.add_out()
[docs] def rem_inp(self, index):
self.delete_input(index)
del self.actions["remove inp"][str(len(self.inputs))]
if self.linked_node is not None:
self.linked_node.rem_out(index)
[docs] def update_event(self, inp=-1):
if self.linked_node is not None:
self.linked_node.set_output_val(inp, self.input(inp))
[docs] def get_state(self) -> dict:
return {
"ID": str(self.ID),
}
[docs] def set_state(self, data: dict, version):
if data["ID"] in self.INSTANCES:
# this happens when some existing node has been copied and pasted.
# we only want to rebuild links when loading a project, considering
# new links when copying nodes might get quite complex
pass
else:
del self.INSTANCES[str(self.ID)] # remove old ref
self.ID = uuid.UUID(data["ID"]) # use original ID
self.INSTANCES[str(self.ID)] = self # set new ref
# resolve possible pending link builds from OUT nodes that happened
# to get initialized earlier
LinkOUT_Node.new_link_in_loaded(self)
[docs] def remove_event(self):
# break existent link
if self.linked_node:
self.linked_node.linked_node = None
self.linked_node = None
[docs]class LinkOUT_Node(NodeBase):
"""The complement to the link IN node"""
title = "link OUT"
version = "v0.1"
init_inputs = [] # no inputs
init_outputs = [] # will be synchronized with linked IN node
INSTANCES = []
PENDING_LINK_BUILDS = {}
# because a link OUT node might get initialized BEFORE it's corresponding
# link IN, it then stores itself together with the ID of the link IN it's
# waiting for in PENDING_LINK_BUILDS
[docs] @classmethod
def new_link_in_loaded(cls, n: LinkIN_Node):
for out_node, in_ID in cls.PENDING_LINK_BUILDS.items():
if in_ID == str(n.ID):
out_node.link_to(n)
def __init__(self, params):
super().__init__(params)
self.display_title = "link"
self.INSTANCES.append(self)
self.linked_node: LinkIN_Node = None
self.actions["link to ID"] = {"method": self.choose_link_ID}
[docs] def choose_link_ID(self):
"""opens a small input dialog for providing a copied link IN ID"""
from qtpy.QtWidgets import QDialog, QMessageBox, QVBoxLayout, QLineEdit
class IDInpDialog(QDialog):
def __init__(self):
super().__init__()
self.id_str = None
self.setLayout(QVBoxLayout())
self.line_edit = QLineEdit()
self.layout().addWidget(self.line_edit)
self.line_edit.returnPressed.connect(self.return_pressed)
def return_pressed(self):
self.id_str = self.line_edit.text()
self.accept()
d = IDInpDialog()
d.exec_()
if d.id_str is not None:
n = LinkIN_Node.INSTANCES.get(d.id_str)
if n is None:
QMessageBox.warning(
title="link failed", text="couldn't find a valid link in node"
)
else:
self.link_to(n)
[docs] def link_to(self, n: LinkIN_Node):
self.linked_node = n
n.linked_node = self
o = len(self.outputs)
i = len(self.linked_node.inputs)
# remove outputs if there are too many
for j in range(i, o):
self.delete_output(0)
# add outputs if there are too few
for j in range(o, i):
self.create_output()
self.update()
[docs] def add_out(self):
# triggered by linked_node
self.create_output()
[docs] def rem_out(self, index):
# triggered by linked_node
self.delete_output(index)
[docs] def update_event(self, inp=-1):
if self.linked_node is None:
return
# update ALL ports
for i in range(len(self.outputs)):
self.set_output_val(i, self.linked_node.input(i))
[docs] def get_state(self) -> dict:
if self.linked_node is None:
return {}
else:
return {
"linked ID": str(self.linked_node.ID),
}
[docs] def set_state(self, data: dict, version):
if len(data) > 0:
n: LinkIN_Node = LinkIN_Node.INSTANCES.get(data["linked ID"])
if n is None:
# means that the OUT node gets initialized before it's link IN
self.PENDING_LINK_BUILDS[self] = data["linked ID"]
elif n.linked_node is None:
# pair up
n.linked_node = self
self.linked_node = n
[docs] def remove_event(self):
# break existent link
if self.linked_node:
self.linked_node.linked_node = None
self.linked_node = None
# -------------------------------------------
nodes = [
Checkpoint_Node,
Button_Node,
Print_Node,
Log_Node,
Clock_Node,
Slider_Node,
Exec_Node,
Eval_Node,
Storage_Node,
LinkIN_Node,
LinkOUT_Node,
Interpreter_Node,
]