Skip to content

AP-25468: Fixed input/output ids for combined tools workflow #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hornm-knime wants to merge 1 commit into
base: master
Choose a base branch
from
+10 −15
Open

AP-25468: Fixed input/output ids for combined tools workflow #36

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 10 additions & 15 deletions org.knime.python3.nodes/src/main/java/org/knime/python3/nodes/ToolExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import org.knime.core.data.filestore.FileStore;
Expand Down Expand Up @@ -266,9 +267,9 @@ PythonPortObject getCombinedToolsWorkflow() {
spec = new WorkflowPortObjectSpec(ws, name, List.of(), List.of());
} else {
var wfm = m_combinedToolsWorkflow.loadAndGetWorkflow();
var inputIds = new ArrayList<String>();
var outputIds = new ArrayList<String>();
var ws = createWorkflowSegmentWithRemovedIONodes(wfm, inputIds, outputIds);
var ws = createWorkflowSegmentWithRemovedIONodes(wfm);
var inputIds = IntStream.range(0, ws.getConnectedInputs().size()).mapToObj(i -> "input-" + i).toList();
var outputIds = IntStream.range(0, ws.getConnectedOutputs().size()).mapToObj(i -> "output-" + i).toList();
spec = new WorkflowPortObjectSpec(ws, wfm.getName(), inputIds, outputIds);
}
var wpo = new WorkflowPortObject(spec);
Expand Down Expand Up @@ -406,16 +407,15 @@ private static WorkflowSegment createEmptyWorkflowSegment(final String name) {
return new WorkflowSegment(createEmptyWorkflow(name), List.of(), List.of(), Set.of());
}

private static WorkflowSegment createWorkflowSegmentWithRemovedIONodes(final WorkflowManager wfm,
final List<String> inputIds, final List<String> outputIds) {
private static WorkflowSegment createWorkflowSegmentWithRemovedIONodes(final WorkflowManager wfm) {
WorkflowManager segmentWfm;
segmentWfm = createEmptyWorkflow("workflow_segment");
var copyContent = WorkflowCopyContent.builder()
.setNodeIDs(wfm.getNodeContainers().stream().map(NodeContainer::getID).toArray(NodeID[]::new)).build();
segmentWfm.copyFromAndPasteHere(wfm, copyContent);
List<Input> inputs = new ArrayList<>();
List<Output> outputs = new ArrayList<>();
removeAndCollectContainerInputsAndOutputs(segmentWfm, inputs, inputIds, outputs, outputIds);
removeAndCollectContainerInputsAndOutputs(segmentWfm, inputs, outputs);
return new WorkflowSegment(segmentWfm, inputs, outputs, Set.of());
}

Expand All @@ -431,25 +431,23 @@ private static WorkflowManager createEmptyWorkflow(final String name) {
}

private static void removeAndCollectContainerInputsAndOutputs(final WorkflowManager wfm, final List<Input> inputs,
final List<String> inputIds, final List<Output> outputs, final List<String> outputIds) {
final List<Output> outputs) {
List<NodeID> nodesToRemove = new ArrayList<>();
for (NodeContainer nc : wfm.getNodeContainers()) {
if (nc instanceof NativeNodeContainer nnc
&& (collectInputs(wfm, inputs, inputIds, nnc) || collectOutputs(wfm, outputs, outputIds, nnc))) {
&& (collectInputs(wfm, inputs, nnc) || collectOutputs(wfm, outputs, nnc))) {
nodesToRemove.add(nnc.getID());
}
}
nodesToRemove.forEach(wfm::removeNode);
}

private static boolean collectOutputs(final WorkflowManager wfm, final List<Output> outputs,
final List<String> outputIds, final NativeNodeContainer nnc) {
final NativeNodeContainer nnc) {
if (nnc.getNodeModel() instanceof DefaultVirtualPortObjectOutNodeModel) {
for (ConnectionContainer cc : wfm.getIncomingConnectionsFor(nnc.getID())) {
outputs.add(new Output(nnc.getInPort(cc.getDestPort()).getPortType(), null,
new PortID(NodeIDSuffix.create(wfm.getID(), cc.getSource()), cc.getSourcePort())));
outputIds.add(wfm.getNodeContainer(cc.getSource()).getOutPort(cc.getSourcePort()).getPortName() + "-"
+ outputIds.size());
}
return true;
} else {
Expand All @@ -458,17 +456,14 @@ private static boolean collectOutputs(final WorkflowManager wfm, final List<Outp
}

private static boolean collectInputs(final WorkflowManager wfm, final List<Input> inputs,
final List<String> inputIds, final NativeNodeContainer nnc) {
final NativeNodeContainer nnc) {
if (nnc.getNodeModel() instanceof DefaultVirtualPortObjectInNodeModel) {
for (var i = 0; i < nnc.getNrOutPorts(); i++) {
Set<PortID> ports = wfm.getOutgoingConnectionsFor(nnc.getID(), i).stream()
.map(cc -> new PortID(NodeIDSuffix.create(wfm.getID(), cc.getDest()), cc.getDestPort()))
.collect(Collectors.toSet());
if (!ports.isEmpty()) {
inputs.add(new Input(nnc.getOutputType(i), null, ports));
var firstPort = ports.iterator().next();
inputIds.add(wfm.getNodeContainer(firstPort.getNodeIDSuffix().prependParent(wfm.getID()))
.getInPort(firstPort.getIndex()).getPortName() + "-" + inputIds.size());
}
}
return true;
Expand Down