base for the application : support custom python application and auto-install dependencies

source/model/base/BaseModel.py

@@ -0,0 +1,123 @@
+import abc
+import gc
+from pathlib import Path
+
+from source.manager import ModelManager
+
+
+class BaseModel(abc.ABC):
+    """
+    Represent a model.
+    """
+
+    def __init__(self, manager: ModelManager, configuration: dict, path: Path):
+        # the environment directory of the model
+        self.path = path
+        # the model manager
+        self.manager = manager
+        # the mimetype of the model responses
+        self.response_mimetype: str = configuration.get("response_type", "application/json")
+
+        self._loaded = False
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__}: {self.name}>"
+
+    @property
+    def name(self):
+        """
+        Get the name of the model
+        :return: the name of the model
+        """
+
+        return self.path.name
+
+    def get_information(self):
+        """
+        Get information about the model
+        :return: information about the model
+        """
+
+        return {
+            "name": self.name,
+            "response_mimetype": self.response_mimetype,
+        }
+
+    def load(self) -> None:
+        """
+        Load the model within the model manager
+        """
+
+        # if we are already loaded, stop
+        if self._loaded:
+            return
+
+        # check if we are the current loaded model
+        if self.manager.current_loaded_model is not self:
+            # unload the previous model
+            if self.manager.current_loaded_model is not None:
+                self.manager.current_loaded_model.unload()
+
+        # model specific loading
+        self._load()
+
+        # declare ourselves as the currently loaded model
+        self.manager.current_loaded_model = self
+
+    @abc.abstractmethod
+    def _load(self):
+        """
+        Load the model
+        Do not call manually, use `load` instead.
+        """
+
+    def unload(self) -> None:
+        """
+        Unload the model within the model manager
+        """
+
+        # if we are not already loaded, stop
+        if not self._loaded:
+            return
+
+        # if we were the currently loaded model of the manager, demote ourselves
+        if self.manager.current_loaded_model is self:
+            self.manager.current_loaded_model = None
+
+        # model specific unloading part
+        self._unload()
+
+        # force the garbage collector to clean the memory
+        gc.collect()
+
+        # mark the model as unloaded
+        self._loaded = False
+
+    @abc.abstractmethod
+    def _unload(self):
+        """
+        Unload the model
+        Do not call manually, use `unload` instead.
+        :return:
+        """
+
+    def infer(self, payload: dict) -> str | bytes:
+        """
+        Infer our payload through the model within the model manager
+        :param payload: the payload to give to the model
+        :return: the response of the model
+        """
+
+        # make sure we are loaded before an inference
+        self.load()
+
+        # model specific inference part
+        return self._infer(payload)
+
+    @abc.abstractmethod
+    def _infer(self, payload: dict) -> str | bytes:
+        """
+        Infer our payload through the model
+        :param payload: the payload to give to the model
+        :return: the response of the model
+        """

source/model/base/__init__.py

@@ -0,0 +1 @@
+from .BaseModel import BaseModel