Replace staged input file with original URL

Signed-off-by: Ben Sherman <bentshermann@gmail.com>

Author: bentsherman

src/main/groovy/nextflow/prov/renderers/BcoRenderer.groovy

@@ -112,7 +112,7 @@ class BcoRenderer implements Renderer {
                     "step_number": task.id,
                     "name": task.hash.toString(),
                     "description": task.name,
-                    "input_list": task.getInputFilesMap().collect { name, source -> [
+                    "input_list": ProvHelper.getTaskInputs(task).collect { name, source -> [
                         "uri": normalizePath(source)
                     ] },
                     "output_list": ProvHelper.getTaskOutputs(task).collect { source -> [

src/main/groovy/nextflow/prov/renderers/DagRenderer.groovy

@@ -66,7 +66,7 @@ class DagRenderer implements Renderer {
     private Map<TaskRun,Vertex> getVertices(Set<TaskRun> tasks) {
         Map<TaskRun,Vertex> result = [:]
         for( final task : tasks ) {
-            final inputs = task.getInputFilesMap()
+            final inputs = ProvHelper.getTaskInputs(task)
             final outputs = ProvHelper.getTaskOutputs(task)
 
             result[task] = new Vertex(result.size(), task.name, inputs, outputs)

src/main/groovy/nextflow/prov/renderers/WrrocRenderer.groovy

@@ -230,7 +230,7 @@ class WrrocRenderer implements Renderer {
 
         final inputFiles = workflowInputs
             .findAll { source ->
-                !ProvHelper.isStagedInput(source, session) && !ProvHelper.isTmpInput(source, session)
+                !ProvHelper.isTmpInput(source, session)
             }
             .collect { source ->
                 final paramName = paramInputFiles[source]
@@ -390,19 +390,6 @@ class WrrocRenderer implements Renderer {
             }
 
         // -- workflow execution
-        final stagedInputs = workflowInputs
-            .findAll { source -> ProvHelper.isStagedInput(source, session) }
-            .collect { source ->
-                final name = getStagedInputName(source, session)
-
-                withoutNulls([
-                    "@id"           : "#stage/${name}",
-                    "@type"         : "CreativeWork",
-                    "name"          : name,
-                    "encodingFormat": getEncodingFormat(source),
-                ])
-            }
-
         final tmpInputs = workflowInputs
             .findAll { source -> ProvHelper.isTmpInput(source, session) }
             .collect { source ->
@@ -417,10 +404,9 @@ class WrrocRenderer implements Renderer {
         final taskCreateActions = tasks
             .collect { task ->
                 final processDef = processLookup[task.processor]
-                final inputs = task.getInputFilesMap().collect { name, source ->
+                final inputs = ProvHelper.getTaskInputs(task).collect { name, source ->
                     final id =
                         source in taskLookup ? getTaskOutputId(taskLookup[source], source)
-                        : ProvHelper.isStagedInput(source, session) ? "#stage/${getStagedInputName(source, session)}"
                         : ProvHelper.isTmpInput(source, session) ? "#tmp/${source.name}"
                         : normalizePath(source)
                     ["@id": id]
@@ -507,7 +493,6 @@ class WrrocRenderer implements Renderer {
                     "mainEntity" : ["@id": mainScriptId],
                     "mentions"   : [
                         ["@id": "#${session.uniqueId}"],
-                        *asReferences(stagedInputs),
                         *asReferences(tmpInputs),
                         *asReferences(taskCreateActions),
                         *asReferences(taskOutputs),
@@ -608,7 +593,6 @@ class WrrocRenderer implements Renderer {
                 *datasetParts,
                 *propertyValues,
                 *controlActions,
-                *stagedInputs,
                 *tmpInputs,
                 *taskCreateActions,
                 *taskOutputs,

src/main/groovy/nextflow/prov/util/ProvHelper.groovy

@@ -17,6 +17,7 @@
 package nextflow.prov.util
 
 import java.nio.file.Path
+import java.util.stream.Collectors
 
 import groovy.transform.CompileStatic
 import nextflow.Session
@@ -59,6 +60,20 @@ class ProvHelper {
         return session.workDir.resolve("stage-${session.uniqueId}")
     }
 
+    /**
+     * Get the input files for a task as a mapping from stage name to source path.
+     *
+     * @param task
+     */
+    static Map<String,Path> getTaskInputs(TaskRun task) {
+        final entries = task.getInputFiles().values().stream()
+            .filter(files -> files != null)
+            .flatMap(files -> files.stream())
+            .map(holder -> Map.entry(holder.getStageName(), holder.getSourcePath()))
+            .toList() as Map.Entry<String,Path>[]
+        return Map.ofEntries(entries)
+    }
+
     /**
      * Get the list of output files for a task.
      *
@@ -77,12 +92,11 @@ class ProvHelper {
      * @param tasks
      */
     static Map<Path,TaskRun> getTaskLookup(Set<TaskRun> tasks) {
-        final result = [:] as Map<Path,TaskRun>
-
-        for( def task : tasks )
+        final Map<Path,TaskRun> result = [:]
+        for( def task : tasks ) {
             for( def output : getTaskOutputs(task) )
                 result[output] = task
-
+        }
         return result
     }
 
@@ -94,28 +108,10 @@ class ProvHelper {
      * @param taskLookup
      */
     static Set<Path> getWorkflowInputs(Set<TaskRun> tasks, Map<Path,TaskRun> taskLookup) {
-        final result = [] as Set<Path>
-
-        tasks.each { task ->
-            task.getInputFilesMap().each { name, path ->
-                if( taskLookup[path] )
-                    return
-
-                result << path
-            }
-        }
-
-        return result
-    }
-
-    /**
-     * Determine whether a task input file was staged into the work directory.
-     *
-     * @param source
-     * @param session
-     */
-    static boolean isStagedInput(Path source, Session session) {
-        return source.startsWith(getStageDir(session))
+        return tasks.stream()
+            .flatMap(task -> getTaskInputs(task).values().stream())
+            .filter(path -> !taskLookup[path])
+            .collect(Collectors.toSet())
     }
 
     /**