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.
This commit is contained in:
Kristian Bendiksen authored and Magne Sjaastad committed 2026-10-05 19:56:36 +02:00
1 parent 4175bfdf7b
commit 65942d455d
15 files changed
+1027 -530

No files matched your search

+1
View File
@@ -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
@@ -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
@@ -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() );
@@ -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 <http://www.gnu.org/licenses/gpl.html>
// for more details.
//
/////////////////////////////////////////////////////////////////////////////////
#include "RimWorkflowDescribeTools.h"
#include <QJsonArray>
#include <QJsonValue>
#include <QMap>
#include <QStringList>
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<QString, QString> typeMap = { { "Case", "EclipseCase" },
{ "Reservoir", "EclipseCase" },
{ "EclipseCase", "EclipseCase" },
{ "WellPath", "WellPath" },
{ "View", "View" },
{ "EclipseView", "View" } };
return typeMap.value( pythonType.mid( separator + 1 ) );
}
@@ -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 <http://www.gnu.org/licenses/gpl.html>
// for more details.
//
/////////////////////////////////////////////////////////////////////////////////
#pragma once
#include <QJsonObject>
#include <QString>
//==================================================================================================
/// 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
@@ -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
@@ -0,0 +1,201 @@
#include "gtest/gtest.h"
#include "RiaTestDataDirectory.h"
#include "RimWorkflowDescribeTools.h"
#include <QFile>
#include <QJsonArray>
#include <QJsonDocument>
#include <QJsonObject>
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() );
}
@@ -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"
}
}
]
}
@@ -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();
@@ -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`).
"""
@@ -8,7 +8,7 @@ import sys
def main() -> int:
if len(sys.argv) < 2:
print(
"usage: python -m rips.taskmaestro_helper <introspect|run> ...",
"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
@@ -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 <workflow_dir>", 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
@@ -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__: <type>, ...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
`<type>` 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.<Name>`; 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
@@ -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"}]
+12
View File
@@ -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.