537 lines
20 KiB
Python
537 lines
20 KiB
Python
from __future__ import annotations
|
|
from typing import Any, Callable
|
|
|
|
LIST_CAPABLE_SOCKET_TYPES = {
|
|
"SLICESOCK_Mesh",
|
|
"SLICESOCK_Paths",
|
|
"SLICESOCK_Object",
|
|
"SLICESOCK_Geometry",
|
|
"NodeSocketObject",
|
|
"NodeSocketString",
|
|
}
|
|
|
|
def is_list_data(val: Any) -> bool:
|
|
"""Checks if val represents a list of items (Object, Mesh, Path, Gcode).
|
|
Note: SlicePaths is a list subclass representing an individual toolpath collection,
|
|
not a list of toolpaths, so it is treated as an atomic item (not list data)."""
|
|
if val is None or isinstance(val, (str, bytes)):
|
|
return False
|
|
if type(val).__name__ == "SlicePaths" or getattr(val, "is_slice_paths", False):
|
|
return False
|
|
return isinstance(val, (list, tuple))
|
|
|
|
def flatten_list(items: Any) -> list[Any]:
|
|
"""Recursively flattens nested lists/tuples, keeping atomic items intact."""
|
|
result: list[Any] = []
|
|
if not is_list_data(items):
|
|
if items is not None:
|
|
result.append(items)
|
|
return result
|
|
for item in items:
|
|
if is_list_data(item):
|
|
result.extend(flatten_list(item))
|
|
elif item is not None:
|
|
result.append(item)
|
|
return result
|
|
|
|
def eval_element_wise(op_func: Callable[[Any], Any], data: Any) -> Any:
|
|
"""If data is a list, applies op_func to each element and returns a list.
|
|
Otherwise applies op_func to data directly."""
|
|
if is_list_data(data):
|
|
return [op_func(item) for item in data]
|
|
return op_func(data)
|
|
|
|
def set_socket_shape(sock: Any, shape: str) -> None:
|
|
"""Safely updates socket display_shape property ('LIST', 'CIRCLE', etc.)."""
|
|
if sock is not None:
|
|
try:
|
|
sock.display_shape = shape
|
|
except (AttributeError, TypeError):
|
|
pass
|
|
|
|
def _follow_reroute_upstream(sock: Any, visited: set | None = None) -> Any:
|
|
"""Follows reroute nodes backwards to find the original source output socket."""
|
|
if visited is None:
|
|
visited = set()
|
|
current = sock
|
|
while (
|
|
current is not None
|
|
and getattr(current, "node", None) is not None
|
|
and getattr(current.node, "bl_idname", "") == "NodeReroute"
|
|
and getattr(current.node, "inputs", None)
|
|
and current.node.inputs[0].is_linked
|
|
and current.node.inputs[0].links
|
|
):
|
|
ptr = current.as_pointer() if hasattr(current, "as_pointer") else id(current)
|
|
if ptr in visited:
|
|
break
|
|
visited.add(ptr)
|
|
current = current.node.inputs[0].links[0].from_socket
|
|
return current
|
|
|
|
def _follow_reroute_downstream(reroute_node: Any, visited: set | None = None) -> list[Any]:
|
|
"""Follows reroute nodes forwards to find all downstream target input sockets."""
|
|
if visited is None:
|
|
visited = set()
|
|
node_ptr = reroute_node.as_pointer() if hasattr(reroute_node, "as_pointer") else id(reroute_node)
|
|
if node_ptr in visited:
|
|
return []
|
|
visited.add(node_ptr)
|
|
|
|
targets: list[Any] = []
|
|
for outp in getattr(reroute_node, "outputs", []):
|
|
for link in getattr(outp, "links", []):
|
|
to_sock = link.to_socket
|
|
to_node = getattr(to_sock, "node", None)
|
|
if to_node and getattr(to_node, "bl_idname", "") == "NodeReroute":
|
|
targets.extend(_follow_reroute_downstream(to_node, visited))
|
|
else:
|
|
targets.append(to_sock)
|
|
return targets
|
|
|
|
def socket_accepts_list(sock: Any, visited: set | None = None) -> bool:
|
|
"""Determines whether a given input socket can accept a list input."""
|
|
if sock is None:
|
|
return False
|
|
if visited is None:
|
|
visited = set()
|
|
sock_ptr = sock.as_pointer() if hasattr(sock, "as_pointer") else id(sock)
|
|
if sock_ptr in visited:
|
|
return False
|
|
visited.add(sock_ptr)
|
|
|
|
node = getattr(sock, "node", None)
|
|
if node is None:
|
|
return False
|
|
|
|
bl_idname = getattr(node, "bl_idname", "")
|
|
|
|
# Strictly forbidden nodes
|
|
if bl_idname in ("SLICENODE_StartPrint", "SLICENODE_PrinterList"):
|
|
return False
|
|
|
|
# Path Operations: only input 0 can accept a list
|
|
if bl_idname == "SLICENODE_PathOperations":
|
|
inputs = getattr(node, "inputs", None)
|
|
if inputs and sock != inputs[0]:
|
|
return False
|
|
return True
|
|
|
|
# Seam Placement: "Mesh" input rejects lists; "Paths" accepts lists
|
|
if bl_idname == "SLICENODE_SeamPlacement":
|
|
if getattr(sock, "name", "") == "Mesh":
|
|
return False
|
|
return True
|
|
|
|
# Socket type check
|
|
sock_type = getattr(sock, "bl_idname", "")
|
|
if sock_type not in LIST_CAPABLE_SOCKET_TYPES:
|
|
return False
|
|
|
|
# String sockets: only Gcode accepts lists
|
|
if sock_type == "NodeSocketString":
|
|
sock_name = getattr(sock, "name", "")
|
|
if bl_idname == "SLICENODE_JoinGcode":
|
|
return True
|
|
if bl_idname in ("SLICENODE_WriteGcode", "SLICENODE_ExecuteGcode"):
|
|
return sock_name == "Gcode"
|
|
if bl_idname == "SLICENODE_GcodeRegexReplace":
|
|
return sock_name == "Gcode"
|
|
if bl_idname == "SLICENODE_GcodesToList":
|
|
return True
|
|
if bl_idname == "NodeGroupOutput":
|
|
return sock_name.lower().startswith("gcode")
|
|
if bl_idname == "SLICENODE_Group":
|
|
inner_tree = getattr(node, "slicer_tree", None) or getattr(node, "node_tree", None)
|
|
if inner_tree:
|
|
return group_input_can_accept_list(inner_tree, sock, visited)
|
|
return False
|
|
return False
|
|
|
|
# Reroute node: check downstream targets
|
|
if bl_idname == "NodeReroute":
|
|
targets = _follow_reroute_downstream(node, visited)
|
|
if not targets:
|
|
return False
|
|
return all(socket_accepts_list(t, visited) for t in targets)
|
|
|
|
# NodeGroupOutput
|
|
if bl_idname == "NodeGroupOutput":
|
|
return True
|
|
|
|
# Nested SLICENODE_Group
|
|
if bl_idname == "SLICENODE_Group":
|
|
inner_tree = getattr(node, "slicer_tree", None) or getattr(node, "node_tree", None)
|
|
if inner_tree:
|
|
return group_input_can_accept_list(inner_tree, sock, visited)
|
|
return False
|
|
|
|
# List joiners accept lists
|
|
if getattr(node, "is_list_output", False):
|
|
return True
|
|
|
|
# Data joiners accept lists
|
|
if bl_idname in ("SLICENODE_JoinMesh", "SLICENODE_JoinMeshGroups", "SLICENODE_JoinGcode"):
|
|
return True
|
|
|
|
# Nodes with list_socket_mapping
|
|
mapping = getattr(node, "list_socket_mapping", None)
|
|
if mapping:
|
|
mapped_in_names = {item[0] for item in mapping}
|
|
if getattr(sock, "name", "") in mapped_in_names:
|
|
return True
|
|
|
|
return False
|
|
|
|
def _get_interface_item(tree: Any, sock_or_idx_or_ident: Any, in_out: str) -> Any:
|
|
attr_name = "inputs" if in_out == 'INPUT' else "outputs"
|
|
if not hasattr(tree, "interface"):
|
|
sockets = getattr(tree, attr_name, [])
|
|
if isinstance(sock_or_idx_or_ident, int) and 0 <= sock_or_idx_or_ident < len(sockets):
|
|
return sockets[sock_or_idx_or_ident]
|
|
ident = getattr(sock_or_idx_or_ident, "identifier", str(sock_or_idx_or_ident))
|
|
for s in sockets:
|
|
if s.identifier == ident or s.name == ident:
|
|
return s
|
|
return None
|
|
|
|
items = [
|
|
item for item in getattr(tree.interface, "items_tree", [])
|
|
if getattr(item, "item_type", "") == 'SOCKET' and getattr(item, "in_out", "") == in_out
|
|
]
|
|
if isinstance(sock_or_idx_or_ident, int):
|
|
return items[sock_or_idx_or_ident] if 0 <= sock_or_idx_or_ident < len(items) else None
|
|
|
|
ident = getattr(sock_or_idx_or_ident, "identifier", None)
|
|
name = getattr(sock_or_idx_or_ident, "name", None)
|
|
if isinstance(sock_or_idx_or_ident, str):
|
|
ident = name = sock_or_idx_or_ident
|
|
|
|
for item in items:
|
|
if ident and getattr(item, "identifier", None) == ident:
|
|
return item
|
|
for item in items:
|
|
if name and getattr(item, "name", None) == name:
|
|
return item
|
|
return None
|
|
|
|
def _get_input_interface_item(tree: Any, sock_or_idx_or_ident: Any) -> Any:
|
|
return _get_interface_item(tree, sock_or_idx_or_ident, 'INPUT')
|
|
|
|
def _get_output_interface_item(tree: Any, sock_or_idx_or_ident: Any) -> Any:
|
|
return _get_interface_item(tree, sock_or_idx_or_ident, 'OUTPUT')
|
|
|
|
def _find_group_input_socket(tree: Any, item: Any) -> Any:
|
|
if not item or not hasattr(tree, "nodes"):
|
|
return None
|
|
ident = getattr(item, "identifier", "")
|
|
name = getattr(item, "name", "")
|
|
for n in tree.nodes:
|
|
if getattr(n, "bl_idname", "") == "NodeGroupInput":
|
|
for outp in getattr(n, "outputs", []):
|
|
if outp.identifier == ident or outp.name == name:
|
|
return outp
|
|
return None
|
|
|
|
def _find_group_output_socket(tree: Any, item: Any) -> Any:
|
|
if not item or not hasattr(tree, "nodes"):
|
|
return None
|
|
ident = getattr(item, "identifier", "")
|
|
name = getattr(item, "name", "")
|
|
for n in tree.nodes:
|
|
if getattr(n, "bl_idname", "") == "NodeGroupOutput":
|
|
for inp in getattr(n, "inputs", []):
|
|
if inp.identifier == ident or inp.name == name:
|
|
return inp
|
|
return None
|
|
|
|
def group_input_can_accept_list(tree: Any, sock_or_idx_or_ident: Any, visited: set | None = None) -> bool:
|
|
"""Determines whether a group tree input socket can accept a list input."""
|
|
if not tree:
|
|
return False
|
|
if visited is None:
|
|
visited = set()
|
|
|
|
item = _get_input_interface_item(tree, sock_or_idx_or_ident)
|
|
if not item:
|
|
return False
|
|
|
|
item_key = (tree.as_pointer() if hasattr(tree, "as_pointer") else id(tree), getattr(item, "identifier", id(item)))
|
|
if item_key in visited:
|
|
return False
|
|
visited.add(item_key)
|
|
|
|
stype = getattr(item, "socket_type", "") or getattr(item, "bl_idname", "")
|
|
if stype not in LIST_CAPABLE_SOCKET_TYPES:
|
|
return False
|
|
|
|
gin_sock = _find_group_input_socket(tree, item)
|
|
if not gin_sock or not getattr(gin_sock, "is_linked", False) or not gin_sock.links:
|
|
return False
|
|
|
|
targets: list[Any] = []
|
|
for link in gin_sock.links:
|
|
to_sock = link.to_socket
|
|
to_node = getattr(to_sock, "node", None)
|
|
if to_node and getattr(to_node, "bl_idname", "") == "NodeReroute":
|
|
targets.extend(_follow_reroute_downstream(to_node, visited))
|
|
else:
|
|
targets.append(to_sock)
|
|
|
|
if not targets:
|
|
return False
|
|
|
|
return all(socket_accepts_list(t, visited) for t in targets)
|
|
|
|
def is_internal_socket_a_list(group_node: Any, tree: Any, in_sock: Any, visited: set | None = None) -> bool:
|
|
"""Determines whether an internal input socket receives a list within a group context."""
|
|
if not in_sock or not getattr(in_sock, "is_linked", False) or not in_sock.links:
|
|
return False
|
|
from_sock = in_sock.links[0].from_socket
|
|
from_sock = _follow_reroute_upstream(from_sock, visited)
|
|
if not from_sock:
|
|
return False
|
|
from_node = getattr(from_sock, "node", None)
|
|
if not from_node:
|
|
return False
|
|
|
|
if getattr(from_node, "is_list_output", False):
|
|
return True
|
|
|
|
if getattr(from_node, "bl_idname", "") == "NodeGroupInput":
|
|
in_item = _get_input_interface_item(tree, from_sock)
|
|
if not in_item:
|
|
return False
|
|
if group_node is not None:
|
|
for grp_in in getattr(group_node, "inputs", []):
|
|
if grp_in.identifier == getattr(in_item, "identifier", None) or grp_in.name == getattr(in_item, "name", None):
|
|
return is_socket_linked_to_list(grp_in)
|
|
return False
|
|
else:
|
|
return group_input_can_accept_list(tree, in_item, visited)
|
|
|
|
if getattr(from_node, "bl_idname", "") == "SLICENODE_Group":
|
|
inner = getattr(from_node, "slicer_tree", None) or getattr(from_node, "node_tree", None)
|
|
if inner:
|
|
return is_group_output_a_list(from_node, inner, from_sock, visited)
|
|
return False
|
|
|
|
if getattr(from_node, "bl_idname", "") == "SLICENODE_JoinGcode":
|
|
for inp in getattr(from_node, "inputs", []):
|
|
if getattr(inp, "name", "") == "Gcode" or getattr(inp, "identifier", "").startswith("Gcode_ms_in_"):
|
|
if is_internal_socket_a_list(group_node, tree, inp, visited):
|
|
return True
|
|
return False
|
|
|
|
mapping = getattr(from_node, "list_socket_mapping", None)
|
|
if mapping:
|
|
sock_name = getattr(from_sock, "name", "")
|
|
for m_in_name, m_out_names in mapping:
|
|
targets = [m_out_names] if isinstance(m_out_names, str) else list(m_out_names)
|
|
if sock_name in targets:
|
|
mapped_inp = getattr(from_node, "inputs", None) and from_node.inputs.get(m_in_name)
|
|
if mapped_inp:
|
|
return is_internal_socket_a_list(group_node, tree, mapped_inp, visited)
|
|
|
|
return getattr(from_sock, "display_shape", "") == "LIST"
|
|
|
|
def is_group_output_a_list(
|
|
group_node: Any,
|
|
tree: Any,
|
|
sock_or_idx_or_ident: Any,
|
|
visited: set | None = None
|
|
) -> bool:
|
|
"""Determines whether a group tree output socket produces a list."""
|
|
if not tree:
|
|
return False
|
|
if visited is None:
|
|
visited = set()
|
|
|
|
item = _get_output_interface_item(tree, sock_or_idx_or_ident)
|
|
if not item:
|
|
return False
|
|
|
|
item_key = (
|
|
tree.as_pointer() if hasattr(tree, "as_pointer") else id(tree),
|
|
getattr(item, "identifier", id(item)),
|
|
id(group_node) if group_node else 0,
|
|
)
|
|
if item_key in visited:
|
|
return False
|
|
visited.add(item_key)
|
|
|
|
gout_sock = _find_group_output_socket(tree, item)
|
|
if not gout_sock or not getattr(gout_sock, "is_linked", False) or not gout_sock.links:
|
|
return False
|
|
|
|
src_sock = gout_sock.links[0].from_socket
|
|
src_sock = _follow_reroute_upstream(src_sock, visited)
|
|
if not src_sock:
|
|
return False
|
|
|
|
src_node = getattr(src_sock, "node", None)
|
|
if not src_node:
|
|
return False
|
|
|
|
src_bl_id = getattr(src_node, "bl_idname", "")
|
|
|
|
if getattr(src_node, "is_list_output", False):
|
|
return True
|
|
|
|
if src_bl_id == "SLICENODE_JoinGcode":
|
|
for inp in getattr(src_node, "inputs", []):
|
|
if getattr(inp, "name", "") == "Gcode" or getattr(inp, "identifier", "").startswith("Gcode_ms_in_"):
|
|
if is_internal_socket_a_list(group_node, tree, inp, visited):
|
|
return True
|
|
return False
|
|
|
|
if src_bl_id == "SLICENODE_Group":
|
|
inner_tree = getattr(src_node, "slicer_tree", None) or getattr(src_node, "node_tree", None)
|
|
if inner_tree:
|
|
return is_group_output_a_list(src_node, inner_tree, src_sock, visited)
|
|
return False
|
|
|
|
if src_bl_id == "NodeGroupInput":
|
|
in_item = _get_input_interface_item(tree, src_sock)
|
|
if not in_item:
|
|
return False
|
|
if group_node is not None:
|
|
for grp_in in getattr(group_node, "inputs", []):
|
|
if grp_in.identifier == getattr(in_item, "identifier", None) or grp_in.name == getattr(in_item, "name", None):
|
|
return is_socket_linked_to_list(grp_in)
|
|
return False
|
|
else:
|
|
return group_input_can_accept_list(tree, in_item, visited)
|
|
|
|
mapping = getattr(src_node, "list_socket_mapping", None)
|
|
if mapping:
|
|
sock_name = getattr(src_sock, "name", "")
|
|
for in_name, out_names in mapping:
|
|
targets = [out_names] if isinstance(out_names, str) else list(out_names)
|
|
if sock_name in targets:
|
|
inp = getattr(src_node, "inputs", None) and src_node.inputs.get(in_name)
|
|
if inp:
|
|
return is_internal_socket_a_list(group_node, tree, inp, visited)
|
|
|
|
return getattr(src_sock, "display_shape", "") == "LIST"
|
|
|
|
def sync_group_node_socket_shapes(group_node: Any) -> None:
|
|
"""Updates display_shape property on a SLICENODE_Group node's inputs and outputs."""
|
|
tree = getattr(group_node, "slicer_tree", None) or getattr(group_node, "node_tree", None)
|
|
if not tree:
|
|
return
|
|
for inp in getattr(group_node, "inputs", []):
|
|
can_list = group_input_can_accept_list(tree, inp)
|
|
set_socket_shape(inp, "LIST" if can_list else "CIRCLE")
|
|
for outp in getattr(group_node, "outputs", []):
|
|
is_list = is_group_output_a_list(group_node, tree, outp)
|
|
set_socket_shape(outp, "LIST" if is_list else "CIRCLE")
|
|
|
|
def sync_group_tree_internal_sockets(tree: Any) -> None:
|
|
"""Updates display_shape property on NodeGroupInput and NodeGroupOutput within a group tree."""
|
|
if not tree or not hasattr(tree, "nodes"):
|
|
return
|
|
for node in tree.nodes:
|
|
bl_id = getattr(node, "bl_idname", "")
|
|
if bl_id == "NodeGroupInput":
|
|
for outp in getattr(node, "outputs", []):
|
|
if outp.name == "" or outp.identifier == "__extend__":
|
|
continue
|
|
can_list = group_input_can_accept_list(tree, outp)
|
|
set_socket_shape(outp, "LIST" if can_list else "CIRCLE")
|
|
elif bl_id == "NodeGroupOutput":
|
|
for inp in getattr(node, "inputs", []):
|
|
if inp.name == "" or inp.identifier == "__extend__":
|
|
continue
|
|
is_list = is_group_output_a_list(None, tree, inp)
|
|
set_socket_shape(inp, "LIST" if is_list else "CIRCLE")
|
|
|
|
def is_output_socket_a_list(sock: Any, visited: set | None = None) -> bool:
|
|
"""Determines whether an output socket produces a list."""
|
|
if sock is None:
|
|
return False
|
|
if visited is None:
|
|
visited = set()
|
|
sock_ptr = sock.as_pointer() if hasattr(sock, "as_pointer") else id(sock)
|
|
if sock_ptr in visited:
|
|
return False
|
|
visited.add(sock_ptr)
|
|
|
|
if getattr(sock, "display_shape", "") == "LIST":
|
|
return True
|
|
|
|
node = getattr(sock, "node", None)
|
|
if node is None:
|
|
return False
|
|
|
|
bl_idname = getattr(node, "bl_idname", "")
|
|
|
|
if getattr(node, "is_list_output", False):
|
|
return True
|
|
|
|
if bl_idname == "NodeReroute":
|
|
inputs = getattr(node, "inputs", None)
|
|
if inputs and inputs[0].is_linked and inputs[0].links:
|
|
return is_output_socket_a_list(inputs[0].links[0].from_socket, visited)
|
|
return False
|
|
|
|
if bl_idname == "SLICENODE_JoinGcode":
|
|
inputs = getattr(node, "inputs", [])
|
|
return any(
|
|
is_socket_linked_to_list(inp)
|
|
for inp in inputs
|
|
if getattr(inp, "name", "") == "Gcode" or getattr(inp, "identifier", "").startswith("Gcode_ms_in_")
|
|
)
|
|
|
|
if bl_idname == "SLICENODE_Group":
|
|
tree = getattr(node, "slicer_tree", None) or getattr(node, "node_tree", None)
|
|
if tree:
|
|
return is_group_output_a_list(node, tree, sock, visited)
|
|
return False
|
|
|
|
if bl_idname == "NodeGroupInput":
|
|
tree = getattr(node, "id_data", None)
|
|
if tree:
|
|
return group_input_can_accept_list(tree, sock, visited)
|
|
return False
|
|
|
|
mapping = getattr(node, "list_socket_mapping", None)
|
|
if mapping:
|
|
sock_name = getattr(sock, "name", "")
|
|
for in_name, out_names in mapping:
|
|
targets = [out_names] if isinstance(out_names, str) else list(out_names)
|
|
if sock_name in targets:
|
|
inp = getattr(node, "inputs", None) and node.inputs.get(in_name)
|
|
if inp and is_socket_linked_to_list(inp):
|
|
return True
|
|
|
|
return False
|
|
|
|
def is_socket_linked_to_list(sock: Any) -> bool:
|
|
"""Determines whether an input socket is linked to an upstream list output."""
|
|
if not getattr(sock, "is_linked", False) or not getattr(sock, "links", None):
|
|
return False
|
|
from_socket = sock.links[0].from_socket
|
|
from_socket = _follow_reroute_upstream(from_socket)
|
|
return is_output_socket_a_list(from_socket)
|
|
|
|
def sync_node_list_sockets(node: Any, mapping: list[tuple[str, list[str] | str]]) -> None:
|
|
"""Updates display_shape on input and output sockets based on mapping.
|
|
mapping is a list of tuples: (input_socket_name, output_socket_names)
|
|
e.g. [('Paths', ['Paths']), ('Mesh', ['Selection', 'Inverted'])]
|
|
"""
|
|
for in_name, out_names in mapping:
|
|
inp = getattr(node, "inputs", None) and node.inputs.get(in_name)
|
|
if inp is None:
|
|
continue
|
|
is_list = is_socket_linked_to_list(inp)
|
|
set_socket_shape(inp, "LIST" if is_list else "CIRCLE")
|
|
|
|
if isinstance(out_names, str):
|
|
targets = [out_names]
|
|
else:
|
|
targets = list(out_names)
|
|
for out_name in targets:
|
|
out_sock = getattr(node, "outputs", None) and node.outputs.get(out_name)
|
|
if out_sock is not None:
|
|
set_socket_shape(out_sock, "LIST" if is_list else "CIRCLE")
|