PluginManager¶

@err_catcher(name=__name__)
def initializePlugins(self, appPlugin: str) -> None:
    """Initialize and load all plugins for a Prism session.

    Searches for and loads the specified app plugin first, then loads all other
    plugins from configured paths. Triggers startup callbacks after loading.

    Args:
        appPlugin: Name of the app plugin to load (e.g., 'Standalone', 'Maya', 'Houdini').
    """
    self.core.unloadedAppPlugins = {}
    self.core.customPlugins = {}
    self.core.unloadedPlugins = {}
    self.core.pluginMetaData = {}
    self.renderfarmPlugins = []

    pluginDirs = self.getPluginDirs()
    appPlugs = self.searchPlugins(
        pluginPaths=pluginDirs["pluginPaths"],
        directories=pluginDirs["searchPaths"],
        pluginNames=[appPlugin],
    )
    if not appPlugs:
        msg = "App plugin %s couldn't be found." % appPlugin
        self.core.popup(msg)
        return

    appPlug = self.loadAppPlugin(
        appPlugs[0]["name"], pluginPath=appPlugs[0]["path"], startup=True
    )
    if not appPlug:
        msg = "App plugin %s couldn't be loaded." % appPlugs[0]["name"]
        self.core.popup(msg)
        return

    self.loadPlugins(
        pluginPaths=pluginDirs["pluginPaths"],
        directories=pluginDirs["searchPaths"],
        force=False,
        ignore=[appPlugs[0]["name"]],
    )
    self.core.callback("onPluginsLoaded")
    if self.core.splashScreen:
        self.core.splashScreen.setStatus("plugins loaded...")

    if self.core.appPlugin and self.core.appPlugin.pluginName != "Standalone":
        # self.core.maxwait = 120
        # self.core.elapsed = 0
        self.core.timer = QTimer()
        result = self.core.startup()
        if result is False:
            self.core.timer.timeout.connect(self.core.startup)
            self.core.timer.start(1000)
    else:
        self.core.startup()

@err_catcher(name=__name__)
def getPluginDirs(self, includeDefaults: bool = True, includeEnv: bool = True, includeConfig: bool = True, enabledOnly: bool = True) ->  Dict[str, List[str]]:
    """Get plugin directories from various sources.

    Args:
        includeDefaults: Include default plugin directories
        includeEnv: Include PRISM_PLUGIN_PATHS and PRISM_PLUGIN_SEARCH_PATHS env vars
        includeConfig: Include user config plugin paths
        enabledOnly: Only include enabled plugins from config

    Returns:
        Dict with 'pluginPaths' and 'searchPaths' lists
    """
    result = {"pluginPaths": [], "searchPaths": []}
    if includeDefaults:
        result["searchPaths"] = self.core.pluginDirs[:]

    if includeEnv:
        envPluginDirs = os.getenv("PRISM_PLUGIN_PATHS", "").split(os.pathsep)
        if envPluginDirs[0]:
            result["pluginPaths"] += envPluginDirs

        envPluginSearchDirs = os.getenv("PRISM_PLUGIN_SEARCH_PATHS", "").split(os.pathsep)
        if envPluginSearchDirs[0]:
            result["searchPaths"] += envPluginSearchDirs

    if includeConfig:
        userPluginDirs = self.core.getConfig(config="PluginPaths") or {}
        if userPluginDirs.get("plugins"):
            if enabledOnly:
                result["pluginPaths"] += [p["path"] for p in userPluginDirs["plugins"] if p.get("enabled", True)]
            else:
                result["pluginPaths"] += userPluginDirs["plugins"]

        if userPluginDirs.get("searchPaths"):
            if enabledOnly:
                result["searchPaths"] += [p["path"] for p in userPluginDirs["searchPaths"] if p.get("enabled", True)]
            else:
                result["searchPaths"] += userPluginDirs["searchPaths"]

    return result

