Skip to content

lib

AYON lib functions.

AYONSecureRegistry

Store information using keyring.

Registry should be used for private data that should be available only for user.

All passed registry names will have added prefix AYON/ to easier identify which data were created by AYON.

Parameters:

Name Type Description Default
name(str)

Name of registry used as identifier for data.

required
Source code in client/ayon_core/lib/local_settings.py
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class AYONSecureRegistry:
    """Store information using keyring.

    Registry should be used for private data that should be available only for
    user.

    All passed registry names will have added prefix `AYON/` to easier
    identify which data were created by AYON.

    Args:
        name(str): Name of registry used as identifier for data.
    """
    def __init__(self, name):
        try:
            import keyring

        except Exception:
            raise NotImplementedError(
                "Python module `keyring` is not available."
            )

        # hack for cx_freeze and Windows keyring backend
        if platform.system().lower() == "windows":
            from keyring.backends import Windows

            keyring.set_keyring(Windows.WinVaultKeyring())

        # Force "AYON" prefix
        self._name = "/".join(("AYON", name))

    def set_item(self, name, value):
        # type: (str, str) -> None
        """Set sensitive item into system's keyring.

        This uses `Keyring module`_ to save sensitive stuff into system's
        keyring.

        Args:
            name (str): Name of the item.
            value (str): Value of the item.

        .. _Keyring module:
            https://github.com/jaraco/keyring

        """
        import keyring

        keyring.set_password(self._name, name, value)

    @lru_cache(maxsize=32)
    def get_item(self, name, default=_PLACEHOLDER):
        """Get value of sensitive item from system's keyring.

        See also `Keyring module`_

        Args:
            name (str): Name of the item.
            default (Any): Default value if item is not available.

        Returns:
            value (str): Value of the item.

        Raises:
            ValueError: If item doesn't exist and default is not defined.

        .. _Keyring module:
            https://github.com/jaraco/keyring

        """
        import keyring

        value = keyring.get_password(self._name, name)
        if value is not None:
            return value

        if default is not _PLACEHOLDER:
            return default

        # NOTE Should raise `KeyError`
        raise ValueError(
            "Item {}:{} does not exist in keyring.".format(self._name, name)
        )

    def delete_item(self, name):
        # type: (str) -> None
        """Delete value stored in system's keyring.

        See also `Keyring module`_

        Args:
            name (str): Name of the item to be deleted.

        .. _Keyring module:
            https://github.com/jaraco/keyring

        """
        import keyring

        self.get_item.cache_clear()
        keyring.delete_password(self._name, name)

delete_item(name)

Delete value stored in system's keyring.

See also Keyring module_

Parameters:

Name Type Description Default
name str

Name of the item to be deleted.

required

.. _Keyring module: https://github.com/jaraco/keyring

Source code in client/ayon_core/lib/local_settings.py
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def delete_item(self, name):
    # type: (str) -> None
    """Delete value stored in system's keyring.

    See also `Keyring module`_

    Args:
        name (str): Name of the item to be deleted.

    .. _Keyring module:
        https://github.com/jaraco/keyring

    """
    import keyring

    self.get_item.cache_clear()
    keyring.delete_password(self._name, name)

get_item(name, default=_PLACEHOLDER) cached

Get value of sensitive item from system's keyring.

See also Keyring module_

Parameters:

Name Type Description Default
name str

Name of the item.

required
default Any

Default value if item is not available.

_PLACEHOLDER

Returns:

Name Type Description
value str

Value of the item.

Raises:

Type Description
ValueError

If item doesn't exist and default is not defined.

.. _Keyring module: https://github.com/jaraco/keyring

Source code in client/ayon_core/lib/local_settings.py
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
@lru_cache(maxsize=32)
def get_item(self, name, default=_PLACEHOLDER):
    """Get value of sensitive item from system's keyring.

    See also `Keyring module`_

    Args:
        name (str): Name of the item.
        default (Any): Default value if item is not available.

    Returns:
        value (str): Value of the item.

    Raises:
        ValueError: If item doesn't exist and default is not defined.

    .. _Keyring module:
        https://github.com/jaraco/keyring

    """
    import keyring

    value = keyring.get_password(self._name, name)
    if value is not None:
        return value

    if default is not _PLACEHOLDER:
        return default

    # NOTE Should raise `KeyError`
    raise ValueError(
        "Item {}:{} does not exist in keyring.".format(self._name, name)
    )

set_item(name, value)

Set sensitive item into system's keyring.

This uses Keyring module_ to save sensitive stuff into system's keyring.

Parameters:

Name Type Description Default
name str

Name of the item.

required
value str

Value of the item.

required

.. _Keyring module: https://github.com/jaraco/keyring

Source code in client/ayon_core/lib/local_settings.py
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def set_item(self, name, value):
    # type: (str, str) -> None
    """Set sensitive item into system's keyring.

    This uses `Keyring module`_ to save sensitive stuff into system's
    keyring.

    Args:
        name (str): Name of the item.
        value (str): Value of the item.

    .. _Keyring module:
        https://github.com/jaraco/keyring

    """
    import keyring

    keyring.set_password(self._name, name, value)

AYONSettingsRegistry

Bases: JSONSettingRegistry

Class handling AYON general settings registry.

Parameters:

Name Type Description Default
name Optional[str]

Name of the registry.

None
Source code in client/ayon_core/lib/local_settings.py
550
551
552
553
554
555
556
557
558
559
560
561
class AYONSettingsRegistry(JSONSettingRegistry):
    """Class handling AYON general settings registry.

    Args:
        name (Optional[str]): Name of the registry.
    """

    def __init__(self, name=None):
        if not name:
            name = "AYON_settings"
        path = get_launcher_storage_dir()
        super(AYONSettingsRegistry, self).__init__(name, path)

AbstractAttrDef

Abstraction of attribute definition.

Each attribute definition must have implemented validation and conversion method.

Attribute definition should have ability to return "default" value. That can be based on passed data into __init__ so is not abstracted to attribute.

QUESTION: How to force to set key attribute?

Parameters:

Name Type Description Default
key str

Under which key will be attribute value stored.

required
default Any

Default value of an attribute.

required
label Optional[str]

Attribute label.

None
tooltip Optional[str]

Attribute tooltip.

None
is_label_horizontal Optional[bool]

UI specific argument. Specify if label is next to value input or ahead.

None
visible Optional[bool]

Item is shown to user (for UI purposes).

None
enabled Optional[bool]

Item is enabled (for UI purposes).

None
hidden Optional[bool]

DEPRECATED: Use 'visible' instead.

None
disabled Optional[bool]

DEPRECATED: Use 'enabled' instead.

None
Source code in client/ayon_core/lib/attribute_definitions.py
 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
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
class AbstractAttrDef(metaclass=AbstractAttrDefMeta):
    """Abstraction of attribute definition.

    Each attribute definition must have implemented validation and
    conversion method.

    Attribute definition should have ability to return "default" value. That
    can be based on passed data into `__init__` so is not abstracted to
    attribute.

    QUESTION:
    How to force to set `key` attribute?

    Args:
        key (str): Under which key will be attribute value stored.
        default (Any): Default value of an attribute.
        label (Optional[str]): Attribute label.
        tooltip (Optional[str]): Attribute tooltip.
        is_label_horizontal (Optional[bool]): UI specific argument. Specify
            if label is next to value input or ahead.
        visible (Optional[bool]): Item is shown to user (for UI purposes).
        enabled (Optional[bool]): Item is enabled (for UI purposes).
        hidden (Optional[bool]): DEPRECATED: Use 'visible' instead.
        disabled (Optional[bool]): DEPRECATED: Use 'enabled' instead.

    """
    type_attributes = []

    is_value_def = True

    def __init__(
        self,
        key: str,
        default: Any,
        label: Optional[str] = None,
        tooltip: Optional[str] = None,
        is_label_horizontal: Optional[bool] = None,
        visible: Optional[bool] = None,
        enabled: Optional[bool] = None,
        hidden: Optional[bool] = None,
        disabled: Optional[bool] = None,
    ):
        if is_label_horizontal is None:
            is_label_horizontal = True

        enabled = _convert_reversed_attr(
            enabled, disabled, "enabled", "disabled", True
        )
        visible = _convert_reversed_attr(
            visible, hidden, "visible", "hidden", True
        )

        self.key: str = key
        self.label: Optional[str] = label
        self.tooltip: Optional[str] = tooltip
        self.default: Any = default
        self.is_label_horizontal: bool = is_label_horizontal
        self.visible: bool = visible
        self.enabled: bool = enabled
        self._id: str = uuid.uuid4().hex

        self.__init__class__ = AbstractAttrDef

    @property
    def id(self) -> str:
        return self._id

    def clone(self) -> "Self":
        data = self.serialize()
        data.pop("type")
        return self.deserialize(data)

    @property
    def hidden(self) -> bool:
        return not self.visible

    @hidden.setter
    def hidden(self, value: bool):
        self.visible = not value

    @property
    def disabled(self) -> bool:
        return not self.enabled

    @disabled.setter
    def disabled(self, value: bool):
        self.enabled = not value

    def __eq__(self, other: Any) -> bool:
        return self.compare_to_def(other)

    def __ne__(self, other: Any) -> bool:
        return not self.compare_to_def(other)

    def compare_to_def(
        self,
        other: Any,
        ignore_default: Optional[bool] = False,
        ignore_enabled: Optional[bool] = False,
        ignore_visible: Optional[bool] = False,
        ignore_def_type_compare: Optional[bool] = False,
    ) -> bool:
        if not isinstance(other, self.__class__) or self.key != other.key:
            return False
        if not ignore_def_type_compare and not self._def_type_compare(other):
            return False
        return (
            (ignore_default or self.default == other.default)
            and (ignore_visible or self.visible == other.visible)
            and (ignore_enabled or self.enabled == other.enabled)
        )

    @abstractmethod
    def is_value_valid(self, value: Any) -> bool:
        """Check if value is valid.

        This should return False if value is not valid based
            on definition type.

        Args:
            value (Any): Value to validate based on definition type.

        Returns:
            bool: True if value is valid.

        """
        pass

    @property
    @abstractmethod
    def type(self) -> str:
        """Attribute definition type also used as identifier of class.

        Returns:
            str: Type of attribute definition.

        """
        pass

    @abstractmethod
    def convert_value(self, value: Any) -> Any:
        """Convert value to a valid one.

        Convert passed value to a valid type. Use default if value can't be
        converted.

        """
        pass

    def serialize(self) -> Dict[str, Any]:
        """Serialize object to data so it's possible to recreate it.

        Returns:
            Dict[str, Any]: Serialized object that can be passed to
                'deserialize' method.

        """
        data = {
            "type": self.type,
            "key": self.key,
            "label": self.label,
            "tooltip": self.tooltip,
            "default": self.default,
            "is_label_horizontal": self.is_label_horizontal,
            "visible": self.visible,
            "enabled": self.enabled
        }
        for attr in self.type_attributes:
            data[attr] = getattr(self, attr)
        return data

    @classmethod
    def deserialize(cls, data: Dict[str, Any]) -> "Self":
        """Recreate object from data.

        Data can be received using 'serialize' method.
        """
        if "type" in data:
            data = dict(data)
            data.pop("type")

        return cls(**data)

    def _def_type_compare(self, other: "Self") -> bool:
        return True

type abstractmethod property

Attribute definition type also used as identifier of class.

Returns:

Name Type Description
str str

Type of attribute definition.

convert_value(value) abstractmethod

Convert value to a valid one.

Convert passed value to a valid type. Use default if value can't be converted.

Source code in client/ayon_core/lib/attribute_definitions.py
238
239
240
241
242
243
244
245
246
@abstractmethod
def convert_value(self, value: Any) -> Any:
    """Convert value to a valid one.

    Convert passed value to a valid type. Use default if value can't be
    converted.

    """
    pass

deserialize(data) classmethod

Recreate object from data.

Data can be received using 'serialize' method.

Source code in client/ayon_core/lib/attribute_definitions.py
270
271
272
273
274
275
276
277
278
279
280
@classmethod
def deserialize(cls, data: Dict[str, Any]) -> "Self":
    """Recreate object from data.

    Data can be received using 'serialize' method.
    """
    if "type" in data:
        data = dict(data)
        data.pop("type")

    return cls(**data)

is_value_valid(value) abstractmethod

Check if value is valid.

This should return False if value is not valid based on definition type.

Parameters:

Name Type Description Default
value Any

Value to validate based on definition type.

required

Returns:

Name Type Description
bool bool

True if value is valid.

Source code in client/ayon_core/lib/attribute_definitions.py
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@abstractmethod
def is_value_valid(self, value: Any) -> bool:
    """Check if value is valid.

    This should return False if value is not valid based
        on definition type.

    Args:
        value (Any): Value to validate based on definition type.

    Returns:
        bool: True if value is valid.

    """
    pass

serialize()

Serialize object to data so it's possible to recreate it.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: Serialized object that can be passed to 'deserialize' method.

Source code in client/ayon_core/lib/attribute_definitions.py
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def serialize(self) -> Dict[str, Any]:
    """Serialize object to data so it's possible to recreate it.

    Returns:
        Dict[str, Any]: Serialized object that can be passed to
            'deserialize' method.

    """
    data = {
        "type": self.type,
        "key": self.key,
        "label": self.label,
        "tooltip": self.tooltip,
        "default": self.default,
        "is_label_horizontal": self.is_label_horizontal,
        "visible": self.visible,
        "enabled": self.enabled
    }
    for attr in self.type_attributes:
        data[attr] = getattr(self, attr)
    return data

BoolDef

Bases: AbstractAttrDef

Boolean representation.

Parameters:

Name Type Description Default
default(bool)

Default value. Set to False if not defined.

required
Source code in client/ayon_core/lib/attribute_definitions.py
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
class BoolDef(AbstractAttrDef):
    """Boolean representation.

    Args:
        default(bool): Default value. Set to `False` if not defined.

    """
    type = "bool"

    def __init__(self, key: str, default: Optional[bool] = None, **kwargs):
        if default is None:
            default = False
        super().__init__(key, default=default, **kwargs)

    def is_value_valid(self, value: Any) -> bool:
        return isinstance(value, bool)

    def convert_value(self, value: Any) -> bool:
        if isinstance(value, bool):
            return value
        return self.default

CacheItem

Simple cache item with lifetime and default factory for default value.

Default factory should return default value that is used on init and on reset.

Parameters:

Name Type Description Default
default_factory Optional[callable]

Function that returns default value used on init and on reset.

None
lifetime Optional[int]

Lifetime of the cache data in seconds. Default lifetime is 120 seconds.

None
Source code in client/ayon_core/lib/cache.py
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
class CacheItem:
    """Simple cache item with lifetime and default factory for default value.

    Default factory should return default value that is used on init
        and on reset.

    Args:
        default_factory (Optional[callable]): Function that returns default
            value used on init and on reset.
        lifetime (Optional[int]): Lifetime of the cache data in seconds.
            Default lifetime is 120 seconds.

    """
    def __init__(self, default_factory=None, lifetime=None):
        if lifetime is None:
            lifetime = 120
        self._lifetime = lifetime
        self._last_update = None
        if default_factory is None:
            default_factory = _default_factory_func
        self._default_factory = default_factory
        self._data = default_factory()

    @property
    def is_valid(self):
        """Is cache valid to use.

        Return:
            bool: True if cache is valid, False otherwise.

        """
        if self._last_update is None:
            return False

        return (time.time() - self._last_update) < self._lifetime

    def set_lifetime(self, lifetime):
        """Change lifetime of cache item.

        Args:
            lifetime (int): Lifetime of the cache data in seconds.
        """

        self._lifetime = lifetime

    def set_invalid(self):
        """Set cache as invalid."""

        self._last_update = None

    def reset(self):
        """Set cache as invalid and reset data."""

        self._last_update = None
        self._data = self._default_factory()

    def get_data(self):
        """Receive cached data.

        Returns:
            Any: Any data that are cached.

        """
        return self._data

    def update_data(self, data):
        """Update cache data.

        Args:
            data (Any): Any data that are cached.

        """
        self._data = data
        self._last_update = time.time()

is_valid property

Is cache valid to use.

Return

bool: True if cache is valid, False otherwise.

get_data()

Receive cached data.

Returns:

Name Type Description
Any

Any data that are cached.

Source code in client/ayon_core/lib/cache.py
70
71
72
73
74
75
76
77
def get_data(self):
    """Receive cached data.

    Returns:
        Any: Any data that are cached.

    """
    return self._data

reset()

Set cache as invalid and reset data.

Source code in client/ayon_core/lib/cache.py
64
65
66
67
68
def reset(self):
    """Set cache as invalid and reset data."""

    self._last_update = None
    self._data = self._default_factory()

set_invalid()

Set cache as invalid.

Source code in client/ayon_core/lib/cache.py
59
60
61
62
def set_invalid(self):
    """Set cache as invalid."""

    self._last_update = None

set_lifetime(lifetime)

Change lifetime of cache item.

Parameters:

Name Type Description Default
lifetime int

Lifetime of the cache data in seconds.

required
Source code in client/ayon_core/lib/cache.py
50
51
52
53
54
55
56
57
def set_lifetime(self, lifetime):
    """Change lifetime of cache item.

    Args:
        lifetime (int): Lifetime of the cache data in seconds.
    """

    self._lifetime = lifetime

update_data(data)

Update cache data.

Parameters:

Name Type Description Default
data Any

Any data that are cached.

required
Source code in client/ayon_core/lib/cache.py
79
80
81
82
83
84
85
86
87
def update_data(self, data):
    """Update cache data.

    Args:
        data (Any): Any data that are cached.

    """
    self._data = data
    self._last_update = time.time()

EnumDef

Bases: AbstractAttrDef

Enumeration of items.

Enumeration of single item from items. Or list of items if multiselection is enabled.

Parameters:

Name Type Description Default
key str

Key under which value is stored.

required
items EnumItemsInputType

Items definition that can be converted using 'prepare_enum_items'.

required
default Optional[Any]

Default value. Must be one key(value) from passed items or list of values for multiselection.

None
multiselection Optional[bool]

If True, multiselection is allowed. Output is list of selected items.

False
placeholder Optional[str]

Placeholder for UI purposes, only for multiselection enumeration.

None
Source code in client/ayon_core/lib/attribute_definitions.py
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
class EnumDef(AbstractAttrDef):
    """Enumeration of items.

    Enumeration of single item from items. Or list of items if multiselection
    is enabled.

    Args:
        key (str): Key under which value is stored.
        items (EnumItemsInputType): Items definition that can be converted
            using 'prepare_enum_items'.
        default (Optional[Any]): Default value. Must be one key(value) from
            passed items or list of values for multiselection.
        multiselection (Optional[bool]): If True, multiselection is allowed.
            Output is list of selected items.
        placeholder (Optional[str]): Placeholder for UI purposes, only for
            multiselection enumeration.

    """
    type = "enum"

    type_attributes = [
        "multiselection",
        "placeholder",
    ]

    def __init__(
        self,
        key: str,
        items: "EnumItemsInputType",
        default: "Union[str, List[Any]]" = None,
        multiselection: Optional[bool] = False,
        placeholder: Optional[str] = None,
        **kwargs
    ):
        if multiselection is None:
            multiselection = False

        if not items and not multiselection:
            raise ValueError(
                f"Empty 'items' value. {self.__class__.__name__} must have"
                " defined values on initialization."
            )

        items = self.prepare_enum_items(items)
        item_values = [item["value"] for item in items]
        item_values_set = set(item_values)

        if multiselection:
            if default is None:
                default = []
            default = list(item_values_set.intersection(default))

        elif default not in item_values:
            default = next(iter(item_values), None)

        super().__init__(key, default=default, **kwargs)

        self.items: List["EnumItemDict"] = items
        self._item_values: Set[Any] = item_values_set
        self.multiselection: bool = multiselection
        self.placeholder: Optional[str] = placeholder

    def convert_value(self, value):
        if not self.multiselection:
            if value in self._item_values:
                return value
            return self.default

        if value is None:
            return copy.deepcopy(self.default)
        return list(self._item_values.intersection(value))

    def is_value_valid(self, value: Any) -> bool:
        """Check if item is available in possible values."""
        if isinstance(value, list):
            if not self.multiselection:
                return False
            return all(value in self._item_values for value in value)

        if self.multiselection:
            return False
        return value in self._item_values

    def serialize(self):
        data = super().serialize()
        data["items"] = copy.deepcopy(self.items)
        return data

    @staticmethod
    def prepare_enum_items(
        items: "EnumItemsInputType"
    ) -> List["EnumItemDict"]:
        """Convert items to unified structure.

        Output is a list where each item is dictionary with 'value'
        and 'label'.

        ```python
        # Example output
        [
            {"label": "Option 1", "value": 1},
            {"label": "Option 2", "value": 2},
            {"label": "Option 3", "value": 3}
        ]
        ```

        Args:
            items (EnumItemsInputType): The items to convert.

        Returns:
            List[EnumItemDict]: Unified structure of items.

        """
        output = []
        if isinstance(items, dict):
            for value, label in items.items():
                output.append({"label": label, "value": value})

        elif isinstance(items, (tuple, list, set)):
            for item in items:
                if isinstance(item, dict):
                    # Validate if 'value' is available
                    if "value" not in item:
                        raise KeyError("Item does not contain 'value' key.")

                    if "label" not in item:
                        item["label"] = str(item["value"])
                elif isinstance(item, (list, tuple)):
                    if len(item) == 2:
                        value, label = item
                    elif len(item) == 1:
                        value = item[0]
                        label = str(value)
                    else:
                        raise ValueError((
                            "Invalid items count {}."
                            " Expected 1 or 2. Value: {}"
                        ).format(len(item), str(item)))

                    item = {"label": label, "value": value}
                else:
                    item = {"label": str(item), "value": item}
                output.append(item)

        else:
            raise TypeError(
                "Unknown type for enum items '{}'".format(type(items))
            )

        return output

    def _def_type_compare(self, other: "EnumDef") -> bool:
        return (
            self.items == other.items
            and self.multiselection == other.multiselection
        )

