Skip to content
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

workflow auto simulate orchestration changes #6309

Merged
merged 17 commits into from
Jan 30, 2025
Merged

workflow auto simulate orchestration changes #6309

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
ac39d7a
clean up workflow-controller
mwdchang Jan 29, 2025
0b3c6b0
handle workflow not found
mwdchang Jan 29, 2025
7052dd5
Fixes for delete-operator and update-positions
mwdchang Jan 30, 2025
3a1f80e
clean up
mwdchang Jan 30, 2025
ef146ed
replace ununload, route change action
mwdchang Jan 30, 2025
3a39896
layout fixes
mwdchang Jan 30, 2025
15533ec
clean up
mwdchang Jan 30, 2025
66ff6f8
tests
mwdchang Jan 30, 2025
c0fd5d7
reapply https://github.com/DARPA-ASKEM/terarium/pull/6270
mwdchang Jan 30, 2025
83a0908
Merge branch 'more-workflow-fixes' into workflow-part4
mwdchang Jan 30, 2025
fdb158a
Merge branch 'cleanup-workflow-controller' into workflow-part4
mwdchang Jan 30, 2025
0c3ca4b
Merge branch 'main' into workflow-part4
mwdchang Jan 30, 2025
d517e15
update
mwdchang Jan 30, 2025
afcb567
Merge remote-tracking branch 'origin/main' into workflow-part4
mwdchang Jan 30, 2025
cbaa11f
fix double ports
mwdchang Jan 30, 2025
3a747f3
linter
mwdchang Jan 30, 2025
e5d8763
fix edge when using quick menu
mwdchang Jan 30, 2025
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
2 changes: 1 addition & 1 deletion ...lient/hmi-client/src/components/workflow/ops/simulate-ciemss/simulate-ciemss-operation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ export const SimulateCiemssOperation: Operation = {
isOptional: true
}
],
outputs: [{ type: 'datasetId' }],
outputs: [{ type: 'datasetId', label: 'Dataset' }],
isRunnable: true,