@err_catcher(name=__name__)
def setPluginPathEnabled(self, path: str, enabled: bool) -> bool:
    """Enable or disable a plugin path in user config.

    Args:
        path: Plugin path to modify.
        enabled: Whether to enable or disable the path.

    Returns:
        bool: True if successful, False otherwise.
    """
    userPluginDirs = self.core.getConfig(config="PluginPaths") or {}
    if not userPluginDirs.get("plugins"):
        return False

    for plugin in userPluginDirs["plugins"]:
        if plugin["path"] != path:
            continue

        plugin["enabled"] = enabled
        self.core.setConfig(data=userPluginDirs, config="PluginPaths")
        return True

    return False

@err_catcher(name=__name__)
def setPluginSearchPathEnabled(self, path: str, enabled: bool) -> bool:
    """Enable or disable a plugin search path in user config.

    Args:
        path: Search path to modify.
        enabled: Whether to enable or disable the path.

    Returns:
        bool: True if successful, False otherwise.
    """
    userPluginDirs = self.core.getConfig(config="PluginPaths") or {}
    if not userPluginDirs.get("plugins"):
        return False

    for plugin in userPluginDirs["searchPaths"]:
        if plugin["path"] != path:
            continue

        plugin["enabled"] = enabled
        self.core.setConfig(data=userPluginDirs, config="PluginPaths")
        return True

    return False

@err_catcher(name=__name__)
def getPluginPath(self, location: str = "root", pluginType: str = "", path: str = "", pluginName: str = "") -> str:
    """Get the file system path for a plugin.

    Args:
        location: Location type ('root', 'computer', 'user', 'project', 'custom').
        pluginType: Optional plugin type ('App', 'Custom', 'Single File').
        path: Custom path if location is 'custom'.
        pluginName: Optional specific plugin name.

    Returns:
        str: Plugin path.
    """
    if location == "root":
        pluginPath = os.path.abspath(
            os.path.join(__file__, os.pardir, os.pardir, os.pardir, "Plugins")
        )
    elif location == "computer":
        pluginPath = self.getComputerPluginPath()
    elif location == "user":
        pluginPath = self.getUserPluginPath()
    elif location == "project":
        pluginPath = self.core.projects.getPluginFolder()
    elif location == "custom":
        pluginPath = path

    if location not in ["computer", "custom", "user", "project"] and pluginType:
        if pluginType == "App":
            dirName = "Apps"
        elif pluginType == "Custom":
            dirName = "Custom"
        else:
            dirName = ""

        if dirName:
            pluginPath = os.path.join(pluginPath, dirName)

    if pluginName:
        pluginPath = os.path.join(pluginPath, pluginName)
        if pluginType == "Single File":
            pluginPath += ".py"

    return pluginPath.replace("\\", "/")

@err_catcher(name=__name__)
def getUserPluginPath(self) -> str:
    """Get the user-specific plugin directory path.

    Returns:
        str: User plugin directory path.
    """
    pluginPath = os.path.join(os.path.dirname(self.core.userini), "plugins")
    return pluginPath

@err_catcher(name=__name__)
def getComputerPluginPath(self) -> str:
    """Get the computer-specific plugin directory path.

    Returns:
        str: Computer plugin directory path.
    """
    pluginPath = os.path.join(self.core.getPrismDataDir(), "plugins")
    return pluginPath

@err_catcher(name=__name__)
def getDefaultPluginPath(self) -> str:
    """Get the default plugin installation directory.

    Checks environment variable and user config, falls back to computer path.

    Returns:
        str: Default plugin directory path.
    """
    path = os.getenv("PRISM_DEFAULT_PLUGIN_PATH")
    if not path:
        path = self.core.getConfig("globals", "defaultPluginPath", config="user")
        if not path:
            path = self.getComputerPluginPath()

    return path

@err_catcher(name=__name__)
def getFallbackPluginPath(self) -> str:
    """Get the fallback plugin directory for failed installations.

    Checks environment variable and user config, falls back to user path.

    Returns:
        str: Fallback plugin directory path.
    """
    path = os.getenv("PRISM_FALLBACK_PLUGIN_PATH")
    if not path:
        path = self.core.getConfig("globals", "fallbackPluginPath", config="user")
        if not path:
            path = self.getUserPluginPath()

    return path

