897 lines
35 KiB
Python
897 lines
35 KiB
Python
from __future__ import annotations
|
|
from typing import Any, Callable
|
|
import numpy as np
|
|
|
|
LIST_CAPABLE_SOCKET_TYPES = {
|
|
"SLICESOCK_Mesh",
|
|
"SLICESOCK_Paths",
|
|
"SLICESOCK_Object",
|
|
"SLICESOCK_Geometry",
|
|
"SLICESOCK_Intrinsics",
|
|
"SLICESOCK_Settings",
|
|
"NodeSocketObject",
|
|
"NodeSocketString",
|
|
"NodeSocketVector",
|
|
"NodeSocketRotation",
|
|
"NodeSocketFloat",
|
|
"NodeSocketInt",
|
|
"NodeSocketColor",
|
|
"NodeSocketBool",
|
|
}
|
|
|
|
def is_vector_value(val: Any) -> bool:
|
|
"""Checks if val represents a single 3D/4D vector, Euler rotation, or color,
|
|
rather than a list of such values."""
|
|
if val is None:
|
|
return False
|
|
val_type = type(val).__name__
|
|
if val_type in ("Vector", "Euler", "Quaternion", "Color"):
|
|
return True
|
|
if isinstance(val, np.ndarray):
|
|
if val.ndim == 1 and val.shape[0] in (3, 4):
|
|
return True
|
|
return False
|
|
if isinstance(val, (list, tuple)):
|
|
if len(val) in (3, 4) and all(isinstance(x, (int, float, np.number)) and not isinstance(x, bool) for x in val):
|
|
return True
|
|
return False
|
|
|
|
def _is_sequence_collection(val: Any) -> bool:
|
|
if isinstance(val, (list, tuple)):
|
|
return True
|
|
if hasattr(val, "__iter__") and not hasattr(val, "shape") and not isinstance(val, (str, bytes, dict, set)):
|
|
cls_name = type(val).__name__
|
|
if "bpy_prop" in cls_name or "Collection" in cls_name or hasattr(val, "foreach_get"):
|
|
return True
|
|
import collections.abc
|
|
if isinstance(val, collections.abc.Sequence):
|
|
return True
|
|
return False
|
|
|
|
def is_input_a_list(val: Any, is_vector: bool = False) -> bool:
|
|
"""Determines whether an input value represents a list of items.
|
|
If is_vector is True, a single vector/rotation/color is treated as an atomic single value."""
|
|
if val is None or isinstance(val, (str, bytes, bool)):
|
|
return False
|
|
if type(val).__name__ == "SlicePaths" or getattr(val, "is_slice_paths", False):
|
|
return False
|
|
if is_vector:
|
|
if is_vector_value(val):
|
|
return False
|
|
return _is_sequence_collection(val)
|
|
return _is_sequence_collection(val)
|
|
|
|
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)."""
|
|
return is_input_a_list(val, is_vector=False)
|
|
|
|
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 prepare_indexed_inputs(
|
|
main_data: Any,
|
|
*secondary_inputs: Any,
|
|
is_vector: list[bool] | None = None,
|
|
input_names: list[str] | None = None,
|
|
is_atomic: list[bool] | None = None,
|
|
) -> tuple[bool, list[Any], list[list[Any]]]:
|
|
"""Prepares and validates main and secondary inputs according to the generalized list architecture:
|
|
- Main input can be a list or a single item.
|
|
- Secondary inputs that are lists must have either the same length as the main input, or be a single value (length 1 or non-list).
|
|
- If a secondary input is a single value, it is broadcast across all elements of the main input.
|
|
- If list lengths mismatch and length != 1, raises ValueError.
|
|
- If is_atomic is True for a secondary input, it is treated as a single atomic data value (e.g. layer height list)
|
|
unless it is explicitly a nested list matching the main input length.
|
|
|
|
Returns:
|
|
(is_main_list, main_items, broadcasted_secondary_lists)
|
|
"""
|
|
main_is_list = is_input_a_list(main_data, is_vector=False)
|
|
main_items = list(main_data) if main_is_list else [main_data]
|
|
n = len(main_items)
|
|
|
|
vec_flags = is_vector or [False] * len(secondary_inputs)
|
|
atom_flags = is_atomic or [False] * len(secondary_inputs)
|
|
names = input_names or [f"input_{i+1}" for i in range(len(secondary_inputs))]
|
|
broadcasted_secondaries: list[list[Any]] = []
|
|
|
|
for i, sec_val in enumerate(secondary_inputs):
|
|
is_vec = vec_flags[i] if i < len(vec_flags) else False
|
|
is_atom = atom_flags[i] if i < len(atom_flags) else False
|
|
name = names[i] if i < len(names) else f"input_{i+1}"
|
|
|
|
if is_atom:
|
|
is_nested_list = (
|
|
isinstance(sec_val, (list, tuple))
|
|
and len(sec_val) > 0
|
|
and isinstance(sec_val[0], (list, tuple))
|
|
)
|
|
if not is_nested_list:
|
|
broadcasted_secondaries.append([sec_val] * n)
|
|
elif len(sec_val) == n:
|
|
broadcasted_secondaries.append(list(sec_val))
|
|
elif len(sec_val) == 1:
|
|
broadcasted_secondaries.append([sec_val[0]] * n)
|
|
else:
|
|
m = len(sec_val)
|
|
expected = "1 or a single value" if not main_is_list else f"{n} or 1"
|
|
main_desc = f"{n} items" if main_is_list else "a single item"
|
|
raise ValueError(
|
|
f"Input list length mismatch: main input has {main_desc}, "
|
|
f"but secondary input '{name}' has {m} items (must be {expected})."
|
|
)
|
|
continue
|
|
|
|
if not is_input_a_list(sec_val, is_vector=is_vec):
|
|
broadcasted_secondaries.append([sec_val] * n)
|
|
elif len(sec_val) == n:
|
|
broadcasted_secondaries.append(list(sec_val))
|
|
elif len(sec_val) == 1:
|
|
broadcasted_secondaries.append([sec_val[0]] * n)
|
|
else:
|
|
m = len(sec_val)
|
|
expected = "1 or a single value" if not main_is_list else f"{n} or 1"
|
|
main_desc = f"{n} items" if main_is_list else "a single item"
|
|
raise ValueError(
|
|
f"Input list length mismatch: main input has {main_desc}, "
|
|
f"but secondary input '{name}' has {m} items (must be {expected})."
|
|
)
|
|
|
|
return main_is_list, main_items, broadcasted_secondaries
|
|
|
|
def eval_element_wise_multivalue(
|
|
op_func: Callable[..., Any],
|
|
main_data: Any,
|
|
*secondary_inputs: Any,
|
|
is_vector: list[bool] | None = None,
|
|
input_names: list[str] | None = None,
|
|
output_index: int | None = None,
|
|
is_atomic: list[bool] | None = None,
|
|
) -> Any:
|
|
"""Evaluates op_func element-wise with support for main and secondary list inputs:
|
|
- If main_data is a list, returns a list of results.
|
|
- If main_data is a single item, returns a single result.
|
|
- Secondary inputs that are lists are matched index per index.
|
|
- Secondary inputs that are single values (or length-1 lists) are broadcast across all main elements.
|
|
- Mismatched secondary list lengths raise ValueError.
|
|
- If output_index is not None, unpacks result[output_index] from the op_func return value.
|
|
"""
|
|
is_main_list, main_items, sec_lists = prepare_indexed_inputs(
|
|
main_data,
|
|
*secondary_inputs,
|
|
is_vector=is_vector,
|
|
input_names=input_names,
|
|
is_atomic=is_atomic,
|
|
)
|
|
|
|
def _call(item: Any, i: int) -> Any:
|
|
res = op_func(item, *(sec[i] for sec in sec_lists))
|
|
if output_index is not None and isinstance(res, tuple):
|
|
try:
|
|
return res[output_index]
|
|
except (IndexError, TypeError):
|
|
return None
|
|
return res
|
|
|
|
if is_main_list:
|
|
return [_call(item, i) for i, item in enumerate(main_items)]
|
|
return _call(main_items[0], 0)
|
|
|
|
def is_socket_vector_type(node: Any, socket_name: str) -> bool:
|
|
"""Checks whether the socket on the node is a vector/rotation/color socket."""
|
|
inputs = getattr(node, "inputs", None)
|
|
if inputs:
|
|
sock = None
|
|
if hasattr(inputs, "get"):
|
|
sock = inputs.get(socket_name)
|
|
elif isinstance(inputs, (list, tuple)):
|
|
for s in inputs:
|
|
if getattr(s, "name", "") == socket_name:
|
|
sock = s
|
|
break
|
|
if sock is not None:
|
|
bl_id = getattr(sock, "bl_idname", "")
|
|
if bl_id in ("NodeSocketVector", "NodeSocketRotation", "NodeSocketColor", "NodeSocketVectorXYZ", "NodeSocketVectorEuler"):
|
|
return True
|
|
if bl_id in ("NodeSocketFloat", "NodeSocketInt", "NodeSocketString", "NodeSocketBool", "SLICESOCK_Mesh", "SLICESOCK_Paths", "SLICESOCK_Settings"):
|
|
return False
|
|
name_lower = socket_name.lower()
|
|
return name_lower in ("translation", "rotation", "scale", "offset", "center", "layer rotation", "color")
|
|
|
|
def eval_multivalue(
|
|
node: Any,
|
|
op_func: Callable[..., Any],
|
|
main_socket: str | None = None,
|
|
secondary_sockets: list[str] | None = None,
|
|
*,
|
|
socket_name: str | None = None,
|
|
default: Any = None,
|
|
is_vector: list[bool] | None = None,
|
|
is_atomic: list[bool] | None = None,
|
|
) -> Any:
|
|
"""Generalized element-wise multi-value evaluator for SliceNode instances."""
|
|
mapping = getattr(node, "list_socket_mapping", None)
|
|
|
|
# Infer main_socket if not specified
|
|
if main_socket is None:
|
|
if mapping and len(mapping) > 0 and mapping[0][0]:
|
|
main_socket = mapping[0][0]
|
|
else:
|
|
inputs = getattr(node, "inputs", None)
|
|
if inputs and len(inputs) > 0:
|
|
main_socket = getattr(inputs[0], "name", "")
|
|
else:
|
|
return default
|
|
|
|
# Determine output index / socket validation
|
|
output_index: int | None = None
|
|
if socket_name is not None and mapping:
|
|
out_spec = next((outs for m_in, outs in mapping if m_in.lower() == (main_socket or "").lower()), None)
|
|
if out_spec is not None:
|
|
if isinstance(out_spec, (list, tuple)):
|
|
matched_idx = next((idx for idx, s in enumerate(out_spec) if s.lower() == socket_name.lower()), None)
|
|
if matched_idx is None:
|
|
return default
|
|
output_index = matched_idx
|
|
elif isinstance(out_spec, str) and out_spec:
|
|
if socket_name.lower() != out_spec.lower():
|
|
return default
|
|
|
|
# Resolve main input
|
|
resolve_fn = getattr(node, "resolve", None)
|
|
if not callable(resolve_fn):
|
|
return default
|
|
|
|
main_data = resolve_fn(main_socket)
|
|
if main_data is None:
|
|
return default
|
|
if isinstance(main_data, list) and len(main_data) == 0:
|
|
return []
|
|
|
|
# Infer secondary_sockets if not specified
|
|
if secondary_sockets is None:
|
|
inputs = getattr(node, "inputs", [])
|
|
secondary_sockets = [s.name for s in inputs if getattr(s, "name", "") != main_socket]
|
|
|
|
# Resolve secondary inputs
|
|
sec_vals = [resolve_fn(sname) for sname in secondary_sockets]
|
|
|
|
# Infer is_vector if not specified
|
|
if is_vector is None:
|
|
vec_flags = [is_socket_vector_type(node, sname) for sname in secondary_sockets]
|
|
else:
|
|
vec_flags = is_vector
|
|
|
|
try:
|
|
return eval_element_wise_multivalue(
|
|
op_func,
|
|
main_data,
|
|
*sec_vals,
|
|
is_vector=vec_flags,
|
|
input_names=secondary_sockets,
|
|
output_index=output_index,
|
|
is_atomic=is_atomic,
|
|
)
|
|
except Exception as e:
|
|
from .blender_slice_nodes_generic import attach_node_failure_info
|
|
known_inputs = ([main_socket] if main_socket else []) + (secondary_sockets or [])
|
|
attach_node_failure_info(node, e, fallback_inputs=known_inputs)
|
|
raise
|
|
|
|
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:
|
|
if getattr(sock, "display_shape", "") != shape:
|
|
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
|
|
|
|
# Socket type check
|
|
sock_type = getattr(sock, "bl_idname", "")
|
|
if sock_type not in LIST_CAPABLE_SOCKET_TYPES:
|
|
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
|
|
|
|
return True
|
|
|
|
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
|
|
|
|
always_outs = getattr(from_node, "always_list_outputs", None)
|
|
if always_outs and getattr(from_sock, "name", "") in always_outs:
|
|
return True
|
|
|
|
if getattr(from_node, "bl_idname", "") in ("SLICENODE_FloatMath", "SLICENODE_IntMath"):
|
|
for inp in getattr(from_node, "inputs", []):
|
|
if is_internal_socket_a_list(group_node, tree, inp, visited):
|
|
return True
|
|
return False
|
|
|
|
if getattr(from_node, "bl_idname", "") == "SLICENODE_PathOperations":
|
|
inputs = getattr(from_node, "inputs", [])
|
|
if inputs:
|
|
return is_internal_socket_a_list(group_node, tree, inputs[0], visited)
|
|
return False
|
|
|
|
if getattr(from_node, "bl_idname", "") == "SLICENODE_SeparateMeshByRange":
|
|
inputs = getattr(from_node, "inputs", None)
|
|
mesh_inp = inputs.get("Mesh") if inputs else None
|
|
if mesh_inp:
|
|
return is_internal_socket_a_list(group_node, tree, mesh_inp, visited)
|
|
return False
|
|
|
|
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
|
|
|
|
always_outs = getattr(src_node, "always_list_outputs", None)
|
|
if always_outs and getattr(src_sock, "name", "") in always_outs:
|
|
return True
|
|
|
|
if src_bl_id in ("SLICENODE_FloatMath", "SLICENODE_IntMath"):
|
|
for inp in getattr(src_node, "inputs", []):
|
|
if is_internal_socket_a_list(group_node, tree, inp, visited):
|
|
return True
|
|
return False
|
|
|
|
if src_bl_id == "SLICENODE_PathOperations":
|
|
inputs = getattr(src_node, "inputs", [])
|
|
if inputs:
|
|
return is_internal_socket_a_list(group_node, tree, inputs[0], visited)
|
|
return False
|
|
|
|
if src_bl_id == "SLICENODE_SeparateMeshByRange":
|
|
inputs = getattr(src_node, "inputs", None)
|
|
mesh_inp = inputs.get("Mesh") if inputs else None
|
|
if mesh_inp:
|
|
return is_internal_socket_a_list(group_node, tree, mesh_inp, visited)
|
|
return False
|
|
|
|
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
|
|
|
|
always_outs = getattr(node, "always_list_outputs", None)
|
|
if always_outs and getattr(sock, "name", "") in always_outs:
|
|
return True
|
|
|
|
if bl_idname in ("SLICENODE_FloatMath", "SLICENODE_IntMath"):
|
|
return any(is_socket_linked_to_list(inp) for inp in getattr(node, "inputs", []))
|
|
|
|
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_PathOperations":
|
|
inputs = getattr(node, "inputs", None)
|
|
return is_socket_linked_to_list(inputs[0]) if inputs else False
|
|
|
|
if bl_idname == "SLICENODE_SeparateMeshByRange":
|
|
inputs = getattr(node, "inputs", None)
|
|
mesh_inp = inputs.get("Mesh") if inputs else None
|
|
return is_socket_linked_to_list(mesh_inp) if mesh_inp else False
|
|
|
|
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 = None) -> None:
|
|
"""Updates display_shape on input and output sockets based on mapping and node rules.
|
|
Ensures list inputs/outputs consistently display as 'LIST', and single values as 'CIRCLE'.
|
|
"""
|
|
if node is None:
|
|
return
|
|
|
|
bl_id = getattr(node, "bl_idname", "")
|
|
if bl_id == "NodeReroute":
|
|
return
|
|
if bl_id == "SLICENODE_Group":
|
|
sync_group_node_socket_shapes(node)
|
|
return
|
|
|
|
if mapping is None:
|
|
mapping = getattr(node, "list_socket_mapping", None) or []
|
|
|
|
always_in = set(getattr(node, "always_list_inputs", []) or [])
|
|
always_out = set(getattr(node, "always_list_outputs", []) or [])
|
|
is_list_out = bool(getattr(node, "is_list_output", False))
|
|
|
|
if bl_id in ("SLICENODE_FloatMath", "SLICENODE_IntMath"):
|
|
inputs = getattr(node, "inputs", [])
|
|
any_list = any(is_socket_linked_to_list(inp) for inp in inputs)
|
|
for inp in inputs:
|
|
set_socket_shape(inp, "LIST" if is_socket_linked_to_list(inp) else "CIRCLE")
|
|
for outp in getattr(node, "outputs", []):
|
|
set_socket_shape(outp, "LIST" if any_list else "CIRCLE")
|
|
return
|
|
|
|
if bl_id == "SLICENODE_JoinGcode":
|
|
inputs = getattr(node, "inputs", [])
|
|
has_list = any(
|
|
is_socket_linked_to_list(inp)
|
|
for inp in inputs
|
|
if getattr(inp, "name", "") == "Gcode" or getattr(inp, "identifier", "").startswith("Gcode_ms_in_")
|
|
)
|
|
for inp in inputs:
|
|
set_socket_shape(inp, "LIST" if is_socket_linked_to_list(inp) else "CIRCLE")
|
|
out_sock = getattr(node, "outputs", None) and node.outputs.get("Gcode")
|
|
if out_sock:
|
|
set_socket_shape(out_sock, "LIST" if has_list else "CIRCLE")
|
|
return
|
|
|
|
if bl_id == "SLICENODE_PathOperations":
|
|
inputs = getattr(node, "inputs", [])
|
|
first_is_list = is_socket_linked_to_list(inputs[0]) if inputs else False
|
|
for inp in inputs:
|
|
set_socket_shape(inp, "LIST" if is_socket_linked_to_list(inp) else "CIRCLE")
|
|
out_sock = getattr(node, "outputs", None) and (node.outputs.get("Paths") or (node.outputs[0] if node.outputs else None))
|
|
if out_sock:
|
|
set_socket_shape(out_sock, "LIST" if first_is_list else "CIRCLE")
|
|
return
|
|
|
|
if bl_id == "SLICENODE_SeparateMeshByRange":
|
|
if hasattr(node, "_update_outputs"):
|
|
node._update_outputs()
|
|
return
|
|
|
|
mapped_inputs = set()
|
|
mapped_outputs = set()
|
|
|
|
for in_name, out_names in mapping:
|
|
mapped_inputs.add(in_name)
|
|
inp = getattr(node, "inputs", None) and node.inputs.get(in_name)
|
|
if inp is None:
|
|
continue
|
|
in_is_list = (in_name in always_in) or is_socket_linked_to_list(inp)
|
|
set_socket_shape(inp, "LIST" if in_is_list else "CIRCLE")
|
|
|
|
if isinstance(out_names, str):
|
|
targets = [out_names]
|
|
else:
|
|
targets = list(out_names)
|
|
for out_name in targets:
|
|
mapped_outputs.add(out_name)
|
|
out_sock = getattr(node, "outputs", None) and node.outputs.get(out_name)
|
|
if out_sock is not None:
|
|
out_is_list = is_list_out or (out_name in always_out) or in_is_list
|
|
set_socket_shape(out_sock, "LIST" if out_is_list else "CIRCLE")
|
|
|
|
for inp in getattr(node, "inputs", []):
|
|
inp_name = getattr(inp, "name", "")
|
|
if inp_name not in mapped_inputs:
|
|
in_is_list = (inp_name in always_in) or is_socket_linked_to_list(inp)
|
|
set_socket_shape(inp, "LIST" if in_is_list else "CIRCLE")
|
|
|
|
for outp in getattr(node, "outputs", []):
|
|
outp_name = getattr(outp, "name", "")
|
|
if outp_name not in mapped_outputs:
|
|
out_is_list = is_list_out or (outp_name in always_out)
|
|
set_socket_shape(outp, "LIST" if out_is_list else "CIRCLE")
|
|
|
|
|
|
def sync_tree_socket_shapes(tree: Any) -> None:
|
|
"""Updates display_shape ('LIST' vs 'CIRCLE') consistently across all nodes in the tree."""
|
|
if not tree or not hasattr(tree, "nodes"):
|
|
return
|
|
|
|
sync_group_tree_internal_sockets(tree)
|
|
|
|
for _ in range(2):
|
|
for node in tree.nodes:
|
|
bl_id = getattr(node, "bl_idname", "")
|
|
if bl_id in ("NodeReroute", "NodeGroupInput", "NodeGroupOutput"):
|
|
continue
|
|
sync_node_list_sockets(node)
|
|
|