annotations/Azaion.Inference/ai_availability_status.pyx

cimport constants_inf
import msgpack
from threading import Lock

AIStatus2Text = {
    AIAvailabilityEnum.NONE: "None",
    AIAvailabilityEnum.DOWNLOADING: "Downloading",
    AIAvailabilityEnum.CONVERTING: "Converting",
    AIAvailabilityEnum.UPLOADING: "Uploading",
    AIAvailabilityEnum.ENABLED: "Enabled",
    AIAvailabilityEnum.WARNING: "Warning",
    AIAvailabilityEnum.ERROR: "Error",
}

cdef class AIAvailabilityStatus:
    def __init__(self):
        self.status = AIAvailabilityEnum.NONE
        self.error_message = None
        self._lock = Lock()

    def __str__(self):
        self._lock.acquire()
        try:
            status_text = AIStatus2Text.get(self.status, "Unknown")
            error_text = self.error_message if self.error_message else ""
            return f"{status_text} {error_text}"
        finally:
            self._lock.release()

    cdef bytes serialize(self):
        self._lock.acquire()
        try:
            return msgpack.packb({
                "s": self.status,
                "m": self.error_message
            })
        finally:
            self._lock.release()

    cdef set_status(self, AIAvailabilityEnum status, str error_message=None):
        log_message = ""
        self._lock.acquire()
        try:
            self.status = status
            self.error_message = error_message
            status_text = AIStatus2Text.get(self.status, "Unknown")
            error_text = self.error_message if self.error_message else ""
            log_message = f"{status_text} {error_text}"
        finally:
            self._lock.release()

        if error_message is not None:
            constants_inf.logerror(<str>error_message)
        else:
            constants_inf.log(<str>log_message)