@err_catcher(name=__name__)
def loadAppPlugin(self, pluginName: str, pluginPath: Optional[str] = None, startup: bool = False) -> Optional[Any]:
    """Load the application plugin for the current DCC.

    Args:
        pluginName: Name of the app plugin (e.g., 'Maya', 'Houdini').
        pluginPath: Optional path to plugin directory.
        startup: Whether this is being called during Prism startup.

    Returns:
        Optional[Any]: Loaded plugin instance, or None if loading failed.
    """
    if self.core.splashScreen:
        self.core.splashScreen.setStatus("loading appPlugin %s..." % pluginName)

    if not pluginPath:
        pluginPath = os.path.join(self.core.pluginPathApp, pluginName, "Scripts")
    else:
        if os.path.basename(pluginPath) != "Scripts":
            pluginPath = os.path.join(pluginPath, "Scripts")

    sys.path.append(pluginPath)
    self.core.appPlugin = None
    try:
        appPlug = getattr(
            __import__("Prism_%s_init" % pluginName), "Prism_Plugin_%s" % pluginName
        )(self.core)
    except Exception as e:
        logger.warning(traceback.format_exc())
        msg = "Failed to load app plugin.\nPlease contact the support.\n\n%s" % e
        self.core.popup(msg)
        return

    if not getattr(appPlug, "isActive", lambda: True)():
        logger.debug("no appPlugin loaded")
        return

    self.core.appPlugin = appPlug

    if not self.core.appPlugin:
        msg = "Prism could not initialize correctly and may not work correctly in this session."
        self.core.popup(msg, severity="error")
        return

    if not getattr(self.core.appPlugin, "enabled", True):
        logger.debug("appplugin disabled")
        return

    self.core.appPlugin.location = "prismRoot"
    self.core.appPlugin.pluginPath = pluginPath

    if (
        not getattr(self.core, "messageParent", None)
    ):
        isHp311 = self.core.appPlugin.pluginName == "Houdini" and sys.version_info.minor == 11
        if isHp311 or QApplication.instance() is not None:
            for arg in self.core.prismArgs:
                if isinstance(arg, dict) and "messageParent" in arg:
                    self.core.messageParent = arg["messageParent"]
                    break
            else:
                self.core.messageParent = QWidget()

    if not self.core.appPlugin.hasQtParent:
        self.core.parentWindows = False
        if self.core.appPlugin.pluginName != "Standalone" and self.core.useOnTop:
            self.core.messageParent.setWindowFlags(
                self.core.messageParent.windowFlags() ^ Qt.WindowStaysOnTopHint
            )

    getattr(self.core.appPlugin, "instantStartup", lambda x: None)(self.core)

    if not startup:
        self.core.appPlugin.startup(self.core)

    self.core.callback("pluginLoaded", args=[self.core.appPlugin])
    logger.debug("loaded app plugin %s" % pluginName)
    return self.core.appPlugin

