compute_module.py

# =====================================================
# Apckeyl Framework
# Version 1.0
# compute_module.py
# =====================================================

"""
Apckeyl Compute Module Contract.

Version 1.0

Базовый программный контракт для вычислительных
модулей Apckeyl Framework.

Этот класс НЕ выполняет реальные вычисления.

Он определяет единый интерфейс, которому должны
соответствовать реальные Compute Modules.
"""


# =====================================================
# Compute Module
# =====================================================

class ComputeModule:

    def __init__(
        self,
        module_id,
        module_name,
        module_version,
        supported_tasks=None,
    ):

        if not module_id:

            raise ValueError(
                "module_id is required"
            )

        if not module_name:

            raise ValueError(
                "module_name is required"
            )

        if not module_version:

            raise ValueError(
                "module_version is required"
            )

        if supported_tasks is None:

            supported_tasks = []

        self.module_id = module_id

        self.module_name = module_name

        self.module_version = module_version

        self.supported_tasks = list(
            supported_tasks
        )


    # =================================================
    # Module Information
    # =================================================

    def get_info(self):

        return {

            "module_id": self.module_id,

            "module_name": self.module_name,

            "module_version": (
                self.module_version
            ),

            "supported_tasks": (
                self.supported_tasks.copy()
            ),
        }


    # =================================================
    # Supported Task Check
    # =================================================

    def supports_task(
        self,
        task_type,
    ):

        return (
            task_type
            in self.supported_tasks
        )


    # =================================================
    # Health Check
    # =================================================

    def health_check(self):

        return {

            "module_id": self.module_id,

            "status": "healthy",
        }


    # =================================================
    # Execute
    # =================================================

    def execute(
        self,
        task,
    ):

        if task is None:

            raise ValueError(
                "Task cannot be None"
            )

        if not isinstance(
            task,
            dict,
        ):

            raise TypeError(
                "Task must be a dictionary"
            )

        task_id = task.get(
            "task_id"
        )

        if not task_id:

            raise ValueError(
                "Task must contain task_id"
            )

        task_type = task.get(
            "task_type"
        )

        if not task_type:

            raise ValueError(
                "Task must contain task_type"
            )

        if not self.supports_task(
            task_type
        ):

            raise ValueError(

                f"Unsupported task type: "
                f"{task_type}"
            )


        # -------------------------------------------------
        # Base contract does not perform computation.
        # -------------------------------------------------

        raise NotImplementedError(
            "ComputeModule.execute() "
            "must be implemented by a "
            "concrete Compute Module"
        )


# =====================================================
# Example Contract Module
# =====================================================

class ExampleComputeModule(
    ComputeModule
):

    def __init__(self):

        super().__init__(

            module_id="example",

            module_name=(
                "Example Compute Module"
            ),

            module_version="1.0",

            supported_tasks=[
                "example_task"
            ],
        )


    # =================================================
    # Example Execute
    # =================================================

    def execute(
        self,
        task,
    ):

        if task is None:

            raise ValueError(
                "Task cannot be None"
            )

        if not isinstance(
            task,
            dict,
        ):

            raise TypeError(
                "Task must be a dictionary"
            )

        task_id = task.get(
            "task_id"
        )

        if not task_id:

            raise ValueError(
                "Task must contain task_id"
            )

        task_type = task.get(
            "task_type"
        )

        if not self.supports_task(
            task_type
        ):

            raise ValueError(

                f"Unsupported task type: "
                f"{task_type}"
            )


        return {

            "task_id": task_id,

            "status": "completed",

            "result": {

                "message": (
                    "Example Compute "
                    "Module executed "
                    "successfully."
                ),
            },
        }


# =====================================================
# Default Example Module
# =====================================================

example_compute_module = (
    ExampleComputeModule()
)