999 lines
29 KiB
Python
999 lines
29 KiB
Python
import uuid
|
|
|
|
import typing as typ
|
|
import typing_extensions as typx
|
|
import json
|
|
import inspect
|
|
|
|
import bpy
|
|
import pydantic as pyd
|
|
|
|
from .. import contracts as ct
|
|
from .. import sockets
|
|
|
|
CACHE: dict[str, typ.Any] = {} ## By Instance UUID
|
|
## NOTE: CACHE does not persist between file loads.
|
|
|
|
_DEFAULT_LOOSE_SOCKET_SER = json.dumps({
|
|
"socket_names": [],
|
|
"socket_def_names": [],
|
|
"models": [],
|
|
})
|
|
|
|
class MaxwellSimNode(bpy.types.Node):
|
|
# Fundamentals
|
|
node_type: ct.NodeType
|
|
bl_idname: str
|
|
use_sim_node_name: bool = False
|
|
bl_label: str
|
|
#draw_label(self) -> str: pass
|
|
|
|
# Style
|
|
bl_description: str = ""
|
|
|
|
#bl_width_default: float = 0.0
|
|
#bl_width_min: float = 0.0
|
|
#bl_width_max: float = 0.0
|
|
|
|
# Sockets
|
|
_output_socket_methods: dict
|
|
|
|
input_sockets: dict[str, ct.schemas.SocketDef] = {}
|
|
output_sockets: dict[str, ct.schemas.SocketDef] = {}
|
|
input_socket_sets: dict[str, dict[str, ct.schemas.SocketDef]] = {}
|
|
output_socket_sets: dict[str, dict[str, ct.schemas.SocketDef]] = {}
|
|
|
|
# Presets
|
|
presets = {}
|
|
|
|
# Managed Objects
|
|
managed_obj_defs: dict[ct.ManagedObjName, ct.schemas.ManagedObjDef] = {}
|
|
|
|
####################
|
|
# - Initialization
|
|
####################
|
|
def __init_subclass__(cls, **kwargs: typ.Any):
|
|
super().__init_subclass__(**kwargs)
|
|
|
|
# Setup Blender ID for Node
|
|
if not hasattr(cls, "node_type"):
|
|
msg = f"Node class {cls} does not define 'node_type', or it is does not have the type {ct.NodeType}"
|
|
raise ValueError(msg)
|
|
cls.bl_idname = str(cls.node_type.value)
|
|
|
|
# Setup Instance ID for Node
|
|
cls.__annotations__["instance_id"] = bpy.props.StringProperty(
|
|
name="Instance ID",
|
|
description="The instance ID of a particular MaxwellSimNode instance, used to index caches",
|
|
default="",
|
|
)
|
|
|
|
# Setup Name Property for Node
|
|
cls.__annotations__["sim_node_name"] = bpy.props.StringProperty(
|
|
name="Sim Node Name",
|
|
description="The name of a particular MaxwellSimNode node, which can be used to help identify data managed by the node",
|
|
default="",
|
|
update=(lambda self, context: self.sync_sim_node_name(context))
|
|
)
|
|
|
|
# Setup Locked Property for Node
|
|
cls.__annotations__["locked"] = bpy.props.BoolProperty(
|
|
name="Locked State",
|
|
description="The lock-state of a particular MaxwellSimNode instance, which determines the node's user editability",
|
|
default=False,
|
|
)
|
|
|
|
# Setup Blender Label for Node
|
|
if not hasattr(cls, "bl_label"):
|
|
msg = f"Node class {cls} does not define 'bl_label'"
|
|
raise ValueError(msg)
|
|
|
|
# Setup Callback Methods
|
|
cls._output_socket_methods = {
|
|
method._index_by: method
|
|
for attr_name in dir(cls)
|
|
if hasattr(
|
|
method := getattr(cls, attr_name),
|
|
"_callback_type"
|
|
) and method._callback_type == "computes_output_socket"
|
|
}
|
|
cls._on_value_changed_methods = {
|
|
method
|
|
for attr_name in dir(cls)
|
|
if hasattr(
|
|
method := getattr(cls, attr_name),
|
|
"_callback_type"
|
|
) and method._callback_type == "on_value_changed"
|
|
}
|
|
cls._on_show_preview = {
|
|
method
|
|
for attr_name in dir(cls)
|
|
if hasattr(
|
|
method := getattr(cls, attr_name),
|
|
"_callback_type"
|
|
) and method._callback_type == "on_show_preview"
|
|
}
|
|
cls._on_show_plot = {
|
|
method
|
|
for attr_name in dir(cls)
|
|
if hasattr(
|
|
method := getattr(cls, attr_name),
|
|
"_callback_type"
|
|
) and method._callback_type == "on_show_plot"
|
|
}
|
|
cls._on_init = {
|
|
method
|
|
for attr_name in dir(cls)
|
|
if hasattr(
|
|
method := getattr(cls, attr_name),
|
|
"_callback_type"
|
|
) and method._callback_type == "on_init"
|
|
}
|
|
|
|
# Setup Socket Set Dropdown
|
|
if not len(cls.input_socket_sets) + len(cls.output_socket_sets) > 0:
|
|
cls.active_socket_set = None
|
|
else:
|
|
## Add Active Socket Set Enum
|
|
socket_set_names = (
|
|
(_input_socket_set_names := list(cls.input_socket_sets.keys()))
|
|
+ [
|
|
output_socket_set_name
|
|
for output_socket_set_name in cls.output_socket_sets.keys()
|
|
if output_socket_set_name not in _input_socket_set_names
|
|
]
|
|
)
|
|
socket_set_ids = [
|
|
socket_set_name.replace(" ", "_").upper()
|
|
for socket_set_name in socket_set_names
|
|
]
|
|
## TODO: Better deriv. of sock.set. ID, ex. ( is currently invalid.
|
|
|
|
## Add Active Socket Set Enum
|
|
cls.__annotations__["active_socket_set"] = bpy.props.EnumProperty(
|
|
name="Active Socket Set",
|
|
description="The active socket set",
|
|
items=[
|
|
(
|
|
socket_set_name,
|
|
socket_set_name,
|
|
socket_set_name,
|
|
)
|
|
for socket_set_id, socket_set_name in zip(
|
|
socket_set_ids,
|
|
socket_set_names,
|
|
)
|
|
],
|
|
default=socket_set_names[0],
|
|
update=lambda self, context: self.sync_active_socket_set(context),
|
|
)
|
|
|
|
# Setup Preset Dropdown
|
|
if not cls.presets:
|
|
cls.active_preset = None
|
|
else:
|
|
## TODO: Check that presets are represented in a socket that is guaranteed to be always available, specifically either a static socket or ALL static socket sets.
|
|
cls.__annotations__["active_preset"] = bpy.props.EnumProperty(
|
|
name="Active Preset",
|
|
description="The active preset",
|
|
items=[
|
|
(
|
|
preset_name,
|
|
preset_def.label,
|
|
preset_def.description,
|
|
)
|
|
for preset_name, preset_def in cls.presets.items()
|
|
],
|
|
default=list(cls.presets.keys())[0],
|
|
update=lambda self, context: (
|
|
self.sync_active_preset()()
|
|
),
|
|
)
|
|
|
|
####################
|
|
# - Generic Properties
|
|
####################
|
|
def sync_active_socket_set(self, context):
|
|
self.sync_sockets()
|
|
self.sync_prop("active_socket_set", context)
|
|
|
|
def sync_sim_node_name(self, context):
|
|
if (mobjs := CACHE[self.instance_id].get("managed_objs")) is None:
|
|
return
|
|
|
|
for mobj_id, mobj in mobjs.items():
|
|
# Retrieve Managed Obj Definition
|
|
mobj_def = self.managed_obj_defs[mobj_id]
|
|
|
|
# Set Managed Obj Name
|
|
mobj.name = mobj_def.name_prefix + self.sim_node_name
|
|
## ManagedObj is allowed to alter the name when setting it.
|
|
## - This will happen whenever the name is taken.
|
|
## - If altered, set the 'sim_node_name' to the altered name.
|
|
## - This will cause recursion, but only once.
|
|
|
|
####################
|
|
# - Managed Object Properties
|
|
####################
|
|
@property
|
|
def managed_objs(self):
|
|
global CACHE
|
|
if not CACHE.get(self.instance_id):
|
|
CACHE[self.instance_id] = {}
|
|
|
|
# If No Managed Objects in CACHE: Initialize Managed Objects
|
|
## - This happens on every ex. file load, init(), etc. .
|
|
## - ManagedObjects MUST the same object by name.
|
|
## - We sync our 'sim_node_name' with all managed objects.
|
|
## - (There is also a class-defined 'name_prefix' to differentiate)
|
|
## - See the 'sim_node_name' w/its sync function.
|
|
if CACHE[self.instance_id].get("managed_objs") is None:
|
|
# Initialize the Managed Object Instance Cache
|
|
CACHE[self.instance_id]["managed_objs"] = {}
|
|
|
|
# Fill w/Managed Objects by Name Socket
|
|
for mobj_id, mobj_def in self.managed_obj_defs.items():
|
|
name = mobj_def.name_prefix + self.sim_node_name
|
|
CACHE[self.instance_id]["managed_objs"][mobj_id] = (
|
|
mobj_def.mk(name)
|
|
)
|
|
|
|
return CACHE[self.instance_id]["managed_objs"]
|
|
|
|
return CACHE[self.instance_id]["managed_objs"]
|
|
|
|
####################
|
|
# - Socket Properties
|
|
####################
|
|
def active_bl_sockets(self, direc: typx.Literal["input", "output"]):
|
|
return self.inputs if direc == "input" else self.outputs
|
|
|
|
def active_socket_set_sockets(
|
|
self,
|
|
direc: typx.Literal["input", "output"],
|
|
) -> dict:
|
|
# No Active Socket Set: Return Nothing
|
|
if not self.active_socket_set: return {}
|
|
|
|
# Retrieve Active Socket Set Sockets
|
|
socket_sets = (
|
|
self.input_socket_sets
|
|
if direc == "input" else self.output_socket_sets
|
|
)
|
|
active_socket_set_sockets = socket_sets.get(
|
|
self.active_socket_set
|
|
)
|
|
|
|
# Return Active Socket Set Sockets (if any)
|
|
if not active_socket_set_sockets: return {}
|
|
return active_socket_set_sockets
|
|
|
|
def active_sockets(self, direc: typx.Literal["input", "output"]):
|
|
static_sockets = (
|
|
self.input_sockets
|
|
if direc == "input"
|
|
else self.output_sockets
|
|
)
|
|
socket_sets = (
|
|
self.input_socket_sets
|
|
if direc == "input"
|
|
else self.output_socket_sets
|
|
)
|
|
loose_sockets = (
|
|
self.loose_input_sockets
|
|
if direc == "input"
|
|
else self.loose_output_sockets
|
|
)
|
|
|
|
return (
|
|
static_sockets
|
|
| self.active_socket_set_sockets(direc=direc)
|
|
| loose_sockets
|
|
)
|
|
|
|
####################
|
|
# - Loose Sockets
|
|
####################
|
|
# Loose Sockets
|
|
## Only Blender props persist as instance data
|
|
ser_loose_input_sockets: bpy.props.StringProperty(
|
|
name="Serialized Loose Input Sockets",
|
|
description="JSON-serialized representation of loose input sockets.",
|
|
default=_DEFAULT_LOOSE_SOCKET_SER,
|
|
)
|
|
ser_loose_output_sockets: bpy.props.StringProperty(
|
|
name="Serialized Loose Input Sockets",
|
|
description="JSON-serialized representation of loose input sockets.",
|
|
default=_DEFAULT_LOOSE_SOCKET_SER,
|
|
)
|
|
|
|
## Internal Serialization/Deserialization Methods (yuck)
|
|
def _ser_loose_sockets(self, deser: dict[str, ct.schemas.SocketDef]) -> str:
|
|
if not all(isinstance(model, pyd.BaseModel) for model in deser.values()):
|
|
msg = "Trying to deserialize loose sockets with invalid SocketDefs (they must be `pydantic` BaseModels)."
|
|
raise ValueError(msg)
|
|
|
|
return json.dumps({
|
|
"socket_names": list(deser.keys()),
|
|
"socket_def_names": [
|
|
model.__class__.__name__
|
|
for model in deser.values()
|
|
],
|
|
"models": [
|
|
model.model_dump()
|
|
for model in deser.values()
|
|
if isinstance(model, pyd.BaseModel)
|
|
],
|
|
}) ## Big reliance on order-preservation of dicts here.)
|
|
def _deser_loose_sockets(self, ser: str) -> dict[str, ct.schemas.SocketDef]:
|
|
semi_deser = json.loads(ser)
|
|
return {
|
|
socket_name: getattr(sockets, socket_def_name)(**model_kwargs)
|
|
for socket_name, socket_def_name, model_kwargs in zip(
|
|
semi_deser["socket_names"],
|
|
semi_deser["socket_def_names"],
|
|
semi_deser["models"],
|
|
)
|
|
if hasattr(sockets, socket_def_name)
|
|
}
|
|
|
|
@property
|
|
def loose_input_sockets(self) -> dict[str, ct.schemas.SocketDef]:
|
|
return self._deser_loose_sockets(self.ser_loose_input_sockets)
|
|
@property
|
|
def loose_output_sockets(self) -> dict[str, ct.schemas.SocketDef]:
|
|
return self._deser_loose_sockets(self.ser_loose_output_sockets)
|
|
## TODO: Some caching may play a role if this is all too slow.
|
|
|
|
@loose_input_sockets.setter
|
|
def loose_input_sockets(
|
|
self, value: dict[str, ct.schemas.SocketDef],
|
|
) -> None:
|
|
if not value: self.ser_loose_input_sockets = _DEFAULT_LOOSE_SOCKET_SER
|
|
else: self.ser_loose_input_sockets = self._ser_loose_sockets(value)
|
|
|
|
# Synchronize Sockets
|
|
self.sync_sockets()
|
|
## TODO: Perhaps re-init() all loose sockets anyway?
|
|
|
|
@loose_output_sockets.setter
|
|
def loose_output_sockets(
|
|
self, value: dict[str, ct.schemas.SocketDef],
|
|
) -> None:
|
|
if not value: self.ser_loose_output_sockets = _DEFAULT_LOOSE_SOCKET_SER
|
|
else: self.ser_loose_output_sockets = self._ser_loose_sockets(value)
|
|
|
|
# Synchronize Sockets
|
|
self.sync_sockets()
|
|
## TODO: Perhaps re-init() all loose sockets anyway?
|
|
|
|
####################
|
|
# - Socket Management
|
|
####################
|
|
def _prune_inactive_sockets(self):
|
|
"""Remove all inactive sockets from the node.
|
|
|
|
**NOTE**: Socket names must be unique within direction, active socket set, and loose socket set.
|
|
"""
|
|
for direc in ["input", "output"]:
|
|
sockets = self.active_sockets(direc)
|
|
bl_sockets = self.active_bl_sockets(direc)
|
|
|
|
# Determine Sockets to Remove
|
|
bl_sockets_to_remove = [
|
|
bl_socket
|
|
for socket_name, bl_socket in bl_sockets.items()
|
|
if socket_name not in sockets
|
|
]
|
|
|
|
# Remove Sockets
|
|
for bl_socket in bl_sockets_to_remove:
|
|
bl_sockets.remove(bl_socket)
|
|
|
|
def _add_new_active_sockets(self):
|
|
"""Add and initialize all non-existing active sockets to the node.
|
|
|
|
Existing sockets within the given direction are not re-created.
|
|
"""
|
|
for direc in ["input", "output"]:
|
|
sockets = self.active_sockets(direc)
|
|
bl_sockets = self.active_bl_sockets(direc)
|
|
|
|
# Define BL Sockets
|
|
created_sockets = {}
|
|
for socket_name, socket_def in sockets.items():
|
|
# Skip Existing Sockets
|
|
if socket_name in bl_sockets: continue
|
|
|
|
# Create BL Socket from Socket
|
|
bl_socket = bl_sockets.new(
|
|
str(socket_def.socket_type.value),
|
|
socket_name,
|
|
)
|
|
bl_socket.display_shape = bl_socket.socket_shape
|
|
## `display_shape` needs to be dynamically set
|
|
|
|
# Record Created Socket
|
|
created_sockets[socket_name] = socket_def
|
|
|
|
# Initialize Just-Created BL Sockets
|
|
for socket_name, socket_def in created_sockets.items():
|
|
socket_def.init(bl_sockets[socket_name])
|
|
|
|
def sync_sockets(self) -> None:
|
|
"""Synchronize the node's sockets with the active sockets.
|
|
|
|
- Any non-existing active socket will be added and initialized.
|
|
- Any existing active socket will not be changed.
|
|
- Any existing inactive socket will be removed.
|
|
|
|
Must be called after any change to socket definitions, including loose
|
|
sockets.
|
|
"""
|
|
self._prune_inactive_sockets()
|
|
self._add_new_active_sockets()
|
|
|
|
####################
|
|
# - Preset Management
|
|
####################
|
|
def sync_active_preset(self) -> None:
|
|
"""Applies the active preset by overwriting the value of
|
|
preset-defined input sockets.
|
|
"""
|
|
if not (preset_def := self.presets.get(self.active_preset)):
|
|
msg = f"Tried to apply active preset, but the active preset ({self.active_preset}) is not in presets ({self.presets})"
|
|
raise RuntimeError(msg)
|
|
|
|
for socket_name, socket_value in preset_def.values.items():
|
|
if not (bl_socket := self.inputs.get(socket_name)):
|
|
msg = f"Tried to set preset socket/value pair ({socket_name}={socket_value}), but socket is not in active input sockets ({self.inputs})"
|
|
raise ValueError(msg)
|
|
|
|
bl_socket.value = socket_value
|
|
## TODO: Lazy-valued presets?
|
|
|
|
####################
|
|
# - UI Methods
|
|
####################
|
|
def draw_buttons(
|
|
self,
|
|
context: bpy.types.Context,
|
|
layout: bpy.types.UILayout,
|
|
) -> None:
|
|
if self.locked: layout.enabled = False
|
|
|
|
if self.active_preset:
|
|
layout.prop(self, "active_preset", text="")
|
|
|
|
if self.active_socket_set:
|
|
layout.prop(self, "active_socket_set", text="")
|
|
|
|
# Draw Name
|
|
col = layout.column(align=False)
|
|
if self.use_sim_node_name:
|
|
row = col.row(align=True)
|
|
row.label(text="", icon="FILE_TEXT")
|
|
row.prop(self, "sim_node_name", text="")
|
|
|
|
# Draw Name
|
|
self.draw_props(context, col)
|
|
self.draw_operators(context, col)
|
|
self.draw_info(context, col)
|
|
|
|
## TODO: Managed Operators instead of this shit
|
|
def draw_props(self, context, layout): pass
|
|
def draw_operators(self, context, layout): pass
|
|
def draw_info(self, context, layout): pass
|
|
|
|
def draw_buttons_ext(self, context, layout): pass
|
|
## TODO: Side panel buttons for fanciness.
|
|
|
|
def draw_plot_settings(self, context, layout):
|
|
if self.locked: layout.enabled = False
|
|
|
|
####################
|
|
# - Data Flow
|
|
####################
|
|
def _compute_input(
|
|
self,
|
|
input_socket_name: ct.SocketName,
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
) -> typ.Any | None:
|
|
"""Computes the data of an input socket, by socket name and data flow kind, by asking the socket nicely via `bl_socket.compute_data`.
|
|
|
|
Args:
|
|
input_socket_name: The name of the input socket, as defined in
|
|
`self.input_sockets`.
|
|
kind: The data flow kind to compute retrieve.
|
|
"""
|
|
if not (bl_socket := self.inputs.get(input_socket_name)):
|
|
return None
|
|
#msg = f"Input socket name {input_socket_name} is not an active input sockets."
|
|
#raise ValueError(msg)
|
|
|
|
return bl_socket.compute_data(kind=kind)
|
|
|
|
def compute_output(
|
|
self,
|
|
output_socket_name: ct.SocketName,
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
) -> typ.Any:
|
|
"""Computes the value of an output socket name, from its socket name.
|
|
|
|
Searches methods decorated with `@computes_output_socket(output_socket_name, kind=..., ...)`, for a perfect match to the pair `socket_name..kind`.
|
|
This method is run to produce the value.
|
|
|
|
Args:
|
|
output_socket_name: The name declaring the output socket,
|
|
for which this method computes the output.
|
|
|
|
Returns:
|
|
The value of the output socket, as computed by the dedicated method
|
|
registered using the `@computes_output_socket` decorator.
|
|
"""
|
|
if not (
|
|
output_socket_method := self._output_socket_methods.get(
|
|
(output_socket_name, kind)
|
|
)
|
|
):
|
|
msg = f"No output method for ({output_socket_name}, {str(kind.value)}"
|
|
raise ValueError(msg)
|
|
|
|
return output_socket_method(self)
|
|
|
|
####################
|
|
# - Action Chain
|
|
####################
|
|
def sync_prop(self, prop_name: str, context: bpy.types.Context):
|
|
"""Called when a property has been updated.
|
|
"""
|
|
if not hasattr(self, prop_name):
|
|
msg = f"Property {prop_name} not defined on socket {self}"
|
|
raise RuntimeError(msg)
|
|
|
|
self.trigger_action("value_changed", prop_name=prop_name)
|
|
|
|
def trigger_action(
|
|
self,
|
|
action: typx.Literal["enable_lock", "disable_lock", "value_changed", "show_preview", "show_plot"],
|
|
socket_name: ct.SocketName | None = None,
|
|
prop_name: ct.SocketName | None = None,
|
|
) -> None:
|
|
"""Reports that the input socket is changed.
|
|
|
|
Invalidates (recursively) the cache of any managed object or
|
|
output socket method that implicitly depends on this input socket.
|
|
"""
|
|
# Forwards Chains
|
|
if action == "value_changed":
|
|
# Run User Callbacks
|
|
## Careful with these, they run BEFORE propagation...
|
|
## ...because later-chain methods may rely on the results of this.
|
|
for method in self._on_value_changed_methods:
|
|
if (
|
|
socket_name
|
|
and socket_name in method._extra_data.get("changed_sockets")
|
|
) or (
|
|
prop_name
|
|
and prop_name in method._extra_data.get("changed_props")
|
|
) or (
|
|
socket_name
|
|
and method._extra_data["changed_loose_input"]
|
|
and socket_name in self.loose_input_sockets
|
|
):
|
|
method(self)
|
|
|
|
# Propagate via Output Sockets
|
|
for bl_socket in self.active_bl_sockets("output"):
|
|
bl_socket.trigger_action(action)
|
|
|
|
# Backwards Chains
|
|
elif action == "enable_lock":
|
|
self.locked = True
|
|
|
|
## Propagate via Input Sockets
|
|
for bl_socket in self.active_bl_sockets("input"):
|
|
bl_socket.trigger_action(action)
|
|
|
|
elif action == "disable_lock":
|
|
self.locked = False
|
|
|
|
## Propagate via Input Sockets
|
|
for bl_socket in self.active_bl_sockets("input"):
|
|
bl_socket.trigger_action(action)
|
|
|
|
elif action == "show_preview":
|
|
# Run User Callbacks
|
|
for method in self._on_show_preview:
|
|
method(self)
|
|
|
|
## Propagate via Input Sockets
|
|
for bl_socket in self.active_bl_sockets("input"):
|
|
bl_socket.trigger_action(action)
|
|
|
|
elif action == "show_plot":
|
|
# Run User Callbacks
|
|
## These shouldn't change any data, BUT...
|
|
## ...because they can stop propagation, they should go first.
|
|
for method in self._on_show_plot:
|
|
method(self)
|
|
if method._extra_data["stop_propagation"]:
|
|
return
|
|
|
|
## Propagate via Input Sockets
|
|
for bl_socket in self.active_bl_sockets("input"):
|
|
bl_socket.trigger_action(action)
|
|
|
|
####################
|
|
# - Blender Node Methods
|
|
####################
|
|
@classmethod
|
|
def poll(cls, node_tree: bpy.types.NodeTree) -> bool:
|
|
"""Run (by Blender) to determine instantiability.
|
|
|
|
Restricted to the MaxwellSimTreeType.
|
|
"""
|
|
|
|
return node_tree.bl_idname == ct.TreeType.MaxwellSim.value
|
|
|
|
def init(self, context: bpy.types.Context):
|
|
"""Run (by Blender) on node creation.
|
|
"""
|
|
global CACHE
|
|
|
|
# Initialize Cache and Instance ID
|
|
self.instance_id = str(uuid.uuid4())
|
|
CACHE[self.instance_id] = {}
|
|
|
|
# Initialize Name
|
|
self.sim_node_name = self.name
|
|
## Only shown in draw_buttons if 'self.use_sim_node_name'
|
|
|
|
# Initialize Sockets
|
|
self.sync_sockets()
|
|
|
|
# Apply Default Preset
|
|
if self.active_preset: self.sync_active_preset()
|
|
|
|
# Callbacks
|
|
for method in self._on_init:
|
|
method(self)
|
|
|
|
def update(self) -> None:
|
|
pass
|
|
|
|
def free(self) -> None:
|
|
"""Run (by Blender) when deleting the node.
|
|
"""
|
|
global CACHE
|
|
if not CACHE.get(self.instance_id):
|
|
CACHE[self.instance_id] = {}
|
|
node_tree = self.id_data
|
|
|
|
# Unlock
|
|
## This is one approach to the "deleted locked nodes" problem.
|
|
## Essentially, deleting a locked node will unlock along input chain.
|
|
## It also counts if any of the input sockets are linked and locked.
|
|
## Thus, we prevent "dangling locks".
|
|
## TODO: Don't even allow deleting a locked node.
|
|
if self.locked or any(
|
|
bl_socket.is_linked and bl_socket.locked
|
|
for bl_socket in self.inputs.values()
|
|
):
|
|
self.trigger_action("disable_lock")
|
|
|
|
# Free Managed Objects
|
|
for managed_obj in self.managed_objs.values():
|
|
managed_obj.free()
|
|
|
|
# Update NodeTree Caches
|
|
## The NodeTree keeps caches to for optimized event triggering.
|
|
## However, ex. deleted nodes also deletes links, without cache update.
|
|
## By reporting that we're deleting the node, the cache stays happy.
|
|
node_tree.sync_node_removed(self)
|
|
|
|
# Finally: Free Instance Cache
|
|
if self.instance_id in CACHE:
|
|
del CACHE[self.instance_id]
|
|
|
|
|
|
|
|
def chain_event_decorator(
|
|
callback_type: typ.Literal[
|
|
"computes_output_socket",
|
|
"on_value_changed",
|
|
"on_show_preview",
|
|
"on_show_plot",
|
|
"on_init",
|
|
],
|
|
index_by: typ.Any | None = None,
|
|
extra_data: dict[str, typ.Any] | None = None,
|
|
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(), ## For now, presume
|
|
output_sockets: set[str] = set(), ## For now, presume
|
|
loose_input_sockets: bool = False,
|
|
loose_output_sockets: bool = False,
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
|
|
req_params: set[str] = set()
|
|
):
|
|
def decorator(method: typ.Callable) -> typ.Callable:
|
|
# Check Function Signature Validity
|
|
func_sig = set(inspect.signature(method).parameters.keys())
|
|
|
|
## Too Little
|
|
if func_sig != req_params and func_sig.issubset(req_params):
|
|
msg = f"Decorated method {method.__name__} is missing arguments {req_params - func_sig}"
|
|
|
|
## Too Much
|
|
if func_sig != req_params and func_sig.issuperset(req_params):
|
|
msg = f"Decorated method {method.__name__} has superfluous arguments {func_sig - req_params}"
|
|
raise ValueError(msg)
|
|
|
|
## Just Right :)
|
|
|
|
# TODO: Check Function Annotation Validity
|
|
# - w/pydantic and/or socket capabilities
|
|
|
|
def decorated(node: MaxwellSimNode):
|
|
# Assemble Keyword Arguments
|
|
method_kw_args = {}
|
|
|
|
## Add Input Sockets
|
|
if input_sockets:
|
|
_input_sockets = {
|
|
input_socket_name: node._compute_input(input_socket_name, kind)
|
|
for input_socket_name in input_sockets
|
|
}
|
|
method_kw_args |= dict(input_sockets=_input_sockets)
|
|
|
|
## Add Output Sockets
|
|
if output_sockets:
|
|
_output_sockets = {
|
|
output_socket_name: node.compute_output(output_socket_name, kind)
|
|
for output_socket_name in output_sockets
|
|
}
|
|
method_kw_args |= dict(output_sockets=_output_sockets)
|
|
|
|
## Add Loose Sockets
|
|
if loose_input_sockets:
|
|
_loose_input_sockets = {
|
|
input_socket_name: node._compute_input(input_socket_name, kind)
|
|
for input_socket_name in node.loose_input_sockets
|
|
}
|
|
method_kw_args |= dict(
|
|
loose_input_sockets=_loose_input_sockets
|
|
)
|
|
if loose_output_sockets:
|
|
_loose_output_sockets = {
|
|
output_socket_name: node.compute_output(output_socket_name, kind)
|
|
for output_socket_name in node.loose_output_sockets
|
|
}
|
|
method_kw_args |= dict(
|
|
loose_output_sockets=_loose_output_sockets
|
|
)
|
|
|
|
## Add Props
|
|
if props:
|
|
_props = {
|
|
prop_name: getattr(node, prop_name)
|
|
for prop_name in props
|
|
}
|
|
method_kw_args |= dict(props=_props)
|
|
|
|
## Add Managed Object
|
|
if managed_objs:
|
|
_managed_objs = {
|
|
managed_obj_name: node.managed_objs[managed_obj_name]
|
|
for managed_obj_name in managed_objs
|
|
}
|
|
method_kw_args |= dict(managed_objs=_managed_objs)
|
|
|
|
# Call Method
|
|
return method(
|
|
node,
|
|
**method_kw_args,
|
|
)
|
|
|
|
# Set Attributes for Discovery
|
|
decorated._callback_type = callback_type
|
|
if index_by:
|
|
decorated._index_by = index_by
|
|
if extra_data:
|
|
decorated._extra_data = extra_data
|
|
|
|
return decorated
|
|
return decorator
|
|
|
|
|
|
####################
|
|
# - Decorator: Output Socket
|
|
####################
|
|
def computes_output_socket(
|
|
output_socket_name: ct.SocketName,
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(),
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
cacheable: bool = True,
|
|
):
|
|
"""Given a socket name, defines a function-that-makes-a-function (aka.
|
|
decorator) which has the name of the socket attached.
|
|
|
|
Must be used as a decorator, ex. `@compute_output_socket("name")`.
|
|
|
|
Args:
|
|
output_socket_name: The name of the output socket to attach the
|
|
decorated method to.
|
|
input_sockets: The values of these input sockets will be computed
|
|
using `_compute_input`, then passed to the decorated function
|
|
as `input_sockets: list[Any]`. If the input socket doesn't exist (ex. is contained in an inactive loose socket or socket set), then None is returned.
|
|
managed_objs: These managed objects will be passed to the
|
|
function as `managed_objs: list[Any]`.
|
|
kind: Requests for this `output_socket_name, DataFlowKind` pair will
|
|
be returned by the decorated function.
|
|
cacheable: The output of th
|
|
be returned by the decorated function.
|
|
|
|
Returns:
|
|
The decorator, which takes the output-socket-computing method
|
|
and returns a new output-socket-computing method, now annotated
|
|
and discoverable by the `MaxwellSimTreeNode`.
|
|
"""
|
|
req_params = {"self"} | (
|
|
{"input_sockets"} if input_sockets else set()
|
|
) | (
|
|
{"props"} if props else set()
|
|
) | (
|
|
{"managed_objs"} if managed_objs else set()
|
|
)
|
|
|
|
return chain_event_decorator(
|
|
callback_type="computes_output_socket",
|
|
index_by=(output_socket_name, kind),
|
|
kind=kind,
|
|
input_sockets=input_sockets,
|
|
props=props,
|
|
managed_objs=managed_objs,
|
|
req_params=req_params,
|
|
)
|
|
|
|
|
|
|
|
####################
|
|
# - Decorator: On Show Preview
|
|
####################
|
|
def on_value_changed(
|
|
socket_name: set[ct.SocketName] | ct.SocketName | None = None,
|
|
prop_name: set[str] | str | None = None,
|
|
any_loose_input_socket: bool = False,
|
|
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(),
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
):
|
|
if sum([
|
|
int(socket_name is not None),
|
|
int(prop_name is not None),
|
|
int(any_loose_input_socket),
|
|
]) > 1:
|
|
msg = "Define only one of socket_name, prop_name or any_loose_input_socket"
|
|
raise ValueError(msg)
|
|
|
|
req_params = {"self"} | (
|
|
{"input_sockets"} if input_sockets else set()
|
|
) | (
|
|
{"loose_input_sockets"} if any_loose_input_socket else set()
|
|
) | (
|
|
{"props"} if props else set()
|
|
) | (
|
|
{"managed_objs"} if managed_objs else set()
|
|
)
|
|
|
|
return chain_event_decorator(
|
|
callback_type="on_value_changed",
|
|
extra_data={
|
|
"changed_sockets": (
|
|
socket_name if isinstance(socket_name, set) else {socket_name}
|
|
),
|
|
"changed_props": (
|
|
prop_name if isinstance(prop_name, set) else {prop_name}
|
|
),
|
|
"changed_loose_input": any_loose_input_socket,
|
|
},
|
|
kind=kind,
|
|
input_sockets=input_sockets,
|
|
loose_input_sockets=any_loose_input_socket,
|
|
props=props,
|
|
managed_objs=managed_objs,
|
|
req_params=req_params,
|
|
)
|
|
|
|
def on_show_preview(
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(), ## For now, presume only same kind
|
|
output_sockets: set[str] = set(), ## For now, presume only same kind
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
):
|
|
req_params = {"self"} | (
|
|
{"input_sockets"} if input_sockets else set()
|
|
) | (
|
|
{"output_sockets"} if output_sockets else set()
|
|
) | (
|
|
{"props"} if props else set()
|
|
) | (
|
|
{"managed_objs"} if managed_objs else set()
|
|
)
|
|
|
|
return chain_event_decorator(
|
|
callback_type="on_show_preview",
|
|
kind=kind,
|
|
input_sockets=input_sockets,
|
|
output_sockets=output_sockets,
|
|
props=props,
|
|
managed_objs=managed_objs,
|
|
req_params=req_params,
|
|
)
|
|
|
|
def on_show_plot(
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(),
|
|
output_sockets: set[str] = set(),
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
stop_propagation: bool = False,
|
|
):
|
|
req_params = {"self"} | (
|
|
{"input_sockets"} if input_sockets else set()
|
|
) | (
|
|
{"output_sockets"} if output_sockets else set()
|
|
) | (
|
|
{"props"} if props else set()
|
|
) | (
|
|
{"managed_objs"} if managed_objs else set()
|
|
)
|
|
|
|
return chain_event_decorator(
|
|
callback_type="on_show_plot",
|
|
extra_data={
|
|
"stop_propagation": stop_propagation,
|
|
},
|
|
kind=kind,
|
|
input_sockets=input_sockets,
|
|
output_sockets=output_sockets,
|
|
props=props,
|
|
managed_objs=managed_objs,
|
|
req_params=req_params,
|
|
)
|
|
|
|
def on_init(
|
|
kind: ct.DataFlowKind = ct.DataFlowKind.Value,
|
|
input_sockets: set[str] = set(),
|
|
output_sockets: set[str] = set(),
|
|
props: set[str] = set(),
|
|
managed_objs: set[str] = set(),
|
|
):
|
|
req_params = {"self"} | (
|
|
{"input_sockets"} if input_sockets else set()
|
|
) | (
|
|
{"output_sockets"} if output_sockets else set()
|
|
) | (
|
|
{"props"} if props else set()
|
|
) | (
|
|
{"managed_objs"} if managed_objs else set()
|
|
)
|
|
|
|
return chain_event_decorator(
|
|
callback_type="on_init",
|
|
kind=kind,
|
|
input_sockets=input_sockets,
|
|
output_sockets=output_sockets,
|
|
props=props,
|
|
managed_objs=managed_objs,
|
|
req_params=req_params,
|
|
)
|