diff --git a/src/griptape_nodes/common/project_templates/default_project_template.py b/src/griptape_nodes/common/project_templates/default_project_template.py index a675e4354..a34b1db0b 100644 --- a/src/griptape_nodes/common/project_templates/default_project_template.py +++ b/src/griptape_nodes/common/project_templates/default_project_template.py @@ -164,6 +164,26 @@ ), fallback=BuiltInSituation.SAVE_FILE, ), + "save_output_directory": SituationTemplate( + name="save_output_directory", + description="Node creates and outputs a directory (versioned with _v001, _v002, …)", + macro="{outputs}/{sub_dirs?:/}{node_name?:_}{dir_name}_v{_index:03}", + policy=SituationPolicy( + on_collision=SituationFilePolicy.CREATE_NEW, + create_dirs=True, + ), + fallback=None, + ), + "save_file_sequence": SituationTemplate( + name="save_file_sequence", + description="Node writes a file sequence; version in directory and entry filenames", + macro="{outputs}/{sub_dirs?:/}{node_name?:_}{file_name_base}_v{_index:03}/{file_name_base}_v{_index:03}_{entry:04}.{file_extension}", + policy=SituationPolicy( + on_collision=SituationFilePolicy.CREATE_NEW, + create_dirs=True, + ), + fallback=None, + ), # Workflows save into the workspace root today for backward compatibility. # Migrating to a dedicated subdirectory is tracked in # https://github.com/griptape-ai/griptape-nodes/issues/2047. diff --git a/src/griptape_nodes/common/project_templates/situation_resolver.py b/src/griptape_nodes/common/project_templates/situation_resolver.py new file mode 100644 index 000000000..dffecf5f3 --- /dev/null +++ b/src/griptape_nodes/common/project_templates/situation_resolver.py @@ -0,0 +1,72 @@ +"""Resolve a project situation name into file-write configuration.""" + +import logging +import typing + +from griptape_nodes.common.project_templates import situation as situation_mod +from griptape_nodes.retained_mode import griptape_nodes as griptape_nodes_mod +from griptape_nodes.retained_mode.events import os_events, project_events + +logger = logging.getLogger("griptape_nodes") + +SITUATION_TO_FILE_POLICY: dict[str, os_events.ExistingFilePolicy] = { + situation_mod.SituationFilePolicy.CREATE_NEW: os_events.ExistingFilePolicy.CREATE_NEW, + situation_mod.SituationFilePolicy.OVERWRITE: os_events.ExistingFilePolicy.OVERWRITE, + situation_mod.SituationFilePolicy.FAIL: os_events.ExistingFilePolicy.FAIL, + situation_mod.SituationFilePolicy.PROMPT: os_events.ExistingFilePolicy.CREATE_NEW, # PROMPT has no direct mapping; fall back to CREATE_NEW +} + + +class ResolvedSituation(typing.NamedTuple): + """Result of looking up a project situation by name. + + Attributes: + macro_template: The macro template string for the situation. + existing_file_policy: Mapped file collision policy. + create_parents: Whether to create intermediate directories. + situation_obj: Raw situation template, or None when the lookup failed and + fallback values are in use. + """ + + macro_template: str + existing_file_policy: os_events.ExistingFilePolicy + create_parents: bool + situation_obj: situation_mod.SituationTemplate | None + + +def resolve_situation( + situation_name: str, + fallback_macro: str, + default_policy: os_events.ExistingFilePolicy = os_events.ExistingFilePolicy.CREATE_NEW, +) -> ResolvedSituation: + """Look up a situation by name and return its resolved configuration. + + Falls back to fallback_macro and default_policy when the situation cannot be loaded. + + Args: + situation_name: Situation name to look up in the current project. + fallback_macro: Macro template to use when the situation cannot be found. + default_policy: ExistingFilePolicy to use in the fallback case. + + Returns: + ResolvedSituation with macro_template, existing_file_policy, create_parents, + and situation_obj (None when falling back). + """ + result = griptape_nodes_mod.GriptapeNodes.handle_request( + project_events.GetSituationRequest(situation_name=situation_name) + ) + if isinstance(result, project_events.GetSituationResultSuccess): + situation_obj = result.situation + return ResolvedSituation( + macro_template=situation_obj.macro, + existing_file_policy=SITUATION_TO_FILE_POLICY.get(situation_obj.policy.on_collision, default_policy), + create_parents=situation_obj.policy.create_dirs, + situation_obj=situation_obj, + ) + logger.error("Failed to load situation '%s', using fallback macro template", situation_name) + return ResolvedSituation( + macro_template=fallback_macro, + existing_file_policy=default_policy, + create_parents=True, + situation_obj=None, + ) diff --git a/src/griptape_nodes/exe_types/param_components/project_directory_parameter.py b/src/griptape_nodes/exe_types/param_components/project_directory_parameter.py new file mode 100644 index 000000000..cd78ab025 --- /dev/null +++ b/src/griptape_nodes/exe_types/param_components/project_directory_parameter.py @@ -0,0 +1,139 @@ +"""ProjectDirectoryParameter - parameter component for project-aware directory creation. + +Provisions a versioned output directory via situation-based macro routing and exposes it as a +DirectoryDestination. Falls back to a sensible default when no situation is configured. +""" + +from griptape_nodes.common import macro_parser +from griptape_nodes.common.project_templates import situation_resolver +from griptape_nodes.exe_types import core_types, node_types +from griptape_nodes.exe_types.param_components import project_output_parameter +from griptape_nodes.files import directory as directory_mod +from griptape_nodes.retained_mode.events import project_events +from griptape_nodes.traits import file_system_picker + +_FALLBACK_DIRECTORY_MACRO = "{outputs}/{node_name?:_}{dir_name}_v{_index:03}" + + +class ProjectDirectoryParameter(project_output_parameter.ProjectOutputParameter): + """Parameter component for project-aware directory creation. + + Adds a directory name parameter to a node that, when processed, returns a + ``DirectoryDestination`` with a versioned macro path for deferred resolution. + + Usage: + # In node __init__: + self._dir_param = ProjectDirectoryParameter( + node=self, + name="output_dir", + default_dirname="renders", + ) + self._dir_param.add_parameter() + + # In node process(): + dest = self._dir_param.build_directory() + directory = dest.create() + self.set_parameter_value("output_dir", directory.location) + """ + + DEFAULT_SITUATION = "save_output_directory" + + def __init__( # noqa: PLR0913 + self, + node: node_types.BaseNode, + name: str, + *, + default_dirname: str, + situation: str = DEFAULT_SITUATION, + allowed_modes: set[core_types.ParameterMode] | None = None, + ui_options: dict | None = None, + ) -> None: + super().__init__( + node, + name, + default_value=default_dirname, + situation=situation, + allowed_modes=allowed_modes, + ui_options=ui_options, + ) + + @property + def _settings_node_type(self) -> str: + return "DirectoryOutputSettings" + + @property + def _settings_value_param_name(self) -> str: + return "dirname" + + @property + def _settings_source_param_name(self) -> str: + return "directory_destination" + + @property + def _parameter_output_type(self) -> str: + return "Directory" + + def _make_parameter_traits(self) -> set: + return { + file_system_picker.FileSystemPicker( + allow_files=False, + allow_directories=True, + allow_create=True, + ) + } + + def build_directory(self, **extra_vars: str | int) -> directory_mod.DirectoryDestination: + """Build a DirectoryDestination from the parameter's current value. + + If an upstream node exposes a ``directory_destination`` attribute, its + ``DirectoryDestination`` is retrieved directly. Otherwise the parameter's + string value is used as the directory name, combined with the situation macro. + + Args: + **extra_vars: Additional variables for the macro (e.g., sub_dirs="renders") + + Returns: + DirectoryDestination with a versioned MacroPath and baked-in policy. + + Raises: + ValueError: If an upstream node exposes ``directory_destination`` but returns None. + """ + upstream = self._get_upstream_destination("directory_destination", "DirectoryDestination") + if upstream is not None: + return upstream # type: ignore[return-value] + + value = self._node.get_parameter_value(self._name) + dirname = value if isinstance(value, str) and value else self._default_value + + if "node_name" not in extra_vars: + extra_vars["node_name"] = self._node.name + + return _build_directory_destination_from_situation(dirname, self._situation_name, **extra_vars) + + +def _build_directory_destination_from_situation( + dirname: str, + situation: str, + **extra_vars: str | int, +) -> directory_mod.DirectoryDestination: + """Build a DirectoryDestination from a project situation template. + + Args: + dirname: Directory name to use as the ``dir_name`` macro variable. + situation: Situation name to look up in the current project. + **extra_vars: Additional macro variables. + + Returns: + DirectoryDestination with a MacroPath and baked-in creation policy. + """ + resolved = situation_resolver.resolve_situation(situation, _FALLBACK_DIRECTORY_MACRO) + variables: dict[str, str | int] = { + "dir_name": dirname, + **extra_vars, + } + macro_path = project_events.MacroPath(macro_parser.ParsedMacro(resolved.macro_template), variables) + return directory_mod.DirectoryDestination( + macro_path, + existing_dir_policy=resolved.existing_file_policy, + create_parents=resolved.create_parents, + ) diff --git a/src/griptape_nodes/exe_types/param_components/project_file_parameter.py b/src/griptape_nodes/exe_types/param_components/project_file_parameter.py index ae95affe7..c62a3cd97 100644 --- a/src/griptape_nodes/exe_types/param_components/project_file_parameter.py +++ b/src/griptape_nodes/exe_types/param_components/project_file_parameter.py @@ -1,25 +1,19 @@ -"""ProjectFileParameter - parameter component for project-aware file saving.""" +"""ProjectFileParameter - parameter component for project-aware file saving. -import logging +Wraps a MacroPath into a FileDestination with a baked-in write policy, deferring +path resolution to execution time via the situation-based macro system. +""" from griptape_nodes.common.project_templates.situation import BuiltInSituation -from griptape_nodes.exe_types.core_types import NodeMessageResult, Parameter, ParameterMode +from griptape_nodes.exe_types.core_types import ParameterMode from griptape_nodes.exe_types.node_types import BaseNode -from griptape_nodes.files.file import FileDestination, FileDestinationProvider +from griptape_nodes.exe_types.param_components.project_output_parameter import ProjectOutputParameter +from griptape_nodes.files.file import FileDestination from griptape_nodes.files.project_file import ProjectFileDestination -from griptape_nodes.retained_mode.events.connection_events import ( - ListConnectionsForNodeRequest, - ListConnectionsForNodeResultSuccess, -) -from griptape_nodes.retained_mode.griptape_nodes import GriptapeNodes -from griptape_nodes.retained_mode.retained_mode import RetainedMode -from griptape_nodes.traits.button import Button, ButtonDetailsMessagePayload from griptape_nodes.traits.file_system_picker import FileSystemPicker -logger = logging.getLogger("griptape_nodes") - -class ProjectFileParameter: +class ProjectFileParameter(ProjectOutputParameter): """Parameter component for project-aware file saving. Adds a file path parameter to a node that, when processed, returns a @@ -51,28 +45,33 @@ def __init__( # noqa: PLR0913 allowed_modes: set[ParameterMode] | None = None, ui_options: dict | None = None, ) -> None: - """Initialize with situation context. + super().__init__( + node, + name, + default_value=default_filename, + situation=situation, + allowed_modes=allowed_modes, + ui_options=ui_options, + ) - Args: - node: Parent node instance - name: Parameter name - default_filename: Default filename if parameter is empty - situation: Situation name (default: "save_node_output") - allowed_modes: Set of allowed parameter modes (default: INPUT, PROPERTY) - ui_options: Optional UI options to pass to the generated parameter - """ - self._node = node - self._name = name - self._situation_name = situation - self._default_filename = default_filename - self._allowed_modes = allowed_modes or {ParameterMode.INPUT, ParameterMode.PROPERTY} - self._ui_options = ui_options - - def add_parameter(self) -> None: - """Create and add the file path parameter to the node.""" - tooltip = f"Output filename (uses '{self._situation_name}' situation template)" - - traits: set = { + @property + def _settings_node_type(self) -> str: + return "FileOutputSettings" + + @property + def _settings_value_param_name(self) -> str: + return "filename" + + @property + def _settings_source_param_name(self) -> str: + return "file_destination" + + @property + def _parameter_output_type(self) -> str: + return "str" + + def _make_parameter_traits(self) -> set: + return { FileSystemPicker( allow_files=True, allow_directories=False, @@ -80,43 +79,17 @@ def add_parameter(self) -> None: ) } - if ParameterMode.INPUT in self._allowed_modes: - traits.add( - Button( - icon="cog", - size="icon", - variant="secondary", - tooltip="Create and connect a FileOutputSettings node", - on_click=self._on_configure_button_clicked, - ) - ) - - parameter = Parameter( - name=self._name, - type="str", - default_value=self._default_filename, - allowed_modes=self._allowed_modes, - tooltip=tooltip, - input_types=["str"], - output_type="str", - traits=traits, - ui_options=self._ui_options, - ) - parameter.on_incoming_connection_removed.append(self._reset_to_default) - - self._node.add_parameter(parameter) - def build_file(self, **extra_vars: str | int) -> FileDestination: """Build a FileDestination with a MacroPath from the parameter's current value. - If an upstream node implements FileDestinationProvider (e.g., FileOutputSettings), - its FileDestination is retrieved directly without deserializing from the wire. - An upstream provider that returns None (misconfigured) raises instead of silently - falling back to the default situation, since that fallback hides user intent. + If an upstream node (e.g. FileOutputSettings) exposes a ``file_destination`` + attribute, its ``FileDestination`` is retrieved directly without deserialising + from the wire. A node that exposes the attribute but returns None raises instead + of silently falling back to the default situation, since that fallback hides + user intent. - Otherwise the parameter's string value is parsed into - file_name_base/file_extension, combined with this component's default - situation, and wrapped in a FileDestination. + Otherwise the parameter's string value is combined with this component's + situation to build a ``FileDestination``. Args: **extra_vars: Additional variables for the macro (e.g., sub_dirs="renders") @@ -125,30 +98,14 @@ def build_file(self, **extra_vars: str | int) -> FileDestination: FileDestination with a MacroPath and baked-in write policy for deferred path resolution Raises: - ValueError: If an upstream FileDestinationProvider is connected but returns None. + ValueError: If an upstream node exposes ``file_destination`` but returns None. """ - result = GriptapeNodes.handle_request(ListConnectionsForNodeRequest(node_name=self._node.name)) - if isinstance(result, ListConnectionsForNodeResultSuccess): - for conn in result.incoming_connections: - if conn.target_parameter_name == self._name: - source_node = GriptapeNodes.ObjectManager().attempt_get_object_by_name(conn.source_node_name) - if isinstance(source_node, FileDestinationProvider): - file_dest = source_node.file_destination - if file_dest is None: - msg = ( - f"Attempted to build file destination for {self._node.name}.{self._name}. " - f"Failed because upstream node '{conn.source_node_name}' provides a " - f"FileDestination but returned None (likely missing a filename)." - ) - raise ValueError(msg) - return file_dest + upstream = self._get_upstream_destination("file_destination", "FileDestination") + if upstream is not None: + return upstream # type: ignore[return-value] value = self._node.get_parameter_value(self._name) - - if isinstance(value, str) and value: - filename = value - else: - filename = self._default_filename + filename = value if isinstance(value, str) and value else self._default_value if "node_name" not in extra_vars: extra_vars["node_name"] = self._node.name @@ -158,79 +115,3 @@ def build_file(self, **extra_vars: str | int) -> FileDestination: self._situation_name, **extra_vars, ) - - def _reset_to_default(self, parameter: Parameter, source_node_name: str, source_parameter_name: str) -> None: # noqa: ARG002 - self._node.set_parameter_value(self._name, self._default_filename) - self._node.publish_update_to_parameter(self._name, self._default_filename) - - def _on_configure_button_clicked( - self, - button: Button, # noqa: ARG002 - button_details: ButtonDetailsMessagePayload, - ) -> NodeMessageResult: - """Create and connect a FileOutputSettings node to this parameter.""" - node_name = self._node.name - - has_incoming = False - result = GriptapeNodes.handle_request(ListConnectionsForNodeRequest(node_name=node_name)) - if isinstance(result, ListConnectionsForNodeResultSuccess): - has_incoming = any(conn.target_parameter_name == self._name for conn in result.incoming_connections) - - if has_incoming: - return NodeMessageResult( - success=False, - details=f"{node_name}: {self._name} parameter already has an incoming connection", - response=button_details, - altered_workflow_state=False, - ) - - # TODO: https://github.com/griptape-ai/griptape-nodes/issues/4097 - # Replace with a non-RM utility for creating sibling nodes relative to a given node. - create_result = RetainedMode.create_node_relative_to( - reference_node_name=node_name, - new_node_type="FileOutputSettings", - offset_side="left", - offset_x=-750, - offset_y=0, - lock=False, - ) - - if not isinstance(create_result, str): - return NodeMessageResult( - success=False, - details=f"{node_name}: Failed to create FileOutputSettings node", - response=button_details, - altered_workflow_state=False, - ) - - configure_node_name = create_result - - configure_node = GriptapeNodes.ObjectManager().attempt_get_object_by_name(configure_node_name) - if configure_node is not None: - configure_node.set_parameter_value("situation", self._situation_name) - configure_node.publish_update_to_parameter("situation", self._situation_name) - - current_filename = self._node.get_parameter_value(self._name) - if isinstance(current_filename, str) and current_filename: - configure_node.set_parameter_value("filename", current_filename) - configure_node.publish_update_to_parameter("filename", current_filename) - - connection_result = RetainedMode.connect( - source=f"{configure_node_name}.file_destination", - destination=f"{node_name}.{self._name}", - ) - - if not connection_result.succeeded(): - return NodeMessageResult( - success=False, - details=f"{node_name}: Failed to connect {configure_node_name}.file_destination to {self._name}", - response=button_details, - altered_workflow_state=True, - ) - - return NodeMessageResult( - success=True, - details=f"{node_name}: Created and connected {configure_node_name}", - response=button_details, - altered_workflow_state=True, - ) diff --git a/src/griptape_nodes/exe_types/param_components/project_file_sequence_parameter.py b/src/griptape_nodes/exe_types/param_components/project_file_sequence_parameter.py new file mode 100644 index 000000000..e6c96ea3d --- /dev/null +++ b/src/griptape_nodes/exe_types/param_components/project_file_sequence_parameter.py @@ -0,0 +1,147 @@ +"""ProjectFileSequenceParameter - parameter component for project-aware file sequence output. + +Provisions a versioned output directory and exposes a FileSequenceDestination whose entry() +method returns a per-frame FileDestination. Situation-based macro routing determines the +directory layout; falls back to a sensible default when no situation is configured. +""" + +from griptape_nodes.common import macro_parser +from griptape_nodes.common.project_templates import situation_resolver +from griptape_nodes.exe_types import core_types, node_types +from griptape_nodes.exe_types.param_components import project_output_parameter +from griptape_nodes.files import file_sequence, path_utils +from griptape_nodes.retained_mode.events import os_events, project_events + +_FALLBACK_SEQUENCE_MACRO = ( + "{outputs}/{node_name?:_}{file_name_base}_v{_index:03}/{file_name_base}_v{_index:03}_{entry:04}.{file_extension}" +) + + +class ProjectFileSequenceParameter(project_output_parameter.ProjectOutputParameter): + """Parameter component for project-aware file sequence output. + + Adds a filename-pattern parameter to a node that, when processed, returns a + ``FileSequenceDestination`` with a versioned macro path for deferred resolution. + + The parameter accepts a filename like ``"render.exr"`` or a ``####`` pattern + like ``"render_####.exr"``. The situation macro wraps it with versioning. + + Usage: + # In node __init__: + self._seq_param = ProjectFileSequenceParameter( + node=self, + name="output_sequence", + default_filename="render.exr", + ) + self._seq_param.add_parameter() + + # In node process(): + dest = self._seq_param.build_sequence() + for i, entry_data in enumerate(entries): + dest.entry(i + 1).write_bytes(entry_data) + seq = dest.file_sequence + if seq is not None: + self.set_parameter_value("output_sequence", seq.location) + """ + + DEFAULT_SITUATION = "save_file_sequence" + + def __init__( # noqa: PLR0913 + self, + node: node_types.BaseNode, + name: str, + *, + default_filename: str, + situation: str = DEFAULT_SITUATION, + allowed_modes: set[core_types.ParameterMode] | None = None, + ui_options: dict | None = None, + ) -> None: + super().__init__( + node, + name, + default_value=default_filename, + situation=situation, + allowed_modes=allowed_modes, + ui_options=ui_options, + ) + + @property + def _settings_node_type(self) -> str: + return "FileSequenceSettings" + + @property + def _settings_value_param_name(self) -> str: + return "filename" + + @property + def _settings_source_param_name(self) -> str: + return "sequence_destination" + + @property + def _parameter_output_type(self) -> str: + return "FileSequence" + + def build_sequence(self, **extra_vars: str | int) -> file_sequence.FileSequenceDestination: + """Build a FileSequenceDestination from the parameter's current value. + + If an upstream node exposes a ``file_sequence_destination`` attribute, its + ``FileSequenceDestination`` is retrieved directly. Otherwise the parameter's + string value (filename or #### pattern) is parsed and combined with the + situation macro. + + Args: + **extra_vars: Additional variables for the macro (e.g., sub_dirs="renders") + + Returns: + FileSequenceDestination with a versioned MacroPath and baked-in policy. + + Raises: + ValueError: If an upstream node exposes ``file_sequence_destination`` but returns None. + FileSequenceError: If no available version index can be found. + """ + upstream = self._get_upstream_destination("file_sequence_destination", "FileSequenceDestination") + if upstream is not None: + return upstream # type: ignore[return-value] + + value = self._node.get_parameter_value(self._name) + filename = value if isinstance(value, str) and value else self._default_value + + if "node_name" not in extra_vars: + extra_vars["node_name"] = self._node.name + + return _build_sequence_destination_from_situation(filename, self._situation_name, **extra_vars) + + +def _build_sequence_destination_from_situation( + filename: str, + situation: str, + **extra_vars: str | int, +) -> file_sequence.FileSequenceDestination: + """Build a FileSequenceDestination from a project situation template. + + Parses the filename (or #### pattern) into parts, looks up the situation, + and builds a versioned destination by updating all available ``_index`` variables. + + Args: + filename: Filename or #### pattern (e.g., ``"render.exr"`` or ``"render_####.exr"``). + situation: Situation name to look up in the current project. + **extra_vars: Additional macro variables. + + Returns: + FileSequenceDestination with a locked version index but unresolved element token. + """ + resolved = situation_resolver.resolve_situation( + situation, _FALLBACK_SEQUENCE_MACRO, os_events.ExistingFilePolicy.OVERWRITE + ) + parts = path_utils.FilenameParts.from_filename(filename) + variables: dict[str, str | int] = { + "file_name_base": parts.stem, + "file_extension": parts.extension, + **extra_vars, + } + macro_path = project_events.MacroPath(macro_parser.ParsedMacro(resolved.macro_template), variables) + return file_sequence.build_versioned_sequence_destination( + macro_path, + existing_file_policy=resolved.existing_file_policy, + create_parents=resolved.create_parents, + ) diff --git a/src/griptape_nodes/exe_types/param_components/project_output_parameter.py b/src/griptape_nodes/exe_types/param_components/project_output_parameter.py new file mode 100644 index 000000000..73cb10962 --- /dev/null +++ b/src/griptape_nodes/exe_types/param_components/project_output_parameter.py @@ -0,0 +1,248 @@ +"""ProjectOutputParameter - shared base class for project-aware output parameter components.""" + +from __future__ import annotations + +import abc +import logging +import typing + +from griptape_nodes.exe_types import core_types +from griptape_nodes.retained_mode import griptape_nodes as griptape_nodes_mod +from griptape_nodes.retained_mode import retained_mode as retained_mode_mod +from griptape_nodes.retained_mode.events import connection_events +from griptape_nodes.traits import button as button_mod + +if typing.TYPE_CHECKING: + from griptape_nodes.exe_types import node_types + +logger = logging.getLogger("griptape_nodes") + + +class ProjectOutputParameter(abc.ABC): + """Shared base for project-aware output parameter components. + + Handles the cog-button pattern (create + connect a settings node) and + upstream provider resolution, which are identical across all output types. + Subclasses supply the output-type-specific behaviour via abstract properties + and implement the ``build_xxx()`` method that returns the concrete destination. + """ + + def __init__( # noqa: PLR0913 + self, + node: node_types.BaseNode, + name: str, + *, + default_value: str, + situation: str, + allowed_modes: set[core_types.ParameterMode] | None = None, + ui_options: dict | None = None, + ) -> None: + """Initialise with situation context. + + Args: + node: Parent node instance. + name: Parameter name. + default_value: Default value when the parameter is empty. + situation: Situation name used to build the output path macro. + allowed_modes: Allowed parameter modes (default: INPUT, PROPERTY). + ui_options: Optional UI options forwarded to the generated parameter. + """ + self._node = node + self._name = name + self._situation_name = situation + self._default_value = default_value + self._allowed_modes = allowed_modes or {core_types.ParameterMode.INPUT, core_types.ParameterMode.PROPERTY} + self._ui_options = ui_options + + # ---- Abstract pieces each subclass must supply ---- + + @property + @abc.abstractmethod + def _settings_node_type(self) -> str: + """Node type to create when the cog button is clicked (e.g. 'FileOutputSettings').""" + + @property + @abc.abstractmethod + def _settings_value_param_name(self) -> str: + """Parameter name on the settings node that holds the filename/dirname (e.g. 'filename').""" + + @property + @abc.abstractmethod + def _settings_source_param_name(self) -> str: + """Output parameter on the settings node to wire to this parameter (e.g. 'file_destination').""" + + @property + @abc.abstractmethod + def _parameter_output_type(self) -> str: + """The output_type string for the generated parameter (e.g. 'str', 'Directory').""" + + # ---- Overridable hooks ---- + + def _make_parameter_traits(self) -> set: + """Return additional traits for the parameter (e.g. a FileSystemPicker). + + Returns an empty set by default; subclasses override to add pickers. + """ + return set() + + # ---- Shared concrete logic ---- + + def add_parameter(self) -> None: + """Create and add the output parameter to the node.""" + tooltip = f"Output path (uses '{self._situation_name}' situation template)" + + traits = self._make_parameter_traits() + if core_types.ParameterMode.INPUT in self._allowed_modes: + traits.add( + button_mod.Button( + icon="cog", + size="icon", + variant="secondary", + tooltip=f"Create and connect a {self._settings_node_type} node", + on_click=self._on_configure_button_clicked, + ) + ) + + parameter = core_types.Parameter( + name=self._name, + type="str", + default_value=self._default_value, + allowed_modes=self._allowed_modes, + tooltip=tooltip, + input_types=["str"], + output_type=self._parameter_output_type, + traits=traits, + ui_options=self._ui_options, + ) + parameter.on_incoming_connection_removed.append(self._reset_to_default) + + self._node.add_parameter(parameter) + + def _reset_to_default( + self, + parameter: core_types.Parameter, # noqa: ARG002 + source_node_name: str, # noqa: ARG002 + source_parameter_name: str, # noqa: ARG002 + ) -> None: + self._node.set_parameter_value(self._name, self._default_value) + self._node.publish_update_to_parameter(self._name, self._default_value) + + def _get_upstream_destination( + self, + destination_attr: str, + destination_type_name: str, + ) -> object | None: + """Return the destination from the first upstream node that exposes ``destination_attr``. + + Returns None if no connected node has the attribute. + + Args: + destination_attr: Attribute name to look for on the upstream node + (e.g. ``'file_destination'``). + destination_type_name: Human-readable type name used in error messages + (e.g. ``'FileDestination'``). + + Raises: + ValueError: If a connected node exposes ``destination_attr`` but returns None. + """ + result = griptape_nodes_mod.GriptapeNodes.handle_request( + connection_events.ListConnectionsForNodeRequest(node_name=self._node.name) + ) + if not isinstance(result, connection_events.ListConnectionsForNodeResultSuccess): + return None + + for conn in result.incoming_connections: + if conn.target_parameter_name != self._name: + continue + source_node = griptape_nodes_mod.GriptapeNodes.ObjectManager().attempt_get_object_by_name( + conn.source_node_name + ) + if source_node is None or not hasattr(source_node, destination_attr): + continue + destination = getattr(source_node, destination_attr) + if destination is None: + msg = ( + f"Attempted to build {destination_type_name} for {self._node.name}.{self._name}. " + f"Failed because upstream node '{conn.source_node_name}' returned None " + f"(likely missing a filename or path)." + ) + raise ValueError(msg) + return destination + + return None + + def _on_configure_button_clicked( + self, + button: button_mod.Button, # noqa: ARG002 + button_details: button_mod.ButtonDetailsMessagePayload, + ) -> core_types.NodeMessageResult: + """Create and connect the appropriate settings node to this parameter.""" + node_name = self._node.name + + has_incoming = False + result = griptape_nodes_mod.GriptapeNodes.handle_request( + connection_events.ListConnectionsForNodeRequest(node_name=node_name) + ) + if isinstance(result, connection_events.ListConnectionsForNodeResultSuccess): + has_incoming = any(conn.target_parameter_name == self._name for conn in result.incoming_connections) + + if has_incoming: + return core_types.NodeMessageResult( + success=False, + details=f"{node_name}: {self._name} parameter already has an incoming connection", + response=button_details, + altered_workflow_state=False, + ) + + # TODO: https://github.com/griptape-ai/griptape-nodes/issues/4097 + # Replace with a non-RM utility for creating sibling nodes relative to a given node. + create_result = retained_mode_mod.RetainedMode.create_node_relative_to( + reference_node_name=node_name, + new_node_type=self._settings_node_type, + offset_side="left", + offset_x=-750, + offset_y=0, + lock=False, + ) + + if not isinstance(create_result, str): + return core_types.NodeMessageResult( + success=False, + details=f"{node_name}: Failed to create {self._settings_node_type} node", + response=button_details, + altered_workflow_state=False, + ) + + configure_node_name = create_result + + configure_node = griptape_nodes_mod.GriptapeNodes.ObjectManager().attempt_get_object_by_name( + configure_node_name + ) + if configure_node is not None: + configure_node.set_parameter_value("situation", self._situation_name) + configure_node.publish_update_to_parameter("situation", self._situation_name) + + current_value = self._node.get_parameter_value(self._name) + if isinstance(current_value, str) and current_value: + configure_node.set_parameter_value(self._settings_value_param_name, current_value) + configure_node.publish_update_to_parameter(self._settings_value_param_name, current_value) + + connection_result = retained_mode_mod.RetainedMode.connect( + source=f"{configure_node_name}.{self._settings_source_param_name}", + destination=f"{node_name}.{self._name}", + ) + + if not connection_result.succeeded(): + return core_types.NodeMessageResult( + success=False, + details=f"{node_name}: Failed to connect {configure_node_name}.{self._settings_source_param_name} to {self._name}", + response=button_details, + altered_workflow_state=True, + ) + + return core_types.NodeMessageResult( + success=True, + details=f"{node_name}: Created and connected {configure_node_name}", + response=button_details, + altered_workflow_state=True, + ) diff --git a/src/griptape_nodes/files/project_file.py b/src/griptape_nodes/files/project_file.py index 71afcc62a..fa75f5d35 100644 --- a/src/griptape_nodes/files/project_file.py +++ b/src/griptape_nodes/files/project_file.py @@ -4,15 +4,12 @@ from pathlib import Path from griptape_nodes.common.macro_parser import ParsedMacro -from griptape_nodes.common.project_templates.situation import SituationFilePolicy +from griptape_nodes.common.project_templates.situation_resolver import resolve_situation from griptape_nodes.files.file import File, FileDestination from griptape_nodes.files.path_utils import FilenameParts -from griptape_nodes.retained_mode.events.os_events import ExistingFilePolicy from griptape_nodes.retained_mode.events.project_events import ( AttemptMapAbsolutePathToProjectRequest, AttemptMapAbsolutePathToProjectResultSuccess, - GetSituationRequest, - GetSituationResultSuccess, MacroPath, ) from griptape_nodes.retained_mode.file_metadata.sidecar_metadata import ( @@ -27,12 +24,12 @@ FALLBACK_MACRO_TEMPLATE = "{outputs}/{node_name?:_}{file_name_base}{_index?:03}.{file_extension}" -SITUATION_TO_FILE_POLICY: dict[str, ExistingFilePolicy] = { - SituationFilePolicy.CREATE_NEW: ExistingFilePolicy.CREATE_NEW, - SituationFilePolicy.OVERWRITE: ExistingFilePolicy.OVERWRITE, - SituationFilePolicy.FAIL: ExistingFilePolicy.FAIL, - SituationFilePolicy.PROMPT: ExistingFilePolicy.CREATE_NEW, # PROMPT has no direct mapping; fall back to CREATE_NEW -} +def _attempt_map_to_project(absolute_path: Path) -> str | None: + """Fire AttemptMapAbsolutePathToProjectRequest; return the mapped macro path string or None.""" + map_result = GriptapeNodes.handle_request(AttemptMapAbsolutePathToProjectRequest(absolute_path=absolute_path)) + if isinstance(map_result, AttemptMapAbsolutePathToProjectResultSuccess) and map_result.mapped_path is not None: + return map_result.mapped_path + return None def _attempt_map_to_project(absolute_path: Path) -> str | None: @@ -79,11 +76,9 @@ def _map_to_macro_file(self, result_file: File) -> File: portable reference via ``file.as_macro()``. Falls back to the original File (absolute path) if mapping is not possible. """ - map_result = GriptapeNodes.handle_request( - AttemptMapAbsolutePathToProjectRequest(absolute_path=Path(result_file.resolve())) - ) - if isinstance(map_result, AttemptMapAbsolutePathToProjectResultSuccess) and map_result.mapped_path is not None: - return File(map_result.mapped_path) + mapped = _attempt_map_to_project(Path(result_file.resolve())) + if mapped is not None: + return File(mapped) return result_file @classmethod @@ -106,20 +101,10 @@ def from_situation( situation: Situation name to look up in the current project. **extra_vars: Additional macro variables (e.g., node_name="MyNode", _index=1). """ - result = GriptapeNodes.handle_request(GetSituationRequest(situation_name=situation)) - - if isinstance(result, GetSituationResultSuccess): - situation_obj = result.situation - macro_template = situation_obj.macro - on_collision = situation_obj.policy.on_collision - existing_file_policy = SITUATION_TO_FILE_POLICY.get(on_collision, ExistingFilePolicy.CREATE_NEW) - create_dirs = situation_obj.policy.create_dirs - else: - logger.error("Failed to load situation '%s', using fallback macro template", situation) - situation_obj = None - macro_template = FALLBACK_MACRO_TEMPLATE - existing_file_policy = ExistingFilePolicy.CREATE_NEW - create_dirs = True + resolved = resolve_situation(situation, FALLBACK_MACRO_TEMPLATE) + situation_obj = resolved.situation_obj + existing_file_policy = resolved.existing_file_policy + create_dirs = resolved.create_parents parts = FilenameParts.from_filename(filename) variables: dict[str, str | int] = { @@ -142,7 +127,7 @@ def from_situation( # caller-supplied variables here. The sidecar records the raw inputs; # anyone re-resolving the path against the current project gets the # same derived values the write used. - macro_path = MacroPath(ParsedMacro(macro_template), variables) + macro_path = MacroPath(ParsedMacro(resolved.macro_template), variables) file_metadata = ( SidecarContent( diff --git a/tests/unit/exe_types/param_components/test_project_directory_parameter.py b/tests/unit/exe_types/param_components/test_project_directory_parameter.py new file mode 100644 index 000000000..e445afc95 --- /dev/null +++ b/tests/unit/exe_types/param_components/test_project_directory_parameter.py @@ -0,0 +1,163 @@ +"""Unit tests for _build_directory_destination_from_situation.""" + +from unittest import mock + +from griptape_nodes.common.project_templates import situation +from griptape_nodes.exe_types.param_components import project_directory_parameter +from griptape_nodes.files import directory as directory_mod +from griptape_nodes.retained_mode.events import os_events, project_events + +HANDLE_REQUEST_PATH = "griptape_nodes.files.project_file.GriptapeNodes.handle_request" + +_POLICY_MAP = { + "CREATE_NEW": situation.SituationFilePolicy.CREATE_NEW, + "OVERWRITE": situation.SituationFilePolicy.OVERWRITE, + "FAIL": situation.SituationFilePolicy.FAIL, +} + + +def _make_situation( + macro: str, + on_collision: str = "CREATE_NEW", + *, + create_dirs: bool = True, +) -> situation.SituationTemplate: + return situation.SituationTemplate( + name="test_situation", + macro=macro, + policy=situation.SituationPolicy(on_collision=_POLICY_MAP[on_collision], create_dirs=create_dirs), + ) + + +class TestBuildDirectoryDestinationFromSituation: + """Tests for _build_directory_destination_from_situation helper.""" + + def test_uses_situation_macro(self) -> None: + sit = _make_situation("{outputs}/{node_name}/{dir_name}_v{_index:03}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert isinstance(dest, directory_mod.DirectoryDestination) + assert isinstance(dest._dir_path, project_events.MacroPath) + assert dest._dir_path.parsed_macro.template == "{outputs}/{node_name}/{dir_name}_v{_index:03}" + + def test_falls_back_to_default_macro_when_situation_not_found(self) -> None: + failure = project_events.GetSituationResultFailure(result_details="not found") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=failure): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "missing_situation" + ) + + assert isinstance(dest._dir_path, project_events.MacroPath) + assert dest._dir_path.parsed_macro.template == project_directory_parameter._FALLBACK_DIRECTORY_MACRO + + def test_wires_dirname_as_macro_variable(self) -> None: + sit = _make_situation("{outputs}/{dir_name}_v{_index:03}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "frames", "save_output_directory" + ) + + assert isinstance(dest._dir_path, project_events.MacroPath) + assert dest._dir_path.variables["dir_name"] == "frames" + + def test_extra_vars_forwarded_to_macro(self) -> None: + sit = _make_situation("{outputs}/{node_name}/{dir_name}_v{_index:03}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory", node_name="MyNode" + ) + + assert isinstance(dest._dir_path, project_events.MacroPath) + assert dest._dir_path.variables["node_name"] == "MyNode" + + def test_situation_overwrite_policy_maps_to_overwrite(self) -> None: + sit = _make_situation("{outputs}/{dir_name}", on_collision="OVERWRITE") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert dest._existing_dir_policy == os_events.ExistingFilePolicy.OVERWRITE + + def test_situation_create_new_policy_maps_to_create_new(self) -> None: + sit = _make_situation("{outputs}/{dir_name}", on_collision="CREATE_NEW") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert dest._existing_dir_policy == os_events.ExistingFilePolicy.CREATE_NEW + + def test_situation_fail_policy_maps_to_fail(self) -> None: + sit = _make_situation("{outputs}/{dir_name}", on_collision="FAIL") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert dest._existing_dir_policy == os_events.ExistingFilePolicy.FAIL + + def test_situation_create_dirs_false_propagated(self) -> None: + sit = _make_situation("{outputs}/{dir_name}", create_dirs=False) + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert dest._create_parents is False + + def test_fallback_uses_create_new_policy(self) -> None: + failure = project_events.GetSituationResultFailure(result_details="not found") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=failure): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "missing_situation" + ) + + assert dest._existing_dir_policy == os_events.ExistingFilePolicy.CREATE_NEW + + def test_multiple_extra_vars_all_forwarded(self) -> None: + sit = _make_situation("{outputs}/{dir_name}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", + "save_output_directory", + node_name="MyNode", + sub_dirs="pass_1", + ) + + assert isinstance(dest._dir_path, project_events.MacroPath) + assert dest._dir_path.variables["node_name"] == "MyNode" + assert dest._dir_path.variables["sub_dirs"] == "pass_1" + assert dest._dir_path.variables["dir_name"] == "renders" + + def test_returns_directory_destination(self) -> None: + sit = _make_situation("{outputs}/{dir_name}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + dest = project_directory_parameter._build_directory_destination_from_situation( + "renders", "save_output_directory" + ) + + assert isinstance(dest, directory_mod.DirectoryDestination) diff --git a/tests/unit/exe_types/param_components/test_project_file_sequence_parameter.py b/tests/unit/exe_types/param_components/test_project_file_sequence_parameter.py new file mode 100644 index 000000000..d2e269bf5 --- /dev/null +++ b/tests/unit/exe_types/param_components/test_project_file_sequence_parameter.py @@ -0,0 +1,199 @@ +"""Unit tests for _build_sequence_destination_from_situation.""" + +from unittest import mock + +from griptape_nodes.common.project_templates import situation +from griptape_nodes.exe_types.param_components import project_file_sequence_parameter +from griptape_nodes.files import file_sequence +from griptape_nodes.retained_mode.events import os_events, project_events + +HANDLE_REQUEST_PATH = "griptape_nodes.files.project_file.GriptapeNodes.handle_request" +BUILD_VERSIONED_PATH = "griptape_nodes.files.file_sequence.build_versioned_sequence_destination" + +_POLICY_MAP = { + "CREATE_NEW": situation.SituationFilePolicy.CREATE_NEW, + "OVERWRITE": situation.SituationFilePolicy.OVERWRITE, + "FAIL": situation.SituationFilePolicy.FAIL, +} + + +def _make_situation( + macro: str, + on_collision: str = "OVERWRITE", + *, + create_dirs: bool = True, +) -> situation.SituationTemplate: + return situation.SituationTemplate( + name="test_situation", + macro=macro, + policy=situation.SituationPolicy(on_collision=_POLICY_MAP[on_collision], create_dirs=create_dirs), + ) + + +class TestBuildSequenceDestinationFromSituation: + """Tests for _build_sequence_destination_from_situation helper.""" + + def test_uses_situation_macro(self) -> None: + situation_macro = ( + "{outputs}/{node_name?:_}{file_name_base}_v{_index:03}/{file_name_base}_{entry:04}.{file_extension}" + ) + sit = _make_situation(situation_macro) + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence" + ) + + call_args = mock_build.call_args + macro_path = call_args.args[0] + assert macro_path.parsed_macro.template == situation_macro + + def test_falls_back_to_default_macro_when_situation_not_found(self) -> None: + failure = project_events.GetSituationResultFailure(result_details="not found") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=failure), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation("frame.exr", "missing_situation") + + call_args = mock_build.call_args + macro_path = call_args.args[0] + assert macro_path.parsed_macro.template == project_file_sequence_parameter._FALLBACK_SEQUENCE_MACRO + + def test_plain_filename_parsed_into_stem_and_extension(self) -> None: + sit = _make_situation("{outputs}/{file_name_base}_{entry:04}.{file_extension}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence" + ) + + macro_path = mock_build.call_args.args[0] + assert macro_path.variables["file_name_base"] == "frame" + assert macro_path.variables["file_extension"] == "exr" + + def test_hash_pattern_filename_converted_before_parsing(self) -> None: + sit = _make_situation("{outputs}/{file_name_base}_{entry:04}.{file_extension}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame_####.exr", "save_file_sequence" + ) + + macro_path = mock_build.call_args.args[0] + assert macro_path.variables["file_extension"] == "exr" + + def test_extra_vars_forwarded_to_macro(self) -> None: + sit = _make_situation("{outputs}/{node_name}/{file_name_base}_{entry:04}.{file_extension}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence", node_name="MyNode" + ) + + macro_path = mock_build.call_args.args[0] + assert macro_path.variables["node_name"] == "MyNode" + + def test_situation_overwrite_policy_forwarded(self) -> None: + sit = _make_situation("{outputs}/{entry:04}.exr", on_collision="OVERWRITE") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence" + ) + + call_kwargs = mock_build.call_args.kwargs + assert call_kwargs["existing_file_policy"] == os_events.ExistingFilePolicy.OVERWRITE + + def test_situation_create_dirs_forwarded(self) -> None: + sit = _make_situation("{outputs}/{entry:04}.exr", create_dirs=False) + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence" + ) + + call_kwargs = mock_build.call_args.kwargs + assert call_kwargs["create_parents"] is False + + def test_fallback_uses_overwrite_policy(self) -> None: + failure = project_events.GetSituationResultFailure(result_details="not found") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=failure), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation("frame.exr", "missing_situation") + + call_kwargs = mock_build.call_args.kwargs + assert call_kwargs["existing_file_policy"] == os_events.ExistingFilePolicy.OVERWRITE + + def test_returns_file_sequence_destination(self) -> None: + sit = _make_situation("{outputs}/{entry:04}.exr") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest), + ): + result = project_file_sequence_parameter._build_sequence_destination_from_situation( + "frame.exr", "save_file_sequence" + ) + + assert result is mock_dest + + def test_multiple_extra_vars_all_forwarded(self) -> None: + sit = _make_situation("{outputs}/{file_name_base}_{entry:04}.{file_extension}") + success = project_events.GetSituationResultSuccess(situation=sit, result_details="ok") + mock_dest = mock.MagicMock(spec=file_sequence.FileSequenceDestination) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(BUILD_VERSIONED_PATH, return_value=mock_dest) as mock_build, + ): + project_file_sequence_parameter._build_sequence_destination_from_situation( + "render.exr", + "save_file_sequence", + node_name="Renderer", + sub_dirs="pass_1", + ) + + macro_path = mock_build.call_args.args[0] + assert macro_path.variables["node_name"] == "Renderer" + assert macro_path.variables["sub_dirs"] == "pass_1" + assert macro_path.variables["file_name_base"] == "render" + assert macro_path.variables["file_extension"] == "exr" diff --git a/tests/unit/exe_types/param_components/test_project_output_parameter.py b/tests/unit/exe_types/param_components/test_project_output_parameter.py new file mode 100644 index 000000000..c85b56128 --- /dev/null +++ b/tests/unit/exe_types/param_components/test_project_output_parameter.py @@ -0,0 +1,191 @@ +"""Unit tests for ProjectOutputParameter._get_upstream_destination.""" + +from unittest import mock + +import pytest + +from griptape_nodes.exe_types import core_types +from griptape_nodes.exe_types.param_components import project_output_parameter +from griptape_nodes.retained_mode.events import connection_events + +HANDLE_REQUEST_PATH = "griptape_nodes.retained_mode.griptape_nodes.GriptapeNodes.handle_request" +OBJECT_MANAGER_PATH = "griptape_nodes.retained_mode.griptape_nodes.GriptapeNodes.ObjectManager" + + +class _ConcreteParam(project_output_parameter.ProjectOutputParameter): + """Minimal concrete subclass for testing the base class.""" + + @property + def _settings_node_type(self) -> str: + return "TestSettings" + + @property + def _settings_value_param_name(self) -> str: + return "value" + + @property + def _settings_source_param_name(self) -> str: + return "test_destination" + + @property + def _parameter_output_type(self) -> str: + return "str" + + +def _make_param(param_name: str = "output") -> _ConcreteParam: + mock_node = mock.MagicMock() + mock_node.name = "MyNode" + return _ConcreteParam(mock_node, param_name, default_value="default.txt", situation="save_node_output") + + +def _make_connections_result( + *connections: connection_events.IncomingConnection, +) -> connection_events.ListConnectionsForNodeResultSuccess: + return connection_events.ListConnectionsForNodeResultSuccess( + result_details="ok", + incoming_connections=list(connections), + outgoing_connections=[], + ) + + +def _make_connection( + target_param: str, + source_node: str = "UpstreamNode", + source_param: str = "test_destination", +) -> connection_events.IncomingConnection: + return connection_events.IncomingConnection( + source_node_name=source_node, + source_parameter_name=source_param, + target_parameter_name=target_param, + ) + + +class TestGetUpstreamDestination: + """Tests for _get_upstream_destination, which finds an upstream provider via hasattr.""" + + def test_returns_none_when_list_connections_fails(self) -> None: + param = _make_param() + failure = connection_events.ListConnectionsForNodeResultFailure(result_details="error") + + with mock.patch(HANDLE_REQUEST_PATH, return_value=failure): + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is None + + def test_returns_none_when_no_incoming_connections(self) -> None: + param = _make_param() + success = _make_connections_result() + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success): + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is None + + def test_returns_none_when_connection_targets_different_parameter(self) -> None: + param = _make_param("output") + conn = _make_connection(target_param="other_param") + success = _make_connections_result(conn) + mock_source = mock.MagicMock(spec=[]) # no attributes + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(OBJECT_MANAGER_PATH) as mock_om, + ): + mock_om.return_value.attempt_get_object_by_name.return_value = mock_source + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is None + + def test_returns_none_when_source_node_not_found(self) -> None: + param = _make_param() + conn = _make_connection(target_param="output") + success = _make_connections_result(conn) + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(OBJECT_MANAGER_PATH) as mock_om, + ): + mock_om.return_value.attempt_get_object_by_name.return_value = None + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is None + + def test_returns_none_when_source_node_lacks_attribute(self) -> None: + param = _make_param() + conn = _make_connection(target_param="output") + success = _make_connections_result(conn) + mock_source = mock.MagicMock(spec=[]) # no attributes at all + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(OBJECT_MANAGER_PATH) as mock_om, + ): + mock_om.return_value.attempt_get_object_by_name.return_value = mock_source + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is None + + def test_returns_destination_when_source_has_attribute(self) -> None: + param = _make_param() + conn = _make_connection(target_param="output") + success = _make_connections_result(conn) + expected_dest = mock.MagicMock() + mock_source = mock.MagicMock(spec=["test_destination"]) + mock_source.test_destination = expected_dest + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(OBJECT_MANAGER_PATH) as mock_om, + ): + mock_om.return_value.attempt_get_object_by_name.return_value = mock_source + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is expected_dest + + def test_raises_when_provider_attribute_returns_none(self) -> None: + param = _make_param() + conn = _make_connection(target_param="output") + success = _make_connections_result(conn) + mock_source = mock.MagicMock(spec=["test_destination"]) + mock_source.test_destination = None + + with mock.patch(HANDLE_REQUEST_PATH, return_value=success), mock.patch(OBJECT_MANAGER_PATH) as mock_om: + mock_om.return_value.attempt_get_object_by_name.return_value = mock_source + with pytest.raises(ValueError, match="UpstreamNode"): + param._get_upstream_destination("test_destination", "TestDestination") + + def test_skips_non_provider_and_returns_provider_destination(self) -> None: + """A non-provider connection followed by a provider: skips first, returns second.""" + param = _make_param() + conn_non_provider = _make_connection(target_param="output", source_node="PlainNode") + conn_provider = _make_connection(target_param="output", source_node="ProviderNode") + success = _make_connections_result(conn_non_provider, conn_provider) + + expected_dest = mock.MagicMock() + plain_source = mock.MagicMock(spec=[]) # no test_destination + provider_source = mock.MagicMock(spec=["test_destination"]) + provider_source.test_destination = expected_dest + + def get_node(name: str) -> mock.MagicMock: + return plain_source if name == "PlainNode" else provider_source + + with ( + mock.patch(HANDLE_REQUEST_PATH, return_value=success), + mock.patch(OBJECT_MANAGER_PATH) as mock_om, + ): + mock_om.return_value.attempt_get_object_by_name.side_effect = get_node + result = param._get_upstream_destination("test_destination", "TestDestination") + + assert result is expected_dest + + def test_allowed_modes_default(self) -> None: + param = _make_param() + assert param._allowed_modes == {core_types.ParameterMode.INPUT, core_types.ParameterMode.PROPERTY} + + def test_custom_allowed_modes(self) -> None: + mock_node = mock.MagicMock() + mock_node.name = "N" + param = _ConcreteParam( + mock_node, "out", default_value="x", situation="s", allowed_modes={core_types.ParameterMode.OUTPUT} + ) + assert param._allowed_modes == {core_types.ParameterMode.OUTPUT}