# this file used all classes and functions of project and create a flow. import os from utils.zoom_in import croped_images,image_enhancements from utils.distance import get_distances from utils.generate_result import get_json_data from object_detection.object_detection import ObjectDetection from activity_detection.activity_detection import ActivityDetection import os from dotenv import load_dotenv from pathlib import Path import logging env_path = Path('.') / '.env' load_dotenv(dotenv_path=env_path) path = { 'ACTIVITY_DET_MODEL_PATH':str(os.getenv('ACTIVITY_DET_MODEL_PATH')), 'OBJECT_DET_MODEL_PATH':str(os.getenv('OBJECT_DET_MODEL_PATH')), } # Configure the logger logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s", filename="logs.log", ) # Create a logger logger = logging.getLogger("pipline") def pipeline(image): """this function takes input as image from streamlit application then performs object detection,cropping image, image enhancement, activity detection, distance estimation and get final results in json format and returns to streamlit application. Args: image (numpy array): get numpy array of image which has 3 channels Returns: final_results: JSON Array which has below object { 'zoomed_img':np.array([]) , 'actual_boxes':[], 'merged_boundries':{}, } """ # detect object of given image using YOLO and get json_data of each object try: # Detect object from image. object_detection = ObjectDetection() logger.info("object detection object is created...") object_detection.set_trained_model_path(path['OBJECT_DET_MODEL_PATH']) logger.info("object detection model path is set...") object_json_data = object_detection.inference(image) logger.info("object detection done successfully...") # get croped_images list which has overlapping boundry box and also get croped single object images croped_images_list,single_object_images = croped_images(image,object_json_data) logger.info("cropping of image is done successfully...") # enhance images of both croped images and single object images enhanced_images,single_object_images = image_enhancements(croped_images_list,single_object_images) logger.info("enhancement of image is done successfully...") # detect activity of person object using image classification activity_detection = ActivityDetection() logger.info('activity detection object is created successfully...') activity_detection.set_trained_model_path(path['ACTIVITY_DET_MODEL_PATH']) logger.info('activity detection model is set') detected_activity = activity_detection.inference(single_object_images) logger.info("detection of activity is done successfully...") # Calculate distances of all objects distances_list = get_distances(object_json_data) logger.info("distance of object is calculated successfully...") # get final json array final_results = get_json_data(object_json_data,enhanced_images,detected_activity,distances_list) logger.info('final result of given image is created successfully...') return final_results except Exception as e: pass