is_value_valid(value)

Check if item is available in possible values.

Source code in client/ayon_core/lib/attribute_definitions.py
611
612
613
614
615
616
617
618
619
620
def is_value_valid(self, value: Any) -> bool:
    """Check if item is available in possible values."""
    if isinstance(value, list):
        if not self.multiselection:
            return False
        return all(value in self._item_values for value in value)

    if self.multiselection:
        return False
    return value in self._item_values

prepare_enum_items(items) staticmethod

Convert items to unified structure.

Output is a list where each item is dictionary with 'value' and 'label'.

# Example output
[
    {"label": "Option 1", "value": 1},
    {"label": "Option 2", "value": 2},
    {"label": "Option 3", "value": 3}
]

Parameters:

Name Type Description Default
items EnumItemsInputType

The items to convert.

required

Returns:

Type Description
List[EnumItemDict]

List[EnumItemDict]: Unified structure of items.

Source code in client/ayon_core/lib/attribute_definitions.py
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
@staticmethod
def prepare_enum_items(
    items: "EnumItemsInputType"
) -> List["EnumItemDict"]:
    """Convert items to unified structure.

    Output is a list where each item is dictionary with 'value'
    and 'label'.

    ```python
    # Example output
    [
        {"label": "Option 1", "value": 1},
        {"label": "Option 2", "value": 2},
        {"label": "Option 3", "value": 3}
    ]
    ```

    Args:
        items (EnumItemsInputType): The items to convert.

    Returns:
        List[EnumItemDict]: Unified structure of items.

    """
    output = []
    if isinstance(items, dict):
        for value, label in items.items():
            output.append({"label": label, "value": value})

    elif isinstance(items, (tuple, list, set)):
        for item in items:
            if isinstance(item, dict):
                # Validate if 'value' is available
                if "value" not in item:
                    raise KeyError("Item does not contain 'value' key.")

                if "label" not in item:
                    item["label"] = str(item["value"])
            elif isinstance(item, (list, tuple)):
                if len(item) == 2:
                    value, label = item
                elif len(item) == 1:
                    value = item[0]
                    label = str(value)
                else:
                    raise ValueError((
                        "Invalid items count {}."
                        " Expected 1 or 2. Value: {}"
                    ).format(len(item), str(item)))

                item = {"label": label, "value": value}
            else:
                item = {"label": str(item), "value": item}
            output.append(item)

    else:
        raise TypeError(
            "Unknown type for enum items '{}'".format(type(items))
        )

    return output

FileDef

Bases: AbstractAttrDef

File definition. It is possible to define filters of allowed file extensions and if supports folders. Args: single_item(bool): Allow only single path item. folders(bool): Allow folder paths. extensions(List[str]): Allow files with extensions. Empty list will allow all extensions and None will disable files completely. extensions_label(str): Custom label shown instead of extensions in UI. default(str, List[str]): Default value.

Source code in client/ayon_core/lib/attribute_definitions.py
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
class FileDef(AbstractAttrDef):
    """File definition.
    It is possible to define filters of allowed file extensions and if supports
    folders.
    Args:
        single_item(bool): Allow only single path item.
        folders(bool): Allow folder paths.
        extensions(List[str]): Allow files with extensions. Empty list will
            allow all extensions and None will disable files completely.
        extensions_label(str): Custom label shown instead of extensions in UI.
        default(str, List[str]): Default value.
    """

    type = "path"
    type_attributes = [
        "single_item",
        "folders",
        "extensions",
        "allow_sequences",
        "extensions_label",
    ]

    def __init__(
        self,
        key: str,
        single_item: Optional[bool] = True,
        folders: Optional[bool] = None,
        extensions: Optional[Iterable[str]] = None,
        allow_sequences: Optional[bool] = True,
        extensions_label: Optional[str] = None,
        default: Optional["Union[FileDefItemDict, List[str]]"] = None,
        **kwargs
    ):
        if folders is None and extensions is None:
            folders = True
            extensions = []

        if default is None:
            if single_item:
                default = FileDefItem.create_empty_item().to_dict()
            else:
                default = []
        else:
            if single_item:
                if isinstance(default, dict):
                    FileDefItem.from_dict(default)

                elif isinstance(default, str):
                    default = FileDefItem.from_paths(
                        [default.strip()], allow_sequences
                    )[0]

                else:
                    raise TypeError((
                        "'default' argument must be 'str' or 'dict' not '{}'"
                    ).format(type(default)))

            else:
                if not isinstance(default, (tuple, list, set)):
                    raise TypeError((
                        "'default' argument must be 'list', 'tuple' or 'set'"
                        ", not '{}'"
                    ).format(type(default)))

        # Change horizontal label
        is_label_horizontal = kwargs.get("is_label_horizontal")
        if is_label_horizontal is None:
            kwargs["is_label_horizontal"] = False

        self.single_item: bool = single_item
        self.folders: bool = folders
        self.extensions: Set[str] = set(extensions)
        self.allow_sequences: bool = allow_sequences
        self.extensions_label: Optional[str] = extensions_label
        super().__init__(key, default=default, **kwargs)

    def __eq__(self, other: Any) -> bool:
        if not super().__eq__(other):
            return False

        return (
            self.single_item == other.single_item
            and self.folders == other.folders
            and self.extensions == other.extensions
            and self.allow_sequences == other.allow_sequences
        )

    def is_value_valid(self, value: Any) -> bool:
        if self.single_item:
            if not isinstance(value, dict):
                return False
            try:
                FileDefItem.from_dict(value)
                return True
            except (ValueError, KeyError):
                return False

        if not isinstance(value, list):
            return False

        for item in value:
            if not isinstance(item, dict):
                return False

            try:
                FileDefItem.from_dict(item)
            except (ValueError, KeyError):
                return False
        return True

    def convert_value(
        self, value: Any
    ) -> "Union[FileDefItemDict, List[FileDefItemDict]]":
        if isinstance(value, (str, dict)):
            value = [value]

        if isinstance(value, (tuple, list, set)):
            string_paths = []
            dict_items = []
            for item in value:
                if isinstance(item, str):
                    string_paths.append(item.strip())
                elif isinstance(item, dict):
                    try:
                        FileDefItem.from_dict(item)
                        dict_items.append(item)
                    except (ValueError, KeyError):
                        pass

            if string_paths:
                file_items = FileDefItem.from_paths(
                    string_paths, self.allow_sequences
                )
                dict_items.extend([
                    file_item.to_dict()
                    for file_item in file_items
                ])

            if not self.single_item:
                return dict_items

            if not dict_items:
                return self.default
            return dict_items[0]

        if self.single_item:
            return FileDefItem.create_empty_item().to_dict()
        return []

FileDefItem

Source code in client/ayon_core/lib/attribute_definitions.py
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
class FileDefItem:
    def __init__(
        self,
        directory: str,
        filenames: List[str],
        frames: Optional[List[int]] = None,
        template: Optional[str] = None,
    ):
        self.directory = directory

        self.filenames = []
        self.is_sequence = False
        self.template = None
        self.frames = []
        self.is_empty = True

        self.set_filenames(filenames, frames, template)

    def __str__(self):
        return json.dumps(self.to_dict())

    def __repr__(self):
        if self.is_empty:
            filename = "< empty >"
        elif self.is_sequence:
            filename = self.template
        else:
            filename = self.filenames[0]

        return "<{}: \"{}\">".format(
            self.__class__.__name__,
            os.path.join(self.directory, filename)
        )

    @property
    def label(self) -> Optional[str]:
        if self.is_empty:
            return None

        if not self.is_sequence:
            return self.filenames[0]

        frame_start = self.frames[0]
        filename_template = os.path.basename(self.template)
        if len(self.frames) == 1:
            return "{} [{}]".format(filename_template, frame_start)

        frame_end = self.frames[-1]
        expected_len = (frame_end - frame_start) + 1
        if expected_len == len(self.frames):
            return "{} [{}-{}]".format(
                filename_template, frame_start, frame_end
            )

        ranges = []
        _frame_start = None
        _frame_end = None
        for frame in range(frame_start, frame_end + 1):
            if frame not in self.frames:
                add_to_ranges = _frame_start is not None
            elif _frame_start is None:
                _frame_start = _frame_end = frame
                add_to_ranges = frame == frame_end
            else:
                _frame_end = frame
                add_to_ranges = frame == frame_end

            if add_to_ranges:
                if _frame_start != _frame_end:
                    _range = "{}-{}".format(_frame_start, _frame_end)
                else:
                    _range = str(_frame_start)
                ranges.append(_range)
                _frame_start = _frame_end = None
        return "{} [{}]".format(
            filename_template, ",".join(ranges)
        )

    def split_sequence(self) -> List["Self"]:
        if not self.is_sequence:
            raise ValueError("Cannot split single file item")

        paths = [
            os.path.join(self.directory, filename)
            for filename in self.filenames
        ]
        return self.from_paths(paths, False)

    @property
    def ext(self) -> Optional[str]:
        if self.is_empty:
            return None
        _, ext = os.path.splitext(self.filenames[0])
        if ext:
            return ext
        return None

    @property
    def lower_ext(self) -> Optional[str]:
        ext = self.ext
        if ext is not None:
            return ext.lower()
        return ext

    @property
    def is_dir(self) -> bool:
        if self.is_empty:
            return False

        # QUESTION a better way how to define folder (in init argument?)
        if self.ext:
            return False
        return True

    def set_directory(self, directory: str):
        self.directory = directory

    def set_filenames(
        self,
        filenames: List[str],
        frames: Optional[List[int]] = None,
        template: Optional[str] = None,
    ):
        if frames is None:
            frames = []
        is_sequence = False
        if frames:
            is_sequence = True

        if is_sequence and not template:
            raise ValueError("Missing template for sequence")

        self.is_empty = len(filenames) == 0
        self.filenames = filenames
        self.template = template
        self.frames = frames
        self.is_sequence = is_sequence

    @classmethod
    def create_empty_item(cls) -> "Self":
        return cls("", "")

    @classmethod
    def from_value(
        cls,
        value: "Union[List[FileDefItemDict], FileDefItemDict]",
        allow_sequences: bool,
    ) -> List["Self"]:
        """Convert passed value to FileDefItem objects.

        Returns:
            list: Created FileDefItem objects.

        """
        # Convert single item to iterable
        if not isinstance(value, (list, tuple, set)):
            value = [value]

        output = []
        str_filepaths = []
        for item in value:
            if isinstance(item, dict):
                item = cls.from_dict(item)

            if isinstance(item, FileDefItem):
                if not allow_sequences and item.is_sequence:
                    output.extend(item.split_sequence())
                else:
                    output.append(item)

            elif isinstance(item, str):
                str_filepaths.append(item)
            else:
                raise TypeError(
                    "Unknown type \"{}\". Can't convert to {}".format(
                        str(type(item)), cls.__name__
                    )
                )

        if str_filepaths:
            output.extend(cls.from_paths(str_filepaths, allow_sequences))

        return output

    @classmethod
    def from_dict(cls, data: "FileDefItemDict") -> "Self":
        return cls(
            data["directory"],
            data["filenames"],
            data.get("frames"),
            data.get("template")
        )

    @classmethod
    def from_paths(
        cls,
        paths: List[str],
        allow_sequences: bool,
    ) -> List["Self"]:
        filenames_by_dir = collections.defaultdict(list)
        for path in paths:
            normalized = os.path.normpath(path)
            directory, filename = os.path.split(normalized)
            filenames_by_dir[directory].append(filename)

        output = []
        for directory, filenames in filenames_by_dir.items():
            if allow_sequences:
                cols, remainders = clique.assemble(filenames)
            else:
                cols = []
                remainders = filenames

            for remainder in remainders:
                output.append(cls(directory, [remainder]))

            for col in cols:
                frames = list(col.indexes)
                paths = [filename for filename in col]
                template = col.format("{head}{padding}{tail}")

                output.append(cls(
                    directory, paths, frames, template
                ))

        return output

    def to_dict(self) -> "FileDefItemDict":
        output = {
            "is_sequence": self.is_sequence,
            "directory": self.directory,
            "filenames": list(self.filenames),
        }
        if self.is_sequence:
            output.update({
                "template": self.template,
                "frames": list(sorted(self.frames)),
            })

        return output

from_value(value, allow_sequences) classmethod

Convert passed value to FileDefItem objects.

Returns:

Name Type Description
list List[Self]

Created FileDefItem objects.

Source code in client/ayon_core/lib/attribute_definitions.py
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
@classmethod
def from_value(
    cls,
    value: "Union[List[FileDefItemDict], FileDefItemDict]",
    allow_sequences: bool,
) -> List["Self"]:
    """Convert passed value to FileDefItem objects.

    Returns:
        list: Created FileDefItem objects.

    """
    # Convert single item to iterable
    if not isinstance(value, (list, tuple, set)):
        value = [value]

    output = []
    str_filepaths = []
    for item in value:
        if isinstance(item, dict):
            item = cls.from_dict(item)

        if isinstance(item, FileDefItem):
            if not allow_sequences and item.is_sequence:
                output.extend(item.split_sequence())
            else:
                output.append(item)

        elif isinstance(item, str):
            str_filepaths.append(item)
        else:
            raise TypeError(
                "Unknown type \"{}\". Can't convert to {}".format(
                    str(type(item)), cls.__name__
                )
            )

    if str_filepaths:
        output.extend(cls.from_paths(str_filepaths, allow_sequences))

    return output

FormatObject

Object that can be used for formatting.

This is base that is valid for to be used in 'StringTemplate' value.

Source code in client/ayon_core/lib/path_templates.py
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
class FormatObject:
    """Object that can be used for formatting.

    This is base that is valid for to be used in 'StringTemplate' value.
    """
    def __init__(self):
        self.value = ""

    def __format__(self, *args, **kwargs):
        return self.value.__format__(*args, **kwargs)

    def __str__(self) -> str:
        return str(self.value)

    def __repr__(self) -> str:
        return self.__str__()

IniSettingRegistry

Bases: ASettingRegistry

Class using :mod:configparser.

This class is using :mod:configparser (ini) files to store items.

Source code in client/ayon_core/lib/local_settings.py
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
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
class IniSettingRegistry(ASettingRegistry):
    """Class using :mod:`configparser`.

    This class is using :mod:`configparser` (ini) files to store items.

    """

    def __init__(self, name, path):
        # type: (str, str) -> IniSettingRegistry
        super(IniSettingRegistry, self).__init__(name)
        # get registry file
        self._registry_file = os.path.join(path, "{}.ini".format(name))
        if not os.path.exists(self._registry_file):
            with open(self._registry_file, mode="w") as cfg:
                print("# Settings registry", cfg)
                now = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
                print("# {}".format(now), cfg)

    def set_item_section(self, section, name, value):
        # type: (str, str, str) -> None
        """Set item to specific section of ini registry.

        If section doesn't exists, it is created.

        Args:
            section (str): Name of section.
            name (str): Name of the item.
            value (str): Value of the item.

        """
        value = str(value)
        config = configparser.ConfigParser()

        config.read(self._registry_file)
        if not config.has_section(section):
            config.add_section(section)
        current = config[section]
        current[name] = value

        with open(self._registry_file, mode="w") as cfg:
            config.write(cfg)

    def _set_item(self, name, value):
        # type: (str, str) -> None
        self.set_item_section("MAIN", name, value)

    def set_item(self, name, value):
        # type: (str, str) -> None
        """Set item to settings ini file.

        This saves item to ``DEFAULT`` section of ini as each item there
        must reside in some section.

        Args:
            name (str): Name of the item.
            value (str): Value of the item.

        """
        # this does the some, overridden just for different docstring.
        # we cast value to str as ini options values must be strings.
        super(IniSettingRegistry, self).set_item(name, str(value))

    def get_item(self, name):
        # type: (str) -> str
        """Gets item from settings ini file.

        This gets settings from ``DEFAULT`` section of ini file as each item
        there must reside in some section.

        Args:
            name (str): Name of the item.

        Returns:
            str: Value of item.

        Raises:
            ValueError: If value doesn't exist.

        """
        return super(IniSettingRegistry, self).get_item(name)

    @lru_cache(maxsize=32)
    def get_item_from_section(self, section, name):
        # type: (str, str) -> str
        """Get item from section of ini file.

        This will read ini file and try to get item value from specified
        section. If that section or item doesn't exist, :exc:`ValueError`
        is risen.

        Args:
            section (str): Name of ini section.
            name (str): Name of the item.

        Returns:
            str: Item value.

        Raises:
            ValueError: If value doesn't exist.

        """
        config = configparser.ConfigParser()
        config.read(self._registry_file)
        try:
            value = config[section][name]
        except KeyError:
            raise ValueError(
                "Registry doesn't contain value {}:{}".format(section, name))
        return value

    def _get_item(self, name):
        # type: (str) -> str
        return self.get_item_from_section("MAIN", name)

    def delete_item_from_section(self, section, name):
        # type: (str, str) -> None
        """Delete item from section in ini file.

        Args:
            section (str): Section name.
            name (str): Name of the item.

        Raises:
            ValueError: If item doesn't exist.

        """
        self.get_item_from_section.cache_clear()
        config = configparser.ConfigParser()
        config.read(self._registry_file)
        try:
            _ = config[section][name]
        except KeyError:
            raise ValueError(
                "Registry doesn't contain value {}:{}".format(section, name))
        config.remove_option(section, name)

        # if section is empty, delete it
        if len(config[section].keys()) == 0:
            config.remove_section(section)

        with open(self._registry_file, mode="w") as cfg:
            config.write(cfg)

    def _delete_item(self, name):
        """Delete item from default section."""
        self.delete_item_from_section("MAIN", name)

delete_item_from_section(section, name)

Delete item from section in ini file.

Parameters:

Name Type Description Default
section str

Section name.

required
name str

Name of the item.

required

Raises:

Type Description
ValueError

If item doesn't exist.

Source code in client/ayon_core/lib/local_settings.py
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def delete_item_from_section(self, section, name):
    # type: (str, str) -> None
    """Delete item from section in ini file.

    Args:
        section (str): Section name.
        name (str): Name of the item.

    Raises:
        ValueError: If item doesn't exist.

    """
    self.get_item_from_section.cache_clear()
    config = configparser.ConfigParser()
    config.read(self._registry_file)
    try:
        _ = config[section][name]
    except KeyError:
        raise ValueError(
            "Registry doesn't contain value {}:{}".format(section, name))
    config.remove_option(section, name)

    # if section is empty, delete it
    if len(config[section].keys()) == 0:
        config.remove_section(section)

    with open(self._registry_file, mode="w") as cfg:
        config.write(cfg)

get_item(name)

Gets item from settings ini file.

This gets settings from DEFAULT section of ini file as each item there must reside in some section.

Parameters:

Name Type Description Default
name str

Name of the item.

required

