src/cache.py

import logging
from multiprocessing import Lock

import torch
import gc
import os

logging.basicConfig(level=logging.INFO)

model_paths = ["./model/model.pth", "../model/model.pth"]


class Singleton:
    """
    Tek bir model örneğini yönetmek için Singleton sınıfı.

    Bu sınıf, yalnızca bir model örneğinin oluşturulmasını sağlar ve modeli yükleme ve serbest bırakma yöntemlerini sağlar.

    Özellikler:
        model_lock (Lock): Model erişimini thread güvenli hale getirmek için bir kilit.
        _device (str): Modelin yükleneceği cihaz (CPU veya CUDA).
        _instance (Singleton): Singleton sınıfının tek bir örneği.
        _model (torch.nn.Module): Yüklenen model örneği.
        __FP16 (bool): Modelin FP16 modunda yüklenip yüklenmediğini gösterir.

    Metodlar:
        __init__(): Singleton örneğini başlatır.
        __new__(): Zaten bir örneği varsa Singleton sınıfının yeni bir örneğini oluşturur.
        load_model(verbose=False): Modeli belirtilen model yolundan yükler.
        release_model(): Yüklenen modeli serbest bırakır.
        get_fp16(): Modelin FP16 modunda yüklenip yüklenmediğini döndürür.
        get_device(): Modelin yüklendiği cihazı döndürür.
    """
    model_lock = Lock()
    _device = None
    _instance = None
    _model = None

    def __init__(self):
        """
        Singleton örneğini başlatır.

        CUDA'nın mevcudiyetine bağlı olarak cihazı ayarlar ve FP16 bayrağını başlatır.
        """
        self.__FP16 = False
        device = "cuda" if torch.cuda.is_available() else "cpu"
        logging.info(f"Device: {device} {torch.version.cuda} {torch.cuda.get_arch_list()}")
        if device == "cuda":
            self.__FP16 = True
        self._device = device

    def __new__(cls):
        """
        Zaten bir örneği varsa Singleton sınıfının yeni bir örneğini oluşturur.

        Returns:
            Singleton: Singleton sınıfının örneği.
        """
        if cls._instance is None:
            cls._instance = super(Singleton, cls).__new__(cls)
        return cls._instance

    def load_model(self, verbose=False):
        """
        Modeli belirtilen model yolundan yükler.

        Parametreler:
            verbose (bool): İşlemleri ayrıntılı olarak loglar.

        Returns:
            torch.nn.Module: Yüklenen model örneği.
        """
        with self.model_lock:
            if self._model is not None:
                if verbose:
                    logging.info(f"Model mevcut.")
            else:
                logging.info(f"Model mevcut değil. Yükleniyor...")
                torch.device(self._device)
                for model_path in model_paths:
                    if os.path.exists(model_path):
                        self._model = torch.load(model_path, map_location=torch.device(self._device))
                self._model.eval()
                if torch.cuda.is_available():
                    logging.info(f"Model Loaded on {self._device}. Allocated memory: {torch.cuda.memory_allocated()}")
                else:
                    logging.info(f"Model Loaded on {self._device}.")
        return self._model

    def release_model(self):
        """
        Yüklenen modeli serbest bırakır.
        """
        with self.model_lock:
            if self._model is not None:
                logging.info(f"Model serbest bırakılıyor...")
                if self._model:
                    del self._model
                gc.collect()
                if torch.cuda.is_available():
                    torch.cuda.empty_cache()
                    torch.cuda.synchronize(self._device)
                    logging.info(f"Model {self._device} üzerinde serbest bırakıldı. Tahsis edilen bellek: {torch.cuda.memory_allocated()}")
                else:
                    logging.info(f"Model {self._device} üzerinde serbest bırakıldı.")
            else:
                logging.info("Serbest bırakılacak model bulunamadı.")

    def get_fp16(self):
        return self.__FP16

    def get_device(self):
        return self._device