@err_catcher(name=__name__)
def loadPlugins(
    self,
    pluginPaths: Optional[List[str]] = None,
    directory: Optional[str] = None,
    directories: Optional[List[str]] = None,
    recursive: bool = False,
    force: bool = True,
    ignore: Optional[List[str]] = None,
    singleFilePlugins: bool = False,
) -> List[Any]:
    """Load plugins from specified paths or directories.

    Args:
        pluginPaths: Specific plugin paths to load
        directory: Single directory to search for plugins
        directories: Multiple directories to search
        recursive: Search subdirectories recursively
        force: Force reload even if already loaded
        ignore: List of plugin names to skip
        singleFilePlugins: Load single-file .py plugins

    Returns:
        List of loaded plugin instances
    """
    ignore = ignore or []
    result = []
    foundPluginPaths = []

    loadPlugins = None
    for arg in self.core.prismArgs:
        if isinstance(arg, dict) and "loadPlugins" in arg:
            loadPlugins = arg["loadPlugins"]
            force = True
            break

    if pluginPaths:
        for pPath in pluginPaths:
            expPath = os.path.expandvars(pPath)
            foundPluginPaths.append(expPath)

    directories = directories or []
    if directory:
        directories.append(directory)

    if directories:
        for dr in directories:
            expDr = os.path.expandvars(dr)
            if not os.path.exists(expDr):
                continue

            if recursive:
                for root, dirs, files in os.walk(expDr):
                    for f in files:
                        if f.endswith("_init.py"):
                            path = os.path.dirname(root)
                            foundPluginPaths.append(path)
                            break
            else:
                for root, dirs, files in os.walk(expDr):
                    for pDir in dirs:
                        if pDir == "PluginEmpty":
                            continue

                        if pDir == self.core.appPlugin.pluginName:
                            continue

                        if pDir.startswith(".") or pDir.startswith("_"):
                            continue

                        path = os.path.join(expDr, pDir)
                        foundPluginPaths.append(path)

                    if singleFilePlugins:
                        for file in files:
                            if file.startswith(".") or file.startswith("_"):
                                continue

                            if not file.endswith(".py"):
                                continue

                            path = os.path.join(expDr, file)
                            foundPluginPaths.append(path)

                    break

    for pluginPath in foundPluginPaths:
        if pluginPath.endswith(".py"):
            if loadPlugins:
                continue
        else:
            pluginName = self.getPluginNameFromPath(pluginPath)
            if pluginName in ignore:
                continue

            if loadPlugins and pluginName not in loadPlugins:
                if "loadPluginMetaData" in self.core.prismArgs:
                    self.loadPluginMetaData(pluginPath)
                continue

            if self.isPluginLoaded(pluginName):
                continue

        result.append(self.loadPlugin(pluginPath, force=force))

    return result

@err_catcher(name=__name__)
def searchPlugins(
    self,
    pluginPaths: Optional[List[str]] = None,
    directory: Optional[str] = None,
    directories: Optional[List[str]] = None,
    recursive: bool = True,
    pluginNames: Optional[List[str]] = None,
) -> List[Dict[str, str]]:
    """Search for plugins in specified paths.

    Args:
        pluginPaths: Optional list of direct plugin folder paths.
        directory: Optional single directory to search.
        directories: Optional list of directories to search.
        recursive: Whether to search recursively. Defaults to True.
        pluginNames: Optional list of specific plugin names to find.

    Returns:
        List[Dict[str, str]]: List of dicts with 'name' and 'path' keys.
    """
    result = []

    if pluginPaths:
        for pPath in pluginPaths:
            pluginName = os.path.basename(pPath)
            if pluginNames and pluginName not in pluginNames:
                continue

            expPath = os.path.expandvars(pPath)
            if not os.path.exists(expPath):
                continue

            pData = {"name": pluginName, "path": expPath}
            result.append(pData)

    directories = directories or []
    if directory:
        directories.append(directory)

    for dr in directories:
        expDr = os.path.expandvars(dr)
        if not os.path.exists(expDr):
            continue

        for root, dirs, files in os.walk(expDr):
            if "Scripts" in dirs:
                dirs[:] = ["Scripts"]
                continue

            dirs[:] = [d for d in dirs if d[0] not in [".", "_"]]
            for f in files:
                if not f.endswith("_init.py"):
                    continue

                dirs[:] = []
                path = os.path.dirname(root)
                pluginName = os.path.basename(path)
                if pluginNames and pluginName not in pluginNames:
                    continue

                pData = {"name": pluginName, "path": path}
                result.append(pData)
                break

            if not recursive:
                break

    return result

@err_catcher(name=__name__)
def activatePlugin(self, path: str) -> Optional[Any]:
    """Activate and load a previously deactivated plugin.

    Args:
        path: Path to plugin directory.

    Returns:
        Optional[Any]: Loaded plugin instance, or None if loading failed.
    """
    if os.path.basename(path) == "Scripts":
        path = os.path.dirname(path)

    pluginName = os.path.basename(path)
    if pluginName in self.core.unloadedPlugins:
        self.core.unloadedPlugins.pop(pluginName)

    logger.debug("activating plugin %s" % pluginName)
    return self.loadPlugin(path)

