Update app.py

app.py

@@ -703,146 +703,6 @@ def choose_top_candidates(embedding_results, phash_results, imgmatch_results, to
 
     return result
 
-    
-def hybrid_similarity_matching(sprite_images_bytes, sprite_ids, min_similarity=None, top_k=5, method_weights=(0.5,0.3,0.2)):
-    from PIL import Image
-    # Local safe defaults
-    embeddings_path = os.path.join(BLOCKS_DIR, "hybrid_embeddings.json")
-    hash_path = os.path.join(BLOCKS_DIR, "phash_data.json")
-    signature_path = os.path.join(BLOCKS_DIR, "signature_data.json")
-
-    # Load embeddings
-    embedding_json = {}
-    if os.path.exists(embeddings_path):
-        with open(embeddings_path, "r", encoding="utf-8") as f:
-            embedding_json = json.load(f)
-
-    # Load phash data (if exists) -> ensure hash_dict variable exists
-    hash_dict = {}
-    if os.path.exists(hash_path):
-        try:
-            with open(hash_path, "r", encoding="utf-8") as f:
-                hash_data = json.load(f)
-                for path, hash_str in hash_data.items():
-                    try:
-                        hash_dict[path] = hash_str
-                    except Exception:
-                        pass
-        except Exception:
-            pass
-
-    # Load signature data (if exists) -> ensure signature_dict exists
-    signature_dict = {}
-    sig_data = {}
-    if os.path.exists(signature_path):
-        try:
-            with open(signature_path, "r", encoding="utf-8") as f:
-                sig_data = json.load(f)
-                for path, sig_list in sig_data.items():
-                    try:
-                        signature_dict[path] = np.array(sig_list)
-                    except Exception:
-                        pass
-        except Exception:
-            pass
-
-    # Parse embeddings into lists
-    paths_list = []
-    embeddings_list = []
-    if isinstance(embedding_json, dict):
-        for p, emb in embedding_json.items():
-            if isinstance(emb, dict):
-                maybe_emb = emb.get("embedding") or emb.get("embeddings") or emb.get("emb")
-                if maybe_emb is None:
-                    continue
-                arr = np.asarray(maybe_emb, dtype=np.float32)
-            elif isinstance(emb, list):
-                arr = np.asarray(emb, dtype=np.float32)
-            else:
-                continue
-            paths_list.append(os.path.normpath(str(p)))
-            embeddings_list.append(arr)
-    elif isinstance(embedding_json, list):
-        for item in embedding_json:
-            if not isinstance(item, dict):
-                continue
-            p = item.get("path") or item.get("image_path") or item.get("file") or item.get("filename") or item.get("img_path")
-            emb = item.get("embeddings") or item.get("embedding") or item.get("features") or item.get("vector") or item.get("emb")
-            if p is None or emb is None:
-                continue
-            paths_list.append(os.path.normpath(str(p)))
-            embeddings_list.append(np.asarray(emb, dtype=np.float32))
-
-    if len(paths_list) == 0:
-        print("⚠ No reference images/embeddings found (this test harness may be running without data)")
-        # Return empty results gracefully
-        return [[] for _ in sprite_images_bytes], [[] for _ in sprite_images_bytes], []
-
-    ref_matrix = np.vstack(embeddings_list).astype(np.float32)
-    
-    # Batch: Get all sprite embeddings, phash, sigs first
-    sprite_emb_list = []
-    sprite_phash_list = []
-    sprite_sig_list = []
-    per_sprite_final_indices = []
-    per_sprite_final_scores = []
-    per_sprite_rerank_debug = []
-    for i, sprite_bytes in enumerate(sprite_images_bytes):
-        sprite_pil = Image.open(sprite_bytes)
-        enhanced_sprite = process_image_cv2_from_pil(sprite_pil, scale=2) or sprite_pil
-        # sprite_emb = get_dinov2_embedding_from_pil(preprocess_for_model(enhanced_sprite)) or np.zeros(ref_matrix.shape[1])
-        # sprite_emb_list.append(sprite_emb)
-        sprite_emb = get_dinov2_embedding_from_pil(preprocess_for_model(enhanced_sprite))
-        sprite_emb = sprite_emb if sprite_emb is not None else np.zeros(ref_matrix.shape[1])
-        sprite_emb_list.append(sprite_emb)
-        # Perceptual hash
-        sprite_hash_arr = preprocess_for_hash(enhanced_sprite)
-        sprite_phash = None
-        if sprite_hash_arr is not None:
-            try: sprite_phash = phash.encode_image(image_array=sprite_hash_arr)
-            except: pass
-        sprite_phash_list.append(sprite_phash)
-        # Signature
-        sprite_sig = None
-        embedding_results, phash_results, imgmatch_results, combined_results = run_query_search_flow(
-            query_b64=sprite_b64_clean[i],
-            processed_dir=BLOCKS_DIR,
-            embeddings_dict=embedding_json,
-            hash_dict=hash_data,
-            signature_obj_map=sig_data,
-            gis=gis,
-            phash=phash,
-            MAX_PHASH_BITS=64,
-            k=5
-        )
-        # Call the advanced re-ranker
-        rerank_result = choose_top_candidates(embedding_results, phash_results, imgmatch_results,
-                                            top_k=top_k, method_weights=method_weights, verbose=True)
-        per_sprite_rerank_debug.append(rerank_result)
-
-        # Selection logic: prefer consensus, else weighted top-1
-        final = None
-        if len(rerank_result["consensus_topk"]) > 0:
-            consensus = rerank_result["consensus_topk"]
-            best = max(consensus, key=lambda p: rerank_result["weighted_scores_full"].get(p, 0.0))
-            final = best
-        else:
-            final = rerank_result["weighted_topk"][0][0] if rerank_result["weighted_topk"] else None
-
-        # Store index and score for downstream use
-        if final is not None and final in paths_list:
-            idx = paths_list.index(final)
-            score = rerank_result["weighted_scores_full"].get(final, 0.0)
-            per_sprite_final_indices.append([idx])
-            per_sprite_final_scores.append([score])
-            print(f"Sprite '{sprite_ids}' FINAL selected: {final} (index {idx}) score={score:.4f}")
-        else:
-            per_sprite_final_indices.append([])
-            per_sprite_final_scores.append([])
-
-    return per_sprite_final_indices, per_sprite_final_scores, paths_list#, per_sprite_rerank_debug
-    
-
 def is_subpath(path: str, base: str) -> bool:
     """Return True if path is inside base (works across OSes)."""
     try:
@@ -1847,7 +1707,6 @@ def similarity_matching(sprites_data: dict, project_folder: str, top_k: int = 1,
     code_blocks_path = r"D:\DEV PATEL\2025\scratch_VLM\scratch_agent\blocks\code_blocks"
     # out_path = r"D:\DEV PATEL\2025\scratch_VLM\scratch_agent\blocks\out_json"
     
-
     project_json_path = os.path.join(project_folder, "project.json")
 
     # -------------------------
@@ -1871,6 +1730,145 @@ def similarity_matching(sprites_data: dict, project_folder: str, top_k: int = 1,
         img.save(buffer, format="PNG")
         buffer.seek(0)
         sprite_images_bytes.append(buffer)
+            
+    def hybrid_similarity_matching(sprite_images_bytes, sprite_ids, min_similarity=None, top_k=5, method_weights=(0.5,0.3,0.2)):
+        from PIL import Image
+        # Local safe defaults
+        embeddings_path = os.path.join(BLOCKS_DIR, "hybrid_embeddings.json")
+        hash_path = os.path.join(BLOCKS_DIR, "phash_data.json")
+        signature_path = os.path.join(BLOCKS_DIR, "signature_data.json")
+    
+        # Load embeddings
+        embedding_json = {}
+        if os.path.exists(embeddings_path):
+            with open(embeddings_path, "r", encoding="utf-8") as f:
+                embedding_json = json.load(f)
+    
+        # Load phash data (if exists) -> ensure hash_dict variable exists
+        hash_dict = {}
+        if os.path.exists(hash_path):
+            try:
+                with open(hash_path, "r", encoding="utf-8") as f:
+                    hash_data = json.load(f)
+                    for path, hash_str in hash_data.items():
+                        try:
+                            hash_dict[path] = hash_str
+                        except Exception:
+                            pass
+            except Exception:
+                pass
+    
+        # Load signature data (if exists) -> ensure signature_dict exists
+        signature_dict = {}
+        sig_data = {}
+        if os.path.exists(signature_path):
+            try:
+                with open(signature_path, "r", encoding="utf-8") as f:
+                    sig_data = json.load(f)
+                    for path, sig_list in sig_data.items():
+                        try:
+                            signature_dict[path] = np.array(sig_list)
+                        except Exception:
+                            pass
+            except Exception:
+                pass
+    
+        # Parse embeddings into lists
+        paths_list = []
+        embeddings_list = []
+        if isinstance(embedding_json, dict):
+            for p, emb in embedding_json.items():
+                if isinstance(emb, dict):
+                    maybe_emb = emb.get("embedding") or emb.get("embeddings") or emb.get("emb")
+                    if maybe_emb is None:
+                        continue
+                    arr = np.asarray(maybe_emb, dtype=np.float32)
+                elif isinstance(emb, list):
+                    arr = np.asarray(emb, dtype=np.float32)
+                else:
+                    continue
+                paths_list.append(os.path.normpath(str(p)))
+                embeddings_list.append(arr)
+        elif isinstance(embedding_json, list):
+            for item in embedding_json:
+                if not isinstance(item, dict):
+                    continue
+                p = item.get("path") or item.get("image_path") or item.get("file") or item.get("filename") or item.get("img_path")
+                emb = item.get("embeddings") or item.get("embedding") or item.get("features") or item.get("vector") or item.get("emb")
+                if p is None or emb is None:
+                    continue
+                paths_list.append(os.path.normpath(str(p)))
+                embeddings_list.append(np.asarray(emb, dtype=np.float32))
+    
+        if len(paths_list) == 0:
+            print("⚠ No reference images/embeddings found (this test harness may be running without data)")
+            # Return empty results gracefully
+            return [[] for _ in sprite_images_bytes], [[] for _ in sprite_images_bytes], []
+    
+        ref_matrix = np.vstack(embeddings_list).astype(np.float32)
+        
+        # Batch: Get all sprite embeddings, phash, sigs first
+        sprite_emb_list = []
+        sprite_phash_list = []
+        sprite_sig_list = []
+        per_sprite_final_indices = []
+        per_sprite_final_scores = []
+        per_sprite_rerank_debug = []
+        for i, sprite_bytes in enumerate(sprite_images_bytes):
+            sprite_pil = Image.open(sprite_bytes)
+            enhanced_sprite = process_image_cv2_from_pil(sprite_pil, scale=2) or sprite_pil
+            # sprite_emb = get_dinov2_embedding_from_pil(preprocess_for_model(enhanced_sprite)) or np.zeros(ref_matrix.shape[1])
+            # sprite_emb_list.append(sprite_emb)
+            sprite_emb = get_dinov2_embedding_from_pil(preprocess_for_model(enhanced_sprite))
+            sprite_emb = sprite_emb if sprite_emb is not None else np.zeros(ref_matrix.shape[1])
+            sprite_emb_list.append(sprite_emb)
+            # Perceptual hash
+            sprite_hash_arr = preprocess_for_hash(enhanced_sprite)
+            sprite_phash = None
+            if sprite_hash_arr is not None:
+                try: sprite_phash = phash.encode_image(image_array=sprite_hash_arr)
+                except: pass
+            sprite_phash_list.append(sprite_phash)
+            # Signature
+            sprite_sig = None
+            embedding_results, phash_results, imgmatch_results, combined_results = run_query_search_flow(
+                query_b64=sprite_b64_clean[i],
+                processed_dir=BLOCKS_DIR,
+                embeddings_dict=embedding_json,
+                hash_dict=hash_data,
+                signature_obj_map=sig_data,
+                gis=gis,
+                phash=phash,
+                MAX_PHASH_BITS=64,
+                k=5
+            )
+            # Call the advanced re-ranker
+            rerank_result = choose_top_candidates(embedding_results, phash_results, imgmatch_results,
+                                                top_k=top_k, method_weights=method_weights, verbose=True)
+            per_sprite_rerank_debug.append(rerank_result)
+    
+            # Selection logic: prefer consensus, else weighted top-1
+            final = None
+            if len(rerank_result["consensus_topk"]) > 0:
+                consensus = rerank_result["consensus_topk"]
+                best = max(consensus, key=lambda p: rerank_result["weighted_scores_full"].get(p, 0.0))
+                final = best
+            else:
+                final = rerank_result["weighted_topk"][0][0] if rerank_result["weighted_topk"] else None
+    
+            # Store index and score for downstream use
+            if final is not None and final in paths_list:
+                idx = paths_list.index(final)
+                score = rerank_result["weighted_scores_full"].get(final, 0.0)
+                per_sprite_final_indices.append([idx])
+                per_sprite_final_scores.append([score])
+                print(f"Sprite '{sprite_ids}' FINAL selected: {final} (index {idx}) score={score:.4f}")
+            else:
+                per_sprite_final_indices.append([])
+                per_sprite_final_scores.append([])
+    
+        return per_sprite_final_indices, per_sprite_final_scores, paths_list#, per_sprite_rerank_debug
+    
 
     # Use hybrid matching system
     per_sprite_matched_indices, per_sprite_scores, paths_list = hybrid_similarity_matching(