Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathshared.py
More file actions
Latest commit
114 lines (92 loc) · 3.21 KB
/
Copy pathshared.py
File metadata and controls
114 lines (92 loc) · 3.21 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
fromcollectionsimportCounter
fromtypingimportAny
NODES_LABEL="nodes"
EDGES_LABEL="edges"
SOURCE_LABEL="source"
SOURCE_PORT_LABEL="sourcePort"
TARGET_LABEL="target"
TARGET_PORT_LABEL="targetPort"
VERSION_NUMBER="0.1.0"
VERSION_LABEL="version"
defget_dict(**kwargs) ->dict:
returndict(kwargs.items())
defget_list(**kwargs) ->list:
returnlist(kwargs.values())
defget_kwargs(lst: list) ->dict:
return {
t[TARGET_PORT_LABEL]: {
SOURCE_LABEL: t[SOURCE_LABEL],
SOURCE_PORT_LABEL: t[SOURCE_PORT_LABEL],
}
fortinlst
}
defget_source_handles(edges_lst: list) ->dict:
source_handle_dict: dict[Any, list] = {}
foredinedges_lst:
ifed[SOURCE_LABEL] notinsource_handle_dict:
source_handle_dict[ed[SOURCE_LABEL]] = []
source_handle_dict[ed[SOURCE_LABEL]].append(ed[SOURCE_PORT_LABEL])
return {
k: list(range(len(v))) iflen(v) >1andall(elisNoneforelinv) elsev
fork, vinsource_handle_dict.items()
}
defconvert_nodes_list_to_dict(nodes_list: list) ->dict:
return {
str(el["id"]): el["value"] if"value"inelelseel["name"]
forelinsorted(nodes_list, key=lambdad: d["id"])
}
defupdate_node_names(workflow_dict: dict) ->dict:
node_names_final_dict= {}
input_nodes= [nforninworkflow_dict[NODES_LABEL] ifn["type"] =="input"]
node_names_dict= {
n["id"]: list(
{
e[TARGET_PORT_LABEL]
foreinworkflow_dict[EDGES_LABEL]
ife[SOURCE_LABEL] ==n["id"]
}
)[0]
fornininput_nodes
}
counter_dict=Counter(node_names_dict.values())
node_names_useage_dict=dict.fromkeys(counter_dict.keys(), -1)
fork, vinnode_names_dict.items():
node_names_useage_dict[v] +=1
ifcounter_dict[v] >1:
node_names_final_dict[k] =v+"_"+str(node_names_useage_dict[v])
else:
node_names_final_dict[k] =v
forninworkflow_dict[NODES_LABEL]:
ifn["type"] =="input":
n["name"] =node_names_final_dict[n["id"]]
returnworkflow_dict
defset_result_node(workflow_dict):
node_id_lst= [n["id"] forninworkflow_dict[NODES_LABEL]]
source_lst=list({e[SOURCE_LABEL] foreinworkflow_dict[EDGES_LABEL]})
end_node_lst= []
forniinnode_id_lst:
ifninotinsource_lst:
end_node_lst.append(ni)
node_id=len(workflow_dict[NODES_LABEL])
workflow_dict[NODES_LABEL].append(
{"id": node_id, "type": "output", "name": "result"}
)
workflow_dict[EDGES_LABEL].append(
{
TARGET_LABEL: node_id,
TARGET_PORT_LABEL: None,
SOURCE_LABEL: end_node_lst[0],
SOURCE_PORT_LABEL: None,
}
)
returnworkflow_dict
defremove_result(workflow_dict):
node_output_id= [
n["id"] forninworkflow_dict[NODES_LABEL] ifn["type"] =="output"
][0]
return {
NODES_LABEL: [nforninworkflow_dict[NODES_LABEL] ifn["type"] !="output"],
EDGES_LABEL: [
eforeinworkflow_dict[EDGES_LABEL] ife[TARGET_LABEL] !=node_output_id
],
}