-
Notifications
You must be signed in to change notification settings - Fork 127
/
integrate_inputlinks_ayon.py
207 lines (165 loc) · 6.85 KB
/
integrate_inputlinks_ayon.py
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
import collections
import pyblish.api
from ayon_api import (
create_link,
make_sure_link_type_exists,
get_versions_links,
)
from openpype import AYON_SERVER_ENABLED
class IntegrateInputLinksAYON(pyblish.api.ContextPlugin):
"""Connecting version level dependency links"""
order = pyblish.api.IntegratorOrder + 0.2
label = "Connect Dependency InputLinks AYON"
def process(self, context):
"""Connect dependency links for all instances, globally
Code steps:
- filter instances that integrated version
- have "versionEntity" entry in data
- separate workfile instance within filtered instances
- when workfile instance is available:
- link all `loadedVersions` as input of the workfile
- link workfile as input of all other integrated versions
- link version's inputs if it's instance have "inputVersions" entry
-
inputVersions:
The "inputVersions" in instance.data should be a list of
version ids (str), which are the dependencies of the publishing
instance that should be extracted from working scene by the DCC
specific publish plugin.
"""
workfile_instance, other_instances = self.split_instances(context)
# Variable where links are stored in submethods
new_links_by_type = collections.defaultdict(list)
self.create_workfile_links(
workfile_instance, other_instances, new_links_by_type)
self.create_generative_links(other_instances, new_links_by_type)
self.create_links_on_server(context, new_links_by_type)
def split_instances(self, context):
workfile_instance = None
other_instances = []
for instance in context:
# Skip inactive instances
if not instance.data.get("publish", True):
continue
version_doc = instance.data.get("versionEntity")
if not version_doc:
self.log.debug(
"Instance {} doesn't have version.".format(instance))
continue
family = instance.data.get("family")
if family == "workfile":
workfile_instance = instance
else:
other_instances.append(instance)
return workfile_instance, other_instances
def add_link(self, new_links_by_type, link_type, input_id, output_id):
"""Add dependency link data into temporary variable.
Args:
new_links_by_type (dict[str, list[dict[str, Any]]]): Object where
output is stored.
link_type (str): Type of link, one of 'reference' or 'generative'
input_id (str): Input version id.
output_id (str): Output version id.
"""
new_links_by_type[link_type].append((input_id, output_id))
def create_workfile_links(
self, workfile_instance, other_instances, new_links_by_type
):
if workfile_instance is None:
self.log.warn("No workfile in this publish session.")
return
workfile_version_id = workfile_instance.data["versionEntity"]["_id"]
# link workfile to all publishing versions
for instance in other_instances:
self.add_link(
new_links_by_type,
"generative",
workfile_version_id,
instance.data["versionEntity"]["_id"],
)
loaded_versions = workfile_instance.context.get("loadedVersions")
if not loaded_versions:
return
# link all loaded versions in scene into workfile
for version in loaded_versions:
self.add_link(
new_links_by_type,
"reference",
version["version"],
workfile_version_id,
)
def create_generative_links(self, other_instances, new_links_by_type):
for instance in other_instances:
input_versions = instance.data.get("inputVersions")
if not input_versions:
continue
version_entity = instance.data["versionEntity"]
for input_version in input_versions:
self.add_link(
new_links_by_type,
"generative",
input_version,
version_entity["_id"],
)
def _get_existing_links(self, project_name, link_type, entity_ids):
"""Find all existing links for given version ids.
Args:
project_name (str): Name of project.
link_type (str): Type of link.
entity_ids (set[str]): Set of version ids.
Returns:
dict[str, set[str]]: Existing links by version id.
"""
output = collections.defaultdict(set)
if not entity_ids:
return output
existing_in_links = get_versions_links(
project_name, entity_ids, [link_type], "output"
)
for entity_id, links in existing_in_links.items():
if not links:
continue
for link in links:
output[entity_id].add(link["entityId"])
return output
def create_links_on_server(self, context, new_links):
"""Create new links on server.
Args:
dict[str, list[tuple[str, str]]]: Version links by link type.
"""
if not new_links:
return
project_name = context.data["projectName"]
# Make sure link types are available on server
for link_type in new_links.keys():
make_sure_link_type_exists(
project_name, link_type, "version", "version"
)
# Create link themselves
for link_type, items in new_links.items():
mapping = collections.defaultdict(set)
# Make sure there are no duplicates of src > dst ids
for item in items:
_input_id, _output_id = item
mapping[_input_id].add(_output_id)
existing_links_by_in_id = self._get_existing_links(
project_name, link_type, set(mapping.keys())
)
for input_id, output_ids in mapping.items():
existing_links = existing_links_by_in_id[input_id]
for output_id in output_ids:
# Skip creation of link if already exists
# NOTE: AYON server does not support
# to have same links
if output_id in existing_links:
continue
create_link(
project_name,
link_type,
input_id,
"version",
output_id,
"version"
)
if not AYON_SERVER_ENABLED:
del IntegrateInputLinksAYON