Fertilizer-Recommendation/src/predictor.py

from src.entity.config_entity import TRANSFORMER_OBJECT_FILE_NAME
from src.entity.config_entity import MODEL_FILE_NAME
from src.entity.config_entity import TARGET_ENCODER_OBJECT_FILE_NAME
from src.exception import FertilizerException
from src.logger import logging 

import os 
import sys 
from glob import glob
from typing import Optional

class ModelResolver:

    def __init__(
        self, 
        model_registry: str = 'saved_models', 
        transformer_dir_name = 'transformer',
        target_encoder_dir_name = 'target_encoder',
        model_dir_name = 'model'
    ):

        self.model_registry = model_registry
        os.makedirs(self.model_registry, exist_ok=True)

        self.transformer_dir_name = transformer_dir_name
        self.target_encoder_dir_name = target_encoder_dir_name
        self.model_dir_name = model_dir_name

    def get_latest_dir_path(self) ->Optional[str]:
        try:
            dir_names = os.listdir(self.model_registry)
            
            if len(dir_names) == 0:
                return None 
            dir_names = list(map(int, dir_names))
            latest_dir_name = max(dir_names)
            
            return os.path.join(self.model_registry, f"{latest_dir_name}")

        except Exception as e:
            raise FertilizerException(e, sys)
    
    def get_latest_model_path(self):
        try:
            latest_dir = self.get_latest_dir_path()
            
            if latest_dir is None:
                raise Exception(f"Model is not available")
            
            return os.path.join(latest_dir, self.model_dir_name, MODEL_FILE_NAME)

        except Exception as e:
            raise FertilizerException(e, sys)

    def get_latest_transformer_path(self):
        try:
            latest_dir = self.get_latest_dir_path()
            if latest_dir is None:
                raise Exception(f"Transformer is not available")
            
            return os.path.join(latest_dir, self.transformer_dir_name, TRANSFORMER_OBJECT_FILE_NAME)

        except Exception as e:
            raise FertilizerException(e, sys)
    
    def get_latest_target_encoder_path(self):
        try:
            latest_dir = self.get_latest_dir_path()
            if latest_dir is None:
                raise Exception(f"Target Encoder is not available")

            return os.path.join(latest_dir, self.target_encoder_dir_name, TARGET_ENCODER_OBJECT_FILE_NAME)

        except Exception as e:
            raise FertilizerException(e, sys)
    
    def get_latest_save_dir_path(self):
        try:
            latest_dir = self.get_latest_dir_path()

            if latest_dir is None:
                return os.path.join(self.model_registry, f"{0}")
            
            latest_dir_num = int(os.path.basename(self.get_latest_dir_path()))

            return os.path.join(self.model_registry, f"{latest_dir_num + 1}")

        except Exception as e:
            raise FertilizerException(e, sys)

    def get_latest_save_model_path(self):
        try:
            latest_dir = self.get_latest_save_dir_path()
            
            return os.path.join(latest_dir, self.model_dir_name, MODEL_FILE_NAME)

        except Exception as e:
            raise FertilizerException(e, sys)

    def get_latest_save_transformer_path(self):
        try:
            latest_dir = self.get_latest_save_dir_path()

            return os.path.join(latest_dir, self.transformer_dir_name, TRANSFORMER_OBJECT_FILE_NAME)
        
        except Exception as e:
            raise FertilizerException(e, sys)

    def get_latest_save_target_encoder_path(self):
        try:
            latest_dir = self.get_latest_save_dir_path()

            return os.path.join(latest_dir, self.target_encoder_dir_name, TARGET_ENCODER_OBJECT_FILE_NAME)

        except Exception as e:
            raise FertilizerException(e, sys)