compute_adapter.py

# =====================================================
# Apckeyl Framework
# Version 1.1
# compute_adapter.py
# =====================================================

"""
Compute Adapter.

Version 1.1

Адаптер между Apckeyl Framework
и конкретным Compute Module.

На этом этапе реальный внешний модуль
Apckeyl_RealESRGAN ещё НЕ подключается.
"""

from compute_module import ComputeModule


# =====================================================
# Compute Adapter
# =====================================================

class ComputeAdapter:

    def __init__(
        self,
        compute_module,
    ):

        if compute_module is None:

            raise ValueError(
                "compute_module is required"
            )

        if not isinstance(
            compute_module,
            ComputeModule,
        ):

            raise TypeError(
                "compute_module must be "
                "an instance of ComputeModule"
            )

        self.compute_module = (
            compute_module
        )


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

    def get_module_info(self):

        return (
            self.compute_module.get_info()
        )


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

    def health_check(self):

        return (
            self.compute_module.health_check()
        )


    # =================================================
    # Task Support
    # =================================================

    def supports_task(
        self,
        task_type,
    ):

        return (
            self.compute_module.supports_task(
                task_type
            )
        )


    # =================================================
    # Execute Task
    # =================================================

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

        return (
            self.compute_module.execute(
                task
            )
        )


# =====================================================
# Adapter Factory
# =====================================================

def create_compute_adapter(
    compute_module,
):

    return ComputeAdapter(
        compute_module
    )