diff --git a/ApplicationLibCode/Commands/WorkflowCommands/CMakeLists_files.cmake b/ApplicationLibCode/Commands/WorkflowCommands/CMakeLists_files.cmake index 5540dd50a6..22db60a3da 100644 --- a/ApplicationLibCode/Commands/WorkflowCommands/CMakeLists_files.cmake +++ b/ApplicationLibCode/Commands/WorkflowCommands/CMakeLists_files.cmake @@ -15,6 +15,10 @@ set(SOURCE_GROUP_SOURCE_FILES ${CMAKE_CURRENT_LIST_DIR}/RicRenameWorkflowTaskFeature.cpp ${CMAKE_CURRENT_LIST_DIR}/RicSetWorkflowResultTaskFeature.cpp ${CMAKE_CURRENT_LIST_DIR}/RicToggleWorkflowOptionalInputFeature.cpp + ${CMAKE_CURRENT_LIST_DIR}/RicMapWorkflowTaskFeature.cpp + ${CMAKE_CURRENT_LIST_DIR}/RicRemoveWorkflowTaskMapFeature.cpp + ${CMAKE_CURRENT_LIST_DIR}/RicRenameWorkflowCollectKeyFeature.cpp + ${CMAKE_CURRENT_LIST_DIR}/RicMoveWorkflowCollectMemberFeature.cpp ) list(APPEND COMMAND_CODE_SOURCE_FILES ${SOURCE_GROUP_SOURCE_FILES}) diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.cpp b/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.cpp new file mode 100644 index 0000000000..8ad239401e --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.cpp @@ -0,0 +1,52 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#include "RicMapWorkflowTaskFeature.h" + +#include "RicWorkflowFeatureTools.h" +#include "RiuWorkflowEditorWidget.h" + +#include + +CAF_CMD_SOURCE_INIT( RicMapWorkflowTaskFeature, "RicMapWorkflowTaskFeature" ); + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RicMapWorkflowTaskFeature::isCommandEnabled() const +{ + return RicWorkflowFeatureTools::editableWorkflowEditor() != nullptr; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicMapWorkflowTaskFeature::onActionTriggered( bool isChecked ) +{ + const QVariantMap data = userData().toMap(); + if ( auto* editor = RicWorkflowFeatureTools::editableWorkflowEditor() ) editor->editTaskMap( data.value( "task" ).toString() ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicMapWorkflowTaskFeature::setupActionLook( QAction* actionToSetup ) +{ + actionToSetup->setText( "Map Over..." ); + actionToSetup->setIcon( QIcon( ":/Refresh.svg" ) ); +} diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.h b/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.h new file mode 100644 index 0000000000..a293a1e8b6 --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicMapWorkflowTaskFeature.h @@ -0,0 +1,34 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#pragma once + +#include "cafCmdFeature.h" + +//================================================================================================== +/// Graph command: acts on the workflow shown in the workflow editor, with the item in the user data +//================================================================================================== +class RicMapWorkflowTaskFeature : public caf::CmdFeature +{ + CAF_CMD_HEADER_INIT; + +protected: + bool isCommandEnabled() const override; + void onActionTriggered( bool isChecked ) override; + void setupActionLook( QAction* actionToSetup ) override; +}; diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.cpp b/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.cpp new file mode 100644 index 0000000000..789e9ae7e3 --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.cpp @@ -0,0 +1,58 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#include "RicMoveWorkflowCollectMemberFeature.h" + +#include "RicWorkflowFeatureTools.h" +#include "RiuWorkflowEditorWidget.h" + +#include + +CAF_CMD_SOURCE_INIT( RicMoveWorkflowCollectMemberFeature, "RicMoveWorkflowCollectMemberFeature" ); + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RicMoveWorkflowCollectMemberFeature::isCommandEnabled() const +{ + return RicWorkflowFeatureTools::editableWorkflowEditor() != nullptr; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicMoveWorkflowCollectMemberFeature::onActionTriggered( bool isChecked ) +{ + const QVariantMap data = userData().toMap(); + if ( auto* editor = RicWorkflowFeatureTools::editableWorkflowEditor() ) + { + editor->moveCollectMember( RiuWorkflowGraphView::ItemRef::forEdge( data.value( "from" ).toString(), + data.value( "output" ).toString(), + data.value( "to" ).toString(), + data.value( "input" ).toString() ), + data.value( "delta" ).toInt() ); + } +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicMoveWorkflowCollectMemberFeature::setupActionLook( QAction* actionToSetup ) +{ + actionToSetup->setText( "Move Earlier" ); +} diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.h b/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.h new file mode 100644 index 0000000000..9b1ae5b3f1 --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicMoveWorkflowCollectMemberFeature.h @@ -0,0 +1,34 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#pragma once + +#include "cafCmdFeature.h" + +//================================================================================================== +/// Graph command: acts on the workflow shown in the workflow editor, with the item in the user data +//================================================================================================== +class RicMoveWorkflowCollectMemberFeature : public caf::CmdFeature +{ + CAF_CMD_HEADER_INIT; + +protected: + bool isCommandEnabled() const override; + void onActionTriggered( bool isChecked ) override; + void setupActionLook( QAction* actionToSetup ) override; +}; diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.cpp b/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.cpp new file mode 100644 index 0000000000..74d34520ad --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.cpp @@ -0,0 +1,51 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#include "RicRemoveWorkflowTaskMapFeature.h" + +#include "RicWorkflowFeatureTools.h" +#include "RiuWorkflowEditorWidget.h" + +#include + +CAF_CMD_SOURCE_INIT( RicRemoveWorkflowTaskMapFeature, "RicRemoveWorkflowTaskMapFeature" ); + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RicRemoveWorkflowTaskMapFeature::isCommandEnabled() const +{ + return RicWorkflowFeatureTools::editableWorkflowEditor() != nullptr; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicRemoveWorkflowTaskMapFeature::onActionTriggered( bool isChecked ) +{ + const QVariantMap data = userData().toMap(); + if ( auto* editor = RicWorkflowFeatureTools::editableWorkflowEditor() ) editor->removeTaskMap( data.value( "task" ).toString() ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicRemoveWorkflowTaskMapFeature::setupActionLook( QAction* actionToSetup ) +{ + actionToSetup->setText( "Remove Map" ); +} diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.h b/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.h new file mode 100644 index 0000000000..e858d68cb4 --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicRemoveWorkflowTaskMapFeature.h @@ -0,0 +1,34 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#pragma once + +#include "cafCmdFeature.h" + +//================================================================================================== +/// Graph command: acts on the workflow shown in the workflow editor, with the item in the user data +//================================================================================================== +class RicRemoveWorkflowTaskMapFeature : public caf::CmdFeature +{ + CAF_CMD_HEADER_INIT; + +protected: + bool isCommandEnabled() const override; + void onActionTriggered( bool isChecked ) override; + void setupActionLook( QAction* actionToSetup ) override; +}; diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.cpp b/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.cpp new file mode 100644 index 0000000000..33460d2e4d --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.cpp @@ -0,0 +1,57 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#include "RicRenameWorkflowCollectKeyFeature.h" + +#include "RicWorkflowFeatureTools.h" +#include "RiuWorkflowEditorWidget.h" + +#include + +CAF_CMD_SOURCE_INIT( RicRenameWorkflowCollectKeyFeature, "RicRenameWorkflowCollectKeyFeature" ); + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RicRenameWorkflowCollectKeyFeature::isCommandEnabled() const +{ + return RicWorkflowFeatureTools::editableWorkflowEditor() != nullptr; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicRenameWorkflowCollectKeyFeature::onActionTriggered( bool isChecked ) +{ + const QVariantMap data = userData().toMap(); + if ( auto* editor = RicWorkflowFeatureTools::editableWorkflowEditor() ) + { + editor->renameCollectKey( RiuWorkflowGraphView::ItemRef::forEdge( data.value( "from" ).toString(), + data.value( "output" ).toString(), + data.value( "to" ).toString(), + data.value( "input" ).toString() ) ); + } +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RicRenameWorkflowCollectKeyFeature::setupActionLook( QAction* actionToSetup ) +{ + actionToSetup->setText( "Rename Key..." ); +} diff --git a/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.h b/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.h new file mode 100644 index 0000000000..190666e6eb --- /dev/null +++ b/ApplicationLibCode/Commands/WorkflowCommands/RicRenameWorkflowCollectKeyFeature.h @@ -0,0 +1,34 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#pragma once + +#include "cafCmdFeature.h" + +//================================================================================================== +/// Graph command: acts on the workflow shown in the workflow editor, with the item in the user data +//================================================================================================== +class RicRenameWorkflowCollectKeyFeature : public caf::CmdFeature +{ + CAF_CMD_HEADER_INIT; + +protected: + bool isCommandEnabled() const override; + void onActionTriggered( bool isChecked ) override; + void setupActionLook( QAction* actionToSetup ) override; +}; diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake b/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake index 81f0528a46..5574346949 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake +++ b/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake @@ -24,6 +24,7 @@ set(SOURCE_GROUP_SOURCE_FILES ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowDateBinding.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowFilePathBinding.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowVec3Binding.cpp + ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowMappingBinding.cpp ) list(APPEND CODE_SOURCE_FILES ${SOURCE_GROUP_SOURCE_FILES}) diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.cpp index df4e1d8e3d..f29bf4ec6e 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.cpp @@ -18,6 +18,8 @@ #include "RimWorkflowDefinition.h" +#include "RimWorkflowPortCompatibility.h" + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -34,6 +36,22 @@ bool RimWorkflowDefinitionEdge::isWholeOutput() const return output.isEmpty(); } +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RimWorkflowDefinitionEdge::isCollected() const +{ + return collect != Collect::None; +} + +//-------------------------------------------------------------------------------------------------- +/// Same tasks and ports, ignoring how the value is collected +//-------------------------------------------------------------------------------------------------- +bool RimWorkflowDefinitionEdge::sameConnection( const RimWorkflowDefinitionEdge& other ) const +{ + return from == other.from && output == other.output && to == other.to && input == other.input; +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -79,12 +97,16 @@ QJsonObject RimWorkflowDefinition::taskType( const QString& taskId ) const } //-------------------------------------------------------------------------------------------------- -/// +/// The task type as seen by the workflow: a mapped task has `dict[str, Output]` as output, and its +/// key and value fields are not inputs //-------------------------------------------------------------------------------------------------- QJsonObject RimWorkflowDefinition::taskTypeForNode( const QString& nodeName ) const { const auto* node = findNode( nodeName ); - return node ? taskType( node->taskId ) : QJsonObject(); + if ( !node ) return {}; + const QJsonObject type = taskType( node->taskId ); + if ( type.isEmpty() || !node->map ) return type; + return RimWorkflowPortCompatibility::mappedTaskType( type, *node->map ); } //-------------------------------------------------------------------------------------------------- diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.h index 80ba45698e..f182544bf3 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.h +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinition.h @@ -23,8 +23,24 @@ #include #include +#include #include +//================================================================================================== +/// A mapped task (`map:` in YAML) runs once per entry of the job config value `over`, a mapping +/// with string keys. Each key goes to the input field `keyAs` and each value to `valueAs`. +/// The output is `dict[str, Output]` keyed like `over`. +//================================================================================================== +struct RimWorkflowTaskMap +{ + QString over{}; + QString keyAs{}; + QString valueAs{}; + QString errorMode = "fail_fast"; + + bool operator==( const RimWorkflowTaskMap& other ) const = default; +}; + //================================================================================================== /// One task instance in a workflow //================================================================================================== @@ -37,7 +53,9 @@ struct RimWorkflowDefinitionNode // Fields the user wants to configure even when they are optional or cannot be derived QStringList explicitConfigFields{}; - // Keys of the YAML entry the editor does not model (nested workflow, map); kept for read-only display + std::optional map{}; + + // Keys of the YAML entry the editor does not model (nested workflow); kept for read-only display QJsonObject extra{}; bool operator==( const RimWorkflowDefinitionNode& other ) const = default; @@ -45,16 +63,29 @@ struct RimWorkflowDefinitionNode //================================================================================================== /// A connection between two tasks. An empty output or input means the whole model. +/// Several edges can collect outputs into one `list[T]` or `dict[str, T]` input (`collect:` in +/// YAML). List members are ordered like the edges; dict members have a key. //================================================================================================== struct RimWorkflowDefinitionEdge { + enum class Collect + { + None, + List, + Dict + }; + QString from{}; QString output{}; QString to{}; QString input{}; + Collect collect = Collect::None; + QString key{}; bool isWholeInput() const; bool isWholeOutput() const; + bool isCollected() const; + bool sameConnection( const RimWorkflowDefinitionEdge& other ) const; bool operator==( const RimWorkflowDefinitionEdge& other ) const = default; }; diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.cpp index 977acdb6a4..e115e20b9d 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.cpp @@ -23,9 +23,12 @@ #include "RimWorkflowTaskCatalog.h" #include +#include #include #include +#include + namespace { const QString explicitConfigKey = "x_ri_explicit_config_fields"; @@ -52,14 +55,56 @@ RimWorkflowDefinitionNode nodeFromJson( const QJsonObject& json ) node.taskId = json.value( "task" ).toString(); node.configFields = stringList( json.value( "config_fields" ) ); node.explicitConfigFields = json.contains( explicitConfigKey ) ? stringList( json.value( explicitConfigKey ) ) : node.configFields; + if ( json.value( "map" ).isObject() ) + { + const QJsonObject map = json.value( "map" ).toObject(); + node.map = RimWorkflowTaskMap{ .over = map.value( "over" ).toString(), + .keyAs = map.value( "key_as" ).toString(), + .valueAs = map.value( "value_as" ).toString(), + .errorMode = map.value( "error_mode" ).toString( "fail_fast" ) }; + } for ( auto it = json.begin(); it != json.end(); ++it ) { - if ( it.key() != "name" && it.key() != "task" && it.key() != "config_fields" && it.key() != explicitConfigKey ) + if ( it.key() != "name" && it.key() != "task" && it.key() != "config_fields" && it.key() != explicitConfigKey && it.key() != "map" ) node.extra.insert( it.key(), it.value() ); } return node; } +QJsonObject mapToJson( const RimWorkflowTaskMap& map ) +{ + return QJsonObject{ { "over", map.over }, { "key_as", map.keyAs }, { "value_as", map.valueAs }, { "error_mode", map.errorMode } }; +} + +QString collectName( RimWorkflowDefinitionEdge::Collect collect ) +{ + switch ( collect ) + { + case RimWorkflowDefinitionEdge::Collect::List: + return "list"; + case RimWorkflowDefinitionEdge::Collect::Dict: + return "dict"; + default: + return {}; + } +} + +RimWorkflowDefinitionEdge edgeFromJson( const QJsonObject& json ) +{ + RimWorkflowDefinitionEdge edge{ .from = json.value( "from" ).toString(), + .output = json.value( "output" ).toString(), + .to = json.value( "to" ).toString(), + .input = json.value( "input" ).toString() }; + const QString collect = json.value( "collect" ).toString(); + if ( collect == "list" ) edge.collect = RimWorkflowDefinitionEdge::Collect::List; + if ( collect == "dict" ) + { + edge.collect = RimWorkflowDefinitionEdge::Collect::Dict; + edge.key = json.value( "key" ).toString(); + } + return edge; +} + QJsonObject nodeToJson( const RimWorkflowDefinitionNode& node ) { QJsonObject json = node.extra; @@ -67,15 +112,19 @@ QJsonObject nodeToJson( const RimWorkflowDefinitionNode& node ) json["task"] = node.taskId; json["config_fields"] = QJsonArray::fromStringList( node.configFields ); json[explicitConfigKey] = QJsonArray::fromStringList( node.explicitConfigFields ); + if ( node.map ) json["map"] = mapToJson( *node.map ); return json; } QJsonObject edgeToJson( const RimWorkflowDefinitionEdge& edge ) { - return QJsonObject{ { "from", edge.from }, - { "output", nullIfEmpty( edge.output ) }, - { "to", edge.to }, - { "input", nullIfEmpty( edge.input ) } }; + QJsonObject json{ { "from", edge.from }, { "output", nullIfEmpty( edge.output ) }, { "to", edge.to }, { "input", nullIfEmpty( edge.input ) } }; + if ( edge.isCollected() ) + { + json["collect"] = collectName( edge.collect ); + json["key"] = edge.collect == RimWorkflowDefinitionEdge::Collect::Dict ? nullIfEmpty( edge.key ) : QJsonValue( QJsonValue::Null ); + } + return json; } QJsonObject issueToJson( const RimWorkflowIssue& issue ) @@ -126,9 +175,11 @@ bool hasCycle( const RimWorkflowDefinition& definition ) void validateConnections( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionNode& node, std::vector& issues ) { - const auto incoming = definition.incomingEdges( node.name ); - int wholeCount = 0; - QSet fieldInputs; + const auto incoming = definition.incomingEdges( node.name ); + int wholeCount = 0; + QSet fieldInputs; + QMap collectKinds; + QMap collectKeys; for ( const auto& edge : incoming ) { if ( edge.isWholeInput() ) @@ -136,7 +187,31 @@ void validateConnections( const RimWorkflowDefinition& definition, const RimWork ++wholeCount; continue; } - if ( fieldInputs.contains( edge.input ) ) + if ( edge.isCollected() ) + { + const QString kind = collectName( edge.collect ); + if ( fieldInputs.contains( edge.input ) && !collectKinds.contains( edge.input ) ) + addIssue( issues, + node.name, + edge.input, + QString( "Input '%1' has both a direct connection and collected members" ).arg( edge.input ) ); + if ( collectKinds.contains( edge.input ) && collectKinds.value( edge.input ) != kind ) + addIssue( issues, node.name, edge.input, QString( "Input '%1' mixes list and keyed members" ).arg( edge.input ) ); + collectKinds.insert( edge.input, kind ); + if ( edge.collect == RimWorkflowDefinitionEdge::Collect::Dict ) + { + if ( edge.key.isEmpty() ) + addIssue( issues, node.name, edge.input, QString( "A member of input '%1' has no key" ).arg( edge.input ) ); + else if ( collectKeys[edge.input].contains( edge.key ) ) + addIssue( issues, node.name, edge.input, QString( "Duplicate key '%1' in input '%2'" ).arg( edge.key, edge.input ) ); + collectKeys[edge.input].append( edge.key ); + } + fieldInputs.insert( edge.input ); + continue; + } + if ( collectKinds.contains( edge.input ) ) + addIssue( issues, node.name, edge.input, QString( "Input '%1' has both a direct connection and collected members" ).arg( edge.input ) ); + else if ( fieldInputs.contains( edge.input ) ) addIssue( issues, node.name, edge.input, QString( "Input '%1' has more than one connection" ).arg( edge.input ) ); fieldInputs.insert( edge.input ); } @@ -148,16 +223,35 @@ void validateConnections( const RimWorkflowDefinition& definition, const RimWork for ( const auto& edge : incoming ) { if ( !definition.findNode( edge.from ) ) continue; + if ( edge.isCollected() ) + { + auto check = RimWorkflowPortCompatibility::checkCollectMember( definition, edge ); + if ( !check ) addIssue( issues, node.name, edge.input, check.error() ); + continue; + } RimWorkflowDefinition others = definition; std::erase( others.edges, edge ); - auto check = RimWorkflowPortCompatibility::canConnect( others, edge.from, edge.output, edge.to, edge.input ); - if ( !check ) addIssue( issues, node.name, edge.input, check.error() ); + auto check = RimWorkflowPortCompatibility::resolveConnection( others, edge.from, edge.output, edge.to, edge.input ); + if ( !check ) + addIssue( issues, node.name, edge.input, check.error() ); + else if ( check->isCollected() ) + addIssue( issues, + node.name, + edge.input, + QString( "%1 is a single member of input '%2'; it must be collected" ).arg( edge.from, edge.input ) ); } } +void validateMap( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionNode& node, std::vector& issues ) +{ + if ( !node.map ) return; + auto check = RimWorkflowDefinitionTools::checkTaskMap( definition, node.name, *node.map ); + if ( !check ) addIssue( issues, node.name, node.map->over, check.error() ); +} + void validateInputs( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionNode& node, std::vector& issues ) { - const QJsonObject taskType = definition.taskType( node.taskId ); + const QJsonObject taskType = definition.taskTypeForNode( node.name ); const QStringList covered = RimWorkflowDefinitionTools::coveredInputFields( definition, node.name ); const auto incoming = definition.incomingEdges( node.name ); @@ -173,7 +267,7 @@ void validateInputs( const RimWorkflowDefinition& definition, const RimWorkflowD const QJsonObject inputSchema = taskType.value( "input_schema" ).toObject(); const bool hasFields = !inputSchema.value( "properties" ).toObject().isEmpty(); - if ( incoming.empty() && hasFields && node.configFields.isEmpty() && !hasFieldIssue ) + if ( incoming.empty() && hasFields && node.configFields.isEmpty() && !hasFieldIssue && !node.map ) { const QString message = RimWorkflowPortCompatibility::inputPorts( taskType ).size() > 1 @@ -216,15 +310,17 @@ QJsonArray inputPortsJson( const RimWorkflowDefinition& definit bool wired = false; for ( const auto& edge : incoming ) wired = wired || edge.input == port.name; - result.append( QJsonObject{ { "name", port.name }, - { "type", port.typeName }, - { "icon", port.iconResource }, - { "description", port.description }, - { "required", port.required }, - { "configurable", port.configurable }, - { "configured", !port.isWhole() && node.configFields.contains( port.name ) }, - { "covered", !port.isWhole() && covered.contains( port.name ) }, - { "wired", wired } } ); + QJsonObject json{ { "name", port.name }, + { "type", port.typeName }, + { "icon", port.iconResource }, + { "description", port.description }, + { "required", port.required }, + { "configurable", port.configurable }, + { "configured", !port.isWhole() && node.configFields.contains( port.name ) }, + { "covered", !port.isWhole() && covered.contains( port.name ) }, + { "wired", wired } }; + if ( const auto target = RimWorkflowPortCompatibility::collectTarget( port ) ) json["collect"] = collectName( target->first ); + result.append( json ); } return result; } @@ -251,7 +347,7 @@ QJsonObject graphTask( const RimWorkflowDefinition& definition, const QString& resultTask, const std::vector& issues ) { - const QJsonObject taskType = definition.taskType( node.taskId ); + const QJsonObject taskType = definition.taskTypeForNode( node.name ); const QJsonObject inputSchema = taskType.value( "input_schema" ).toObject(); const auto inputPorts = RimWorkflowPortCompatibility::inputPorts( taskType ); const auto outputPorts = RimWorkflowPortCompatibility::outputPorts( taskType ); @@ -272,6 +368,31 @@ QJsonObject graphTask( const RimWorkflowDefinition& definition, for ( const QString& field : node.configFields ) configFields.append( RimWorkflowSchemaTools::configFieldSchema( field, inputSchema, required, configValues ) ); + // The mapping a mapped task runs over is a config value, shown as an input of the task + QJsonArray inputPortList = inputPortsJson( definition, node, inputPorts, definition.incomingEdges( node.name ) ); + QJsonObject inputTypes = RimWorkflowPortCompatibility::portTypes( inputPorts ); + QJsonObject inputIcons = RimWorkflowPortCompatibility::portIcons( inputPorts ); + QJsonObject mapJson; + if ( node.map ) + { + const QJsonObject overSchema = RimWorkflowDefinitionTools::mapOverSchema( definition, node.name ); + configFields.append( overSchema ); + mapJson = mapToJson( *node.map ); + mapJson["type"] = overSchema.value( "type_name" ).toString(); + inputTypes[node.map->over] = overSchema.value( "type_name" ).toString(); + if ( overSchema.contains( "icon" ) ) inputIcons[node.map->over] = overSchema.value( "icon" ).toString(); + inputPortList.prepend( QJsonObject{ { "name", node.map->over }, + { "type", overSchema.value( "type_name" ).toString() }, + { "icon", overSchema.value( "icon" ).toString() }, + { "description", overSchema.value( "description" ).toString() }, + { "required", true }, + { "configurable", true }, + { "configured", true }, + { "covered", false }, + { "wired", false }, + { "map_over", true } } ); + } + QJsonArray taskIssues; for ( const auto& issue : issues ) { @@ -288,12 +409,13 @@ QJsonObject graphTask( const RimWorkflowDefinition& definition, { "config_fields", configFields }, { "accepts_input", !inputSchema.value( "properties" ).toObject().isEmpty() }, { "whole_input", inputPorts.size() == 1 && !inputSchema.value( "properties" ).toObject().isEmpty() }, - { "input_ports", inputPortsJson( definition, node, inputPorts, definition.incomingEdges( node.name ) ) }, + { "input_ports", inputPortList }, { "output_ports", outputPortsJson( outputPorts, definition.outgoingEdges( node.name ) ) }, - { "input_types", RimWorkflowPortCompatibility::portTypes( inputPorts ) }, + { "input_types", inputTypes }, { "output_types", RimWorkflowPortCompatibility::portTypes( outputPorts ) }, - { "input_icons", RimWorkflowPortCompatibility::portIcons( inputPorts ) }, + { "input_icons", inputIcons }, { "output_icons", RimWorkflowPortCompatibility::portIcons( outputPorts ) }, + { "map", mapJson.isEmpty() ? QJsonValue( QJsonValue::Null ) : QJsonValue( mapJson ) }, { "is_result", node.name == resultTask }, { "issues", taskIssues } }; } @@ -336,13 +458,7 @@ std::expected RimWorkflowDefinitionTools::fromJs } for ( const QJsonValue& value : json.value( "edges" ).toArray() ) - { - const QJsonObject edge = value.toObject(); - definition.edges.push_back( { .from = edge.value( "from" ).toString(), - .output = edge.value( "output" ).toString(), - .to = edge.value( "to" ).toString(), - .input = edge.value( "input" ).toString() } ); - } + definition.edges.push_back( edgeFromJson( value.toObject() ) ); return definition; } @@ -402,7 +518,7 @@ RimWorkflowDefinition RimWorkflowDefinitionTools::editableCopy( const RimWorkflo } //-------------------------------------------------------------------------------------------------- -/// Parts of the workflow the editor does not model (nested workflows, `map:`). A copy of such a +/// Parts of the workflow the editor does not model (nested workflows). A copy of such a /// workflow stays read-only. //-------------------------------------------------------------------------------------------------- QStringList RimWorkflowDefinitionTools::structuralReadOnlyReasons( const RimWorkflowDefinition& definition ) @@ -529,7 +645,7 @@ QStringList RimWorkflowDefinitionTools::deriveConfigFields( const RimWorkflowDef { const auto* node = definition.findNode( nodeName ); if ( !node ) return {}; - const QJsonObject taskType = definition.taskType( node->taskId ); + const QJsonObject taskType = definition.taskTypeForNode( nodeName ); if ( taskType.isEmpty() ) return node->configFields; for ( const auto& edge : definition.incomingEdges( nodeName ) ) @@ -545,7 +661,7 @@ QStringList RimWorkflowDefinitionTools::deriveConfigFields( const RimWorkflowDef QStringList derived; for ( const QString& field : RimWorkflowSchemaTools::propertyNames( inputSchema ) ) { - if ( covered.contains( field ) ) continue; + if ( covered.contains( field ) || ( node->map && field == node->map->over ) ) continue; const bool configurable = RimWorkflowSchemaTools::isConfigurable( properties.value( field ).toObject(), inputSchema ); if ( node->explicitConfigFields.contains( field ) || ( required.contains( field ) && configurable ) ) derived.append( field ); } @@ -622,20 +738,25 @@ RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::removeTask( c } //-------------------------------------------------------------------------------------------------- -/// Add a connection. An existing connection to the same input is replaced. +/// Add a connection. An existing connection to the same input is replaced, except when the new +/// connection is another member of a collected (list or dict) input. //-------------------------------------------------------------------------------------------------- RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::connect( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge ) { if ( !definition.editable ) return std::unexpected( "The workflow is read-only" ); - auto check = RimWorkflowPortCompatibility::canConnect( definition, edge.from, edge.output, edge.to, edge.input ); - if ( !check ) return std::unexpected( check.error() ); + auto resolved = RimWorkflowPortCompatibility::resolveConnection( definition, edge.from, edge.output, edge.to, edge.input ); + if ( !resolved ) return std::unexpected( resolved.error() ); RimWorkflowDefinition result = definition; std::erase_if( result.edges, - [&edge]( const RimWorkflowDefinitionEdge& existing ) { return existing.to == edge.to && existing.input == edge.input; } ); - result.edges.push_back( edge ); + [&resolved]( const RimWorkflowDefinitionEdge& existing ) + { + return existing.to == resolved->to && existing.input == resolved->input && + ( !resolved->isCollected() || existing.collect != resolved->collect ); + } ); + result.edges.push_back( *resolved ); return normalize( result ); } @@ -648,10 +769,181 @@ RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::disconnect( c if ( !definition.editable ) return std::unexpected( "The workflow is read-only" ); RimWorkflowDefinition result = definition; - if ( std::erase( result.edges, edge ) == 0 ) return std::unexpected( "The connection does not exist" ); + if ( std::erase_if( result.edges, [&edge]( const RimWorkflowDefinitionEdge& existing ) { return existing.sameConnection( edge ); } ) == 0 ) + return std::unexpected( "The connection does not exist" ); return normalize( result ); } +//-------------------------------------------------------------------------------------------------- +/// Rename the key of a member of a `dict[str, T]` input +//-------------------------------------------------------------------------------------------------- +RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::setCollectKey( const RimWorkflowDefinition& definition, + const RimWorkflowDefinitionEdge& edge, + const QString& key ) +{ + if ( !definition.editable ) return std::unexpected( "The workflow is read-only" ); + if ( key.trimmed().isEmpty() ) return std::unexpected( "The key cannot be empty" ); + + RimWorkflowDefinition result = definition; + auto it = + std::find_if( result.edges.begin(), result.edges.end(), [&edge]( const auto& existing ) { return existing.sameConnection( edge ); } ); + if ( it == result.edges.end() ) return std::unexpected( "The connection does not exist" ); + if ( it->collect != RimWorkflowDefinitionEdge::Collect::Dict ) return std::unexpected( "The connection is not a keyed member" ); + + for ( const auto& other : result.edges ) + { + if ( &other != &*it && other.to == it->to && other.input == it->input && other.key == key ) + return std::unexpected( QString( "Input '%1' already has a member with key '%2'" ).arg( it->input, key ) ); + } + it->key = key; + return result; +} + +//-------------------------------------------------------------------------------------------------- +/// Move a member of a `list[T]` input `delta` positions within the list +//-------------------------------------------------------------------------------------------------- +RimWorkflowDefinitionTools::EditResult + RimWorkflowDefinitionTools::moveCollectMember( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge, int delta ) +{ + if ( !definition.editable ) return std::unexpected( "The workflow is read-only" ); + + RimWorkflowDefinition result = definition; + std::vector members; + size_t position = 0; + bool found = false; + for ( size_t i = 0; i < result.edges.size(); ++i ) + { + const auto& existing = result.edges[i]; + if ( existing.to != edge.to || existing.input != edge.input || !existing.isCollected() ) continue; + if ( existing.sameConnection( edge ) ) + { + position = members.size(); + found = true; + } + members.push_back( i ); + } + if ( !found ) return std::unexpected( "The connection is not a collected member" ); + + const int target = static_cast( position ) + delta; + if ( target < 0 || target >= static_cast( members.size() ) ) return std::unexpected( "The member cannot be moved further" ); + std::swap( result.edges[members[position]], result.edges[members[static_cast( target )]] ); + return result; +} + +//-------------------------------------------------------------------------------------------------- +/// Checks that a map fits the task: the key and value fields are inputs of the task, and `over` +/// is a new config field name +//-------------------------------------------------------------------------------------------------- +std::expected + RimWorkflowDefinitionTools::checkTaskMap( const RimWorkflowDefinition& definition, const QString& nodeName, const RimWorkflowTaskMap& map ) +{ + const auto* node = definition.findNode( nodeName ); + if ( !node ) return std::unexpected( QString( "Unknown task '%1'" ).arg( nodeName ) ); + + if ( !isIdentifier( map.over ) ) return std::unexpected( QString( "'%1' is not a valid name for the mapping" ).arg( map.over ) ); + if ( map.keyAs.isEmpty() || map.valueAs.isEmpty() ) return std::unexpected( "Choose the inputs that receive the key and the value" ); + if ( map.keyAs == map.valueAs ) return std::unexpected( "The same input cannot receive both the key and the value" ); + if ( map.errorMode != "fail_fast" && map.errorMode != "collect_all" ) + return std::unexpected( QString( "Unknown error mode '%1'" ).arg( map.errorMode ) ); + + const QJsonObject taskType = definition.taskType( node->taskId ); + if ( taskType.isEmpty() ) return {}; + + const QJsonObject inputSchema = taskType.value( "input_schema" ).toObject(); + const QJsonObject properties = inputSchema.value( "properties" ).toObject(); + for ( const QString& field : { map.keyAs, map.valueAs } ) + { + if ( !properties.contains( field ) ) return std::unexpected( QString( "'%1' has no input '%2'" ).arg( nodeName, field ) ); + } + if ( properties.contains( map.over ) ) + return std::unexpected( QString( "'%1' is an input of '%2'; choose another name for the mapping" ).arg( map.over, nodeName ) ); + + const QJsonObject keySchema = properties.value( map.keyAs ).toObject(); + const QJsonObject strSchema{ { "type", "string" } }; + if ( !RimWorkflowPortCompatibility::isCompatible( strSchema, strSchema, keySchema, inputSchema ) ) + return std::unexpected( QString( "Input '%1' receives the key and must accept a string" ).arg( map.keyAs ) ); + return {}; +} + +//-------------------------------------------------------------------------------------------------- +/// Config field schema of the mapping a mapped task runs over: a `dict[str, Value]`, where the +/// value type is the type of the `value_as` input +//-------------------------------------------------------------------------------------------------- +QJsonObject RimWorkflowDefinitionTools::mapOverSchema( const RimWorkflowDefinition& definition, const QString& nodeName ) +{ + const auto* node = definition.findNode( nodeName ); + if ( !node || !node->map ) return {}; + + const QJsonObject inputSchema = definition.taskType( node->taskId ).value( "input_schema" ).toObject(); + const QJsonObject configValues = definition.inputs.value( nodeName ).toObject(); + + QJsonObject entry{ { "name", node->map->over }, { "type", "object" }, { "required", true }, { "map_over", true } }; + if ( configValues.contains( node->map->over ) ) entry["default"] = configValues.value( node->map->over ); + + QString valueType = "Any"; + if ( !node->map->valueAs.isEmpty() ) + { + const QJsonObject value = RimWorkflowSchemaTools::configFieldSchema( node->map->valueAs, inputSchema, {}, {} ); + entry["value_type"] = value.value( "type" ); + if ( value.contains( "resinsight_type" ) ) entry["value_resinsight_type"] = value.value( "resinsight_type" ); + if ( value.contains( "format" ) ) entry["value_format"] = value.value( "format" ); + + const QJsonObject property = inputSchema.value( "properties" ).toObject().value( node->map->valueAs ).toObject(); + valueType = RimWorkflowPortCompatibility::typeName( property, inputSchema ); + const QString icon = RimWorkflowPortCompatibility::iconResource( property, inputSchema ); + if ( !icon.isEmpty() ) entry["icon"] = icon; + } + entry["type_name"] = QString( "dict[str, %1]" ).arg( valueType ); + entry["description"] = + QString( "The items '%1' runs for; each key is passed as '%2' and each value as '%3'" ).arg( nodeName, node->map->keyAs, node->map->valueAs ); + return entry; +} + +//-------------------------------------------------------------------------------------------------- +/// Run a task once per item of a mapping, or remove the map with std::nullopt +//-------------------------------------------------------------------------------------------------- +RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::setTaskMap( const RimWorkflowDefinition& definition, + const QString& nodeName, + const std::optional& map ) +{ + if ( !definition.editable ) return std::unexpected( "The workflow is read-only" ); + const auto* node = definition.findNode( nodeName ); + if ( !node ) return std::unexpected( QString( "Unknown task '%1'" ).arg( nodeName ) ); + + RimWorkflowDefinition result = definition; + auto* target = result.findNode( nodeName ); + QJsonObject values = result.inputs.value( nodeName ).toObject(); + if ( node->map ) values.remove( node->map->over ); + + if ( map ) + { + if ( auto check = checkTaskMap( definition, nodeName, *map ); !check ) return std::unexpected( check.error() ); + for ( const auto& edge : definition.incomingEdges( nodeName ) ) + { + if ( edge.isWholeInput() ) + return std::unexpected( + QString( "'%1' takes its whole input from '%2'; disconnect it before mapping" ).arg( nodeName, edge.from ) ); + } + + // The key and value inputs are filled by the map + std::erase_if( result.edges, + [&]( const RimWorkflowDefinitionEdge& edge ) + { return edge.to == nodeName && ( edge.input == map->keyAs || edge.input == map->valueAs ); } ); + target->explicitConfigFields.removeAll( map->keyAs ); + target->explicitConfigFields.removeAll( map->valueAs ); + values.remove( map->keyAs ); + values.remove( map->valueAs ); + if ( node->map && node->map->over != map->over && definition.inputs.value( nodeName ).toObject().contains( node->map->over ) ) + values[map->over] = definition.inputs.value( nodeName ).toObject().value( node->map->over ); + } + target->map = map; + result.inputs.insert( nodeName, values ); + + // Downstream connections may no longer fit the changed output type + result = normalize( result ); + return result; +} + //-------------------------------------------------------------------------------------------------- /// Rename a task instance, updating connections, the result task and the input values //-------------------------------------------------------------------------------------------------- @@ -718,7 +1010,7 @@ RimWorkflowDefinitionTools::EditResult RimWorkflowDefinitionTools::setOptionalIn const auto* node = definition.findNode( nodeName ); if ( !node ) return std::unexpected( QString( "Unknown task '%1'" ).arg( nodeName ) ); - const auto port = RimWorkflowPortCompatibility::inputPort( definition.taskType( node->taskId ), field ); + const auto port = RimWorkflowPortCompatibility::inputPort( definition.taskTypeForNode( nodeName ), field ); if ( !port || port->isWhole() ) return std::unexpected( QString( "'%1' has no input '%2'" ).arg( nodeName, field ) ); if ( port->required ) return std::unexpected( QString( "Input '%1' is required and always configured unless connected" ).arg( field ) ); @@ -761,7 +1053,7 @@ RimWorkflowDefinition RimWorkflowDefinitionTools::autoWireResInsightInputs( cons }; QStringList inputsToWire; - const auto ports = RimWorkflowPortCompatibility::inputPorts( definition.taskType( node->taskId ) ); + const auto ports = RimWorkflowPortCompatibility::inputPorts( definition.taskTypeForNode( nodeName ) ); for ( const auto& port : ports ) { if ( port.isWhole() || wiredInputs.contains( port.name ) || !takesConnection( port ) ) continue; @@ -832,6 +1124,7 @@ std::vector RimWorkflowDefinitionTools::validate( const RimWor } validateConnections( definition, node, issues ); validateInputs( definition, node, issues ); + validateMap( definition, node, issues ); } if ( !definition.resultTask.isEmpty() && !names.contains( definition.resultTask ) ) @@ -869,12 +1162,25 @@ QJsonObject RimWorkflowDefinitionTools::graphFromDefinition( const RimWorkflowDe for ( const auto& node : definition.nodes ) tasks.append( graphTask( definition, node, resultTask, issues ) ); + // Position of each list member within its input + QMap listIndex; + QMap listCount; + for ( const auto& edge : definition.edges ) + { + if ( edge.collect == RimWorkflowDefinitionEdge::Collect::List ) listIndex[&edge] = listCount[edge.to + "." + edge.input]++; + } + QJsonArray edges; for ( const auto& edge : definition.edges ) { QJsonObject json{ { "from", edge.from }, { "to", edge.to } }; if ( !edge.input.isEmpty() ) json["input"] = edge.input; if ( !edge.output.isEmpty() ) json["output"] = edge.output; + if ( edge.isCollected() ) + { + json["collect"] = collectName( edge.collect ); + json["label"] = edge.collect == RimWorkflowDefinitionEdge::Collect::Dict ? edge.key : QString( "[%1]" ).arg( listIndex[&edge] ); + } edges.append( json ); } diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.h index 42c5ce20bd..b3c1f13cea 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.h +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDefinitionTools.h @@ -26,6 +26,7 @@ #include #include +#include #include class RimWorkflowTaskCatalog; @@ -70,9 +71,14 @@ EditResult addTask( const RimWorkflowDefinition& definition, EditResult removeTask( const RimWorkflowDefinition& definition, const QString& nodeName ); EditResult connect( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge ); EditResult disconnect( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge ); -EditResult renameTask( const RimWorkflowDefinition& definition, const QString& oldName, const QString& newName ); -EditResult renameWorkflow( const RimWorkflowDefinition& definition, const QString& newName ); -EditResult setResultTask( const RimWorkflowDefinition& definition, const QString& nodeName ); +EditResult setCollectKey( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge, const QString& key ); +EditResult moveCollectMember( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge, int delta ); +EditResult setTaskMap( const RimWorkflowDefinition& definition, const QString& nodeName, const std::optional& map ); +std::expected checkTaskMap( const RimWorkflowDefinition& definition, const QString& nodeName, const RimWorkflowTaskMap& map ); +QJsonObject mapOverSchema( const RimWorkflowDefinition& definition, const QString& nodeName ); +EditResult renameTask( const RimWorkflowDefinition& definition, const QString& oldName, const QString& newName ); +EditResult renameWorkflow( const RimWorkflowDefinition& definition, const QString& newName ); +EditResult setResultTask( const RimWorkflowDefinition& definition, const QString& nodeName ); EditResult setOptionalInputConfigured( const RimWorkflowDefinition& definition, const QString& nodeName, const QString& field, bool configured ); RimWorkflowDefinition autoWireResInsightInputs( const RimWorkflowDefinition& definition, const RimWorkflowTaskCatalog& catalog, const QString& nodeName ); diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.cpp index de33d8e93b..63d8f5d219 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.cpp @@ -209,6 +209,35 @@ void RimWorkflowJob::updateTaskState( const QString& runId, const QString& taskN if ( auto* window = RiuMainWindow::instance() ) window->workflowJobStateChanged( this ); } +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QMap> RimWorkflowJob::itemStates() const +{ + return m_itemStates; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QMap> RimWorkflowJob::itemErrors() const +{ + return m_itemErrors; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RimWorkflowJob::updateItemState( const QString& runId, const QString& taskName, const QString& item, const QString& state, const QString& error ) +{ + if ( runId != m_runId || !m_taskStates.contains( taskName ) ) return; + if ( state != "running" && state != "completed" && state != "failed" ) return; + + m_itemStates[taskName][item] = state; + if ( state == "failed" ) m_itemErrors[taskName][item] = error; + if ( auto* window = RiuMainWindow::instance() ) window->workflowJobStateChanged( this ); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -226,7 +255,7 @@ void RimWorkflowJob::finishRun( const QString& runId, bool succeeded, bool cance if ( cancelled ) m_runStatus = "Cancelled"; else if ( succeeded ) - m_runStatus = "Completed"; + m_runStatus.clear(); else if ( !m_runStatus.startsWith( "Failed:" ) ) m_runStatus = "Failed (see workflow log)"; m_activeTask.clear(); @@ -430,6 +459,8 @@ void RimWorkflowJob::runJob() m_runId = QUuid::createUuid().toString( QUuid::WithoutBraces ); m_taskStates.clear(); m_taskErrors.clear(); + m_itemStates.clear(); + m_itemErrors.clear(); for ( const QJsonValue& value : workflow->graph().value( "tasks" ).toArray() ) { const QString taskName = value.toObject().value( "name" ).toString(); @@ -455,6 +486,14 @@ void RimWorkflowJob::runJob() { if ( safeJob && eventRunId == runId ) safeJob->updateTaskState( runId, taskName, state, error ); } ); + QObject::connect( m_runner, + &RiuWorkflowJobRunner::mapItemStateChanged, + m_runner, + [safeJob, + runId]( const QString& eventRunId, const QString& taskName, const QString& item, const QString& state, const QString& error ) + { + if ( safeJob && eventRunId == runId ) safeJob->updateItemState( runId, taskName, item, state, error ); + } ); QObject::connect( m_runner, &RiuWorkflowJobRunner::runFinished, m_runner, diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.h index f7a76b1eb1..17527ddb0e 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.h +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowJob.h @@ -58,7 +58,12 @@ public: QMap taskErrors() const; QString runStatus() const; void updateTaskState( const QString& runId, const QString& taskName, const QString& state, const QString& error ); - void finishRun( const QString& runId, bool succeeded, bool cancelled ); + + // States and errors of the items of mapped tasks, by task and item key + QMap> itemStates() const; + QMap> itemErrors() const; + void updateItemState( const QString& runId, const QString& taskName, const QString& item, const QString& state, const QString& error ); + void finishRun( const QString& runId, bool succeeded, bool cancelled ); protected: void fieldChangedByUi( const caf::PdmFieldHandle* changedField, const QVariant& oldValue, const QVariant& newValue ) override; @@ -74,6 +79,8 @@ private: QMap m_detachedValues; QMap m_taskStates; QMap m_taskErrors; + QMap> m_itemStates; + QMap> m_itemErrors; QString m_runStatus; QString m_activeTask; QString m_runId; diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.cpp new file mode 100644 index 0000000000..a9c94ed796 --- /dev/null +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.cpp @@ -0,0 +1,240 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#include "RimWorkflowMappingBinding.h" + +#include "RimEclipseCase.h" +#include "RimEclipseCaseTools.h" +#include "RimEclipseView.h" +#include "RimProject.h" +#include "RimTools.h" +#include "RimWellPath.h" + +#include "cafPdmUiTreeSelectionEditor.h" + +#include +#include +#include + +CAF_PDM_SOURCE_INIT( RimWorkflowMappingBinding, "WorkflowMappingBinding" ); + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +RimWorkflowMappingBinding::RimWorkflowMappingBinding() +{ + CAF_PDM_InitFieldNoDefault( &m_valueType, "ValueType", "Value Type" ); + m_valueType.uiCapability()->setUiHidden( true ); + + CAF_PDM_InitFieldNoDefault( &m_text, "Text", "Items" ); + m_text.uiCapability()->setUiToolTip( "A JSON object, for example {\"low\": 0.5, \"high\": 2.0}" ); + + CAF_PDM_InitFieldNoDefault( &m_cases, "Cases", "Cases" ); + CAF_PDM_InitFieldNoDefault( &m_wellPaths, "WellPaths", "Well Paths" ); + CAF_PDM_InitFieldNoDefault( &m_views, "Views", "Views" ); + for ( caf::PdmFieldHandle* field : std::initializer_list{ &m_cases, &m_wellPaths, &m_views } ) + { + field->uiCapability()->setUiEditorTypeName( caf::PdmUiTreeSelectionEditor::uiEditorTypeName() ); + field->uiCapability()->setUiLabelPosition( caf::PdmUiItemInfo::LabelPosition::TOP ); + } +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +caf::PdmFieldHandle* RimWorkflowMappingBinding::valueField() +{ + if ( m_valueType() == "EclipseCase" ) return &m_cases; + if ( m_valueType() == "WellPath" ) return &m_wellPaths; + if ( m_valueType() == "View" ) return &m_views; + return &m_text; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +bool RimWorkflowMappingBinding::isObjectReference() const +{ + return !m_valueType().isEmpty(); +} + +//-------------------------------------------------------------------------------------------------- +/// The selected objects as `{key: reference}`, keyed by their names made unique +//-------------------------------------------------------------------------------------------------- +std::vector> RimWorkflowMappingBinding::references() const +{ + std::vector> items; + if ( m_valueType() == "EclipseCase" ) + { + for ( RimEclipseCase* eclipseCase : m_cases.ptrReferencedObjectsByType() ) + { + if ( eclipseCase ) + items.push_back( { eclipseCase->caseUserDescription(), + QJsonObject{ { "__resinsight_ref__", "EclipseCase" }, { "case_id", eclipseCase->caseId() } } } ); + } + } + else if ( m_valueType() == "WellPath" ) + { + for ( RimWellPath* wellPath : m_wellPaths.ptrReferencedObjectsByType() ) + { + if ( wellPath ) + items.push_back( + { wellPath->name(), QJsonObject{ { "__resinsight_ref__", "WellPath" }, { "well_path_name", wellPath->name() } } } ); + } + } + else if ( m_valueType() == "View" ) + { + for ( RimEclipseView* view : m_views.ptrReferencedObjectsByType() ) + { + if ( view ) items.push_back( { view->name(), QJsonObject{ { "__resinsight_ref__", "View" }, { "view_id", view->id() } } } ); + } + } + + QSet used; + for ( auto& [key, reference] : items ) + { + const QString base = key.isEmpty() ? QString( "item" ) : key; + key = base; + for ( int index = 2; used.contains( key ); ++index ) + key = QString( "%1_%2" ).arg( base ).arg( index ); + used.insert( key ); + } + return items; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QJsonValue RimWorkflowMappingBinding::toJsonValue() const +{ + if ( isObjectReference() ) + { + QJsonObject mapping; + for ( const auto& [key, reference] : references() ) + mapping[key] = reference; + return mapping.isEmpty() ? QJsonValue( QJsonValue::Null ) : QJsonValue( mapping ); + } + + if ( !hasValue() ) return QJsonValue::Null; + const QJsonDocument document = QJsonDocument::fromJson( m_text().toUtf8() ); + return document.isObject() ? QJsonValue( document.object() ) : QJsonValue( m_text() ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QString RimWorkflowMappingBinding::displayValue() const +{ + QStringList keys; + if ( isObjectReference() ) + { + for ( const auto& [key, reference] : references() ) + keys.append( key ); + } + else + { + const QJsonValue value = toJsonValue(); + if ( !value.isObject() ) return hasValue() ? "(not a JSON object)" : "(not set)"; + keys = value.toObject().keys(); + } + if ( keys.isEmpty() ) return "(no items)"; + + const QString count = keys.size() == 1 ? QString( "1 item" ) : QString( "%1 items" ).arg( keys.size() ); + QString names = keys.mid( 0, 3 ).join( ", " ); + if ( keys.size() > 3 ) names += ", …"; + return QString( "%1: %2" ).arg( count, names ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QList RimWorkflowMappingBinding::calculateValueOptions( const caf::PdmFieldHandle* fieldNeedingOptions ) +{ + QList options; + if ( fieldNeedingOptions == &m_cases ) RimTools::eclipseCaseOptionItems( &options ); + if ( fieldNeedingOptions == &m_wellPaths ) RimTools::wellPathOptionItems( &options ); + if ( fieldNeedingOptions == &m_views ) + { + for ( RimEclipseCase* eclipseCase : RimEclipseCaseTools::eclipseCases() ) + { + if ( !eclipseCase ) continue; + for ( RimEclipseView* view : eclipseCase->reservoirViews() ) + { + if ( view ) + options.push_back( + caf::PdmOptionItemInfo( QString( "%1 / %2" ).arg( eclipseCase->caseUserDescription(), view->name() ), view ) ); + } + } + } + return options; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RimWorkflowMappingBinding::applySchema( const QJsonObject& fieldSchema ) +{ + m_valueType = fieldSchema.value( "value_resinsight_type" ).toString(); + RimWorkflowFieldBinding::applySchema( fieldSchema ); + + const QJsonObject mapping = fieldSchema.value( "default" ).toObject(); + if ( !isObjectReference() ) + { + if ( fieldSchema.value( "default" ).isObject() ) + m_text = QString::fromUtf8( QJsonDocument( mapping ).toJson( QJsonDocument::Compact ) ); + return; + } + + // Restore the objects of a mapping of references + QSet caseIds; + QSet viewIds; + QSet wellPathNames; + for ( const QJsonValue& value : mapping ) + { + const QJsonObject reference = value.toObject(); + if ( reference.value( "__resinsight_ref__" ).toString() == "EclipseCase" ) + caseIds.insert( reference.value( "case_id" ).toInt( -1 ) ); + if ( reference.value( "__resinsight_ref__" ).toString() == "View" ) viewIds.insert( reference.value( "view_id" ).toInt( -1 ) ); + if ( reference.value( "__resinsight_ref__" ).toString() == "WellPath" ) + wellPathNames.insert( reference.value( "well_path_name" ).toString() ); + } + if ( caseIds.isEmpty() && viewIds.isEmpty() && wellPathNames.isEmpty() ) return; + + std::vector cases; + std::vector views; + for ( RimEclipseCase* eclipseCase : RimEclipseCaseTools::eclipseCases() ) + { + if ( !eclipseCase ) continue; + if ( caseIds.contains( eclipseCase->caseId() ) ) cases.push_back( eclipseCase ); + for ( RimEclipseView* view : eclipseCase->reservoirViews() ) + { + if ( view && viewIds.contains( view->id() ) ) views.push_back( view ); + } + } + std::vector wellPaths; + if ( auto* project = RimProject::current() ) + { + for ( RimWellPath* wellPath : project->allWellPaths() ) + { + if ( wellPath && wellPathNames.contains( wellPath->name() ) ) wellPaths.push_back( wellPath ); + } + } + m_cases.setValue( cases ); + m_views.setValue( views ); + m_wellPaths.setValue( wellPaths ); +} diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.h new file mode 100644 index 0000000000..724d142049 --- /dev/null +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowMappingBinding.h @@ -0,0 +1,57 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// Copyright (C) 2026- Equinor ASA +// +// ResInsight is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY +// WARRANTY; without even the implied warranty of MERCHANTABILITY or +// FITNESS FOR A PARTICULAR PURPOSE. +// +// See the GNU General Public License at +// for more details. +// +///////////////////////////////////////////////////////////////////////////////// + +#pragma once + +#include "RimWorkflowFieldBinding.h" + +#include "cafPdmPtrArrayField.h" + +class RimEclipseCase; +class RimEclipseView; +class RimWellPath; + +//================================================================================================== +/// The `dict[str, Value]` a mapped task runs over. ResInsight objects are picked from the project +/// and keyed by name; other values are entered as a JSON object. +//================================================================================================== +class RimWorkflowMappingBinding : public RimWorkflowFieldBinding +{ + CAF_PDM_HEADER_INIT; + +public: + RimWorkflowMappingBinding(); + + void applySchema( const QJsonObject& fieldSchema ) override; + QJsonValue toJsonValue() const override; + bool isObjectReference() const override; + QString displayValue() const override; + + caf::PdmFieldHandle* valueField() override; + +private: + QList calculateValueOptions( const caf::PdmFieldHandle* fieldNeedingOptions ) override; + + std::vector> references() const; + + caf::PdmField m_valueType; + caf::PdmField m_text; + caf::PdmPtrArrayField m_cases; + caf::PdmPtrArrayField m_wellPaths; + caf::PdmPtrArrayField m_views; +}; diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.cpp index 32cfc52cfd..6c68ca66b0 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.cpp @@ -336,6 +336,8 @@ QString RimWorkflowPortCompatibility::typeName( const QJsonObject& schema, const if ( resolved.contains( "x-ri-python-type" ) ) return shortName( resolved.value( "x-ri-python-type" ).toString() ); const QString type = resolved.value( "type" ).toString(); + if ( type == "object" && resolved.value( "additionalProperties" ).isObject() && !resolved.contains( "properties" ) ) + return QString( "dict[str, %1]" ).arg( typeName( resolved.value( "additionalProperties" ).toObject(), rootSchema ) ); if ( type == "object" ) return resolved.value( "title" ).toString( "object" ); if ( type == "array" ) return QString( "list[%1]" ).arg( typeName( resolved.value( "items" ).toObject(), rootSchema ) ); @@ -368,6 +370,9 @@ QString RimWorkflowPortCompatibility::iconResource( const QJsonObject& schema, c } } if ( resolved.value( "type" ).toString() == "array" ) return iconResource( resolved.value( "items" ).toObject(), rootSchema ); + if ( resolved.value( "type" ).toString() == "object" && resolved.value( "additionalProperties" ).isObject() && + !resolved.contains( "properties" ) ) + return iconResource( resolved.value( "additionalProperties" ).toObject(), rootSchema ); static const QMap icons = { { "Instance", ":/AppLogo48x48.png" }, { "Case", ":/Case48x48.png" }, @@ -462,17 +467,96 @@ bool RimWorkflowPortCompatibility::wouldCreateCycle( const RimWorkflowDefinition } //-------------------------------------------------------------------------------------------------- -/// Checks a new connection. An existing connection to the same input is replaced and ignored here. +/// A mapped task runs once per item: the key and value fields are filled by the map and are not +/// inputs, and the output is `dict[str, Output]` //-------------------------------------------------------------------------------------------------- -std::expected RimWorkflowPortCompatibility::canConnect( const RimWorkflowDefinition& definition, - const QString& from, - const QString& output, - const QString& to, - const QString& input ) +QJsonObject RimWorkflowPortCompatibility::mappedTaskType( const QJsonObject& taskType, const RimWorkflowTaskMap& map ) { + QJsonObject inputSchema = taskType.value( "input_schema" ).toObject(); + QJsonObject properties = inputSchema.value( "properties" ).toObject(); + properties.remove( map.keyAs ); + properties.remove( map.valueAs ); + inputSchema["properties"] = properties; + + QJsonArray required; + for ( const QJsonValue& field : inputSchema.value( "required" ).toArray() ) + { + if ( field.toString() != map.keyAs && field.toString() != map.valueAs ) required.append( field ); + } + inputSchema["required"] = required; + + QJsonObject result = taskType; + result["input_schema"] = inputSchema; + result["output_schema"] = mappedOutputSchema( taskType.value( "output_schema" ).toObject() ); + return result; +} + +//-------------------------------------------------------------------------------------------------- +/// `MappedOutput[Output]`, a `dict[str, Output]`. The output model moves into `$defs`, so its own +/// references still resolve against the new root. +//-------------------------------------------------------------------------------------------------- +QJsonObject RimWorkflowPortCompatibility::mappedOutputSchema( const QJsonObject& outputSchema ) +{ + QJsonObject definitions = outputSchema.value( "$defs" ).toObject(); + QJsonObject item = outputSchema; + item.remove( "$defs" ); + + QString itemKey = "MappedItem"; + while ( definitions.contains( itemKey ) ) + itemKey += "_"; + definitions[itemKey] = item; + + return QJsonObject{ { "type", "object" }, + { "title", QString( "dict[str, %1]" ).arg( typeName( outputSchema, outputSchema ) ) }, + { "additionalProperties", QJsonObject{ { "$ref", "#/$defs/" + itemKey } } }, + { "$defs", definitions }, + { "x-ri-mapped-output", true } }; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +std::optional> + RimWorkflowPortCompatibility::collectTarget( const RimWorkflowPort& expected ) +{ + if ( expected.isWhole() ) return std::nullopt; + + const QJsonObject resolved = RimWorkflowSchemaTools::resolveReferences( expected.schema, expected.rootSchema ); + QJsonArray options; + for ( const char* unionKey : { "anyOf", "oneOf" } ) + { + for ( const QJsonValue& option : resolved.value( unionKey ).toArray() ) + options.append( option ); + } + if ( options.isEmpty() ) options.append( resolved ); + + for ( const QJsonValue& value : options ) + { + const QJsonObject option = RimWorkflowSchemaTools::resolveReferences( value.toObject(), expected.rootSchema ); + const QString type = option.value( "type" ).toString(); + if ( type == "array" && option.value( "items" ).isObject() ) + return std::make_pair( RimWorkflowDefinitionEdge::Collect::List, option.value( "items" ).toObject() ); + if ( type == "object" && option.value( "additionalProperties" ).isObject() && !option.contains( "properties" ) ) + return std::make_pair( RimWorkflowDefinitionEdge::Collect::Dict, option.value( "additionalProperties" ).toObject() ); + } + return std::nullopt; +} + +//-------------------------------------------------------------------------------------------------- +/// Checks a new connection and returns the edge to add +//-------------------------------------------------------------------------------------------------- +std::expected RimWorkflowPortCompatibility::resolveConnection( const RimWorkflowDefinition& definition, + const QString& from, + const QString& output, + const QString& to, + const QString& input ) +{ + using Collect = RimWorkflowDefinitionEdge::Collect; + if ( from == to ) return std::unexpected( "A task cannot be connected to itself" ); if ( !definition.findNode( from ) ) return std::unexpected( QString( "Unknown task '%1'" ).arg( from ) ); - if ( !definition.findNode( to ) ) return std::unexpected( QString( "Unknown task '%1'" ).arg( to ) ); + const auto* downstream = definition.findNode( to ); + if ( !downstream ) return std::unexpected( QString( "Unknown task '%1'" ).arg( to ) ); const QJsonObject upstreamType = definition.taskTypeForNode( from ); const QJsonObject downstreamType = definition.taskTypeForNode( to ); @@ -483,27 +567,109 @@ std::expected RimWorkflowPortCompatibility::canConnect( const Rim if ( !producedPort ) return std::unexpected( QString( "'%1' has no output '%2'" ).arg( from, output ) ); const auto expectedPort = inputPort( downstreamType, input ); if ( !expectedPort ) return std::unexpected( QString( "'%1' has no input '%2'" ).arg( to, input ) ); + if ( downstream->map && input.isEmpty() ) + return std::unexpected( QString( "'%1' is mapped and takes its inputs field by field" ).arg( to ) ); - // Edges into the same input are replaced by the new connection - RimWorkflowDefinition remaining = definition; - std::erase_if( remaining.edges, [&]( const RimWorkflowDefinitionEdge& edge ) { return edge.to == to && edge.input == input; } ); - if ( wouldCreateCycle( remaining, from, to ) ) return std::unexpected( "The connection would create a cycle" ); - - for ( const auto& edge : remaining.incomingEdges( to ) ) + RimWorkflowDefinitionEdge edge{ .from = from, .output = output, .to = to, .input = input }; + if ( !input.isEmpty() && !isCompatible( *producedPort, *expectedPort ) ) { - if ( input.isEmpty() && !edge.isWholeInput() ) - return std::unexpected( QString( "'%1' already has individual inputs connected and cannot also take a whole input" ).arg( to ) ); - if ( !input.isEmpty() && edge.isWholeInput() ) - return std::unexpected( QString( "'%1' takes its whole input from '%2'; disconnect it first" ).arg( to, edge.from ) ); + const auto target = collectTarget( *expectedPort ); + if ( target && isCompatible( producedPort->schema, producedPort->rootSchema, target->second, expectedPort->rootSchema ) ) + { + edge.collect = target->first; + } + else + { + const QString source = output.isEmpty() ? from : from + "." + output; + return std::unexpected( + QString( "%1 is %2, but %3.%4 expects %5" ).arg( source, producedPort->typeName, to, input, expectedPort->typeName ) ); + } } - if ( input.isEmpty() && output.isEmpty() ) return checkWholeToWhole( upstreamType, downstreamType, from, to ); + // Edges into the same input are replaced by the new connection, except other collected members + RimWorkflowDefinition remaining = definition; + std::erase_if( remaining.edges, + [&]( const RimWorkflowDefinitionEdge& existing ) + { return existing.to == to && existing.input == input && ( !edge.isCollected() || existing.collect != edge.collect ); } ); + if ( wouldCreateCycle( remaining, from, to ) ) return std::unexpected( "The connection would create a cycle" ); - if ( !isCompatible( *producedPort, *expectedPort ) ) + QStringList keys; + for ( const auto& existing : remaining.incomingEdges( to ) ) { - const QString source = output.isEmpty() ? from : from + "." + output; - const QString target = input.isEmpty() ? to : to + "." + input; - return std::unexpected( QString( "%1 is %2, but %3 expects %4" ).arg( source, producedPort->typeName, target, expectedPort->typeName ) ); + if ( input.isEmpty() && !existing.isWholeInput() ) + return std::unexpected( QString( "'%1' already has individual inputs connected and cannot also take a whole input" ).arg( to ) ); + if ( !input.isEmpty() && existing.isWholeInput() ) + return std::unexpected( QString( "'%1' takes its whole input from '%2'; disconnect it first" ).arg( to, existing.from ) ); + if ( existing.input != input ) continue; + if ( existing.sameConnection( edge ) ) + return std::unexpected( QString( "%1 is already collected into %2.%3" ).arg( from, to, input ) ); + keys.append( existing.key ); + } + + if ( input.isEmpty() && output.isEmpty() ) + { + if ( auto check = checkWholeToWhole( upstreamType, downstreamType, from, to ); !check ) return std::unexpected( check.error() ); + return edge; + } + if ( input.isEmpty() && !isCompatible( *producedPort, *expectedPort ) ) + { + return std::unexpected( + QString( "%1.%2 is %3, but %4 expects %5" ).arg( from, output, producedPort->typeName, to, expectedPort->typeName ) ); + } + + if ( edge.collect == Collect::Dict ) + { + // The upstream task name is the default key + edge.key = output.isEmpty() ? from : from + "_" + output; + for ( int index = 2; keys.contains( edge.key ); ++index ) + edge.key = QString( "%1_%2" ).arg( output.isEmpty() ? from : from + "_" + output ).arg( index ); + } + return edge; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +std::expected RimWorkflowPortCompatibility::canConnect( const RimWorkflowDefinition& definition, + const QString& from, + const QString& output, + const QString& to, + const QString& input ) +{ + auto edge = resolveConnection( definition, from, output, to, input ); + if ( !edge ) return std::unexpected( edge.error() ); + return {}; +} + +//-------------------------------------------------------------------------------------------------- +/// Type check of one member of a collected input +//-------------------------------------------------------------------------------------------------- +std::expected RimWorkflowPortCompatibility::checkCollectMember( const RimWorkflowDefinition& definition, + const RimWorkflowDefinitionEdge& edge ) +{ + const QJsonObject upstreamType = definition.taskTypeForNode( edge.from ); + const QJsonObject downstreamType = definition.taskTypeForNode( edge.to ); + if ( upstreamType.isEmpty() || downstreamType.isEmpty() ) return {}; + + const auto producedPort = outputPort( upstreamType, edge.output ); + if ( !producedPort ) return std::unexpected( QString( "'%1' has no output '%2'" ).arg( edge.from, edge.output ) ); + const auto expectedPort = inputPort( downstreamType, edge.input ); + if ( !expectedPort || expectedPort->isWhole() ) + return std::unexpected( QString( "'%1' has no input '%2'" ).arg( edge.to, edge.input ) ); + + const auto target = collectTarget( *expectedPort ); + const bool isList = edge.collect == RimWorkflowDefinitionEdge::Collect::List; + if ( !target || target->first != edge.collect ) + { + return std::unexpected( QString( "Input '%1' is %2 and cannot collect %3" ) + .arg( edge.input, expectedPort->typeName, isList ? "a list of members" : "keyed members" ) ); + } + if ( !isCompatible( producedPort->schema, producedPort->rootSchema, target->second, expectedPort->rootSchema ) ) + { + const QString source = edge.output.isEmpty() ? edge.from : edge.from + "." + edge.output; + return std::unexpected( + QString( "%1 is %2, but the members of %3.%4 must be %5" ) + .arg( source, producedPort->typeName, edge.to, edge.input, typeName( target->second, expectedPort->rootSchema ) ) ); } return {}; } diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.h index 97a3dd37ca..a86a068df8 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.h +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowPortCompatibility.h @@ -18,6 +18,8 @@ #pragma once +#include "RimWorkflowDefinition.h" + #include #include @@ -25,8 +27,6 @@ #include #include -struct RimWorkflowDefinition; - //================================================================================================== /// A connectable input or output of a task. An empty name is the whole input or output model. //================================================================================================== @@ -66,6 +66,22 @@ QJsonObject portIcons( const std::vector& ports ); QStringList coveredByWholeOutput( const QJsonObject& upstreamTaskType, const QJsonObject& downstreamTaskType ); bool wouldCreateCycle( const RimWorkflowDefinition& definition, const QString& from, const QString& to ); +// Mapped tasks +QJsonObject mappedTaskType( const QJsonObject& taskType, const RimWorkflowTaskMap& map ); +QJsonObject mappedOutputSchema( const QJsonObject& outputSchema ); + +// Collected inputs: the kind and element schema of a `list[T]` or `dict[str, T]` input +std::optional> collectTarget( const RimWorkflowPort& expected ); + +// The edge to add for a new connection, collecting into a list or dict input when the upstream +// value is an element of it. An existing connection to the same input is replaced, except for +// other members of the same collection. +std::expected resolveConnection( const RimWorkflowDefinition& definition, + const QString& from, + const QString& output, + const QString& to, + const QString& input ); std::expected canConnect( const RimWorkflowDefinition& definition, const QString& from, const QString& output, const QString& to, const QString& input ); +std::expected checkCollectMember( const RimWorkflowDefinition& definition, const RimWorkflowDefinitionEdge& edge ); } // namespace RimWorkflowPortCompatibility diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowTaskInput.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowTaskInput.cpp index 2e31f5a1e5..65ea8d5a22 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowTaskInput.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowTaskInput.cpp @@ -24,6 +24,7 @@ #include "RimWorkflowFilePathBinding.h" #include "RimWorkflowFloatBinding.h" #include "RimWorkflowIntBinding.h" +#include "RimWorkflowMappingBinding.h" #include "RimWorkflowStringBinding.h" #include "RimWorkflowVec3Binding.h" #include "RimWorkflowViewBinding.h" @@ -95,6 +96,8 @@ void RimWorkflowTaskInput::setTaskInfo( const QString& taskType, const QString& //-------------------------------------------------------------------------------------------------- QString RimWorkflowTaskInput::bindingClassKeyword( const QJsonObject& fieldSchema ) { + if ( fieldSchema.value( "map_over" ).toBool() ) return RimWorkflowMappingBinding::classKeywordStatic(); + const QString resinsightType = fieldSchema.value( "resinsight_type" ).toString(); if ( resinsightType == "EclipseCase" ) return RimWorkflowCaseBinding::classKeywordStatic(); if ( resinsightType == "WellPath" ) return RimWorkflowWellPathBinding::classKeywordStatic(); @@ -129,6 +132,7 @@ RimWorkflowFieldBinding* RimWorkflowTaskInput::createBinding( const QJsonObject& if ( keyword == RimWorkflowIntBinding::classKeywordStatic() ) return new RimWorkflowIntBinding; if ( keyword == RimWorkflowFloatBinding::classKeywordStatic() ) return new RimWorkflowFloatBinding; if ( keyword == RimWorkflowArrayBinding::classKeywordStatic() ) return new RimWorkflowArrayBinding; + if ( keyword == RimWorkflowMappingBinding::classKeywordStatic() ) return new RimWorkflowMappingBinding; return new RimWorkflowStringBinding; } diff --git a/ApplicationLibCode/UnitTests/RimWorkflowDefinition-Test.cpp b/ApplicationLibCode/UnitTests/RimWorkflowDefinition-Test.cpp index b00585994e..08fa934c5b 100644 --- a/ApplicationLibCode/UnitTests/RimWorkflowDefinition-Test.cpp +++ b/ApplicationLibCode/UnitTests/RimWorkflowDefinition-Test.cpp @@ -373,3 +373,208 @@ TEST( RimWorkflowDefinitionEdit, names ) EXPECT_EQ( "resinsight_load_model", RimWorkflowDefinitionTools::identifierFrom( "resinsight.load-model" ) ); EXPECT_EQ( "_3d", RimWorkflowDefinitionTools::identifierFrom( "3d" ) ); } + +namespace +{ +QJsonObject numberModel( const QString& title, const QString& field, const QJsonObject& fieldSchema ) +{ + return QJsonObject{ { "type", "object" }, + { "title", title }, + { "properties", QJsonObject{ { field, fieldSchema } } }, + { "required", QJsonArray{ field } } }; +} + +// produce(name: str) -> value: float; scale(name: str, factor: float) -> value: float; +// sum_list(values: list[float]); sum_dict(values: dict[str, float]) +RimWorkflowDefinition collectDefinition() +{ + const QJsonObject number{ { "type", "number" } }; + const QJsonObject output = numberModel( "Value", "value", number ); + const QJsonObject nameInput{ { "type", "object" }, + { "title", "Named" }, + { "properties", QJsonObject{ { "name", QJsonObject{ { "type", "string" } } }, { "factor", number } } }, + { "required", QJsonArray{ "name", "factor" } } }; + + auto definition = RimWorkflowDefinitionTools::createEmpty( "collect" ); + definition.taskTypes.insert( "t.produce", QJsonObject{ { "name", "produce" }, { "input_schema", nameInput }, { "output_schema", output } } ); + definition.taskTypes.insert( "t.sum_list", + QJsonObject{ { "name", "sum_list" }, + { "input_schema", + numberModel( "ListInput", "values", QJsonObject{ { "type", "array" }, { "items", number } } ) }, + { "output_schema", output } } ); + definition.taskTypes.insert( "t.sum_dict", + QJsonObject{ { "name", "sum_dict" }, + { "input_schema", + numberModel( "DictInput", + "values", + QJsonObject{ { "type", "object" }, { "additionalProperties", number } } ) }, + { "output_schema", output } } ); + for ( const auto& [name, taskId] : std::vector>{ { "a", "t.produce" }, + { "b", "t.produce" }, + { "c", "t.produce" }, + { "sum_list", "t.sum_list" }, + { "sum_dict", "t.sum_dict" } } ) + { + definition.nodes.push_back( { .name = name, .taskId = taskId } ); + definition.inputs.insert( name, QJsonObject() ); + } + definition.taskTypes.insert( "t.gather", + QJsonObject{ { "name", "gather" }, + { "input_schema", + QJsonObject{ { "type", "object" }, + { "title", "GatherInput" }, + { "properties", + QJsonObject{ { "results", + QJsonObject{ { "type", "object" }, + { "additionalProperties", + QJsonObject{ { "$ref", "#/$defs/Value" } } } } } } }, + { "required", QJsonArray{ "results" } }, + { "$defs", QJsonObject{ { "Value", output } } } } }, + { "output_schema", output } } ); + definition.nodes.push_back( { .name = "gather", .taskId = "t.gather" } ); + definition.inputs.insert( "gather", QJsonObject() ); + return RimWorkflowDefinitionTools::normalize( definition ); +} + +QStringList taskIssues( const RimWorkflowDefinition& definition, const QString& task ) +{ + std::vector issues; + for ( const auto& issue : RimWorkflowDefinitionTools::validate( definition ) ) + { + if ( issue.task == task ) issues.push_back( issue ); + } + return issueMessages( issues ); +} +} // namespace + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDefinitionEdit, collectIntoList ) +{ + using Collect = RimWorkflowDefinitionEdge::Collect; + auto definition = collectDefinition(); + + definition = + edited( RimWorkflowDefinitionTools::connect( definition, { .from = "a", .output = "value", .to = "sum_list", .input = "values" } ) ); + definition = + edited( RimWorkflowDefinitionTools::connect( definition, { .from = "b", .output = "value", .to = "sum_list", .input = "values" } ) ); + ASSERT_EQ( 2u, definition.incomingEdges( "sum_list" ).size() ); + EXPECT_EQ( Collect::List, definition.incomingEdges( "sum_list" )[0].collect ); + EXPECT_EQ( "a", definition.incomingEdges( "sum_list" )[0].from ); + EXPECT_TRUE( definition.findNode( "sum_list" )->configFields.isEmpty() ); + EXPECT_TRUE( taskIssues( definition, "sum_list" ).isEmpty() ) << taskIssues( definition, "sum_list" ).join( "\n" ).toStdString(); + + // Collecting the same output twice is rejected + EXPECT_FALSE( RimWorkflowDefinitionTools::connect( definition, { .from = "a", .output = "value", .to = "sum_list", .input = "values" } ) ); + + // Reorder the members + const RimWorkflowDefinitionEdge second{ .from = "b", .output = "value", .to = "sum_list", .input = "values" }; + definition = edited( RimWorkflowDefinitionTools::moveCollectMember( definition, second, -1 ) ); + EXPECT_EQ( "b", definition.incomingEdges( "sum_list" )[0].from ); + EXPECT_FALSE( RimWorkflowDefinitionTools::moveCollectMember( definition, second, -1 ) ); + + // Index labels in the graph + const QJsonArray edges = RimWorkflowDefinitionTools::graphFromDefinition( definition, true ).value( "edges" ).toArray(); + QStringList labels; + for ( const QJsonValue& edge : edges ) + labels.append( edge.toObject().value( "from" ).toString() + edge.toObject().value( "label" ).toString() ); + EXPECT_TRUE( labels.contains( "b[0]" ) ); + EXPECT_TRUE( labels.contains( "a[1]" ) ); + + // Disconnecting ignores how the member is collected + definition = edited( + RimWorkflowDefinitionTools::disconnect( definition, { .from = "a", .output = "value", .to = "sum_list", .input = "values" } ) ); + EXPECT_EQ( 1u, definition.incomingEdges( "sum_list" ).size() ); + + // JSON round trip keeps the collected edges + const auto roundTrip = RimWorkflowDefinitionTools::fromJson( RimWorkflowDefinitionTools::toJson( definition ) ); + ASSERT_TRUE( roundTrip.has_value() ); + EXPECT_EQ( definition.edges, roundTrip->edges ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDefinitionEdit, collectIntoDict ) +{ + using Collect = RimWorkflowDefinitionEdge::Collect; + auto definition = collectDefinition(); + + definition = + edited( RimWorkflowDefinitionTools::connect( definition, { .from = "a", .output = "value", .to = "sum_dict", .input = "values" } ) ); + definition = + edited( RimWorkflowDefinitionTools::connect( definition, { .from = "b", .output = "value", .to = "sum_dict", .input = "values" } ) ); + const auto incoming = definition.incomingEdges( "sum_dict" ); + ASSERT_EQ( 2u, incoming.size() ); + EXPECT_EQ( Collect::Dict, incoming[0].collect ); + EXPECT_EQ( "a_value", incoming[0].key ); + EXPECT_EQ( "b_value", incoming[1].key ); + + const RimWorkflowDefinitionEdge first{ .from = "a", .output = "value", .to = "sum_dict", .input = "values" }; + EXPECT_FALSE( RimWorkflowDefinitionTools::setCollectKey( definition, first, "b_value" ) ); + EXPECT_FALSE( RimWorkflowDefinitionTools::setCollectKey( definition, first, "" ) ); + definition = edited( RimWorkflowDefinitionTools::setCollectKey( definition, first, "low" ) ); + EXPECT_EQ( "low", definition.incomingEdges( "sum_dict" )[0].key ); + EXPECT_TRUE( taskIssues( definition, "sum_dict" ).isEmpty() ) << taskIssues( definition, "sum_dict" ).join( "\n" ).toStdString(); + + const auto roundTrip = RimWorkflowDefinitionTools::fromJson( RimWorkflowDefinitionTools::toJson( definition ) ); + ASSERT_TRUE( roundTrip.has_value() ); + EXPECT_EQ( definition.edges, roundTrip->edges ); + + // Duplicate keys are reported + definition.edges[1].key = "low"; + EXPECT_FALSE( taskIssues( definition, "sum_dict" ).isEmpty() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDefinitionEdit, mapTask ) +{ + auto definition = collectDefinition(); + + // The key input must accept a string + EXPECT_FALSE( + RimWorkflowDefinitionTools::setTaskMap( definition, "a", RimWorkflowTaskMap{ .over = "items", .keyAs = "factor", .valueAs = "name" } ) ); + EXPECT_FALSE( + RimWorkflowDefinitionTools::setTaskMap( definition, "a", RimWorkflowTaskMap{ .over = "name", .keyAs = "name", .valueAs = "factor" } ) ); + EXPECT_FALSE( + RimWorkflowDefinitionTools::setTaskMap( definition, "a", RimWorkflowTaskMap{ .over = "items", .keyAs = "name", .valueAs = "name" } ) ); + + const RimWorkflowTaskMap map{ .over = "items", .keyAs = "name", .valueAs = "factor" }; + definition = edited( RimWorkflowDefinitionTools::setTaskMap( definition, "a", map ) ); + ASSERT_TRUE( definition.findNode( "a" )->map.has_value() ); + EXPECT_TRUE( definition.findNode( "a" )->configFields.isEmpty() ); + + // The mapped output is a dict[str, Value]: it feeds a dict[str, Value] input directly + const QJsonObject mappedType = definition.taskTypeForNode( "a" ); + EXPECT_EQ( "dict[str, Value]", RimWorkflowPortCompatibility::outputPorts( mappedType ).front().typeName ); + EXPECT_FALSE( RimWorkflowPortCompatibility::canConnect( definition, "a", "", "sum_dict", "values" ) ); + EXPECT_FALSE( RimWorkflowPortCompatibility::canConnect( definition, "a", "", "sum_list", "values" ) ); + definition = edited( RimWorkflowDefinitionTools::connect( definition, { .from = "a", .to = "gather", .input = "results" } ) ); + EXPECT_FALSE( definition.incomingEdges( "gather" ).front().isCollected() ); + // A mapped task takes its inputs field by field + EXPECT_FALSE( RimWorkflowPortCompatibility::canConnect( definition, "b", "", "a", "" ) ); + + // The mapping is a config value shown as an input port + const QJsonObject task = taskByName( RimWorkflowDefinitionTools::graphFromDefinition( definition, true ), "a" ); + EXPECT_EQ( "items", task.value( "map" ).toObject().value( "over" ).toString() ); + EXPECT_EQ( "dict[str, float]", task.value( "input_types" ).toObject().value( "items" ).toString() ); + bool hasOverField = false; + for ( const QJsonValue& field : task.value( "config_fields" ).toArray() ) + { + if ( field.toObject().value( "name" ).toString() == "items" ) + { + hasOverField = field.toObject().value( "map_over" ).toBool(); + EXPECT_EQ( "number", field.toObject().value( "value_type" ).toString() ); + } + } + EXPECT_TRUE( hasOverField ); + + // JSON round trip + const auto roundTrip = RimWorkflowDefinitionTools::fromJson( RimWorkflowDefinitionTools::toJson( definition ) ); + ASSERT_TRUE( roundTrip.has_value() ); + EXPECT_EQ( definition, roundTrip.value() ); + + // Removing the map restores the inputs and drops the mapping value + definition.inputs.insert( "a", QJsonObject{ { "items", QJsonObject{ { "x", 1.0 } } } } ); + definition = edited( RimWorkflowDefinitionTools::setTaskMap( definition, "a", std::nullopt ) ); + EXPECT_FALSE( definition.findNode( "a" )->map.has_value() ); + EXPECT_EQ( QStringList( { "name", "factor" } ), definition.findNode( "a" )->configFields ); + EXPECT_FALSE( definition.inputs.value( "a" ).toObject().contains( "items" ) ); +} diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.cpp b/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.cpp index 94ae29f421..28112ceabb 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.cpp +++ b/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.cpp @@ -37,10 +37,16 @@ #include "cafCmdFeatureManager.h" #include +#include +#include +#include +#include #include #include #include +#include #include +#include #include #include @@ -240,6 +246,14 @@ void RiuWorkflowEditorWidget::updateJobState() const auto errors = m_job->taskErrors(); for ( auto it = states.cbegin(); it != states.cend(); ++it ) m_graphView->setTaskState( it.key(), it.value(), errors.value( it.key() ) ); + + const auto itemStates = m_job->itemStates(); + const auto itemErrors = m_job->itemErrors(); + for ( auto task = itemStates.cbegin(); task != itemStates.cend(); ++task ) + { + for ( auto item = task.value().cbegin(); item != task.value().cend(); ++item ) + m_graphView->setTaskItemState( task.key(), item.key(), item.value(), itemErrors.value( task.key() ).value( item.key() ) ); + } m_graphView->setRunStatus( m_job->runStatus() ); // A running job locks the workflow @@ -425,6 +439,161 @@ void RiuWorkflowEditorWidget::connectPorts( const QString& from, const QString& applyEdit( RimWorkflowDefinitionTools::connect( m_workflow->definition(), edge ) ); } +//-------------------------------------------------------------------------------------------------- +/// Shows why the workflow cannot be edited +//-------------------------------------------------------------------------------------------------- +bool RiuWorkflowEditorWidget::canEdit() +{ + if ( !m_workflow ) return false; + if ( m_workflow->isEditable() && !m_workflow->isLocked() ) return true; + + showMessage( m_workflow->isLocked() ? QString( "The workflow is locked while a job is running" ) : m_workflow->readOnlyReason() ); + return false; +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +const RimWorkflowDefinitionEdge* RiuWorkflowEditorWidget::findEdge( const RiuWorkflowGraphView::ItemRef& item ) const +{ + if ( !m_workflow || item.kind != RiuWorkflowGraphView::ItemRef::Kind::Edge ) return nullptr; + + const RimWorkflowDefinitionEdge wanted{ .from = item.from, .output = item.output, .to = item.to, .input = item.input }; + for ( const auto& edge : m_workflow->definition().edges ) + { + if ( edge.sameConnection( wanted ) ) return &edge; + } + return nullptr; +} + +//-------------------------------------------------------------------------------------------------- +/// Run the task once per item of a mapping. The dialog edits the current map, if any. +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::editTaskMap( const QString& taskName ) +{ + if ( !canEdit() ) return; + const RimWorkflowDefinition& definition = m_workflow->definition(); + const RimWorkflowDefinitionNode* node = definition.findNode( taskName ); + if ( !node ) return; + + QStringList fields; + for ( const auto& port : RimWorkflowPortCompatibility::inputPorts( definition.taskType( node->taskId ) ) ) + { + if ( !port.isWhole() ) fields.append( port.name ); + } + if ( fields.size() < 2 ) + { + showMessage( QString( "Task '%1' needs two inputs to receive the key and the value of each item" ).arg( taskName ) ); + return; + } + + const RimWorkflowTaskMap current = node->map.value_or( RimWorkflowTaskMap{ .over = "items" } ); + + QDialog dialog( this ); + dialog.setWindowTitle( QString( "Map Over - %1" ).arg( taskName ) ); + auto* form = new QFormLayout( &dialog ); + + auto* overEdit = new QLineEdit( current.over, &dialog ); + overEdit->setToolTip( "Name of the configured mapping. Each key and value of the mapping runs the task once." ); + form->addRow( "Map over:", overEdit ); + + auto addFieldCombo = [&]( const QString& label, const QString& value, const QString& fallback, const QString& toolTip ) + { + auto* combo = new QComboBox( &dialog ); + combo->addItems( fields ); + combo->setCurrentText( fields.contains( value ) ? value : fallback ); + combo->setToolTip( toolTip ); + form->addRow( label, combo ); + return combo; + }; + auto* keyCombo = addFieldCombo( "Key goes to:", current.keyAs, fields.front(), "The input that receives the key of each item" ); + auto* valueCombo = addFieldCombo( "Value goes to:", current.valueAs, fields.at( 1 ), "The input that receives the value of each item" ); + + auto* errorCombo = new QComboBox( &dialog ); + errorCombo->addItem( "Stop at the first failed item", "fail_fast" ); + errorCombo->addItem( "Run every item, then fail if any failed", "collect_all" ); + errorCombo->setCurrentIndex( std::max( 0, errorCombo->findData( current.errorMode ) ) ); + form->addRow( "On failure:", errorCombo ); + + auto* errorLabel = new QLabel( &dialog ); + errorLabel->setStyleSheet( "color: #b00020;" ); + errorLabel->setWordWrap( true ); + form->addRow( errorLabel ); + + auto* buttons = new QDialogButtonBox( QDialogButtonBox::Ok | QDialogButtonBox::Cancel, &dialog ); + form->addRow( buttons ); + + auto mapFromDialog = [&]() + { + return RimWorkflowTaskMap{ .over = overEdit->text().trimmed(), + .keyAs = keyCombo->currentText(), + .valueAs = valueCombo->currentText(), + .errorMode = errorCombo->currentData().toString() }; + }; + auto updateValidity = [&]() + { + const auto valid = RimWorkflowDefinitionTools::checkTaskMap( definition, taskName, mapFromDialog() ); + errorLabel->setText( valid ? QString() : valid.error() ); + errorLabel->setVisible( !valid ); + buttons->button( QDialogButtonBox::Ok )->setEnabled( valid.has_value() ); + }; + QObject::connect( overEdit, &QLineEdit::textChanged, &dialog, updateValidity ); + QObject::connect( keyCombo, &QComboBox::currentIndexChanged, &dialog, updateValidity ); + QObject::connect( valueCombo, &QComboBox::currentIndexChanged, &dialog, updateValidity ); + QObject::connect( buttons, &QDialogButtonBox::accepted, &dialog, &QDialog::accept ); + QObject::connect( buttons, &QDialogButtonBox::rejected, &dialog, &QDialog::reject ); + updateValidity(); + + if ( dialog.exec() != QDialog::Accepted ) return; + + const RimWorkflowTaskMap map = mapFromDialog(); + if ( node->map && *node->map == map ) return; + applyEdit( RimWorkflowDefinitionTools::setTaskMap( definition, taskName, map ) ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::removeTaskMap( const QString& taskName ) +{ + if ( !canEdit() ) return; + applyEdit( RimWorkflowDefinitionTools::setTaskMap( m_workflow->definition(), taskName, std::nullopt ) ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::renameCollectKey( const RiuWorkflowGraphView::ItemRef& item ) +{ + if ( !canEdit() ) return; + const RimWorkflowDefinitionEdge* edge = findEdge( item ); + if ( !edge || edge->collect != RimWorkflowDefinitionEdge::Collect::Dict ) return; + + bool ok = false; + const QString key = QInputDialog::getText( this, + "Rename Key", + QString( "Key of '%1' in '%2':" ).arg( edge->from, edge->input ), + QLineEdit::Normal, + edge->key, + &ok ) + .trimmed(); + if ( !ok || key == edge->key ) return; + + applyEdit( RimWorkflowDefinitionTools::setCollectKey( m_workflow->definition(), *edge, key ) ); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::moveCollectMember( const RiuWorkflowGraphView::ItemRef& item, int delta ) +{ + if ( !canEdit() ) return; + const RimWorkflowDefinitionEdge* edge = findEdge( item ); + if ( !edge ) return; + + applyEdit( RimWorkflowDefinitionTools::moveCollectMember( m_workflow->definition(), *edge, delta ) ); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -478,12 +647,14 @@ void RiuWorkflowEditorWidget::showContextMenu( const RiuWorkflowGraphView::ItemR menu.addAction( action ); } appendOptionalInputsMenu( &menu, item.task ); + appendMapMenu( &menu, item.task ); menu.addSeparator(); if ( auto* action = featureAction( "RicDeleteWorkflowTaskFeature", "Delete Task", data ) ) menu.addAction( action ); } else if ( item.kind == RiuWorkflowGraphView::ItemRef::Kind::Edge && editable ) { const QVariantMap data{ { "from", item.from }, { "output", item.output }, { "to", item.to }, { "input", item.input } }; + appendCollectMenu( &menu, item ); if ( auto* action = featureAction( "RicDeleteWorkflowConnectionFeature", "Delete Connection", data ) ) menu.addAction( action ); } else @@ -553,6 +724,64 @@ void RiuWorkflowEditorWidget::appendAddTaskMenu( QMenu* menu, const QPointF& sce addMenu->setToolTipsVisible( true ); } +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::appendMapMenu( QMenu* menu, const QString& taskName ) +{ + const RimWorkflowDefinitionNode* node = m_workflow->definition().findNode( taskName ); + if ( !node ) return; + + const QVariantMap data{ { "task", taskName } }; + if ( auto* action = featureAction( "RicMapWorkflowTaskFeature", node->map ? "Edit Map..." : "Map Over...", data ) ) + { + action->setToolTip( "Run the task once for each item of a mapping and collect the outputs by key" ); + menu->addAction( action ); + } + if ( node->map ) + { + if ( auto* action = featureAction( "RicRemoveWorkflowTaskMapFeature", "Remove Map", data ) ) menu->addAction( action ); + } +} + +//-------------------------------------------------------------------------------------------------- +/// Collected connections: dict members have a key, list members an order +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowEditorWidget::appendCollectMenu( QMenu* menu, const RiuWorkflowGraphView::ItemRef& item ) +{ + const RimWorkflowDefinitionEdge* edge = findEdge( item ); + if ( !edge || !edge->isCollected() ) return; + + const QVariantMap data{ { "from", item.from }, { "output", item.output }, { "to", item.to }, { "input", item.input } }; + if ( edge->collect == RimWorkflowDefinitionEdge::Collect::Dict ) + { + if ( auto* action = featureAction( "RicRenameWorkflowCollectKeyFeature", "Rename Key...", data ) ) menu->addAction( action ); + } + else + { + int index = 0; + int count = 0; + for ( const auto& other : m_workflow->definition().edges ) + { + if ( other.to != edge->to || other.input != edge->input || !other.isCollected() ) continue; + if ( &other == edge ) index = count; + ++count; + } + + for ( const auto& [text, delta] : { std::pair{ QString( "Move Earlier" ), -1 }, std::pair{ QString( "Move Later" ), 1 } } ) + { + QVariantMap moveData = data; + moveData["delta"] = delta; + if ( auto* action = featureAction( "RicMoveWorkflowCollectMemberFeature", text, moveData ) ) + { + action->setEnabled( index + delta >= 0 && index + delta < count ); + menu->addAction( action ); + } + } + } + menu->addSeparator(); +} + //-------------------------------------------------------------------------------------------------- /// Optional inputs are configured by the user only when checked; connected inputs are disabled //-------------------------------------------------------------------------------------------------- @@ -563,7 +792,7 @@ void RiuWorkflowEditorWidget::appendOptionalInputsMenu( QMenu* menu, const QStri if ( !node ) return; std::vector optionalPorts; - for ( const auto& port : RimWorkflowPortCompatibility::inputPorts( definition.taskType( node->taskId ) ) ) + for ( const auto& port : RimWorkflowPortCompatibility::inputPorts( definition.taskTypeForNode( taskName ) ) ) { if ( !port.isWhole() && !port.required ) optionalPorts.push_back( port ); } diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.h b/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.h index 0b50b6f8ec..83e5e0fa98 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.h +++ b/ApplicationLibCode/UserInterface/RiuWorkflowEditorWidget.h @@ -34,6 +34,7 @@ class QSplitter; class RimWorkflow; class RimWorkflowJob; struct RimWorkflowDefinition; +struct RimWorkflowDefinitionEdge; class RiuWorkflowTaskPalette; class RiuWorkflowValidationRunner; @@ -69,6 +70,10 @@ public: void renameTask( const QString& taskName ); void setResultTask( const QString& taskName ); void setOptionalInputConfigured( const QString& taskName, const QString& field, bool configured ); + void editTaskMap( const QString& taskName ); + void removeTaskMap( const QString& taskName ); + void renameCollectKey( const RiuWorkflowGraphView::ItemRef& edge ); + void moveCollectMember( const RiuWorkflowGraphView::ItemRef& edge, int delta ); void connectPorts( const QString& from, const QString& output, const QString& to, const QString& input ); void showMessage( const QString& message ); @@ -80,6 +85,12 @@ private: void showWorkflow( RiuWorkflowGraphView::ViewportPolicy policy ); void showContextMenu( const RiuWorkflowGraphView::ItemRef& item, const QPoint& globalPos, const QPointF& scenePos ); void appendAddTaskMenu( QMenu* menu, const QPointF& scenePos ); + void appendMapMenu( QMenu* menu, const QString& taskName ); + void appendCollectMenu( QMenu* menu, const RiuWorkflowGraphView::ItemRef& item ); + bool canEdit(); + + const RimWorkflowDefinitionEdge* findEdge( const RiuWorkflowGraphView::ItemRef& item ) const; + void appendOptionalInputsMenu( QMenu* menu, const QString& taskName ); void updateActions(); void updateStatus(); diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp index 30a262bb4c..c826f1ffd7 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp +++ b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp @@ -51,6 +51,8 @@ constexpr qreal nodeWidth = 240.0; constexpr qreal configWidth = 190.0; constexpr qreal portStep = 40.0; constexpr qreal portTop = 51.0; +constexpr qreal subtitleGap = 16.0; +constexpr qreal stackOffset = 5.0; constexpr qreal columnStep = 650.0; constexpr qreal configGap = 110.0; constexpr qreal rowGap = 55.0; @@ -73,6 +75,7 @@ const QColor errorColor( 190, 45, 45 ); const QColor warningColor( 200, 130, 20 ); const QColor edgeColor( 75, 95, 115 ); const QColor selectedEdgeColor( 30, 120, 220 ); +const QColor mapColor( 95, 80, 160 ); QString keyForField( const QString& field ) { @@ -127,7 +130,8 @@ struct PortInfo QString toolTip; QString typeName{}; QString iconResource{}; - Style style = Style::Normal; + Style style = Style::Normal; + bool collect = false; }; //================================================================================================== @@ -169,7 +173,11 @@ private: class GraphNode : public QGraphicsRectItem { public: - GraphNode( const QString& name, const std::vector& inputs, const std::vector& outputs, bool isConfig = false ); + GraphNode( const QString& name, + const std::vector& inputs, + const std::vector& outputs, + bool isConfig = false, + const QString& subtitle = {} ); qreal height() const; QPointF portPosition( const QString& key, bool output ) const; @@ -179,6 +187,8 @@ public: std::vector ports() const; void setConfigValue( const QString& fieldName, const QString& value ); void setTaskState( const QString& state, const QString& error = {} ); + void setItemState( const QString& item, const QString& state, const QString& error = {} ); + void setSubtitleToolTip( const QString& toolTip ); void setTitle( const QString& title, const QString& toolTip ); void setIssues( const QJsonArray& issues ); void setUnknownType( bool unknown ); @@ -200,6 +210,11 @@ private: QGraphicsTextItem* m_title = nullptr; QGraphicsTextItem* m_stateLabel = nullptr; QGraphicsTextItem* m_issueLabel = nullptr; + QGraphicsTextItem* m_subtitle = nullptr; + std::vector m_stack; + qreal m_portTop = portTop; + QMap m_itemStates; + QMap m_itemErrors; QString m_state; QString m_stateError; QString m_baseToolTip; @@ -224,6 +239,7 @@ public: GraphNode* toNode() const { return m_to; } QString outputField() const { return fieldForKey( m_outputKey ); } QString inputField() const { return fieldForKey( m_inputKey ); } + void setLabel( const QString& label, const QString& toolTip ); protected: QVariant itemChange( GraphicsItemChange change, const QVariant& value ) override; @@ -237,6 +253,7 @@ private: QString m_inputKey; bool m_isConfig; QGraphicsPolygonItem* m_arrow; + QGraphicsTextItem* m_label = nullptr; }; //-------------------------------------------------------------------------------------------------- @@ -270,6 +287,17 @@ PortItem::PortItem( GraphNode* node, const PortInfo& info, bool output, bool con setHighlight( Highlight::None ); setAcceptedMouseButtons( Qt::NoButton ); setZValue( 2 ); + + // An outer ring marks inputs that collect several connections + if ( info.collect ) + { + const qreal ring = portRadius + 3.0; + auto* outer = new QGraphicsEllipseItem( center.x() - ring, center.y() - ring, 2 * ring, 2 * ring, node ); + outer->setPen( QPen( color, 1.2 ) ); + outer->setBrush( Qt::NoBrush ); + outer->setAcceptedMouseButtons( Qt::NoButton ); + outer->setZValue( 1.5 ); + } } //-------------------------------------------------------------------------------------------------- @@ -302,14 +330,19 @@ void PortItem::setHighlight( Highlight highlight, const QString& reason ) //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- -GraphNode::GraphNode( const QString& name, const std::vector& inputs, const std::vector& outputs, bool isConfig ) +GraphNode::GraphNode( const QString& name, + const std::vector& inputs, + const std::vector& outputs, + bool isConfig, + const QString& subtitle ) : m_name( name ) , m_isConfig( isConfig ) , m_width( isConfig ? configWidth : nodeWidth ) + , m_portTop( subtitle.isEmpty() ? portTop : portTop + subtitleGap ) , m_baseToolTip( name ) { const qsizetype rows = std::max( { qsizetype( 1 ), qsizetype( inputs.size() ), qsizetype( outputs.size() ) } ); - setRect( 0, 0, m_width, portTop + portStep * rows + 12 ); + setRect( 0, 0, m_width, m_portTop + portStep * rows + 12 ); setFlags( ItemIsMovable | ItemIsSelectable | ItemSendsGeometryChanges ); setZValue( 1 ); @@ -336,6 +369,28 @@ GraphNode::GraphNode( const QString& name, const std::vector& inputs, m_issueLabel->setAcceptedMouseButtons( Qt::NoButton ); } + // A mapped task runs once per item: drawn as a stack of boxes with a line saying what it maps over + if ( !subtitle.isEmpty() ) + { + for ( int depth = 2; depth >= 1; --depth ) + { + auto* card = new QGraphicsRectItem( rect().translated( depth * stackOffset, depth * stackOffset ), this ); + card->setFlag( ItemStacksBehindParent ); + card->setAcceptedMouseButtons( Qt::NoButton ); + m_stack.push_back( card ); + } + + m_subtitle = new QGraphicsTextItem( this ); + QFont font; + font.setPointSize( 8 ); + font.setItalic( true ); + m_subtitle->setFont( font ); + m_subtitle->setDefaultTextColor( mapColor ); + m_subtitle->setPlainText( QFontMetrics( font ).elidedText( subtitle, Qt::ElideRight, m_width - 20 ) ); + m_subtitle->setPos( 10, 24 ); + m_subtitle->setAcceptedMouseButtons( Qt::NoButton ); + } + addPorts( inputs, false ); addPorts( outputs, true ); setTitle( isConfig ? "Config: " + name : name, {} ); @@ -370,7 +425,7 @@ void GraphNode::addPorts( const std::vector& ports, bool output ) { const PortInfo& info = ports[row]; const qreal x = output ? m_width : 0.0; - const qreal y = portTop + row * portStep; + const qreal y = m_portTop + row * portStep; auto* port = new PortItem( this, info, output, m_isConfig, QPointF( x, y ) ); portMap.insert( info.key, port ); @@ -473,9 +528,36 @@ void GraphNode::setTaskState( const QString& state, const QString& error ) if ( m_isConfig ) return; m_state = state; m_stateError = error; + if ( state.isEmpty() ) + { + m_itemStates.clear(); + m_itemErrors.clear(); + } applyStyle(); } +//-------------------------------------------------------------------------------------------------- +/// The state of one item of a mapped task +//-------------------------------------------------------------------------------------------------- +void GraphNode::setItemState( const QString& item, const QString& state, const QString& error ) +{ + if ( m_isConfig ) return; + m_itemStates[item] = state; + if ( error.isEmpty() ) + m_itemErrors.remove( item ); + else + m_itemErrors[item] = error; + applyStyle(); +} + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void GraphNode::setSubtitleToolTip( const QString& toolTip ) +{ + if ( m_subtitle ) m_subtitle->setToolTip( toolTip ); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -555,11 +637,31 @@ void GraphNode::applyStyle() pen.setStyle( style ); setPen( pen ); setBrush( background ); + for ( QGraphicsRectItem* card : m_stack ) + { + card->setPen( QPen( border, 1.0 ) ); + card->setBrush( background.darker( 104 ) ); + } + // Item progress of a mapped task, such as "Running 3✓ 1✗" + int done = 0; + int failed = 0; + for ( const QString& itemState : m_itemStates ) + { + if ( itemState == "completed" ) ++done; + if ( itemState == "failed" ) ++failed; + } if ( m_stateLabel ) { + QString text = m_state.isEmpty() ? "" : m_state.at( 0 ).toUpper() + m_state.mid( 1 ); + if ( !m_itemStates.isEmpty() ) + { + text += QString( " %1/%2" ).arg( done ).arg( m_itemStates.size() ); + if ( failed > 0 ) text += QString::fromUtf8( " ✗%1" ).arg( failed ); + } m_stateLabel->setDefaultTextColor( border ); - m_stateLabel->setPlainText( m_state.isEmpty() ? "" : m_state.at( 0 ).toUpper() + m_state.mid( 1 ) ); + m_stateLabel->setPlainText( text ); + m_stateLabel->setPos( m_width - m_stateLabel->boundingRect().width() - 6, 5 ); } if ( m_issueLabel ) @@ -574,6 +676,17 @@ void GraphNode::applyStyle() QStringList toolTip{ m_baseToolTip }; if ( !m_stateError.isEmpty() ) toolTip << m_stateError; + if ( !m_itemStates.isEmpty() ) + { + QStringList items; + for ( auto it = m_itemStates.cbegin(); it != m_itemStates.cend(); ++it ) + { + QString line = QString( "%1: %2" ).arg( it.key(), it.value() ); + if ( m_itemErrors.contains( it.key() ) ) line += " - " + m_itemErrors.value( it.key() ); + items << line; + } + toolTip << "Items:\n" + items.join( "\n" ); + } if ( !m_issueText.isEmpty() ) toolTip << m_issueText; setToolTip( toolTip.join( "\n\n" ) ); } @@ -650,6 +763,26 @@ GraphEdge::GraphEdge( GraphNode* from, QString outputKey, GraphNode* to, QString updatePath(); } +//-------------------------------------------------------------------------------------------------- +/// A label near the target, such as the index or key of a collected member +//-------------------------------------------------------------------------------------------------- +void GraphEdge::setLabel( const QString& label, const QString& toolTip ) +{ + if ( label.isEmpty() ) return; + if ( !m_label ) + { + m_label = new QGraphicsTextItem( this ); + QFont font; + font.setPointSize( 8 ); + m_label->setFont( font ); + m_label->setDefaultTextColor( edgeColor ); + m_label->setAcceptedMouseButtons( Qt::NoButton ); + } + m_label->setPlainText( label ); + if ( !toolTip.isEmpty() ) m_label->setToolTip( toolTip ); + updatePath(); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -674,6 +807,14 @@ void GraphEdge::updatePath() path.cubicTo( start + QPointF( bend, 0 ), end - QPointF( bend, 0 ), end ); setPath( path ); m_arrow->setPolygon( QPolygonF{ end, end + QPointF( -9, -5 ), end + QPointF( -9, 5 ) } ); + + // Place the label on the curve shortly before the target + if ( m_label ) + { + const QPointF anchor = path.pointAtPercent( 0.85 ); + const QRectF bounds = m_label->boundingRect(); + m_label->setPos( anchor.x() - bounds.width() / 2.0, anchor.y() - bounds.height() ); + } } //-------------------------------------------------------------------------------------------------- @@ -734,6 +875,18 @@ std::pair, std::vector> editModePorts( const QJs if ( port.value( "configured" ).toBool() ) toolTip << "Set in the job configuration"; if ( port.value( "covered" ).toBool() ) toolTip << "Provided by a whole-output connection"; info.toolTip = toolTip.join( "\n" ); + if ( !port.value( "collect" ).toString().isEmpty() ) + { + info.collect = true; + toolTip << ( port.value( "collect" ).toString() == "dict" ? "Connect several outputs to collect them, each under its own key" + : "Connect several outputs to collect them into a list" ); + info.toolTip = toolTip.join( "\n" ); + } + if ( port.value( "map_over" ).toBool() ) + { + info.toolTip = QString( "%1: %2\n%3\nThe task runs once per item; set the items in the job configuration" ) + .arg( info.label, info.typeName, port.value( "description" ).toString() ); + } if ( !provided && !field.isEmpty() ) info.style = required ? PortInfo::Style::Missing : PortInfo::Style::Hollow; inputs.push_back( info ); } @@ -991,7 +1144,23 @@ void RiuWorkflowGraphView::showGraph( const QJsonObject& graph, const QString& e outputs = portInfos( out, true, data.value( "output_types" ).toObject(), data.value( "output_icons" ).toObject() ); } - auto* node = new GraphNode( name, inputs, outputs ); + const QJsonObject map = data.value( "map" ).toObject(); + QString subtitle; + if ( !map.isEmpty() ) subtitle = QString::fromUtf8( "⟳ map over %1" ).arg( map.value( "over" ).toString() ); + auto* node = new GraphNode( name, inputs, outputs, false, subtitle ); + if ( !map.isEmpty() ) + { + QStringList mapToolTip{ + QString( "Runs once per item of '%1' (%2)" ).arg( map.value( "over" ).toString(), map.value( "type" ).toString() ) }; + if ( !map.value( "key_as" ).toString().isEmpty() ) mapToolTip << QString( "Key → %1" ).arg( map.value( "key_as" ).toString() ); + if ( !map.value( "value_as" ).toString().isEmpty() ) + mapToolTip << QString( "Value → %1" ).arg( map.value( "value_as" ).toString() ); + mapToolTip << ( map.value( "error_mode" ).toString() == "collect_all" + ? "Every item runs; the task fails afterwards if any item failed" + : "The first failed item stops the task" ); + mapToolTip << "The output is a dict with one entry per item"; + node->setSubtitleToolTip( mapToolTip.join( "\n" ) ); + } const QString taskId = data.value( "task_id" ).toString(); const QString description = data.value( "description" ).toString(); QStringList toolTip{ name }; @@ -1092,12 +1261,20 @@ void RiuWorkflowGraphView::showGraph( const QJsonObject& graph, const QString& e const QString from = edge.value( "from" ).toString(); const QString to = edge.value( "to" ).toString(); if ( !m_taskNodes.contains( from ) || !m_taskNodes.contains( to ) || from == to ) continue; - m_scene->addItem( new GraphEdge( m_taskNodes.value( from ), + auto* graphEdge = new GraphEdge( m_taskNodes.value( from ), keyForField( edge.value( "output" ).toString() ), m_taskNodes.value( to ), keyForField( edge.value( "input" ).toString() ), false, - selectableEdges ) ); + selectableEdges ); + if ( edge.contains( "collect" ) ) + { + const bool keyed = edge.value( "collect" ).toString() == "dict"; + graphEdge->setLabel( edge.value( "label" ).toString(), + keyed ? QString( "Collected under the key '%1'" ).arg( edge.value( "label" ).toString() ) + : QString( "Collected as item %1 of the list" ).arg( edge.value( "label" ).toString() ) ); + } + m_scene->addItem( graphEdge ); } for ( auto it = m_configNodes.cbegin(); it != m_configNodes.cend(); ++it ) @@ -1237,6 +1414,14 @@ void RiuWorkflowGraphView::setTaskState( const QString& taskName, const QString& if ( auto* node = m_taskNodes.value( taskName, nullptr ) ) node->setTaskState( state, error ); } +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +void RiuWorkflowGraphView::setTaskItemState( const QString& taskName, const QString& item, const QString& state, const QString& error ) +{ + if ( auto* node = m_taskNodes.value( taskName, nullptr ) ) node->setItemState( item, state, error ); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.h b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.h index 3722f7c78b..dcaabdd44f 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.h +++ b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.h @@ -97,6 +97,7 @@ public: void setTaskInputValue( const QString& taskName, const QString& fieldName, const QString& value ); void resetTaskStates(); void setTaskState( const QString& taskName, const QString& state, const QString& error = {} ); + void setTaskItemState( const QString& taskName, const QString& item, const QString& state, const QString& error = {} ); void setRunStatus( const QString& status ); signals: diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.cpp b/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.cpp index 15fddbf32d..3e078f0225 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.cpp +++ b/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.cpp @@ -169,6 +169,16 @@ void RiuWorkflowJobRunner::drainLines( QByteArray& buffer, RILogLevel level ) event.value( "error" ).toString() ); continue; } + if ( event.value( "event" ) == "map_item_state" && event.value( "run_id" ).isString() && event.value( "task" ).isString() && + event.value( "item" ).isString() && event.value( "state" ).isString() ) + { + emit mapItemStateChanged( event.value( "run_id" ).toString(), + event.value( "task" ).toString(), + event.value( "item" ).toString(), + event.value( "state" ).toString(), + event.value( "error" ).toString() ); + continue; + } } } diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.h b/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.h index ff525f706e..21fe7e7901 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.h +++ b/ApplicationLibCode/UserInterface/RiuWorkflowJobRunner.h @@ -38,6 +38,7 @@ public: signals: void taskStateChanged( const QString& runId, const QString& taskName, const QString& state, const QString& error ); + void mapItemStateChanged( const QString& runId, const QString& taskName, const QString& item, const QString& state, const QString& error ); void runFinished( bool succeeded, bool cancelled ); private slots: diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/definition.py b/GrpcInterface/Python/rips/taskmaestro_helper/definition.py index c511d3ad01..581f31636b 100644 --- a/GrpcInterface/Python/rips/taskmaestro_helper/definition.py +++ b/GrpcInterface/Python/rips/taskmaestro_helper/definition.py @@ -35,9 +35,29 @@ BACKUP_SUFFIX = ".bak" def _edge( - upstream: str, output: str | None, downstream: str, input_field: str | None + upstream: str, + output: str | None, + downstream: str, + input_field: str | None, + *, + collect: str | None = None, + key: str | None = None, ) -> dict[str, Any]: - return {"from": upstream, "output": output, "to": downstream, "input": input_field} + edge = {"from": upstream, "output": output, "to": downstream, "input": input_field} + if collect is not None: + # A member of a collected input: "list" members keep their order, "dict" members a key + edge["collect"] = collect + edge["key"] = key + return edge + + +def _map_json(task_map: Any) -> dict[str, Any]: + return { + "over": task_map.over, + "key_as": task_map.key_as, + "value_as": task_map.value_as, + "error_mode": task_map.error_mode, + } def _empty_definition(name: str) -> dict[str, Any]: @@ -202,8 +222,7 @@ def _load_nodes( if cls is not None: _add_task_type(definition, entry.task, cls) if entry.map is not None: - node["map"] = entry.map.model_dump() - definition["readonly_reasons"].append(f"Task '{name}' is mapped ('map:')") + node["map"] = _map_json(entry.map) if any(existing["name"] == name for existing in definition["nodes"]): raise HelperError(f"Duplicate task name '{name}'", task=name) node["name"] = name @@ -271,9 +290,7 @@ def _load_edges( if isinstance(deps, dict): for field, raw in deps.items(): if isinstance(raw, dict) and set(raw) == {"collect"}: - definition["readonly_reasons"].append( - f"Task '{name}' collects outputs into '{field}' ('collect:')" - ) + _load_collect(definition, candidates, raw["collect"], name, field) continue ref, output = _field_reference(raw, name, field) upstream = _resolve_reference(candidates, ref, name) @@ -284,6 +301,33 @@ def _load_edges( definition["edges"].append(_edge(upstream, output, name, None)) +def _load_collect( + definition: dict[str, Any], + candidates: dict[str, set[str]], + members: Any, + name: str, + field: str, +) -> None: + if isinstance(members, list): + keyed: list[tuple[str | None, Any]] = [(None, member) for member in members] + kind = "list" + elif isinstance(members, dict): + keyed = [(str(key), member) for key, member in members.items()] + kind = "dict" + else: + raise HelperError( + f"'collect' for field '{field}' on task '{name}' must be a list or mapping", + task=name, + field=field, + ) + for key, member in keyed: + ref, output = _field_reference(member, name, field) + upstream = _resolve_reference(candidates, ref, name) + definition["edges"].append( + _edge(upstream, output, name, field, collect=kind, key=key) + ) + + def _load_inputs(definition: dict[str, Any], input_path: Path | None) -> None: raw_input: Any = {} if input_path is not None and input_path.exists(): @@ -442,8 +486,9 @@ def definition_from_workflow( else: node["task"] = task_id _add_task_type(definition, task_id, cls) - if workflow.get_task_map(name) is not None: - reasons.append(f"Task '{name}' is mapped") + task_map = workflow.get_task_map(name) + if task_map is not None: + node["map"] = _map_json(task_map) definition["nodes"].append(node) definition["inputs"][name] = {} @@ -457,8 +502,23 @@ def definition_from_workflow( else: for field, ref in deps.items(): if isinstance(ref, CollectionRef): - reasons.append(f"Task '{name}' collects outputs into '{field}'") - has_dependents.update(item.task_name for item in ref.output_refs()) + kind = "list" if ref.kind == "positional" else "dict" + members = ( + [(None, member) for member in ref.positional_members] + if ref.kind == "positional" + else list(ref.keyed_members) + ) + for key, member in members: + definition["edges"].append( + _edge( + member.task_name, + member.output_field, + name, + field, + collect=kind, + key=key, + ) + ) elif isinstance(ref, tuple): definition["edges"].append(_edge(ref[0], ref[1], name, field)) else: @@ -569,6 +629,9 @@ def _dependencies(definition: dict[str, Any], node_name: str) -> Any: return None deps: dict[str, Any] = {} for edge in fields: + if edge.get("collect"): + _add_collect_member(deps, edge, node_name) + continue if edge["input"] in deps: raise HelperError( f"Input '{node_name}.{edge['input']}' has more than one connection", @@ -579,6 +642,77 @@ def _dependencies(definition: dict[str, Any], node_name: str) -> Any: return deps +def _add_collect_member( + deps: dict[str, Any], edge: dict[str, Any], node_name: str +) -> None: + field = edge["input"] + kind = edge["collect"] + if kind not in ("list", "dict"): + raise HelperError( + f"Unknown collect kind '{kind}' for input '{node_name}.{field}'", + task=node_name, + field=field, + ) + existing = deps.get(field) + if existing is None: + existing = {"collect": [] if kind == "list" else {}} + deps[field] = existing + members = existing.get("collect") if isinstance(existing, dict) else None + if not isinstance(members, list if kind == "list" else dict): + raise HelperError( + f"Input '{node_name}.{field}' mixes collected and direct connections, " + "or list and keyed members", + task=node_name, + field=field, + ) + reference = _reference(edge["from"], edge.get("output")) + if kind == "list": + members.append(reference) + return + key = edge.get("key") + if not isinstance(key, str) or not key: + raise HelperError( + f"A keyed member of '{node_name}.{field}' has no key", + task=node_name, + field=field, + ) + if key in members: + raise HelperError( + f"Duplicate key '{key}' in '{node_name}.{field}'", + task=node_name, + field=field, + ) + members[key] = reference + + +def _map_document(task_map: Any, node_name: str) -> dict[str, Any]: + if not isinstance(task_map, dict): + raise HelperError(f"Invalid map for task '{node_name}'", task=node_name) + document: dict[str, Any] = {} + for key in ("over", "key_as", "value_as"): + value = task_map.get(key) + if not _is_identifier(value): + raise HelperError( + f"Invalid map '{key}' for task '{node_name}': {value!r}", task=node_name + ) + document[key] = value + if document["key_as"] == document["value_as"]: + raise HelperError( + f"The map of task '{node_name}' uses '{document['key_as']}' for both the key " + "and the value", + task=node_name, + ) + error_mode = task_map.get("error_mode") or "fail_fast" + if error_mode not in ("fail_fast", "collect_all"): + raise HelperError( + f"Invalid map error mode '{error_mode}' for task '{node_name}'", + task=node_name, + ) + if error_mode != "fail_fast": + document["error_mode"] = error_mode + return document + + def workflow_document(definition: dict[str, Any]) -> dict[str, Any]: """Build the ``workflow.yaml`` document for a definition.""" _check_saveable(definition) @@ -595,6 +729,8 @@ def workflow_document(definition: dict[str, Any]) -> dict[str, Any]: entry["depends_on"] = deps if node.get("config_fields"): entry["config_fields"] = _FlowList(node["config_fields"]) + if node.get("map"): + entry["map"] = _map_document(node["map"], node["name"]) tasks.append(entry) section: dict[str, Any] = {"name": definition["name"]} @@ -614,9 +750,11 @@ def input_document(definition: dict[str, Any]) -> dict[str, Any]: document: dict[str, Any] = {} for node in definition["nodes"]: values = inputs.get(node["name"]) or {} - config_fields = node.get("config_fields") or [] + fields = list(node.get("config_fields") or []) + if node.get("map"): + fields.append(node["map"].get("over")) document[node["name"]] = strip_refs( - {field: value for field, value in values.items() if field in config_fields} + {field: value for field, value in values.items() if field in fields} ) return document diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/run.py b/GrpcInterface/Python/rips/taskmaestro_helper/run.py index afc027ebd3..d91bb30814 100644 --- a/GrpcInterface/Python/rips/taskmaestro_helper/run.py +++ b/GrpcInterface/Python/rips/taskmaestro_helper/run.py @@ -26,8 +26,14 @@ from .refs import REF_MARKER PROGRESS_PREFIX = "@@RI_WORKFLOW_EVENT@@" -def _emit_progress(run_id: str, task: str, state: str, error: str = "") -> None: +def _emit_progress( + run_id: str, task: str, state: str, error: str = "", item: str | None = None +) -> None: payload = {"event": "task_state", "run_id": run_id, "task": task, "state": state} + if item is not None: + # One item of a mapped task (`map:`); the task itself reports its own states + payload["event"] = "map_item_state" + payload["item"] = item if error: payload["error"] = error sys.stdout.write(PROGRESS_PREFIX + json.dumps(payload) + "\n") @@ -49,6 +55,15 @@ class ProgressHook: def on_task_fail(self, job: Any, task: Any, error: Exception) -> None: _emit_progress(self.run_id, task.name, "failed", str(error)) + def on_map_item_start(self, job: Any, task: Any, key: str) -> None: + _emit_progress(self.run_id, task.name, "running", item=key) + + def on_map_item_complete(self, job: Any, task: Any, key: str, output: Any) -> None: + _emit_progress(self.run_id, task.name, "completed", item=key) + + def on_map_item_fail(self, job: Any, task: Any, key: str, error: Exception) -> None: + _emit_progress(self.run_id, task.name, "failed", str(error), item=key) + def _emit(event: str, **fields: Any) -> None: sys.stdout.write(json.dumps({"event": event, **fields}) + "\n") diff --git a/GrpcInterface/Python/rips/tests/test_data/taskmaestro_workflows/tm_helper_pipeline.py b/GrpcInterface/Python/rips/tests/test_data/taskmaestro_workflows/tm_helper_pipeline.py index 1193e5ea57..c3c0ccbf6e 100644 --- a/GrpcInterface/Python/rips/tests/test_data/taskmaestro_workflows/tm_helper_pipeline.py +++ b/GrpcInterface/Python/rips/tests/test_data/taskmaestro_workflows/tm_helper_pipeline.py @@ -37,6 +37,12 @@ class Values(BaseModel): values: list[float] +class KeyedValues(BaseModel): + """Keyed numbers to sum.""" + + values: dict[str, float] + + class Item(BaseModel): """One mapped item.""" @@ -102,6 +108,15 @@ class Sum(Task[Values, Number]): return Number(value=sum(input.values)) +class KeyedSum(Task[KeyedValues, Number]): + """Sum keyed numbers.""" + + name = "keyed_sum" + + def run(self, input: KeyedValues, ctx: ExecutionContext) -> Number: + return Number(value=sum(input.values.values())) + + class PerItem(Task[Item, Number]): """Return the value of a mapped item.""" diff --git a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py index 81b2be4f11..80ad7eebac 100644 --- a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py +++ b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py @@ -135,6 +135,64 @@ def test_progress_hook_reports_runner_failure( assert events[-1]["error"] == "bad range" +def test_progress_hook_reports_mapped_items( + capsys: pytest.CaptureFixture[str], +) -> None: + from pydantic import BaseModel + from taskmaestro import EmptyConfig, ExecutionContext, Job, Runner, Task, Workflow + from taskmaestro.job import JobConfiguration + from taskmaestro.mapping import TaskMap + + from rips.taskmaestro_helper.run import PROGRESS_PREFIX, ProgressHook + + class Item(BaseModel): + key: str + value: int + + class Out(BaseModel): + value: int + + class Check(Task[Item, Out]): + name = "check" + + def run(self, input: Item, ctx: ExecutionContext) -> Out: + if input.value < 0: + raise ValueError("negative") + return Out(value=input.value) + + workflow = ( + Workflow.builder("mapped") + .add_task( + Check, + name="each", + mapped_over=TaskMap( + over="items", key_as="key", value_as="value", error_mode="collect_all" + ), + ) + .build() + ) + config = JobConfiguration({"each": {"items": {"a": 1, "b": -1}}}) + Runner(hooks=[ProgressHook("run-map")]).run( + Job(workflow, EmptyConfig(), job_configuration=config), ctx=ExecutionContext() + ) + events = [ + json.loads(line.removeprefix(PROGRESS_PREFIX)) + for line in capsys.readouterr().out.splitlines() + if line.startswith(PROGRESS_PREFIX) + ] + assert [ + (event["event"], event.get("item"), event["state"]) for event in events + ] == [ + ("task_state", None, "running"), + ("map_item_state", "a", "running"), + ("map_item_state", "a", "completed"), + ("map_item_state", "b", "running"), + ("map_item_state", "b", "failed"), + ("task_state", None, "failed"), + ] + assert events[4]["error"] == "negative" + + def test_run_helper_emits_progress_without_grpc( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: diff --git a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper_definition.py b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper_definition.py index 13c04ca2a1..3e02e37e7a 100644 --- a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper_definition.py +++ b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper_definition.py @@ -13,6 +13,7 @@ from pathlib import Path from typing import Any import pytest +import yaml pytest.importorskip("taskmaestro") @@ -421,34 +422,171 @@ def test_save_rejects_read_only_and_empty_workflows(tmp_path: Path) -> None: write_definition({**definition, "nodes": []}, tmp_path / "out") +# --- Mapped tasks and collected inputs --- + + +def test_round_trip_positional_collect(tmp_path: Path) -> None: + source = _write_workflow( + tmp_path / "src", + f"""\ + workflow: + name: collected + tasks: + - task: {PIPELINE}.Constant + name: a + - task: {PIPELINE}.Constant + name: b + - task: {PIPELINE}.Sum + name: total + depends_on: + values: + collect: [[b, value], [a, value]] + """, + ) + first, _ = _round_trip(source, tmp_path / "out") + assert [edge for edge in first["edges"] if edge["to"] == "total"] == [ + { + "from": "b", + "output": "value", + "to": "total", + "input": "values", + "collect": "list", + "key": None, + }, + { + "from": "a", + "output": "value", + "to": "total", + "input": "values", + "collect": "list", + "key": None, + }, + ] + raw = yaml.safe_load((tmp_path / "out" / "workflow.yaml").read_text()) + assert raw["workflow"]["tasks"][2]["depends_on"] == { + "values": {"collect": [["b", "value"], ["a", "value"]]} + } + + +def test_round_trip_keyed_collect(tmp_path: Path) -> None: + source = _write_workflow( + tmp_path / "src", + f"""\ + workflow: + name: keyed + tasks: + - task: {PIPELINE}.Constant + name: a + - task: {PIPELINE}.Constant + name: b + - task: {PIPELINE}.KeyedSum + name: total + depends_on: + values: + collect: {{first: [a, value], second: [b, value]}} + """, + ) + first, _ = _round_trip(source, tmp_path / "out") + assert [ + (edge["from"], edge["collect"], edge["key"]) + for edge in first["edges"] + if edge["to"] == "total" + ] == [("a", "dict", "first"), ("b", "dict", "second")] + raw = yaml.safe_load((tmp_path / "out" / "workflow.yaml").read_text()) + assert raw["workflow"]["tasks"][2]["depends_on"] == { + "values": {"collect": {"first": ["a", "value"], "second": ["b", "value"]}} + } + + +def test_save_rejects_inconsistent_collect(tmp_path: Path) -> None: + from rips.taskmaestro_helper._compat import HelperError + + source = _write_workflow( + tmp_path / "src", + f"""\ + workflow: + name: keyed + tasks: + - task: {PIPELINE}.Constant + name: a + - task: {PIPELINE}.Constant + name: b + - task: {PIPELINE}.KeyedSum + name: total + depends_on: + values: + collect: {{first: [a, value], second: [b, value]}} + """, + ) + definition = load_definition(source) + definition["edges"][-1]["key"] = "first" + with pytest.raises(HelperError, match="Duplicate key"): + write_definition(definition, tmp_path / "out") + definition["edges"][-1]["collect"] = "list" + with pytest.raises(HelperError, match="mixes"): + write_definition(definition, tmp_path / "out") + + +def test_round_trip_mapped_task(tmp_path: Path) -> None: + source = _write_workflow( + tmp_path / "src", + f"""\ + workflow: + name: mapped + tasks: + - task: {PIPELINE}.PerItem + name: items + map: {{over: numbers, key_as: key, value_as: value, error_mode: collect_all}} + """, + "items:\n numbers:\n a: 1.0\n b: 2.0\n", + ) + first, _ = _round_trip(source, tmp_path / "out") + assert first["nodes"][0]["map"] == { + "over": "numbers", + "key_as": "key", + "value_as": "value", + "error_mode": "collect_all", + } + assert first["nodes"][0]["config_fields"] == [] + assert first["inputs"] == {"items": {"numbers": {"a": 1.0, "b": 2.0}}} + raw = yaml.safe_load((tmp_path / "out" / "workflow.yaml").read_text()) + assert raw["workflow"]["tasks"][0]["map"] == { + "over": "numbers", + "key_as": "key", + "value_as": "value", + "error_mode": "collect_all", + } + assert yaml.safe_load((tmp_path / "out" / "input.yaml").read_text()) == { + "items": {"numbers": {"a": 1.0, "b": 2.0}} + } + + +def test_save_rejects_invalid_map(tmp_path: Path) -> None: + from rips.taskmaestro_helper._compat import HelperError + + source = _write_workflow( + tmp_path / "src", + f"""\ + workflow: + name: mapped + tasks: + - task: {PIPELINE}.PerItem + name: items + map: {{over: numbers, key_as: key, value_as: value}} + """, + ) + definition = load_definition(source) + definition["nodes"][0]["map"]["value_as"] = "key" + with pytest.raises(HelperError, match="both the key"): + write_definition(definition, tmp_path / "out") + + # --- Read-only workflows --- @pytest.mark.parametrize( ("tasks", "reason"), [ - ( - f"""\ - - task: {PIPELINE}.Constant - name: a - - task: {PIPELINE}.Constant - name: b - - task: {PIPELINE}.Sum - name: total - depends_on: - values: - collect: [[a, value], [b, value]] - """, - "collect", - ), - ( - f"""\ - - task: {PIPELINE}.PerItem - name: items - map: {{over: items, key_as: key, value_as: value}} - """, - "map", - ), ( """\ - workflow: inner.yaml diff --git a/README-experimental-workflow-ui.md b/README-experimental-workflow-ui.md index e4097641c9..51385632d2 100644 --- a/README-experimental-workflow-ui.md +++ b/README-experimental-workflow-ui.md @@ -102,6 +102,14 @@ for Undo, Redo, Save, Save As, and Duplicate as Editable. config input, shown in the property view when the task is selected. Optional inputs can be turned on under **Optional Inputs** in the task's context menu. +- **Collect outputs:** an input that takes a list or a dict (ring around the + port) accepts several connections. Each connection is labelled with its + position or key. Right-click a connection to **Rename Key** or **Move + Earlier/Later**. +- **Map over items:** right-click a task and choose **Map Over** to run it + once per item of a mapping. Choose the inputs that receive the key and the + value of each item. The items are entered in the property view, either as + JSON or by selecting cases, well paths or views. **Remove Map** undoes it. - **Rename, Set as Result Task, Delete:** right-click a task or connection. F2 renames and Del deletes the selection. Ctrl+Z and Ctrl+Y undo and redo. @@ -129,8 +137,8 @@ Use **Rescan Workflows** on the **Workflows** node after changing workflow folders or installing packages. ResInsight asks before closing or rescanning when there are unsaved edits. -Workflows that use `collect`, `map`, or nested workflows open read-only, and -the status line explains why. They can still be run. +Workflows that use nested workflows open read-only, and the status line +explains why. They can still be run. ## Troubleshooting diff --git a/docs/workflow-ui-design.md b/docs/workflow-ui-design.md index ac2f5a46e2..efe98cf98d 100644 --- a/docs/workflow-ui-design.md +++ b/docs/workflow-ui-design.md @@ -115,8 +115,8 @@ when: - it is **installed** (registered by a package). Use *Duplicate as Editable* to get an editable copy. -- it uses features the editor does not support: `collect`, `map`, nested - `workflow:` entries, or task classes that cannot be imported. +- it uses features the editor does not support: nested `workflow:` entries, + or task classes that cannot be imported. Read-only workflows can still be run. @@ -178,6 +178,33 @@ In an editable workflow: Duplicate as Editable and Fit. There is no toolbar. - Ctrl+Z/Ctrl+Y undo and redo; Ctrl+S saves. +### Collect and map + +`collect` (taskmaestro 0.3) gathers several outputs into one input. In the +definition each member is a separate edge with `collect: list|dict` (and a +`key` for dicts); the order of the list edges is the list order. +`RimWorkflowPortCompatibility::resolveConnection` decides whether a drop is +a plain edge or a collect member: an input typed `list[T]` or +`dict[str, T]` collects outputs of type `T`, while an output of the whole +container type still connects directly. Collecting inputs are drawn with a +ring around the port; member edges are labelled `[i]` or with their key. +Dict keys default to the source task name and are made unique. Right-click +a member for Rename Key… (dict) or Move Earlier/Later (list). + +`map` runs a task once per item of a configured mapping. Right-click a task +for Map Over… (Edit Map… when mapped) and Remove Map. The dialog picks the +mapping's config name (`over`), the inputs that receive the key (`key_as`, +must accept a string) and the value (`value_as`), and the error mode +(`fail_fast` or `collect_all`); `checkTaskMap` validates it while typing. +`taskTypeForNode` gives the effective task type of a mapped node: `key_as` +and `value_as` are no longer inputs, and the output becomes +`dict[str, Output]`. `graphTask` adds `over` as a config field and port +(`map_over: true`) typed `dict[str, ]` (`mapOverSchema`). +The node is drawn as a stack with a "⟳ map over X" subtitle. The `over` value is edited with `RimWorkflowMappingBinding`: a JSON +object, or a multi-selection of cases, well paths or views keyed by name. +During a run the helper emits `map_item_state` events; the node shows +"done/seen" counts and failures, and the tooltip lists each item. + Selecting a task shows its `RimWorkflowTaskInput` from the displayed job in the property view. Editing is locked while a job of the workflow is running. During a run, task states are drawn on the nodes. @@ -427,7 +454,7 @@ Touched (existing files): - **New edit operation**: add a pure function to `RimWorkflowDefinitionTools` with a unit test, call it through `RimWorkflow::applyEdit`, and expose it with a `Ric*WorkflowFeature` that reads its arguments from `userData()`. -- **Supporting `collect`/`map`/nested workflows in the editor**: extend the +- **Supporting nested workflows in the editor**: extend the definition JSON and `RimWorkflowDefinitionNode`, drop the read-only reason in `definition.py` and `structuralReadOnlyReasons`, and teach `save` to write them.