Update app.py

app.py

@@ -2,7 +2,302 @@ import sys
 sys.stdout.reconfigure(line_buffering=True) 
 
 
-from pipeline import app
+import os
+import numpy as np
+import requests
+import cv2
+from skimage import feature
+from io import BytesIO
+import traceback
 
+from flask import Flask, request, jsonify
+from PIL import Image
+
+# import deep learning libraries 
+import torch
+from transformers import AutoProcessor, AutoModelForZeroShotObjectDetection, AutoTokenizer, AutoModel
+from segment_anything import SamPredictor, sam_model_registry
+
+app = Flask(__name__)
+
+# sum = 1
+FEATURE_WEIGHTS = {
+    "shape": 0.4,
+    "color": 0.5,
+    "texture": 0.1
+}
+
+# threshold
+FINAL_SCORE_THRESHOLD = 0.5
+
+
+# load all models
+print("="*50)
+print("🚀 Initializing application and loading models...")
+device_name = os.environ.get("device", "cpu")
+device = torch.device('cuda' if 'cuda' in device_name and torch.cuda.is_available() else 'cpu')
+print(f"🧠 Using device: {device}")
+
+print("...Loading Grounding DINO model...")
+gnd_model_id = "IDEA-Research/grounding-dino-tiny"
+processor_gnd = AutoProcessor.from_pretrained(gnd_model_id)
+model_gnd = AutoModelForZeroShotObjectDetection.from_pretrained(gnd_model_id).to(device)
+
+print("...Loading Segment Anything (SAM) model...")
+sam_checkpoint = "sam_vit_b_01ec64.pth" 
+sam_model = sam_model_registry["vit_b"](checkpoint=sam_checkpoint).to(device)
+predictor = SamPredictor(sam_model)
+
+print("...Loading BGE model for text embeddings...")
+bge_model_id = "BAAI/bge-small-en-v1.5"
+tokenizer_text = AutoTokenizer.from_pretrained(bge_model_id)
+model_text = AutoModel.from_pretrained(bge_model_id).to(device)
+print("✅ All models loaded successfully.")
+print("="*50)
+
+
+# helper functions
+
+def get_canonical_label(object_name_phrase: str) -> str:
+    print(f"\n  [Label] Extracting label for: '{object_name_phrase}'")
+    label = object_name_phrase.strip().lower().split()[-1]
+    label = ''.join(filter(str.isalpha, label))
+    print(f"  [Label] ✅ Extracted label: '{label}'")
+    return label if label else "unknown"
+
+def download_image_from_url(image_url: str) -> Image.Image:
+    print(f"  [Download] Downloading image from: {image_url[:80]}...")
+    response = requests.get(image_url)
+    response.raise_for_status()
+    image = Image.open(BytesIO(response.content))
+    image_rgb = image.convert("RGB")
+    print("  [Download] ✅ Image downloaded and standardized to RGB.")
+    return image_rgb
+
+def detect_and_crop(image: Image.Image, object_name: str) -> Image.Image:
+    print(f"\n  [Detect & Crop] Starting detection for object: '{object_name}'")
+    image_np = np.array(image.convert("RGB"))
+    height, width = image_np.shape[:2]
+    prompt = [[f"a {object_name}"]]
+    inputs = processor_gnd(images=image, text=prompt, return_tensors="pt").to(device)
+    with torch.no_grad():
+        outputs = model_gnd(**inputs)
+    results = processor_gnd.post_process_grounded_object_detection(
+        outputs, inputs.input_ids, box_threshold=0.4, text_threshold=0.3, target_sizes=[(height, width)]
+    )
+    if not results or len(results[0]['boxes']) == 0:
+        print("  [Detect & Crop] ⚠ Warning: Grounding DINO did not detect the object. Using full image.")
+        return image
+    result = results[0]
+    scores = result['scores']
+    max_idx = int(torch.argmax(scores))
+    box = result['boxes'][max_idx].cpu().numpy().astype(int)
+    print(f"  [Detect & Crop] ✅ Object detected with confidence: {scores[max_idx]:.2f}, Box: {box}")
+    x1, y1, x2, y2 = box
+    
+    predictor.set_image(image_np)
+    box_prompt = np.array([[x1, y1, x2, y2]])
+    masks, _, _ = predictor.predict(box=box_prompt, multimask_output=False)
+    mask = masks[0]
+    
+    mask_bool = mask > 0
+    cropped_img_rgba = np.zeros((height, width, 4), dtype=np.uint8)
+    cropped_img_rgba[:, :, :3] = image_np
+    cropped_img_rgba[:, :, 3] = mask_bool * 255
+    
+    cropped_img_rgba = cropped_img_rgba[y1:y2, x1:x2]
+
+    object_image = Image.fromarray(cropped_img_rgba, 'RGBA')
+    return object_image
+
+def extract_features(segmented_image: Image.Image) -> dict:
+    image_rgba = np.array(segmented_image)
+    if image_rgba.shape[2] != 4:
+        raise ValueError("Segmented image must be RGBA")
+    
+    b, g, r, a = cv2.split(image_rgba)
+    image_rgb = cv2.merge((b, g, r))
+    mask = a
+    
+    gray = cv2.cvtColor(image_rgb, cv2.COLOR_BGR2GRAY)
+    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    hu_moments = cv2.HuMoments(cv2.moments(contours[0])).flatten() if contours else np.zeros(7)
+    
+    color_hist = cv2.calcHist([image_rgb], [0, 1, 2], mask, [8, 8, 8], [0, 256, 0, 256, 0, 256])
+    cv2.normalize(color_hist, color_hist)
+    color_hist = color_hist.flatten()
+    
+    gray_masked = cv2.bitwise_and(gray, gray, mask=mask)
+    lbp = feature.local_binary_pattern(gray_masked, P=24, R=3, method="uniform")
+    (texture_hist, _) = np.histogram(lbp.ravel(), bins=np.arange(0, 27), range=(0, 26))
+    texture_hist = texture_hist.astype("float32")
+    texture_hist /= (texture_hist.sum() + 1e-6)
+    
+    return {
+        "shape_features": hu_moments.tolist(),
+        "color_features": color_hist.tolist(),
+        "texture_features": texture_hist.tolist()
+    }
+
+def get_text_embedding(text: str) -> list:
+    print(f"  [Embedding] Generating text embedding for: '{text[:50]}...'")
+    text_with_instruction = f"Represent this sentence for searching relevant passages: {text}"
+    inputs = tokenizer_text(text_with_instruction, return_tensors='pt', padding=True, truncation=True, max_length=512).to(device)
+    with torch.no_grad():
+        outputs = model_text(**inputs)
+    embedding = outputs.last_hidden_state[:, 0, :]
+    embedding = torch.nn.functional.normalize(embedding, p=2, dim=1)
+    print("  [Embedding] ✅ Text embedding generated.")
+    return embedding.cpu().numpy()[0].tolist()
+
+def cosine_similarity(vec1: np.ndarray, vec2: np.ndarray) -> float:
+    return float(np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2)))
+
+# API endpoints
+
+@app.route('/process', methods=['POST'])
+def process_item():
+    """
+    Receives item details, processes them, and returns all computed features.
+    This is called when a new item is created in the Node.js backend.
+    """
+    print("\n" + "="*50)
+    print("➡ [Request] Received new request to /process")
+    try:
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "Invalid JSON payload"}), 400
+
+        object_name = data.get('objectName')
+        description = data.get('objectDescription')
+        image_url = data.get('objectImage') # This can now be null
+
+        if not all([object_name, description]):
+            return jsonify({"error": "objectName and objectDescription are required."}), 400
+
+        # process text based features
+        canonical_label = get_canonical_label(object_name)
+        text_embedding = get_text_embedding(description)
+
+        response_data = {
+            "canonicalLabel": canonical_label,
+            "text_embedding": text_embedding,
+        }
+
+        # process visual features ONLY if an image_url is provided
+        if image_url:
+            print("--- Image URL provided, processing visual features... ---")
+            image = download_image_from_url(image_url)
+            object_crop = detect_and_crop(image, canonical_label)
+            visual_features = extract_features(object_crop)
+            # Add visual features to the response
+            response_data.update(visual_features)
+        else:
+            print("--- No image URL provided, skipping visual feature extraction. ---")
+        
+        print("✅ Successfully processed item.")
+        print("="*50)
+        return jsonify(response_data), 200
+
+    except Exception as e:
+        print(f"❌ Error in /process: {e}")
+        traceback.print_exc()
+        return jsonify({"error": str(e)}), 500
+
+def stretch_image_score(score):
+    if score < 0.4 or score == 1.0:
+        return score
+    # increase confidence
+    return 0.7 + (score - 0.4) * (0.99 - 0.7) / (1.0 - 0.4)
+
+@app.route('/compare', methods=['POST'])
+def compare_items():
+    print("\n" + "="*50)
+    print("➡ [Request] Received new request to /compare")
+    try:
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "Invalid JSON payload"}), 400
+        
+        query_item = data.get('queryItem')
+        search_list = data.get('searchList')
+
+        if not all([query_item, search_list]):
+            return jsonify({"error": "queryItem and searchList are required."}), 400
+        
+        query_text_emb = np.array(query_item['text_embedding'])
+        results = []
+        print(f"--- Comparing 1 query item against {len(search_list)} items ---")
+
+        for item in search_list:
+            item_id = item.get('_id')
+            print(f"\n  [Checking] Item ID: {item_id}")
+            try:
+                # Text comparison is always done
+                text_emb_found = np.array(item['text_embedding'])
+                text_score = cosine_similarity(query_text_emb, text_emb_found)
+                print(f"     - Text Score: {text_score:.4f}")
+
+                # --- NEW: Check if BOTH items have visual features ---
+                has_query_image = 'shape_features' in query_item and query_item['shape_features']
+                has_item_image = 'shape_features' in item and item['shape_features']
+
+                if has_query_image and has_item_image:
+                    print("     - Both items have images. Performing visual comparison.")
+                    # If both have images, proceed with full comparison
+                    query_shape_feat = np.array(query_item['shape_features'])
+                    query_color_feat = np.array(query_item['color_features']).astype("float32")
+                    query_texture_feat = np.array(query_item['texture_features']).astype("float32")
+                    
+                    found_shape = np.array(item['shape_features'])
+                    found_color = np.array(item['color_features']).astype("float32")
+                    found_texture = np.array(item['texture_features']).astype("float32")
+
+                    shape_dist = cv2.matchShapes(query_shape_feat, found_shape, cv2.CONTOURS_MATCH_I1, 0.0)
+                    shape_score = 1.0 / (1.0 + shape_dist)
+                    color_score = cv2.compareHist(query_color_feat, found_color, cv2.HISTCMP_CORREL)
+                    texture_score = cv2.compareHist(query_texture_feat, found_texture, cv2.HISTCMP_CORREL)
+
+                    raw_image_score = (FEATURE_WEIGHTS["shape"] * shape_score +
+                                   FEATURE_WEIGHTS["color"] * color_score +
+                                   FEATURE_WEIGHTS["texture"] * texture_score)
+
+                    image_score = stretch_image_score(raw_image_score)
+                    
+                    # Weighted average of image and text scores
+                    final_score = 0.4 * image_score + 0.6 * text_score
+                    print(f"     - Image Score: {image_score:.4f} | Final Score: {final_score:.4f}")
+                
+                else:
+                    # If one or both items lack an image, the final score is JUST the text score
+                    print("     - One or both items missing image. Using text score only.")
+                    final_score = text_score
+
+                # Check if the final score meets the threshold
+                if final_score >= FINAL_SCORE_THRESHOLD:
+                    print(f"     - ✅ ACCEPTED (Score >= {FINAL_SCORE_THRESHOLD})")
+                    results.append({
+                        "_id": item_id,
+                        "score": round(final_score, 4),
+                        "objectName": item.get("objectName"),
+                        "objectDescription": item.get("objectDescription"),
+                        "objectImage": item.get("objectImage"),
+                    })
+                else:
+                    print(f"     - ❌ REJECTED (Score < {FINAL_SCORE_THRESHOLD})")
+
+            except Exception as e:
+                print(f"  [Skipping] Item {item_id} due to processing error: {e}")
+                continue
+        
+        results.sort(key=lambda x: x["score"], reverse=True)
+        print(f"\n✅ Search complete. Found {len(results)} potential matches.")
+        print("="*50)
+        return jsonify({"matches": results}), 200
+
+    except Exception as e:
+        print(f"❌ Error in /compare: {e}")
+        traceback.print_exc()
+        return jsonify({"error": str(e)}), 500
 if __name__ == '__main__':
     app.run(host='0.0.0.0', port=7860)