ayon_applications

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: dict[str, Any], group: ApplicationGroup):
        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 = f"{group.name}/{name}"
        self.full_label = full_label
        self.arguments = arguments
        self.executables = executables
        self._environment = env
        self.redirect_output = data.get("redirect_output", True)

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

    @property
    def environment(self) -> dict[str, str]:
        return copy.deepcopy(self._environment)

    @property
    def manager(self) -> "ApplicationManager":
        return self.group.manager

    @property
    def host_name(self) -> Optional[str]:
        return self.group.host_name

    @property
    def icon(self) -> dict[str, str] | None:
        return self.group.icon

    @property
    def is_host(self) -> bool:
        return self.group.is_host

    def find_executable(self) -> Optional[ApplicationExecutable]:
        """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) -> Optional[subprocess.Popen]:
        """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)

class ApplicationExecutable:
    """Representation of executable loaded from settings."""

    def __init__(self, executable: str):
        # 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) -> str:
        return self.executable_path

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

    @staticmethod
    def macos_executable_prep(executable: str) -> str:
        """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) -> list[str]:
        return [self.executable_path]

    def _realpath(self) -> Optional[str]:
        """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.isfile(self.executable_path):
            return self.executable_path
        return None

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

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 valid or 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)

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: str,
        data: dict[str, Any],
        manager: "ApplicationManager",
    ):
        icon = manager.get_app_icon(name)

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

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

        self.enabled = data["enabled"]
        self.label = label
        self.icon: dict[str, str] | None = 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) -> str:
        return f"<{self.__class__.__name__}> - {self.name}"

    def __iter__(self) -> Generator["Application", None, None]:
        for variant in self.variants.values():
            yield variant

    @property
    def environment(self) -> dict[str, str]:
        return copy.deepcopy(self._environment)

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: Application,
        executable: ApplicationExecutable,
        env_group: Optional[str] = None,
        launch_type: Optional[str] = None,
        **data,
    ):
        from .process import ProcessManager

        # Application object
        self.application: Application = application

        self.addons_manager: AddonsManager = AddonsManager()
        self.process_manager: ProcessManager = ProcessManager()
        self.redirect_output: bool = application.redirect_output

        # Logger
        self.log: logging.Logger = Logger.get_logger(
            f"{self.__class__.__name__}-{application.full_name}"
        )

        self.executable: ApplicationExecutable = executable

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

        if env_group is None:
            env_group = DEFAULT_ENV_SUBGROUP

        self.env_group: str = env_group

        self.data: dict[str, Any] = dict(data)
        project_name = self.data.get("project_name")
        if project_name and "project_settings" not in self.data:
            self.data["project_settings"] = get_project_settings(
                project_name
            )

        launch_args = []
        if executable is not None:
            launch_args = executable.as_args()
        # subprocess.Popen launch arguments (first argument in constructor)
        self.launch_args: list[str] = 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(
                f"Passed `env` kwarg has invalid type: {type(src_env)}."
                " Expected: `dict`. Using `os.environ` instead."
            )
            src_env = None

        if src_env is None:
            src_env = os.environ

        ignored_env = {"QT_API", "PYTHONHOME"}
        env = {
            key: str(value)
            for key, value in src_env.items()
            if key not in ignored_env
        }
        # subprocess.Popen keyword arguments
        self.kwargs: dict[str, Any] = {"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

        # TODO: add type hints
        # note that these need to be None in order to trigger discovery
        # when 'discover_launch_hooks' is called
        self.prelaunch_hooks = None
        self.postlaunch_hooks = None

        self.process: Optional[Popen] = None
        self._prelaunch_hooks_executed = False

    @property
    def env(self) -> dict[str, str]:
        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: dict[str, str]) -> None:
        if not isinstance(value, dict):
            raise TypeError(
                f"'env' attribute expect 'dict' object. Got: {type(value)}"
            )
        self.kwargs["env"] = value

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

        """
        return self.addons_manager

    def _collect_addons_launch_hook_paths(self) -> list[str]:
        """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"
                    f" type {type(hook_paths)}. Expected {expected_types}"
                )
                continue

            output.extend(hook_paths)
        return output

    def paths_to_launch_hooks(self) -> list[str]:
        """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: bool = False) -> None:
        """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(f"- {path}" for path in paths)
        ))

        all_classes: dict[str, list[Type[Union[PreLaunchHook, PostLaunchHook]]]] = {  # noqa: E501
            "pre": [],
            "post": []
        }
        for path in paths:
            if not os.path.exists(path):
                self.log.info(
                    f"Path to launch hooks does not exist: \"{path}\""
                )
                continue

            result = modules_from_path(path)
            # Future compatibility using 'ModulesResult'
            # TODO Remove when ayon-core > 1.9.10 is required
            if hasattr(result, "modules"):
                modules = [item.module for item in result.modules]
            else:
                modules_info, _crashed = result
                modules = [module for _, module in modules_info]

            for 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:"
                            f" {klass.__name__}"
                        )
                        continue

                    if inspect.isabstract(hook):
                        self.log.debug(
                            f"Skipped abstract hook: {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(
                        f"Initialization of hook failed: {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(
            f"Found {len(self.prelaunch_hooks)} prelaunch"
            f" and {len(self.postlaunch_hooks)} postlaunch hooks."
        )

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

    @property
    def host_name(self) -> str:
        return self.application.host_name

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

    @property
    def manager(self) -> ApplicationManager:
        return self.application.manager

    def _run_process(self) -> subprocess.Popen:
        """Run the process with the given launch arguments and keyword args.

        This method will handle the process differently based on the platform
        it is running on. It will create a temporary file for output on
        Windows and macos, while on Linux it will use a mid-process to launch
        the application with the provided arguments and environment variables.

        It will pass file paths to temporary files to the mid-process where
        the process output and pid will be stored.

        Returns:
            subprocess.Popen: The process object created by Popen.

        """
        # Windows and macOS have easier process start
        low_platform = platform.system().lower()
        if low_platform in ("windows", "darwin"):
            return self._execute_with_stdout()
        # 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 mid-process
        # - store arguments to a json and pass path to json as last argument
        # - pass environments to set
        app_env = self.kwargs.pop("env", {})
        # create temporary file path passed to mid-process

        output_file = None
        if self.redirect_output:
            with tempfile.NamedTemporaryFile(
                mode="w",
                prefix=f"ayon_{self.application.host_name}_output_",
                suffix=".txt",
                delete=False,
                encoding="utf-8",
            ) as temp_file:
                output_file = temp_file.name
        # create temporary file to read back pid
        with tempfile.NamedTemporaryFile(
            mode="w",
            prefix="ayon_pid_",
            suffix=".txt",
            delete=False,
            encoding="utf-8",
        ) as pid_temp_file:
            pid_file = pid_temp_file.name

        json_data = {
            "args": self.launch_args,
            "env": app_env,
            "pid_file": pid_file,
        }
        if output_file:
            json_data["stdout"] = output_file
            json_data["stderr"] = output_file

        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 the temp file
        with tempfile.NamedTemporaryFile(
            mode="w", prefix="ay_app_args", suffix=".json", delete=False
        ) as json_temp:
            json_temp_filepath = json_temp.name
            json.dump(json_data, json_temp)

        launch_args.append(json_temp_filepath)

        # 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()

        # read back pid from the json file
        try:
            with open(json_temp_filepath, encoding="utf-8") as stream:
                json_data = json.load(stream)

                try:
                    import psutil
                except ImportError:
                    psutil = None

                pid_from_mid = json_data.get("pid")
                executable = Path(str(self.executable))
                start_time = None
                if pid_from_mid and psutil:
                    start_time = (
                        self.process_manager.get_process_start_time_by_pid(
                            pid_from_mid)
                    )
                    executable = (
                        self.process_manager.get_executable_path_by_pid(
                            pid_from_mid)
                    ) or executable

                from .process import ProcessInfo

                process_info = ProcessInfo(
                    name=self.application.full_name,
                    executable=executable,
                    args=self.launch_args,
                    env=app_env,
                    cwd=self.kwargs.get("cwd") or os.getcwd(),
                    pid=pid_from_mid,
                    output=Path(output_file) if self.redirect_output else None,
                    start_time=start_time,
                )
                # Store process info to the database
                self.process_manager.store_process_info(process_info)
        except OSError:
            self.log.exception(
                "Failed to read process info from JSON file: %s"
            )

        # Remove the temp file
        os.remove(json_temp_filepath)
        # Return process which is already terminated
        return process

    def run_prelaunch_hooks(self) -> None:
        """Run prelaunch hooks.

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

        Raises:
            RuntimeError: When prelaunch hooks were already executed.

        """
        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 hook in self.prelaunch_hooks:
            self.log.debug(
                f"Executing prelaunch hook: {hook.__class__.__name__}"
            )
            hook.execute()
        self._prelaunch_hooks_executed = True

    def launch(self) -> Optional[subprocess.Popen]:
        """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 None

        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 = f" ({len(args)})"
        self.log.info(
            f'Launching "{self.application.full_name}"'
            f" with args{args_len_str}: {args}"
        )
        self.launch_args = args

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

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

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

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

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

        return self.process

    @staticmethod
    def clear_launch_args(args: list) -> list[str]:
        """Collect launch arguments to final order.

        Launch argument should be a 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.

        Returns:
            list: Unpacked arguments.

        """
        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

    def _execute_with_stdout(self) -> subprocess.Popen:
        """Run the process with stdout and stderr redirected to a file.

        Stores process information to the database.

        Returns:
            subprocess.Popen: The process object created by Popen.
        """
        from .process import ProcessInfo

        process_info = ProcessInfo(
            name=self.application.full_name,
            executable=Path(str(self.executable)),
            args=self.launch_args,
            env=self.kwargs.get("env", {}),
            cwd=self.kwargs.get("cwd") or os.getcwd(),
            output=None,
            pid=None,
            start_time=None,
        )

        if self.redirect_output:
            with tempfile.NamedTemporaryFile(
                mode="w",
                prefix=f"ayon_{self.application.host_name}_output_",
                suffix=".txt",
                delete=False,
                encoding="utf-8",
            ) as temp_file:
                temp_file_path = temp_file.name

            with open(temp_file_path, "wb") as tmp_file:
                self.kwargs["stdout"] = tmp_file
                self.kwargs["stderr"] = tmp_file
                process = subprocess.Popen(self.launch_args, **self.kwargs)
                process_info.output = Path(temp_file_path)
        else:
            process = subprocess.Popen(self.launch_args, **self.kwargs)

        start_time = self.process_manager.get_process_start_time(process)
        process_info.pid = process.pid
        process_info.start_time = start_time
        # Store process info to the database
        self.process_manager.store_process_info(process_info)

        return process

class ApplicationLaunchFailed(Exception):
    """Application launch failed due to known reason.

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

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: Optional[dict[str, Any]] = None):
        self.log = Logger.get_logger(self.__class__.__name__)

        self.app_groups: dict[str, ApplicationGroup] = {}
        self.applications: dict[str, Application] = {}
        self.tool_groups: dict[str, EnvironmentToolGroup] = {}
        self.tools: dict[str, EnvironmentTool] = {}

        self._app_group_info: dict[str, GroupAppInfo] = {}

        self._studio_settings = studio_settings

        self.refresh()

    def get_app_label(self, group_name: str) -> str:
        """Get label for application group by name.

        Args:
            group_name (str): Application group name.

        Returns:
            str: Application label.

        """
        group_info = self._app_group_info.get(group_name)
        if group_info is not None and group_info.label:
            return group_info.label
        return group_name

    def get_app_icon(self, group_name: str) -> dict[str, str] | None:
        """Get icon for application group by name.

        Args:
            group_name (str): Application name.

        Returns:
            dict[str, str] | None: Application icon definition.

        """
        group_info = self._app_group_info.get(group_name)
        if group_info is None:
            return None
        return group_info.icon

    def set_studio_settings(self, studio_settings: dict[str, Any]) -> None:
        """Ability to change init system settings.

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

        self.refresh()

    def refresh(self) -> None:
        """Refresh applications from settings."""
        from ayon_applications import ApplicationsAddon

        self.app_groups.clear()
        self.applications.clear()
        self.tool_groups.clear()
        self.tools.clear()
        self._app_group_info.clear()

        app_group_info = {}
        app_items = ApplicationsAddon.get_application_items()
        for app_item in app_items:
            group_name = app_item["full_name"].split("/")[0]
            if group_name in app_group_info:
                continue

            app_group_info[group_name] = GroupAppInfo(
                group_name,
                label=app_item["group_label"],
                icon=app_item["icon"],
            )

        self._app_group_info = app_group_info

        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(
                    f"Additional application '{app_name}' is already"
                    " in built-in applications."
                )
            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: str
    ) -> Optional[ApplicationGroup]:
        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: str, **data
    ) -> "ApplicationLaunchContext":
        """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: "ApplicationLaunchContext"
    ) -> Optional[subprocess.Popen]:
        """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) -> Optional[subprocess.Popen]:
        """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 an explanation message,
                traceback should not be needed.

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