Skip to content

delivery

Functions useful for delivery of published representations.

check_destination_path(repre_id, anatomy, anatomy_data, datetime_data, template_name)

Try to create destination path based on 'template_name'.

In the case that path cannot be filled, template contains unmatched keys, provide error message to filter out repre later.

Parameters:

Name Type Description Default
repre_id str

Representation id.

required
anatomy Anatomy

Project anatomy.

required
anatomy_data dict

Template data to fill anatomy templates.

required
datetime_data dict

Values with actual date.

required
template_name str

Name of template which should be used from anatomy templates.

required

Returns: Dict[str, List[str]]: Report of happened errors. Key is message title value is detailed information.

Source code in client/ayon_core/pipeline/delivery.py
 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
def check_destination_path(
    repre_id,
    anatomy,
    anatomy_data,
    datetime_data,
    template_name
):
    """ Try to create destination path based on 'template_name'.

    In the case that path cannot be filled, template contains unmatched
    keys, provide error message to filter out repre later.

    Args:
        repre_id (str): Representation id.
        anatomy (Anatomy): Project anatomy.
        anatomy_data (dict): Template data to fill anatomy templates.
        datetime_data (dict): Values with actual date.
        template_name (str): Name of template which should be used from anatomy
            templates.
    Returns:
        Dict[str, List[str]]: Report of happened errors. Key is message title
            value is detailed information.
    """

    anatomy_data.update(datetime_data)
    path_template = anatomy.get_template_item(
        "delivery", template_name, "path"
    )
    dest_path = path_template.format(anatomy_data)
    report_items = collections.defaultdict(list)

    if not dest_path.solved:
        msg = (
            "Missing keys in Representation's context"
            " for anatomy template \"{}\"."
        ).format(template_name)

        sub_msg = (
            "Representation: {}<br>"
        ).format(repre_id)

        if dest_path.missing_keys:
            keys = ", ".join(dest_path.missing_keys)
            sub_msg += (
                "- Missing keys: \"{}\"<br>"
            ).format(keys)

        if dest_path.invalid_types:
            items = []
            for key, value in dest_path.invalid_types.items():
                items.append("\"{}\" {}".format(key, str(value)))

            keys = ", ".join(items)
            sub_msg += (
                "- Invalid value DataType: \"{}\"<br>"
            ).format(keys)

        report_items[msg].append(sub_msg)

    return report_items

deliver_sequence(src_path, repre, anatomy, template_name, anatomy_data, format_dict, report_items, log, has_renumbered_frame=False, new_frame_start=0)

For Pype2(mainly - works in 3 too) where representation might not contain files.

Uses listing physical files (not 'files' on repre as a)might not be
 present, b)might not be reliable for representation and copying them.

 TODO Should be refactored when files are sufficient to drive all
 representations.

Parameters:

Name Type Description Default
src_path(str)

path of source representation file

required
repre dict

full representation

required
template_name string

user selected delivery template name

required
anatomy_data dict

data from repre to fill anatomy with

required
format_dict dict

root dictionary with names and values

required
report_items defaultdict

to return error messages

required
log Logger

for log printing

required

Returns:

Type Description

(collections.defaultdict, int)

Source code in client/ayon_core/pipeline/delivery.py
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def deliver_sequence(
    src_path,
    repre,
    anatomy,
    template_name,
    anatomy_data,
    format_dict,
    report_items,
    log,
    has_renumbered_frame=False,
    new_frame_start=0
):
    """ For Pype2(mainly - works in 3 too) where representation might not
        contain files.

        Uses listing physical files (not 'files' on repre as a)might not be
         present, b)might not be reliable for representation and copying them.

         TODO Should be refactored when files are sufficient to drive all
         representations.

    Args:
        src_path(str): path of source representation file
        repre (dict): full representation
        anatomy (Anatomy)
        template_name (string): user selected delivery template name
        anatomy_data (dict): data from repre to fill anatomy with
        format_dict (dict): root dictionary with names and values
        report_items (collections.defaultdict): to return error messages
        log (logging.Logger): for log printing

    Returns:
        (collections.defaultdict, int)
    """

    src_path = os.path.normpath(src_path.replace("\\", "/"))

    def hash_path_exist(myPath):
        res = myPath.replace('#', '*')
        glob_search_results = glob.glob(res)
        if len(glob_search_results) > 0:
            return True
        return False

    if not hash_path_exist(src_path):
        msg = "{} doesn't exist for {}".format(
            src_path, repre["id"])
        report_items["Source file was not found"].append(msg)
        return report_items, 0

    delivery_template = anatomy.get_template_item(
        "delivery", template_name, "path", default=None
    )
    if delivery_template is None:
        msg = (
            "Delivery template \"{}\" in anatomy of project \"{}\""
            " was not found"
        ).format(template_name, anatomy.project_name)
        report_items[""].append(msg)
        return report_items, 0

    # Check if 'frame' key is available in template which is required
    #   for sequence delivery
    if "{frame" not in delivery_template.template:
        msg = (
            "Delivery template \"{}\" in anatomy of project \"{}\""
            "does not contain '{{frame}}' key to fill. Delivery of sequence"
            " can't be processed."
        ).format(template_name, anatomy.project_name)
        report_items[""].append(msg)
        return report_items, 0

    dir_path, file_name = os.path.split(str(src_path))

    context = repre["context"]
    ext = context.get("ext", context.get("representation"))

    if not ext:
        msg = "Source extension not found, cannot find collection"
        report_items[msg].append(src_path)
        log.warning("{} <{}>".format(msg, context))
        return report_items, 0

    ext = "." + ext
    # context.representation could be .psd
    ext = ext.replace("..", ".")

    src_collections, remainder = clique.assemble(os.listdir(dir_path))
    src_collection = None
    for col in src_collections:
        if col.tail != ext:
            continue

        src_collection = col
        break

    if src_collection is None:
        msg = "Source collection of files was not found"
        report_items[msg].append(src_path)
        log.warning("{} <{}>".format(msg, src_path))
        return report_items, 0

    frame_indicator = "@####@"

    anatomy_data = copy.deepcopy(anatomy_data)
    anatomy_data["frame"] = frame_indicator
    if format_dict:
        anatomy_data["root"] = format_dict["root"]
    delivery_path = delivery_template.format_strict(anatomy_data)

    delivery_path = os.path.normpath(delivery_path.replace("\\", "/"))
    delivery_folder = os.path.dirname(delivery_path)
    dst_head, dst_tail = delivery_path.split(frame_indicator)
    dst_padding = src_collection.padding
    dst_collection = clique.Collection(
        head=dst_head,
        tail=dst_tail,
        padding=dst_padding
    )

    if not os.path.exists(delivery_folder):
        os.makedirs(delivery_folder)

    src_head = src_collection.head
    src_tail = src_collection.tail
    uploaded = 0
    first_frame = min(src_collection.indexes)
    for index in src_collection.indexes:
        src_padding = src_collection.format("{padding}") % index
        src_file_name = "{}{}{}".format(src_head, src_padding, src_tail)
        src = os.path.normpath(
            os.path.join(dir_path, src_file_name)
        )
        dst_index = index
        if has_renumbered_frame:
            # Calculate offset between first frame and current frame
            # - '0' for first frame
            offset = new_frame_start - first_frame
            # Add offset to new frame start
            dst_index = index + offset
            if dst_index < 0:
                msg = "Renumber frame has a smaller number than original frame"     # noqa
                report_items[msg].append(src_file_name)
                log.warning("{} <{}>".format(msg, context))
                return report_items, 0
        dst_padding = dst_collection.format("{padding}") % dst_index
        dst = "{}{}{}".format(dst_head, dst_padding, dst_tail)
        log.debug("Copying single: {} -> {}".format(src, dst))
        _copy_file(src, dst)

        uploaded += 1

    return report_items, uploaded