@err_catcher(name=__name__)
def loadPlugin(self, path: Optional[str] = None, name: Optional[str] = None, force: bool = True, activate: Optional[bool] = None, showWarnings: bool = False) -> Optional[Any]:
    """Load a single plugin from path or name.

    Args:
        path: Optional path to plugin directory or .py file.
        name: Optional plugin name to search for and load.
        force: Whether to reload if already loaded. Defaults to True.
        activate: Whether to activate if plugin was inactive. Defaults to None.
        showWarnings: Whether to show warning popups. Defaults to False.

    Returns:
        Optional[Any]: Loaded plugin instance, or None if loading failed.
    """
    # logger.debug("about to load plugin: %s - %s" % (path, name))
    if not path:
        if name:
            pluginPaths = self.searchPluginPaths(name)
            if not pluginPaths:
                msg = "couldn't find plugin: %s" % name
                if showWarnings:
                    self.core.popup(msg)
                else:
                    logger.debug(msg)

                return

            for pluginPath in pluginPaths:
                result = self.loadPlugin(path=pluginPath, name=name, force=force, activate=activate)
                if result:
                    return result

            return

        if not path:
            msg = "invalid pluginpath: \"%s\"" % path
            if showWarnings:
                self.core.popup(msg)
            else:
                logger.debug(msg)

            return

    if os.path.normpath(path).startswith(os.path.normpath(self.core.prismRoot)):
        location = "prismRoot"
    elif path.startswith(getattr(self.core, "projectPath", ())):
        location = "prismProject"
    else:
        location = "custom"

    notAutoLoadedPlugins = self.getNotAutoLoadPlugins()

    if path.endswith(".py"):
        dirpath = os.path.dirname(path)
        if dirpath not in sys.path:
            sys.path.append(dirpath)

        pluginName = os.path.basename(os.path.splitext(path)[0]).replace("Prism_Plugin_", "")
        if self.core.splashScreen:
            self.core.splashScreen.setStatus("loading plugin %s..." % pluginName)

        initPath = path
        pluginPath = path
    else:
        if os.path.basename(path) == "Scripts":
            path = os.path.dirname(path)

        pluginName = os.path.basename(path)
        if pluginName == "PluginEmpty":
            return

        if self.core.splashScreen:
            self.core.splashScreen.setStatus("loading plugin %s..." % pluginName)

        if pluginName == "LoadExternalPlugins":
            result = self.core.getConfig("plugins", "load_deprExternalPlugins")
            if result is None:
                qstr = 'Deprecated plugin found: "LoadExternalPlugins"\nLoading this plugin can cause errors if you haven\'t modified it to work with this Prism version.\n\nAre you sure you want to load this plugin? (if unsure click "No")'
                answer = self.core.popupQuestion(qstr, buttons=["Yes", "No"])
                if answer == "No":
                    self.core.setConfig("plugins", "load_deprExternalPlugins", False)
                    return
                else:
                    self.core.setConfig("plugins", "load_deprExternalPlugins", True)
            elif not result:
                return

        if self.core.getPlugin(pluginName):
            if force:
                self.unloadPlugin(pluginName)
            else:
                msg = "plugin is already loaded: \"%s\"" % pluginName
                if showWarnings:
                    self.core.popup(msg)
                else:
                    logger.warning(msg)

                return

        # logger.debug(pluginName)
        initmodule = "Prism_%s_init" % pluginName
        pluginPath = os.path.join(path, "Scripts")
        initPath = os.path.join(pluginPath, initmodule + ".py")

        if pluginName in notAutoLoadedPlugins and not force:
            if not os.path.exists(path):
                msg = "pluginpath doesn't exist: %s" % path
                if showWarnings:
                    self.core.popup(msg)
                else:
                    logger.debug(msg)

                return

            if activate:
                return self.activatePlugin(path)

            self.core.unloadedPlugins[pluginName] = UnloadedPlugin(self.core, pluginName, path=pluginPath, location=location)
            msg = "skipped loading plugin %s - autoload of this plugin is disabled in the preferences" % pluginName
            if showWarnings:
                self.core.popup(msg)
            else:
                logger.debug(msg)

            return

        if self.core.appPlugin and (pluginName == self.core.appPlugin.pluginName):
            return

        if not (
            os.path.exists(initPath)
            or os.path.exists(initPath.replace("_init", "_init_unloaded"))
        ):
            # self.core.unloadedPlugins[pluginName] = UnloadedPlugin(self.core, pluginName, path=pluginPath, location=location)
            msg = "skipped loading plugin %s - folder doesn't contain a valid plugin (no init script) - check your plugin configuration. %s " % (pluginName, path)
            if showWarnings:
                self.core.popup(msg)
            else:
                logger.debug(msg)

            return

    if os.path.dirname(initPath) not in sys.path:
        sys.path.append(os.path.dirname(initPath))

    try:
        if path.endswith(".py"):
            if not os.path.exists(path):
                msg = "pluginpath doesn't exist: %s" % path
                if showWarnings:
                    self.core.popup(msg)
                else:
                    logger.debug(msg)

                return

            plugModule = __import__(pluginName)
            if hasattr(plugModule, "name"):
                pluginName = plugModule.name

            if pluginName in notAutoLoadedPlugins and not force:
                if activate:
                    return self.activatePlugin(path)

                self.core.unloadedPlugins[pluginName] = UnloadedPlugin(self.core, pluginName, path=pluginPath, location=location)
                msg = "skipped loading plugin %s - autoload of this plugin is disabled in the preferences" % pluginName
                if showWarnings:
                    self.core.popup(msg)
                else:
                    logger.debug(msg)

                return

            if hasattr(plugModule, "classname"):
                classname = plugModule.classname
            else:
                classname = "Prism_%s" % pluginName

            pPlug = getattr(plugModule, classname)(self.core)
            pPlug.pluginName = pluginName
        elif os.path.exists(initPath.replace("_init", "_init_unloaded")):
            pPlug = getattr(
                __import__("Prism_%s_init_unloaded" % (pluginName)),
                "Prism_%s_unloaded" % pluginName,
            )(self.core)
        else:
            pPlug = getattr(
                __import__("Prism_%s_init" % (pluginName)), "Prism_%s" % pluginName
            )(self.core)
    except:
        msg = "Failed to load plugin: %s" % pluginName
        detailMsg = msg + "\n\n" + traceback.format_exc()
        logger.debug(detailMsg)
        result = self.core.popupQuestion(
            msg,
            buttons=["Details", "Close"],
            icon=QMessageBox.Warning,
            default="Close",
        )

        if result == "Details":
            self.core.showErrorDetailPopup(detailMsg)

        for arg in self.core.prismArgs:
            if isinstance(arg, dict) and "errorCallback" in arg:
                arg["errorCallback"](detailMsg)
                break

        self.core.unloadedPlugins[pluginName] = UnloadedPlugin(self.core, pluginName, path=pluginPath, location=location)
        return

    if hasattr(pPlug, "platforms") and platform.system() not in pPlug.platforms:
        msg = "skipped loading plugin %s - plugin doesn't support this OS" % pPlug.pluginName
        if showWarnings:
            self.core.popup(msg)
        else:
            logger.debug(msg)

        return

    if pluginName in self.core.unloadedPlugins:
        self.core.unloadedPlugins.pop(pluginName)

    pPlug.location = location
    pPlug.pluginPath = pluginPath

    if hasattr(pPlug, "pluginType") and pPlug.pluginType in ["App"]:
        self.core.unloadedAppPlugins[pPlug.pluginName] = pPlug
    else:
        if not getattr(pPlug, "isActive", lambda: True)():
            self.core.unloadedPlugins[pPlug.pluginName] = pPlug
            msg = "plugin \"%s\" is inactive" % pPlug.pluginName
            if showWarnings:
                self.core.popup(msg)
            else:
                logger.debug(msg)

            return

        if not hasattr(pPlug, "pluginType") or pPlug.pluginType in ["Custom"]:
            self.core.customPlugins[pPlug.pluginName] = pPlug

    if self.core.pb:
        self.core.pb.sceneBrowser.refreshAppFilters()

    self.core.callback("pluginLoaded", args=[pPlug])
    logger.debug("loaded plugin %s" % pPlug.pluginName)
    return pPlug