Source code for cgl.plugins.smart_task

from __future__ import annotations

import importlib
import inspect
import logging
import os
from pathlib import Path

from cgl.core.config.query import AlchemyConfigManager

CFG = AlchemyConfigManager()




class SmartTask(object):
    """
    This is a template for a "task" within the cookbook.  It covers common areas when dealing with digital assets
    specific to different tasks.
    """

    scene = None
    path_object = None
    msd_path = ""
    msd_dict = None
    task = None
    task_entity_type = None
    task_object = None
    dcc = None

    def __init__(
        self,
        msd=None,
        task_object=None,
        path_object=None,
        dcc=None,
        task=None,
        scene=None,
        **kwargs,
    ):
        """
        Args:
            dcc: the dcc we are working in
            task: the task we are working on
            path_object: the path object we are working on
            origin_task: the task this SmartTask is originating from (for example, if we are exporting a camera from a
            previz task.
            **kwargs: any additional arguments we want to pass in

        """
        self.msd = msd
        self.task = task
        self.task_object = task_object
        self.path_object = path_object
        self.dcc = dcc
        self.scene = scene
        caller_file = inspect.stack()[1].filename
        for key in kwargs:
            setattr(self, key, kwargs[key])
        if not self.scene:
            self.set_scene()
        if not self.path_object:
            self.set_path_object()
        if not self.dcc:
            self.dcc = get_dcc_from_task_file(caller_file)

        if not self.task:
            if not self.task_object:
                self.task = os.path.basename(os.path.dirname(caller_file))
            else:
                self.task = self.task_object.task
        if self.task and not self.task_object:
            self.set_task_object()

        if not self.task_object:
            raise ValueError(
                "Task object not set. Please provide a task_object or set the task."
            )

        # if self.path_object:
        #     if not self.task_object:
        #         self.origin_task = self.path_object.task
        #         self.set_task_object()
        # else:
        #     raise ValueError(f"Could not set path_object for DCC '{self.dcc}'")



    def set_scene(self):
        try:
            logging.debug(
                f"[set_scene()] setting path object for dcc: {self.dcc} {self.task}"
            )
            module_path = f"cgl.plugins.{self.dcc}.alchemy"
            logging.info("[set_scene()] loading: {}".format(module_path))
            module = importlib.import_module(module_path)
            self.scene_class = getattr(module, "Scene")  # <-- store it
            if self.dcc != "alchemy":
                self.scene = self.scene_class()

        except (ModuleNotFoundError, AttributeError) as e:
            raise ImportError(
                f"[set_scene()] Could not load Scene class for DCC '{self.dcc}': {e}"
            )




    def set_path_object(self):

        self.path_object = self.scene.path_object




    def get_task_object(self, task, latest=True):
        """Returns a PathObject with the task set to the input task
        Args:
            task: the task we want to set the PathObject to
        Returns:
            PathObject: the PathObject with the task set to the input task

        """

        if self.path_object.task != task:
            entity_type = CFG.get_task_entity_type(task)
            return self.path_object.copy(
                tree="render",
                entity_type=entity_type,
                user="publish",
                task=task,
                latest=latest,
                set_filename=True,
            )
        else:
            print("Copying path object for task.")
            return self.path_object.copy()




    def set_task_object(self):
        """
        Gets the Unreal Engine Task Object for this task.
        Returns:

        """
        if self.task and not self.task_object:
            print(f"------>>>> Setting task object for task: {self.task}")
            self.task_object = self.get_task_object(task=self.task)
            self.msd_path = self.task_object.get_msd_path()
            self.msd_dict = self.task_object.get_msd_dict()
        else:
            raise ValueError("Task not set")


    def _import(self, file_path: str | Path, reference: bool = False):
        """Imports the file into the scene - this function should be smart enough to handle various file types
        as well as.

        """
        if os.path.exists(file_path):
            self.Scene().import_file(file_path, reference=reference)
            return file_path
        else:
            logging.info(
                f"No published file found for "
                f"{self.path_object.sequence}/{self.path_object.shot}/{self.path_object.task}"
            )
            return None



    def import_latest(self, task=None, reference=False, **kwargs):
        """
        Imports the lastest published version of whatever files this task is responsible for.

        This method is typically overridden in the specific task class to handle the import logic.
        """
        print(
            "\nDefault import_latest method called. Override this in your task "
            "class to implement specific import logic."
        )
        print(f"MSD path: {self.msd_path}")
        print(f"MSD dict: {self.msd_dict}")




    def get_msd_data(self):
        """
        creates the msd dictionary for the task being called. This is used at publish time to get the data
        to be wrtten to the .msd file.
        Args:
            task_name:
        """
        pass




    def export_task(self):
        """
        Basic export method for tasks

        This should call the customizable '_export()' method per file.

        This method is typically overridden in the specific task class to handle the export logic.
        """
        self.select()
        render_folder = self.task_object.get_render_path(dirname=True)
        print(
            f"emplement your own export_tas function - for example Render folder: {render_folder}"
        )

        # self._export(render_folder)

    def _export(self, file_path):
        """Exports the task

        This must be customized per dcc and per task.
        The Scene() class in the dcc plugin should have an export method that handles the export logic.
        """
        self.Scene().export(file_path)



    def render(self):
        """ """
        pass




    def select(self):
        """
        Selects the task in the scene

        This must be customized per dcc
        """
        pass




    def version_up(self):
        """
        Version up the task. Override this method in the task class if you want to add additional functionality.
        """
        pass






def get_dcc_from_task_file(caller_file):
    """
    Gets the dcc from the caller file path.

    Args:
        caller_file: the file path of the caller

    Returns:
        str: the dcc name
    """
    # Extract the dcc from the caller file path
    if "cookbook" in caller_file:
        # we're using a cookbook smart task
        dcc = caller_file.split("cookbook")[-1]
        dcc = dcc.split("\\")[1]
        return dcc
    else:
        raise (
            f"Dealing with a legacy SmartTask() from the core plugins repository - "
            f"move to cookbook.\n\t {caller_file}"
        )





def get_task_class(software, task):
    """
    gets the class that relates to the specified task, if no task is specified the task for the current scene will
    be used.

    Args:
        software: the dcc in the plugins directory where we can find Task().get_msd_data()
        task: the task to get the class for

    Returns:
        class or None if the class does not exist

    """
    module = "cookbook.{}.tasks.{}.{}".format(software, task, task)
    logging.info(f"[get_task_class] Loading task class from module: {module}")
    module_name = task
    try:
        loaded_module = importlib.import_module(module, module_name)
        class_ = getattr(loaded_module, "Task")
        return class_
    except ModuleNotFoundError:
        logging.info(f"Module: {module} does not exist")
        return None





def get_legacy_task_class(software, task):
    """
    Gets the legacy task class for the specified software and task.
    This is used for tasks that are not in the cookbook structure.

    Args:
        software: the dcc in the plugins directory where we can find Task().get_msd_data()
        task: the task to get the class for

    Returns:
        class or None if the class does not exist
    """
    module = "cgl.plugins.{}.tasks.{}".format(software, task)
    print(f"Loading legacy task class from module: {module}")
    module_name = task
    try:
        loaded_module = importlib.import_module(module, module_name)
        class_ = getattr(loaded_module, "Task")
        return class_
    except ModuleNotFoundError:
        logging.info(f"Legacy Module: {module} does not exist")
        return None



if __name__ == "__main__":
    caller_file = r"E:\Alchemy\CGL\config\cookbook\blender\tasks\srb\task.py"
    print(get_dcc_from_task_file(caller_file))