Deploy merged demo: representative images (#42), t-SNE exact solver (#45), PCA reproducibility (#46), decoupled projection/KMeans + thread pipeline, demo header/footer

Dockerfile

@@ -23,8 +23,10 @@ RUN pip install --no-cache-dir -r requirements-space.txt
 # App source (apps/ and shared/ are pushed to the Space repo alongside this file).
 COPY --chown=user . $HOME/app
 
-# Demo data is mounted at /data (see README).
-ENV EMB_EXPLORER_DEMO_DATA_ROOT=/data
+# Demo data is mounted at /data (see README); EMB_EXPLORER_DEMO=1 turns on the
+# hosted-demo chrome (header/footer) so it stays dormant in normal local use.
+ENV EMB_EXPLORER_DEMO_DATA_ROOT=/data \
+    EMB_EXPLORER_DEMO=1
 
 EXPOSE 7860
 CMD ["streamlit", "run", "apps/precalculated/app.py", \

README.md

@@ -7,6 +7,21 @@ sdk: docker
 app_port: 7860
 pinned: false
 license: mit
+short_description: Filter, project, and cluster precalculated image embeddings
+tags:
+  - biodiversity
+  - embeddings
+  - bioclip
+  - clustering
+  - dimensionality-reduction
+  - umap
+  - tsne
+  - visualization
+  - imageomics
+datasets:
+  - imageomics/TreeOfLife-200M-Embeddings
+models:
+  - imageomics/bioclip-2
 ---
 
 # Image Embedding Explorer — Precalculated Demo

apps/embed_explore/app.py

@@ -7,11 +7,6 @@ cluster them, and explore the results visually.
 
 import streamlit as st
 
-from apps.embed_explore.components.sidebar import render_clustering_sidebar
-from apps.embed_explore.components.image_preview import render_image_preview
-from shared.components.summary import render_clustering_summary
-from shared.components.visualization import render_scatter_plot
-
 
 def main():
     """CLI entry point — launches the Streamlit server."""
@@ -25,6 +20,11 @@ def main():
 
 def app():
     """Streamlit application layout."""
+    from apps.embed_explore.components.sidebar import render_clustering_sidebar
+    from apps.embed_explore.components.image_preview import render_image_preview
+    from shared.components.summary import render_clustering_summary
+    from shared.components.visualization import render_scatter_plot
+
     st.set_page_config(
         layout="wide",
         page_title="Embed & Explore",

apps/embed_explore/components/image_preview.py

@@ -19,24 +19,31 @@ def render_image_preview():
 
     valid_paths = st.session_state.get("valid_paths", None)
     labels = st.session_state.get("labels", None)
-    selected_idx = st.session_state.get("selected_image_idx", 0)
+    kmeans_col = st.session_state.get("kmeans_column", None)
+    selected_idx = st.session_state.get("selected_image_idx", None)
 
     if (
         valid_paths is not None and
-        labels is not None and
         selected_idx is not None and
         0 <= selected_idx < len(valid_paths)
     ):
         img_path = valid_paths[selected_idx]
-        cluster = labels[selected_idx] if labels is not None else "?"
+        cluster = labels[selected_idx] if labels is not None else None
 
-        # Log only when image changes
         if _last_displayed_path != img_path:
-            logger.info(f"[Image] Loading local file: {os.path.basename(img_path)} (cluster={cluster})")
+            log_msg = f"[Image] Loading local file: {os.path.basename(img_path)}"
+            if cluster is not None:
+                log_msg += f" (cluster={cluster})"
+            logger.info(log_msg)
             _last_displayed_path = img_path
 
-        st.image(img_path, caption=f"Cluster {cluster}: {os.path.basename(img_path)}", width='stretch')
+        caption = os.path.basename(img_path)
+        if cluster is not None and kmeans_col:
+            caption = f"{kmeans_col}={cluster}: {caption}"
+
+        st.image(img_path, caption=caption, width='stretch')
         st.markdown(f"**File:** `{os.path.basename(img_path)}`")
-        st.markdown(f"**Cluster:** `{cluster}`")
+        if cluster is not None and kmeans_col:
+            st.markdown(f"**{kmeans_col}:** `{cluster}`")
     else:
-        st.info("Image preview will appear here after you select a cluster point.")
+        st.info("Image preview will appear here after you select a point in the scatter.")

apps/embed_explore/components/sidebar.py

@@ -4,13 +4,20 @@ Sidebar components for the embed_explore application.
 
 import streamlit as st
 import os
-from typing import Tuple, List, Optional
+import time
+import hashlib
+import numpy as np
+import pandas as pd
+from typing import Tuple, Optional
 
 from shared.services.embedding_service import EmbeddingService
 from shared.services.clustering_service import ClusteringService
 from shared.services.file_service import FileService
 from shared.lib.progress import StreamlitProgressContext
-from shared.components.clustering_controls import render_clustering_backend_controls, render_basic_clustering_controls
+from shared.components.clustering_controls import (
+    render_projection_controls,
+    render_kmeans_controls,
+)
 from shared.utils.backend import check_cuda_available, resolve_backend, is_oom_error
 from shared.utils.logging_config import get_logger
 
@@ -74,10 +81,11 @@ def render_embedding_section() -> Tuple[bool, Optional[str], Optional[str], int,
                         st.session_state.valid_paths = valid_paths
                         st.session_state.last_image_dir = image_dir
                         st.session_state.embedding_complete = True
-                        # Reset clustering/selection state
+                        # Reset projection/clustering/selection state for the new embeddings
                         st.session_state.labels = None
+                        st.session_state.kmeans_column = None
                         st.session_state.data = None
-                        st.session_state.selected_image_idx = 0
+                        st.session_state.selected_image_idx = None
 
                 except Exception as e:
                     st.error(f"Error during embedding: {e}")
@@ -89,152 +97,239 @@ def render_embedding_section() -> Tuple[bool, Optional[str], Optional[str], int,
     return embed_button, image_dir, model_name, n_workers, batch_size
 
 
-def render_clustering_section(n_workers: int = 1) -> Tuple[bool, int, str]:
-    """
-    Render the clustering section of the sidebar.
+def render_projection_section():
+    """Render the 2D projection section."""
+    with st.expander("Project to 2D", expanded=False):
+        embeddings = st.session_state.get("embeddings", None)
+        valid_paths = st.session_state.get("valid_paths", None)
 
-    Args:
-        n_workers: Number of workers for parallel processing
+        if embeddings is None or valid_paths is None or len(valid_paths) < 2:
+            st.info("Run embedding first to enable projection.")
+            return
 
-    Returns:
-        Tuple of (cluster_button_clicked, n_clusters, reduction_method)
-    """
-    with st.expander("Cluster", expanded=False):
-        # Basic clustering controls
-        n_clusters, reduction_method = render_basic_clustering_controls()
-
-        # Backend and advanced controls
-        dim_reduction_backend, clustering_backend, n_workers_clustering, seed = render_clustering_backend_controls()
-
-        cluster_button = st.button("Run Clustering", type="primary")
-
-        # Handle clustering execution
-        if cluster_button:
-            embeddings = st.session_state.get("embeddings", None)
-            valid_paths = st.session_state.get("valid_paths", None)
-
-            if embeddings is not None and valid_paths is not None and len(valid_paths) > 1:
-                run_clustering_with_fallback(
-                    embeddings, valid_paths, n_clusters, reduction_method,
-                    n_workers_clustering, dim_reduction_backend, clustering_backend, seed
-                )
-            else:
-                st.error("Please run embedding first.")
-
-    return cluster_button, n_clusters, reduction_method
-
-
-def run_clustering_with_fallback(
-    embeddings,
-    valid_paths,
-    n_clusters: int,
-    reduction_method: str,
-    n_workers: int,
-    dim_reduction_backend: str,
-    clustering_backend: str,
-    seed: Optional[int]
-):
-    """
-    Run clustering with robust error handling and automatic fallbacks.
+        n_samples, emb_dim = embeddings.shape
+        st.markdown(f"**Ready to project:** {n_samples:,} images ({emb_dim}-dim embeddings)")
 
-    Uses ClusteringService.run_clustering_safe() which transparently
-    handles GPU errors by falling back to CPU-based sklearn backends.
-    """
-    cuda_available, device_info = check_cuda_available()
-    actual_dim_backend = resolve_backend(dim_reduction_backend, "reduction")
-    actual_cluster_backend = resolve_backend(clustering_backend, "clustering")
+        reduction_method = st.selectbox(
+            "Dimensionality Reduction",
+            ["TSNE", "PCA", "UMAP"],
+            help="Method to project high-dimensional embeddings to 2D for visualization.",
+        )
+
+        dim_reduction_backend, seed = render_projection_controls()
+
+        if st.button("Project to 2D", type="primary"):
+            _run_projection(embeddings, valid_paths, reduction_method, dim_reduction_backend, seed)
+
+
+def render_kmeans_section():
+    """Render the optional KMeans clustering section."""
+    with st.expander("KMeans Clustering", expanded=False):
+        df_plot = st.session_state.get("data", None)
+        embeddings = st.session_state.get("embeddings", None)
+
+        if df_plot is None or embeddings is None:
+            st.info("Run projection first to enable KMeans.")
+            return
+
+        emb_dim = embeddings.shape[1]
+        st.markdown(f"**{len(df_plot):,} points** ({emb_dim}-dim embeddings)")
+
+        n_clusters = st.slider("Number of clusters", 2, min(100, max(2, len(df_plot) // 2)), 5)
+
+        clustering_backend, n_workers, seed = render_kmeans_controls()
+
+        if st.button("Run KMeans", type="primary"):
+            _run_kmeans(embeddings, n_clusters, clustering_backend, n_workers, seed)
+
+
+def _run_projection(embeddings, valid_paths, reduction_method, dim_reduction_backend, seed):
+    """Run dim reduction and create the 2D scatter plot dataframe."""
+    try:
+        cuda_available, device_info = check_cuda_available()
+        actual_backend = resolve_backend(dim_reduction_backend, "reduction")
+
+        logger.info("=" * 60)
+        logger.info("PROJECTION START")
+        logger.info(f"Device: {device_info} (CUDA: {'Yes' if cuda_available else 'No'})")
+        logger.info(f"Backend: {actual_backend} (requested: {dim_reduction_backend})")
+
+        t_start = time.time()
+        n_samples, emb_dim = embeddings.shape
+        logger.info(f"Records: {n_samples:,} | Dim: {emb_dim}")
+
+        with st.spinner(f"Running {reduction_method}..."):
+            reduced = ClusteringService.run_dim_reduction_safe(
+                embeddings, reduction_method,
+                n_workers=8, dim_reduction_backend=actual_backend, seed=seed
+            )
+
+        t_total = time.time() - t_start
+        logger.info(f"Projection complete in {t_total:.2f}s")
+
+        # Build plot dataframe (no cluster column)
+        df_plot = pd.DataFrame({
+            "x": reduced[:, 0],
+            "y": reduced[:, 1],
+            "image_path": valid_paths,
+            "file_name": [os.path.basename(p) for p in valid_paths],
+            "idx": range(len(valid_paths)),
+        })
+
+        # Carry over any prior KMeans columns from the previous df_plot (if length matches)
+        prev_df = st.session_state.get("data")
+        if prev_df is not None and len(prev_df) == len(df_plot):
+            for col in prev_df.columns:
+                if col.startswith("KMeans (k="):
+                    df_plot[col] = prev_df[col].values
+
+        data_hash = hashlib.md5(f"{len(df_plot)}_{reduction_method}_{t_total}".encode()).hexdigest()[:8]
+        st.session_state.data = df_plot
+        st.session_state.data_version = data_hash
+        st.session_state.selected_image_idx = None
+
+        logger.info("=" * 60)
+        st.success(f"Projected {n_samples:,} points to 2D using {reduction_method}.")
+
+    except (RuntimeError, OSError) as e:
+        if is_oom_error(e):
+            st.error("**GPU Out of Memory**")
+            st.info("Try: Reduce dataset size, use 'sklearn' backend, or try PCA.")
+            logger.exception("GPU OOM during projection")
+        else:
+            st.error(f"Error during projection: {e}")
+            logger.exception("Projection error")
+    except MemoryError:
+        st.error("**System Out of Memory** - Reduce dataset size")
+        logger.exception("System memory exhausted during projection")
+    except Exception as e:
+        st.error(f"Error: {e}")
+        logger.exception("Unexpected projection error")
 
-    logger.info(f"Starting clustering: samples={len(embeddings)}, clusters={n_clusters}, "
-                f"reduction={reduction_method}, device={device_info}")
-    logger.info(f"Backends: dim_reduction={actual_dim_backend}, clustering={actual_cluster_backend}")
 
+def _run_kmeans(embeddings, n_clusters, clustering_backend, n_workers, seed):
+    """Run KMeans on already-extracted embeddings and add labels to df_plot."""
     try:
-        with st.spinner(f"Running {reduction_method} + KMeans ({actual_dim_backend}/{actual_cluster_backend})..."):
-            df_plot, labels = ClusteringService.run_clustering_safe(
-                embeddings, valid_paths, n_clusters, reduction_method,
-                n_workers, actual_dim_backend, actual_cluster_backend, seed
+        actual_backend = resolve_backend(clustering_backend, "clustering")
+        logger.info(f"KMeans: k={n_clusters}, backend={actual_backend}")
+
+        with st.spinner(f"Running KMeans (k={n_clusters})..."):
+            labels = ClusteringService.run_kmeans_only_safe(
+                embeddings, n_clusters,
+                n_workers=n_workers, clustering_backend=actual_backend, seed=seed
             )
 
-        # Store results
+        df_plot = st.session_state.data
+        kmeans_col = f"KMeans (k={n_clusters})"
+
+        df_plot[kmeans_col] = labels.astype(str)
         st.session_state.data = df_plot
         st.session_state.labels = labels
-        st.session_state.selected_image_idx = 0
+        st.session_state.kmeans_column = kmeans_col
 
-        # Compute and store clustering summary
+        # Compute clustering summary on the full embedding space.
+        # Cache by kmeans_col so multiple KMeans runs can each have their own
+        # summary + representatives that the user can switch between.
         logger.info("Computing clustering summary statistics...")
         summary_df, representatives = ClusteringService.generate_clustering_summary(
             embeddings, labels, df_plot
         )
-        st.session_state.clustering_summary = summary_df
-        st.session_state.clustering_representatives = representatives
-        logger.info(f"Clustering summary computed: {len(summary_df)} clusters")
+        summaries = st.session_state.get("clustering_summaries", {})
+        reps_by_col = st.session_state.get("clustering_representatives_by_col", {})
+        summaries[kmeans_col] = summary_df
+        reps_by_col[kmeans_col] = representatives
+        st.session_state.clustering_summaries = summaries
+        st.session_state.clustering_representatives_by_col = reps_by_col
+        logger.info(f"Clustering summary computed for {kmeans_col}: {len(summary_df)} clusters")
 
-        st.success(f"Clustering complete! Found {n_clusters} clusters.")
+        logger.info(f"KMeans complete: {len(np.unique(labels))} clusters")
+        st.success(f"KMeans complete! {len(np.unique(labels))} clusters assigned.")
 
     except (RuntimeError, OSError) as e:
         if is_oom_error(e):
-            st.error("**GPU Out of Memory** - Dataset too large for GPU")
-            st.info("Try: Reduce dataset size, or select 'sklearn' backend")
-            logger.exception("GPU OOM error during clustering")
+            st.error("**GPU Out of Memory**")
+            logger.exception("GPU OOM during KMeans")
         else:
-            st.error(f"Error during clustering: {e}")
-            logger.exception("Clustering error")
-
+            st.error(f"Error during KMeans: {e}")
+            logger.exception("KMeans error")
     except MemoryError:
         st.error("**System Out of Memory** - Reduce dataset size")
-        logger.exception("System memory exhausted during clustering")
-
+        logger.exception("System memory exhausted during KMeans")
     except Exception as e:
-        st.error(f"Error during clustering: {e}")
-        logger.exception("Unexpected clustering error")
+        st.error(f"Error: {e}")
+        logger.exception("Unexpected KMeans error")
 
 
-def render_save_section():
-    """Render the save operations section of the sidebar."""
-    # --- Save images from a specific cluster utility ---
-    save_status_placeholder = st.empty()
-    with st.expander("Save Images from Specific Cluster", expanded=True):
-        df_plot = st.session_state.get("data", None)
-        labels = st.session_state.get("labels", None)
-
-        if df_plot is not None and labels is not None:
-            available_clusters = sorted(df_plot['cluster'].unique(), key=lambda x: int(x))
-            selected_clusters = st.multiselect(
-                "Select cluster(s) to save",
-                available_clusters,
-                default=available_clusters[:1] if available_clusters else [],
-                key="save_cluster_select"
-            )
-            save_dir = st.text_input(
-                "Directory to save selected cluster images",
-                value="cluster_selected_output",
-                key="save_cluster_dir"
-            )
-            save_cluster_button = st.button("Save images", key="save_cluster_btn")
+def _get_available_kmeans_cols(df_plot) -> list:
+    """Return KMeans columns in df_plot sorted by k value."""
+    if df_plot is None:
+        return []
+    return sorted(
+        [c for c in df_plot.columns if c.startswith("KMeans (k=")],
+        key=lambda c: int(c.split("=")[1].rstrip(")")),
+    )
 
-            # Handle save execution
-            if save_cluster_button and selected_clusters:
-                cluster_rows = df_plot[df_plot['cluster'].isin(selected_clusters)]
-                max_workers = st.session_state.get("num_threads", 8)
 
-                with StreamlitProgressContext(
-                    save_status_placeholder,
-                    f"Images from cluster(s) {', '.join(map(str, selected_clusters))} saved successfully!"
-                ) as progress:
-                    try:
-                        save_summary_df, csv_path = FileService.save_cluster_images(
-                            cluster_rows, save_dir, max_workers, progress_callback=progress
-                        )
-                        st.info(f"Summary CSV saved at {csv_path}")
-
-                    except Exception as e:
-                        save_status_placeholder.error(f"Error saving images: {e}")
+def render_save_section():
+    """Render the save operations section of the sidebar.
 
-            elif save_cluster_button:
-                save_status_placeholder.warning("Please select at least one cluster.")
+    Both 'Save Images from Specific Cluster' and 'Repartition Images by Cluster'
+    require at least one KMeans run. When multiple KMeans runs exist, the user
+    picks which one to operate on via a shared selector at the top.
+    """
+    df_plot = st.session_state.get("data", None)
+    kmeans_cols = _get_available_kmeans_cols(df_plot)
+
+    if not kmeans_cols:
+        st.info("Run KMeans first to enable saving by cluster.")
+        return
+
+    # Shared selector: which KMeans run drives both save operations
+    default_idx = len(kmeans_cols) - 1  # most recent run
+    selected_kmeans_col = st.selectbox(
+        "KMeans result",
+        options=kmeans_cols,
+        index=default_idx,
+        key="save_kmeans_selector",
+        help="Pick which KMeans run to use for save / repartition.",
+    )
 
-        else:
-            st.info("Run clustering first to enable this utility.")
+    # --- Save images from a specific cluster utility ---
+    save_status_placeholder = st.empty()
+    with st.expander("Save Images from Specific Cluster", expanded=True):
+        available_clusters = sorted(df_plot[selected_kmeans_col].unique(), key=lambda x: int(x))
+        selected_clusters = st.multiselect(
+            "Select cluster(s) to save",
+            available_clusters,
+            default=available_clusters[:1] if available_clusters else [],
+            key="save_cluster_select",
+        )
+        save_dir = st.text_input(
+            "Directory to save selected cluster images",
+            value="cluster_selected_output",
+            key="save_cluster_dir",
+        )
+        save_cluster_button = st.button("Save images", key="save_cluster_btn")
+
+        if save_cluster_button and selected_clusters:
+            cluster_rows = df_plot[df_plot[selected_kmeans_col].isin(selected_clusters)].copy()
+            # FileService expects a 'cluster' column
+            cluster_rows["cluster"] = cluster_rows[selected_kmeans_col]
+            max_workers = st.session_state.get("num_threads", 8)
+
+            with StreamlitProgressContext(
+                save_status_placeholder,
+                f"Images from cluster(s) {', '.join(map(str, selected_clusters))} saved successfully!"
+            ) as progress:
+                try:
+                    save_summary_df, csv_path = FileService.save_cluster_images(
+                        cluster_rows, save_dir, max_workers, progress_callback=progress
+                    )
+                    st.info(f"Summary CSV saved at {csv_path}")
+                except Exception as e:
+                    save_status_placeholder.error(f"Error saving images: {e}")
+        elif save_cluster_button:
+            save_status_placeholder.warning("Please select at least one cluster.")
 
     # --- Repartition expander and status ---
     repartition_status_placeholder = st.empty()
@@ -243,7 +338,7 @@ def render_save_section():
         repartition_dir = st.text_input(
             "Directory",
             value="repartitioned_output",
-            key="repartition_dir"
+            key="repartition_dir",
         )
         max_workers = st.number_input(
             "Number of threads (higher = faster, try 8-32)",
@@ -251,49 +346,34 @@ def render_save_section():
             max_value=64,
             value=8,
             step=1,
-            key="num_threads"
+            key="num_threads",
         )
         repartition_button = st.button("Repartition images by cluster", key="repartition_btn")
 
-        # Handle repartition execution
         if repartition_button:
-            df_plot = st.session_state.get("data", None)
-
-            if df_plot is None or len(df_plot) < 1:
-                repartition_status_placeholder.warning("Please run clustering first before repartitioning images.")
-            else:
-                with StreamlitProgressContext(
-                    repartition_status_placeholder,
-                    f"Repartition complete! Images organized in {repartition_dir}"
-                ) as progress:
-                    try:
-                        repartition_summary_df, csv_path = FileService.repartition_images_by_cluster(
-                            df_plot, repartition_dir, max_workers, progress_callback=progress
-                        )
-                        st.info(f"Summary CSV saved at {csv_path}")
-
-                    except Exception as e:
-                        repartition_status_placeholder.error(f"Error repartitioning images: {e}")
+            df_for_repartition = df_plot.copy()
+            df_for_repartition["cluster"] = df_for_repartition[selected_kmeans_col]
+            with StreamlitProgressContext(
+                repartition_status_placeholder,
+                f"Repartition complete! Images organized in {repartition_dir}",
+            ) as progress:
+                try:
+                    repartition_summary_df, csv_path = FileService.repartition_images_by_cluster(
+                        df_for_repartition, repartition_dir, max_workers, progress_callback=progress
+                    )
+                    st.info(f"Summary CSV saved at {csv_path}")
+                except Exception as e:
+                    repartition_status_placeholder.error(f"Error repartitioning images: {e}")
 
 
 def render_clustering_sidebar():
-    """Render the complete clustering sidebar with all sections."""
+    """Render the complete sidebar with embed / project / KMeans / save sections."""
     tab_compute, tab_save = st.tabs(["Compute", "Save"])
 
     with tab_compute:
-        embed_button, image_dir, model_name, n_workers, batch_size = render_embedding_section()
-        cluster_button, n_clusters, reduction_method = render_clustering_section(n_workers)
+        render_embedding_section()
+        render_projection_section()
+        render_kmeans_section()
 
     with tab_save:
         render_save_section()
-
-    return {
-        'embed_button': embed_button,
-        'image_dir': image_dir,
-        'model_name': model_name,
-        'n_workers': n_workers,
-        'batch_size': batch_size,
-        'cluster_button': cluster_button,
-        'n_clusters': n_clusters,
-        'reduction_method': reduction_method,
-    }

apps/precalculated/app.py

@@ -7,16 +7,6 @@ Features dynamic filter generation based on available columns.
 
 import streamlit as st
 
-from apps.precalculated.components.sidebar import (
-    render_file_section,
-    render_dynamic_filters,
-    render_projection_section,
-    render_kmeans_section,
-)
-from apps.precalculated.components.data_preview import render_data_preview
-from shared.components.visualization import render_scatter_plot
-from shared.components.summary import render_clustering_summary
-
 
 def main():
     """CLI entry point — launches the Streamlit server."""
@@ -30,6 +20,24 @@ def main():
 
 def app():
     """Streamlit application layout."""
+    from apps.precalculated.components.sidebar import (
+        render_file_section,
+        render_dynamic_filters,
+        render_projection_section,
+        render_kmeans_section,
+    )
+    from apps.precalculated.components.data_preview import (
+        render_data_preview,
+        render_cluster_representatives,
+    )
+    from shared.components.visualization import render_scatter_plot
+    from shared.components.summary import render_clustering_summary
+    from shared.components.demo_chrome import (
+        is_demo_mode,
+        render_demo_header,
+        render_demo_footer,
+    )
+
     st.set_page_config(
         layout="wide",
         page_title="Precalculated Embeddings Explorer",
@@ -45,12 +53,15 @@ def app():
                 del st.session_state[key]
         st.session_state.page_type = "precalculated_app"
 
-    # Header
-    st.title("📊 Precalculated Embeddings Explorer")
-    st.markdown(
-        "Load parquet files with embeddings, apply dynamic filters, and cluster for visualization. "
-        "Filters are automatically generated based on your data columns."
-    )
+    # Header — demo chrome when hosted, otherwise the standard title.
+    if is_demo_mode():
+        render_demo_header()
+    else:
+        st.title("📊 Precalculated Embeddings Explorer")
+        st.markdown(
+            "Load parquet files with embeddings, apply dynamic filters, and cluster for visualization. "
+            "Filters are automatically generated based on your data columns."
+        )
 
     # Row 1: File loading
     render_file_section()
@@ -71,9 +82,14 @@ def app():
     with col_preview:
         render_data_preview()
 
-    # Bottom: Taxonomy summary
+    # Bottom: Taxonomy summary + representative images
     st.markdown("---")
     render_clustering_summary(show_taxonomy=True)
+    render_cluster_representatives()
+
+    # Demo-only attribution / funding footer.
+    if is_demo_mode():
+        render_demo_footer()
 
 
 if __name__ == "__main__":

apps/precalculated/components/data_preview.py

@@ -6,80 +6,21 @@ Dynamically displays all available metadata fields.
 import streamlit as st
 import pandas as pd
 import numpy as np
-import requests
-import time
-from typing import Optional
-from PIL import Image
-from io import BytesIO
 
 from shared.utils.logging_config import get_logger
+from shared.utils.representatives import find_cluster_representatives
+from shared.utils.images import (
+    IMAGE_URL_COLUMNS,
+    fetch_images_concurrent,
+    get_image_from_url,
+    resolve_record_image_url,
+    _IMAGE_CACHE,
+)
+from shared.components.representatives import render_representative_images
 
 logger = get_logger(__name__)
 
 
-@st.cache_data(ttl=300, show_spinner=False)
-def _fetch_image_from_url_cached(url: str, timeout: int = 5) -> Optional[bytes]:
-    """Internal cached function to fetch image bytes."""
-    if not url or not isinstance(url, str):
-        return None
-
-    try:
-        if not url.startswith(('http://', 'https://')):
-            return None
-
-        response = requests.get(url, timeout=timeout, stream=True)
-        response.raise_for_status()
-
-        content_type = response.headers.get('content-type', '').lower()
-        if not content_type.startswith('image/'):
-            return None
-
-        return response.content
-
-    except Exception:
-        return None
-
-
-def fetch_image_from_url(url: str, timeout: int = 5) -> Optional[bytes]:
-    """
-    Fetch an image from a URL with logging.
-    Uses caching internally but logs the request.
-    """
-    if not url or not isinstance(url, str):
-        return None
-
-    if not url.startswith(('http://', 'https://')):
-        logger.warning(f"[Image] Invalid URL scheme: {url[:50]}...")
-        return None
-
-    logger.info(f"[Image] Fetching: {url[:80]}...")
-    start_time = time.time()
-
-    result = _fetch_image_from_url_cached(url, timeout)
-
-    elapsed = time.time() - start_time
-    if result:
-        logger.info(f"[Image] Loaded: {len(result)/1024:.1f}KB in {elapsed:.3f}s")
-    else:
-        logger.warning(f"[Image] Failed to load: {url[:50]}...")
-
-    return result
-
-
-def get_image_from_url(url: str) -> Optional[Image.Image]:
-    """Get image from URL with caching and logging."""
-    image_bytes = fetch_image_from_url(url)
-    if image_bytes:
-        try:
-            image = Image.open(BytesIO(image_bytes))
-            logger.info(f"[Image] Opened: {image.size[0]}x{image.size[1]} {image.mode}")
-            return image
-        except Exception as e:
-            logger.error(f"[Image] Failed to open: {e}")
-            return None
-    return None
-
-
 def render_data_preview():
     """Render the data preview panel (record details on point click)."""
     df_plot = st.session_state.get("data", None)
@@ -110,15 +51,12 @@ def render_data_preview():
 
         st.markdown("### Record Details")
 
-        # Try to display image if identifier/url column exists (cached to prevent re-fetch)
-        image_cols = ['identifier', 'image_url', 'url', 'img_url', 'image']
-        for img_col in image_cols:
-            if img_col in record.index and pd.notna(record[img_col]):
-                url = record[img_col]
-                image = get_image_from_url(url)
-                if image is not None:
-                    st.image(image, width=280)
-                    break
+        # Try to display image if an image URL column exists (process-cached).
+        url = resolve_record_image_url(record)
+        if url:
+            image = get_image_from_url(url)
+            if image is not None:
+                st.image(image, width=280)
 
         st.markdown(f"**UUID:** `{selected_uuid}`")
 
@@ -276,3 +214,85 @@ def render_cluster_analysis():
         st.code(tree_output, language="text")
     else:
         st.info(f"No valid '{color_by}' values to compare with KMeans clusters.")
+
+
+def render_cluster_representatives():
+    """Render representative images per KMeans cluster for the precalculated app.
+
+    Representatives are the members closest to each cluster centroid (computed
+    on the full-dimensional embeddings). Images are fetched from each record's
+    URL column; URLs that fail to load are skipped and the next-closest
+    candidate is tried (fallback), so transient/broken URLs don't leave gaps.
+    """
+    df_plot = st.session_state.get("data", None)
+    embeddings = st.session_state.get("embeddings", None)
+    if df_plot is None or embeddings is None:
+        return
+
+    kmeans_cols = sorted(
+        [c for c in df_plot.columns if c.startswith("KMeans (k=")],
+        key=lambda c: int(c.split("=")[1].rstrip(")")),
+    )
+    if not kmeans_cols:
+        return  # nothing to show until a KMeans run exists
+
+    st.markdown("### Representative Images")
+    st.caption(
+        "Members closest to each cluster centroid. Images load from each "
+        "record's URL; unreachable images are skipped automatically."
+    )
+
+    selected_col = st.selectbox(
+        "KMeans result",
+        options=kmeans_cols,
+        index=len(kmeans_cols) - 1,
+        key="representatives_kmeans_selector",
+        help="Which KMeans run to show representatives for.",
+    )
+
+    # Guard: embeddings must align row-for-row with df_plot.
+    if len(embeddings) != len(df_plot):
+        st.info("Re-run projection and KMeans to view representatives.")
+        return
+
+    n_per_cluster = 3
+    representatives = find_cluster_representatives(
+        embeddings, df_plot[selected_col].values, n_per_cluster=n_per_cluster
+    )
+
+    # Warm the cache concurrently. Representatives are oversampled for fallback,
+    # but we only need a few successes per cluster — prefetch a prefix (2x the
+    # display count) in parallel. Deeper fallback candidates (rare) resolve
+    # on-demand below.
+    prefetch_per_cluster = n_per_cluster * 2
+    prefetch_urls = [
+        resolve_record_image_url(df_plot.iloc[idx])
+        for idxs in representatives.values()
+        for idx in idxs[:prefetch_per_cluster]
+    ]
+    with st.spinner("Loading representative images..."):
+        fetch_images_concurrent([u for u in prefetch_urls if u])
+
+    def _resolve(idx):
+        url = resolve_record_image_url(df_plot.iloc[idx])
+        if not url:
+            return None
+        # Prefetched URLs hit the process cache; anything deeper falls back to
+        # a single synchronous fetch (also cached).
+        if url in _IMAGE_CACHE:
+            return _IMAGE_CACHE[url]
+        return get_image_from_url(url)
+
+    def _caption(idx):
+        row = df_plot.iloc[idx]
+        for col in ("scientific_name", "species", "common_name", "uuid"):
+            if col in row.index and pd.notna(row[col]):
+                return str(row[col])
+        return None
+
+    render_representative_images(
+        representatives,
+        resolve_image=_resolve,
+        n_per_cluster=n_per_cluster,
+        caption_fn=_caption,
+    )

shared/__init__.py

@@ -2,4 +2,11 @@
 Shared utilities and services for the emb-explorer applications.
 """
 
-__version__ = "0.1.0"
+from importlib.metadata import PackageNotFoundError, version as _version
+
+try:
+    # Single source of truth: the version declared in pyproject.toml,
+    # read from the installed package metadata.
+    __version__ = _version("emb-explorer")
+except PackageNotFoundError:  # running from a source tree without an install
+    __version__ = "0.0.0+unknown"

shared/components/demo_chrome.py

@@ -0,0 +1,76 @@
+"""Demo-only page chrome (header + footer) for the hosted Hugging Face Space.
+
+Rendered only when ``EMB_EXPLORER_DEMO=1`` (set by the Space Dockerfile), so the
+normal local apps are unaffected. Kept additive and self-contained so it merges
+cleanly across branches.
+"""
+
+import os
+
+import streamlit as st
+
+REPO_URL = "https://github.com/Imageomics/emb-explorer"
+EMBEDDINGS_DATASET_URL = "https://huggingface.co/datasets/imageomics/TreeOfLife-200M-Embeddings"
+BIOCLIP2_URL = "https://huggingface.co/imageomics/bioclip-2"
+BIOCLIP2_SITE_URL = "https://imageomics.github.io/bioclip-2/"
+PYBIOCLIP_URL = "https://github.com/Imageomics/pybioclip"
+TOL200M_URL = "https://huggingface.co/datasets/imageomics/TreeOfLife-200M"
+IMAGEOMICS_URL = "https://imageomics.org"
+NSF_AWARD_URL = "https://www.nsf.gov/awardsearch/showAward?AWD_ID=2118240"
+
+
+def is_demo_mode() -> bool:
+    """True when running as the hosted demo (the Space sets EMB_EXPLORER_DEMO=1)."""
+    return os.environ.get("EMB_EXPLORER_DEMO", "0") == "1"
+
+
+def render_demo_header() -> None:
+    """Demo title + one-line plain-text intro (no links; links live in the footer)."""
+    st.title("🔍 Image Embedding Explorer (Demo)")
+    st.markdown(
+        "A hosted demo of the Image Embedding Explorer. Explore precalculated "
+        "BioCLIP 2 embeddings from a curated subset of the TreeOfLife-200M image "
+        "collection."
+    )
+
+
+_FOOTER_CSS = """
+<style>
+.demo-footer { margin-top: 2rem; padding-top: 1rem;
+               border-top: 1px solid rgba(128, 128, 128, 0.3);
+               opacity: 0.6; font-size: 0.82rem; line-height: 1.5; }
+.demo-footer:hover { opacity: 0.9; }
+.demo-footer a { color: #3f9b6e; text-decoration: none; }
+.demo-footer a:hover { text-decoration: underline; }
+</style>
+"""
+
+
+def render_demo_footer() -> None:
+    """Muted, bordered attribution / funding footer (Imageomics standard text).
+
+    Adopts the bioclip-image-search footer, with the emb-explorer source repo
+    and the TreeOfLife-200M-Embeddings dataset added.
+    """
+    st.markdown(_FOOTER_CSS, unsafe_allow_html=True)
+    st.markdown(
+        '<div class="demo-footer">'
+        f'This demo is built with the <a href="{REPO_URL}">emb-explorer</a> source '
+        f'repository and explores a curated subset of the '
+        f'<a href="{EMBEDDINGS_DATASET_URL}">TreeOfLife-200M-Embeddings</a> dataset. '
+        f'For more information on the <a href="{BIOCLIP2_URL}">BioCLIP&nbsp;2</a> model '
+        f'creation, see our <a href="{BIOCLIP2_SITE_URL}">BioCLIP&nbsp;2 Project website</a>, '
+        'and for easier programmatic integration of BioCLIP&nbsp;2, checkout '
+        f'<a href="{PYBIOCLIP_URL}">pybioclip</a>. To learn more about the data, check out '
+        f'our <a href="{TOL200M_URL}">TreeOfLife-200M Dataset</a>.'
+        '<br><br>'
+        f'This work was supported by the <a href="{IMAGEOMICS_URL}">Imageomics Institute</a>, '
+        "which is funded by the US National Science Foundation's Harnessing the Data "
+        f'Revolution (HDR) program under <a href="{NSF_AWARD_URL}">Award&nbsp;#2118240</a> '
+        '(Imageomics: A New Frontier of Biological Information Powered by Knowledge-Guided '
+        'Machine Learning). Any opinions, findings and conclusions or recommendations '
+        'expressed in this material are those of the author(s) and do not necessarily '
+        'reflect the views of the National Science Foundation.'
+        '</div>',
+        unsafe_allow_html=True,
+    )

shared/components/representatives.py

@@ -0,0 +1,76 @@
+"""Shared renderer for per-cluster representative images.
+
+Both apps surface representative images differently:
+- embed_explore resolves a local image file path.
+- precalculated fetches a remote image URL (which can fail).
+
+This renderer is source-agnostic: the caller passes a `resolve_image(idx)`
+callable that returns something `st.image` can display (a PIL image, a path,
+or bytes) or `None` when the image is unavailable. The renderer walks each
+cluster's ranked candidate indices and collects up to `n_per_cluster`
+successful images, skipping any that resolve to `None` — the shared fallback.
+"""
+
+from typing import Any, Callable, Dict, List, Optional
+
+import streamlit as st
+
+from shared.utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+def _sorted_cluster_ids(representatives: Dict[object, List[int]]) -> List[object]:
+    """Sort cluster ids numerically when possible, else as strings."""
+    keys = list(representatives.keys())
+    try:
+        return sorted(keys, key=lambda k: int(k))
+    except (ValueError, TypeError):
+        return sorted(keys, key=str)
+
+
+def render_representative_images(
+    representatives: Dict[object, List[int]],
+    resolve_image: Callable[[int], Optional[Any]],
+    n_per_cluster: int = 3,
+    caption_fn: Optional[Callable[[int], str]] = None,
+    columns: int = 3,
+) -> None:
+    """Render up to `n_per_cluster` representative images per cluster.
+
+    Args:
+        representatives: {cluster_id: [ranked candidate global indices]}, as
+            returned by `find_cluster_representatives`.
+        resolve_image: idx -> displayable (PIL image / path / bytes) or None.
+            None means "unavailable" and the renderer falls back to the next
+            candidate.
+        n_per_cluster: number of images to show per cluster.
+        caption_fn: optional idx -> caption string.
+        columns: images per row.
+    """
+    for cluster_id in _sorted_cluster_ids(representatives):
+        candidates = representatives[cluster_id]
+        st.markdown(f"**Cluster {cluster_id}**")
+
+        # Walk ranked candidates, collecting successful resolutions until we
+        # have n_per_cluster (or run out of candidates).
+        shown: List[tuple] = []  # (displayable, caption)
+        for idx in candidates:
+            if len(shown) >= n_per_cluster:
+                break
+            try:
+                img = resolve_image(idx)
+            except Exception as e:  # never let one bad image break the panel
+                logger.debug(f"resolve_image({idx}) raised: {e}")
+                img = None
+            if img is not None:
+                caption = caption_fn(idx) if caption_fn else None
+                shown.append((img, caption))
+
+        if not shown:
+            st.caption("No images available for this cluster.")
+            continue
+
+        cols = st.columns(min(columns, len(shown)))
+        for i, (img, caption) in enumerate(shown):
+            cols[i % len(cols)].image(img, caption=caption, width="stretch")

shared/components/summary.py

@@ -6,6 +6,7 @@ import streamlit as st
 import os
 import pandas as pd
 from shared.utils.taxonomy_tree import build_taxonomic_tree, format_tree_string, get_tree_statistics
+from shared.components.representatives import render_representative_images
 from shared.utils.logging_config import get_logger
 
 logger = get_logger(__name__)
@@ -144,47 +145,80 @@ def render_taxonomic_tree_summary():
 
 
 def render_clustering_summary(show_taxonomy=False):
-    """Render the clustering summary panel using cached results from clustering action."""
+    """Render the clustering summary panel using cached results per KMeans run.
+
+    For the embed_explore app, when multiple KMeans runs exist on df_plot,
+    the user can pick which run's summary + representative images to display.
+    Summaries are cached per kmeans_col by `_run_kmeans` so switching is instant.
+    """
     df_plot = st.session_state.get("data", None)
-    labels = st.session_state.get("labels", None)
 
-    # Get pre-computed summary from session state (computed when clustering was run)
-    summary_df = st.session_state.get("clustering_summary", None)
-    representatives = st.session_state.get("clustering_representatives", None)
-
-    if df_plot is not None:
-        has_images = 'image_path' in df_plot.columns
-
-        if has_images:
-            # embed_explore app: show full clustering summary with representative images
-            if labels is not None:
-                st.subheader("Clustering Summary")
-
-                if summary_df is not None and representatives is not None:
-                    logger.debug("Displaying cached clustering summary")
-                    st.dataframe(summary_df, hide_index=True, width='stretch')
-
-                    st.markdown("#### Representative Images")
-                    for row in summary_df.itertuples():
-                        k = row.Cluster
-                        st.markdown(f"**Cluster {k}**")
-                        img_cols = st.columns(3)
-                        for i, img_idx in enumerate(representatives[k]):
-                            img_path = df_plot.iloc[img_idx]["image_path"]
-                            logger.debug(f"Displaying representative image: {img_path}")
-                            img_cols[i].image(
-                                img_path,
-                                width='stretch',
-                                caption=os.path.basename(img_path)
-                            )
-                else:
-                    st.info("Clustering summary will be computed when you run clustering.")
-        else:
-            # Precalculated app: show taxonomy tree (works with or without KMeans)
-            if show_taxonomy:
-                filtered_df = st.session_state.get("filtered_df_for_clustering", None)
-                if filtered_df is not None:
-                    render_taxonomic_tree_summary()
+    if df_plot is None:
+        st.info("Summary will appear here after projection.")
+        return
+
+    has_images = 'image_path' in df_plot.columns
 
+    if has_images:
+        # embed_explore app: full clustering summary with representative images
+        kmeans_cols = sorted(
+            [c for c in df_plot.columns if c.startswith("KMeans (k=")],
+            key=lambda c: int(c.split("=")[1].rstrip(")")),
+        )
+
+        if not kmeans_cols:
+            st.subheader("Clustering Summary")
+            st.info("Run KMeans to see the clustering summary and representative images.")
+            return
+
+        summaries = st.session_state.get("clustering_summaries", {}) or {}
+        reps_by_col = st.session_state.get("clustering_representatives_by_col", {}) or {}
+
+        st.subheader("Clustering Summary")
+        default_idx = len(kmeans_cols) - 1  # most recent run
+        selected_kmeans_col = st.selectbox(
+            "KMeans result",
+            options=kmeans_cols,
+            index=default_idx,
+            key="summary_kmeans_selector",
+            help="Select which KMeans run to view summary + representative images for.",
+        )
+
+        summary_df = summaries.get(selected_kmeans_col)
+        representatives = reps_by_col.get(selected_kmeans_col)
+
+        if summary_df is None or representatives is None:
+            st.info(
+                f"No cached summary for {selected_kmeans_col}. "
+                "Re-run KMeans with this k to regenerate it."
+            )
+            return
+
+        logger.debug(f"Displaying cached clustering summary for {selected_kmeans_col}")
+        st.dataframe(summary_df, hide_index=True, width='stretch')
+
+        st.markdown("#### Representative Images")
+
+        def _resolve_local_image(idx):
+            """Return the local image path if it exists, else None (fallback)."""
+            path = df_plot.iloc[idx]["image_path"]
+            if isinstance(path, str) and os.path.exists(path):
+                return path
+            return None
+
+        def _local_caption(idx):
+            path = df_plot.iloc[idx]["image_path"]
+            return os.path.basename(path) if isinstance(path, str) else None
+
+        render_representative_images(
+            representatives,
+            resolve_image=_resolve_local_image,
+            n_per_cluster=3,
+            caption_fn=_local_caption,
+        )
     else:
-        st.info("Summary will appear here after projection.")
+        # Precalculated app: show taxonomy tree (works with or without KMeans)
+        if show_taxonomy:
+            filtered_df = st.session_state.get("filtered_df_for_clustering", None)
+            if filtered_df is not None:
+                render_taxonomic_tree_summary()

shared/components/visualization.py

@@ -77,54 +77,51 @@ def _render_chart_fragment(df_plot):
         else:
             heatmap_bins = 40  # Default, not used
 
-    # Determine color column
-    if is_precalculated:
-        # Build list of colorable columns
-        skip_color_cols = {'x', 'y', 'idx', 'uuid', 'emb', 'embedding', 'embeddings', 'vector',
-                           'identifier', 'image_url', 'url', 'img_url', 'image'}
-        colorable_cols = [c for c in df_plot.columns
-                          if c not in skip_color_cols and df_plot[c].nunique() <= 100]
-
-        # Sort KMeans columns to front (all runs, sorted by k)
-        kmeans_cols = sorted(
-            [c for c in colorable_cols if c.startswith("KMeans (k=")],
-            key=lambda c: int(c.split("=")[1].rstrip(")"))
+    # Determine color column — same dropdown pattern for both apps.
+    # Build list of colorable columns (skip technical/identifier columns).
+    skip_color_cols = {'x', 'y', 'idx', 'uuid', 'emb', 'embedding', 'embeddings', 'vector',
+                       'identifier', 'image_url', 'url', 'img_url', 'image',
+                       'image_path', 'file_name'}
+    colorable_cols = [c for c in df_plot.columns
+                      if c not in skip_color_cols and df_plot[c].nunique() <= 100]
+
+    # Sort KMeans columns to front (all runs, sorted by k)
+    kmeans_cols = sorted(
+        [c for c in colorable_cols if c.startswith("KMeans (k=")],
+        key=lambda c: int(c.split("=")[1].rstrip(")"))
+    )
+    other_cols = [c for c in colorable_cols if not c.startswith("KMeans (k=")]
+    colorable_cols = kmeans_cols + other_cols
+
+    # Build unique count lookup for display
+    col_nunique = {c: df_plot[c].nunique() for c in colorable_cols}
+
+    if colorable_cols:
+        color_col = st.selectbox(
+            "Color by",
+            options=["(none)"] + colorable_cols,
+            index=0,
+            key="color_by_column",
+            format_func=lambda c: c if c == "(none)" else f"{c} ({col_nunique[c]})",
+            help="Select a column to color the points by"
         )
-        other_cols = [c for c in colorable_cols if not c.startswith("KMeans (k=")]
-        colorable_cols = kmeans_cols + other_cols
-
-        # Build unique count lookup for display
-        col_nunique = {c: df_plot[c].nunique() for c in colorable_cols}
-
-        if colorable_cols:
-            color_col = st.selectbox(
-                "Color by",
-                options=["(none)"] + colorable_cols,
-                index=0,
-                key="color_by_column",
-                format_func=lambda c: c if c == "(none)" else f"{c} ({col_nunique[c]})",
-                help="Select a column to color the points by"
-            )
-            if color_col == "(none)":
-                color_col = None
-        else:
+        if color_col == "(none)":
             color_col = None
-
-        # Warning for high cardinality
-        if color_col and df_plot[color_col].nunique() > 20:
-            st.warning(f"'{color_col}' has {df_plot[color_col].nunique()} unique values. Colors may repeat.")
-
-        # Trigger full page rerun when color changes (so bottom section updates).
-        # Use a sentinel to distinguish "never set" from "set to None".
-        _sentinel = object()
-        prev_color = st.session_state.get("_prev_color_by", _sentinel)
-        if color_col != prev_color:
-            st.session_state["_prev_color_by"] = color_col
-            if prev_color is not _sentinel:
-                st.rerun(scope="app")
     else:
-        # embed_explore app: always color by cluster
-        color_col = 'cluster' if 'cluster' in df_plot.columns else None
+        color_col = None
+
+    # Warning for high cardinality
+    if color_col and df_plot[color_col].nunique() > 20:
+        st.warning(f"'{color_col}' has {df_plot[color_col].nunique()} unique values. Colors may repeat.")
+
+    # Trigger full page rerun when color changes (so bottom section updates).
+    # Use a sentinel to distinguish "never set" from "set to None".
+    _sentinel = object()
+    prev_color = st.session_state.get("_prev_color_by", _sentinel)
+    if color_col != prev_color:
+        st.session_state["_prev_color_by"] = color_col
+        if prev_color is not _sentinel:
+            st.rerun(scope="app")
 
     point_selector = alt.selection_point(fields=["idx"], name="point_selection")
 
@@ -133,13 +130,11 @@ def _render_chart_fragment(df_plot):
     skip_cols = {'x', 'y', 'idx', 'emb', 'embedding', 'embeddings', 'vector',
                  'uuid', 'identifier', 'image_url', 'url', 'img_url', 'image'}
 
-    # For embed_explore, include cluster/cluster_name in tooltip
-    if not is_precalculated:
-        if 'cluster_name' in df_plot.columns:
-            tooltip_fields.append('cluster_name:N')
-        elif 'cluster' in df_plot.columns:
-            tooltip_fields.append('cluster:N')
-        skip_cols.update({'cluster', 'cluster_name'})
+    # For embed_explore, include the file_name in the tooltip for quick reference
+    if not is_precalculated and 'file_name' in df_plot.columns:
+        tooltip_fields.append('file_name:N')
+        skip_cols.add('file_name')
+    skip_cols.add('image_path')
 
     # Add the color column first if set (and not already in tooltip)
     if color_col and color_col not in skip_cols:

shared/services/clustering_service.py

@@ -196,25 +196,21 @@ class ClusteringService:
         Returns:
             Tuple of (summary dataframe, representatives dict)
         """
+        from shared.utils.representatives import find_cluster_representatives
+
         logger.info("Generating clustering summary statistics")
         cluster_ids = np.unique(labels)
         logger.debug(f"Found {len(cluster_ids)} unique clusters")
-        summary_data = []
-        representatives = {}
 
+        # Ranked representative candidates per cluster (shared utility).
+        representatives = find_cluster_representatives(embeddings, labels)
+
+        summary_data = []
         for k in cluster_ids:
             idxs = np.where(labels == k)[0]
             cluster_embeds = embeddings[idxs]
             centroid = cluster_embeds.mean(axis=0)
-
-            # Internal variance
             variance = np.mean(np.sum((cluster_embeds - centroid) ** 2, axis=1))
-
-            # Find 3 closest images
-            dists = np.sum((cluster_embeds - centroid) ** 2, axis=1)
-            closest_indices = idxs[np.argsort(dists)[:3]]
-            representatives[k] = closest_indices
-
             summary_data.append({
                 "Cluster": int(k),
                 "Count": len(idxs),

shared/services/embedding_service.py

@@ -3,8 +3,68 @@ Embedding generation service.
 
 Heavy libraries (torch, open_clip) are imported lazily inside methods
 to avoid slowing down app startup.
+
+Device-aware concurrency:
+
+PyTorch has two kinds of parallelism built in, we focus on the intra-op
+parallelism which is relevant to the embedding pipeline:
+
+Intra-op is the parallelism inside a single operation. One op, say
+`Normalize` on a `[3, 244, 244]` tensor, or a big matrix multiply, splits its
+own work across multiple threads (via an openMP/MKL thread pool).
+`torch.get_num_threads()` queries how many threads one op may use, and
+`torch.set_num_threads(n)` sets it.
+
+A single `preprocess(img)` is a chain of torch ops (resize -> to_tensor ->
+normalize). With the default intra-op thread settings, each of those ops can
+fan its work out across all CPU cores. So ONE preprocess call of one image
+can momentarily spin up ~`cpu_count` threads to do that tiny bit of math. 
+
+^^^ Why that's wasteful here? 
+
+Since we already have our own parallelism layer at image level: the
+`ThreadPoolExecutor` runs `workers` threads, one image per thread, and each
+thread calls `preprocess(img)`. If each preprocess call fans out across all
+CPU cores, then `workers` threads can easily oversubscribe the CPU with
+`workers * cpu_count` threads. This causes contention and can actually slow
+down the whole process. 
+
+```
+Layer 1 (ThreadPoolExecutor):  16 worker threads, each handling one image preprocess
+Layer 2 (torch intra-op):      x Each preprocess call can use up to `cpu_count` threads
+                               ========================================================
+                               Total threads = 16 (workers) * cpu_count (intra-op) =>
+                               Potentially 256 threads on a 16-core machine,
+                               causing oversubscription and slowdown.
+```
+
+By setting `torch.set_num_threads(1)`, we ensure that each preprocess call
+runs single-thread, no internal spliting. All parallelism comes cleanly from
+one place -  the `ThreadPoolExecutor`. Instead of two nested layers that
+multiply into a thread explosion, each core does one useful thing (decode a
+whole image) with no scheduling thrash and no per-op thread-launch overhead. 
+
+```
+Layer 1 (ThreadPoolExecutor):  16 worker threads, each handling one image preprocess
+Layer 2 (torch intra-op):      x 1 (each op runs single-threaded, instantly)
+                               ========================================================
+                               Total threads = 16 (workers) * 1 (intra-op) =>
+                               Potentially 16 threads on a 16-core machine,
+                               fully utilizing the CPU without oversubscription.
+```
+
+What Intra-op is good for? 
+
+Intra-op parallelism is excellent for big ops. On the CPU-only path, the
+forward pass of the model is the bottleneck, and it benefits from intra-op
+parallelism. So we leave torch's intra-op threads alone on CPU, and cap the
+worker threads to a small number (2) to avoid too much contention. On GPU,
+the forward pass is fast and doesn't need CPU cores, so we maximize worker
+threads for decoding and set intra-op to 1 to avoid oversubscription.
+
 """
 
+import os
 import numpy as np
 import streamlit as st
 import time
@@ -79,90 +139,96 @@ class EmbeddingService:
         model_name: str,
         batch_size: int,
         n_workers: int,
-        progress_callback: Optional[Callable[[float, str], None]] = None
+        progress_callback: Optional[Callable[[float, str], None]] = None,
+        recursive: bool = False,
     ) -> Tuple[np.ndarray, List[str]]:
         """
         Generate embeddings for images in a directory.
 
+        Preprocessing runs on a thread pool (GIL-light) overlapped with the model
+        forward pass — no multiprocessing, so behavior is identical on every OS.
+
         Args:
             image_dir: Path to directory containing images
             model_name: Name of the model to use
-            batch_size: Batch size for processing
-            n_workers: Number of worker processes
+            batch_size: Batch size for the forward pass
+            n_workers: Max preprocessing threads (capped per device, see below)
             progress_callback: Optional callback for progress updates
+            recursive: Recurse into subdirectories when listing images
 
         Returns:
             Tuple of (embeddings array, list of valid image paths)
         """
         import torch
-        from hpc_inference.datasets.image_folder_dataset import ImageFolderDataset
+        from shared.utils.image_pipeline import embed_image_folder
 
         logger.info(f"Starting embedding generation: dir={image_dir}, model={model_name}, "
-                    f"batch_size={batch_size}, n_workers={n_workers}")
+                    f"batch_size={batch_size}, n_workers={n_workers}, recursive={recursive}")
         total_start = time.time()
 
         if progress_callback:
             progress_callback(0.0, "Listing images...")
 
-        image_paths = list_image_files(image_dir)
-        logger.info(f"Found {len(image_paths)} images in {image_dir}")
+        image_paths = list_image_files(image_dir, recursive=recursive)
+        total = len(image_paths)
+        logger.info(f"Found {total} images in {image_dir}")
 
         if progress_callback:
-            progress_callback(0.1, f"Found {len(image_paths)} images. Loading model...")
+            progress_callback(0.05, f"Found {total} images. Loading model...")
 
         torch_device = "cuda" if torch.cuda.is_available() else "cpu"
+        device = torch.device(torch_device)
         logger.info(f"Using device: {torch_device}")
         model, preprocess = EmbeddingService.load_model_unified(model_name, torch_device)
 
-        if progress_callback:
-            progress_callback(0.2, "Creating dataset...")
-
-        # Create dataset & DataLoader
-        dataset = ImageFolderDataset(
-            image_dir=image_dir,
-            preprocess=preprocess,
-            uuid_mode="fullpath",
-            rank=0,
-            world_size=1,
-            evenly_distribute=True,
-            validate=True
-        )
-        dataloader = torch.utils.data.DataLoader(
-            dataset,
-            batch_size=batch_size,
-            shuffle=False,
-            num_workers=n_workers,
-            pin_memory=True
-        )
+        # Device-aware concurrency:
+        cpu_count = os.cpu_count() or 1
+        prev_threads = None
+
+        if device.type == "cuda":
+            # GPU: feed the GPU with parallel decode, avoid per-op oversubscription.
+            # - preprocess threads: wide
+            # - torch intra-op threads: forced to 1
+            
+            # Set the number of preprocessing threads, clamped by three ceilings:
+            # 1) the user-requested n_workers 
+            # 2) the number of CPU cores
+            # 3) never more threads than images
+            workers = max(1, min(n_workers, cpu_count, max(total, 1)))
+            
+            prev_threads = torch.get_num_threads()
+            torch.set_num_threads(1)
 
-        total = len(image_paths)
-        valid_paths = []
-        embeddings = []
-
-        processed = 0
-        with torch.no_grad():
-            for batch_paths, batch_imgs in dataloader:
-                batch_imgs = batch_imgs.to(torch_device, non_blocking=True)
-                batch_embeds = model.encode_image(batch_imgs).cpu().numpy()
-                embeddings.append(batch_embeds)
-                valid_paths.extend(batch_paths)
-                processed += len(batch_paths)
-
-                if progress_callback:
-                    progress = 0.2 + (processed / total) * 0.8  # Use 20% to 100% for actual processing
-                    progress_callback(progress, f"Embedding {processed}/{total}")
-
-        # Stack embeddings if available
-        if embeddings:
-            embeddings = np.vstack(embeddings)
         else:
-            embeddings = np.empty((0, model.visual.output_dim))
+            # CPU: the CPU forward is the bottleneck, needs the cores, 
+            # so keep preprocess pool small and leave torch threads alone.
+            workers = max(1, min(2, n_workers, max(total, 1)))
+
+        # Map the pipeline's 0..1 progress into the 0.1..1.0 band (model load took 0..0.1).
+        def _embed_progress(frac: float, msg: str):
+            if progress_callback:
+                progress_callback(0.1 + 0.9 * frac, msg)
+
+        try:
+            embeddings, valid_paths = embed_image_folder(
+                image_paths,
+                model,
+                preprocess,
+                device,
+                batch_size=batch_size,
+                n_workers=workers,
+                progress_callback=_embed_progress,
+            )
+        finally:
+            if prev_threads is not None:
+                torch.set_num_threads(prev_threads)
 
         if progress_callback:
             progress_callback(1.0, f"Complete! Generated {embeddings.shape[0]} embeddings")
 
         total_elapsed = time.time() - total_start
-        logger.info(f"Embedding generation completed: {embeddings.shape[0]} embeddings in {total_elapsed:.2f}s "
-                    f"({embeddings.shape[0] / total_elapsed:.1f} images/sec)")
+        rate = embeddings.shape[0] / total_elapsed if total_elapsed > 0 else 0.0
+        logger.info(f"Embedding generation completed: {embeddings.shape[0]} embeddings in "
+                    f"{total_elapsed:.2f}s ({rate:.1f} images/sec)")
 
         return embeddings, valid_paths

shared/utils/backend.py

@@ -99,9 +99,8 @@ def resolve_backend(backend: str, operation: str = "general") -> str:
         return backend
 
     cuda_available, device_info = check_cuda_available()
-    has_cuml = check_cuml_available()
-
-    if cuda_available and has_cuml:
+    # Only probe for cuML when CUDA is actually available.
+    if cuda_available and check_cuml_available():
         resolved = "cuml"
         logger.info(f"Auto-resolved {operation} backend to cuML (GPU: {device_info})")
     else:

shared/utils/clustering.py

@@ -199,7 +199,9 @@ def _reduce_dim_sklearn(embeddings: np.ndarray, method: str, seed: Optional[int]
     effective_workers = -1 if n_workers > 1 else n_workers
 
     if method.upper() == "PCA":
-        reducer = PCA(n_components=2)
+        # Pass random_state so the randomized SVD solver (auto-selected for
+        # large inputs) is reproducible when a seed is set; None keeps it random.
+        reducer = PCA(n_components=2, random_state=seed)
     elif method.upper() == "TSNE":
         # Adjust perplexity to be valid for the sample size
         n_samples = embeddings.shape[0]
@@ -244,16 +246,21 @@ def _reduce_dim_cuml(embeddings: np.ndarray, method: str, seed: Optional[int], n
 
         if method.upper() == "PCA":
             from cuml.decomposition import PCA as cuPCA
+            # cuML PCA takes no random_state and needs none: its full-SVD solver
+            # is deterministic, so results are already reproducible run-to-run.
             reducer = cuPCA(n_components=2)
         elif method.upper() == "TSNE":
             from cuml.manifold import TSNE as cuTSNE
             n_samples = embeddings.shape[0]
             perplexity = min(30, max(5, n_samples // 3))
 
+            # Force the exact solver: cuML's default Barnes-Hut collapses to a
+            # ~1D line on near-homogeneous data (#40). exact is O(N^2) but fine
+            # at our interactive scale; a faster Barnes-Hut-with-guard can come later.
             if seed is not None:
-                reducer = cuTSNE(n_components=2, perplexity=perplexity, random_state=seed)
+                reducer = cuTSNE(n_components=2, perplexity=perplexity, method="exact", random_state=seed)
             else:
-                reducer = cuTSNE(n_components=2, perplexity=perplexity)
+                reducer = cuTSNE(n_components=2, perplexity=perplexity, method="exact")
         else:
             raise ValueError("Unsupported method. Choose 'PCA', 'TSNE', or 'UMAP'.")
 

shared/utils/image_pipeline.py

@@ -0,0 +1,154 @@
+"""Thread-parallel image embedding pipeline.
+
+Turns a list of image paths into embeddings on a single machine. Preprocessing
+(decode + transform) runs on a thread pool; the model forward runs on the
+calling thread that owns the device. Each batch is preprocessed while the
+previous batch runs through the model (a one-batch prefetch), so CPU decoding
+and the device forward overlap.
+
+Threads — rather than worker processes — carry the preprocessing because the
+work is GIL-light: PIL decode and torchvision tensor ops release the GIL, so a
+thread pool scales nearly linearly. Staying in one process means no per-image
+data crosses a process boundary and there is no worker-spawn cost, so small
+folders are cheap and behavior does not depend on the OS.
+
+This module is Streamlit-free and unit-testable.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures as cf
+from collections import deque
+from typing import Callable, List, Optional, Tuple
+
+import numpy as np
+import torch
+from PIL import Image
+
+from shared.utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+def _output_dim(model) -> int:
+    """Best-effort embedding width, for shaping an empty result."""
+    return int(getattr(getattr(model, "visual", None), "output_dim", 0) or 0)
+
+
+def _preprocess_one(path: str, preprocess: Callable, color_mode: str):
+    """Decode + preprocess a single image.
+
+    Returns ``(path, tensor)`` on success or ``(path, None)`` if the file can't
+    be read/decoded. Pure and device-free, so it is safe on worker threads.
+    """
+    try:
+        with Image.open(path) as im:
+            img = im.convert(color_mode)
+        return path, preprocess(img)
+    except Exception as e:
+        logger.warning(f"[Embed] Skipping unreadable image {path}: {e}")
+        return path, None
+
+
+def embed_image_folder(
+    image_paths: List[str],
+    model,
+    preprocess: Callable,
+    device: torch.device,
+    *,
+    batch_size: int = 32,
+    n_workers: int = 8,
+    prefetch_batches: int = 1,
+    color_mode: str = "RGB",
+    progress_callback: Optional[Callable[[float, str], None]] = None,
+) -> Tuple[np.ndarray, List[str]]:
+    """Embed a list of image paths, overlapping preprocessing with the forward.
+
+    Preprocessing runs on a ``ThreadPoolExecutor``; the model forward runs on the
+    calling thread (which owns ``device``). Up to ``prefetch_batches`` batches are
+    preprocessed ahead of the batch currently being run through the model.
+
+    Unreadable images are skipped (and logged), so the returned embeddings may
+    have fewer rows than ``image_paths``. ``embeddings[i]`` corresponds to
+    ``valid_paths[i]``.
+
+    Args:
+        image_paths: Image file paths to embed.
+        model: Model exposing ``encode_image(tensor) -> tensor``.
+        preprocess: Callable mapping a PIL image to a CHW tensor.
+        device: Torch device the model lives on.
+        batch_size: Images per forward pass.
+        n_workers: Preprocessing threads.
+        prefetch_batches: Batches to preprocess ahead of the forward (overlap).
+        color_mode: PIL convert mode applied before preprocessing.
+        progress_callback: Optional ``(fraction, message)`` progress sink.
+
+    Returns:
+        ``(embeddings [N, D] float array, valid_paths [N])``.
+    """
+    total = len(image_paths)
+    if total == 0:
+        return np.empty((0, _output_dim(model)), dtype=np.float32), []
+
+    batches = [image_paths[i:i + batch_size] for i in range(0, total, batch_size)]
+    window = max(1, prefetch_batches + 1)
+
+    emb_chunks: List[np.ndarray] = []
+    valid_paths: List[str] = []
+    processed = 0
+    
+    # concurrent.futures.ThreadPoolExecutor(max_workers=n_workers)
+    # spins up `n_workers` OS threads sitting idle, waiting for work...
+    # hand it work with ex.submit(fn, *args), which returns a Future immediately, 
+    # and runs fn(*args) on a worker thread when it gets scheduled by the OS...
+    with cf.ThreadPoolExecutor(max_workers=n_workers) as ex:
+        
+        # non-blocking, starts the preprocessing of a batch on the pool
+        def submit(batch: List[str]) -> List[cf.Future]:
+            return [ex.submit(_preprocess_one, p, preprocess, color_mode) for p in batch]
+
+        # Prime the pipeline so the first forward already has successors decoding.
+        # pending is a queue of lists of futures, one list per batch.
+        pending: deque = deque()
+        next_idx = 0
+        while next_idx < len(batches) and len(pending) < window:
+            pending.append(submit(batches[next_idx]))
+            next_idx += 1
+        # pending is now a full window of batches being preprocessed
+        
+        with torch.no_grad():
+            while pending:
+                # Take the oldest in-flight batch
+                futures = pending.popleft()              
+                # Refill the window first: these batches preprocess on the pool
+                # while we run the current batch through the model below.
+                if next_idx < len(batches):
+                    pending.append(submit(batches[next_idx]))
+                    next_idx += 1
+
+                # If the worker alreadt=y finished, returns immediately; 
+                # otherwise blocks until the batch is ready.
+                results = [f.result() for f in futures]     
+                batch_paths = [p for p, t in results if t is not None]
+                tensors = [t for _, t in results if t is not None]
+
+                if tensors:
+                    x = torch.stack(tensors).to(device)
+                    feats = model.encode_image(x).cpu().numpy()
+                    emb_chunks.append(feats)
+                    valid_paths.extend(batch_paths)
+
+                processed += len(futures)
+                if progress_callback:
+                    progress_callback(processed / total, f"Embedding {processed}/{total}")
+
+    if emb_chunks:
+        embeddings = np.vstack(emb_chunks)
+    else:
+        embeddings = np.empty((0, _output_dim(model)), dtype=np.float32)
+
+    logger.info(
+        f"[Embed] {embeddings.shape[0]}/{total} images embedded "
+        f"({total - embeddings.shape[0]} skipped)"
+    )
+    return embeddings, valid_paths

shared/utils/images.py

@@ -0,0 +1,197 @@
+"""Shared image-fetching utilities.
+
+App-agnostic helpers for resolving and fetching record images from remote
+URLs. Kept free of Streamlit so the helpers run safely on worker threads and
+in any app (precalculated, a URL-based embed_explore, the demo Space, ...).
+
+In-app fetch flow
+-----------------
+A record (parquet row) holds an image URL in one of ``IMAGE_URL_COLUMNS``.
+Two call paths consume these:
+
+1. Cluster representatives (bulk, eager).
+   ``render_cluster_representatives`` resolves a URL per candidate with
+   ``resolve_record_image_url`` and warms the cache up front via
+   ``fetch_images_concurrent`` (thread pool, 8 workers). Each thread calls
+   ``download_image_bytes`` -> ``bytes_to_image`` and stores the PIL image
+   (or ``None`` on failure) in ``_IMAGE_CACHE``. The renderer then reads
+   results straight from the cache; broken URLs are skipped and the next
+   candidate is tried.
+
+2. Click preview (single, lazy).
+   ``render_data_preview`` resolves one URL and calls ``get_image_from_url``,
+   which serves the cached image if present and otherwise does a single
+   synchronous ``download_image_bytes`` -> ``bytes_to_image`` and caches it.
+
+So both paths share one fetch primitive and one cache; the only difference is
+concurrent prefetch vs. on-demand single fetch.
+
+Why a process-level cache (not ``@st.cache_data``)
+--------------------------------------------------
+- The bulk path fetches from worker threads, where ``st.*`` calls are unsafe;
+  a plain module-level dict is thread-friendly and lets both paths share the
+  same entries.
+- It survives Streamlit reruns within the process, so panning/clicking does
+  not refetch. A soft FIFO cap (``_IMAGE_CACHE_MAX``) bounds memory. 
+  Trimming only happens at the end of a fetch call, not on every insertion.
+- ``None`` is cached as a known miss, so a dead URL is fetched at most once.
+
+The single shared ``requests.Session`` carries the project User-Agent so data
+hosts can identify / allowlist us.
+"""
+
+import concurrent.futures
+import time
+from io import BytesIO
+from typing import Dict, Iterable, Optional
+
+import requests
+from PIL import Image
+
+from shared import __version__ as _EMB_EXPLORER_VERSION
+from shared.utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+# Columns checked, in order, for an image URL when resolving a record's image.
+IMAGE_URL_COLUMNS = ['identifier', 'image_url', 'url', 'img_url', 'image']
+
+# Be a polite client: identify the app and link the repo so data hosts can
+# contact us / allowlist us if needed.
+USER_AGENT = (
+    f"emb-explorer/{_EMB_EXPLORER_VERSION} "
+    "(+https://github.com/Imageomics/emb-explorer)"
+)
+
+_session: Optional[requests.Session] = None
+
+
+def _get_session() -> requests.Session:
+    """Lazily build a shared requests.Session carrying our User-Agent."""
+    global _session
+    if _session is None:
+        s = requests.Session()
+        s.headers.update({"User-Agent": USER_AGENT})
+        _session = s
+    return _session
+
+
+def download_image_bytes(url: str, timeout: int = 5) -> Optional[bytes]:
+    """Fetch raw image bytes via the shared session. None on any failure.
+
+    Contains no Streamlit calls, so it is safe to run from worker threads.
+    """
+    if not isinstance(url, str) or not url.startswith(('http://', 'https://')):
+        return None
+    try:
+        resp = _get_session().get(url, timeout=timeout, stream=True)
+        resp.raise_for_status()
+        if not resp.headers.get('content-type', '').lower().startswith('image/'):
+            return None
+        return resp.content
+    except Exception:
+        return None
+
+
+def bytes_to_image(data: Optional[bytes]) -> Optional[Image.Image]:
+    """Decode image bytes to a PIL image, or None on failure."""
+    if not data:
+        return None
+    try:
+        return Image.open(BytesIO(data))
+    except Exception as e:
+        logger.error(f"[Image] Failed to open: {e}")
+        return None
+
+
+# Process-level cache for fetched images. Survives Streamlit reruns within the
+# process; value is a PIL image or None (known miss).
+_IMAGE_CACHE: Dict[str, Optional[Image.Image]] = {}
+_IMAGE_CACHE_MAX = 512
+
+
+def _trim_cache() -> None:
+    """Soft FIFO cap so the cache doesn't grow unbounded across sessions."""
+    if len(_IMAGE_CACHE) > _IMAGE_CACHE_MAX:
+        for k in list(_IMAGE_CACHE.keys())[: len(_IMAGE_CACHE) - _IMAGE_CACHE_MAX]:
+            _IMAGE_CACHE.pop(k, None)
+
+
+def fetch_images_concurrent(
+    urls: Iterable[str], max_workers: int = 8, timeout: int = 5
+) -> Dict[str, Optional[Image.Image]]:
+    """Fetch many image URLs concurrently with a thread pool.
+
+    Returns {url: PIL image or None}. Per-URL results are cached in a
+    process-level dict so reruns and overlapping clusters don't refetch.
+    Threads only do HTTP + PIL decode (no st.* calls), which is Streamlit-safe.
+    """
+    unique = [u for u in dict.fromkeys(urls) if isinstance(u, str) and u]
+    missing = [u for u in unique if u not in _IMAGE_CACHE]
+
+    if missing:
+        t0 = time.time()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
+            future_to_url = {
+                ex.submit(download_image_bytes, u, timeout): u for u in missing
+            }
+            for fut in concurrent.futures.as_completed(future_to_url):
+                u = future_to_url[fut]
+                try:
+                    _IMAGE_CACHE[u] = bytes_to_image(fut.result())
+                except Exception:
+                    _IMAGE_CACHE[u] = None
+        ok = sum(1 for u in missing if _IMAGE_CACHE.get(u) is not None)
+        logger.info(
+            f"[Image] Concurrently fetched {len(missing)} url(s) in "
+            f"{time.time() - t0:.2f}s ({ok} ok)"
+        )
+        _trim_cache()
+
+    return {u: _IMAGE_CACHE.get(u) for u in unique}
+
+
+def get_image_from_url(url: str, timeout: int = 5) -> Optional[Image.Image]:
+    """Get a single image from a URL, using the process cache.
+
+    Logs the request; results (including misses) are cached so repeated
+    lookups and the concurrent path share one cache.
+    """
+    if not url or not isinstance(url, str):
+        return None
+    if url in _IMAGE_CACHE:
+        return _IMAGE_CACHE[url]
+    if not url.startswith(('http://', 'https://')):
+        logger.warning(f"[Image] Invalid URL scheme: {url[:50]}...")
+        return None
+
+    logger.info(f"[Image] Fetching: {url[:80]}...")
+    start_time = time.time()
+    image = bytes_to_image(download_image_bytes(url, timeout))
+    elapsed = time.time() - start_time
+    if image is not None:
+        logger.info(f"[Image] Loaded in {elapsed:.3f}s")
+    else:
+        logger.warning(f"[Image] Failed to load: {url[:50]}...")
+
+    _IMAGE_CACHE[url] = image
+    _trim_cache()
+    return image
+
+
+def resolve_record_image_url(row) -> Optional[str]:
+    """Return the first valid HTTP(S) image URL from a record/row, else None.
+
+    `row` is anything supporting `col in row` membership and `row[col]`
+    indexing (e.g. a pandas Series or a dict).
+    """
+    for col in IMAGE_URL_COLUMNS:
+        try:
+            present = col in row.index
+        except AttributeError:
+            present = col in row
+        if present:
+            val = row[col]
+            if isinstance(val, str) and val.startswith(('http://', 'https://')):
+                return val
+    return None

shared/utils/io.py

@@ -1,22 +1,37 @@
 import os
 import shutil
 
-def list_image_files(image_dir, allowed_extensions=('jpg', 'jpeg', 'png')):
+# Image extensions we attempt to load (PIL-decodable raster formats).
+IMAGE_EXTENSIONS = ('.jpg', '.jpeg', '.png', '.bmp', '.tif', '.tiff', '.webp')
+
+
+def list_image_files(image_dir, allowed_extensions=IMAGE_EXTENSIONS, recursive=False):
     """
     List image file paths in a directory with allowed extensions.
 
     Args:
         image_dir (str): Path to the directory containing images.
-        allowed_extensions (tuple, optional): Allowed file extensions. Defaults to ('jpg', 'jpeg', 'png').
+        allowed_extensions (tuple, optional): Allowed file extensions (lowercase,
+            leading dot). Defaults to IMAGE_EXTENSIONS.
+        recursive (bool, optional): Recurse into subdirectories. Defaults to False.
 
     Returns:
-        list: List of full file paths for images with allowed extensions.
+        list: Sorted list of full file paths for images with allowed extensions.
     """
-    return [
-        os.path.join(image_dir, f)
-        for f in os.listdir(image_dir)
-        if f.lower().endswith(allowed_extensions)
-    ]
+    if recursive:
+        paths = [
+            os.path.join(root, f)
+            for root, _, files in os.walk(image_dir)
+            for f in files
+            if f.lower().endswith(allowed_extensions)
+        ]
+    else:
+        paths = [
+            os.path.join(image_dir, f)
+            for f in os.listdir(image_dir)
+            if f.lower().endswith(allowed_extensions)
+        ]
+    return sorted(paths)
     
 def copy_image(row, repartition_dir):
     """

shared/utils/models.py

@@ -4,9 +4,11 @@ def list_available_models():
     # Create list of all models
     models_data = []
     
-    # Add special models first
+    # Add special models first (Imageomics BioCLIP family)
     models_data.extend([
         {"name": "hf-hub:imageomics/bioclip-2", "pretrained": None},
+        {"name": "hf-hub:imageomics/bioclip-2.5-vith14", "pretrained": None},
+        {"name": "hf-hub:imageomics/biocap", "pretrained": None},
         {"name": "hf-hub:imageomics/bioclip", "pretrained": None}
     ])
     

shared/utils/representatives.py

@@ -0,0 +1,62 @@
+"""Find representative members of clusters.
+
+Given embeddings and cluster labels, rank each cluster's members by proximity
+to the cluster centroid. Returns more candidates than strictly requested
+(oversampled) so callers that render images can skip candidates whose image
+fails to load and still show the desired number per cluster.
+"""
+
+from typing import Dict, List
+
+import numpy as np
+
+from shared.utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+def find_cluster_representatives(
+    embeddings: np.ndarray,
+    labels,
+    n_per_cluster: int = 3,
+    oversample: int = 4,
+) -> Dict[object, List[int]]:
+    """Rank each cluster's members by closeness to the cluster centroid.
+
+    Args:
+        embeddings: (N, D) array of embeddings (row i aligns with label i).
+        labels: array-like of length N with cluster labels (int or str).
+        n_per_cluster: how many representatives the caller intends to show.
+        oversample: multiplier for how many candidate indices to return per
+            cluster (n_per_cluster * oversample), so failed image loads can be
+            skipped while still surfacing n_per_cluster images.
+
+    Returns:
+        Dict mapping each cluster label to a list of global indices into
+        `embeddings`, ordered closest-to-centroid first, capped at
+        n_per_cluster * oversample (or the cluster size, whichever is smaller).
+    """
+    labels = np.asarray(labels)
+    embeddings = np.asarray(embeddings)
+    n_candidates = max(n_per_cluster * oversample, n_per_cluster)
+
+    representatives: Dict[object, List[int]] = {}
+    for cluster_id in np.unique(labels):
+        member_idxs = np.where(labels == cluster_id)[0]
+        if member_idxs.size == 0:
+            continue
+        cluster_embeds = embeddings[member_idxs]
+        centroid = cluster_embeds.mean(axis=0)
+        
+        # Compute squared Euclidean distance to the centroid for each member.
+        dists = np.sum((cluster_embeds - centroid) ** 2, axis=1)
+        order = np.argsort(dists)[:n_candidates]
+        # Keep the label's native Python type for clean dict keys / display.
+        key = cluster_id.item() if hasattr(cluster_id, "item") else cluster_id
+        representatives[key] = member_idxs[order].tolist()
+
+    logger.debug(
+        f"Found representatives for {len(representatives)} clusters "
+        f"(up to {n_candidates} candidates each)"
+    )
+    return representatives