#!/usr/bin/env python """DeepFashion text-search comparison demo for CLIP vs Hyper3-CLIP in HyperView.""" from __future__ import annotations import os import re import time from collections import Counter from pathlib import Path from typing import Any from datasets import load_dataset from PIL import Image, ImageOps import hyperview as hv from hyperview.core.sample import Sample SPACE_DIR = Path(__file__).resolve().parent SPACE_HOST = os.environ.get("HYPERVIEW_HOST", "127.0.0.1") SPACE_PORT = int(os.environ.get("HYPERVIEW_PORT", "6262")) WORKSPACE_ID = os.environ.get("HYPERVIEW_WORKSPACE_ID", "fashion-retail-search-v062-samples-visible") DATASET_NAME = os.environ.get("HYPERVIEW_DATASET_NAME", "deepfashion_text_search_clip_hyper3clip") EXTENSION_DIR = SPACE_DIR / ".hyperview" / "extensions" / "fashion-search-readout" HF_DATASET = os.environ.get("DEEPFASHION_HF_DATASET", "Marqo/deepfashion-inshop") HF_SPLIT = os.environ.get("DEEPFASHION_HF_SPLIT", "data") SAMPLES_PER_CATEGORY = int(os.environ.get("DEEPFASHION_SAMPLES_PER_CATEGORY", "45")) MAX_SAMPLES = int(os.environ.get("DEEPFASHION_MAX_SAMPLES", "700")) IMAGE_MAX_SIZE = (768, 768) FORCE_SAMPLE_REFRESH = os.environ.get("HYPERVIEW_DEEPFASHION_FORCE_REFRESH", "").lower() in { "1", "true", "yes", } ALLOW_CANDIDATE_FALLBACK = os.environ.get("HYPERVIEW_ALLOW_CANDIDATE_FALLBACK", "1").lower() in { "1", "true", "yes", } ENABLE_CONTEXT_MAPS = os.environ.get("FASHION_ENABLE_CONTEXT_MAPS", "1").lower() in { "1", "true", "yes", } EMBEDDING_MAX_ATTEMPTS = max(1, int(os.environ.get("HYPERVIEW_EMBEDDING_MAX_ATTEMPTS", "4"))) EMBEDDING_RETRY_DELAY_SECONDS = float(os.environ.get("HYPERVIEW_EMBEDDING_RETRY_DELAY_SECONDS", "15")) RUNTIME_WARNINGS: list[str] = [] DEFAULT_EXAMPLE_ID = os.environ.get("FASHION_DEFAULT_EXAMPLE_ID", "light-denim-leggings") MODEL_SPECS = [ { "key": "clip", "display_name": os.environ.get("FASHION_BASELINE_DISPLAY_NAME", "CLIP"), "button_label": os.environ.get("FASHION_BASELINE_BUTTON_LABEL", "Inspect CLIP neighbors"), "provider": os.environ.get("FASHION_BASELINE_PROVIDER", "embed-anything"), "model": os.environ.get("FASHION_BASELINE_MODEL", "openai/clip-vit-base-patch32"), "layout": os.environ.get("FASHION_BASELINE_LAYOUT", "euclidean:2d"), "geometry": os.environ.get("FASHION_BASELINE_GEOMETRY", "euclidean"), "layout_dimension": int(os.environ.get("FASHION_BASELINE_LAYOUT_DIMENSION", "2")), "metric": os.environ.get("FASHION_BASELINE_METRIC", "cosine"), "panel_title": os.environ.get("FASHION_BASELINE_PANEL_TITLE", "CLIP - Fashion Catalog Map"), }, { "key": "candidate", "display_name": os.environ.get("FASHION_CANDIDATE_DISPLAY_NAME", "Hyper3-CLIP"), "button_label": os.environ.get("FASHION_CANDIDATE_BUTTON_LABEL", "Inspect Hyper3-CLIP neighbors"), "provider": os.environ.get("FASHION_CANDIDATE_PROVIDER", "hyper-models"), "model": os.environ.get("FASHION_CANDIDATE_MODEL", "hyper3-clip-v0.5"), "layout": os.environ.get("FASHION_CANDIDATE_LAYOUT", "poincare:2d"), "geometry": os.environ.get("FASHION_CANDIDATE_GEOMETRY", "poincare"), "layout_dimension": int(os.environ.get("FASHION_CANDIDATE_LAYOUT_DIMENSION", "2")), "metric": os.environ.get("FASHION_CANDIDATE_METRIC", "cosine"), "panel_title": os.environ.get("FASHION_CANDIDATE_PANEL_TITLE", "Hyper3-CLIP - Fashion Catalog Map"), }, ] TEXT_SEARCH_EXAMPLES = [ { "id": "light-denim-leggings", "title": "Light denim leggings", "targetItemId": "WOMEN_Leggings_id_00001867_02_3_back", "targetProduct": "WOMEN_Leggings_id_00001867_02", "targetTitle": "women's light denim leggings", "family": "Specific typed product search", "query": "women's light denim leggings with skinny fit, zipper details, five-pocket construction, pockets", "hyper3Rank": 1, "clipRank": 32, "hyper3Text": "Exact target is the first result.", "clipText": "Top results drift to dark denim, black, and similar blue leggings before the exact item appears.", }, { "id": "olive-navy-pants", "title": "Olive and navy drawstring pants", "targetItemId": "MEN_Pants_id_00001468_03_6_flat", "targetProduct": "MEN_Pants_id_00001468_03", "targetTitle": "men's olive and navy drawstring pants", "family": "Specific typed product search", "query": "men's olive and navy pants with drawstring waist, pockets, striped pattern, knit fabric", "hyper3Rank": 1, "clipRank": 56, "hyper3Text": "Exact target is the first result.", "clipText": "CLIP ranks burgundy pants and visually similar pants before the requested product.", }, { "id": "cream-blue-halter-blouse", "title": "Cream and blue halter blouse", "targetItemId": "WOMEN_Blouses_Shirts_id_00007161_02_1_front", "targetProduct": "WOMEN_Blouses_Shirts_id_00007161_02", "targetTitle": "cream and blue halter blouse", "family": "Attribute-heavy apparel search", "query": "women's cream and blue blouse with halter neckline, floral pattern, striped pattern, tribal print", "hyper3Rank": 4, "clipRank": 33, "hyper3Text": "Target views appear in the top 10.", "clipText": "CLIP retrieves broadly similar tops but misses the exact blouse in the first screen.", }, ] DEMO_RESULT_ITEM_IDS = { "MEN_Pants_id_00001468_03_6_flat", "MEN_Pants_id_00001468_04_6_flat", "MEN_Pants_id_00004045_03_2_side", "MEN_Pants_id_00004045_04_1_front", "MEN_Pants_id_00004045_09_3_back", "MEN_Pants_id_00004045_11_1_front", "MEN_Pants_id_00004045_11_2_side", "MEN_Pants_id_00004045_12_1_front", "MEN_Pants_id_00004045_12_2_side", "MEN_Pants_id_00004045_12_3_back", "MEN_Pants_id_00004045_12_7_additional", "MEN_Shirts_Polos_id_00007027_01_6_flat", "MEN_Sweaters_id_00005177_03_2_side", "MEN_Sweaters_id_00005177_03_3_back", "MEN_Sweaters_id_00005177_03_4_full", "WOMEN_Blouses_Shirts_id_00003641_01_1_front", "WOMEN_Blouses_Shirts_id_00006345_01_7_additional", "WOMEN_Blouses_Shirts_id_00007049_01_7_additional", "WOMEN_Blouses_Shirts_id_00007161_02_1_front", "WOMEN_Cardigans_id_00000521_02_3_back", "WOMEN_Denim_id_00000152_04_1_front", "WOMEN_Denim_id_00000152_04_2_side", "WOMEN_Denim_id_00002338_02_7_additional", "WOMEN_Denim_id_00002338_03_1_front", "WOMEN_Denim_id_00002338_03_3_back", "WOMEN_Denim_id_00002338_03_7_additional", "WOMEN_Denim_id_00005673_02_3_back", "WOMEN_Dresses_id_00006961_02_1_front", "WOMEN_Leggings_id_00001412_01_2_side", "WOMEN_Leggings_id_00001867_02_3_back", "WOMEN_Leggings_id_00002130_02_2_side", "WOMEN_Leggings_id_00003850_01_2_side", "WOMEN_Leggings_id_00003908_07_2_side", "WOMEN_Leggings_id_00003908_08_2_side", "WOMEN_Leggings_id_00004562_01_3_back", "WOMEN_Pants_id_00000053_02_1_front", "WOMEN_Pants_id_00001574_02_3_back", "WOMEN_Rompers_Jumpsuits_id_00004432_02_3_back", "WOMEN_Rompers_Jumpsuits_id_00004653_02_2_side", "WOMEN_Rompers_Jumpsuits_id_00005484_01_3_back", "WOMEN_Sweaters_id_00003304_01_1_front", "WOMEN_Sweaters_id_00003304_01_2_side", "WOMEN_Tees_Tanks_id_00000676_01_1_front", "WOMEN_Tees_Tanks_id_00000676_01_2_side", } def media_root() -> Path: root = Path(os.environ.get("HYPERVIEW_MEDIA_DIR", str(SPACE_DIR / "demo_data" / "media"))) path = root / DATASET_NAME path.mkdir(parents=True, exist_ok=True) return path def product_key(item_id: str) -> str: return re.sub(r"_\d+_[A-Za-z]+$", "", str(item_id)) def safe_sample_id(item_id: str) -> str: return re.sub(r"[^A-Za-z0-9_.-]+", "_", str(item_id)).strip("_")[:96] def readable(value: Any) -> str: text = str(value or "").replace("_", " ").replace("-", " ") return re.sub(r"\s+", " ", text).strip() def save_image(image: Image.Image, destination: Path) -> None: if destination.exists() and destination.stat().st_size > 0 and not FORCE_SAMPLE_REFRESH: return tmp_path = destination.with_suffix(destination.suffix + ".tmp") image = ImageOps.exif_transpose(image).convert("RGB") image.thumbnail(IMAGE_MAX_SIZE, Image.Resampling.LANCZOS) image.save(tmp_path, format="JPEG", quality=92, optimize=True) tmp_path.replace(destination) def select_deepfashion_records() -> list[dict[str, Any]]: print(f"Loading DeepFashion split {HF_SPLIT!r} from {HF_DATASET}...", flush=True) source = load_dataset(HF_DATASET, split=HF_SPLIT) required_products = {example["targetProduct"] for example in TEXT_SEARCH_EXAMPLES} required_item_ids = {example["targetItemId"] for example in TEXT_SEARCH_EXAMPLES} | DEMO_RESULT_ITEM_IDS selected: list[dict[str, Any]] = [] seen: set[str] = set() category_counts: Counter[str] = Counter() for index, row in enumerate(source): item_id = str(row["item_ID"]) category = str(row.get("category2") or "unknown") product = product_key(item_id) required = product in required_products or item_id in required_item_ids balanced = category_counts[category] < SAMPLES_PER_CATEGORY and len(selected) < MAX_SAMPLES if not required and not balanced: continue if item_id in seen: continue selected.append({"index": index, **row}) seen.add(item_id) category_counts[category] += 1 missing = sorted(required_item_ids - seen) if missing: raise RuntimeError(f"Missing required demo items from DeepFashion: {missing}") print(f"Selected {len(selected)} DeepFashion images: {dict(category_counts)}", flush=True) return selected def add_deepfashion_samples(dataset: hv.Dataset) -> None: existing_ids = {sample.id for sample in dataset.samples} media_dir = media_root() added = 0 updated = 0 skipped_existing = 0 records = select_deepfashion_records() samples: list[Sample] = [] for record in records: item_id = str(record["item_ID"]) sample_id = safe_sample_id(item_id) existed = sample_id in existing_ids if existed and not FORCE_SAMPLE_REFRESH: skipped_existing += 1 continue destination = media_dir / f"{sample_id}.jpg" save_image(record["image"], destination) category = readable(record.get("category2") or "unknown").lower() color = readable(record.get("color") or "unknown") metadata = { "item_id": item_id, "product_key": product_key(item_id), "gender": readable(record.get("category1") or "unknown"), "category": category, "subcategory": readable(record.get("category3") or "unknown"), "color": color, "description": readable(record.get("description") or ""), "text": readable(record.get("text") or ""), "source_dataset": HF_DATASET, "split": HF_SPLIT, } samples.append( Sample( id=sample_id, filepath=str(destination), label=category, metadata=metadata, ) ) if existed: updated += 1 else: existing_ids.add(sample_id) added += 1 dataset.add_samples(samples, skip_existing=False) if skipped_existing: print(f"Skipped {skipped_existing} existing DeepFashion sample rows.", flush=True) print(f"Prepared DeepFashion samples ({added} added, {updated} updated).", flush=True) def compute_embeddings_with_retry(dataset: hv.Dataset, spec: dict[str, Any]) -> str: for attempt in range(1, EMBEDDING_MAX_ATTEMPTS + 1): try: return dataset.compute_embeddings( model=spec["model"], provider=spec["provider"], batch_size=32, show_progress=True, ) except BaseException as exc: if isinstance(exc, (KeyboardInterrupt, SystemExit)): raise if attempt >= EMBEDDING_MAX_ATTEMPTS: raise delay = EMBEDDING_RETRY_DELAY_SECONDS * attempt print( f"Embedding load failed for {spec['display_name']} " f"({type(exc).__name__}: {exc}). Retrying in {delay:.0f}s " f"({attempt + 1}/{EMBEDDING_MAX_ATTEMPTS})...", flush=True, ) time.sleep(delay) raise RuntimeError(f"Failed to compute embeddings for {spec['display_name']}") def ensure_layouts(dataset: hv.Dataset) -> dict[str, str]: layouts: dict[str, str] = {} for spec in MODEL_SPECS: print(f"Ensuring {spec['display_name']} embeddings...", flush=True) try: space_key = compute_embeddings_with_retry(dataset, spec) except BaseException as exc: if isinstance(exc, (KeyboardInterrupt, SystemExit)): raise if spec["key"] == "candidate" and ALLOW_CANDIDATE_FALLBACK and "clip" in layouts: warning = ( f"Hyper3-CLIP embeddings are unavailable ({type(exc).__name__}: {exc}). " "Showing the CLIP layout as a clearly labeled fallback so the Space can start." ) print(warning, flush=True) RUNTIME_WARNINGS.append(warning) fallback_layout_key = layouts["clip"] spec.update( { "display_name": "Hyper3-CLIP unavailable (CLIP fallback)", "button_label": "Inspect CLIP fallback neighbors", "geometry": MODEL_SPECS[0]["geometry"], "layout_dimension": MODEL_SPECS[0]["layout_dimension"], "panel_title": "Hyper3-CLIP unavailable - showing CLIP fallback", "fallback": True, "layout_key": fallback_layout_key, } ) layouts[spec["key"]] = fallback_layout_key continue raise print(f"Ensuring {spec['display_name']} layout...", flush=True) layout_key = dataset.compute_visualization( space_key=space_key, layout=spec["layout"], n_neighbors=20, min_dist=0.08, metric=spec["metric"], ) spec["layout_key"] = layout_key layouts[spec["key"]] = layout_key return layouts def build_dataset() -> tuple[hv.Dataset, dict[str, str]]: dataset = hv.Dataset(DATASET_NAME) add_deepfashion_samples(dataset) if ENABLE_CONTEXT_MAPS: layouts = ensure_layouts(dataset) else: layouts = {} return dataset, layouts def model_panel_props(layouts: dict[str, str]) -> list[dict[str, Any]]: props = [] for spec in MODEL_SPECS: layout_key = layouts.get(spec["key"]) props.append( { "key": spec["key"], "displayName": spec["display_name"], "buttonLabel": spec["button_label"], "layoutKey": layout_key, } ) return props def neighbor_summary(dataset: hv.Dataset, sample_id: str, model_key: str) -> dict[str, Any]: spec = next((item for item in MODEL_SPECS if item["key"] == model_key), None) if spec is None: return {} query = dataset[sample_id] layout_key = spec.get("layout_key") if layout_key is None: return {} neighbors = dataset.find_similar(sample_id, k=10, layout_key=str(layout_key)) query_product = query.metadata.get("product_key") query_category = query.metadata.get("category") product_hits = sum(1 for sample, _distance in neighbors if sample.metadata.get("product_key") == query_product) category_hits = sum(1 for sample, _distance in neighbors if sample.metadata.get("category") == query_category) return {"productHits": product_hits, "categoryHits": category_hits, "total": len(neighbors)} def build_examples(dataset: hv.Dataset) -> list[dict[str, Any]]: examples = [] candidate_is_fallback = any(spec["key"] == "candidate" and spec.get("fallback") for spec in MODEL_SPECS) for item in TEXT_SEARCH_EXAMPLES: sample_id = safe_sample_id(item["targetItemId"]) if sample_id not in {sample.id for sample in dataset.samples}: continue candidate_text = ( "Hyper3-CLIP is unavailable in this runtime, so this button shows the CLIP fallback neighborhood." if candidate_is_fallback else item["hyper3Text"] ) examples.append( { "id": item["id"], "title": item["title"], "family": item["family"], "query": item["query"], "queryId": sample_id, "targetTitle": item["targetTitle"], "summaries": { "clip": { "rank": item["clipRank"], "text": item["clipText"], **neighbor_summary(dataset, sample_id, "clip"), }, "candidate": { "rank": item["hyper3Rank"], "text": candidate_text, **neighbor_summary(dataset, sample_id, "candidate"), }, }, } ) return examples def build_demo_view(dataset: hv.Dataset, layouts: dict[str, str]) -> hv.ui.View: shared_props = { "models": model_panel_props(layouts), "examples": build_examples(dataset), "initialExampleId": DEFAULT_EXAMPLE_ID, "warnings": RUNTIME_WARNINGS, "metrics": { "typedQueryCount": 180, "typedCandidateImages": 1120, "hit10Hyper3Only": 23, "hit10ClipOnly": 19, "strongHyper3Wins": 13, "strongClipWins": 9, "imageRetrievalMapHyper3": 0.407, "imageRetrievalMapClip": 0.240, "typedHit1Hyper3": 0.244, "typedHit1Clip": 0.233, "typedHit10Hyper3": 0.572, "typedHit10Clip": 0.550, "typedCategoryP10Hyper3": 0.594, "typedCategoryP10Clip": 0.561, "typedMrrHyper3": 0.358, "typedMrrClip": 0.344, }, } results_panel = hv.ui.ExtensionPanel( id="fashion-ranked-results", title="Ranked Search Results", extension="fashion-search-readout", panel="fashion-comparison", position="center", layout=hv.ui.PanelLayout( width=int(os.environ.get("FASHION_RESULTS_WIDTH", "620")), min_width=500, ), props={ **shared_props, "mode": "results", }, ) samples_panel = hv.ui.Samples( id="grid", title="Samples", position="center", reference_panel_id="fashion-ranked-results", direction="right", layout=hv.ui.PanelLayout( width=int(os.environ.get("FASHION_SAMPLES_WIDTH", "660")), min_width=420, min_height=480, ), ) if not ENABLE_CONTEXT_MAPS: return hv.ui.View(results_panel, samples_panel, active_panel="fashion-ranked-results") clip_spec = MODEL_SPECS[0] candidate_spec = MODEL_SPECS[1] map_layout = hv.ui.PanelLayout( height=int(os.environ.get("FASHION_MAP_HEIGHT", "180")), min_height=150, min_width=220, ) clip_map = hv.ui.Scatter( id="fashion-map-clip", title="Context Map: CLIP", layout_key=layouts["clip"], position="center", reference_panel_id="grid", direction="below", geometry=clip_spec["geometry"], layout_dimension=clip_spec["layout_dimension"], layout=map_layout, ) candidate_map = hv.ui.Scatter( id="fashion-map-hyper3", title="Context Map: Hyper3", layout_key=layouts["candidate"], position="center", reference_panel_id="fashion-map-clip", direction="right", geometry=candidate_spec["geometry"], layout_dimension=candidate_spec["layout_dimension"], layout=map_layout, ) return hv.ui.View( results_panel, samples_panel, clip_map, candidate_map, active_panel="fashion-ranked-results", ) def initial_target_sample_id() -> str | None: example = next( (item for item in TEXT_SEARCH_EXAMPLES if item["id"] == DEFAULT_EXAMPLE_ID), TEXT_SEARCH_EXAMPLES[0] if TEXT_SEARCH_EXAMPLES else None, ) if example is None: return None return safe_sample_id(example["targetItemId"]) def launch_demo(dataset: hv.Dataset, layouts: dict[str, str]) -> hv.Session: session = hv.launch( dataset, host=SPACE_HOST, port=SPACE_PORT, open_browser=False, workspace_id=WORKSPACE_ID, block=False, ) print("Installing DeepFashion demo extension...", flush=True) session.ui.add_extension(EXTENSION_DIR, workspace_id=WORKSPACE_ID) print("Applying DeepFashion retail search demo view...", flush=True) session.ui.apply_view(build_demo_view(dataset, layouts), workspace_id=WORKSPACE_ID) if ENABLE_CONTEXT_MAPS and layouts: session.ui.set_active_layout(layouts["clip"], workspace_id=WORKSPACE_ID) sample_id = initial_target_sample_id() if sample_id: session.ui.set_selection([sample_id], workspace_id=WORKSPACE_ID) print(f"\nHyperView DeepFashion text-search demo is running at {session.url}", flush=True) if ENABLE_CONTEXT_MAPS: print(" Samples and nearest neighbors stay visible; scatter maps use the actual CLIP/Hyper3 layouts.", flush=True) else: print(" Samples stay visible; ranked text-search results are the main demo.", flush=True) return session def main() -> None: dataset, layouts = build_dataset() if layouts: print("Layouts:", flush=True) for spec in MODEL_SPECS: print(f" {spec['display_name']}: {layouts[spec['key']]}", flush=True) else: print("Context maps disabled; skipping embedding/layout startup.", flush=True) session = launch_demo(dataset, layouts) session.wait() if __name__ == "__main__": main()