initState: () => {
Expand Down
2 changes: 1 addition & 1 deletion ...rc/components/workflow/ops/simulate-ensemble-ciemss/simulate-ensemble-ciemss-operation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ export const SimulateEnsembleCiemssOperation: Operation = {
documentationUrl: DOCUMENTATION_URL,
imageUrl: simulateEnsembleCiemss,
inputs: [{ type: 'modelConfigId', label: 'Model configuration' }],
outputs: [{ type: 'datasetId' }],
outputs: [{ type: 'datasetId', label: 'Dataset' }],
isRunnable: true,
uniqueInputs: true,

Expand Down
107 changes: 66 additions & 41 deletions packages/client/hmi-client/src/components/workflow/tera-workflow.vue
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@
@dragging="(event) => updatePosition(node, event)"
@dragend="
isDragging = false;
saveWorkflowHandler();
debounceSaveWorkflowPositions();
"
>
<tera-operator
Expand Down Expand Up @@ -180,7 +180,7 @@

<script setup lang="ts">
import { cloneDeep, isArray, intersection, debounce } from 'lodash';
import { computed, onMounted, onUnmounted, ref, watch } from 'vue';
import { computed, onMounted, onUnmounted, ref, watch, nextTick } from 'vue';
import TeraInfiniteCanvas from '@/components/widgets/tera-infinite-canvas.vue';
import TeraCanvasItem from '@/components/widgets/tera-canvas-item.vue';
import type { Position } from '@/types/common';
Expand Down Expand Up @@ -215,7 +215,7 @@ import { useRouter, useRoute } from 'vue-router';
import { MenuItem } from 'primevue/menuitem';
import * as EventService from '@/services/event';
import { useProjects } from '@/composables/project';
import useAuthStore from '@/stores/auth';
// import useAuthStore from '@/stores/auth';
import { cloneNoteBookSession } from '@/services/notebook-session';
import * as SimulateCiemssOp from '@/components/workflow/ops/simulate-ciemss/mod';
import * as StratifyMiraOp from '@/components/workflow/ops/stratify-mira/mod';
Expand All @@ -239,7 +239,7 @@ import { activeProjectId } from '@/composables/activeProject';

const WORKFLOW_SAVE_INTERVAL = 4000;

const currentUserId = useAuthStore().user?.id;
// const currentUserId = useAuthStore().user?.id;

const registry = new workflowService.WorkflowRegistry();
registry.registerOp(SimulateCiemssOp);
Expand Down Expand Up @@ -306,33 +306,32 @@ async function updateWorkflowName(newName: string) {
wf.value.load(await workflowService.getWorkflow(props.assetId));
}

// eslint-disable-next-line
const _saveWorkflow = async () => {
await workflowService.saveWorkflow(wf.value.dump(), currentProjectId.value ?? undefined);
// wf.value.update(updated);
};
// eslint-disable-next-line
const _updateWorkflow = (event: ClientEvent<any>) => {
if (event.data.id !== wf.value.getId()) {
return;
}

const delayUpdate = isDragging || event.userId === currentUserId;
wf.value.update(event.data as Workflow, delayUpdate);
// const delayUpdate = isDragging || event.userId === currentUserId;
wf.value.update(event.data as Workflow);
};

const nodeStateMap: Map<string, any> = new Map();
const saveWorkflowDebounced = debounce(_saveWorkflow, 400);
const updateWorkflowHandler = debounce(_updateWorkflow, 250);
const saveNodeStateHandler = debounce(async () => {
const updatedWorkflow = await workflowService.updateState(wf.value.getId(), nodeStateMap);
nodeStateMap.clear();
wf.value.update(updatedWorkflow, false);

(updatedWorkflow as Workflow).nodes.forEach((node) => {
if (node.isDeleted === false) {
wf.value.updateNodeState(node.id, node.state);
}
});
}, 250);

const saveWorkflowHandler = () => {
saveWorkflowDebounced();
};
const debounceSaveWorkflowPositions = debounce(() => {
saveWorkflowPositions();
}, 100);

async function appendInput(
node: WorkflowNode<any>,
Expand All @@ -352,7 +351,7 @@ async function appendInput(
};

const updatedWorkflow = await workflowService.appendInput(wf.value.getId(), node.id, inputPort);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
}

/**
Expand Down Expand Up @@ -389,7 +388,14 @@ async function appendOutput(
};

const updatedWorkflow = await workflowService.appendOutput(wf.value.getId(), node.id, outputPort, newState);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);

// We want to try to wait here, because we replace default dummy outputs we might
// try to do id-lookup to a non-existing element
nextTick().then(() => {
relinkEdges(null);
debounceSaveWorkflowPositions();
});
}

function updateWorkflowNodeState(node: WorkflowNode<any> | null, state: any) {
Expand All @@ -414,12 +420,12 @@ async function updateWorkflowNodeStatus(node: WorkflowNode<any> | null, status:
const payload: Map<string, OperatorStatus> = new Map([[node.id, status]]);

const updatedWorkflow = await workflowService.updateStatus(wf.value.getId(), payload);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
}

async function selectOutput(node: WorkflowNode<any> | null, selectedOutputId: string) {
const updatedWorkflow = await workflowService.selectOutput(wf.value.getId(), node!.id, selectedOutputId);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
}

// Route is mutated then watcher is triggered to open or close the drilldown
Expand Down Expand Up @@ -456,14 +462,16 @@ const closeDrilldown = async () => {

const removeNode = async (nodeId: string) => {
const updatedWorkflow = await workflowService.removeNodes(wf.value.getId(), [nodeId]);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
};

const duplicateBranch = (nodeId: string) => {
const duplicateBranch = async (nodeId: string) => {
wf.value.branchWorkflow(nodeId);

cloneNoteBookSessions();
saveWorkflowHandler();

const updatedWorkflow = await workflowService.saveWorkflow(wf.value.dump(), currentProjectId.value ?? undefined);
wf.value.update(updatedWorkflow);
};

// We need to clone data-transform sessions, unlike other operators that are
Expand Down Expand Up @@ -498,7 +506,7 @@ const addOperatorToWorkflow: Function =
size: nodeSize
});
const updatedWorkflow = await workflowService.addNode(wf.value.getId(), node);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);

return node;
};
Expand Down Expand Up @@ -533,15 +541,16 @@ async function onMenuSelection(operatorType: string, menuNode: WorkflowNode<any>
target: newNode.id,
targetPortId: inputPorts[0].id,
points: [
{ x: currentPortPosition.x, y: currentPortPosition.y },
{ x: currentPortPosition.x, y: currentPortPosition.y }
{ x: 0, y: 0 },
{ x: 0, y: 0 }
]
};

const updatedWorkflow = await workflowService.addEdge(wf.value.getId(), edgePayload);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);

// Force edges to re-evaluate
await nextTick();
relinkEdges(null);
}
}
Expand Down Expand Up @@ -684,7 +693,7 @@ async function onDrop(event: DragEvent) {
}
const operator = workflowService.newOperator(wf.value.getId(), operation, newNodePosition, { state });
const updatedWorkflow = await workflowService.addNode(wf.value.getId(), operator);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
}
}

Expand Down Expand Up @@ -740,7 +749,7 @@ async function createNewEdge(node: WorkflowNode<any>, port: WorkflowPort, direct
}

const updatedWorkflow = await workflowService.addEdge(wf.value.getId(), edgePayload);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
cancelNewEdge();
}
}
Expand All @@ -754,7 +763,7 @@ async function removeEdges(portId: string) {
wf.value.getId(),
edges.map((e) => e.id)
);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
}

function onCanvasClick() {
Expand Down Expand Up @@ -802,6 +811,8 @@ function relinkEdges(node: WorkflowNode<any> | null) {
const targetNode = nodeMap.get(edge.target as string);
const targetPortElem = getPortElement(edge.targetPortId as string);

if (!sourcePortElem || !targetPortElem) continue;

edge.points[0].x = sourceNode!.x + sourceNode!.width + sourcePortElem.offsetWidth * 0.5;
edge.points[0].y = sourceNode!.y + sourcePortElem.offsetTop + sourcePortElem.offsetHeight * 0.5;
edge.points[1].x = targetNode!.x + targetPortElem.offsetWidth * 0.5;
Expand Down Expand Up @@ -873,6 +884,24 @@ function updateEdgePositions(node: WorkflowNode<any>, { x, y }) {
});
}

const saveWorkflowPositions = async () => {
const nodes = new Map(wf.value.getNodes().map((n) => [n.id, { x: n.x, y: n.y }]));
const edges = new Map(
wf.value.getEdges().map((e) => {
const start = e.points[0];
const end = e.points[1];
return [
e.id,
[
{ x: start.x, y: start.y },
{ x: end.x, y: end.y }
]
];
})
);
await workflowService.updatePositions(wf.value.getId(), nodes, edges);
};

const updatePosition = (node: WorkflowNode<any>, { x, y }) => {
const teraNode = teraOperatorRefs.value.find((operatorNode) => operatorNode.id === node.id);
if (teraNode.isEditing ?? false) {
Expand All @@ -893,7 +922,7 @@ const addAnnotationToWorkflow = async () => {
type: '',
textSize: 12
});
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
};

const updateAnnotationPosition = (annotation: WorkflowAnnotation, event: any) => {
Expand All @@ -903,12 +932,12 @@ const updateAnnotationPosition = (annotation: WorkflowAnnotation, event: any) =>

const updateAnnotation = async (annotation: WorkflowAnnotation) => {
const updatedWorkflow = await workflowService.addOrUpdateAnnotation(wf.value.getId(), annotation);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
};

const removeAnnotation = async (annotationId: string) => {
const updatedWorkflow = await workflowService.removeAnnotation(wf.value.getId(), annotationId);
wf.value.update(updatedWorkflow, false);
wf.value.update(updatedWorkflow);
};

function interpolatePointsForCurve(a: Position, b: Position): Position[] {
Expand All @@ -925,10 +954,6 @@ const pathFn = d3
// Get around typescript complaints
const drawPath = (v: any) => pathFn(v) as string;

const unloadCheck = () => {
workflowService.saveWorkflow(wf.value.dump());
};

const handleDrilldown = () => {
const operatorId = route.query?.operator?.toString();
if (operatorId) {
Expand Down Expand Up @@ -979,7 +1004,7 @@ watch(
async (newId, oldId) => {
// Save previous workflow, if applicable
if (newId !== oldId && oldId) {
workflowService.saveWorkflow(wf.value.dump());
saveWorkflowPositions();
workflowService.setLocalStorageTransform(wf.value.getId(), canvasTransform);
}

Expand Down Expand Up @@ -1015,7 +1040,7 @@ onMounted(() => {
}

document.addEventListener('mousemove', mouseUpdate);
window.addEventListener('beforeunload', unloadCheck);

saveTimer = setInterval(async () => {
workflowService.setLocalStorageTransform(wf.value.getId(), canvasTransform);
}, WORKFLOW_SAVE_INTERVAL);
Expand All @@ -1025,7 +1050,7 @@ onMounted(() => {
});

onUnmounted(() => {
workflowService.saveWorkflow(wf.value.dump());
saveWorkflowPositions();
if (saveTimer) {
clearInterval(saveTimer);
}
Expand All @@ -1035,7 +1060,7 @@ onUnmounted(() => {
workflowService.setLocalStorageTransform(wf.value.getId(), canvasTransform);
}
document.removeEventListener('mousemove', mouseUpdate);
window.removeEventListener('beforeunload', unloadCheck);
workflowService.saveWorkflow(wf.value.dump());
});
</script>

Expand Down
Loading