Returns:

Name Type Description
str

Value of item.

Raises:

Type Description
ValueError

If value doesn't exist.

Source code in client/ayon_core/lib/local_settings.py
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
def get_item(self, name):
    # type: (str) -> str
    """Gets item from settings ini file.

    This gets settings from ``DEFAULT`` section of ini file as each item
    there must reside in some section.

    Args:
        name (str): Name of the item.

    Returns:
        str: Value of item.

    Raises:
        ValueError: If value doesn't exist.

    """
    return super(IniSettingRegistry, self).get_item(name)

get_item_from_section(section, name) cached

Get item from section of ini file.

This will read ini file and try to get item value from specified section. If that section or item doesn't exist, :exc:ValueError is risen.

Parameters:

Name Type Description Default
section str

Name of ini section.

required
name str

Name of the item.

required

Returns:

Name Type Description
str

Item value.

Raises:

Type Description
ValueError

If value doesn't exist.

Source code in client/ayon_core/lib/local_settings.py
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
@lru_cache(maxsize=32)
def get_item_from_section(self, section, name):
    # type: (str, str) -> str
    """Get item from section of ini file.

    This will read ini file and try to get item value from specified
    section. If that section or item doesn't exist, :exc:`ValueError`
    is risen.

    Args:
        section (str): Name of ini section.
        name (str): Name of the item.

    Returns:
        str: Item value.

    Raises:
        ValueError: If value doesn't exist.

    """
    config = configparser.ConfigParser()
    config.read(self._registry_file)
    try:
        value = config[section][name]
    except KeyError:
        raise ValueError(
            "Registry doesn't contain value {}:{}".format(section, name))
    return value

set_item(name, value)

Set item to settings ini file.

This saves item to DEFAULT section of ini as each item there must reside in some section.

Parameters:

Name Type Description Default
name str

Name of the item.

required
value str

Value of the item.

required
Source code in client/ayon_core/lib/local_settings.py
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def set_item(self, name, value):
    # type: (str, str) -> None
    """Set item to settings ini file.

    This saves item to ``DEFAULT`` section of ini as each item there
    must reside in some section.

    Args:
        name (str): Name of the item.
        value (str): Value of the item.

    """
    # this does the some, overridden just for different docstring.
    # we cast value to str as ini options values must be strings.
    super(IniSettingRegistry, self).set_item(name, str(value))

set_item_section(section, name, value)

Set item to specific section of ini registry.

If section doesn't exists, it is created.

Parameters:

Name Type Description Default
section str

Name of section.

required
name str

Name of the item.

required
value str

Value of the item.

required
Source code in client/ayon_core/lib/local_settings.py
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def set_item_section(self, section, name, value):
    # type: (str, str, str) -> None
    """Set item to specific section of ini registry.

    If section doesn't exists, it is created.

    Args:
        section (str): Name of section.
        name (str): Name of the item.
        value (str): Value of the item.

    """
    value = str(value)
    config = configparser.ConfigParser()

    config.read(self._registry_file)
    if not config.has_section(section):
        config.add_section(section)
    current = config[section]
    current[name] = value

    with open(self._registry_file, mode="w") as cfg:
        config.write(cfg)

JSONSettingRegistry

Bases: ASettingRegistry

Class using json file as storage.

Source code in client/ayon_core/lib/local_settings.py
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
class JSONSettingRegistry(ASettingRegistry):
    """Class using json file as storage."""

    def __init__(self, name, path):
        # type: (str, str) -> JSONSettingRegistry
        super(JSONSettingRegistry, self).__init__(name)
        #: str: name of registry file
        self._registry_file = os.path.join(path, "{}.json".format(name))
        now = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
        header = {
            "__metadata__": {"generated": now},
            "registry": {}
        }

        if not os.path.exists(os.path.dirname(self._registry_file)):
            os.makedirs(os.path.dirname(self._registry_file), exist_ok=True)
        if not os.path.exists(self._registry_file):
            with open(self._registry_file, mode="w") as cfg:
                json.dump(header, cfg, indent=4)

    @lru_cache(maxsize=32)
    def _get_item(self, name):
        # type: (str) -> object
        """Get item value from registry json.

        Note:
            See :meth:`ayon_core.lib.JSONSettingRegistry.get_item`

        """
        with open(self._registry_file, mode="r") as cfg:
            data = json.load(cfg)
            try:
                value = data["registry"][name]
            except KeyError:
                raise ValueError(
                    "Registry doesn't contain value {}".format(name))
        return value

    def get_item(self, name):
        # type: (str) -> object
        """Get item value from registry json.

        Args:
            name (str): Name of the item.

        Returns:
            value of the item

        Raises:
            ValueError: If item is not found in registry file.

        """
        return self._get_item(name)

    def _set_item(self, name, value):
        # type: (str, object) -> None
        """Set item value to registry json.

        Note:
            See :meth:`ayon_core.lib.JSONSettingRegistry.set_item`

        """
        with open(self._registry_file, "r+") as cfg:
            data = json.load(cfg)
            data["registry"][name] = value
            cfg.truncate(0)
            cfg.seek(0)
            json.dump(data, cfg, indent=4)

    def set_item(self, name, value):
        # type: (str, object) -> None
        """Set item and its value into json registry file.

        Args:
            name (str): name of the item.
            value (Any): value of the item.

        """
        self._set_item(name, value)

    def _delete_item(self, name):
        # type: (str) -> None
        self._get_item.cache_clear()
        with open(self._registry_file, "r+") as cfg:
            data = json.load(cfg)
            del data["registry"][name]
            cfg.truncate(0)
            cfg.seek(0)
            json.dump(data, cfg, indent=4)

get_item(name)

Get item value from registry json.

Parameters:

Name Type Description Default
name str

Name of the item.

required

Returns:

Type Description

value of the item

Raises:

Type Description
ValueError

If item is not found in registry file.

Source code in client/ayon_core/lib/local_settings.py
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
def get_item(self, name):
    # type: (str) -> object
    """Get item value from registry json.

    Args:
        name (str): Name of the item.

    Returns:
        value of the item

    Raises:
        ValueError: If item is not found in registry file.

    """
    return self._get_item(name)

set_item(name, value)

Set item and its value into json registry file.

Parameters:

Name Type Description Default
name str

name of the item.

required
value Any

value of the item.

required
Source code in client/ayon_core/lib/local_settings.py
528
529
530
531
532
533
534
535
536
537
def set_item(self, name, value):
    # type: (str, object) -> None
    """Set item and its value into json registry file.

    Args:
        name (str): name of the item.
        value (Any): value of the item.

    """
    self._set_item(name, value)

Logger

Source code in client/ayon_core/lib/log.py
 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
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
class Logger:
    DFT = '%(levelname)s >>> { %(name)s }: [ %(message)s ] '
    DBG = "  - { %(name)s }: [ %(message)s ] "
    INF = ">>> [ %(message)s ] "
    WRN = "*** WRN: >>> { %(name)s }: [ %(message)s ] "
    ERR = "!!! ERR: %(asctime)s >>> { %(name)s }: [ %(message)s ] "
    CRI = "!!! CRI: %(asctime)s >>> { %(name)s }: [ %(message)s ] "

    FORMAT_FILE = {
        logging.INFO: INF,
        logging.DEBUG: DBG,
        logging.WARNING: WRN,
        logging.ERROR: ERR,
        logging.CRITICAL: CRI,
    }

    # Is static class initialized
    initialized = False
    _init_lock = threading.Lock()

    # Logging level - AYON_LOG_LEVEL
    log_level = None

    # Data same for all record documents
    process_data = None
    # Cached process name or ability to set different process name
    _process_name = None

    @classmethod
    def get_logger(cls, name=None):
        if not cls.initialized:
            cls.initialize()

        logger = logging.getLogger(name or "__main__")

        logger.setLevel(cls.log_level)

        add_console_handler = True

        for handler in logger.handlers:
            if isinstance(handler, LogStreamHandler):
                add_console_handler = False

        if add_console_handler:
            logger.addHandler(cls._get_console_handler())

        # Do not propagate logs to root logger
        logger.propagate = False

        return logger

    @classmethod
    def _get_console_handler(cls):
        formatter = LogFormatter(cls.FORMAT_FILE)
        console_handler = LogStreamHandler()

        console_handler.set_name("LogStreamHandler")
        console_handler.setFormatter(formatter)
        return console_handler

    @classmethod
    def initialize(cls):
        # TODO update already created loggers on re-initialization
        if not cls._init_lock.locked():
            with cls._init_lock:
                cls._initialize()
        else:
            # If lock is locked wait until is finished
            while cls._init_lock.locked():
                time.sleep(0.1)

    @classmethod
    def _initialize(cls):
        # Change initialization state to prevent runtime changes
        # if is executed during runtime
        cls.initialized = False

        # Define what is logging level
        log_level = os.getenv("AYON_LOG_LEVEL")
        if not log_level:
            # Check AYON_DEBUG for debug level
            op_debug = os.getenv("AYON_DEBUG")
            if op_debug and int(op_debug) > 0:
                log_level = 10
            else:
                log_level = 20
        cls.log_level = int(log_level)

        # Mark as initialized
        cls.initialized = True

    @classmethod
    def get_process_data(cls):
        """Data about current process which should be same for all records.

        Process data are used for each record sent to mongo database.
        """
        if cls.process_data is not None:
            return copy.deepcopy(cls.process_data)

        if not cls.initialized:
            cls.initialize()

        host_name = socket.gethostname()
        try:
            host_ip = socket.gethostbyname(host_name)
        except socket.gaierror:
            host_ip = "127.0.0.1"

        process_name = cls.get_process_name()

        cls.process_data = {
            "hostname": host_name,
            "hostip": host_ip,
            "username": getpass.getuser(),
            "system_name": platform.system(),
            "process_name": process_name
        }
        return copy.deepcopy(cls.process_data)

    @classmethod
    def set_process_name(cls, process_name):
        """Set process name for mongo logs."""
        # Just change the attribute
        cls._process_name = process_name
        # Update process data if are already set
        if cls.process_data is not None:
            cls.process_data["process_name"] = process_name

    @classmethod
    def get_process_name(cls):
        """Process name that is like "label" of a process.

        AYON logging can be used from OpenPyppe itself of from hosts.
        Even in AYON process it's good to know if logs are from tray or
        from other cli commands. This should help to identify that information.
        """
        if cls._process_name is not None:
            return cls._process_name

        # Get process name
        process_name = os.environ.get("AYON_APP_NAME")
        if not process_name:
            try:
                import psutil
                process = psutil.Process(os.getpid())
                process_name = process.name()

            except ImportError:
                pass

        if not process_name:
            process_name = os.path.basename(sys.executable)

        cls._process_name = process_name
        return cls._process_name

get_process_data() classmethod

Data about current process which should be same for all records.

Process data are used for each record sent to mongo database.

Source code in client/ayon_core/lib/log.py
185
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
@classmethod
def get_process_data(cls):
    """Data about current process which should be same for all records.

    Process data are used for each record sent to mongo database.
    """
    if cls.process_data is not None:
        return copy.deepcopy(cls.process_data)

    if not cls.initialized:
        cls.initialize()

    host_name = socket.gethostname()
    try:
        host_ip = socket.gethostbyname(host_name)
    except socket.gaierror:
        host_ip = "127.0.0.1"

    process_name = cls.get_process_name()

    cls.process_data = {
        "hostname": host_name,
        "hostip": host_ip,
        "username": getpass.getuser(),
        "system_name": platform.system(),
        "process_name": process_name
    }
    return copy.deepcopy(cls.process_data)

get_process_name() classmethod

Process name that is like "label" of a process.

AYON logging can be used from OpenPyppe itself of from hosts. Even in AYON process it's good to know if logs are from tray or from other cli commands. This should help to identify that information.

Source code in client/ayon_core/lib/log.py
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
@classmethod
def get_process_name(cls):
    """Process name that is like "label" of a process.

    AYON logging can be used from OpenPyppe itself of from hosts.
    Even in AYON process it's good to know if logs are from tray or
    from other cli commands. This should help to identify that information.
    """
    if cls._process_name is not None:
        return cls._process_name

    # Get process name
    process_name = os.environ.get("AYON_APP_NAME")
    if not process_name:
        try:
            import psutil
            process = psutil.Process(os.getpid())
            process_name = process.name()

        except ImportError:
            pass

    if not process_name:
        process_name = os.path.basename(sys.executable)

    cls._process_name = process_name
    return cls._process_name

set_process_name(process_name) classmethod

Set process name for mongo logs.

Source code in client/ayon_core/lib/log.py
214
215
216
217
218
219
220
221
@classmethod
def set_process_name(cls, process_name):
    """Set process name for mongo logs."""
    # Just change the attribute
    cls._process_name = process_name
    # Update process data if are already set
    if cls.process_data is not None:
        cls.process_data["process_name"] = process_name

NestedCacheItem

Helper for cached items stored in nested structure.

Example

cache = NestedCacheItem(levels=2, default_factory=lambda: 0) cache["a"]["b"].is_valid False cache["a"]["b"].get_data() 0 cache["a"]["b"] = 1 cache["a"]["b"].is_valid True cache["a"]["b"].get_data() 1 cache.reset() cache["a"]["b"].is_valid False

Parameters:

Name Type Description Default
levels int

Number of nested levels where read cache is stored.

1
default_factory Optional[callable]

Function that returns default value used on init and on reset.

None
lifetime Optional[int]

Lifetime of the cache data in seconds. Default value is based on default value of 'CacheItem'.

None
_init_info Optional[InitInfo]

Private argument. Init info for nested cache where created from parent item.

None
Source code in client/ayon_core/lib/cache.py
 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
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
class NestedCacheItem:
    """Helper for cached items stored in nested structure.

    Example:
        >>> cache = NestedCacheItem(levels=2, default_factory=lambda: 0)
        >>> cache["a"]["b"].is_valid
        False
        >>> cache["a"]["b"].get_data()
        0
        >>> cache["a"]["b"] = 1
        >>> cache["a"]["b"].is_valid
        True
        >>> cache["a"]["b"].get_data()
        1
        >>> cache.reset()
        >>> cache["a"]["b"].is_valid
        False

    Args:
        levels (int): Number of nested levels where read cache is stored.
        default_factory (Optional[callable]): Function that returns default
            value used on init and on reset.
        lifetime (Optional[int]): Lifetime of the cache data in seconds.
            Default value is based on default value of 'CacheItem'.
        _init_info (Optional[InitInfo]): Private argument. Init info for
            nested cache where created from parent item.

    """
    def __init__(
        self, levels=1, default_factory=None, lifetime=None, _init_info=None
    ):
        if levels < 1:
            raise ValueError("Nested levels must be greater than 0")
        self._data_by_key = {}
        if _init_info is None:
            _init_info = InitInfo(default_factory, lifetime)
        self._init_info = _init_info
        self._levels = levels

    def __getitem__(self, key):
        """Get cached data.

        Args:
            key (str): Key of the cache item.

        Returns:
            Union[NestedCacheItem, CacheItem]: Cache item.

        """
        cache = self._data_by_key.get(key)
        if cache is None:
            if self._levels > 1:
                cache = NestedCacheItem(
                    levels=self._levels - 1,
                    _init_info=self._init_info
                )
            else:
                cache = CacheItem(
                    self._init_info.default_factory,
                    self._init_info.lifetime
                )
            self._data_by_key[key] = cache
        return cache

    def __setitem__(self, key, value):
        """Update cached data.

        Args:
            key (str): Key of the cache item.
            value (Any): Any data that are cached.

        """
        if self._levels > 1:
            raise AttributeError((
                "{} does not support '__setitem__'. Lower nested level by {}"
            ).format(self.__class__.__name__, self._levels - 1))
        cache = self[key]
        cache.update_data(value)

    def get(self, key):
        """Get cached data.

        Args:
            key (str): Key of the cache item.

        Returns:
            Union[NestedCacheItem, CacheItem]: Cache item.

        """
        return self[key]

    def cached_count(self):
        """Amount of cached items.

        Returns:
            int: Amount of cached items.

        """
        return len(self._data_by_key)

    def clear_key(self, key):
        """Clear cached item by key.

        Args:
            key (str): Key of the cache item.

        """
        self._data_by_key.pop(key, None)

    def clear_invalid(self):
        """Clear all invalid cache items.

        Note:
            To clear all cache items use 'reset'.

        """
        changed = {}
        children_are_nested = self._levels > 1
        for key, cache in tuple(self._data_by_key.items()):
            if children_are_nested:
                output = cache.clear_invalid()
                if output:
                    changed[key] = output
                if not cache.cached_count():
                    self._data_by_key.pop(key)
            elif not cache.is_valid:
                changed[key] = cache.get_data()
                self._data_by_key.pop(key)
        return changed

    def reset(self):
        """Reset cache.

        Note:
            To clear only invalid cache items use 'clear_invalid'.

        """
        self._data_by_key = {}

    def set_lifetime(self, lifetime):
        """Change lifetime of all children cache items.

        Args:
            lifetime (int): Lifetime of the cache data in seconds.

        """
        self._init_info.lifetime = lifetime
        for cache in self._data_by_key.values():
            cache.set_lifetime(lifetime)

    @property
    def is_valid(self):
        """Raise reasonable error when called on wrong level.

        Raises:
            AttributeError: If called on nested cache item.

        """
        raise AttributeError((
            "{} does not support 'is_valid'. Lower nested level by '{}'"
        ).format(self.__class__.__name__, self._levels))

is_valid property

Raise reasonable error when called on wrong level.

Raises:

Type Description
AttributeError

If called on nested cache item.

__getitem__(key)

Get cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required

Returns:

Type Description

Union[NestedCacheItem, CacheItem]: Cache item.

Source code in client/ayon_core/lib/cache.py
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def __getitem__(self, key):
    """Get cached data.

    Args:
        key (str): Key of the cache item.

    Returns:
        Union[NestedCacheItem, CacheItem]: Cache item.

    """
    cache = self._data_by_key.get(key)
    if cache is None:
        if self._levels > 1:
            cache = NestedCacheItem(
                levels=self._levels - 1,
                _init_info=self._init_info
            )
        else:
            cache = CacheItem(
                self._init_info.default_factory,
                self._init_info.lifetime
            )
        self._data_by_key[key] = cache
    return cache

__setitem__(key, value)

Update cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required
value Any

Any data that are cached.

required
Source code in client/ayon_core/lib/cache.py
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def __setitem__(self, key, value):
    """Update cached data.

    Args:
        key (str): Key of the cache item.
        value (Any): Any data that are cached.

    """
    if self._levels > 1:
        raise AttributeError((
            "{} does not support '__setitem__'. Lower nested level by {}"
        ).format(self.__class__.__name__, self._levels - 1))
    cache = self[key]
    cache.update_data(value)

cached_count()

Amount of cached items.

Returns:

Name Type Description
int

Amount of cached items.

Source code in client/ayon_core/lib/cache.py
181
182
183
184
185
186
187
188
def cached_count(self):
    """Amount of cached items.

    Returns:
        int: Amount of cached items.

    """
    return len(self._data_by_key)

clear_invalid()

Clear all invalid cache items.

Note

To clear all cache items use 'reset'.

Source code in client/ayon_core/lib/cache.py
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def clear_invalid(self):
    """Clear all invalid cache items.

    Note:
        To clear all cache items use 'reset'.

    """
    changed = {}
    children_are_nested = self._levels > 1
    for key, cache in tuple(self._data_by_key.items()):
        if children_are_nested:
            output = cache.clear_invalid()
            if output:
                changed[key] = output
            if not cache.cached_count():
                self._data_by_key.pop(key)
        elif not cache.is_valid:
            changed[key] = cache.get_data()
            self._data_by_key.pop(key)
    return changed

clear_key(key)

Clear cached item by key.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required
Source code in client/ayon_core/lib/cache.py
190
191
192
193
194
195
196
197
def clear_key(self, key):
    """Clear cached item by key.

    Args:
        key (str): Key of the cache item.

    """
    self._data_by_key.pop(key, None)

get(key)

Get cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required

Returns:

Type Description

Union[NestedCacheItem, CacheItem]: Cache item.

Source code in client/ayon_core/lib/cache.py
169
170
171
172
173
174
175
176
177
178
179
def get(self, key):
    """Get cached data.

    Args:
        key (str): Key of the cache item.

    Returns:
        Union[NestedCacheItem, CacheItem]: Cache item.

    """
    return self[key]

reset()

Reset cache.

Note

