Skip to content

ayon_applications

Application

Hold information about application.

Object by itself does nothing special.

Parameters:

Name Type Description Default
data dict

Data for the version containing information about executables, variant label or if is enabled. Only required key is executables.

required
group ApplicationGroup

App group object that created the application and under which application belongs.

required
Source code in client/ayon_applications/defs.py
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
class Application:
    """Hold information about application.

    Object by itself does nothing special.

    Args:
        data (dict): Data for the version containing information about
            executables, variant label or if is enabled.
            Only required key is `executables`.
        group (ApplicationGroup): App group object that created the application
            and under which application belongs.

    """
    def __init__(self, data, group):
        self._data = data
        name = data["name"]
        label = data["label"] or name
        enabled = False
        if group.enabled:
            enabled = data.get("enabled", True)

        if group.label:
            full_label = " ".join((group.label, label))
        else:
            full_label = label
        env = {}
        try:
            env = json.loads(data["environment"])
        except Exception:
            pass

        arguments = data["arguments"]
        if isinstance(arguments, dict):
            arguments = arguments.get(platform.system().lower())

        if not arguments:
            arguments = []

        _executables = data["executables"].get(platform.system().lower(), [])
        executables = [
            ApplicationExecutable(executable)
            for executable in _executables
        ]

        self.group = group

        self.name = name
        self.label = label
        self.enabled = enabled
        self.use_python_2 = data.get("use_python_2", False)

        self.full_name = "/".join((group.name, name))
        self.full_label = full_label
        self.arguments = arguments
        self.executables = executables
        self._environment = env

    def __repr__(self):
        return "<{}> - {}".format(self.__class__.__name__, self.full_name)

    @property
    def environment(self):
        return copy.deepcopy(self._environment)

    @property
    def manager(self):
        return self.group.manager

    @property
    def host_name(self):
        return self.group.host_name

    @property
    def icon(self):
        return self.group.icon

    @property
    def is_host(self):
        return self.group.is_host

    def find_executable(self):
        """Try to find existing executable for application.

        Returns (str): Path to executable from `executables` or None if any
            exists.
        """
        for executable in self.executables:
            if executable.exists():
                return executable
        return None

    def launch(self, *args, **kwargs):
        """Launch the application.

        For this purpose is used manager's launch method to keep logic at one
        place.

        Arguments must match with manager's launch method. That's why *args
        **kwargs are used.

        Returns:
            subprocess.Popen: Return executed process as Popen object.
        """
        return self.manager.launch(self.full_name, *args, **kwargs)

find_executable()

Try to find existing executable for application.

Returns (str): Path to executable from executables or None if any exists.

Source code in client/ayon_applications/defs.py
283
284
285
286
287
288
289
290
291
292
def find_executable(self):
    """Try to find existing executable for application.

    Returns (str): Path to executable from `executables` or None if any
        exists.
    """
    for executable in self.executables:
        if executable.exists():
            return executable
    return None

launch(*args, **kwargs)

Launch the application.

For this purpose is used manager's launch method to keep logic at one place.

Arguments must match with manager's launch method. That's why args *kwargs are used.

Returns:

Type Description

subprocess.Popen: Return executed process as Popen object.

Source code in client/ayon_applications/defs.py
294
295
296
297
298
299
300
301
302
303
304
305
306
def launch(self, *args, **kwargs):
    """Launch the application.

    For this purpose is used manager's launch method to keep logic at one
    place.

    Arguments must match with manager's launch method. That's why *args
    **kwargs are used.

    Returns:
        subprocess.Popen: Return executed process as Popen object.
    """
    return self.manager.launch(self.full_name, *args, **kwargs)

ApplicationExecutable

Representation of executable loaded from settings.

Source code in client/ayon_applications/defs.py
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class ApplicationExecutable:
    """Representation of executable loaded from settings."""

    def __init__(self, executable):
        # Try to format executable with environments
        try:
            executable = executable.format(**os.environ)
        except Exception:
            pass

        # On MacOS check if exists path to executable when ends with `.app`
        # - it is common that path will lead to "/Applications/Blender" but
        #   real path is "/Applications/Blender.app"
        if platform.system().lower() == "darwin":
            executable = self.macos_executable_prep(executable)

        self.executable_path = executable

    def __str__(self):
        return self.executable_path

    def __repr__(self):
        return "<{}> {}".format(self.__class__.__name__, self.executable_path)

    @staticmethod
    def macos_executable_prep(executable):
        """Try to find full path to executable file.

        Real executable is stored in '*.app/Contents/MacOS/<executable>'.

        Having path to '*.app' gives ability to read it's plist info and
        use "CFBundleExecutable" key from plist to know what is "executable."

        Plist is stored in '*.app/Contents/Info.plist'.

        This is because some '*.app' directories don't have same permissions
        as real executable.
        """
        # Try to find if there is `.app` file
        if not os.path.exists(executable):
            _executable = executable + ".app"
            if os.path.exists(_executable):
                executable = _executable

        # Try to find real executable if executable has `Contents` subfolder
        contents_dir = os.path.join(executable, "Contents")
        if os.path.exists(contents_dir):
            executable_filename = None
            # Load plist file and check for bundle executable
            plist_filepath = os.path.join(contents_dir, "Info.plist")
            if os.path.exists(plist_filepath):
                import plistlib

                if hasattr(plistlib, "load"):
                    with open(plist_filepath, "rb") as stream:
                        parsed_plist = plistlib.load(stream)
                else:
                    parsed_plist = plistlib.readPlist(plist_filepath)
                executable_filename = parsed_plist.get("CFBundleExecutable")

            if executable_filename:
                executable = os.path.join(
                    contents_dir, "MacOS", executable_filename
                )

        return executable

    def as_args(self):
        return [self.executable_path]

    def _realpath(self):
        """Check if path is valid executable path."""
        # Check for executable in PATH
        result = find_executable(self.executable_path)
        if result is not None:
            return result

        # This is not 100% validation but it is better than remove ability to
        #   launch .bat, .sh or extentionless files
        if os.path.exists(self.executable_path):
            return self.executable_path
        return None

    def exists(self):
        if not self.executable_path:
            return False
        return bool(self._realpath())

macos_executable_prep(executable) staticmethod

Try to find full path to executable file.

Real executable is stored in '*.app/Contents/MacOS/'.

Having path to '*.app' gives ability to read it's plist info and use "CFBundleExecutable" key from plist to know what is "executable."

Plist is stored in '*.app/Contents/Info.plist'.

This is because some '*.app' directories don't have same permissions as real executable.

Source code in client/ayon_applications/defs.py
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
@staticmethod
def macos_executable_prep(executable):
    """Try to find full path to executable file.

    Real executable is stored in '*.app/Contents/MacOS/<executable>'.

    Having path to '*.app' gives ability to read it's plist info and
    use "CFBundleExecutable" key from plist to know what is "executable."

    Plist is stored in '*.app/Contents/Info.plist'.

    This is because some '*.app' directories don't have same permissions
    as real executable.
    """
    # Try to find if there is `.app` file
    if not os.path.exists(executable):
        _executable = executable + ".app"
        if os.path.exists(_executable):
            executable = _executable

    # Try to find real executable if executable has `Contents` subfolder
    contents_dir = os.path.join(executable, "Contents")
    if os.path.exists(contents_dir):
        executable_filename = None
        # Load plist file and check for bundle executable
        plist_filepath = os.path.join(contents_dir, "Info.plist")
        if os.path.exists(plist_filepath):
            import plistlib

            if hasattr(plistlib, "load"):
                with open(plist_filepath, "rb") as stream:
                    parsed_plist = plistlib.load(stream)
            else:
                parsed_plist = plistlib.readPlist(plist_filepath)
            executable_filename = parsed_plist.get("CFBundleExecutable")

        if executable_filename:
            executable = os.path.join(
                contents_dir, "MacOS", executable_filename
            )

    return executable

ApplicationExecutableNotFound

Bases: Exception

Defined executable paths are not available on the machine.

