planogram updated

Data/config.py

@@ -0,0 +1,11 @@
+from ultralytics import YOLO
+
+blankModel = YOLO('Models/blankDet_v1.1.pt')
+frameModel = YOLO('Models/seg_v1.1.pt')
+frame_shape = (416, 416)  # Set the input frame shape
+EXPAND_RATIO = 0.1  # Inner padding
+OUTER_PADDING_SIZE = 5  # Outer padding size
+seg_conf = 0.60
+det_conf = 0.50
+classes_to_delete = ['ghw']  
+expected_segments = ['ff', 'bg', 'sw', 'fns']

Models/best.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ead1f561a024f3d416b360a008ec196aa56a29e6a57dd2932c8fb9e5515f90e3
+size 87689203

Models/blankDet.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6b0d01752063565222f6ff4d9c656d2700bf60c1ee22b1713ce8c6abd703f0cb
+size 87685289

Models/blankDet_v1.1.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4e44928ff0203bded268c72d8e05fb6b9f827e6a5c759e4ab0b74514e3520a78
+size 52253474

Models/seg_v1.1.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6c7ea9a6887702e4998e222100a3d4e3a04d4838f24f42db587bff6680d5ae2c
+size 92340083

Utils/segment.py

@@ -0,0 +1,130 @@
+import cv2
+import os
+import uuid
+import numpy as np
+from Data.config import *
+
+class CropModel:
+
+    def __init__(self, model, model_input_shape=(frame_shape[0], frame_shape[1], 3)):
+        self.model = model
+        self.model_input_shape = model_input_shape
+        self.conf_threshold = seg_conf
+
+    def image_prediction_mask(self, image):
+        predict = self.model.predict(image)
+
+        # Extract the masks and move to CPU only if necessary
+        mask_tensor = predict[0].masks.data[0]  # Assuming this is the mask tensor
+        if mask_tensor.is_cuda:
+            mask = mask_tensor.cpu().numpy() * 255  # Convert to NumPy array after moving to CPU
+        else:
+            mask = mask_tensor.numpy() * 255  # No need to move to CPU if it's already on CPU
+
+        mask = mask.astype("uint8")
+
+        # Extract class IDs and class names from the predictions
+        class_ids_tensor = predict[0].boxes.cls
+        if class_ids_tensor.is_cuda:
+            class_ids = class_ids_tensor.cpu().numpy()  # Move to CPU
+        else:
+            class_ids = class_ids_tensor.numpy()
+
+        class_names = [self.model.names[int(cls_id)] for cls_id in class_ids]  # Convert indices to names
+
+        # If any class is found, return the first one
+        if class_names:
+            class_name = class_names[0]  # Return the first detected class
+        else:
+            class_name = "Not Found"  # If no class is detected
+
+        return mask, class_name
+
+    
+    @staticmethod
+    def get_mask_corner_points(mask):
+        _, thresh = cv2.threshold(mask, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
+        contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
+
+        if not contours:
+            return None
+
+        cnt = max(contours, key=cv2.contourArea)
+        cnt_approx = cv2.approxPolyDP(cnt, 0.03 * cv2.arcLength(cnt, True), True)
+
+        return cnt_approx.reshape((4, 2)) if len(cnt_approx) == 4 else None
+
+    @staticmethod
+    def get_order_points(points):
+        rect = np.zeros((4, 2), dtype="float32")
+        s = points.sum(axis=1)
+        diff = np.diff(points, axis=1)
+
+        rect[0] = points[np.argmin(s)]
+        rect[2] = points[np.argmax(s)]
+        rect[1] = points[np.argmin(diff)]
+        rect[3] = points[np.argmax(diff)]
+
+        return rect
+
+    @staticmethod
+    def expand_bounding_box(points, expand_ratio):
+        center = np.mean(points, axis=0)
+        expanded_points = points + (points - center) * expand_ratio
+        return expanded_points.astype("float32")
+
+    def point_transform(self, image, points):
+        ordered_points = self.get_order_points(points)
+        expanded_points = self.expand_bounding_box(ordered_points, EXPAND_RATIO)
+        height, width = image.shape[:2]
+
+        dst = np.array([[0, 0], [width - 1, 0], [width - 1, height - 1], [0, height - 1]], dtype="float32")
+        M = cv2.getPerspectiveTransform(expanded_points, dst)
+        warped_image = cv2.warpPerspective(image, M, (width, height))
+
+        return warped_image
+
+    @staticmethod
+    def add_padding_to_image(image, padding_size):
+        return cv2.copyMakeBorder(image, padding_size, padding_size, padding_size, padding_size, cv2.BORDER_CONSTANT, value=[0, 0, 0])
+
+    # def get_predicted_warped_image(self, image):
+    #     mask, class_name = self.image_prediction_mask(image)
+    #     corner_points = self.get_mask_corner_points(mask)
+
+    #     if corner_points is None:
+    #         return None, class_name
+
+    #     warped_image = self.point_transform(image, corner_points)
+    #     padded_image = self.add_padding_to_image(warped_image, OUTER_PADDING_SIZE)
+
+
+    #     return padded_image, class_name
+
+    def get_predicted_warped_image(self, image, save_dir="store"):
+        mask, class_name = self.image_prediction_mask(image)
+        corner_points = self.get_mask_corner_points(mask)
+
+        if corner_points is None:
+            return None, class_name
+
+        # Perform the transformation
+        warped_image = self.point_transform(image, corner_points)
+
+        # Add padding to the image
+        padded_image = self.add_padding_to_image(warped_image, OUTER_PADDING_SIZE)
+
+        # Ensure the directory exists
+        if not os.path.exists(save_dir):
+            os.makedirs(save_dir)
+
+        # Generate a random image name
+        random_name = f"image_{uuid.uuid4().hex[:8]}.jpg"  # First 8 characters of a UUID
+        save_path = os.path.join(save_dir, random_name)
+
+        # Save the padded image
+        cv2.imwrite(save_path, padded_image)
+        print(f"Image saved at: {save_path}")
+
+        return padded_image, class_name
+

Utils/sorting.py

@@ -0,0 +1,134 @@
+from Data.config import *
+import logging
+
+
+class SortModel:
+    def __init__(self, model, classes_to_delete=classes_to_delete, conf_threshold=det_conf, expected_segments=expected_segments):
+        self.model = model
+        self.classes_to_delete = classes_to_delete
+        self.conf_threshold = conf_threshold
+        self.expected_segments = expected_segments
+
+    def get_center(self, detection):
+        _, x_min, y_min, x_max, y_max = detection
+        center_x = (x_min + x_max) / 2
+        center_y = (y_min + y_max) / 2
+        return center_x, center_y
+
+    def sort_and_group_detections(self, detections):
+        detections_with_centers = [(d[0], *self.get_center(d)) for d in detections]
+        sorted_detections = sorted(detections_with_centers, key=lambda x: (x[1], x[2]))
+
+        if not sorted_detections:
+            return []
+
+        threshold_x = (sorted_detections[0][1] * 0.5)
+        rows = []
+        current_row = []
+        current_x = sorted_detections[0][1]
+
+        for detection in sorted_detections:
+            class_name, center_x, center_y = detection
+
+            if abs(center_x - current_x) > threshold_x:
+                rows.append(sorted(current_row, key=lambda x: x[2]))
+                current_row = []
+                current_x = center_x
+
+            current_row.append(detection)
+
+        if current_row:
+            rows.append(sorted(current_row, key=lambda x: x[2]))
+
+        max_columns = max(len(row) for row in rows) if rows else 0
+        grid_matrix = []
+        for row in rows:
+            grid_row = [d[0] for d in row]
+            grid_row.extend([''] * (max_columns - len(row)))
+            grid_matrix.append(grid_row)
+
+        transposed_matrix = list(map(list, zip(*grid_matrix)))
+        return transposed_matrix
+
+    @staticmethod
+    def sequence(matrix, expected_segments):
+        if not expected_segments:
+            return False
+
+        for sequence in matrix:
+            segment_index = 0
+            for item in sequence:
+                if item == expected_segments[segment_index]:
+                    continue
+                elif segment_index < len(expected_segments) - 1 and item == expected_segments[segment_index + 1]:
+                    segment_index += 1
+                else:
+                    return False
+            if segment_index != len(expected_segments) - 1:
+                return False
+        return True
+
+    def process_image(self, image_path, predicted_class):
+        # Run the detection on an image
+        results = self.model(image_path)
+    
+        planogram_ghw_count = 0
+        planogram_blanks_count = 0
+        planogram_ghw = ''
+        planogram_valid_sequence = ''
+        yolo_detections = []
+
+        for result in results:
+            for box in result.boxes:
+                class_name = self.model.names[int(box.cls[0])]
+                
+                if box.conf[0] >= self.conf_threshold:
+                    if class_name in self.classes_to_delete:
+                        planogram_ghw_count += 1
+                        continue      
+
+                if box.conf[0] >= self.conf_threshold:
+                    x_min, y_min, x_max, y_max = box.xyxy[0]
+                    yolo_detections.append((class_name, x_min.item(), y_min.item(), x_max.item(), y_max.item()))
+        
+        planogram_blanks_count = len(yolo_detections)
+
+        grid_matrix = self.sort_and_group_detections(yolo_detections)
+        
+        # Print the matrix
+        print("Grid Matrix:")
+        for row in grid_matrix:
+            print(row)
+        print("\n")
+
+        if planogram_blanks_count == planogram_ghw_count:
+            planogram_ghw = "yes"
+        else:
+            planogram_ghw = "no"
+
+        planogram_valid_sequence = self.sequence(grid_matrix, self.expected_segments)
+        if planogram_valid_sequence:
+            planogram_valid_sequence = "yes"
+        else:
+            planogram_valid_sequence = "no"
+
+        return {
+            "Planogram Blanks Count": planogram_blanks_count,
+            "Planogram GHW Count": planogram_ghw_count,
+            "Planogram Valid Sequence": planogram_valid_sequence,
+            "Planogram GHW": planogram_ghw,
+            "Planogram Name": predicted_class,
+
+            "POSM Blanks Count": "N/A",
+            "POSM GHW Count": "N/A",
+            "POSM Valid Sequence": "N/A",
+            "POSM GHW": "N/A",
+            "POSM Name": "N/A",
+        }
+
+
+
+
+
+
+

api.py

@@ -0,0 +1,76 @@
+from fastapi import FastAPI
+from pydantic import BaseModel
+from typing import Union, List
+import uvicorn
+import logging
+from datetime import datetime
+import pytz
+import torch
+
+from main import ImageProcessor
+image_processor = ImageProcessor()  # Initialize the image processor
+
+logging.basicConfig(filename="drinksLog.log", filemode='w')
+logger = logging.getLogger("drinks")
+logger.setLevel(logging.DEBUG)
+file_handler = logging.FileHandler("drinksLog.log")
+logger.addHandler(file_handler)
+
+app = FastAPI()
+
+class RequestBody(BaseModel):
+    cat: str
+    img: str
+
+class RequestData(BaseModel):
+    body: Union[RequestBody, List[RequestBody]]
+
+@app.get("/status")
+async def status():
+    return {"status": "AI Server is running"}
+
+# Function to process the image based on the category
+async def process_image(item: RequestBody):
+    category = item.cat
+    img_url = item.img
+
+    if category == "posm":
+        result = await image_processor.process_image(img_url)
+        return result
+    elif category == "planogram":
+        result = await image_processor.process_image(img_url)
+        return result
+    else:
+        return {"error": f"Unsupported category {category}"}
+
+
+@app.post("/bats")
+async def detect_items(request_data: RequestData):
+    try:
+        # Initialize an empty list to hold the processed results
+        results = []
+
+        # Check if data.body is a list or a single item
+        if isinstance(request_data.body, list):
+            # If body is already a list, iterate through each item
+            for item in request_data.body:
+                # Process each item and append the result to results
+                processed_result = await process_image(item)  # Use await here
+                results.append(processed_result)
+        else:
+            # If body is a single item, process it directly
+            processed_result = await process_image(request_data.body)  # Use await here
+            results.append(processed_result)
+        
+        # Return the results as a JSON response
+        return results
+
+    except Exception as e:
+        logger.error(f"Error during detection: {str(e)}")
+        return {"error": "An error occurred during detection"}
+
+if __name__ == "__main__":
+    try:
+        uvicorn.run(app, host="127.0.0.1", port=4444)
+    finally:
+        torch.cuda.empty_cache()

main.py

@@ -0,0 +1,71 @@
+# main.py
+import json
+import pandas as pd
+from PIL import Image
+import cv2
+import numpy as np
+from aiohttp import ClientSession
+from io import BytesIO
+import asyncio
+import logging
+from Data.config import *
+from Utils.segment import CropModel
+from Utils.sorting import SortModel
+from Data.config import frameModel
+from Data.config import blankModel
+import matplotlib.pyplot as plt
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+
+class ImageFetcher:
+    @staticmethod
+    async def fetch_image(url, session):
+        try:
+            async with session.get(url) as response:
+                if response.status == 200:
+                    img_data = await response.read()
+                    return Image.open(BytesIO(img_data))
+                else:
+                    logging.error(f"Failed to fetch image from {url}, status code: {response.status}")
+                    return None
+        except Exception as e:
+            logging.error(f"Exception during image fetching from {url}: {e}")
+            return None
+
+
+class ImageProcessor:
+    def __init__(self):
+        self.frameModel = frameModel
+        self.blankModel = blankModel
+
+    async def process_image(self, img_url):
+        async with ClientSession() as session:
+            image = await ImageFetcher.fetch_image(img_url, session)
+            if image is None:
+                return {"error": "Failed to fetch image"}
+
+            image = np.array(image)  # Convert PIL image to NumPy array if needed
+            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+            image = cv2.resize(image, (frame_shape[0], frame_shape[1]))  # Resize to match the frame shape
+
+            try:
+                crop_model = CropModel(model=frameModel)  # Initialize with your model
+                padded_image, other_class_name = crop_model.get_predicted_warped_image(image)
+
+                if other_class_name:
+                    print(f"Other class name: {other_class_name}")
+                else:
+                    print("No other class detected.")
+
+                # plt.imshow(padded_image)
+                # plt.axis('off')  # Hide axes
+                # plt.show() 
+
+                sort_model = SortModel(model=blankModel, classes_to_delete=classes_to_delete, conf_threshold=det_conf, expected_segments=expected_segments) 
+                sequence_str = sort_model.process_image(padded_image, other_class_name)
+                
+                return {"result": sequence_str}
+            except Exception as e:
+                logging.error(f"Exception during image processing: {e}")
+                return {"error": "Image processing failed."}

note.py

@@ -0,0 +1,122 @@
+from Data.config import *
+import logging
+
+
+class SortModel:
+    def __init__(self, model, classes_to_delete=classes_to_delete, conf_threshold=det_conf, expected_segments=expected_segments):
+        self.model = model
+        self.classes_to_delete = classes_to_delete
+        self.conf_threshold = conf_threshold
+        self.expected_segments = expected_segments
+
+    def get_center(self, detection):
+        _, x_min, y_min, x_max, y_max = detection
+        center_x = (x_min + x_max) / 2
+        center_y = (y_min + y_max) / 2
+        return center_x, center_y
+
+    def sort_and_group_detections(self, detections):
+        detections_with_centers = [(d[0], *self.get_center(d)) for d in detections]
+        sorted_detections = sorted(detections_with_centers, key=lambda x: (x[1], x[2]))
+
+        if not sorted_detections:
+            return []
+
+        threshold_x = (sorted_detections[0][1] * 0.5)
+        rows = []
+        current_row = []
+        current_x = sorted_detections[0][1]
+
+        for detection in sorted_detections:
+            class_name, center_x, center_y = detection
+
+            if abs(center_x - current_x) > threshold_x:
+                rows.append(sorted(current_row, key=lambda x: x[2]))
+                current_row = []
+                current_x = center_x
+
+            current_row.append(detection)
+
+        if current_row:
+            rows.append(sorted(current_row, key=lambda x: x[2]))
+
+        max_columns = max(len(row) for row in rows) if rows else 0
+        grid_matrix = []
+        for row in rows:
+            grid_row = [d[0] for d in row]
+            grid_row.extend([''] * (max_columns - len(row)))
+            grid_matrix.append(grid_row)
+
+        transposed_matrix = list(map(list, zip(*grid_matrix)))
+        return transposed_matrix
+
+    @staticmethod
+    def sequence(matrix, expected_segments):
+        if not expected_segments:
+            return False
+
+        for sequence in matrix:
+            segment_index = 0
+            for item in sequence:
+                if item == expected_segments[segment_index]:
+                    continue
+                elif segment_index < len(expected_segments) - 1 and item == expected_segments[segment_index + 1]:
+                    segment_index += 1
+                else:
+                    return False
+            if segment_index != len(expected_segments) - 1:
+                return False
+        return True
+
+    def process_image(self, image_path, predicted_class):
+        # Run the detection on an image
+        results = self.model(image_path)
+    
+        planogram_ghw_count = 0
+        planogram_blanks_count = 0
+        planogram_ghw = ''
+        planogram_valid_sequence = ''
+        yolo_detections = []
+
+        for result in results:
+            for box in result.boxes:
+                class_name = self.model.names[int(box.cls[0])]
+                
+                if class_name in self.classes_to_delete:
+                    planogram_ghw_count += 1
+                    continue
+
+                if box.conf[0] >= self.conf_threshold:
+                    x_min, y_min, x_max, y_max = box.xyxy[0]
+                    yolo_detections.append((class_name, x_min.item(), y_min.item(), x_max.item(), y_max.item()))
+        
+        planogram_blanks_count = len(yolo_detections)
+
+        # Sort and group detections
+        grid_matrix = self.sort_and_group_detections(yolo_detections)
+        print(f"ddddddddd{yolo_detections}")
+        
+        # Print the matrix
+        print("Grid Matrix:")
+        for row in grid_matrix:
+            print(row)
+        print("\n")
+
+        if planogram_blanks_count == planogram_ghw_count:
+            planogram_ghw = "yes"
+        else:
+            planogram_ghw = "no"
+
+        planogram_valid_sequence = self.sequence(grid_matrix, self.expected_segments)
+        if planogram_valid_sequence:
+            planogram_valid_sequence = "y"
+        else:
+            planogram_valid_sequence = "n/a"
+
+        return {
+            "Planogram Blanks Count": planogram_blanks_count,
+            "Planogram GHW Count": planogram_ghw_count,
+            "Planogram Valid Sequence": planogram_valid_sequence,
+            "Planogram GHW": planogram_ghw,
+            "Class Name": predicted_class
+        }