To clear only invalid cache items use 'clear_invalid'.

Source code in client/ayon_core/lib/cache.py
220
221
222
223
224
225
226
227
def reset(self):
    """Reset cache.

    Note:
        To clear only invalid cache items use 'clear_invalid'.

    """
    self._data_by_key = {}

set_lifetime(lifetime)

Change lifetime of all children cache items.

Parameters:

Name Type Description Default
lifetime int

Lifetime of the cache data in seconds.

required
Source code in client/ayon_core/lib/cache.py
229
230
231
232
233
234
235
236
237
238
def set_lifetime(self, lifetime):
    """Change lifetime of all children cache items.

    Args:
        lifetime (int): Lifetime of the cache data in seconds.

    """
    self._init_info.lifetime = lifetime
    for cache in self._data_by_key.values():
        cache.set_lifetime(lifetime)

NumberDef

Bases: AbstractAttrDef

Number definition.

Number can have defined minimum/maximum value and decimal points. Value is integer if decimals are 0.

Parameters:

Name Type Description Default
minimum(int, float

Minimum possible value.

required
maximum(int, float

Maximum possible value.

required
decimals(int)

Maximum decimal points of value.

required
default(int, float

Default value for conversion.

required
Source code in client/ayon_core/lib/attribute_definitions.py
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
class NumberDef(AbstractAttrDef):
    """Number definition.

    Number can have defined minimum/maximum value and decimal points. Value
    is integer if decimals are 0.

    Args:
        minimum(int, float): Minimum possible value.
        maximum(int, float): Maximum possible value.
        decimals(int): Maximum decimal points of value.
        default(int, float): Default value for conversion.

    """
    type = "number"
    type_attributes = [
        "minimum",
        "maximum",
        "decimals"
    ]

    def __init__(
        self,
        key: str,
        minimum: Optional[IntFloatType] = None,
        maximum: Optional[IntFloatType] = None,
        decimals: Optional[int] = None,
        default: Optional[IntFloatType] = None,
        **kwargs
    ):
        minimum = 0 if minimum is None else minimum
        maximum = 999999 if maximum is None else maximum
        # Swap min/max when are passed in opposite order
        if minimum > maximum:
            maximum, minimum = minimum, maximum

        if default is None:
            default = 0

        elif not isinstance(default, (int, float)):
            raise TypeError((
                "'default' argument must be 'int' or 'float', not '{}'"
            ).format(type(default)))

        # Fix default value by mim/max values
        if default < minimum:
            default = minimum

        elif default > maximum:
            default = maximum

        super().__init__(key, default=default, **kwargs)

        self.minimum: IntFloatType = minimum
        self.maximum: IntFloatType = maximum
        self.decimals: int = 0 if decimals is None else decimals

    def is_value_valid(self, value: Any) -> bool:
        if self.decimals == 0:
            if not isinstance(value, int):
                return False
        elif not isinstance(value, float):
            return False
        if self.minimum > value > self.maximum:
            return False
        return True

    def convert_value(self, value: Any) -> IntFloatType:
        if isinstance(value, str):
            try:
                value = float(value)
            except Exception:
                pass

        if not isinstance(value, (int, float)):
            return self.default

        if self.decimals == 0:
            return int(value)
        return round(float(value), self.decimals)

    def _def_type_compare(self, other: "NumberDef") -> bool:
        return (
            self.decimals == other.decimals
            and self.maximum == other.maximum
            and self.maximum == other.maximum
        )

StringTemplate

String that can be formatted.

Source code in client/ayon_core/lib/path_templates.py
 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
208
209
210
211
212
213
class StringTemplate:
    """String that can be formatted."""
    def __init__(self, template: str):
        if not isinstance(template, str):
            raise TypeError(
                f"<{self.__class__.__name__}> argument must be a string,"
                f" not {str(type(template))}."
            )

        self._template: str = template
        parts = []
        formatter = Formatter()

        for item in formatter.parse(template):
            literal_text, field_name, format_spec, conversion = item
            if literal_text:
                parts.append(literal_text)
            if field_name:
                parts.append(
                    FormattingPart(field_name, format_spec, conversion)
                )

        new_parts = []
        for part in parts:
            if not isinstance(part, str):
                new_parts.append(part)
                continue

            substr = ""
            for char in part:
                if char not in ("<", ">"):
                    substr += char
                else:
                    if substr:
                        new_parts.append(substr)
                    new_parts.append(char)
                    substr = ""
            if substr:
                new_parts.append(substr)

        self._parts: List["Union[str, OptionalPart, FormattingPart]"] = (
            self.find_optional_parts(new_parts)
        )

    def __str__(self) -> str:
        return self.template

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}> {self.template}"

    def __contains__(self, other: str) -> bool:
        return other in self.template

    def replace(self, *args, **kwargs):
        self._template = self.template.replace(*args, **kwargs)
        return self

    @property
    def template(self) -> str:
        return self._template

    def format(self, data: Dict[str, Any]) -> "TemplateResult":
        """ Figure out with whole formatting.

        Separate advanced keys (*Like '{project[name]}') from string which must
        be formatted separately in case of missing or incomplete keys in data.

        Args:
            data (dict): Containing keys to be filled into template.

        Returns:
            TemplateResult: Filled or partially filled template containing all
                data needed or missing for filling template.

        """
        result = TemplatePartResult()
        for part in self._parts:
            if isinstance(part, str):
                result.add_output(part)
            else:
                part.format(data, result)

        invalid_types = result.invalid_types
        invalid_types.update(result.invalid_optional_types)
        invalid_types = result.split_keys_to_subdicts(invalid_types)

        missing_keys = result.missing_keys
        missing_keys |= result.missing_optional_keys

        solved = result.solved
        used_values = result.get_clean_used_values()

        return TemplateResult(
            result.output,
            self.template,
            solved,
            used_values,
            missing_keys,
            invalid_types
        )

    def format_strict(self, data: Dict[str, Any]) -> "TemplateResult":
        result = self.format(data)
        result.validate()
        return result

    @classmethod
    def format_template(
        cls, template: str, data: Dict[str, Any]
    ) -> "TemplateResult":
        objected_template = cls(template)
        return objected_template.format(data)

    @classmethod
    def format_strict_template(
        cls, template: str, data: Dict[str, Any]
    ) -> "TemplateResult":
        objected_template = cls(template)
        return objected_template.format_strict(data)

    @staticmethod
    def find_optional_parts(
        parts: List["Union[str, FormattingPart]"]
    ) -> List["Union[str, OptionalPart, FormattingPart]"]:
        new_parts = []
        tmp_parts = {}
        counted_symb = -1
        for part in parts:
            if part == "<":
                counted_symb += 1
                tmp_parts[counted_symb] = []

            elif part == ">":
                if counted_symb > -1:
                    parts = tmp_parts.pop(counted_symb)
                    counted_symb -= 1
                    # If part contains only single string keep value
                    #   unchanged
                    if parts:
                        # Remove optional start char
                        parts.pop(0)

                    if not parts:
                        value = "<>"
                    elif (
                        len(parts) == 1
                        and isinstance(parts[0], str)
                    ):
                        value = "<{}>".format(parts[0])
                    else:
                        value = OptionalPart(parts)

                    if counted_symb < 0:
                        out_parts = new_parts
                    else:
                        out_parts = tmp_parts[counted_symb]
                    # Store value
                    out_parts.append(value)
                    continue

            if counted_symb < 0:
                new_parts.append(part)
            else:
                tmp_parts[counted_symb].append(part)

        if tmp_parts:
            for idx in sorted(tmp_parts.keys()):
                new_parts.extend(tmp_parts[idx])
        return new_parts

format(data)

Figure out with whole formatting.

Separate advanced keys (*Like '{project[name]}') from string which must be formatted separately in case of missing or incomplete keys in data.

Parameters:

Name Type Description Default
data dict

Containing keys to be filled into template.

required

Returns:

Name Type Description
TemplateResult TemplateResult

Filled or partially filled template containing all data needed or missing for filling template.

Source code in client/ayon_core/lib/path_templates.py
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
def format(self, data: Dict[str, Any]) -> "TemplateResult":
    """ Figure out with whole formatting.

    Separate advanced keys (*Like '{project[name]}') from string which must
    be formatted separately in case of missing or incomplete keys in data.

    Args:
        data (dict): Containing keys to be filled into template.

    Returns:
        TemplateResult: Filled or partially filled template containing all
            data needed or missing for filling template.

    """
    result = TemplatePartResult()
    for part in self._parts:
        if isinstance(part, str):
            result.add_output(part)
        else:
            part.format(data, result)

    invalid_types = result.invalid_types
    invalid_types.update(result.invalid_optional_types)
    invalid_types = result.split_keys_to_subdicts(invalid_types)

    missing_keys = result.missing_keys
    missing_keys |= result.missing_optional_keys

    solved = result.solved
    used_values = result.get_clean_used_values()

    return TemplateResult(
        result.output,
        self.template,
        solved,
        used_values,
        missing_keys,
        invalid_types
    )

TemplateUnsolved

Bases: Exception

Exception for unsolved template when strict is set to True.

Source code in client/ayon_core/lib/path_templates.py
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
class TemplateUnsolved(Exception):
    """Exception for unsolved template when strict is set to True."""

    msg = "Template \"{0}\" is unsolved.{1}{2}"
    invalid_types_msg = " Keys with invalid data type: `{0}`."
    missing_keys_msg = " Missing keys: \"{0}\"."

    def __init__(self, template, missing_keys, invalid_types):
        invalid_type_items = []
        for _key, _type in invalid_types.items():
            invalid_type_items.append(f"\"{_key}\" {str(_type)}")

        invalid_types_msg = ""
        if invalid_type_items:
            invalid_types_msg = self.invalid_types_msg.format(
                ", ".join(invalid_type_items)
            )

        missing_keys_msg = ""
        if missing_keys:
            missing_keys_msg = self.missing_keys_msg.format(
                ", ".join(missing_keys)
            )
        super().__init__(
            self.msg.format(template, missing_keys_msg, invalid_types_msg)
        )

TextDef

Bases: AbstractAttrDef

Text definition.

Text can have multiline option so end-line characters are allowed regex validation can be applied placeholder for UI purposes and default value.

Regex validation is not part of attribute implementation.

Parameters:

Name Type Description Default
multiline(bool)

Text has single or multiline support.

required
regex(str, Pattern

Regex validation.

required
placeholder(str)

UI placeholder for attribute.

required
default(str, None

Default value. Empty string used when not defined.

required
Source code in client/ayon_core/lib/attribute_definitions.py
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
class TextDef(AbstractAttrDef):
    """Text definition.

    Text can have multiline option so end-line characters are allowed regex
    validation can be applied placeholder for UI purposes and default value.

    Regex validation is not part of attribute implementation.

    Args:
        multiline(bool): Text has single or multiline support.
        regex(str, re.Pattern): Regex validation.
        placeholder(str): UI placeholder for attribute.
        default(str, None): Default value. Empty string used when not defined.

    """
    type = "text"
    type_attributes = [
        "multiline",
        "placeholder",
    ]

    def __init__(
        self,
        key: str,
        multiline: Optional[bool] = None,
        regex: Optional[str] = None,
        placeholder: Optional[str] = None,
        default: Optional[str] = None,
        **kwargs
    ):
        if default is None:
            default = ""

        super().__init__(key, default=default, **kwargs)

        if multiline is None:
            multiline = False

        elif not isinstance(default, str):
            raise TypeError((
                f"'default' argument must be a str, not '{type(default)}'"
            ))

        if isinstance(regex, str):
            regex = re.compile(regex)

        self.multiline: bool = multiline
        self.placeholder: Optional[str] = placeholder
        self.regex: Optional["Pattern"] = regex

    def is_value_valid(self, value: Any) -> bool:
        if not isinstance(value, str):
            return False
        if self.regex and not self.regex.match(value):
            return False
        return True

    def convert_value(self, value: Any) -> str:
        if isinstance(value, str):
            return value
        return self.default

    def serialize(self) -> Dict[str, Any]:
        data = super().serialize()
        regex = None
        if self.regex is not None:
            regex = self.regex.pattern
        data["regex"] = regex
        data["multiline"] = self.multiline
        data["placeholder"] = self.placeholder
        return data

    def _def_type_compare(self, other: "TextDef") -> bool:
        return (
            self.multiline == other.multiline
            and self.regex == other.regex
        )

ToolNotFoundError

Bases: Exception

Raised when tool arguments are not found.

Source code in client/ayon_core/lib/vendor_bin_utils.py
 9
10
class ToolNotFoundError(Exception):
    """Raised when tool arguments are not found."""

UnknownDef

Bases: AbstractAttrDef

Definition is not known because definition is not available.

This attribute can be used to keep existing data unchanged but does not have known definition of type.

Source code in client/ayon_core/lib/attribute_definitions.py
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class UnknownDef(AbstractAttrDef):
    """Definition is not known because definition is not available.

    This attribute can be used to keep existing data unchanged but does not
    have known definition of type.

    """
    type = "unknown"

    def __init__(self, key: str, default: Optional[Any] = None, **kwargs):
        kwargs["default"] = default
        super().__init__(key, **kwargs)

    def is_value_valid(self, value: Any) -> bool:
        return True

    def convert_value(self, value: Any) -> Any:
        return value

classes_from_module(superclass, module)

Return plug-ins from module

Parameters:

Name Type Description Default
superclass superclass

Superclass of subclasses to look for

required
module ModuleType

Imported module where to look for 'superclass' subclasses.

required

Returns:

Type Description

List of plug-ins, or empty list if none is found.

Source code in client/ayon_core/lib/python_module_tools.py
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def classes_from_module(superclass, module):
    """Return plug-ins from module

    Arguments:
        superclass (superclass): Superclass of subclasses to look for
        module (types.ModuleType): Imported module where to look for
            'superclass' subclasses.

    Returns:
        List of plug-ins, or empty list if none is found.

    """

    classes = list()
    for name in dir(module):
        # It could be anything at this point
        obj = getattr(module, name)
        if not inspect.isclass(obj) or obj is superclass:
            continue

        if issubclass(obj, superclass):
            classes.append(obj)

    return classes

collect_frames(files)

Returns dict of source path and its frame, if from sequence

Uses clique as most precise solution, used when anatomy template that created files is not known.

Assumption is that frames are separated by '.', negative frames are not allowed.

Parameters:

Name Type Description Default
files(list) or (set with single value

list of source paths

required

Returns:

Name Type Description
dict

{'/folder/product_v001.0001.png': '0001', ....}

Source code in client/ayon_core/lib/path_tools.py
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
def collect_frames(files):
    """Returns dict of source path and its frame, if from sequence

    Uses clique as most precise solution, used when anatomy template that
    created files is not known.

    Assumption is that frames are separated by '.', negative frames are not
    allowed.

    Args:
        files(list) or (set with single value): list of source paths

    Returns:
        dict: {'/folder/product_v001.0001.png': '0001', ....}
    """

    # clique.PATTERNS["frames"] supports only `.1001.exr` not `_1001.exr` so
    # we use a customized pattern.
    pattern = "[_.](?P<index>(?P<padding>0*)\\d+)\\.\\D+\\d?$"
    patterns = [pattern]
    collections, remainder = clique.assemble(
        files, minimum_items=1, patterns=patterns)

    sources_and_frames = {}
    if collections:
        for collection in collections:
            src_head = collection.head
            src_tail = collection.tail

            for index in collection.indexes:
                src_frame = collection.format("{padding}") % index
                src_file_name = "{}{}{}".format(
                    src_head, src_frame, src_tail)
                sources_and_frames[src_file_name] = src_frame
    else:
        sources_and_frames[remainder.pop()] = None

    return sources_and_frames

compile_list_of_regexes(in_list)

Convert strings in entered list to compiled regex objects.

Source code in client/ayon_core/lib/profiles_filtering.py
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def compile_list_of_regexes(in_list):
    """Convert strings in entered list to compiled regex objects."""
    regexes = list()
    if not in_list:
        return regexes

    for item in in_list:
        if not item:
            continue
        try:
            regexes.append(re.compile(item))
        except TypeError:
            print((
                "Invalid type \"{}\" value \"{}\"."
                " Expected string based object. Skipping."
            ).format(str(type(item)), str(item)))
    return regexes

convert_ffprobe_fps_to_float(value)

Convert string value of frame rate to float.

Copy of 'convert_ffprobe_fps_value' which raises exceptions on invalid value, does not convert value to string and does not return "Unknown" string.

Parameters:

Name Type Description Default
value str

Value to be converted.

required

Returns:

Name Type Description
Float

Converted frame rate in float. If divisor in value is '0' then '0.0' is returned.

Raises:

Type Description
ValueError

Passed value is invalid for conversion.

Source code in client/ayon_core/lib/transcoding.py
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
def convert_ffprobe_fps_to_float(value):
    """Convert string value of frame rate to float.

    Copy of 'convert_ffprobe_fps_value' which raises exceptions on invalid
    value, does not convert value to string and does not return "Unknown"
    string.

    Args:
        value (str): Value to be converted.

    Returns:
        Float: Converted frame rate in float. If divisor in value is '0' then
            '0.0' is returned.

    Raises:
        ValueError: Passed value is invalid for conversion.
    """

    if not value:
        raise ValueError("Got empty value.")

    items = value.split("/")
    if len(items) == 1:
        return float(items[0])

    if len(items) > 2:
        raise ValueError((
            "FPS expression contains multiple dividers \"{}\"."
        ).format(value))

    dividend = float(items.pop(0))
    divisor = float(items.pop(0))
    if divisor == 0.0:
        return 0.0
    return dividend / divisor

convert_ffprobe_fps_value(str_value)

Returns (str) value of fps from ffprobe frame format (120/1)

Source code in client/ayon_core/lib/transcoding.py
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
def convert_ffprobe_fps_value(str_value):
    """Returns (str) value of fps from ffprobe frame format (120/1)"""
    if str_value == "0/0":
        print("WARNING: Source has \"r_frame_rate\" value set to \"0/0\".")
        return "Unknown"

    items = str_value.split("/")
    if len(items) == 1:
        fps = float(items[0])

    elif len(items) == 2:
        fps = float(items[0]) / float(items[1])

    # Check if fps is integer or float number
    if int(fps) == fps:
        fps = int(fps)

    return str(fps)

convert_for_ffmpeg(first_input_path, output_dir, input_frame_start=None, input_frame_end=None, logger=None)

Convert source file to format supported in ffmpeg.

Currently can convert only exrs.

Parameters:

Name Type Description Default
first_input_path str

Path to first file of a sequence or a single file path for non-sequential input.

required
output_dir str

Path to directory where output will be rendered. Must not be same as input's directory.

required
input_frame_start int

Frame start of input.

None
input_frame_end int

Frame end of input.

None
logger Logger

Logger used for logging.

None

Raises:

Type Description
ValueError

If input filepath has extension not supported by function. Currently is supported only ".exr" extension.

Source code in client/ayon_core/lib/transcoding.py
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
def convert_for_ffmpeg(
    first_input_path,
    output_dir,
    input_frame_start=None,
    input_frame_end=None,
    logger=None
):
    """Convert source file to format supported in ffmpeg.

    Currently can convert only exrs.

    Args:
        first_input_path (str): Path to first file of a sequence or a single
            file path for non-sequential input.
        output_dir (str): Path to directory where output will be rendered.
            Must not be same as input's directory.
        input_frame_start (int): Frame start of input.
        input_frame_end (int): Frame end of input.
        logger (logging.Logger): Logger used for logging.

    Raises:
        ValueError: If input filepath has extension not supported by function.
            Currently is supported only ".exr" extension.
    """
    if logger is None:
        logger = logging.getLogger(__name__)

    logger.warning((
        "DEPRECATED: 'ayon_core.lib.transcoding.convert_for_ffmpeg' is"
        " deprecated function of conversion for FFMpeg. Please replace usage"
        " with 'ayon_core.lib.transcoding.convert_input_paths_for_ffmpeg'"
    ))

    ext = os.path.splitext(first_input_path)[1].lower()
    if ext != ".exr":
        raise ValueError((
            "Function 'convert_for_ffmpeg' currently support only"
            " \".exr\" extension. Got \"{}\"."
        ).format(ext))

    is_sequence = False
    if input_frame_start is not None and input_frame_end is not None:
        is_sequence = int(input_frame_end) != int(input_frame_start)

    input_info = get_oiio_info_for_input(first_input_path, logger=logger)

    # Change compression only if source compression is "dwaa" or "dwab"
    #   - they're not supported in ffmpeg
    compression = input_info["attribs"].get("compression")
    if compression in ("dwaa", "dwab"):
        compression = "none"

    # Prepare subprocess arguments
    oiio_cmd = get_oiio_tool_args(
        "oiiotool",
        # Don't add any additional attributes
        "--nosoftwareattrib",
    )
    # Add input compression if available
    if compression:
        oiio_cmd.extend(["--compression", compression])

    # Collect channels to export
    input_arg, channels_arg = get_oiio_input_and_channel_args(input_info)

    oiio_cmd.extend([
        input_arg, first_input_path,
        # Tell oiiotool which channels should be put to top stack (and output)
        "--ch", channels_arg,
        # Use first subimage
        "--subimage", "0"
    ])

    # Add frame definitions to arguments
    if is_sequence:
        oiio_cmd.extend([
            "--frames", "{}-{}".format(input_frame_start, input_frame_end)
        ])

    for attr_name, attr_value in input_info["attribs"].items():
        if not isinstance(attr_value, str):
            continue

        # Remove attributes that have string value longer than allowed length
        #   for ffmpeg or when contain prohibited symbols
        erase_reason = "Missing reason"
        erase_attribute = False
        if len(attr_value) > MAX_FFMPEG_STRING_LEN:
            erase_reason = "has too long value ({} chars).".format(
                len(attr_value)
            )
            erase_attribute = True

        if not erase_attribute:
            for char in NOT_ALLOWED_FFMPEG_CHARS:
                if char in attr_value:
                    erase_attribute = True
                    erase_reason = (
                        "contains unsupported character \"{}\"."
                    ).format(char)
                    break

        if erase_attribute:
            # Set attribute to empty string
            logger.info((
                "Removed attribute \"{}\" from metadata because {}."
            ).format(attr_name, erase_reason))
            oiio_cmd.extend(["--eraseattrib", attr_name])

    # Add last argument - path to output
    if is_sequence:
        ext = os.path.splitext(first_input_path)[1]
        base_filename = "tmp.%{:0>2}d{}".format(
            len(str(input_frame_end)), ext
        )
    else:
        base_filename = os.path.basename(first_input_path)
    output_path = os.path.join(output_dir, base_filename)
    oiio_cmd.extend([
        "-o", output_path
    ])

    logger.debug("Conversion command: {}".format(" ".join(oiio_cmd)))
    run_subprocess(oiio_cmd, logger=logger)

convert_input_paths_for_ffmpeg(input_paths, output_dir, logger=None)

Convert source file to format supported in ffmpeg.

Currently can convert only exrs. The input filepaths should be files with same type. Information about input is loaded only from first found file.

Filenames of input files are kept so make sure that output directory is not the same directory as input files have. - This way it can handle gaps and can keep input filenames without handling frame template

Parameters:

Name Type Description Default
input_paths str

Paths that should be converted. It is expected that contains single file or image sequence of same type.

required
output_dir str

Path to directory where output will be rendered. Must not be same as input's directory.

required
logger Logger

Logger used for logging.

None

Raises:

Type Description
ValueError

If input filepath has extension not supported by function. Currently is supported only ".exr" extension.

Source code in client/ayon_core/lib/transcoding.py
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
def convert_input_paths_for_ffmpeg(
    input_paths,
    output_dir,
    logger=None
):
    """Convert source file to format supported in ffmpeg.

    Currently can convert only exrs. The input filepaths should be files
    with same type. Information about input is loaded only from first found
    file.

    Filenames of input files are kept so make sure that output directory
    is not the same directory as input files have.
    - This way it can handle gaps and can keep input filenames without handling
        frame template

    Args:
        input_paths (str): Paths that should be converted. It is expected that
            contains single file or image sequence of same type.
        output_dir (str): Path to directory where output will be rendered.
            Must not be same as input's directory.
        logger (logging.Logger): Logger used for logging.

    Raises:
        ValueError: If input filepath has extension not supported by function.
            Currently is supported only ".exr" extension.
    """
    if logger is None:
        logger = logging.getLogger(__name__)

    first_input_path = input_paths[0]
    ext = os.path.splitext(first_input_path)[1].lower()

    if ext != ".exr":
        raise ValueError((
            "Function 'convert_for_ffmpeg' currently support only"
            " \".exr\" extension. Got \"{}\"."
        ).format(ext))

    input_info = get_oiio_info_for_input(first_input_path, logger=logger)

    # Change compression only if source compression is "dwaa" or "dwab"
    #   - they're not supported in ffmpeg
    compression = input_info["attribs"].get("compression")
    if compression in ("dwaa", "dwab"):
        compression = "none"

    # Collect channels to export
    input_arg, channels_arg = get_oiio_input_and_channel_args(input_info)

    for input_path in input_paths:
        # Prepare subprocess arguments
        oiio_cmd = get_oiio_tool_args(
            "oiiotool",
            # Don't add any additional attributes
            "--nosoftwareattrib",
        )
        # Add input compression if available
        if compression:
            oiio_cmd.extend(["--compression", compression])

        oiio_cmd.extend([
            input_arg, input_path,
            # Tell oiiotool which channels should be put to top stack
            #   (and output)
            "--ch", channels_arg,
            # Use first subimage
            "--subimage", "0"
        ])

        for attr_name, attr_value in input_info["attribs"].items():
            if not isinstance(attr_value, str):
                continue

            # Remove attributes that have string value longer than allowed
            #   length for ffmpeg or when containing prohibited symbols
            erase_reason = "Missing reason"
            erase_attribute = False
            if len(attr_value) > MAX_FFMPEG_STRING_LEN:
                erase_reason = "has too long value ({} chars).".format(
                    len(attr_value)
                )
                erase_attribute = True

            if not erase_attribute:
                for char in NOT_ALLOWED_FFMPEG_CHARS:
                    if char in attr_value:
                        erase_attribute = True
                        erase_reason = (
                            "contains unsupported character \"{}\"."
                        ).format(char)
                        break

            if erase_attribute:
                # Set attribute to empty string
                logger.info((
                    "Removed attribute \"{}\" from metadata because {}."
                ).format(attr_name, erase_reason))
                oiio_cmd.extend(["--eraseattrib", attr_name])

        # Add last argument - path to output
        base_filename = os.path.basename(input_path)
        output_path = os.path.join(output_dir, base_filename)
        oiio_cmd.extend([
            "-o", output_path
        ])

        logger.debug("Conversion command: {}".format(" ".join(oiio_cmd)))
        run_subprocess(oiio_cmd, logger=logger)

Create hardlink of file.

Parameters:

Name Type Description Default
src_path(str)

Full path to a file which is used as source for hardlink.

required
dst_path(str)

Full path to a file where a link of source will be added.

required
Source code in client/ayon_core/lib/path_tools.py
31
32
33
34
35
36
37
38
39
40
def create_hard_link(src_path, dst_path):
    """Create hardlink of file.

    Args:
        src_path(str): Full path to a file which is used as source for
            hardlink.
        dst_path(str): Full path to a file where a link of source will be
            added.
    """
    os.link(src_path, dst_path)

emit_event(topic, data=None, source=None)

Emit event with topic and data.

Arg

topic(str): Event's topic. data(dict): Event's additional data. Optional. source(str): Who emitted the topic. Optional.

Returns:

Name Type Description
Event

Object of event that was emitted.

Source code in client/ayon_core/lib/events.py
709
710
711
712
713
714
715
716
717
718
719
720
721
def emit_event(topic, data=None, source=None):
    """Emit event with topic and data.

    Arg:
        topic(str): Event's topic.
        data(dict): Event's additional data. Optional.
        source(str): Who emitted the topic. Optional.

    Returns:
        Event: Object of event that was emitted.
    """

    return GlobalEventSystem.emit(topic, data, source)

env_value_to_bool(env_key=None, value=None, default=False)

Convert environment variable value to boolean.

Function is based on value of the environemt variable. Value is lowered so function is not case sensitive.

Returns:

Name Type Description
bool bool

If value match to one of ["true", "yes", "1"] result if True but if value match to ["false", "no", "0"] result is False else default value is returned.

Source code in client/ayon_core/lib/env_tools.py
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
def env_value_to_bool(
    env_key: Optional[str] = None,
    value: Optional[str] = None,
    default: bool = False,
) -> bool:
    """Convert environment variable value to boolean.

    Function is based on value of the environemt variable. Value is lowered
    so function is not case sensitive.

    Returns:
        bool: If value match to one of ["true", "yes", "1"] result if True
            but if value match to ["false", "no", "0"] result is False else
            default value is returned.

    """
    if value is None and env_key is None:
        return default

    if value is None:
        value = os.environ.get(env_key)

    if value is not None:
        value = str(value).lower()
        if value in ("true", "yes", "1", "on"):
            return True
        elif value in ("false", "no", "0", "off"):
            return False
    return default

filter_profiles(profiles_data, key_values, keys_order=None, logger=None)

Filter profiles by entered key -> values.

Profile if marked with score for each key/value from key_values with points -1, 0 or 1. - if profile contain the key and profile's value contain value from key_values then profile gets 1 point - if profile does not contain the key or profile's value is empty or contain "" then got 0 point - if profile contain the key, profile's value is not empty and does not contain "" and value from key_values is not available in the value then got -1 point

If profile gets -1 point at any time then is skipped and not used for output. Profile with higher score is returned. If there are multiple profiles with same score then first in order is used (order of profiles matter).

Parameters:

Name Type Description Default
profiles_data list

Profile definitions as dictionaries.

required
key_values dict

Mapping of Key <-> Value. Key is checked if is available in profile and if Value is matching it's values.

required
keys_order (list, tuple)

Order of keys from key_values which matters only when multiple profiles have same score.

None
logger Logger

Optionally can be passed different logger.

None

Returns:

Type Description

dict/None: Return most matching profile or None if none of profiles match at least one criteria.

Source code in client/ayon_core/lib/profiles_filtering.py
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def filter_profiles(profiles_data, key_values, keys_order=None, logger=None):
    """ Filter profiles by entered key -> values.

    Profile if marked with score for each key/value from `key_values` with
    points -1, 0 or 1.
    - if profile contain the key and profile's value contain value from
        `key_values` then profile gets 1 point
    - if profile does not contain the key or profile's value is empty or
        contain "*" then got 0 point
    - if profile contain the key, profile's value is not empty and does not
        contain "*" and value from `key_values` is not available in the value
        then got -1 point

    If profile gets -1 point at any time then is skipped and not used for
    output. Profile with higher score is returned. If there are multiple
    profiles with same score then first in order is used (order of profiles
    matter).

    Args:
        profiles_data (list): Profile definitions as dictionaries.
        key_values (dict): Mapping of Key <-> Value. Key is checked if is
            available in profile and if Value is matching it's values.
        keys_order (list, tuple): Order of keys from `key_values` which matters
            only when multiple profiles have same score.
        logger (logging.Logger): Optionally can be passed different logger.

    Returns:
        dict/None: Return most matching profile or None if none of profiles
            match at least one criteria.
    """
    if not profiles_data:
        return None

    if not logger:
        logger = log

    if not keys_order:
        keys_order = tuple(key_values.keys())
    else:
        _keys_order = list(keys_order)
        # Make all keys from `key_values` are passed
        for key in key_values.keys():
            if key not in _keys_order:
                _keys_order.append(key)
        keys_order = tuple(_keys_order)

    log_parts = " | ".join([
        "{}: \"{}\"".format(*item)
        for item in key_values.items()
    ])

    logger.debug(
        "Looking for matching profile for: {}".format(log_parts)
    )

    matching_profiles = None
    highest_profile_points = -1
    # Each profile get 1 point for each matching filter. Profile with most
    # points is returned. For cases when more than one profile will match
    # are also stored ordered lists of matching values.
    for profile in profiles_data:
        profile_points = 0
        profile_scores = []

        for key in keys_order:
            value = key_values[key]
            match = validate_value_by_regexes(value, profile.get(key))
            if match == -1:
                profile_value = profile.get(key) or []
                logger.debug(
                    "\"{}\" not found in \"{}\": {}".format(value, key,
                                                            profile_value)
                )
                profile_points = -1
                break

            profile_points += match
            profile_scores.append(bool(match))

        if (
            profile_points < 0
            or profile_points < highest_profile_points
        ):
            continue

        if profile_points > highest_profile_points:
            matching_profiles = []
            highest_profile_points = profile_points

        if profile_points == highest_profile_points:
            matching_profiles.append((profile, profile_scores))

    if not matching_profiles:
        logger.debug(
            "None of profiles match your setup. {}".format(log_parts)
        )
        return None

    if len(matching_profiles) > 1:
        logger.debug(
            "More than one profile match your setup. {}".format(log_parts)
        )

    profile = _profile_exclusion(matching_profiles, logger)
    if profile:
        logger.debug(
            "Profile selected: {}".format(profile)
        )
    return profile

find_executable(executable)

Find full path to executable.

Also tries additional extensions if passed executable does not contain one.

Paths where it is looked for executable is defined by 'PATH' environment variable, 'os.confstr("CS_PATH")' or 'os.defpath'.

Parameters:

Name Type Description Default
executable(str)

Name of executable with or without extension. Can be path to file.

required

Returns:

Type Description

Union[str, None]: Full path to executable with extension which was found otherwise None.

Source code in client/ayon_core/lib/vendor_bin_utils.py
 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
def find_executable(executable):
    """Find full path to executable.

    Also tries additional extensions if passed executable does not contain one.

    Paths where it is looked for executable is defined by 'PATH' environment
    variable, 'os.confstr("CS_PATH")' or 'os.defpath'.

    Args:
        executable(str): Name of executable with or without extension. Can be
            path to file.

    Returns:
        Union[str, None]: Full path to executable with extension which was
            found otherwise None.
    """

    # Skip if passed path is file
    if is_file_executable(executable):
        return executable

    low_platform = platform.system().lower()
    _, ext = os.path.splitext(executable)

    # Prepare extensions to check
    exts = set()
    if ext:
        exts.add(ext.lower())

    else:
        # Add other possible extension variants only if passed executable
        #   does not have any
        if low_platform == "windows":
            exts |= {".exe", ".ps1", ".bat"}
            for ext in os.getenv("PATHEXT", "").split(os.pathsep):
                exts.add(ext.lower())

        else:
            exts |= {".sh"}

    # Executable is a path but there may be missing extension
    #   - this can happen primarily on windows where
    #       e.g. "ffmpeg" should be "ffmpeg.exe"
    exe_dir, exe_filename = os.path.split(executable)
    if exe_dir and os.path.isdir(exe_dir):
        for filename in os.listdir(exe_dir):
            filepath = os.path.join(exe_dir, filename)
            basename, ext = os.path.splitext(filename)
            if (
                basename == exe_filename
                and ext.lower() in exts
                and is_file_executable(filepath)
            ):
                return filepath

    # Get paths where to look for executable
    path_str = os.environ.get("PATH", None)
    if path_str is None:
        if hasattr(os, "confstr"):
            path_str = os.confstr("CS_PATH")
        elif hasattr(os, "defpath"):
            path_str = os.defpath

    if not path_str:
        return None

    paths = path_str.split(os.pathsep)
    for path in paths:
        if not os.path.isdir(path):
            continue
        for filename in os.listdir(path):
            filepath = os.path.abspath(os.path.join(path, filename))
            # Filename matches executable exactly
            if filename == executable and is_file_executable(filepath):
                return filepath

            basename, ext = os.path.splitext(filename)
            if (
                basename == executable
                and ext.lower() in exts
                and is_file_executable(filepath)
            ):
                return filepath

    return None

format_file_size(file_size, suffix=None)

Returns formatted string with size in appropriate unit.

Parameters:

Name Type Description Default
file_size int

Size of file in bytes.

required
suffix str

Suffix for formatted size. Default is 'B' (as bytes).

None

Returns:

Name Type Description
str

Formatted size using proper unit and passed suffix (e.g. 7 MiB).

Source code in client/ayon_core/lib/path_tools.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def format_file_size(file_size, suffix=None):
    """Returns formatted string with size in appropriate unit.

    Args:
        file_size (int): Size of file in bytes.
        suffix (str): Suffix for formatted size. Default is 'B' (as bytes).

    Returns:
        str: Formatted size using proper unit and passed suffix (e.g. 7 MiB).
    """

    if suffix is None:
        suffix = "B"

    for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
        if abs(file_size) < 1024.0:
            return "%3.1f%s%s" % (file_size, unit, suffix)
        file_size /= 1024.0
    return "%.1f%s%s" % (file_size, "Yi", suffix)

get_addons_resources_dir(addon_name, *args)

Get directory for storing resources for addons.

Some addons might need to store ad-hoc resources that are not part of addon client package (e.g. because of size). Studio might define dedicated directory to store them with 'AYON_ADDONS_RESOURCES_DIR' environment variable. By default, is used 'addons_resources' in launcher storage (might be shared across platforms).

Parameters:

Name Type Description Default
addon_name str

Addon name.

required
*args str

Subfolders in resources directory.

()

Returns:

Name Type Description
str str

Path to resources directory.

Source code in client/ayon_core/lib/local_settings.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def get_addons_resources_dir(addon_name: str, *args) -> str:
    """Get directory for storing resources for addons.

    Some addons might need to store ad-hoc resources that are not part of
        addon client package (e.g. because of size). Studio might define
        dedicated directory to store them with 'AYON_ADDONS_RESOURCES_DIR'
        environment variable. By default, is used 'addons_resources' in
        launcher storage (might be shared across platforms).

    Args:
        addon_name (str): Addon name.
        *args (str): Subfolders in resources directory.

    Returns:
        str: Path to resources directory.

    """
    addons_resources_dir = os.getenv("AYON_ADDONS_RESOURCES_DIR")
    if not addons_resources_dir:
        addons_resources_dir = get_launcher_storage_dir("addons_resources")

    return os.path.join(addons_resources_dir, addon_name, *args)

get_ayon_launcher_args(*args)

Arguments to run AYON launcher process.

Arguments for subprocess when need to spawn new AYON launcher process.

Reasons

AYON launcher started from code has different executable set to virtual env python and must have path to script as first argument which is not needed for built application.

Parameters:

Name Type Description Default
*args str

Any arguments that will be added after executables.

()

Returns:

Type Description

list[str]: List of arguments to run AYON launcher process.

Source code in client/ayon_core/lib/execute.py
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def get_ayon_launcher_args(*args):
    """Arguments to run AYON launcher process.

    Arguments for subprocess when need to spawn new AYON launcher process.

    Reasons:
        AYON launcher started from code has different executable set to
            virtual env python and must have path to script as first argument
            which is not needed for built application.

    Args:
        *args (str): Any arguments that will be added after executables.

    Returns:
        list[str]: List of arguments to run AYON launcher process.

    """
    executable = os.environ["AYON_EXECUTABLE"]
    launch_args = [executable]

    executable_filename = os.path.basename(executable)
    if "python" in executable_filename.lower():
        filepath = os.path.join(os.environ["AYON_ROOT"], "start.py")
        launch_args.append(filepath)

    if args:
        launch_args.extend(args)

    return launch_args

get_ayon_username()

AYON username used for templates and publishing.

Uses curet ayon api username.

Returns:

Name Type Description
str

Username.

Source code in client/ayon_core/lib/local_settings.py
591
592
593
594
595
596
597
598
599
600
def get_ayon_username():
    """AYON username used for templates and publishing.

    Uses curet ayon api username.

    Returns:
        str: Username.

    """
    return ayon_api.get_user()["name"]

get_datetime_data(datetime_obj=None)

Returns current datetime data as dictionary.

Parameters:

Name Type Description Default
datetime_obj datetime

Specific datetime object

None

Returns:

Name Type Description
dict

prepared date & time data

Available keys

"d" - in shortest possible way. "dd" - with 2 digits. "ddd" - shortened week day. e.g.: Mon, ... "dddd" - full name of week day. e.g.: Monday, ... "m" - in shortest possible way. e.g.: 1 if January "mm" - with 2 digits. "mmm" - shortened month name. e.g.: Jan, ... "mmmm" - full month name. e.g.: January, ... "yy" - shortened year. e.g.: 19, 20, ... "yyyy" - full year. e.g.: 2019, 2020, ... "H" - shortened hours. "HH" - with 2 digits. "h" - shortened hours. "hh" - with 2 digits. "ht" - AM or PM. "M" - shortened minutes. "MM" - with 2 digits. "S" - shortened seconds. "SS" - with 2 digits.

Source code in client/ayon_core/lib/dateutils.py
 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
def get_datetime_data(datetime_obj=None):
    """Returns current datetime data as dictionary.

    Args:
        datetime_obj (datetime): Specific datetime object

    Returns:
        dict: prepared date & time data

    Available keys:
        "d" - <Day of month number> in shortest possible way.
        "dd" - <Day of month number> with 2 digits.
        "ddd" - <Week day name> shortened week day. e.g.: `Mon`, ...
        "dddd" - <Week day name> full name of week day. e.g.: `Monday`, ...
        "m" - <Month number> in shortest possible way. e.g.: `1` if January
        "mm" - <Month number> with 2 digits.
        "mmm" - <Month name> shortened month name. e.g.: `Jan`, ...
        "mmmm" - <Month name> full month name. e.g.: `January`, ...
        "yy" - <Year number> shortened year. e.g.: `19`, `20`, ...
        "yyyy" - <Year number> full year. e.g.: `2019`, `2020`, ...
        "H" - <Hours number 24-hour> shortened hours.
        "HH" - <Hours number 24-hour> with 2 digits.
        "h" - <Hours number 12-hour> shortened hours.
        "hh" - <Hours number 12-hour> with 2 digits.
        "ht" - <Midday type> AM or PM.
        "M" - <Minutes number> shortened minutes.
        "MM" - <Minutes number> with 2 digits.
        "S" - <Seconds number> shortened seconds.
        "SS" - <Seconds number> with 2 digits.
    """

    if not datetime_obj:
        datetime_obj = datetime.datetime.now()

    year = datetime_obj.strftime("%Y")

    month = datetime_obj.strftime("%m")
    month_name_full = datetime_obj.strftime("%B")
    month_name_short = datetime_obj.strftime("%b")
    day = datetime_obj.strftime("%d")

    weekday_full = datetime_obj.strftime("%A")
    weekday_short = datetime_obj.strftime("%a")

    hours = datetime_obj.strftime("%H")
    hours_midday = datetime_obj.strftime("%I")
    hour_midday_type = datetime_obj.strftime("%p")
    minutes = datetime_obj.strftime("%M")
    seconds = datetime_obj.strftime("%S")

    return {
        "d": str(int(day)),
        "dd": str(day),
        "ddd": weekday_short,
        "dddd": weekday_full,
        "m": str(int(month)),
        "mm": str(month),
        "mmm": month_name_short,
        "mmmm": month_name_full,
        "yy": str(year[2:]),
        "yyyy": str(year),
        "H": str(int(hours)),
        "HH": str(hours),
        "h": str(int(hours_midday)),
        "hh": str(hours_midday),
        "ht": hour_midday_type,
        "M": str(int(minutes)),
        "MM": str(minutes),
        "S": str(int(seconds)),
        "SS": str(seconds),
    }

get_ffmpeg_codec_args(ffprobe_data, source_ffmpeg_cmd=None, logger=None)

Copy codec from input metadata for output.

Parameters:

Name Type Description Default
ffprobe_data(dict)

Data received from ffprobe.

required
source_ffmpeg_cmd(str)

Command that created input if available.

required
Source code in client/ayon_core/lib/transcoding.py
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
def get_ffmpeg_codec_args(ffprobe_data, source_ffmpeg_cmd=None, logger=None):
    """Copy codec from input metadata for output.

    Args:
        ffprobe_data(dict): Data received from ffprobe.
        source_ffmpeg_cmd(str): Command that created input if available.
    """
    if logger is None:
        logger = logging.getLogger(__name__)

    video_stream = None
    no_audio_stream = None
    for stream in ffprobe_data["streams"]:
        codec_type = stream["codec_type"]
        if codec_type == "video":
            video_stream = stream
            break
        elif no_audio_stream is None and codec_type != "audio":
            no_audio_stream = stream

    if video_stream is None:
        if no_audio_stream is None:
            logger.warning(
                "Couldn't find stream that is not an audio file."
            )
            return []
        logger.info(
            "Didn't find video stream. Using first non audio stream."
        )
        video_stream = no_audio_stream

    codec_name = video_stream.get("codec_name")
    # Codec "prores"
    if codec_name == "prores":
        return _ffmpeg_prores_codec_args(video_stream, source_ffmpeg_cmd)

    # Codec "h264"
    if codec_name == "h264":
        return _ffmpeg_h264_codec_args(video_stream, source_ffmpeg_cmd)

    # Coded DNxHD
    if codec_name == "dnxhd":
        return _ffmpeg_dnxhd_codec_args(video_stream, source_ffmpeg_cmd)

    output = []
    if codec_name:
        output.extend(["-codec:v", codec_name])

    bit_rate = video_stream.get("bit_rate")
    if bit_rate:
        output.extend(["-b:v", bit_rate])

    pix_fmt = video_stream.get("pix_fmt")
    if pix_fmt:
        output.extend(["-pix_fmt", pix_fmt])

    output.extend(["-g", "1"])

    return output

get_ffmpeg_format_args(ffprobe_data, source_ffmpeg_cmd=None)

Copy format from input metadata for output.

Parameters:

Name Type Description Default
ffprobe_data(dict)

Data received from ffprobe.

required
source_ffmpeg_cmd(str)

Command that created input if available.

required
Source code in client/ayon_core/lib/transcoding.py
838
839
840
841
842
843
844
845
846
847
848
def get_ffmpeg_format_args(ffprobe_data, source_ffmpeg_cmd=None):
    """Copy format from input metadata for output.

    Args:
        ffprobe_data(dict): Data received from ffprobe.
        source_ffmpeg_cmd(str): Command that created input if available.
    """
    input_format = ffprobe_data.get("format") or {}
    if input_format.get("format_name") == "mxf":
        return _ffmpeg_mxf_format_args(ffprobe_data, source_ffmpeg_cmd)
    return []

get_ffmpeg_tool_args(tool_name, *extra_args)

Arguments to launch FFmpeg tool.

Parameters:

Name Type Description Default
tool_name str

Tool name 'ffmpeg', 'ffprobe', exc.

required
*extra_args str

Extra arguments to add to after tool arguments.

()

Returns:

Type Description

list[str]: List of arguments.

Source code in client/ayon_core/lib/vendor_bin_utils.py
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def get_ffmpeg_tool_args(tool_name, *extra_args):
    """Arguments to launch FFmpeg tool.

    Args:
        tool_name (str): Tool name 'ffmpeg', 'ffprobe', exc.
        *extra_args (str): Extra arguments to add to after tool arguments.

    Returns:
        list[str]: List of arguments.
    """

    extra_args = list(extra_args)

    args = _get_ayon_ffmpeg_tool_args(tool_name)
    if args:
        return args + extra_args

    executable_path = get_ffmpeg_tool_path(tool_name)
    if executable_path:
        return [executable_path] + extra_args
    raise ToolNotFoundError(
        "FFmpeg '{}' tool not found.".format(tool_name)
    )

get_ffmpeg_tool_path(tool='ffmpeg')

Path to vendorized FFmpeg executable.

Parameters:

Name Type Description Default
tool str

Tool name 'ffmpeg', 'ffprobe', etc. Default is "ffmpeg".

'ffmpeg'

Returns:

Name Type Description
str

Full path to ffmpeg executable.

Source code in client/ayon_core/lib/vendor_bin_utils.py
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def get_ffmpeg_tool_path(tool="ffmpeg"):
    """Path to vendorized FFmpeg executable.

    Args:
        tool (str): Tool name 'ffmpeg', 'ffprobe', etc.
            Default is "ffmpeg".

    Returns:
        str: Full path to ffmpeg executable.
    """

    if CachedToolPaths.is_tool_cached(tool):
        return CachedToolPaths.get_executable_path(tool)

    args = _get_ayon_ffmpeg_tool_args(tool)
    if args is not None:
        if len(args) > 1:
            raise ValueError(
                "AYON ffmpeg arguments consist of multiple arguments."
            )
        tool_executable_path = args[0]
        CachedToolPaths.cache_executable_path(tool, tool_executable_path)
        return tool_executable_path

    custom_paths_str = os.environ.get("AYON_FFMPEG_PATHS") or ""
    tool_executable_path = find_tool_in_custom_paths(
        custom_paths_str.split(os.pathsep),
        tool,
        _ffmpeg_executable_validation
    )

    # Look to PATH for the tool
    if not tool_executable_path:
        from_path = find_executable(tool)
        if from_path and _ffmpeg_executable_validation(from_path):
            tool_executable_path = from_path

    CachedToolPaths.cache_executable_path(tool, tool_executable_path)
    return tool_executable_path

get_ffprobe_data(path_to_file, logger=None)

Load data about entered filepath via ffprobe.

Parameters:

Name Type Description Default
path_to_file str

absolute path

required
logger Logger

injected logger, if empty new is created

None
Source code in client/ayon_core/lib/transcoding.py
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
def get_ffprobe_data(path_to_file, logger=None):
    """Load data about entered filepath via ffprobe.

    Args:
        path_to_file (str): absolute path
        logger (logging.Logger): injected logger, if empty new is created
    """
    if not logger:
        logger = logging.getLogger(__name__)
    logger.debug(
        "Getting information about input \"{}\".".format(path_to_file)
    )
    ffprobe_args = get_ffmpeg_tool_args("ffprobe")
    args = ffprobe_args + [
        "-hide_banner",
        "-loglevel", "fatal",
        "-show_error",
        "-show_format",
        "-show_streams",
        "-show_programs",
        "-show_chapters",
        "-show_private_data",
        "-print_format", "json",
        path_to_file
    ]

    logger.debug("FFprobe command: {}".format(
        subprocess.list2cmdline(args)
    ))
    kwargs = {
        "stdout": subprocess.PIPE,
        "stderr": subprocess.PIPE,
    }
    if platform.system().lower() == "windows":
        kwargs["creationflags"] = (
            subprocess.CREATE_NEW_PROCESS_GROUP
            | getattr(subprocess, "DETACHED_PROCESS", 0)
            | getattr(subprocess, "CREATE_NO_WINDOW", 0)
        )

    popen = subprocess.Popen(args, **kwargs)

    popen_stdout, popen_stderr = popen.communicate()
    if popen_stdout:
        logger.debug("FFprobe stdout:\n{}".format(
            popen_stdout.decode("utf-8")
        ))

    if popen_stderr:
        logger.warning("FFprobe stderr:\n{}".format(
            popen_stderr.decode("utf-8")
        ))

    return json.loads(popen_stdout)

get_ffprobe_streams(path_to_file, logger=None)

Load streams from entered filepath via ffprobe.

Parameters:

Name Type Description Default
path_to_file str

absolute path

required
logger Logger

injected logger, if empty new is created

None
Source code in client/ayon_core/lib/transcoding.py
828
829
830
831
832
833
834
835
def get_ffprobe_streams(path_to_file, logger=None):
    """Load streams from entered filepath via ffprobe.

    Args:
        path_to_file (str): absolute path
        logger (logging.Logger): injected logger, if empty new is created
    """
    return get_ffprobe_data(path_to_file, logger)["streams"]

get_last_version_from_path(path_dir, filter)

Find last version of given directory content.

Parameters:

Name Type Description Default
path_dir str

directory path

required
filter list

list of strings used as file name filter

required

Returns:

Name Type Description
str

file name with last version

Example

last_version_file = get_last_version_from_path( "/project/shots/shot01/work", ["shot01", "compositing", "nk"])

Source code in client/ayon_core/lib/path_tools.py
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
def get_last_version_from_path(path_dir, filter):
    """Find last version of given directory content.

    Args:
        path_dir (str): directory path
        filter (list): list of strings used as file name filter

    Returns:
        str: file name with last version

    Example:
        last_version_file = get_last_version_from_path(
            "/project/shots/shot01/work", ["shot01", "compositing", "nk"])
    """

    assert os.path.isdir(path_dir), "`path_dir` argument needs to be directory"
    assert isinstance(filter, list) and (
        len(filter) != 0), "`filter` argument needs to be list and not empty"

    filtered_files = list()

    # form regex for filtering
    pattern = r".*".join(filter)

    for file in os.listdir(path_dir):
        if not re.findall(pattern, file):
            continue
        filtered_files.append(file)

    if filtered_files:
        filtered_files.sort()
        return filtered_files[-1]

    return None

get_launcher_local_dir(*subdirs)

Get local directory for launcher.

Local directory is used for storing machine or user specific data.

The location is user specific.

Note

This function should be called at least once on bootstrap.

Parameters:

Name Type Description Default
*subdirs str

Subdirectories relative to local dir.

()

Returns:

Name Type Description
str str

Path to local directory.

Source code in client/ayon_core/lib/local_settings.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_launcher_local_dir(*subdirs: str) -> str:
    """Get local directory for launcher.

    Local directory is used for storing machine or user specific data.

    The location is user specific.

    Note:
        This function should be called at least once on bootstrap.

    Args:
        *subdirs (str): Subdirectories relative to local dir.

    Returns:
        str: Path to local directory.

    """
    storage_dir = os.getenv("AYON_LAUNCHER_LOCAL_DIR")
    if not storage_dir:
        storage_dir = _get_ayon_appdirs()

    return os.path.join(storage_dir, *subdirs)

get_launcher_storage_dir(*subdirs)

Get storage directory for launcher.

Storage directory is used for storing shims, addons, dependencies, etc.

It is not recommended, but the location can be shared across multiple machines.

Note

This function should be called at least once on bootstrap.

Parameters:

Name Type Description Default
*subdirs str

Subdirectories relative to storage dir.

()

Returns:

Name Type Description
str str

Path to storage directory.

Source code in client/ayon_core/lib/local_settings.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_launcher_storage_dir(*subdirs: str) -> str:
    """Get storage directory for launcher.

    Storage directory is used for storing shims, addons, dependencies, etc.

    It is not recommended, but the location can be shared across
        multiple machines.

    Note:
        This function should be called at least once on bootstrap.

    Args:
        *subdirs (str): Subdirectories relative to storage dir.

    Returns:
        str: Path to storage directory.

    """
    storage_dir = os.getenv("AYON_LAUNCHER_STORAGE_DIR")
    if not storage_dir:
        storage_dir = _get_ayon_appdirs()

    return os.path.join(storage_dir, *subdirs)

get_linux_launcher_args(*args)

Path to application mid process executable.

This function should be able as arguments are different when used from code and build.

It is possible that this function is used in AYON build which does not have yet the new executable. In that case 'None' is returned.

Todos

Replace by script in scripts for ayon-launcher.

Parameters:

Name Type Description Default
args iterable

List of additional arguments added after executable argument.

()

Returns:

Name Type Description
list

Executables with possible positional argument to script when called from code.

Source code in client/ayon_core/lib/execute.py
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def get_linux_launcher_args(*args):
    """Path to application mid process executable.

    This function should be able as arguments are different when used
    from code and build.

    It is possible that this function is used in AYON build which does
    not have yet the new executable. In that case 'None' is returned.

    Todos:
        Replace by script in scripts for ayon-launcher.

    Args:
        args (iterable): List of additional arguments added after executable
            argument.

    Returns:
        list: Executables with possible positional argument to script when
            called from code.
    """
    filename = "app_launcher"
    executable = os.environ["AYON_EXECUTABLE"]

    executable_filename = os.path.basename(executable)
    if "python" in executable_filename.lower():
        root = os.environ["AYON_ROOT"]
        script_path = os.path.join(root, "{}.py".format(filename))
        launch_args = [executable, script_path]
    else:
        new_executable = os.path.join(
            os.path.dirname(executable),
            filename
        )
        executable_path = find_executable(new_executable)
        if executable_path is None:
            return None
        launch_args = [executable_path]

    if args:
        launch_args.extend(args)

    return launch_args

get_local_site_id()

Get local site identifier.

Identifier is created if does not exist yet.

Source code in client/ayon_core/lib/local_settings.py
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
def get_local_site_id():
    """Get local site identifier.

    Identifier is created if does not exist yet.
    """
    # used for background syncing
    site_id = os.environ.get("AYON_SITE_ID")
    if site_id:
        return site_id

    site_id_path = get_launcher_local_dir("site_id")
    if os.path.exists(site_id_path):
        with open(site_id_path, "r") as stream:
            site_id = stream.read()

    if site_id:
        return site_id

    try:
        from ayon_common.utils import get_local_site_id as _get_local_site_id
        site_id = _get_local_site_id()
    except ImportError:
        raise ValueError("Couldn't access local site id")

    return site_id

get_media_mime_type(filepath)

Determine Mime-Type of a file.

Parameters:

Name Type Description Default
filepath str

Path to file.

required

Returns:

Type Description
Optional[str]

Optional[str]: Mime type or None if is unknown mime type.

Source code in client/ayon_core/lib/transcoding.py
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
def get_media_mime_type(filepath: str) -> Optional[str]:
    """Determine Mime-Type of a file.

    Args:
        filepath (str): Path to file.

    Returns:
        Optional[str]: Mime type or None if is unknown mime type.

    """
    if not filepath or not os.path.exists(filepath):
        return None

    with open(filepath, "rb") as stream:
        content = stream.read()

    content_len = len(content)
    # Pre-validation (largest definition check)
    # - hopefully there cannot be media defined in less than 12 bytes
    if content_len < 12:
        return None

    # FTYP
    if content[4:8] == b"ftyp":
        return _get_media_mime_type_from_ftyp(content)

    # BMP
    if content[0:2] == b"BM":
        return "image/bmp"

    # Tiff
    if content[0:2] in (b"MM", b"II"):
        return "tiff"

    # PNG
    if content[0:4] == b"\211PNG":
        return "image/png"

    # SVG
    if b'xmlns="http://www.w3.org/2000/svg"' in content:
        return "image/svg+xml"

    # JPEG, JFIF or Exif
    if (
        content[0:4] == b"\xff\xd8\xff\xdb"
        or content[6:10] in (b"JFIF", b"Exif")
    ):
        return "image/jpeg"

    # Webp
    if content[0:4] == b"RIFF" and content[8:12] == b"WEBP":
        return "image/webp"

    # Gif
    if content[0:6] in (b"GIF87a", b"GIF89a"):
        return "gif"

    # Adobe PhotoShop file (8B > Adobe, PS > PhotoShop)
    if content[0:4] == b"8BPS":
        return "image/vnd.adobe.photoshop"

    # Windows ICO > this might be wild guess as multiple files can start
    #   with this header
    if content[0:4] == b"\x00\x00\x01\x00":
        return "image/x-icon"
    return None

get_oiio_tool_args(tool_name, *extra_args)

Arguments to launch OpenImageIO tool.

Parameters:

Name Type Description Default
tool_name str

Tool name 'oiiotool', 'maketx', etc.

required
*extra_args str

Extra arguments to add to after tool arguments.

()

Returns:

Type Description

list[str]: List of arguments.

Source code in client/ayon_core/lib/vendor_bin_utils.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def get_oiio_tool_args(tool_name, *extra_args):
    """Arguments to launch OpenImageIO tool.

    Args:
        tool_name (str): Tool name 'oiiotool', 'maketx', etc.
        *extra_args (str): Extra arguments to add to after tool arguments.

    Returns:
        list[str]: List of arguments.
    """

    extra_args = list(extra_args)

    args = _get_ayon_oiio_tool_args(tool_name)
    if args:
        return args + extra_args

    path = get_oiio_tools_path(tool_name)
    if path:
        return [path] + extra_args
    raise ToolNotFoundError(
        "OIIO '{}' tool not found.".format(tool_name)
    )

get_oiio_tools_path(tool='oiiotool')

Path to OpenImageIO tool executables.

On Windows it adds .exe extension if missing from tool argument.

Parameters:

Name Type Description Default
tool string

Tool name 'oiiotool', 'maketx', etc. Default is "oiiotool".

'oiiotool'
Source code in client/ayon_core/lib/vendor_bin_utils.py
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
def get_oiio_tools_path(tool="oiiotool"):
    """Path to OpenImageIO tool executables.

    On Windows it adds .exe extension if missing from tool argument.

    Args:
        tool (string): Tool name 'oiiotool', 'maketx', etc.
            Default is "oiiotool".
    """

    if CachedToolPaths.is_tool_cached(tool):
        return CachedToolPaths.get_executable_path(tool)

    args = _get_ayon_oiio_tool_args(tool)
    if args:
        if len(args) > 1:
            raise ValueError(
                "AYON oiio arguments consist of multiple arguments."
            )
        tool_executable_path = args[0]
        CachedToolPaths.cache_executable_path(tool, tool_executable_path)
        return tool_executable_path

    custom_paths_str = os.environ.get("AYON_OIIO_PATHS") or ""
    tool_executable_path = find_tool_in_custom_paths(
        custom_paths_str.split(os.pathsep),
        tool,
        _oiio_executable_validation
    )

    # Look to PATH for the tool
    if not tool_executable_path:
        from_path = find_executable(tool)
        if from_path and _oiio_executable_validation(from_path):
            tool_executable_path = from_path

    CachedToolPaths.cache_executable_path(tool, tool_executable_path)
    return tool_executable_path

get_paths_from_environ(env_key=None, env_value=None, return_first=False)

Return existing paths from specific environment variable.

Parameters:

Name Type Description Default
env_key Optional[str]

Environment key where should look for paths.

None
env_value Optional[str]

Value of environment variable. Argument env_key is skipped if this argument is entered.

None
return_first bool

Return first found value or return list of found paths. None or empty list returned if nothing found.

False

Returns:

Type Description
Optional[Union[str, list[str]]]

Optional[Union[str, list[str]]]: Result of found path/s.

Source code in client/ayon_core/lib/env_tools.py
 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
def get_paths_from_environ(
    env_key: Optional[str] = None,
    env_value: Optional[str] = None,
    return_first: bool = False,
) -> Optional[Union[str, list[str]]]:
    """Return existing paths from specific environment variable.

    Args:
        env_key (Optional[str]): Environment key where should look for paths.
        env_value (Optional[str]): Value of environment variable.
            Argument `env_key` is skipped if this argument is entered.
        return_first (bool): Return first found value or return list of found
            paths. `None` or empty list returned if nothing found.

    Returns:
        Optional[Union[str, list[str]]]: Result of found path/s.

    """
    existing_paths = []
    if not env_key and not env_value:
        if return_first:
            return None
        return existing_paths

    if env_value is None:
        env_value = os.environ.get(env_key) or ""

    path_items = env_value.split(os.pathsep)
    for path in path_items:
        # Skip empty string
        if not path:
            continue
        # Normalize path
        path = os.path.normpath(path)
        # Check if path exists
        if os.path.exists(path):
            # Return path if `return_first` is set to True
            if return_first:
                return path
            # Store path
            existing_paths.append(path)

    # Return None if none of paths exists
    if return_first:
        return None
    # Return all existing paths from environment variable
    return existing_paths

get_rescaled_command_arguments(application, input_path, target_width, target_height, target_par=None, bg_color=None, log=None)

Get command arguments for rescaling input to target size.

Parameters:

Name Type Description Default
application str

Application for which command should be created. Currently supported are "ffmpeg" and "oiiotool".

required
input_path str

Path to input file.

required
target_width int

Width of target.

required
target_height int

Height of target.

required
target_par Optional[float]

Pixel aspect ratio of target.

None
bg_color Optional[list[int]]

List of 8bit int values for background color. Should be in range 0 - 255.

None
log Optional[Logger]

Logger used for logging.

None

Returns:

Type Description

list[str]: List of command arguments.

Source code in client/ayon_core/lib/transcoding.py
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
def get_rescaled_command_arguments(
        application,
        input_path,
        target_width,
        target_height,
        target_par=None,
        bg_color=None,
        log=None
):
    """Get command arguments for rescaling input to target size.

    Args:
        application (str): Application for which command should be created.
            Currently supported are "ffmpeg" and "oiiotool".
        input_path (str): Path to input file.
        target_width (int): Width of target.
        target_height (int): Height of target.
        target_par (Optional[float]): Pixel aspect ratio of target.
        bg_color (Optional[list[int]]): List of 8bit int values for
            background color. Should be in range 0 - 255.
        log (Optional[logging.Logger]): Logger used for logging.

    Returns:
        list[str]: List of command arguments.
    """
    command_args = []
    target_par = target_par or 1.0
    input_par = 1.0

    input_height, input_width, stream_input_par = _get_image_dimensions(
        application, input_path, log)
    if stream_input_par:
        input_par = (
            float(stream_input_par.split(":")[0])
            / float(stream_input_par.split(":")[1])
        )
    # recalculating input and target width
    input_width = int(input_width * input_par)
    target_width = int(target_width * target_par)

    # calculate aspect ratios
    target_aspect = float(target_width) / target_height
    input_aspect = float(input_width) / input_height

    # calculate scale size
    scale_size = float(input_width) / target_width
    if input_aspect < target_aspect:
        scale_size = float(input_height) / target_height

    # calculate rescaled width and height
    rescaled_width = int(input_width / scale_size)
    rescaled_height = int(input_height / scale_size)

    # calculate width and height shift
    rescaled_width_shift = int((target_width - rescaled_width) / 2)
    rescaled_height_shift = int((target_height - rescaled_height) / 2)

    if application == "ffmpeg":
        # create scale command
        scale = "scale={0}:{1}".format(input_width, input_height)
        pad = "pad={0}:{1}:({2}-iw)/2:({3}-ih)/2".format(
            target_width,
            target_height,
            target_width,
            target_height
        )
        if input_width > target_width or input_height > target_height:
            scale = "scale={0}:{1}".format(rescaled_width, rescaled_height)
            pad = "pad={0}:{1}:{2}:{3}".format(
                target_width,
                target_height,
                rescaled_width_shift,
                rescaled_height_shift
            )

        if bg_color:
            color = convert_color_values(application, bg_color)
            pad += ":{0}".format(color)
        command_args.extend(["-vf", "{0},{1}".format(scale, pad)])

    elif application == "oiiotool":
        input_info = get_oiio_info_for_input(input_path, logger=log)
        # Collect channels to export
        _, channels_arg = get_oiio_input_and_channel_args(
            input_info, alpha_default=1.0)

        command_args.extend([
            # Tell oiiotool which channels should be put to top stack
            #   (and output)
            "--ch", channels_arg,
            # Use first subimage
            "--subimage", "0"
        ])

        if input_par != 1.0:
            command_args.extend(["--pixelaspect", "1"])

        width_shift = int((target_width - input_width) / 2)
        height_shift = int((target_height - input_height) / 2)

        # default resample is not scaling source image
        resample = [
            "--resize",
            "{0}x{1}".format(input_width, input_height),
            "--origin",
            "+{0}+{1}".format(width_shift, height_shift),
        ]
        # scaled source image to target size
        if input_width > target_width or input_height > target_height:
            # form resample command
            resample = [
                "--resize:filter=lanczos3",
                "{0}x{1}".format(rescaled_width, rescaled_height),
                "--origin",
                "+{0}+{1}".format(rescaled_width_shift, rescaled_height_shift),
            ]
        command_args.extend(resample)

        fullsize = [
            "--fullsize",
            "{0}x{1}".format(target_width, target_height)
        ]
        if bg_color:
            color = convert_color_values(application, bg_color)

            fullsize.extend([
                "--pattern",
                "constant:color={0}".format(color),
                "{0}x{1}".format(target_width, target_height),
                "4",  # 4 channels
                "--over"
            ])
        command_args.extend(fullsize)

    else:
        raise ValueError(
            "\"application\" input argument should "
            "be either \"ffmpeg\" or \"oiiotool\""
        )

    return command_args

get_timestamp(datetime_obj=None)

Get standardized timestamp from datetime object.

Parameters:

Name Type Description Default
datetime_obj datetime

Object of datetime. Current time is used if not passed.

None
Source code in client/ayon_core/lib/dateutils.py
79
80
81
82
83
84
85
86
87
88
89
90
91
def get_timestamp(datetime_obj=None):
    """Get standardized timestamp from datetime object.

    Args:
        datetime_obj (datetime.datetime): Object of datetime. Current time
            is used if not passed.
    """

    if datetime_obj is None:
        datetime_obj = datetime.datetime.now()
    return datetime_obj.strftime(
        "%Y%m%dT%H%M%SZ"
    )

get_transcode_temp_directory()

Creates temporary folder for transcoding.

Its local, in case of farm it is 'local' to the farm machine.

Should be much faster, needs to be cleaned up later.

Source code in client/ayon_core/lib/transcoding.py
70
71
72
73
74
75
76
77
78
79
def get_transcode_temp_directory():
    """Creates temporary folder for transcoding.

    Its local, in case of farm it is 'local' to the farm machine.

    Should be much faster, needs to be cleaned up later.
    """
    return os.path.normpath(
        tempfile.mkdtemp(prefix="op_transcoding_")
    )

get_version_from_path(file)

Find version number in file path string.

Parameters:

Name Type Description Default
file str

file path

required

Returns:

Name Type Description
str

version number in string ('001')

Source code in client/ayon_core/lib/path_tools.py
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_version_from_path(file):
    """Find version number in file path string.

    Args:
        file (str): file path

    Returns:
        str: version number in string ('001')
    """

    pattern = re.compile(r"[\._]v([0-9]+)", re.IGNORECASE)
    try:
        return pattern.findall(file)[-1]
    except IndexError:
        log.error(
            "templates:get_version_from_workfile:"
            "`{}` missing version string."
            "Example `v004`".format(file)
        )

import_filepath(filepath, module_name=None, sys_module_name=None)

Import python file as python module.

Parameters:

Name Type Description Default
filepath str

Path to python file.

required
module_name str

Name of loaded module. Only for Python 3. By default is filled with filename of filepath.

None
sys_module_name str

Name of module in sys.modules where to store loaded module. By default is None so module is not added to sys.modules.

None

Todo (antirotor): We should add the module to the sys.modules always but we need to be careful about it and test it properly.

Source code in client/ayon_core/lib/python_module_tools.py
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
def import_filepath(
        filepath: str,
        module_name: Optional[str] = None,
        sys_module_name: Optional[str] = None) -> types.ModuleType:
    """Import python file as python module.

    Args:
        filepath (str): Path to python file.
        module_name (str): Name of loaded module. Only for Python 3. By default
            is filled with filename of filepath.
        sys_module_name (str): Name of module in `sys.modules` where to store
            loaded module. By default is None so module is not added to
            `sys.modules`.

    Todo (antirotor): We should add the module to the sys.modules always but
        we need to be careful about it and test it properly.

    """
    if module_name is None:
        module_name = os.path.splitext(os.path.basename(filepath))[0]

    # Prepare module object where content of file will be parsed
    module = types.ModuleType(module_name)
    module.__file__ = filepath

    # Use loader so module has full specs
    module_loader = importlib.machinery.SourceFileLoader(
        module_name, filepath
    )
    # only add to sys.modules if requested
    if sys_module_name:
        sys.modules[sys_module_name] = module
    module_loader.exec_module(module)
    return module

import_module_from_dirpath(dirpath, folder_name, dst_module_name=None)

Import passed directory as a python module.

Imported module can be assigned as a child attribute of already loaded module from sys.modules if has support of setattr. That is not default behavior of python modules so parent module must be a custom module with that ability.

It is not possible to reimport already cached module. If you need to reimport module you have to remove it from caches manually.

Parameters:

Name Type Description Default
dirpath str

Parent directory path of loaded folder.

required
folder_name str

Folder name which should be imported inside passed directory.

required
dst_module_name str

Parent module name under which can be loaded module added.

None
Source code in client/ayon_core/lib/python_module_tools.py
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
def import_module_from_dirpath(
        dirpath, folder_name, dst_module_name=None):
    """Import passed directory as a python module.

    Imported module can be assigned as a child attribute of already loaded
    module from `sys.modules` if has support of `setattr`. That is not default
    behavior of python modules so parent module must be a custom module with
    that ability.

    It is not possible to reimport already cached module. If you need to
    reimport module you have to remove it from caches manually.

    Args:
        dirpath (str): Parent directory path of loaded folder.
        folder_name (str): Folder name which should be imported inside passed
            directory.
        dst_module_name (str): Parent module name under which can be loaded
            module added.

    """
    # Import passed dirpath as python module
    if dst_module_name:
        full_module_name = "{}.{}".format(dst_module_name, folder_name)
        dst_module = sys.modules[dst_module_name]
    else:
        full_module_name = folder_name
        dst_module = None

    # Skip import if is already imported
    if full_module_name in sys.modules:
        return sys.modules[full_module_name]

    import importlib.util
    from importlib._bootstrap_external import PathFinder

    # Find loader for passed path and name
    loader = PathFinder.find_module(full_module_name, [dirpath])

    # Load specs of module
    spec = importlib.util.spec_from_loader(
        full_module_name, loader, origin=dirpath
    )

    # Create module based on specs
    module = importlib.util.module_from_spec(spec)

    # Store module to destination module and `sys.modules`
    # WARNING this mus be done before module execution
    if dst_module is not None:
        setattr(dst_module, folder_name, module)

    sys.modules[full_module_name] = module

    # Execute module import
    loader.exec_module(module)

    return module

initialize_ayon_connection(force=False)

Initialize global AYON api connection.

Create global connection in ayon_api module and set site id and client version. Is silently skipped if already happened.

Parameters:

Name Type Description Default
force Optional[bool]

Force reinitialize connection. Defaults to False.

False
Source code in client/ayon_core/lib/ayon_connection.py
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
def initialize_ayon_connection(force=False):
    """Initialize global AYON api connection.

    Create global connection in ayon_api module and set site id
        and client version.
    Is silently skipped if already happened.

    Args:
        force (Optional[bool]): Force reinitialize connection.
            Defaults to False.

    """
    if not force and _Cache.initialized:
        return

    _Cache.initialized = True
    ayon_api_version = (
        semver.VersionInfo.parse(ayon_api.__version__).to_tuple()
    )
    # TODO remove mokey patching after when AYON api is safely updated
    fix_before_1_0_2 = ayon_api_version < (1, 0, 2)
    # Monkey patching to fix 'get_last_version_by_product_name'
    if fix_before_1_0_2:
        ayon_api.ServerAPI.get_last_versions = (
            _new_get_last_versions
        )
        ayon_api.ServerAPI.get_last_version_by_product_id = (
            _new_get_last_version_by_product_id
        )
        ayon_api.ServerAPI.get_last_version_by_product_name = (
            _new_get_last_version_by_product_name
        )

    site_id = get_local_site_id()
    version = os.getenv("AYON_VERSION")
    if ayon_api.is_connection_created():
        con = ayon_api.get_server_api_connection()
        # Monkey patching to fix 'get_last_version_by_product_name'
        if fix_before_1_0_2:
            def _lvs_wrapper(*args, **kwargs):
                return _new_get_last_versions(
                    con, *args, **kwargs
                )
            def _lv_by_pi_wrapper(*args, **kwargs):
                return _new_get_last_version_by_product_id(
                    con, *args, **kwargs
                )
            def _lv_by_pn_wrapper(*args, **kwargs):
                return _new_get_last_version_by_product_name(
                    con, *args, **kwargs
                )
            con.get_last_versions = _lvs_wrapper
            con.get_last_version_by_product_id = _lv_by_pi_wrapper
            con.get_last_version_by_product_name = _lv_by_pn_wrapper
        con.set_site_id(site_id)
        con.set_client_version(version)
    else:
        ayon_api.create_connection(site_id, version)

is_dev_mode_enabled()

Dev mode is enabled in AYON.

Returns:

Name Type Description
bool

True if dev mode is enabled.

Source code in client/ayon_core/lib/ayon_info.py
 99
100
101
102
103
104
105
106
def is_dev_mode_enabled():
    """Dev mode is enabled in AYON.

    Returns:
        bool: True if dev mode is enabled.
    """

    return os.getenv("AYON_USE_DEV") == "1"

is_func_signature_supported(func, *args, **kwargs)

Check if a function signature supports passed args and kwargs.

This check does not actually call the function, just look if function can be called with the arguments.

Notes

This does NOT check if the function would work with passed arguments only if they can be passed in. If function have args, *kwargs in parameters, this will always return 'True'.

Example

def my_function(my_number): ... return my_number + 1 ... is_func_signature_supported(my_function, 1) True is_func_signature_supported(my_function, 1, 2) False is_func_signature_supported(my_function, my_number=1) True is_func_signature_supported(my_function, number=1) False is_func_signature_supported(my_function, "string") True def my_other_function(args, kwargs): ... my_function(args, **kwargs) ... is_func_signature_supported( ... my_other_function, ... "string", ... 1, ... other=None ... ) True

Parameters:

Name Type Description Default
func Callable

A function where the signature should be tested.

required
*args Any

Positional arguments for function signature.

()
**kwargs Any

Keyword arguments for function signature.

{}

Returns:

Name Type Description
bool

Function can pass in arguments.

Source code in client/ayon_core/lib/python_module_tools.py
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
def is_func_signature_supported(func, *args, **kwargs):
    """Check if a function signature supports passed args and kwargs.

    This check does not actually call the function, just look if function can
    be called with the arguments.

    Notes:
        This does NOT check if the function would work with passed arguments
            only if they can be passed in. If function have *args, **kwargs
            in parameters, this will always return 'True'.

    Example:
        >>> def my_function(my_number):
        ...     return my_number + 1
        ...
        >>> is_func_signature_supported(my_function, 1)
        True
        >>> is_func_signature_supported(my_function, 1, 2)
        False
        >>> is_func_signature_supported(my_function, my_number=1)
        True
        >>> is_func_signature_supported(my_function, number=1)
        False
        >>> is_func_signature_supported(my_function, "string")
        True
        >>> def my_other_function(*args, **kwargs):
        ...     my_function(*args, **kwargs)
        ...
        >>> is_func_signature_supported(
        ...     my_other_function,
        ...     "string",
        ...     1,
        ...     other=None
        ... )
        True

    Args:
        func (Callable): A function where the signature should be tested.
        *args (Any): Positional arguments for function signature.
        **kwargs (Any): Keyword arguments for function signature.

    Returns:
        bool: Function can pass in arguments.

    """
    sig = inspect.signature(func)
    try:
        sig.bind(*args, **kwargs)
        return True
    except TypeError:
        pass
    return False

is_in_ayon_launcher_process()

Determine if current process is running from AYON launcher.

Returns:

Name Type Description
bool

True if running from AYON launcher.

Source code in client/ayon_core/lib/ayon_info.py
29
30
31
32
33
34
35
36
37
38
def is_in_ayon_launcher_process():
    """Determine if current process is running from AYON launcher.

    Returns:
        bool: True if running from AYON launcher.

    """
    ayon_executable_path = os.path.normpath(os.environ["AYON_EXECUTABLE"])
    executable_path = os.path.normpath(sys.executable)
    return ayon_executable_path == executable_path

is_in_tests()

Process is running in automatic tests mode.

Returns:

Name Type Description
bool

True if running in tests.

Source code in client/ayon_core/lib/ayon_info.py
89
90
91
92
93
94
95
96
def is_in_tests():
    """Process is running in automatic tests mode.

    Returns:
        bool: True if running in tests.

    """
    return os.environ.get("AYON_IN_TESTS") == "1"

is_oiio_supported()

Checks if oiiotool is configured for this platform.

Returns:

Name Type Description
bool

OIIO tool executable is available.

Source code in client/ayon_core/lib/vendor_bin_utils.py
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def is_oiio_supported():
    """Checks if oiiotool is configured for this platform.

    Returns:
        bool: OIIO tool executable is available.
    """

    try:
        args = get_oiio_tool_args("oiiotool")
    except ToolNotFoundError:
        args = None
    if not args:
        log.debug("OIIOTool is not configured or not present.")
        return False
    return _oiio_executable_validation(args)

is_running_from_build()

Determine if current process is running from build or code.

Returns:

Name Type Description
bool

True if running from build.

Source code in client/ayon_core/lib/ayon_info.py
41
42
43
44
45
46
47
48
49
50
51
52
def is_running_from_build():
    """Determine if current process is running from build or code.

    Returns:
        bool: True if running from build.

    """
    executable_path = os.environ["AYON_EXECUTABLE"]
    executable_filename = os.path.basename(executable_path)
    if "python" in executable_filename.lower():
        return False
    return True

is_using_ayon_console()

AYON launcher console executable is used.

This function make sense only on Windows platform. For other platforms always returns True. True is also returned if process is running from code.

AYON launcher on windows has 2 executable files. First 'ayon_console.exe' works as 'python.exe' executable, the second 'ayon.exe' works as 'pythonw.exe' executable. The difference is way how stdout/stderr is handled (especially when calling subprocess).

Returns:

Name Type Description
bool

True if console executable is used.

Source code in client/ayon_core/lib/ayon_info.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def is_using_ayon_console():
    """AYON launcher console executable is used.

    This function make sense only on Windows platform. For other platforms
    always returns True. True is also returned if process is running from
    code.

    AYON launcher on windows has 2 executable files. First 'ayon_console.exe'
    works as 'python.exe' executable, the second 'ayon.exe' works as
    'pythonw.exe' executable. The difference is way how stdout/stderr is
    handled (especially when calling subprocess).

    Returns:
        bool: True if console executable is used.

    """
    if (
        platform.system().lower() != "windows"
        or is_running_from_build()
    ):
        return True
    executable_path = os.environ["AYON_EXECUTABLE"]
    executable_filename = os.path.basename(executable_path)
    return "ayon_console" in executable_filename

modules_from_path(folder_path)

Get python scripts as modules from a path.

Parameters:

Name Type Description Default
path str

Path to folder containing python scripts.

required

Returns:

Type Description

tuple: First list contains successfully imported modules and second list contains tuples of path and exception.

Source code in client/ayon_core/lib/python_module_tools.py
 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
def modules_from_path(folder_path):
    """Get python scripts as modules from a path.

    Arguments:
        path (str): Path to folder containing python scripts.

    Returns:
        tuple<list, list>: First list contains successfully imported modules
            and second list contains tuples of path and exception.
    """
    crashed = []
    modules = []
    output = (modules, crashed)
    # Just skip and return empty list if path is not set
    if not folder_path:
        return output

    # Do not allow relative imports
    if folder_path.startswith("."):
        log.warning((
            "BUG: Relative paths are not allowed for security reasons. {}"
        ).format(folder_path))
        return output

    folder_path = os.path.normpath(folder_path)

    if not os.path.isdir(folder_path):
        log.warning("Not a directory path: {}".format(folder_path))
        return output

    for filename in os.listdir(folder_path):
        # Ignore files which start with underscore
        if filename.startswith("_"):
            continue

        mod_name, mod_ext = os.path.splitext(filename)
        if not mod_ext == ".py":
            continue

        full_path = os.path.join(folder_path, filename)
        if not os.path.isfile(full_path):
            continue

        try:
            module = import_filepath(full_path, mod_name)
            modules.append((full_path, module))

        except Exception:
            crashed.append((full_path, sys.exc_info()))
            log.warning(
                "Failed to load path: \"{0}\"".format(full_path),
                exc_info=True
            )
            continue

    return output

path_to_subprocess_arg(path)

Prepare path for subprocess arguments.

Returned path can be wrapped with quotes or kept as is.

Parameters:

Name Type Description Default
path str

Path to be converted.

required

Returns:

Name Type Description
str

Converted path.

Source code in client/ayon_core/lib/execute.py
317
318
319
320
321
322
323
324
325
326
327
328
def path_to_subprocess_arg(path):
    """Prepare path for subprocess arguments.

    Returned path can be wrapped with quotes or kept as is.

    Args:
        path (str): Path to be converted.

    Returns:
        str: Converted path.
    """
    return subprocess.list2cmdline([path])

prepare_template_data(fill_pairs)

Prepares formatted data for filling template.

It produces multiple variants of keys (key, Key, KEY) to control format of filled template.

Example

src_data = { ... "host": "maya", ... } output = prepare_template_data(src_data) sorted(list(output.items())) # sort & list conversion for tests [('HOST', 'MAYA'), ('Host', 'Maya'), ('host', 'maya')]

Parameters:

Name Type Description Default
fill_pairs Union[dict[str, Any], Iterable[Tuple[str, Any]]]

The value that are prepared for template.

required

Returns:

Type Description

dict[str, str]: Prepared values for template.

Source code in client/ayon_core/lib/plugin_tools.py
 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
def prepare_template_data(fill_pairs):
    """Prepares formatted data for filling template.

    It produces multiple variants of keys (key, Key, KEY) to control
    format of filled template.

    Example:
        >>> src_data = {
        ...    "host": "maya",
        ... }
        >>> output = prepare_template_data(src_data)
        >>> sorted(list(output.items())) # sort & list conversion for tests
        [('HOST', 'MAYA'), ('Host', 'Maya'), ('host', 'maya')]

    Args:
        fill_pairs (Union[dict[str, Any], Iterable[Tuple[str, Any]]]): The
            value that are prepared for template.

    Returns:
        dict[str, str]: Prepared values for template.
    """

    valid_items = _separate_keys_and_value(fill_pairs)
    output = {}
    for item in valid_items:
        keys, value = item
        # Convert only string values
        if isinstance(value, str):
            upper_value = value.upper()
            capitalized_value = _capitalize_value(value)
        else:
            upper_value = capitalized_value = value

        first_key = keys.pop(0)
        if not keys:
            output[first_key] = value
            output[first_key.upper()] = upper_value
            output[first_key.capitalize()] = capitalized_value
            continue

        # Prepare 'normal', 'upper' and 'capitalized' variables
        normal = output.setdefault(first_key, {})
        capitalized = output.setdefault(first_key.capitalize(), {})
        upper = output.setdefault(first_key.upper(), {})

        keys_deque = collections.deque(keys)
        while keys_deque:
            key = keys_deque.popleft()
            upper_key = key
            if isinstance(key, str):
                upper_key = key.upper()

            if not keys_deque:
                # Fill value on last key
                upper[upper_key] = upper_value
                capitalized[key] = capitalized_value
                normal[key] = value
            else:
                normal = normal.setdefault(key, {})
                capitalized = capitalized.setdefault(key, {})
                upper = upper.setdefault(upper_key, {})
    return output

recursive_bases_from_class(klass)

Extract all bases from entered class.

Source code in client/ayon_core/lib/python_module_tools.py
107
108
109
110
111
112
113
114
def recursive_bases_from_class(klass):
    """Extract all bases from entered class."""
    result = []
    bases = klass.__bases__
    result.extend(bases)
    for base in bases:
        result.extend(recursive_bases_from_class(base))
    return result

register_event_callback(topic, callback)

Add callback that will be executed on specific topic.

Parameters:

Name Type Description Default
topic(str)

Topic on which will callback be triggered.

required
callback(function)

Callback that will be triggered when a topic is triggered. Callback should expect none or 1 argument where Event object is passed.

required

Returns:

Name Type Description
EventCallback

Object wrapping the callback. It can be used to enable/disable listening to a topic or remove the callback from the topic completely.

Source code in client/ayon_core/lib/events.py
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
def register_event_callback(topic, callback):
    """Add callback that will be executed on specific topic.

    Args:
        topic(str): Topic on which will callback be triggered.
        callback(function): Callback that will be triggered when a topic
            is triggered. Callback should expect none or 1 argument where
            `Event` object is passed.

    Returns:
        EventCallback: Object wrapping the callback. It can be used to
            enable/disable listening to a topic or remove the callback from
            the topic completely.
    """

    return GlobalEventSystem.add_callback(topic, callback)

run_ayon_launcher_process(*args, add_sys_paths=False, **kwargs)

Execute AYON process with passed arguments and wait.

Wrapper for 'run_process' which prepends AYON executable arguments before passed arguments and define environments if are not passed.

Values from 'os.environ' are used for environments if are not passed. They are cleaned using 'clean_envs_for_ayon_process' function.

Example:

run_ayon_process("run", "<path to .py script>")

Parameters:

Name Type Description Default
*args str

ayon-launcher cli arguments.

()
**kwargs Any

Keyword arguments for subprocess.Popen.

{}

Returns:

Name Type Description
str

Full output of subprocess concatenated stdout and stderr.

Source code in client/ayon_core/lib/execute.py
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
def run_ayon_launcher_process(*args, add_sys_paths=False, **kwargs):
    """Execute AYON process with passed arguments and wait.

    Wrapper for 'run_process' which prepends AYON executable arguments
    before passed arguments and define environments if are not passed.

    Values from 'os.environ' are used for environments if are not passed.
    They are cleaned using 'clean_envs_for_ayon_process' function.

    Example:
    ```
    run_ayon_process("run", "<path to .py script>")
    ```

    Args:
        *args (str): ayon-launcher cli arguments.
        **kwargs (Any): Keyword arguments for subprocess.Popen.

    Returns:
        str: Full output of subprocess concatenated stdout and stderr.

    """
    args = get_ayon_launcher_args(*args)
    env = kwargs.pop("env", None)
    # Keep env untouched if are passed and not empty
    if not env:
        # Skip envs that can affect AYON launcher process
        # - fill more if you find more
        env = clean_envs_for_ayon_process(os.environ)

    if add_sys_paths:
        new_pythonpath = list(sys.path)
        lookup_set = set(new_pythonpath)
        for path in (env.get("PYTHONPATH") or "").split(os.pathsep):
            if path and path not in lookup_set:
                new_pythonpath.append(path)
                lookup_set.add(path)
        env["PYTHONPATH"] = os.pathsep.join(new_pythonpath)

    return run_subprocess(args, env=env, **kwargs)

run_detached_process(args, **kwargs)

Execute process with passed arguments as separated process.

Example

run_detached_process("run", "./path_to.py")

Parameters:

Name Type Description Default
args Iterable[str]

AYON cli arguments.

required
**kwargs dict

Keyword arguments for subprocess.Popen.

{}

Returns:

Type Description

subprocess.Popen: Pointer to launched process but it is possible that launched process is already killed (on linux).

Source code in client/ayon_core/lib/execute.py
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
def run_detached_process(args, **kwargs):
    """Execute process with passed arguments as separated process.

    Example:
        >>> run_detached_process("run", "./path_to.py")


    Args:
        args (Iterable[str]): AYON cli arguments.
        **kwargs (dict): Keyword arguments for subprocess.Popen.

    Returns:
        subprocess.Popen: Pointer to launched process but it is possible that
            launched process is already killed (on linux).

    """
    env = kwargs.pop("env", None)
    # Keep env untouched if are passed and not empty
    if not env:
        env = os.environ

    # Create copy of passed env
    kwargs["env"] = {k: v for k, v in env.items()}

    low_platform = platform.system().lower()
    if low_platform == "darwin":
        new_args = ["open", "-na", args.pop(0), "--args"]
        new_args.extend(args)
        args = new_args

    elif low_platform == "windows":
        flags = (
            subprocess.CREATE_NEW_PROCESS_GROUP
            | subprocess.DETACHED_PROCESS
        )
        kwargs["creationflags"] = flags

        if not sys.stdout:
            kwargs["stdout"] = subprocess.DEVNULL
            kwargs["stderr"] = subprocess.DEVNULL

    elif low_platform == "linux" and get_linux_launcher_args() is not None:
        json_data = {
            "args": args,
            "env": kwargs.pop("env")
        }
        json_temp = tempfile.NamedTemporaryFile(
            mode="w", prefix="op_app_args", suffix=".json", delete=False
        )
        json_temp.close()
        json_temp_filpath = json_temp.name
        with open(json_temp_filpath, "w") as stream:
            json.dump(json_data, stream)

        new_args = get_linux_launcher_args()
        new_args.append(json_temp_filpath)

        # Create mid-process which will launch application
        process = subprocess.Popen(new_args, **kwargs)
        # Wait until the process finishes
        #   - This is important! The process would stay in "open" state.
        process.wait()
        # Remove the temp file
        os.remove(json_temp_filpath)
        # Return process which is already terminated
        return process

    process = subprocess.Popen(args, **kwargs)
    return process

run_subprocess(*args, **kwargs)

Convenience method for getting output errors for subprocess.

Output logged when process finish.

Entered arguments and keyword arguments are passed to subprocess Popen.

On windows are 'creationflags' filled with flags that should cause ignore creation of new window.

Parameters:

Name Type Description Default
*args

Variable length argument list passed to Popen.

()
**kwargs

Arbitrary keyword arguments passed to Popen. Is possible to pass logging.Logger object under "logger" to use custom logger for output.

{}

Returns:

Name Type Description
str

Full output of subprocess concatenated stdout and stderr.

Raises:

Type Description
RuntimeError

Exception is raised if process finished with nonzero return code.

Source code in client/ayon_core/lib/execute.py
 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
def run_subprocess(*args, **kwargs):
    """Convenience method for getting output errors for subprocess.

    Output logged when process finish.

    Entered arguments and keyword arguments are passed to subprocess Popen.

    On windows are 'creationflags' filled with flags that should cause ignore
    creation of new window.

    Args:
        *args: Variable length argument list passed to Popen.
        **kwargs : Arbitrary keyword arguments passed to Popen. Is possible to
            pass `logging.Logger` object under "logger" to use custom logger
            for output.

    Returns:
        str: Full output of subprocess concatenated stdout and stderr.

    Raises:
        RuntimeError: Exception is raised if process finished with nonzero
            return code.

    """
    # Modify creation flags on windows to hide console window if in UI mode
    if (
        platform.system().lower() == "windows"
        and "creationflags" not in kwargs
        # shell=True already tries to hide the console window
        # and passing these creationflags then shows the window again
        # so we avoid it for shell=True cases
        and kwargs.get("shell") is not True
    ):
        kwargs["creationflags"] = (
            subprocess.CREATE_NEW_PROCESS_GROUP
            | getattr(subprocess, "DETACHED_PROCESS", 0)
            | getattr(subprocess, "CREATE_NO_WINDOW", 0)
        )

    # Escape special characters in certain shells
    if (
        kwargs.get("shell") is True
        and len(args) == 1
        and isinstance(args[0], str)
    ):
        # Escape parentheses for bash
        if os.getenv("SHELL") in ("/bin/bash", "/bin/sh"):
            new_arg = (
                args[0]
                .replace("(", "\\(")
                .replace(")", "\\)")
            )
            args = (new_arg,)
        # Escape & on Windows in shell with `cmd.exe` using ^&
        elif (
            platform.system().lower() == "windows"
            and os.getenv("COMSPEC").endswith("cmd.exe")
        ):
            new_arg = args[0].replace("&", "^&")
            args = (new_arg, )

    # Get environments from kwarg or use current process environments if were
    # not passed.
    env = kwargs.get("env") or os.environ
    # Make sure environment contains only strings
    filtered_env = {str(k): str(v) for k, v in env.items()}

    # Use lib's logger if was not passed with kwargs.
    logger = kwargs.pop("logger", None)
    if logger is None:
        logger = Logger.get_logger("run_subprocess")

    # set overrides
    kwargs["stdout"] = kwargs.get("stdout", subprocess.PIPE)
    kwargs["stderr"] = kwargs.get("stderr", subprocess.PIPE)
    kwargs["stdin"] = kwargs.get("stdin", subprocess.PIPE)
    kwargs["env"] = filtered_env

    proc = subprocess.Popen(*args, **kwargs)

    full_output = ""
    _stdout, _stderr = proc.communicate()
    if _stdout:
        _stdout = _stdout.decode("utf-8", errors="backslashreplace")
        full_output += _stdout
        logger.debug(_stdout)

    if _stderr:
        _stderr = _stderr.decode("utf-8", errors="backslashreplace")
        # Add additional line break if output already contains stdout
        if full_output:
            full_output += "\n"
        full_output += _stderr
        logger.info(_stderr)

    if proc.returncode != 0:
        exc_msg = "Executing arguments was not successful: \"{}\"".format(args)
        if _stdout:
            exc_msg += "\n\nOutput:\n{}".format(_stdout)

        if _stderr:
            exc_msg += "Error:\n{}".format(_stderr)

        raise RuntimeError(exc_msg)

    return full_output

should_convert_for_ffmpeg(src_filepath)

Find out if input should be converted for ffmpeg.

Currently cares only about exr inputs and is based on OpenImageIO.

Returns:

Type Description

bool/NoneType: True if should be converted, False if should not and None if can't determine.

Source code in client/ayon_core/lib/transcoding.py
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
def should_convert_for_ffmpeg(src_filepath):
    """Find out if input should be converted for ffmpeg.

    Currently cares only about exr inputs and is based on OpenImageIO.

    Returns:
        bool/NoneType: True if should be converted, False if should not and
            None if can't determine.
    """
    # Care only about exr at this moment
    ext = os.path.splitext(src_filepath)[-1].lower()
    if ext != ".exr":
        return False

    # Can't determine if should convert or not without oiio_tool
    if not is_oiio_supported():
        return None

    # Load info about file from oiio tool
    input_info = get_oiio_info_for_input(src_filepath)
    if not input_info:
        return None

    subimages = input_info.get("subimages")
    if subimages is not None and subimages > 1:
        return True

    # Check compression
    compression = input_info["attribs"].get("compression")
    if compression in ("dwaa", "dwab"):
        return True

    # Check channels
    channel_names = input_info["channelnames"]
    review_channels = get_convert_rgb_channels(channel_names)
    if review_channels is None:
        return None

    for attr_value in input_info["attribs"].values():
        if not isinstance(attr_value, str):
            continue

        if len(attr_value) > MAX_FFMPEG_STRING_LEN:
            return True

        for char in NOT_ALLOWED_FFMPEG_CHARS:
            if char in attr_value:
                return True
    return False

source_hash(filepath, *args)

Generate simple identifier for a source file. This is used to identify whether a source file has previously been processe into the pipeline, e.g. a texture. The hash is based on source filepath, modification time and file size. This is only used to identify whether a specific source file was already published before from the same location with the same modification date. We opt to do it this way as opposed to Avalanch C4 hash as this is much faster and predictable enough for all our production use cases. Args: filepath (str): The source file path. You can specify additional arguments in the function to allow for specific 'processing' values to be included.

Source code in client/ayon_core/lib/plugin_tools.py
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def source_hash(filepath, *args):
    """Generate simple identifier for a source file.
    This is used to identify whether a source file has previously been
    processe into the pipeline, e.g. a texture.
    The hash is based on source filepath, modification time and file size.
    This is only used to identify whether a specific source file was already
    published before from the same location with the same modification date.
    We opt to do it this way as opposed to Avalanch C4 hash as this is much
    faster and predictable enough for all our production use cases.
    Args:
        filepath (str): The source file path.
    You can specify additional arguments in the function
    to allow for specific 'processing' values to be included.
    """
    # We replace dots with comma because . cannot be a key in a pymongo dict.
    file_name = os.path.basename(filepath)
    time = str(os.path.getmtime(filepath))
    size = str(os.path.getsize(filepath))
    return "|".join([file_name, time, size] + list(args)).replace(".", ",")

version_up(filepath)

Version up filepath to a new non-existing version.

Parses for a version identifier like _v001 or .v001 When no version present _v001 is appended as suffix.

Parameters:

Name Type Description Default
filepath str

full url

required

Returns:

Type Description
str

filepath with increased version number

Source code in client/ayon_core/lib/path_tools.py
 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
def version_up(filepath):
    """Version up filepath to a new non-existing version.

    Parses for a version identifier like `_v001` or `.v001`
    When no version present _v001 is appended as suffix.

    Args:
        filepath (str): full url

    Returns:
        (str): filepath with increased version number

    """
    dirname = os.path.dirname(filepath)
    basename, ext = os.path.splitext(os.path.basename(filepath))

    regex = r"[._]v\d+"
    matches = re.findall(regex, str(basename), re.IGNORECASE)
    if not matches:
        log.info("Creating version...")
        new_label = "_v{version:03d}".format(version=1)
        new_basename = "{}{}".format(basename, new_label)
    else:
        label = matches[-1]
        version = re.search(r"\d+", label).group()
        padding = len(version)

        new_version = int(version) + 1
        new_version = '{version:0{padding}d}'.format(version=new_version,
                                                     padding=padding)
        new_label = label.replace(version, new_version, 1)
        new_basename = _rreplace(basename, label, new_label)
    new_filename = "{}{}".format(new_basename, ext)
    new_filename = os.path.join(dirname, new_filename)
    new_filename = os.path.normpath(new_filename)

    if new_filename == filepath:
        raise RuntimeError("Created path is the same as current file,"
                           "this is a bug")

    # We check for version clashes against the current file for any file
    # that matches completely in name up to the {version} label found. Thus
    # if source file was test_v001_test.txt we want to also check clashes
    # against test_v002.txt but do want to preserve the part after the version
    # label for our new filename
    clash_basename = new_basename
    if not clash_basename.endswith(new_label):
        index = (clash_basename.find(new_label))
        index += len(new_label)
        clash_basename = clash_basename[:index]

    for file in os.listdir(dirname):
        if file.endswith(ext) and file.startswith(clash_basename):
            log.info("Skipping existing version %s" % new_label)
            return version_up(new_filename)

    log.info("New version %s" % new_label)
    return new_filename