Source code in client/ayon_applications/exceptions.py
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
class ApplicationExecutableNotFound(Exception):
    """Defined executable paths are not available on the machine."""

    def __init__(self, application):
        self.application = application
        details = None
        if not application.executables:
            msg = (
                "Executable paths for application \"{}\"({}) are not set."
            )
        else:
            msg = (
                "Defined executable paths for application \"{}\"({})"
                " are not available on this machine."
            )
            details = "Defined paths:"
            for executable in application.executables:
                details += "\n- " + executable.executable_path

        self.msg = msg.format(application.full_label, application.full_name)
        self.details = details

        exc_mgs = str(self.msg)
        if details:
            # Is good idea to pass new line symbol to exception message?
            exc_mgs += "\n\n" + details
        self.exc_msg = exc_mgs
        super().__init__(exc_mgs)

ApplicationGroup

Hold information about application group.

Application group wraps different versions(variants) of application. e.g. "maya" is group and "maya_2020" is variant.

Group hold host_name which is implementation name used in AYON. Also holds enabled if whole app group is enabled or icon for application icon path in resources.

Group has also environment which hold same environments for all variants.

Parameters:

Name Type Description Default
name str

Groups' name.

required
data dict

Group defying data loaded from settings.

required
manager ApplicationManager

Manager that created the group.

required
Source code in client/ayon_applications/defs.py
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
class ApplicationGroup:
    """Hold information about application group.

    Application group wraps different versions(variants) of application.
    e.g. "maya" is group and "maya_2020" is variant.

    Group hold `host_name` which is implementation name used in AYON. Also
    holds `enabled` if whole app group is enabled or `icon` for application
    icon path in resources.

    Group has also `environment` which hold same environments for all variants.

    Args:
        name (str): Groups' name.
        data (dict): Group defying data loaded from settings.
        manager (ApplicationManager): Manager that created the group.
    """
    def __init__(self, name, data, manager):
        icon = ICONS_BY_GROUP_NAME.get(name)
        if not icon:
            icon = data.get("icon")

        label = LABELS_BY_GROUP_NAME.get(name)
        if not label:
            label = data.get("label")

        self.name = name
        self.manager = manager
        self._data = data

        self.enabled = data["enabled"]
        self.label = label
        self.icon = icon
        env = {}
        try:
            env = json.loads(data["environment"])
        except Exception:
            pass
        self._environment = env

        host_name = data["host_name"] or None
        self.is_host = host_name is not None
        self.host_name = host_name

        settings_variants = data["variants"]
        variants = {}
        for variant_data in settings_variants:
            app_variant = Application(variant_data, self)
            variants[app_variant.name] = app_variant

        self.variants = variants

    def __repr__(self):
        return "<{}> - {}".format(self.__class__.__name__, self.name)

    def __iter__(self):
        for variant in self.variants.values():
            yield variant

    @property
    def environment(self):
        return copy.deepcopy(self._environment)

ApplicationLaunchContext

Context of launching application.

Main purpose of context is to prepare launch arguments and keyword arguments for new process. Most important part of keyword arguments preparations are environment variables.

During the whole process is possible to use data attribute to store object usable in multiple places.

Launch arguments are strings in list. It is possible to "chain" argument when order of them matters. That is possible to do with adding list where order is right and should not change. NOTE: This is recommendation, not requirement. e.g.: ["nuke.exe", "--NukeX"] -> In this case any part of process may insert argument between nuke.exe and --NukeX. To keep them together it is better to wrap them in another list: [["nuke.exe", "--NukeX"]].

Notes

It is possible to use launch context only to prepare environment variables. In that case executable may be None and can be used 'run_prelaunch_hooks' method to run prelaunch hooks which prepare them.

Parameters:

Name Type Description Default
application Application

Application definition.

required
executable ApplicationExecutable

Object with path to executable.

required
env_group Optional[str]

Environment variable group. If not set 'DEFAULT_ENV_SUBGROUP' is used.

None
launch_type Optional[str]

Launch type. If not set 'local' is used.

None
**data dict

Any additional data. Data may be used during preparation to store objects usable in multiple places.