deliver_single_file(src_path, repre, anatomy, template_name, anatomy_data, format_dict, report_items, log)

Copy single file to calculated path based on template

Parameters:

Name Type Description Default
src_path(str)

path of source representation file

required
repre dict

full repre, used only in deliver_sequence, here only as to share same signature

required
template_name string

user selected delivery template name

required
anatomy_data dict

data from repre to fill anatomy with

required
format_dict dict

root dictionary with names and values

required
report_items defaultdict

to return error messages

required
log Logger

for log printing

required

Returns:

Type Description

(collections.defaultdict, int)

Source code in client/ayon_core/pipeline/delivery.py
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
def deliver_single_file(
    src_path,
    repre,
    anatomy,
    template_name,
    anatomy_data,
    format_dict,
    report_items,
    log
):
    """Copy single file to calculated path based on template

    Args:
        src_path(str): path of source representation file
        repre (dict): full repre, used only in deliver_sequence, here only
            as to share same signature
        anatomy (Anatomy)
        template_name (string): user selected delivery template name
        anatomy_data (dict): data from repre to fill anatomy with
        format_dict (dict): root dictionary with names and values
        report_items (collections.defaultdict): to return error messages
        log (logging.Logger): for log printing

    Returns:
        (collections.defaultdict, int)
    """

    # Make sure path is valid for all platforms
    src_path = os.path.normpath(src_path.replace("\\", "/"))

    if not os.path.exists(src_path):
        msg = "{} doesn't exist for {}".format(src_path, repre["id"])
        report_items["Source file was not found"].append(msg)
        return report_items, 0

    if format_dict:
        anatomy_data = copy.deepcopy(anatomy_data)
        anatomy_data["root"] = format_dict["root"]
    template_obj = anatomy.get_template_item(
        "delivery", template_name, "path"
    )
    delivery_path = template_obj.format_strict(anatomy_data)

    # Backwards compatibility when extension contained `.`
    delivery_path = delivery_path.replace("..", ".")
    # Make sure path is valid for all platforms
    delivery_path = os.path.normpath(delivery_path.replace("\\", "/"))
    # Remove newlines from the end of the string to avoid OSError during copy
    delivery_path = delivery_path.rstrip()

    delivery_folder = os.path.dirname(delivery_path)
    if not os.path.exists(delivery_folder):
        os.makedirs(delivery_folder)

    log.debug("Copying single: {} -> {}".format(src_path, delivery_path))
    _copy_file(src_path, delivery_path)

    return report_items, 1

get_format_dict(anatomy, location_path)

Returns replaced root values from user provider value.

Parameters:

Name Type Description Default
anatomy Anatomy

Project anatomy.

required
location_path str

User provided value.

required

Returns:

Type Description
dict

Prepared data for formatting of a template.

Source code in client/ayon_core/pipeline/delivery.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_format_dict(anatomy, location_path):
    """Returns replaced root values from user provider value.

    Args:
        anatomy (Anatomy): Project anatomy.
        location_path (str): User provided value.

    Returns:
        (dict): Prepared data for formatting of a template.
    """

    format_dict = {}
    if not location_path:
        return format_dict

    location_path = location_path.replace("\\", "/")
    root_names = anatomy.root_names_from_templates(
        anatomy.templates["delivery"]
    )
    format_dict["root"] = {}
    for name in root_names:
        format_dict["root"][name] = location_path
    return format_dict