Update app.py

app.py

@@ -703,6 +703,146 @@ 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:
@@ -1731,144 +1871,6 @@ 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(