From 65942d455d41f33f21cf5ca7a47c3716db0c6a72 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 30 Sep 2026 14:03:53 +0200 Subject: [PATCH] Workflows: Read workflow schemas with the taskmaestro CLI Replace the rips.taskmaestro_helper introspect subcommand with `python -m taskmaestro workflow describe --json`. RimWorkflowDescribeTools converts the describe output (JSON schemas, depends_on, config_values) into the workflow graph used by the Workflow UI. This also adds support for collect dependencies and maps rips.Reservoir to EclipseCase pickers. --- ApplicationLibCode/CMakeLists.txt | 1 + .../Workflow/CMakeLists_files.cmake | 2 + .../ProjectDataModel/Workflow/RimWorkflow.cpp | 49 +- .../Workflow/RimWorkflowDescribeTools.cpp | 280 +++++++++++ .../Workflow/RimWorkflowDescribeTools.h | 33 ++ ApplicationLibCode/UnitTests/CMakeLists.txt | 1 + .../RimWorkflowDescribeTools-Test.cpp | 201 ++++++++ .../RimWorkflowDescribeTools/describe.json | 453 ++++++++++++++++++ .../UserInterface/RiuWorkflowGraphView.cpp | 2 +- .../rips/taskmaestro_helper/__init__.py | 6 +- .../rips/taskmaestro_helper/__main__.py | 7 +- .../rips/taskmaestro_helper/introspect.py | 231 --------- .../Python/rips/taskmaestro_helper/refs.py | 54 +-- .../rips/tests/test_taskmaestro_helper.py | 225 +-------- README-experimental-workflow-ui.md | 12 + 15 files changed, 1027 insertions(+), 530 deletions(-) create mode 100644 ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.cpp create mode 100644 ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.h create mode 100644 ApplicationLibCode/UnitTests/RimWorkflowDescribeTools-Test.cpp create mode 100644 ApplicationLibCode/UnitTests/TestData/RimWorkflowDescribeTools/describe.json delete mode 100644 GrpcInterface/Python/rips/taskmaestro_helper/introspect.py diff --git a/ApplicationLibCode/CMakeLists.txt b/ApplicationLibCode/CMakeLists.txt index eea1db653f..27fd0d438f 100644 --- a/ApplicationLibCode/CMakeLists.txt +++ b/ApplicationLibCode/CMakeLists.txt @@ -448,6 +448,7 @@ target_include_directories( ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModel/WellMeasurement ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModel/WellPath ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModel/WellEvents + ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModel/Workflow ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModelCommands ${CMAKE_CURRENT_SOURCE_DIR}/ProjectDataModelCommands/CommandRouter ${CMAKE_CURRENT_SOURCE_DIR}/ReservoirDataModel diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake b/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake index 171dea07e8..4da73ffda6 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake +++ b/ApplicationLibCode/ProjectDataModel/Workflow/CMakeLists_files.cmake @@ -1,6 +1,7 @@ set(SOURCE_GROUP_HEADER_FILES ${CMAKE_CURRENT_LIST_DIR}/RimWorkflow.h ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowCollection.h + ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowDescribeTools.h ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowJob.h ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowTaskInput.h ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowFieldBinding.h @@ -19,6 +20,7 @@ set(SOURCE_GROUP_HEADER_FILES set(SOURCE_GROUP_SOURCE_FILES ${CMAKE_CURRENT_LIST_DIR}/RimWorkflow.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowCollection.cpp + ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowDescribeTools.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowJob.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowTaskInput.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowFieldBinding.cpp diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflow.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflow.cpp index efb0b0ae63..beae967fc1 100644 --- a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflow.cpp +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflow.cpp @@ -18,11 +18,13 @@ #include "RimWorkflow.h" +#include "RimWorkflowDescribeTools.h" #include "RimWorkflowJob.h" #include "RimWorkflowTaskInput.h" #include "RiaLogging.h" #include "RiaPreferences.h" +#include "RiaQStringFormatter.h" #include "cafCmdFeatureMenuBuilder.h" @@ -123,7 +125,17 @@ bool RimWorkflow::loadFromDirectory( QString* errorMessage ) return false; } - QStringList args{ "-m", "rips.taskmaestro_helper", "introspect", dir }; + const QDir workflowDir( dir ); + if ( !workflowDir.exists( "workflow.yaml" ) ) + { + m_loadError = "Missing workflow.yaml"; + RiaLogging::warning( std::format( "Workflow '{}': {}", dir, m_loadError() ) ); + if ( errorMessage ) *errorMessage = m_loadError; + return false; + } + + QStringList args{ "-m", "taskmaestro", "workflow", "describe", workflowDir.absoluteFilePath( "workflow.yaml" ), "--json" }; + if ( workflowDir.exists( "input.yaml" ) ) args << "--input" << workflowDir.absoluteFilePath( "input.yaml" ); QProcess proc; QProcessEnvironment env = QProcessEnvironment::systemEnvironment(); @@ -139,39 +151,42 @@ bool RimWorkflow::loadFromDirectory( QString* errorMessage ) if ( !proc.waitForFinished( 30000 ) ) { proc.kill(); - m_loadError = "Introspect helper timed out"; + m_loadError = "taskmaestro workflow describe timed out"; RiaLogging::warning( QString( "Workflow '%1': %2" ).arg( dir, m_loadError() ).toStdString() ); if ( errorMessage ) *errorMessage = m_loadError; return false; } - const QByteArray stdoutBytes = proc.readAllStandardOutput(); - if ( proc.exitStatus() != QProcess::NormalExit || proc.exitCode() != 0 ) - { - m_loadError = QString::fromUtf8( proc.readAllStandardError() ).trimmed(); - if ( m_loadError().isEmpty() ) m_loadError = "Introspect helper failed"; - RiaLogging::warning( QString( "Workflow '%1' introspect failed: %2" ).arg( dir, m_loadError() ).toStdString() ); - if ( errorMessage ) *errorMessage = m_loadError; - return false; - } - + const QByteArray stdoutBytes = proc.readAllStandardOutput(); QJsonParseError parseErr{}; const QJsonDocument doc = QJsonDocument::fromJson( stdoutBytes, &parseErr ); - if ( parseErr.error != QJsonParseError::NoError || !doc.isObject() ) + + if ( proc.exitStatus() != QProcess::NormalExit || proc.exitCode() != 0 ) { - m_loadError = QString( "Invalid JSON from introspect: %1" ).arg( parseErr.errorString() ); + // In JSON mode, configuration errors are reported on stdout; other errors on stderr + m_loadError = RimWorkflowDescribeTools::errorFromDescribe( doc.object() ); + if ( m_loadError().isEmpty() ) m_loadError = QString::fromUtf8( proc.readAllStandardError() ).trimmed(); + if ( m_loadError().isEmpty() ) m_loadError = "taskmaestro workflow describe failed"; + RiaLogging::warning( std::format( "Workflow '{}' describe failed: {}", dir, m_loadError() ) ); if ( errorMessage ) *errorMessage = m_loadError; return false; } - const QJsonObject root = doc.object(); - if ( !root.value( "tasks" ).isArray() || !root.value( "edges" ).isArray() ) + if ( parseErr.error != QJsonParseError::NoError || !doc.isObject() ) + { + m_loadError = QString( "Invalid JSON from taskmaestro workflow describe: %1" ).arg( parseErr.errorString() ); + if ( errorMessage ) *errorMessage = m_loadError; + return false; + } + + if ( !doc.object().value( "tasks" ).isArray() ) { m_loadError = "Invalid workflow graph schema"; if ( errorMessage ) *errorMessage = m_loadError; return false; } - m_graph = root; + const QJsonObject root = RimWorkflowDescribeTools::graphFromDescribe( doc.object() ); + m_graph = root; if ( root.contains( "name" ) ) m_name = root.value( "name" ).toString(); if ( root.contains( "description" ) ) m_description = root.value( "description" ).toString(); setUiName( m_name() ); diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.cpp b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.cpp new file mode 100644 index 0000000000..3ee4fba194 --- /dev/null +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.cpp @@ -0,0 +1,280 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// 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 "RimWorkflowDescribeTools.h" + +#include +#include +#include +#include + +namespace +{ +QJsonObject resolveRef( const QString& ref, const QJsonObject& rootSchema ) +{ + const QString prefix = "#/$defs/"; + if ( !ref.startsWith( prefix ) ) return {}; + return rootSchema.value( "$defs" ).toObject().value( ref.mid( prefix.size() ) ).toObject(); +} + +//-------------------------------------------------------------------------------------------------- +/// Follow $ref and unwrap single-item allOf and Optional (anyOf with null) to the underlying type +//-------------------------------------------------------------------------------------------------- +QJsonObject underlyingSchema( QJsonObject schema, const QJsonObject& rootSchema ) +{ + constexpr int maxDepth = 16; + for ( int depth = 0; depth < maxDepth; ++depth ) + { + if ( schema.contains( "$ref" ) ) + { + schema = resolveRef( schema.value( "$ref" ).toString(), rootSchema ); + continue; + } + + const QJsonArray allOf = schema.value( "allOf" ).toArray(); + if ( allOf.size() == 1 ) + { + schema = allOf.first().toObject(); + continue; + } + + if ( schema.value( "anyOf" ).isArray() ) + { + QJsonArray nonNull; + for ( const QJsonValue& arm : schema.value( "anyOf" ).toArray() ) + { + if ( arm.toObject().value( "type" ).toString() != "null" ) nonNull.append( arm ); + } + if ( nonNull.size() == 1 ) + { + schema = nonNull.first().toObject(); + continue; + } + } + break; + } + return schema; +} + +bool isOpaque( const QJsonObject& schema, const QJsonObject& rootSchema ) +{ + return underlyingSchema( schema, rootSchema ).value( "x-taskmaestro-opaque" ).toBool( false ); +} + +//-------------------------------------------------------------------------------------------------- +/// Python type of a runtime-only object, either bare or wrapped in an ObjectModel `value` field +//-------------------------------------------------------------------------------------------------- +QString objectPythonType( const QJsonObject& typeSchema, const QJsonObject& rootSchema ) +{ + if ( typeSchema.contains( "x-taskmaestro-python-type" ) ) return typeSchema.value( "x-taskmaestro-python-type" ).toString(); + + const QJsonObject valueSchema = typeSchema.value( "properties" ).toObject().value( "value" ).toObject(); + if ( valueSchema.isEmpty() ) return {}; + return underlyingSchema( valueSchema, rootSchema ).value( "x-taskmaestro-python-type" ).toString(); +} + +QString jsonType( const QJsonObject& typeSchema ) +{ + const QString type = typeSchema.value( "type" ).toString(); + if ( type == "boolean" || type == "integer" || type == "number" || type == "array" ) return type; + return "string"; +} + +QJsonObject configFieldSchema( const QString& fieldName, + const QJsonObject& inputSchema, + const QStringList& requiredFields, + const QJsonObject& configValues ) +{ + QJsonObject entry{ { "name", fieldName } }; + + const QJsonObject properties = inputSchema.value( "properties" ).toObject(); + if ( !properties.contains( fieldName ) ) + { + entry["type"] = "string"; + entry["required"] = true; + entry["error"] = "field not found in input model"; + return entry; + } + + const QJsonObject property = properties.value( fieldName ).toObject(); + const QJsonObject typeSchema = underlyingSchema( property, inputSchema ); + const QString resinsightType = RimWorkflowDescribeTools::resinsightTypeFromPythonType( objectPythonType( typeSchema, inputSchema ) ); + + entry["type"] = resinsightType.isEmpty() ? jsonType( typeSchema ) : "object"; + entry["required"] = requiredFields.contains( fieldName ); + + const QString description = property.value( "description" ).toString(); + if ( !description.isEmpty() ) entry["description"] = description; + + // Prefer the value already in input.yaml; fall back to the model default + if ( configValues.contains( fieldName ) ) + entry["default"] = configValues.value( fieldName ); + else if ( property.contains( "default" ) ) + entry["default"] = property.value( "default" ); + + const QString format = typeSchema.value( "format" ).toString(); + if ( !format.isEmpty() ) entry["format"] = format; + + if ( !resinsightType.isEmpty() ) entry["resinsight_type"] = resinsightType; + + return entry; +} + +void appendEdge( QJsonArray& edges, const QJsonObject& ref, const QString& to, const QString& input ) +{ + const QString from = ref.value( "task" ).toString(); + if ( from.isEmpty() ) return; + + QJsonObject edge{ { "from", from }, { "to", to } }; + if ( !input.isEmpty() ) edge["input"] = input; + const QString output = ref.value( "field" ).toString(); + if ( !output.isEmpty() ) edge["output"] = output; + edges.append( edge ); +} + +void appendDependencyEdges( QJsonArray& edges, const QString& taskName, const QJsonValue& dependsOn ) +{ + const QJsonObject deps = dependsOn.toObject(); + if ( deps.isEmpty() ) return; + + // Single upstream task: {"task": ..., "field": ...} + if ( deps.value( "task" ).isString() ) + { + appendEdge( edges, deps, taskName, {} ); + return; + } + + for ( auto it = deps.begin(); it != deps.end(); ++it ) + { + const QJsonObject ref = it.value().toObject(); + if ( !ref.contains( "collect" ) ) + { + appendEdge( edges, ref, taskName, it.key() ); + continue; + } + + const QJsonValue members = ref.value( "collect" ).toObject().value( "members" ); + if ( members.isArray() ) + { + for ( const QJsonValue& member : members.toArray() ) + appendEdge( edges, member.toObject(), taskName, it.key() ); + } + else + { + for ( const QJsonValue& member : members.toObject() ) + appendEdge( edges, member.toObject(), taskName, it.key() ); + } + } +} + +QJsonObject taskFromDescribe( const QJsonObject& task ) +{ + const QJsonObject inputSchema = task.value( "input_schema" ).toObject(); + const QJsonObject outputSchema = task.value( "output_schema" ).toObject(); + + QJsonArray inputs; + for ( const QString& key : inputSchema.value( "properties" ).toObject().keys() ) + inputs.append( key ); + + // ObjectModel wraps its object in an opaque `value` field, which is not a routable output + QJsonArray outputs; + const QJsonObject outputProperties = outputSchema.value( "properties" ).toObject(); + for ( auto it = outputProperties.begin(); it != outputProperties.end(); ++it ) + { + if ( it.key() == "value" && isOpaque( it.value().toObject(), outputSchema ) ) continue; + outputs.append( it.key() ); + } + + QStringList requiredFields; + for ( const QJsonValue& value : inputSchema.value( "required" ).toArray() ) + requiredFields.append( value.toString() ); + + const QJsonObject configValues = task.value( "config_values" ).toObject(); + + QJsonArray configFields; + for ( const QJsonValue& value : task.value( "config_fields" ).toArray() ) + { + const QString fieldName = value.toString(); + if ( fieldName.isEmpty() ) continue; + configFields.append( configFieldSchema( fieldName, inputSchema, requiredFields, configValues ) ); + } + + return QJsonObject{ { "name", task.value( "name" ).toString() }, + { "inputs", inputs }, + { "outputs", outputs }, + { "config_fields", configFields } }; +} +} // namespace + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +QJsonObject RimWorkflowDescribeTools::graphFromDescribe( const QJsonObject& describe ) +{ + QJsonArray tasks; + QJsonArray edges; + for ( const QJsonValue& value : describe.value( "tasks" ).toArray() ) + { + const QJsonObject task = value.toObject(); + tasks.append( taskFromDescribe( task ) ); + appendDependencyEdges( edges, task.value( "name" ).toString(), task.value( "depends_on" ) ); + } + + return QJsonObject{ { "name", describe.value( "workflow" ).toString() }, { "description", "" }, { "tasks", tasks }, { "edges", edges } }; +} + +//-------------------------------------------------------------------------------------------------- +/// Error message from a `{"status": "invalid", "error": {...}}` document, empty if there is none +//-------------------------------------------------------------------------------------------------- +QString RimWorkflowDescribeTools::errorFromDescribe( const QJsonObject& describe ) +{ + if ( describe.value( "status" ).toString() != "invalid" ) return {}; + + const QJsonObject error = describe.value( "error" ).toObject(); + QString message = error.value( "message" ).toString( "Workflow configuration is invalid" ); + + QStringList details; + const QString task = error.value( "task" ).toString(); + if ( !task.isEmpty() ) details.append( QString( "task '%1'" ).arg( task ) ); + const QString field = error.value( "field" ).toString(); + if ( !field.isEmpty() ) details.append( QString( "field '%1'" ).arg( field ) ); + if ( !details.isEmpty() ) message += " (" + details.join( ", " ) + ")"; + + return message; +} + +//-------------------------------------------------------------------------------------------------- +/// Map a Python class path such as `rips.generated.generated_classes.WellPath` to the ResInsight +/// object type used to select the field binding. Returns an empty string for unsupported types. +//-------------------------------------------------------------------------------------------------- +QString RimWorkflowDescribeTools::resinsightTypeFromPythonType( const QString& pythonType ) +{ + const int separator = pythonType.lastIndexOf( '.' ); + if ( separator < 0 ) return {}; + + const QString module = pythonType.left( separator ); + if ( module != "rips" && !module.startsWith( "rips." ) ) return {}; + + static const QMap typeMap = { { "Case", "EclipseCase" }, + { "Reservoir", "EclipseCase" }, + { "EclipseCase", "EclipseCase" }, + { "WellPath", "WellPath" }, + { "View", "View" }, + { "EclipseView", "View" } }; + return typeMap.value( pythonType.mid( separator + 1 ) ); +} diff --git a/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.h b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.h new file mode 100644 index 0000000000..8fd407de14 --- /dev/null +++ b/ApplicationLibCode/ProjectDataModel/Workflow/RimWorkflowDescribeTools.h @@ -0,0 +1,33 @@ +///////////////////////////////////////////////////////////////////////////////// +// +// 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 +#include + +//================================================================================================== +/// Converts the output of `taskmaestro workflow describe --json` into the workflow graph used by +/// the Workflow UI: { name, description, tasks: [{ name, inputs, outputs, config_fields }], edges } +//================================================================================================== +namespace RimWorkflowDescribeTools +{ +QJsonObject graphFromDescribe( const QJsonObject& describe ); +QString errorFromDescribe( const QJsonObject& describe ); +QString resinsightTypeFromPythonType( const QString& pythonType ); +} // namespace RimWorkflowDescribeTools diff --git a/ApplicationLibCode/UnitTests/CMakeLists.txt b/ApplicationLibCode/UnitTests/CMakeLists.txt index 2b291622ed..1ebc3bd0b9 100644 --- a/ApplicationLibCode/UnitTests/CMakeLists.txt +++ b/ApplicationLibCode/UnitTests/CMakeLists.txt @@ -96,6 +96,7 @@ set(SOURCE_UNITTEST_FILES ${CMAKE_CURRENT_LIST_DIR}/RimWellPathGeometryDef-Test.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWellRftPlot-Test.cpp ${CMAKE_CURRENT_LIST_DIR}/RimWellTargetMapping-Test.cpp + ${CMAKE_CURRENT_LIST_DIR}/RimWorkflowDescribeTools-Test.cpp ${CMAKE_CURRENT_LIST_DIR}/RimDataFilterCollection-Test.cpp ${CMAKE_CURRENT_LIST_DIR}/RimIntersectionCollection-Test.cpp ${CMAKE_CURRENT_LIST_DIR}/RimDockWindowController-Test.cpp diff --git a/ApplicationLibCode/UnitTests/RimWorkflowDescribeTools-Test.cpp b/ApplicationLibCode/UnitTests/RimWorkflowDescribeTools-Test.cpp new file mode 100644 index 0000000000..d9c4ee94e5 --- /dev/null +++ b/ApplicationLibCode/UnitTests/RimWorkflowDescribeTools-Test.cpp @@ -0,0 +1,201 @@ +#include "gtest/gtest.h" + +#include "RiaTestDataDirectory.h" +#include "RimWorkflowDescribeTools.h" + +#include +#include +#include +#include + +namespace +{ +// Output of `taskmaestro workflow describe workflow.yaml --input input.yaml --json` for a +// synthetic workflow: start -> greet_a, greet_b -> merge (collect), merge <- start.count +QJsonObject describeFixture() +{ + QFile file( QString( "%1/RimWorkflowDescribeTools/describe.json" ).arg( TEST_DATA_DIR ) ); + if ( !file.open( QIODevice::ReadOnly ) ) return {}; + return QJsonDocument::fromJson( file.readAll() ).object(); +} + +QJsonObject taskByName( const QJsonObject& graph, const QString& name ) +{ + for ( const QJsonValue& task : graph.value( "tasks" ).toArray() ) + { + if ( task.toObject().value( "name" ).toString() == name ) return task.toObject(); + } + return {}; +} + +QJsonObject fieldByName( const QJsonObject& task, const QString& name ) +{ + for ( const QJsonValue& field : task.value( "config_fields" ).toArray() ) + { + if ( field.toObject().value( "name" ).toString() == name ) return field.toObject(); + } + return {}; +} +} // namespace + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, tasksInputsAndOutputs ) +{ + const QJsonObject describe = describeFixture(); + ASSERT_FALSE( describe.isEmpty() ); + + const QJsonObject graph = RimWorkflowDescribeTools::graphFromDescribe( describe ); + EXPECT_EQ( "synthetic", graph.value( "name" ).toString() ); + + QStringList names; + for ( const QJsonValue& task : graph.value( "tasks" ).toArray() ) + names.append( task.toObject().value( "name" ).toString() ); + EXPECT_EQ( QStringList( { "start", "greet_a", "greet_b", "merge" } ), names ); + + // The opaque ObjectModel `value` is not an output port + EXPECT_EQ( QJsonArray( { "count" } ), taskByName( graph, "start" ).value( "outputs" ).toArray() ); + EXPECT_EQ( QJsonArray( { "message" } ), taskByName( graph, "greet_a" ).value( "outputs" ).toArray() ); + EXPECT_EQ( QJsonArray( { "count", "messages" } ), taskByName( graph, "merge" ).value( "inputs" ).toArray() ); + EXPECT_TRUE( taskByName( graph, "start" ).value( "config_fields" ).toArray().isEmpty() ); + EXPECT_EQ( 9, taskByName( graph, "greet_a" ).value( "config_fields" ).toArray().size() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, configFieldTypesAndValues ) +{ + const QJsonObject graph = RimWorkflowDescribeTools::graphFromDescribe( describeFixture() ); + const QJsonObject greetA = taskByName( graph, "greet_a" ); + const QJsonObject greetB = taskByName( graph, "greet_b" ); + + // input.yaml value wins over the model default + const QJsonObject name = fieldByName( greetA, "name" ); + EXPECT_EQ( "string", name.value( "type" ).toString() ); + EXPECT_EQ( "alice", name.value( "default" ).toString() ); + EXPECT_EQ( "Person to greet", name.value( "description" ).toString() ); + EXPECT_FALSE( name.value( "required" ).toBool() ); + EXPECT_EQ( "world", fieldByName( greetB, "name" ).value( "default" ).toString() ); + + const QJsonObject times = fieldByName( greetA, "times" ); + EXPECT_EQ( "integer", times.value( "type" ).toString() ); + EXPECT_TRUE( times.value( "required" ).toBool() ); + EXPECT_EQ( 3, times.value( "default" ).toInt() ); + EXPECT_FALSE( fieldByName( greetB, "times" ).contains( "default" ) ); + + // Optional[int] is peeled; an explicit null in input.yaml is preserved + const QJsonObject optionalCount = fieldByName( greetA, "optional_count" ); + EXPECT_EQ( "integer", optionalCount.value( "type" ).toString() ); + EXPECT_TRUE( optionalCount.contains( "default" ) ); + EXPECT_TRUE( optionalCount.value( "default" ).isNull() ); + + EXPECT_EQ( "array", fieldByName( greetA, "values" ).value( "type" ).toString() ); + + const QJsonObject when = fieldByName( greetA, "when" ); + EXPECT_EQ( "date", when.value( "format" ).toString() ); + EXPECT_EQ( "2024-01-01", when.value( "default" ).toString() ); + + EXPECT_EQ( "path", fieldByName( greetA, "out_file" ).value( "format" ).toString() ); + EXPECT_EQ( "/tmp/hi.txt", fieldByName( greetA, "out_file" ).value( "default" ).toString() ); + EXPECT_EQ( "directory-path", fieldByName( greetA, "out_dir" ).value( "format" ).toString() ); + EXPECT_EQ( "/tmp", fieldByName( greetA, "out_dir" ).value( "default" ).toString() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, resinsightObjectFields ) +{ + const QJsonObject graph = RimWorkflowDescribeTools::graphFromDescribe( describeFixture() ); + const QJsonObject greetA = taskByName( graph, "greet_a" ); + + const QJsonObject wellPath = fieldByName( greetA, "well_path" ); + EXPECT_EQ( "object", wellPath.value( "type" ).toString() ); + EXPECT_EQ( "WellPath", wellPath.value( "resinsight_type" ).toString() ); + EXPECT_EQ( "Well path to use", wellPath.value( "description" ).toString() ); + EXPECT_EQ( "WellPath", wellPath.value( "default" ).toObject().value( "__resinsight_ref__" ).toString() ); + EXPECT_EQ( "B-2H", wellPath.value( "default" ).toObject().value( "well_path_name" ).toString() ); + + const QJsonObject gridCase = fieldByName( greetA, "case" ); + EXPECT_EQ( "EclipseCase", gridCase.value( "resinsight_type" ).toString() ); + EXPECT_TRUE( gridCase.value( "required" ).toBool() ); + + EXPECT_FALSE( fieldByName( greetA, "name" ).contains( "resinsight_type" ) ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, dependencyEdges ) +{ + const QJsonObject graph = RimWorkflowDescribeTools::graphFromDescribe( describeFixture() ); + + const QJsonArray expected = { + QJsonObject{ { "from", "start" }, { "to", "greet_a" } }, + QJsonObject{ { "from", "start" }, { "to", "greet_b" } }, + QJsonObject{ { "from", "start" }, { "to", "merge" }, { "input", "count" }, { "output", "count" } }, + QJsonObject{ { "from", "greet_a" }, { "to", "merge" }, { "input", "messages" }, { "output", "message" } }, + QJsonObject{ { "from", "greet_b" }, { "to", "merge" }, { "input", "messages" }, { "output", "message" } }, + }; + EXPECT_EQ( expected, graph.value( "edges" ).toArray() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, singleFieldAndKeyedCollectionEdges ) +{ + const QJsonObject keyedMembers{ { "a", QJsonObject{ { "task", "first" }, { "field", QJsonValue() } } }, + { "b", QJsonObject{ { "task", "second" }, { "field", "value" } } } }; + const QJsonObject describe{ + { "workflow", "routing" }, + { "tasks", + QJsonArray{ QJsonObject{ { "name", "next" }, { "depends_on", QJsonObject{ { "task", "producer" }, { "field", "value" } } } }, + QJsonObject{ { "name", "merge" }, + { "depends_on", + QJsonObject{ { "items", + QJsonObject{ { "collect", + QJsonObject{ { "kind", "keyed" }, { "members", keyedMembers } } } } } } } } } }, + }; + + const QJsonArray expected = { + QJsonObject{ { "from", "producer" }, { "to", "next" }, { "output", "value" } }, + QJsonObject{ { "from", "first" }, { "to", "merge" }, { "input", "items" } }, + QJsonObject{ { "from", "second" }, { "to", "merge" }, { "input", "items" }, { "output", "value" } }, + }; + EXPECT_EQ( expected, RimWorkflowDescribeTools::graphFromDescribe( describe ).value( "edges" ).toArray() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, missingConfigFieldIsReported ) +{ + const QJsonObject describe{ { "workflow", "broken" }, + { "tasks", + QJsonArray{ QJsonObject{ { "name", "task" }, + { "config_fields", QJsonArray{ "ghost" } }, + { "input_schema", QJsonObject{ { "properties", QJsonObject{} } } } } } } }; + + const QJsonObject field = fieldByName( taskByName( RimWorkflowDescribeTools::graphFromDescribe( describe ), "task" ), "ghost" ); + EXPECT_EQ( "string", field.value( "type" ).toString() ); + EXPECT_TRUE( field.value( "required" ).toBool() ); + EXPECT_FALSE( field.value( "error" ).toString().isEmpty() ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, errorFromDescribe ) +{ + EXPECT_TRUE( RimWorkflowDescribeTools::errorFromDescribe( describeFixture() ).isEmpty() ); + + const QJsonObject invalid{ { "status", "invalid" }, + { "error", + QJsonObject{ { "message", "Workflow configuration is invalid" }, { "task", "greet" }, { "field", "times" } } } }; + EXPECT_EQ( "Workflow configuration is invalid (task 'greet', field 'times')", RimWorkflowDescribeTools::errorFromDescribe( invalid ) ); + + const QJsonObject invalidWithoutDetails{ { "status", "invalid" }, + { "error", QJsonObject{ { "message", "Workflow configuration is invalid" } } } }; + EXPECT_EQ( "Workflow configuration is invalid", RimWorkflowDescribeTools::errorFromDescribe( invalidWithoutDetails ) ); +} + +//-------------------------------------------------------------------------------------------------- +TEST( RimWorkflowDescribeTools, resinsightTypeFromPythonType ) +{ + EXPECT_EQ( "EclipseCase", RimWorkflowDescribeTools::resinsightTypeFromPythonType( "rips.generated.generated_classes.Reservoir" ) ); + EXPECT_EQ( "EclipseCase", RimWorkflowDescribeTools::resinsightTypeFromPythonType( "rips.case.Case" ) ); + EXPECT_EQ( "WellPath", RimWorkflowDescribeTools::resinsightTypeFromPythonType( "rips.generated.generated_classes.WellPath" ) ); + EXPECT_EQ( "View", RimWorkflowDescribeTools::resinsightTypeFromPythonType( "rips.generated.generated_classes.EclipseView" ) ); + EXPECT_TRUE( RimWorkflowDescribeTools::resinsightTypeFromPythonType( "rips.instance.Instance" ).isEmpty() ); + EXPECT_TRUE( RimWorkflowDescribeTools::resinsightTypeFromPythonType( "mypackage.WellPath" ).isEmpty() ); + EXPECT_TRUE( RimWorkflowDescribeTools::resinsightTypeFromPythonType( "WellPath" ).isEmpty() ); +} diff --git a/ApplicationLibCode/UnitTests/TestData/RimWorkflowDescribeTools/describe.json b/ApplicationLibCode/UnitTests/TestData/RimWorkflowDescribeTools/describe.json new file mode 100644 index 0000000000..8f2bee9bbd --- /dev/null +++ b/ApplicationLibCode/UnitTests/TestData/RimWorkflowDescribeTools/describe.json @@ -0,0 +1,453 @@ +{ + "workflow": "synthetic", + "result_task": "merge", + "tasks": [ + { + "name": "start", + "python_type": "pipeline.Start", + "depends_on": null, + "config_fields": [], + "provided_config_fields": [], + "missing_config_fields": [], + "config_values": {}, + "required_input_fields": [], + "map": null, + "input_schema": { + "description": "Sentinel config for workflows where all root tasks use JobConfiguration.", + "properties": {}, + "title": "EmptyConfig", + "type": "object" + }, + "output_schema": { + "properties": { + "value": { + "not": {}, + "title": "Value", + "x-taskmaestro-opaque": true, + "x-taskmaestro-python-type": "rips.generated.generated_classes.WellPath" + }, + "count": { + "default": 0, + "title": "Count", + "type": "integer" + } + }, + "required": [ + "value" + ], + "title": "StartOutput", + "type": "object" + } + }, + { + "name": "greet_a", + "python_type": "pipeline.Greet", + "depends_on": { + "task": "start", + "field": null + }, + "config_fields": [ + "case", + "name", + "optional_count", + "out_dir", + "out_file", + "times", + "values", + "well_path", + "when" + ], + "provided_config_fields": [ + "name", + "optional_count", + "out_dir", + "times", + "well_path" + ], + "missing_config_fields": [ + "case", + "out_file", + "values", + "when" + ], + "config_values": { + "name": "alice", + "optional_count": null, + "out_dir": "/tmp", + "times": 3, + "well_path": { + "__resinsight_ref__": "WellPath", + "well_path_name": "B-2H" + } + }, + "required_input_fields": [ + "case", + "out_dir", + "times", + "well_path" + ], + "map": null, + "input_schema": { + "$defs": { + "GridCase": { + "properties": { + "value": { + "not": {}, + "title": "Value", + "x-taskmaestro-opaque": true, + "x-taskmaestro-python-type": "rips.generated.generated_classes.Reservoir" + } + }, + "required": [ + "value" + ], + "title": "GridCase", + "type": "object" + }, + "WellPath": { + "properties": { + "value": { + "not": {}, + "title": "Value", + "x-taskmaestro-opaque": true, + "x-taskmaestro-python-type": "rips.generated.generated_classes.WellPath" + } + }, + "required": [ + "value" + ], + "title": "WellPath", + "type": "object" + } + }, + "properties": { + "source": { + "anyOf": [ + { + "$ref": "#/$defs/WellPath" + }, + { + "type": "null" + } + ], + "default": null + }, + "well_path": { + "$ref": "#/$defs/WellPath", + "description": "Well path to use" + }, + "case": { + "$ref": "#/$defs/GridCase", + "description": "Case to use" + }, + "name": { + "default": "world", + "description": "Person to greet", + "title": "Name", + "type": "string" + }, + "times": { + "description": "How many times", + "minimum": 1, + "title": "Times", + "type": "integer" + }, + "optional_count": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Optional Count" + }, + "values": { + "items": { + "type": "integer" + }, + "title": "Values", + "type": "array" + }, + "when": { + "default": "2024-01-01", + "description": "Greet date", + "format": "date", + "title": "When", + "type": "string" + }, + "out_file": { + "default": "/tmp/hi.txt", + "description": "Where to write", + "format": "path", + "title": "Out File", + "type": "string" + }, + "out_dir": { + "description": "Where to write logs", + "format": "directory-path", + "title": "Out Dir", + "type": "string" + } + }, + "required": [ + "well_path", + "case", + "times", + "out_dir" + ], + "title": "GreetInput", + "type": "object" + }, + "output_schema": { + "properties": { + "message": { + "title": "Message", + "type": "string" + } + }, + "required": [ + "message" + ], + "title": "GreetOutput", + "type": "object" + } + }, + { + "name": "greet_b", + "python_type": "pipeline.Greet", + "depends_on": { + "task": "start", + "field": null + }, + "config_fields": [ + "case", + "name", + "optional_count", + "out_dir", + "out_file", + "times", + "values", + "well_path", + "when" + ], + "provided_config_fields": [], + "missing_config_fields": [ + "case", + "name", + "optional_count", + "out_dir", + "out_file", + "times", + "values", + "well_path", + "when" + ], + "config_values": {}, + "required_input_fields": [ + "case", + "out_dir", + "times", + "well_path" + ], + "map": null, + "input_schema": { + "$defs": { + "GridCase": { + "properties": { + "value": { + "not": {}, + "title": "Value", + "x-taskmaestro-opaque": true, + "x-taskmaestro-python-type": "rips.generated.generated_classes.Reservoir" + } + }, + "required": [ + "value" + ], + "title": "GridCase", + "type": "object" + }, + "WellPath": { + "properties": { + "value": { + "not": {}, + "title": "Value", + "x-taskmaestro-opaque": true, + "x-taskmaestro-python-type": "rips.generated.generated_classes.WellPath" + } + }, + "required": [ + "value" + ], + "title": "WellPath", + "type": "object" + } + }, + "properties": { + "source": { + "anyOf": [ + { + "$ref": "#/$defs/WellPath" + }, + { + "type": "null" + } + ], + "default": null + }, + "well_path": { + "$ref": "#/$defs/WellPath", + "description": "Well path to use" + }, + "case": { + "$ref": "#/$defs/GridCase", + "description": "Case to use" + }, + "name": { + "default": "world", + "description": "Person to greet", + "title": "Name", + "type": "string" + }, + "times": { + "description": "How many times", + "minimum": 1, + "title": "Times", + "type": "integer" + }, + "optional_count": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Optional Count" + }, + "values": { + "items": { + "type": "integer" + }, + "title": "Values", + "type": "array" + }, + "when": { + "default": "2024-01-01", + "description": "Greet date", + "format": "date", + "title": "When", + "type": "string" + }, + "out_file": { + "default": "/tmp/hi.txt", + "description": "Where to write", + "format": "path", + "title": "Out File", + "type": "string" + }, + "out_dir": { + "description": "Where to write logs", + "format": "directory-path", + "title": "Out Dir", + "type": "string" + } + }, + "required": [ + "well_path", + "case", + "times", + "out_dir" + ], + "title": "GreetInput", + "type": "object" + }, + "output_schema": { + "properties": { + "message": { + "title": "Message", + "type": "string" + } + }, + "required": [ + "message" + ], + "title": "GreetOutput", + "type": "object" + } + }, + { + "name": "merge", + "python_type": "pipeline.Merge", + "depends_on": { + "messages": { + "collect": { + "kind": "positional", + "members": [ + { + "task": "greet_a", + "field": "message" + }, + { + "task": "greet_b", + "field": "message" + } + ] + } + }, + "count": { + "task": "start", + "field": "count" + } + }, + "config_fields": [], + "provided_config_fields": [], + "missing_config_fields": [], + "config_values": {}, + "required_input_fields": [ + "count", + "messages" + ], + "map": null, + "input_schema": { + "properties": { + "messages": { + "items": { + "type": "string" + }, + "title": "Messages", + "type": "array" + }, + "count": { + "title": "Count", + "type": "integer" + } + }, + "required": [ + "messages", + "count" + ], + "title": "MergeInput", + "type": "object" + }, + "output_schema": { + "properties": { + "message": { + "title": "Message", + "type": "string" + } + }, + "required": [ + "message" + ], + "title": "GreetOutput", + "type": "object" + } + } + ] +} diff --git a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp index 06eece6348..83a60952b9 100644 --- a/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp +++ b/ApplicationLibCode/UserInterface/RiuWorkflowGraphView.cpp @@ -357,7 +357,7 @@ void RiuWorkflowGraphView::showGraph( const QJsonObject& graph, const QString& e inputs[name].append( configured ); } - // The Python helper provides tasks in topological order. + // taskmaestro describes tasks in topological order. for ( const QJsonValue& task : tasks ) { const QString name = task.toObject().value( "name" ).toString(); diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/__init__.py b/GrpcInterface/Python/rips/taskmaestro_helper/__init__.py index a818a29ea4..cc4a3947c6 100644 --- a/GrpcInterface/Python/rips/taskmaestro_helper/__init__.py +++ b/GrpcInterface/Python/rips/taskmaestro_helper/__init__.py @@ -1,6 +1,6 @@ """Bridge between ResInsight and the taskmaestro workflow library. -ResInsight invokes this package as a subprocess to introspect workflow -schemas and to execute workflows. The Python side exists so ResInsight -itself does not need to parse YAML or know about Pydantic. +ResInsight invokes this package as a subprocess to execute workflows against +a running ResInsight. Workflow schemas are read with the taskmaestro CLI +(`python -m taskmaestro workflow describe`). """ diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/__main__.py b/GrpcInterface/Python/rips/taskmaestro_helper/__main__.py index cbf574f257..5396a160bd 100644 --- a/GrpcInterface/Python/rips/taskmaestro_helper/__main__.py +++ b/GrpcInterface/Python/rips/taskmaestro_helper/__main__.py @@ -8,7 +8,7 @@ import sys def main() -> int: if len(sys.argv) < 2: print( - "usage: python -m rips.taskmaestro_helper ...", + "usage: python -m rips.taskmaestro_helper run ...", file=sys.stderr, ) return 2 @@ -16,11 +16,6 @@ def main() -> int: subcommand = sys.argv[1] rest = sys.argv[2:] - if subcommand == "introspect": - from . import introspect - - return introspect.main(rest) - if subcommand == "run": from . import run diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/introspect.py b/GrpcInterface/Python/rips/taskmaestro_helper/introspect.py deleted file mode 100644 index 4eb5ad9dc0..0000000000 --- a/GrpcInterface/Python/rips/taskmaestro_helper/introspect.py +++ /dev/null @@ -1,231 +0,0 @@ -"""Introspect a taskmaestro workflow directory and emit a JSON schema. - -The schema is consumed by ResInsight to auto-generate property-editor -fields for each task's config inputs. Only fields listed in the workflow's -`config_fields` are inspected, so models that also reference non-Pydantic -types (e.g. live `rips` objects passed via `depends_on`) do not break -introspection. -""" - -from __future__ import annotations - -import datetime -import enum -import json -import pathlib -import sys -from pathlib import Path -from types import UnionType -from typing import Annotated, Any, Union, get_args, get_origin - -from pydantic import TypeAdapter -from pydantic.fields import FieldInfo - -from .refs import annotation_to_resinsight_type - - -_SCALAR_TYPE_MAP: dict[type, str] = { - str: "string", - bool: "boolean", - int: "integer", - float: "number", -} - - -def _format_for(field_info: FieldInfo) -> str | None: - """Ask Pydantic for the JSON-schema `format` of a field, if any. - - Pydantic stores e.g. `pydantic.DirectoryPath` (which is - `Annotated[Path, PathType('dir')]`) as `annotation=Path` and - `metadata=[PathType('dir')]`. We reassemble both before asking - TypeAdapter, so format markers like `directory-path` survive. - """ - annotation = field_info.annotation - if field_info.metadata: - annotation = Annotated[tuple([annotation, *field_info.metadata])] - try: - return TypeAdapter(annotation).json_schema().get("format") - except Exception: - return None - - -def _annotation_type(annotation: object) -> str: - """Map a Pydantic field annotation to a JSON-schema-style type label.""" - if isinstance(annotation, type) and annotation in _SCALAR_TYPE_MAP: - return _SCALAR_TYPE_MAP[annotation] - - # Optional[T] / T | None — peel the union and map the non-None arm - origin = get_origin(annotation) - if origin in (Union, UnionType): - non_none = [a for a in get_args(annotation) if a is not type(None)] - if len(non_none) == 1: - return _annotation_type(non_none[0]) - - if origin in (list, tuple, set, frozenset): - return "array" - - if isinstance(annotation, type): - # Recognised rips object types are reported as "object" with a separate marker. - if annotation_to_resinsight_type(annotation) is not None: - return "object" - - return "string" - - -_SENTINEL = object() - - -def _serialize_default(value: Any) -> Any: - """Convert a Python value into a JSON-friendly form for the schema's `default`.""" - if isinstance(value, (datetime.date, datetime.datetime)): - return value.isoformat() - if isinstance(value, pathlib.PurePath): - return str(value) - if isinstance(value, enum.Enum): - return _serialize_default(value.value) - if isinstance(value, dict): - return {key: _serialize_default(item) for key, item in value.items()} - if isinstance(value, (list, tuple, set, frozenset)): - return [_serialize_default(item) for item in value] - return value - - -def _field_schema( - field_name: str, field_info: FieldInfo, input_value: Any = _SENTINEL -) -> dict[str, Any]: - entry: dict[str, Any] = { - "name": field_name, - "type": _annotation_type(field_info.annotation), - "required": field_info.is_required(), - } - if field_info.description: - entry["description"] = field_info.description - - # Prefer the value the user already has in input.yaml; fall back to the - # Pydantic-side default so the UI is pre-filled in either case. - if input_value is not _SENTINEL: - entry["default"] = _serialize_default(input_value) - elif not field_info.is_required(): - default = field_info.get_default(call_default_factory=True) - entry["default"] = _serialize_default(default) - - fmt = _format_for(field_info) - if fmt is not None: - entry["format"] = fmt - ri_type = annotation_to_resinsight_type(field_info.annotation) - if ri_type is not None: - entry["resinsight_type"] = ri_type - return entry - - -def _dependency_edges(task_name: str, deps: Any) -> list[dict[str, str]]: - """Flatten taskmaestro's resolved dependency forms into directed edges.""" - if deps is None: - return [] - if isinstance(deps, str): - return [{"from": deps, "to": task_name}] - if isinstance(deps, tuple): - return [{"from": deps[0], "to": task_name, "output": deps[1]}] - - edges = [] - for input_name, upstream in sorted(deps.items()): - edge = { - "from": upstream[0] if isinstance(upstream, tuple) else upstream, - "to": task_name, - "input": input_name, - } - if isinstance(upstream, tuple): - edge["output"] = upstream[1] - edges.append(edge) - return edges - - -def _output_fields(output_type: type) -> list[str]: - """Expose output fields, but not ObjectModel's internal wrapped value.""" - from taskmaestro import ObjectModel - - fields = set(output_type.model_fields) - if issubclass(output_type, ObjectModel): - fields.discard("value") - return sorted(fields) - - -def collect_schema(workflow_dir: Path) -> dict[str, Any]: - """Load the workflow at `workflow_dir` and produce its UI schema.""" - workflow_yaml = workflow_dir / "workflow.yaml" - input_yaml = workflow_dir / "input.yaml" - if not workflow_yaml.is_file(): - raise FileNotFoundError(f"Missing workflow.yaml in {workflow_dir}") - - # Workflow source files (e.g. pipeline.py) live next to workflow.yaml and - # are referenced as plain dotted paths in workflow.yaml — make them importable. - workflow_dir_str = str(workflow_dir) - if workflow_dir_str not in sys.path: - sys.path.insert(0, workflow_dir_str) - - from taskmaestro.task import get_input_type, get_output_type - from taskmaestro.yaml_config import _load_workflow_only - - wf, jc = _load_workflow_only( - workflow_yaml, - input_yaml if input_yaml.is_file() else None, - ) - - tasks: list[dict[str, Any]] = [] - edges: list[dict[str, str]] = [] - for task_name, task_cls in wf.topological_order(): - edges.extend(_dependency_edges(task_name, wf.get_dependencies(task_name))) - config_field_names = wf.get_config_fields(task_name) - task_config = jc.get_config_for_task(task_name) if jc is not None else {} - - input_type = get_input_type(task_cls) - config_fields: list[dict[str, Any]] = [] - for field_name in sorted(config_field_names): - field_info = input_type.model_fields.get(field_name) - if field_info is None: - config_fields.append( - { - "name": field_name, - "type": "string", - "required": True, - "error": "field not found in input model", - } - ) - continue - input_value = task_config.get(field_name, _SENTINEL) - config_fields.append( - _field_schema(field_name, field_info, input_value=input_value) - ) - - tasks.append( - { - "name": task_name, - "inputs": sorted(input_type.model_fields), - "outputs": _output_fields(get_output_type(task_cls)), - "config_fields": config_fields, - } - ) - - return { - "name": wf.name, - "description": "", - "tasks": tasks, - "edges": edges, - } - - -def main(argv: list[str]) -> int: - if len(argv) != 1: - print("usage: introspect ", file=sys.stderr) - return 2 - workflow_dir = Path(argv[0]).resolve() - try: - schema = collect_schema(workflow_dir) - except Exception as exc: - print( - json.dumps({"error": str(exc), "type": type(exc).__name__}), file=sys.stderr - ) - return 1 - json.dump(schema, sys.stdout, indent=2) - sys.stdout.write("\n") - return 0 diff --git a/GrpcInterface/Python/rips/taskmaestro_helper/refs.py b/GrpcInterface/Python/rips/taskmaestro_helper/refs.py index de5b1d3c09..afa2d8f030 100644 --- a/GrpcInterface/Python/rips/taskmaestro_helper/refs.py +++ b/GrpcInterface/Python/rips/taskmaestro_helper/refs.py @@ -1,56 +1,12 @@ -"""Mapping between Pydantic field annotations and ResInsight object types. +"""ResInsight object references in taskmaestro input files. -The whitelist below is the single source of truth for which `rips` types -the helper recognises as object pickers. It is referenced both by -`introspect` (to add the `resinsight_type` marker on field schemas) and -by `run` (to resolve `__resinsight_ref__` values back to live `rips` -objects). +The Workflow UI writes a `{__resinsight_ref__: , ...id...}` map into +input.yaml for fields that refer to a ResInsight object. `run` resolves these +maps back to live `rips` objects. The C++ side maps Python types to the +`` labels in RimWorkflowDescribeTools. """ from __future__ import annotations -# Maps rips class name -> resinsight_type label sent to the C++ side. -# The label is also the C++ binding subclass selector (RimWorkflow{Label}Binding). -# Rips classes live in `rips.generated.generated_classes` but are re-exported as -# `rips.`; we match by class name only since the module is generated. -RIPS_TYPE_WHITELIST: dict[str, str] = { - "Case": "EclipseCase", - "EclipseCase": "EclipseCase", - "WellPath": "WellPath", - "View": "View", - "EclipseView": "View", -} - -_RIPS_MODULE_PREFIXES = ("rips.", "rips") - # Marker used inside input.yaml to denote a ResInsight object reference. REF_MARKER = "__resinsight_ref__" - - -def annotation_to_resinsight_type(annotation: object) -> str | None: - """Return the resinsight_type label for a Pydantic field annotation, or None. - - Handles both bare rips classes and ObjectModel[rips.X] wrappers used by - the example workflow (e.g. `class GridCase(ObjectModel[rips.EclipseCase])`). - """ - candidates: list[type] = [] - if isinstance(annotation, type): - candidates.append(annotation) - # Walk the MRO for ObjectModel[T] parameterizations: pydantic stores the - # type argument on the intermediate class in __pydantic_generic_metadata__. - for cls in annotation.__mro__: - md = getattr(cls, "__pydantic_generic_metadata__", None) - if not md: - continue - for arg in md.get("args", ()) or (): - if isinstance(arg, type): - candidates.append(arg) - - for cls in candidates: - module = getattr(cls, "__module__", "") or "" - if not module.startswith(_RIPS_MODULE_PREFIXES): - continue - label = RIPS_TYPE_WHITELIST.get(cls.__name__) - if label is not None: - return label - return None diff --git a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py index bc6021f3b8..81b2be4f11 100644 --- a/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py +++ b/GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py @@ -1,15 +1,12 @@ """Unit tests for rips.taskmaestro_helper. -These tests do not require a running ResInsight; they exercise the -Pydantic introspection logic in isolation against a synthetic workflow -written into a tmp_path so they stay self-contained. +These tests do not require a running ResInsight; they exercise reference +resolution and progress reporting for the `run` subcommand in isolation. """ from __future__ import annotations import json -import subprocess -import sys from pathlib import Path import pytest @@ -17,203 +14,6 @@ import pytest pytest.importorskip("taskmaestro") -SYNTHETIC_PIPELINE = ''' -"""Synthetic workflow for taskmaestro_helper tests.""" - -from __future__ import annotations - -import datetime -import pathlib - -import pydantic -from pydantic import BaseModel, Field - -from taskmaestro import EmptyConfig, ExecutionContext, Task - - -class GreetInput(BaseModel): - name: str = Field(description="Person to greet", default="world") - times: int = Field(description="How many times", ge=1) - when: datetime.date = Field(description="Greet date", default=datetime.date(2024, 1, 1)) - out_file: pathlib.Path = Field(description="Where to write greeting", default=pathlib.Path("/tmp/hi.txt")) - out_dir: pydantic.DirectoryPath = Field(description="Where to write logs") - - -class GreetOutput(BaseModel): - message: str - - -class Greet(Task[GreetInput, GreetOutput]): - name = "greet" - - def run(self, input: GreetInput, ctx: ExecutionContext) -> GreetOutput: # pragma: no cover - return GreetOutput(message=f"hi {input.name}" * input.times) - - -class StartInput(BaseModel): - pass - - -class StartOutput(BaseModel): - pass - - -class Start(Task[StartInput, StartOutput]): - name = "start" - - def run(self, input: StartInput, ctx: ExecutionContext) -> StartOutput: # pragma: no cover - return StartOutput() -''' - - -SYNTHETIC_WORKFLOW_YAML = """workflow: - name: synthetic - tasks: - - task: pipeline.Start - - task: pipeline.Greet - depends_on: pipeline.Start - config_fields: [name, times, when, out_file, out_dir] -""" - - -SYNTHETIC_INPUT_YAML = """greet: - name: alice - times: 3 - out_dir: /tmp -""" - - -@pytest.fixture -def workflow_dir(tmp_path: Path) -> Path: - (tmp_path / "pipeline.py").write_text(SYNTHETIC_PIPELINE) - (tmp_path / "workflow.yaml").write_text(SYNTHETIC_WORKFLOW_YAML) - (tmp_path / "input.yaml").write_text(SYNTHETIC_INPUT_YAML) - return tmp_path - - -def test_collect_schema_returns_workflow_name_and_tasks(workflow_dir: Path) -> None: - from rips.taskmaestro_helper.introspect import collect_schema - - schema = collect_schema(workflow_dir) - assert schema["name"] == "synthetic" - task_names = [t["name"] for t in schema["tasks"]] - assert task_names == ["start", "greet"] - assert schema["tasks"][0]["config_fields"] == [] - assert schema["tasks"][0]["inputs"] == [] - assert schema["tasks"][0]["outputs"] == [] - assert schema["tasks"][1]["inputs"] == [ - "name", - "out_dir", - "out_file", - "times", - "when", - ] - assert schema["tasks"][1]["outputs"] == ["message"] - assert schema["edges"] == [{"from": "start", "to": "greet"}] - - -def test_collect_schema_extracts_field_types_and_metadata(workflow_dir: Path) -> None: - from rips.taskmaestro_helper.introspect import collect_schema - - schema = collect_schema(workflow_dir) - greet = schema["tasks"][1] - fields_by_name = {f["name"]: f for f in greet["config_fields"]} - - # input.yaml has `name: alice`, so that wins over the Pydantic default "world" - assert fields_by_name["name"]["type"] == "string" - assert fields_by_name["name"]["default"] == "alice" - assert fields_by_name["name"]["description"] == "Person to greet" - assert fields_by_name["name"]["required"] is False - - # input.yaml has `times: 3`; field has no Pydantic default but is required - assert fields_by_name["times"]["type"] == "integer" - assert fields_by_name["times"]["required"] is True - assert fields_by_name["times"]["default"] == 3 - assert "resinsight_type" not in fields_by_name["times"] - - # input.yaml has no `when:` block, so Pydantic default kicks in - assert fields_by_name["when"]["type"] == "string" - assert fields_by_name["when"]["format"] == "date" - assert fields_by_name["when"]["default"] == "2024-01-01" - - # input.yaml has no out_file; Pydantic default flows through - assert fields_by_name["out_file"]["type"] == "string" - assert fields_by_name["out_file"]["format"] == "path" - assert fields_by_name["out_file"]["default"] == "/tmp/hi.txt" - - # input.yaml has `out_dir: /tmp`; pre-fills even though field is required - assert fields_by_name["out_dir"]["type"] == "string" - assert fields_by_name["out_dir"]["format"] == "directory-path" - assert fields_by_name["out_dir"]["required"] is True - assert fields_by_name["out_dir"]["default"] == "/tmp" - - -def test_field_schema_preserves_arrays_and_null_defaults() -> None: - from pydantic import BaseModel, Field - - from rips.taskmaestro_helper.introspect import _field_schema - - class Config(BaseModel): - values: list[int] = Field(default_factory=lambda: [1, 2]) - optional_count: int | None = None - - values = _field_schema("values", Config.model_fields["values"]) - optional_count = _field_schema( - "optional_count", Config.model_fields["optional_count"] - ) - - assert values["type"] == "array" - assert values["default"] == [1, 2] - assert optional_count["type"] == "integer" - assert "default" in optional_count - assert optional_count["default"] is None - - -def test_field_schema_preserves_explicit_null_input() -> None: - from pydantic import BaseModel - - from rips.taskmaestro_helper.introspect import _field_schema - - class Config(BaseModel): - optional_name: str | None = "fallback" - - schema = _field_schema( - "optional_name", Config.model_fields["optional_name"], input_value=None - ) - assert "default" in schema - assert schema["default"] is None - - -def test_object_model_value_is_not_an_output_port() -> None: - from pydantic import BaseModel - from taskmaestro import ObjectModel - - from rips.taskmaestro_helper.introspect import _output_fields - - class WrappedOutput(ObjectModel[str]): - other: int - - class PlainOutput(BaseModel): - value: str - other: int - - assert _output_fields(WrappedOutput) == ["other"] - assert _output_fields(PlainOutput) == ["other", "value"] - - -def test_dependency_edges_cover_fan_in_and_field_routing() -> None: - from rips.taskmaestro_helper.introspect import _dependency_edges - - assert _dependency_edges("root", None) == [] - assert _dependency_edges("next", ("producer", "value")) == [ - {"from": "producer", "to": "next", "output": "value"} - ] - assert _dependency_edges("merge", {"b": ("second", "value"), "a": "first"}) == [ - {"from": "first", "to": "merge", "input": "a"}, - {"from": "second", "to": "merge", "input": "b", "output": "value"}, - ] - - def test_resolve_refs_substitutes_object_model_value() -> None: from rips.taskmaestro_helper.run import resolve_refs @@ -374,24 +174,3 @@ def test_run_helper_emits_progress_without_grpc( ("done", "completed"), ] assert all(event["run_id"] == "run-456" for event in events) - - -def test_main_emits_json_to_stdout(workflow_dir: Path) -> None: - rips_root = Path(__file__).resolve().parents[2] - proc = subprocess.run( - [ - sys.executable, - "-m", - "rips.taskmaestro_helper", - "introspect", - str(workflow_dir), - ], - cwd=rips_root, - capture_output=True, - text=True, - check=True, - ) - payload = json.loads(proc.stdout) - assert payload["name"] == "synthetic" - assert {t["name"] for t in payload["tasks"]} == {"start", "greet"} - assert payload["edges"] == [{"from": "start", "to": "greet"}] diff --git a/README-experimental-workflow-ui.md b/README-experimental-workflow-ui.md index 86e8dee8bb..ad6201fcfe 100644 --- a/README-experimental-workflow-ui.md +++ b/README-experimental-workflow-ui.md @@ -34,6 +34,15 @@ git clone https://github.com/OPM/taskmaestro-resinsight.git \ python -m pip install -e ~/resinsight-workflows/taskmaestro-resinsight ``` +ResInsight reads workflows with `python -m taskmaestro workflow describe`, which +needs a `taskmaestro` version that reports `config_values` +([OPM/taskmaestro#13](https://github.com/OPM/taskmaestro/pull/13)). Until that +is released, install it from Git: + +```bash +python -m pip install "git+https://github.com/OPM/taskmaestro.git@main" +``` + ## 2. Register the workflows ResInsight discovers workflows in `~/.taskmaestro/workflows`: @@ -70,5 +79,8 @@ tree. Select a workflow, provide its inputs, and click **Run**. `~/.taskmaestro/workflows` contains `workflow.yaml`, then restart ResInsight. - **A Python module cannot be found:** Confirm that **Python Executable Location** points to the virtual environment created above. +- **A workflow fails to load with `invalid choice: 'workflow'`, or its inputs + are not pre-filled from `input.yaml`:** The installed `taskmaestro` is too + old; reinstall it as described above. - **The workflow cannot connect:** Enable **Python Script Server** and restart ResInsight.