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 | class CollectFromProvidedFiles(pyblish.api.ContextPlugin):
""" Collect instances prepared by the Publish node."""
label = "Collect From Provided Files"
order = pyblish.api.CollectorOrder - 0.5
hosts = ["workflow"]
@staticmethod
def _get_paths(
file_entry: Union[str, ImageSequence, Video]
) -> Union[str, List[str]]:
if isinstance(file_entry, str):
return file_entry
if isinstance(file_entry, Video):
return file_entry.path
if isinstance(file_entry, ImageSequence):
paths = list(file_entry)
if len(paths) == 1:
return paths[0] # single-frame image sequence
return paths
raise TypeError(f"Unsupported file entry provided: {file_entry}")
def process(self, context):
context.data["currentFile"] = os.path.join(os.getcwd(), "<workflow>")
instances_to_collect = context.data.pop(
"ayonWorkflowInstances",
None
)
if not instances_to_collect:
return
mandatory_keys = {
"product_name",
"product_type",
"product_base_type",
"variant",
"file_groups",
}
for instance_to_collect in instances_to_collect:
if (
not isinstance(instance_to_collect, dict)
or not mandatory_keys.issubset(set(instance_to_collect.keys()))
):
raise KnownPublishError(
f"Invalid instance to be collected: {instances_to_collect}"
f" Missing mandatory keys: {mandatory_keys}."
)
product_type = instance_to_collect["product_type"]
product_base_type = instance_to_collect["product_base_type"]
instance_data = {
"publish": True,
"active": True,
"label": instance_to_collect["product_name"],
"name": instance_to_collect["product_name"],
"productName": instance_to_collect["product_name"],
"productType": product_type,
"productBaseType": product_base_type,
"family": product_base_type,
"families": [product_base_type],
"folderPath": context.data["folderPath"],
"task": context.data.get("taskName"),
"variant": instance_to_collect["variant"],
"representations": [],
}
if isinstance(instance_to_collect.get("instance_data"), dict):
self.log.debug(
"Updating instance data from provided data: "
f"{instance_to_collect['instance_data']}"
)
instance_data.update(instance_to_collect["instance_data"])
# Collect instances from file groups
for file_group in instance_to_collect["file_groups"]:
# consolidate all inputs as RepresentationItem
if not isinstance(file_group, RepresentationItem):
file_group = RepresentationItem(input_media=file_group)
paths = self._get_paths(file_group.input_media)
repre_dict = file_group.to_repre_dict()
# Get representation extension.
path = paths[0] if isinstance(paths, list) else paths
_, ext = os.path.splitext(path)
ext = ext.strip(".")
if isinstance(paths, list):
files = [os.path.basename(pth) for pth in paths]
else:
files = os.path.basename(paths)
repre = {
"name": ext,
"ext": ext,
"files": files,
"stagingDir": os.path.dirname(path),
**repre_dict,
}
instance_data["representations"].append(repre)
instance = context.create_instance(instance_data["productName"])
instance.data.update(instance_data)
self.log.debug(
f"Collected instance: {instance_data['productName']}"
)
|