{}
Source code in client/ayon_applications/manager.py
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
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
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
457
458
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
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
class ApplicationLaunchContext:
    """Context of launching application.

    Main purpose of context is to prepare launch arguments and keyword
    arguments for new process. Most important part of keyword arguments
    preparations are environment variables.

    During the whole process is possible to use `data` attribute to store
    object usable in multiple places.

    Launch arguments are strings in list. It is possible to "chain" argument
    when order of them matters. That is possible to do with adding list where
    order is right and should not change.
    NOTE: This is recommendation, not requirement.
    e.g.: `["nuke.exe", "--NukeX"]` -> In this case any part of process may
    insert argument between `nuke.exe` and `--NukeX`. To keep them together
    it is better to wrap them in another list: `[["nuke.exe", "--NukeX"]]`.

    Notes:
        It is possible to use launch context only to prepare environment
            variables. In that case `executable` may be None and can be used
            'run_prelaunch_hooks' method to run prelaunch hooks which prepare
            them.

    Args:
        application (Application): Application definition.
        executable (ApplicationExecutable): Object with path to executable.
        env_group (Optional[str]): Environment variable group. If not set
            'DEFAULT_ENV_SUBGROUP' is used.
        launch_type (Optional[str]): Launch type. If not set 'local' is used.
        **data (dict): Any additional data. Data may be used during
            preparation to store objects usable in multiple places.
    """

    def __init__(
        self,
        application,
        executable,
        env_group=None,
        launch_type=None,
        **data
    ):
        # Application object
        self.application = application

        self.addons_manager = AddonsManager()

        # Logger
        logger_name = "{}-{}".format(self.__class__.__name__,
                                     self.application.full_name)
        self.log = Logger.get_logger(logger_name)

        self.executable = executable

        if launch_type is None:
            launch_type = LaunchTypes.local
        self.launch_type = launch_type

        if env_group is None:
            env_group = DEFAULT_ENV_SUBGROUP

        self.env_group = env_group

        self.data = dict(data)

        launch_args = []
        if executable is not None:
            launch_args = executable.as_args()
        # subprocess.Popen launch arguments (first argument in constructor)
        self.launch_args = launch_args
        self.launch_args.extend(application.arguments)
        if self.data.get("app_args"):
            self.launch_args.extend(self.data.pop("app_args"))

        # Handle launch environemtns
        src_env = self.data.pop("env", None)
        if src_env is not None and not isinstance(src_env, dict):
            self.log.warning((
                "Passed `env` kwarg has invalid type: {}. Expected: `dict`."
                " Using `os.environ` instead."
            ).format(str(type(src_env))))
            src_env = None

        if src_env is None:
            src_env = os.environ

        ignored_env = {"QT_API", }
        env = {
            key: str(value)
            for key, value in src_env.items()
            if key not in ignored_env
        }
        # subprocess.Popen keyword arguments
        self.kwargs = {"env": env}

        if platform.system().lower() == "windows":
            # Detach new process from currently running process on Windows
            flags = (
                subprocess.CREATE_NEW_PROCESS_GROUP
                | subprocess.DETACHED_PROCESS
            )
            self.kwargs["creationflags"] = flags

        if not sys.stdout:
            self.kwargs["stdout"] = subprocess.DEVNULL
            self.kwargs["stderr"] = subprocess.DEVNULL

        self.prelaunch_hooks = None
        self.postlaunch_hooks = None

        self.process = None
        self._prelaunch_hooks_executed = False

    @property
    def env(self):
        if (
            "env" not in self.kwargs
            or self.kwargs["env"] is None
        ):
            self.kwargs["env"] = {}
        return self.kwargs["env"]

    @env.setter
    def env(self, value):
        if not isinstance(value, dict):
            raise ValueError(
                "'env' attribute expect 'dict' object. Got: {}".format(
                    str(type(value))
                )
            )
        self.kwargs["env"] = value

    @property
    def modules_manager(self):
        """
        Deprecated:
            Use 'addons_manager' instead.

        """
        return self.addons_manager

    def _collect_addons_launch_hook_paths(self):
        """Helper to collect application launch hooks from addons.

        Module have to have implemented 'get_launch_hook_paths' method which
        can expect application as argument or nothing.

        Returns:
            List[str]: Paths to launch hook directories.
        """

        expected_types = (list, tuple, set)

        output = []
        for module in self.addons_manager.get_enabled_addons():
            # Skip module if does not have implemented 'get_launch_hook_paths'
            func = getattr(module, "get_launch_hook_paths", None)
            if func is None:
                continue

            func = module.get_launch_hook_paths
            if hasattr(inspect, "signature"):
                sig = inspect.signature(func)
                expect_args = len(sig.parameters) > 0
            else:
                expect_args = len(inspect.getargspec(func)[0]) > 0

            # Pass application argument if method expect it.
            try:
                if expect_args:
                    hook_paths = func(self.application)
                else:
                    hook_paths = func()
            except Exception:
                self.log.warning(
                    "Failed to call 'get_launch_hook_paths'",
                    exc_info=True
                )
                continue

            if not hook_paths:
                continue

            # Convert string to list
            if isinstance(hook_paths, str):
                hook_paths = [hook_paths]

            # Skip invalid types
            if not isinstance(hook_paths, expected_types):
                self.log.warning((
                    "Result of `get_launch_hook_paths`"
                    " has invalid type {}. Expected {}"
                ).format(type(hook_paths), expected_types))
                continue

            output.extend(hook_paths)
        return output

    def paths_to_launch_hooks(self):
        """Directory paths where to look for launch hooks."""
        # This method has potential to be part of application manager (maybe).
        paths = []

        # TODO load additional studio paths from settings
        global_hooks_dir = os.path.join(AYON_CORE_ROOT, "hooks")

        hooks_dirs = [
            global_hooks_dir
        ]
        if self.host_name:
            # If host requires launch hooks and is module then launch hooks
            #   should be collected using 'collect_launch_hook_paths'
            #   - module have to implement 'get_launch_hook_paths'
            host_module = self.addons_manager.get_host_addon(self.host_name)
            if not host_module:
                hooks_dirs.append(os.path.join(
                    AYON_CORE_ROOT, "hosts", self.host_name, "hooks"
                ))

        for path in hooks_dirs:
            if (
                os.path.exists(path)
                and os.path.isdir(path)
                and path not in paths
            ):
                paths.append(path)

        # Load modules paths
        paths.extend(self._collect_addons_launch_hook_paths())

        return paths

    def discover_launch_hooks(self, force=False):
        """Load and prepare launch hooks."""
        if (
            self.prelaunch_hooks is not None
            or self.postlaunch_hooks is not None
        ):
            if not force:
                self.log.info("Launch hooks were already discovered.")
                return

            self.prelaunch_hooks.clear()
            self.postlaunch_hooks.clear()

        self.log.debug("Discovery of launch hooks started.")

        paths = self.paths_to_launch_hooks()
        self.log.debug("Paths searched for launch hooks:\n{}".format(
            "\n".join("- {}".format(path) for path in paths)
        ))

        all_classes = {
            "pre": [],
            "post": []
        }
        for path in paths:
            if not os.path.exists(path):
                self.log.info(
                    "Path to launch hooks does not exist: \"{}\"".format(path)
                )
                continue

            modules, _crashed = modules_from_path(path)
            for _filepath, module in modules:
                all_classes["pre"].extend(
                    classes_from_module(PreLaunchHook, module)
                )
                all_classes["post"].extend(
                    classes_from_module(PostLaunchHook, module)
                )

        for launch_type, classes in all_classes.items():
            hooks_with_order = []
            hooks_without_order = []
            for klass in classes:
                try:
                    hook = klass(self)
                    if not hook.is_valid:
                        self.log.debug(
                            "Skipped hook invalid for current launch context: "
                            "{}".format(klass.__name__)
                        )
                        continue

                    if inspect.isabstract(hook):
                        self.log.debug("Skipped abstract hook: {}".format(
                            klass.__name__
                        ))
                        continue

                    # Separate hooks by pre/post class
                    if hook.order is None:
                        hooks_without_order.append(hook)
                    else:
                        hooks_with_order.append(hook)

                except Exception:
                    self.log.warning(
                        "Initialization of hook failed: "
                        "{}".format(klass.__name__),
                        exc_info=True
                    )

            # Sort hooks with order by order
            ordered_hooks = list(sorted(
                hooks_with_order, key=lambda obj: obj.order
            ))
            # Extend ordered hooks with hooks without defined order
            ordered_hooks.extend(hooks_without_order)

            if launch_type == "pre":
                self.prelaunch_hooks = ordered_hooks
            else:
                self.postlaunch_hooks = ordered_hooks

        self.log.debug("Found {} prelaunch and {} postlaunch hooks.".format(
            len(self.prelaunch_hooks), len(self.postlaunch_hooks)
        ))

    @property
    def app_name(self):
        return self.application.name

    @property
    def host_name(self):
        return self.application.host_name

    @property
    def app_group(self):
        return self.application.group

    @property
    def manager(self):
        return self.application.manager

    def _run_process(self):
        # Windows and MacOS have easier process start
        low_platform = platform.system().lower()
        if low_platform in ("windows", "darwin"):
            return subprocess.Popen(self.launch_args, **self.kwargs)

        # Linux uses mid process
        # - it is possible that the mid process executable is not
        #   available for this version of AYON in that case use standard
        #   launch
        launch_args = get_linux_launcher_args()
        if launch_args is None:
            return subprocess.Popen(self.launch_args, **self.kwargs)

        # Prepare data that will be passed to midprocess
        # - store arguments to a json and pass path to json as last argument
        # - pass environments to set
        app_env = self.kwargs.pop("env", {})
        json_data = {
            "args": self.launch_args,
            "env": app_env
        }
        if app_env:
            # Filter environments of subprocess
            self.kwargs["env"] = {
                key: value
                for key, value in os.environ.items()
                if key in app_env
            }

        # Create temp file
        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)

        launch_args.append(json_temp_filpath)

        # Create mid-process which will launch application
        process = subprocess.Popen(launch_args, **self.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

    def run_prelaunch_hooks(self):
        """Run prelaunch hooks.

        This method will be executed only once, any future calls will skip
            the processing.
        """

        if self._prelaunch_hooks_executed:
            self.log.warning("Prelaunch hooks were already executed.")
            return
        # Discover launch hooks
        self.discover_launch_hooks()

        # Execute prelaunch hooks
        for prelaunch_hook in self.prelaunch_hooks:
            self.log.debug("Executing prelaunch hook: {}".format(
                str(prelaunch_hook.__class__.__name__)
            ))
            prelaunch_hook.execute()
        self._prelaunch_hooks_executed = True

    def launch(self):
        """Collect data for new process and then create it.

        This method must not be executed more than once.

        Returns:
            subprocess.Popen: Created process as Popen object.
        """
        if self.process is not None:
            self.log.warning("Application was already launched.")
            return

        if not self._prelaunch_hooks_executed:
            self.run_prelaunch_hooks()

        self.log.debug("All prelaunch hook executed. Starting new process.")

        # Prepare subprocess args
        args_len_str = ""
        if isinstance(self.launch_args, str):
            args = self.launch_args
        else:
            args = self.clear_launch_args(self.launch_args)
            args_len_str = " ({})".format(len(args))
        self.log.info(
            "Launching \"{}\" with args{}: {}".format(
                self.application.full_name, args_len_str, args
            )
        )
        self.launch_args = args

        # Run process
        self.process = self._run_process()

        # Process post launch hooks
        for postlaunch_hook in self.postlaunch_hooks:
            self.log.debug("Executing postlaunch hook: {}".format(
                str(postlaunch_hook.__class__.__name__)
            ))

            # TODO how to handle errors?
            # - store to variable to let them accessible?
            try:
                postlaunch_hook.execute()

            except Exception:
                self.log.warning(
                    "After launch procedures were not successful.",
                    exc_info=True
                )

        self.log.debug("Launch of {} finished.".format(
            self.application.full_name
        ))

        return self.process

    @staticmethod
    def clear_launch_args(args):
        """Collect launch arguments to final order.

        Launch argument should be list that may contain another lists this
        function will upack inner lists and keep ordering.

        ```
        # source
        [ [ arg1, [ arg2, arg3 ] ], arg4, [arg5, arg6]]
        # result
        [ arg1, arg2, arg3, arg4, arg5, arg6]

        Args:
            args (list): Source arguments in list may contain inner lists.

        Return:
            list: Unpacked arguments.
        """
        if isinstance(args, str):
            return args
        all_cleared = False
        while not all_cleared:
            all_cleared = True
            new_args = []
            for arg in args:
                if isinstance(arg, (list, tuple, set)):
                    all_cleared = False
                    for _arg in arg:
                        new_args.append(_arg)
                else:
                    new_args.append(arg)
            args = new_args

        return args

modules_manager property

Deprecated

Use 'addons_manager' instead.

clear_launch_args(args) staticmethod

Collect launch arguments to final order.

Launch argument should be list that may contain another lists this function will upack inner lists and keep ordering.

```

source

[ [ arg1, [ arg2, arg3 ] ], arg4, [arg5, arg6]]

result

[ arg1, arg2, arg3, arg4, arg5, arg6]

Args: args (list): Source arguments in list may contain inner lists.

Return: list: Unpacked arguments.

Source code in client/ayon_applications/manager.py
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
@staticmethod
def clear_launch_args(args):
    """Collect launch arguments to final order.

    Launch argument should be list that may contain another lists this
    function will upack inner lists and keep ordering.

    ```
    # source
    [ [ arg1, [ arg2, arg3 ] ], arg4, [arg5, arg6]]
    # result
    [ arg1, arg2, arg3, arg4, arg5, arg6]

    Args:
        args (list): Source arguments in list may contain inner lists.

    Return:
        list: Unpacked arguments.
    """
    if isinstance(args, str):
        return args
    all_cleared = False
    while not all_cleared:
        all_cleared = True
        new_args = []
        for arg in args:
            if isinstance(arg, (list, tuple, set)):
                all_cleared = False
                for _arg in arg:
                    new_args.append(_arg)
            else:
                new_args.append(arg)
        args = new_args

    return args

discover_launch_hooks(force=False)

Load and prepare launch hooks.

Source code in client/ayon_applications/manager.py
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
458
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
def discover_launch_hooks(self, force=False):
    """Load and prepare launch hooks."""
    if (
        self.prelaunch_hooks is not None
        or self.postlaunch_hooks is not None
    ):
        if not force:
            self.log.info("Launch hooks were already discovered.")
            return

        self.prelaunch_hooks.clear()
        self.postlaunch_hooks.clear()

    self.log.debug("Discovery of launch hooks started.")

    paths = self.paths_to_launch_hooks()
    self.log.debug("Paths searched for launch hooks:\n{}".format(
        "\n".join("- {}".format(path) for path in paths)
    ))

    all_classes = {
        "pre": [],
        "post": []
    }
    for path in paths:
        if not os.path.exists(path):
            self.log.info(
                "Path to launch hooks does not exist: \"{}\"".format(path)
            )
            continue

        modules, _crashed = modules_from_path(path)
        for _filepath, module in modules:
            all_classes["pre"].extend(
                classes_from_module(PreLaunchHook, module)
            )
            all_classes["post"].extend(
                classes_from_module(PostLaunchHook, module)
            )

    for launch_type, classes in all_classes.items():
        hooks_with_order = []
        hooks_without_order = []
        for klass in classes:
            try:
                hook = klass(self)
                if not hook.is_valid:
                    self.log.debug(
                        "Skipped hook invalid for current launch context: "
                        "{}".format(klass.__name__)
                    )
                    continue

                if inspect.isabstract(hook):
                    self.log.debug("Skipped abstract hook: {}".format(
                        klass.__name__
                    ))
                    continue

                # Separate hooks by pre/post class
                if hook.order is None:
                    hooks_without_order.append(hook)
                else:
                    hooks_with_order.append(hook)

            except Exception:
                self.log.warning(
                    "Initialization of hook failed: "
                    "{}".format(klass.__name__),
                    exc_info=True
                )

        # Sort hooks with order by order
        ordered_hooks = list(sorted(
            hooks_with_order, key=lambda obj: obj.order
        ))
        # Extend ordered hooks with hooks without defined order
        ordered_hooks.extend(hooks_without_order)

        if launch_type == "pre":
            self.prelaunch_hooks = ordered_hooks
        else:
            self.postlaunch_hooks = ordered_hooks

    self.log.debug("Found {} prelaunch and {} postlaunch hooks.".format(
        len(self.prelaunch_hooks), len(self.postlaunch_hooks)
    ))

launch()

Collect data for new process and then create it.

This method must not be executed more than once.

Returns:

Type Description

subprocess.Popen: Created process as Popen object.

Source code in client/ayon_applications/manager.py
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
def launch(self):
    """Collect data for new process and then create it.

    This method must not be executed more than once.

    Returns:
        subprocess.Popen: Created process as Popen object.
    """
    if self.process is not None:
        self.log.warning("Application was already launched.")
        return

    if not self._prelaunch_hooks_executed:
        self.run_prelaunch_hooks()

    self.log.debug("All prelaunch hook executed. Starting new process.")

    # Prepare subprocess args
    args_len_str = ""
    if isinstance(self.launch_args, str):
        args = self.launch_args
    else:
        args = self.clear_launch_args(self.launch_args)
        args_len_str = " ({})".format(len(args))
    self.log.info(
        "Launching \"{}\" with args{}: {}".format(
            self.application.full_name, args_len_str, args
        )
    )
    self.launch_args = args

    # Run process
    self.process = self._run_process()

    # Process post launch hooks
    for postlaunch_hook in self.postlaunch_hooks:
        self.log.debug("Executing postlaunch hook: {}".format(
            str(postlaunch_hook.__class__.__name__)
        ))

        # TODO how to handle errors?
        # - store to variable to let them accessible?
        try:
            postlaunch_hook.execute()

        except Exception:
            self.log.warning(
                "After launch procedures were not successful.",
                exc_info=True
            )

    self.log.debug("Launch of {} finished.".format(
        self.application.full_name
    ))

    return self.process

paths_to_launch_hooks()

Directory paths where to look for launch hooks.

Source code in client/ayon_applications/manager.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
def paths_to_launch_hooks(self):
    """Directory paths where to look for launch hooks."""
    # This method has potential to be part of application manager (maybe).
    paths = []

    # TODO load additional studio paths from settings
    global_hooks_dir = os.path.join(AYON_CORE_ROOT, "hooks")

    hooks_dirs = [
        global_hooks_dir
    ]
    if self.host_name:
        # If host requires launch hooks and is module then launch hooks
        #   should be collected using 'collect_launch_hook_paths'
        #   - module have to implement 'get_launch_hook_paths'
        host_module = self.addons_manager.get_host_addon(self.host_name)
        if not host_module:
            hooks_dirs.append(os.path.join(
                AYON_CORE_ROOT, "hosts", self.host_name, "hooks"
            ))

    for path in hooks_dirs:
        if (
            os.path.exists(path)
            and os.path.isdir(path)
            and path not in paths
        ):
            paths.append(path)

    # Load modules paths
    paths.extend(self._collect_addons_launch_hook_paths())

    return paths

run_prelaunch_hooks()

Run prelaunch hooks.

This method will be executed only once, any future calls will skip the processing.

Source code in client/ayon_applications/manager.py
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
def run_prelaunch_hooks(self):
    """Run prelaunch hooks.

    This method will be executed only once, any future calls will skip
        the processing.
    """

    if self._prelaunch_hooks_executed:
        self.log.warning("Prelaunch hooks were already executed.")
        return
    # Discover launch hooks
    self.discover_launch_hooks()

    # Execute prelaunch hooks
    for prelaunch_hook in self.prelaunch_hooks:
        self.log.debug("Executing prelaunch hook: {}".format(
            str(prelaunch_hook.__class__.__name__)
        ))
        prelaunch_hook.execute()
    self._prelaunch_hooks_executed = True

ApplicationLaunchFailed

Bases: Exception

Application launch failed due to known reason.

Message should be self explanatory as traceback won't be shown.

Source code in client/ayon_applications/exceptions.py
41
42
43
44
45
46
class ApplicationLaunchFailed(Exception):
    """Application launch failed due to known reason.

    Message should be self explanatory as traceback won't be shown.
    """
    pass

ApplicationManager

Load applications and tools and store them by their full name.

Parameters:

Name Type Description Default
studio_settings dict

Preloaded studio settings. When passed manager will always use these values. Gives ability to create manager using different settings.

None
Source code in client/ayon_applications/manager.py
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class ApplicationManager:
    """Load applications and tools and store them by their full name.

    Args:
        studio_settings (dict): Preloaded studio settings. When passed manager
            will always use these values. Gives ability to create manager
            using different settings.
    """

    def __init__(self, studio_settings=None):
        self.log = Logger.get_logger(self.__class__.__name__)

        self.app_groups = {}
        self.applications = {}
        self.tool_groups = {}
        self.tools = {}

        self._studio_settings = studio_settings

        self.refresh()

    def set_studio_settings(self, studio_settings):
        """Ability to change init system settings.

        This will trigger refresh of manager.
        """
        self._studio_settings = studio_settings

        self.refresh()

    def refresh(self):
        """Refresh applications from settings."""
        self.app_groups.clear()
        self.applications.clear()
        self.tool_groups.clear()
        self.tools.clear()

        if self._studio_settings is not None:
            settings = copy.deepcopy(self._studio_settings)
        else:
            settings = get_studio_settings(
                clear_metadata=False, exclude_locals=False
            )

        applications_addon_settings = settings["applications"]

        # Prepare known applications
        app_defs = applications_addon_settings["applications"]
        additional_apps = app_defs.pop("additional_apps")
        for additional_app in additional_apps:
            app_name = additional_app.pop("name")
            if app_name in app_defs:
                self.log.warning((
                    "Additional application '{}' is already"
                    " in built-in applications."
                ).format(app_name))
            app_defs[app_name] = additional_app

        for group_name, variant_defs in app_defs.items():
            group = ApplicationGroup(group_name, variant_defs, self)
            self.app_groups[group_name] = group
            for app in group:
                self.applications[app.full_name] = app

        tools_definitions = applications_addon_settings["tool_groups"]
        for tool_group_data in tools_definitions:
            group = EnvironmentToolGroup(tool_group_data, self)
            self.tool_groups[group.name] = group
            for tool in group:
                self.tools[tool.full_name] = tool

    def find_latest_available_variant_for_group(self, group_name):
        group = self.app_groups.get(group_name)
        if group is None or not group.enabled:
            return None

        output = None
        for _, variant in reversed(sorted(group.variants.items())):
            executable = variant.find_executable()
            if executable:
                output = variant
                break
        return output

    def create_launch_context(self, app_name, **data):
        """Prepare launch context for application.

        Args:
            app_name (str): Name of application that should be launched.
            **data (Any): Any additional data. Data may be used during

        Returns:
            ApplicationLaunchContext: Launch context for application.

        Raises:
            ApplicationNotFound: Application was not found by entered name.
        """

        app = self.applications.get(app_name)
        if not app:
            raise ApplicationNotFound(app_name)

        executable = app.find_executable()

        return ApplicationLaunchContext(
            app, executable, **data
        )

    def launch_with_context(self, launch_context):
        """Launch application using existing launch context.

        Args:
            launch_context (ApplicationLaunchContext): Prepared launch
                context.
        """

        if not launch_context.executable:
            raise ApplicationExecutableNotFound(launch_context.application)
        return launch_context.launch()

    def launch(self, app_name, **data):
        """Launch procedure.

        For host application it's expected to contain "project_name",
        "folder_path" and "task_name".

        Args:
            app_name (str): Name of application that should be launched.
            **data (Any): Any additional data. Data may be used during
                preparation to store objects usable in multiple places.

        Raises:
            ApplicationNotFound: Application was not found by entered
                argument `app_name`.
            ApplicationExecutableNotFound: Executables in application definition
                were not found on this machine.
            ApplicationLaunchFailed: Something important for application launch
                failed. Exception should contain explanation message,
                traceback should not be needed.
        """

        context = self.create_launch_context(app_name, **data)
        return self.launch_with_context(context)

create_launch_context(app_name, **data)

Prepare launch context for application.

Parameters:

Name Type Description Default
app_name str

Name of application that should be launched.

required
**data Any

Any additional data. Data may be used during

{}

Returns:

Name Type Description
ApplicationLaunchContext

Launch context for application.

Raises:

Type Description
ApplicationNotFound

Application was not found by entered name.

Source code in client/ayon_applications/manager.py
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def create_launch_context(self, app_name, **data):
    """Prepare launch context for application.

    Args:
        app_name (str): Name of application that should be launched.
        **data (Any): Any additional data. Data may be used during

    Returns:
        ApplicationLaunchContext: Launch context for application.

    Raises:
        ApplicationNotFound: Application was not found by entered name.
    """

    app = self.applications.get(app_name)
    if not app:
        raise ApplicationNotFound(app_name)

    executable = app.find_executable()

    return ApplicationLaunchContext(
        app, executable, **data
    )

launch(app_name, **data)

Launch procedure.

For host application it's expected to contain "project_name", "folder_path" and "task_name".

Parameters:

Name Type Description Default
app_name str

Name of application that should be launched.

required
**data Any

Any additional data. Data may be used during preparation to store objects usable in multiple places.

{}

Raises:

Type Description
ApplicationNotFound

Application was not found by entered argument app_name.

ApplicationExecutableNotFound

Executables in application definition were not found on this machine.

ApplicationLaunchFailed

Something important for application launch failed. Exception should contain explanation message, traceback should not be needed.

Source code in client/ayon_applications/manager.py
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def launch(self, app_name, **data):
    """Launch procedure.

    For host application it's expected to contain "project_name",
    "folder_path" and "task_name".

    Args:
        app_name (str): Name of application that should be launched.
        **data (Any): Any additional data. Data may be used during
            preparation to store objects usable in multiple places.

    Raises:
        ApplicationNotFound: Application was not found by entered
            argument `app_name`.
        ApplicationExecutableNotFound: Executables in application definition
            were not found on this machine.
        ApplicationLaunchFailed: Something important for application launch
            failed. Exception should contain explanation message,
            traceback should not be needed.
    """

    context = self.create_launch_context(app_name, **data)
    return self.launch_with_context(context)

launch_with_context(launch_context)

Launch application using existing launch context.

Parameters:

Name Type Description Default
launch_context ApplicationLaunchContext

Prepared launch context.

required
Source code in client/ayon_applications/manager.py
137
138
139
140
141
142
143
144
145
146
147
def launch_with_context(self, launch_context):
    """Launch application using existing launch context.

    Args:
        launch_context (ApplicationLaunchContext): Prepared launch
            context.
    """

    if not launch_context.executable:
        raise ApplicationExecutableNotFound(launch_context.application)
    return launch_context.launch()

refresh()

Refresh applications from settings.

Source code in client/ayon_applications/manager.py
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
def refresh(self):
    """Refresh applications from settings."""
    self.app_groups.clear()
    self.applications.clear()
    self.tool_groups.clear()
    self.tools.clear()

    if self._studio_settings is not None:
        settings = copy.deepcopy(self._studio_settings)
    else:
        settings = get_studio_settings(
            clear_metadata=False, exclude_locals=False
        )

    applications_addon_settings = settings["applications"]

    # Prepare known applications
    app_defs = applications_addon_settings["applications"]
    additional_apps = app_defs.pop("additional_apps")
    for additional_app in additional_apps:
        app_name = additional_app.pop("name")
        if app_name in app_defs:
            self.log.warning((
                "Additional application '{}' is already"
                " in built-in applications."
            ).format(app_name))
        app_defs[app_name] = additional_app

    for group_name, variant_defs in app_defs.items():
        group = ApplicationGroup(group_name, variant_defs, self)
        self.app_groups[group_name] = group
        for app in group:
            self.applications[app.full_name] = app

    tools_definitions = applications_addon_settings["tool_groups"]
    for tool_group_data in tools_definitions:
        group = EnvironmentToolGroup(tool_group_data, self)
        self.tool_groups[group.name] = group
        for tool in group:
            self.tools[tool.full_name] = tool

set_studio_settings(studio_settings)

Ability to change init system settings.

This will trigger refresh of manager.

Source code in client/ayon_applications/manager.py
50
51
52
53
54
55
56
57
def set_studio_settings(self, studio_settings):
    """Ability to change init system settings.

    This will trigger refresh of manager.
    """
    self._studio_settings = studio_settings

    self.refresh()

ApplicationNotFound

Bases: Exception

Application was not found in ApplicationManager by name.

Source code in client/ayon_applications/exceptions.py
1
2
3
4
5
6
7
8
class ApplicationNotFound(Exception):
    """Application was not found in ApplicationManager by name."""

    def __init__(self, app_name):
        self.app_name = app_name
        super().__init__(
            "Application \"{}\" was not found.".format(app_name)
        )

ApplicationsAddon

Bases: AYONAddon, IPluginPaths

Source code in client/ayon_applications/addon.py
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
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
class ApplicationsAddon(AYONAddon, IPluginPaths):
    name = "applications"
    version = __version__

    def initialize(self, settings):
        # TODO remove when addon is removed from ayon-core
        self.enabled = self.name in settings

    def get_app_environments_for_context(
        self,
        project_name,
        folder_path,
        task_name,
        full_app_name,
        env_group=None,
        launch_type=None,
        env=None,
    ):
        """Calculate environment variables for launch context.

        Args:
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.
            full_app_name (str): Full application name.
            env_group (Optional[str]): Environment group.
            launch_type (Optional[str]): Launch type.
            env (Optional[dict[str, str]]): Environment variables to update.

        Returns:
            dict[str, str]: Environment variables for context.

        """
        from ayon_applications.utils import get_app_environments_for_context

        if not full_app_name:
            return {}

        return get_app_environments_for_context(
            project_name,
            folder_path,
            task_name,
            full_app_name,
            env_group=env_group,
            launch_type=launch_type,
            env=env,
            addons_manager=self.manager
        )

    def get_farm_publish_environment_variables(
        self,
        project_name,
        folder_path,
        task_name,
        full_app_name=None,
        env_group=None,
    ):
        """Calculate environment variables for farm publish.

        Args:
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.
            env_group (Optional[str]): Environment group.
            full_app_name (Optional[str]): Full application name. Value from
                environment variable 'AYON_APP_NAME' is used if 'None' is
                passed.

        Returns:
            dict[str, str]: Environment variables for farm publish.

        """
        if full_app_name is None:
            full_app_name = os.getenv("AYON_APP_NAME")

        return self.get_app_environments_for_context(
            project_name,
            folder_path,
            task_name,
            full_app_name,
            env_group=env_group,
            launch_type=LaunchTypes.farm_publish
        )

    def get_applications_manager(self, settings=None):
        """Get applications manager.

        Args:
            settings (Optional[dict]): Studio/project settings.

        Returns:
            ApplicationManager: Applications manager.

        """
        return ApplicationManager(settings)

    def get_plugin_paths(self):
        plugins_dir = os.path.join(APPLICATIONS_ADDON_ROOT, "plugins")
        return {
            "actions": [os.path.join(plugins_dir, "launcher_actions")],
            "publish": [os.path.join(plugins_dir, "publish")]
        }

    def get_launch_hook_paths(self, app):
        return [
            os.path.join(APPLICATIONS_ADDON_ROOT, "hooks")
        ]

    def get_app_icon_path(self, icon_filename):
        """Get icon path.

        Args:
            icon_filename (str): Icon filename.

        Returns:
            Union[str, None]: Icon path or None if not found.

        """
        return get_app_icon_path(icon_filename)

    def get_app_icon_url(self, icon_filename, server=False):
        """Get icon path.

        Method does not validate if icon filename exist on server.

        Args:
            icon_filename (str): Icon name.
            server (Optional[bool]): Return url to AYON server.

        Returns:
            Union[str, None]: Icon path or None is server url is not
                available.

        """
        if not icon_filename:
            return None
        icon_name = os.path.basename(icon_filename)
        if server:
            base_url = ayon_api.get_base_url()
            return (
                f"{base_url}/addons/{self.name}/{self.version}"
                f"/public/icons/{icon_name}"
            )
        server_url = os.getenv("AYON_WEBSERVER_URL")
        if not server_url:
            return None
        return "/".join([
            server_url, "addons", self.name, "icons", icon_name
        ])

    def get_applications_action_classes(self):
        """Get application action classes for launcher tool.

        This method should be used only by launcher tool. Please do not use it
        in other places as its implementation is not optimal, and might
        change or be removed.

        Returns:
            list[ApplicationAction]: List of application action classes.

        """
        from .action import ApplicationAction

        actions = []

        manager = self.get_applications_manager()
        for full_name, application in manager.applications.items():
            if not application.enabled:
                continue

            icon = self.get_app_icon_path(application.icon)

            action = type(
                "app_{}".format(full_name),
                (ApplicationAction,),
                {
                    "identifier": "application.{}".format(full_name),
                    "application": application,
                    "name": application.name,
                    "label": application.group.label,
                    "label_variant": application.label,
                    "group": None,
                    "icon": icon,
                    "color": getattr(application, "color", None),
                    "order": getattr(application, "order", None) or 0,
                    "data": {}
                }
            )
            actions.append(action)
        return actions

    def launch_application(
        self, app_name, project_name, folder_path, task_name
    ):
        """Launch application.

        Args:
            app_name (str): Full application name e.g. 'maya/2024'.
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.

        """
        ensure_addons_are_process_ready(
            addon_name=self.name,
            addon_version=self.version,
            project_name=project_name,
        )
        headless = is_headless_mode_enabled()

        # TODO handle raise errors
        failed = True
        message = None
        detail = None
        try:
            app_manager = self.get_applications_manager()
            app_manager.launch(
                app_name,
                project_name=project_name,
                folder_path=folder_path,
                task_name=task_name,
            )
            failed = False

        except (
            ApplicationLaunchFailed,
            ApplicationExecutableNotFound,
            ApplicationNotFound,
        ) as exc:
            message = str(exc)
            self.log.warning(f"Application launch failed: {message}")

        except Exception as exc:
            message = "An unexpected error happened"
            detail = "".join(traceback.format_exception(*sys.exc_info()))
            self.log.warning(
                f"Application launch failed: {str(exc)}",
                exc_info=True
            )

        if not failed:
            return

        if not headless:
            self._show_launch_error_dialog(message, detail)
        sys.exit(1)

    def webserver_initialization(self, manager):
        """Initialize webserver.

        Args:
            manager (WebServerManager): Webserver manager.

        """
        static_prefix = f"/addons/{self.name}/icons"
        manager.add_static(
            static_prefix, os.path.join(APPLICATIONS_ADDON_ROOT, "icons")
        )

    # --- CLI ---
    def cli(self, addon_click_group):
        main_group = click_wrap.group(
            self._cli_main, name=self.name, help="Applications addon"
        )
        (
            main_group.command(
                self._cli_extract_environments,
                name="extractenvironments",
                help=(
                    "Extract environment variables for context into json file"
                )
            )
            .argument("output_json_path")
            .option("--project", help="Project name", default=None)
            .option("--folder", help="Folder path", default=None)
            .option("--task", help="Task name", default=None)
            .option("--app", help="Full application name", default=None)
            .option(
                "--envgroup",
                help="Environment group (e.g. \"farm\")",
                default=None
            )
        )
        (
            main_group.command(
                self._cli_launch_context_names,
                name="launch",
                help="Launch application"
            )
            .option("--app", required=True, help="Full application name")
            .option("--project", required=True, help="Project name")
            .option("--folder", required=True, help="Folder path")
            .option("--task", required=True, help="Task name")
        )
        # Convert main command to click object and add it to parent group
        (
            main_group.command(
                self._cli_launch_with_task_id,
                name="launch-by-id",
                help="Launch application"
            )
            .option("--app", required=True, help="Full application name")
            .option("--project", required=True, help="Project name")
            .option("--task-id", required=True, help="Task id")
        )
        # Convert main command to click object and add it to parent group
        addon_click_group.add_command(
            main_group.to_click_obj()
        )

    def _cli_main(self):
        pass

    def _cli_extract_environments(
        self, output_json_path, project, folder, task, app, envgroup
    ):
        """Produces json file with environment based on project and app.

        Called by farm integration to propagate environment into farm jobs.

        Args:
            output_json_path (str): Output json file path.
            project (str): Project name.
            folder (str): Folder path.
            task (str): Task name.
            app (str): Full application name e.g. 'maya/2024'.
            envgroup (str): Environment group.

        """
        if all((project, folder, task, app)):
            env = self.get_farm_publish_environment_variables(
                project, folder, task, app, env_group=envgroup,
            )
        else:
            env = os.environ.copy()

        output_dir = os.path.dirname(output_json_path)
        os.makedirs(output_dir, exist_ok=True)

        with open(output_json_path, "w") as file_stream:
            json.dump(env, file_stream, indent=4)

    def _cli_launch_context_names(self, project, folder, task, app):
        """Launch application.

        Args:
            project (str): Project name.
            folder (str): Folder path.
            task (str): Task name.
            app (str): Full application name e.g. 'maya/2024'.

        """
        self.launch_application(app, project, folder, task)


    def _cli_launch_with_task_id(self, project, task_id, app):
        """Launch application.

        Args:
            project (str): Project name.
            task_id (str): Task id.
            app (str): Full application name e.g. 'maya/2024'.

        """
        task_entity = ayon_api.get_task_by_id(
            project, task_id, fields={"name", "folderId"}
        )
        folder_entity = ayon_api.get_folder_by_id(
            project, task_entity["folderId"], fields={"path"}
        )
        self.launch_application(
            app, project, folder_entity["path"], task_entity["name"]
        )

    def _show_launch_error_dialog(self, message, detail):
        script_path = os.path.join(
            APPLICATIONS_ADDON_ROOT, "ui", "launch_failed_dialog.py"
        )
        with tempfile.NamedTemporaryFile("w", delete=False) as tmp:
            tmp_path = tmp.name
            json.dump(
                {"message": message, "detail": detail},
                tmp.file
            )

        try:
            run_ayon_launcher_process(
                "--skip-bootstrap",
                script_path,
                tmp_path,
                add_sys_paths=True,
                creationflags=0,
            )

        finally:
            os.remove(tmp_path)

get_app_environments_for_context(project_name, folder_path, task_name, full_app_name, env_group=None, launch_type=None, env=None)

Calculate environment variables for launch context.

Parameters:

Name Type Description Default
project_name str

Project name.

required
folder_path str

Folder path.

required
task_name str

Task name.

required
full_app_name str

Full application name.

required
env_group Optional[str]

Environment group.

None
launch_type Optional[str]

Launch type.

None
env Optional[dict[str, str]]

Environment variables to update.

None

Returns:

Type Description

dict[str, str]: Environment variables for context.

Source code in client/ayon_applications/addon.py
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
def get_app_environments_for_context(
    self,
    project_name,
    folder_path,
    task_name,
    full_app_name,
    env_group=None,
    launch_type=None,
    env=None,
):
    """Calculate environment variables for launch context.

    Args:
        project_name (str): Project name.
        folder_path (str): Folder path.
        task_name (str): Task name.
        full_app_name (str): Full application name.
        env_group (Optional[str]): Environment group.
        launch_type (Optional[str]): Launch type.
        env (Optional[dict[str, str]]): Environment variables to update.

    Returns:
        dict[str, str]: Environment variables for context.

    """
    from ayon_applications.utils import get_app_environments_for_context

    if not full_app_name:
        return {}

    return get_app_environments_for_context(
        project_name,
        folder_path,
        task_name,
        full_app_name,
        env_group=env_group,
        launch_type=launch_type,
        env=env,
        addons_manager=self.manager
    )

get_app_icon_path(icon_filename)

Get icon path.

Parameters:

Name Type Description Default
icon_filename str

Icon filename.

required

Returns:

Type Description

Union[str, None]: Icon path or None if not found.

Source code in client/ayon_applications/addon.py
140
141
142
143
144
145
146
147
148
149
150
def get_app_icon_path(self, icon_filename):
    """Get icon path.

    Args:
        icon_filename (str): Icon filename.

    Returns:
        Union[str, None]: Icon path or None if not found.

    """
    return get_app_icon_path(icon_filename)

get_app_icon_url(icon_filename, server=False)

Get icon path.

Method does not validate if icon filename exist on server.

Parameters:

Name Type Description Default
icon_filename str

Icon name.

required
server Optional[bool]

Return url to AYON server.

False

Returns:

Type Description

Union[str, None]: Icon path or None is server url is not available.

Source code in client/ayon_applications/addon.py
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
def get_app_icon_url(self, icon_filename, server=False):
    """Get icon path.

    Method does not validate if icon filename exist on server.

    Args:
        icon_filename (str): Icon name.
        server (Optional[bool]): Return url to AYON server.

    Returns:
        Union[str, None]: Icon path or None is server url is not
            available.

    """
    if not icon_filename:
        return None
    icon_name = os.path.basename(icon_filename)
    if server:
        base_url = ayon_api.get_base_url()
        return (
            f"{base_url}/addons/{self.name}/{self.version}"
            f"/public/icons/{icon_name}"
        )
    server_url = os.getenv("AYON_WEBSERVER_URL")
    if not server_url:
        return None
    return "/".join([
        server_url, "addons", self.name, "icons", icon_name
    ])

get_applications_action_classes()

Get application action classes for launcher tool.

This method should be used only by launcher tool. Please do not use it in other places as its implementation is not optimal, and might change or be removed.

Returns:

Type Description

list[ApplicationAction]: List of application action classes.

Source code in client/ayon_applications/addon.py
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
def get_applications_action_classes(self):
    """Get application action classes for launcher tool.

    This method should be used only by launcher tool. Please do not use it
    in other places as its implementation is not optimal, and might
    change or be removed.

    Returns:
        list[ApplicationAction]: List of application action classes.

    """
    from .action import ApplicationAction

    actions = []

    manager = self.get_applications_manager()
    for full_name, application in manager.applications.items():
        if not application.enabled:
            continue

        icon = self.get_app_icon_path(application.icon)

        action = type(
            "app_{}".format(full_name),
            (ApplicationAction,),
            {
                "identifier": "application.{}".format(full_name),
                "application": application,
                "name": application.name,
                "label": application.group.label,
                "label_variant": application.label,
                "group": None,
                "icon": icon,
                "color": getattr(application, "color", None),
                "order": getattr(application, "order", None) or 0,
                "data": {}
            }
        )
        actions.append(action)
    return actions

get_applications_manager(settings=None)

Get applications manager.

Parameters:

Name Type Description Default
settings Optional[dict]

Studio/project settings.

None

Returns:

Name Type Description
ApplicationManager

Applications manager.

Source code in client/ayon_applications/addon.py
116
117
118
119
120
121
122
123
124
125
126
def get_applications_manager(self, settings=None):
    """Get applications manager.

    Args:
        settings (Optional[dict]): Studio/project settings.

    Returns:
        ApplicationManager: Applications manager.

    """
    return ApplicationManager(settings)

get_farm_publish_environment_variables(project_name, folder_path, task_name, full_app_name=None, env_group=None)

Calculate environment variables for farm publish.

Parameters:

Name Type Description Default
project_name str

Project name.

required
folder_path str

Folder path.

required
task_name str

Task name.

required
env_group Optional[str]

Environment group.

None
full_app_name Optional[str]

Full application name. Value from environment variable 'AYON_APP_NAME' is used if 'None' is passed.

None

Returns:

Type Description

dict[str, str]: Environment variables for farm publish.

Source code in client/ayon_applications/addon.py
 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
def get_farm_publish_environment_variables(
    self,
    project_name,
    folder_path,
    task_name,
    full_app_name=None,
    env_group=None,
):
    """Calculate environment variables for farm publish.

    Args:
        project_name (str): Project name.
        folder_path (str): Folder path.
        task_name (str): Task name.
        env_group (Optional[str]): Environment group.
        full_app_name (Optional[str]): Full application name. Value from
            environment variable 'AYON_APP_NAME' is used if 'None' is
            passed.

    Returns:
        dict[str, str]: Environment variables for farm publish.

    """
    if full_app_name is None:
        full_app_name = os.getenv("AYON_APP_NAME")

    return self.get_app_environments_for_context(
        project_name,
        folder_path,
        task_name,
        full_app_name,
        env_group=env_group,
        launch_type=LaunchTypes.farm_publish
    )

launch_application(app_name, project_name, folder_path, task_name)

Launch application.

Parameters:

Name Type Description Default
app_name str

Full application name e.g. 'maya/2024'.

required
project_name str

Project name.

required
folder_path str

Folder path.

required
task_name str

Task name.

required
Source code in client/ayon_applications/addon.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
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
def launch_application(
    self, app_name, project_name, folder_path, task_name
):
    """Launch application.

    Args:
        app_name (str): Full application name e.g. 'maya/2024'.
        project_name (str): Project name.
        folder_path (str): Folder path.
        task_name (str): Task name.

    """
    ensure_addons_are_process_ready(
        addon_name=self.name,
        addon_version=self.version,
        project_name=project_name,
    )
    headless = is_headless_mode_enabled()

    # TODO handle raise errors
    failed = True
    message = None
    detail = None
    try:
        app_manager = self.get_applications_manager()
        app_manager.launch(
            app_name,
            project_name=project_name,
            folder_path=folder_path,
            task_name=task_name,
        )
        failed = False

    except (
        ApplicationLaunchFailed,
        ApplicationExecutableNotFound,
        ApplicationNotFound,
    ) as exc:
        message = str(exc)
        self.log.warning(f"Application launch failed: {message}")

    except Exception as exc:
        message = "An unexpected error happened"
        detail = "".join(traceback.format_exception(*sys.exc_info()))
        self.log.warning(
            f"Application launch failed: {str(exc)}",
            exc_info=True
        )

    if not failed:
        return

    if not headless:
        self._show_launch_error_dialog(message, detail)
    sys.exit(1)

webserver_initialization(manager)

Initialize webserver.

Parameters:

Name Type Description Default
manager WebServerManager

Webserver manager.

required
Source code in client/ayon_applications/addon.py
279
280
281
282
283
284
285
286
287
288
289
def webserver_initialization(self, manager):
    """Initialize webserver.

    Args:
        manager (WebServerManager): Webserver manager.

    """
    static_prefix = f"/addons/{self.name}/icons"
    manager.add_static(
        static_prefix, os.path.join(APPLICATIONS_ADDON_ROOT, "icons")
    )

EnvironmentTool

Hold information about application tool.

Structure of tool information.

Parameters:

Name Type Description Default
variant_data dict

Variant data with environments and host and app variant filters.

required
group EnvironmentToolGroup

Name of group which wraps tool.

required
Source code in client/ayon_applications/defs.py
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
class EnvironmentTool:
    """Hold information about application tool.

    Structure of tool information.

    Args:
        variant_data (dict): Variant data with environments and
            host and app variant filters.
        group (EnvironmentToolGroup): Name of group which wraps tool.
    """

    def __init__(self, variant_data, group):
        # Backwards compatibility 3.9.1 - 3.9.2
        # - 'variant_data' contained only environments but contain also host
        #   and application variant filters
        name = variant_data["name"]
        label = variant_data["label"]
        host_names = variant_data["host_names"]
        app_variants = variant_data["app_variants"]

        environment = {}
        try:
            environment = json.loads(variant_data["environment"])
        except Exception:
            pass

        self.host_names = host_names
        self.app_variants = app_variants
        self.name = name
        self.variant_label = label
        self.label = " ".join((group.label, label))
        self.group = group

        self._environment = environment
        self.full_name = "/".join((group.name, name))

    def __repr__(self):
        return "<{}> - {}".format(self.__class__.__name__, self.full_name)

    @property
    def environment(self):
        return copy.deepcopy(self._environment)

    def is_valid_for_app(self, app):
        """Is tool valid for application.

        Args:
            app (Application): Application for which are prepared environments.
        """
        if self.app_variants and app.full_name not in self.app_variants:
            return False

        if self.host_names and app.host_name not in self.host_names:
            return False
        return True

is_valid_for_app(app)

Is tool valid for application.

Parameters:

Name Type Description Default
app Application

Application for which are prepared environments.

required
Source code in client/ayon_applications/defs.py
401
402
403
404
405
406
407
408
409
410
411
412
def is_valid_for_app(self, app):
    """Is tool valid for application.

    Args:
        app (Application): Application for which are prepared environments.
    """
    if self.app_variants and app.full_name not in self.app_variants:
        return False

    if self.host_names and app.host_name not in self.host_names:
        return False
    return True

EnvironmentToolGroup

Hold information about environment tool group.

Environment tool group may hold different variants of same tool and set environments that are same for all of them.

e.g. "mtoa" may have different versions but all environments except one are same.

Parameters:

Name Type Description Default
data dict

Group information with variants.

required
manager ApplicationManager

Manager that creates the group.

required
Source code in client/ayon_applications/defs.py
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
class EnvironmentToolGroup:
    """Hold information about environment tool group.

    Environment tool group may hold different variants of same tool and set
    environments that are same for all of them.

    e.g. "mtoa" may have different versions but all environments except one
        are same.

    Args:
        data (dict): Group information with variants.
        manager (ApplicationManager): Manager that creates the group.
    """

    def __init__(self, data, manager):
        name = data["name"]
        label = data["label"]

        self.name = name
        self.label = label
        self._data = data
        self.manager = manager

        environment = {}
        try:
            environment = json.loads(data["environment"])
        except Exception:
            pass
        self._environment = environment

        variants = data.get("variants") or []
        variants_by_name = {}
        for variant_data in variants:
            tool = EnvironmentTool(variant_data, self)
            variants_by_name[tool.name] = tool
        self.variants = variants_by_name

    def __repr__(self):
        return "<{}> - {}".format(self.__class__.__name__, self.name)

    def __iter__(self):
        for variant in self.variants.values():
            yield variant

    @property
    def environment(self):
        return copy.deepcopy(self._environment)

LaunchTypes

Launch types are filters for pre/post-launch hooks.

Please use these variables in case they'll change values.

Source code in client/ayon_applications/defs.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class LaunchTypes:
    """Launch types are filters for pre/post-launch hooks.

    Please use these variables in case they'll change values.
    """

    # Local launch - application is launched on local machine
    local = "local"
    # Farm render job - application is on farm
    farm_render = "farm-render"
    # Farm publish job - integration post-render job
    farm_publish = "farm-publish"
    # Remote launch - application is launched on remote machine from which
    #     can be started publishing
    remote = "remote"
    # Automated launch - application is launched with automated publishing
    automated = "automated"

UndefinedApplicationExecutable

Bases: ApplicationExecutable

Some applications do not require executable path from settings.

In that case this class is used to "fake" existing executable.

Source code in client/ayon_applications/defs.py
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class UndefinedApplicationExecutable(ApplicationExecutable):
    """Some applications do not require executable path from settings.

    In that case this class is used to "fake" existing executable.
    """
    def __init__(self):
        pass

    def __str__(self):
        return self.__class__.__name__

    def __repr__(self):
        return "<{}>".format(self.__class__.__name__)

    def as_args(self):
        return []

    def exists(self):
        return True