diff --git a/README.md b/README.md index 2cff7ad..b05f75f 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,39 @@ Visual exploration and clustering tool for image embeddings. Users can either br **Precalculated Embeddings** - Load parquet files (or directories of parquets) with precomputed embeddings, apply dynamic cascading filters, and explore clusters with taxonomy tree navigation. See [Data Format](docs/DATA_FORMAT.md) for the expected schema and [Backend Pipeline](docs/BACKEND_PIPELINE.md) for how embeddings flow through clustering and visualization. +**Representative Images** - After KMeans, both apps show the three records closest to each cluster's centroid. Each tile carries the record name, a short id that copies the full id on click, and a source link that opens the image in a new tab; images that fail to load are skipped in favor of the next-closest record, and images over the 25 MiB download cap are shown as a placeholder with their source link intact. The example below is KMeans on [BioCLIP 2](https://huggingface.co/imageomics/bioclip-2) embeddings of a sample of canid images. See [Representative Images](docs/REPRESENTATIVE_IMAGES.md) for how selection and fallback work. + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
Cluster 1
Canis simensis
dbb9ca8f · source		Canis simensis
70f9ab9f · source		Canis simensis
4de7a1cf · source
Ethiopian wolfEthiopian wolfEthiopian wolf
Cluster 2
Canis lycaon
ea070e30 · source		Canis lycaon
55af17f4 · source		Canis lycaon
a7edd26f · source
paw print in snowpaw print in snowplaceholder shown for an image over the download cap
+ +Photos [CC BY-NC 4.0](https://creativecommons.org/licenses/by-nc/4.0/deed.en): Björn Reese, Miguel A. Casado, Bill Fintel, corium. + ## Installation Requires Python 3.10–3.13. We recommend using [uv](https://github.com/astral-sh/uv) for package management. diff --git a/apps/precalculated/app.py b/apps/precalculated/app.py index 9d19650..2dd6de8 100644 --- a/apps/precalculated/app.py +++ b/apps/precalculated/app.py @@ -26,7 +26,10 @@ def app(): render_projection_section, render_kmeans_section, ) - from apps.precalculated.components.data_preview import render_data_preview + 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 @@ -71,9 +74,10 @@ 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() if __name__ == "__main__": diff --git a/apps/precalculated/components/data_preview.py b/apps/precalculated/components/data_preview.py index f855a8a..d145cdf 100644 --- a/apps/precalculated/components/data_preview.py +++ b/apps/precalculated/components/data_preview.py @@ -6,80 +6,20 @@ 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 ( + ImageTooLarge, + fetch_images_concurrent, + get_record_image, + resolve_record_image_url, +) +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 +50,17 @@ 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 an image, falling back across the record's URL + # columns until one loads (process-cached). An over-cap image keeps + # its URL visible so the user can still open it in a browser. + image = get_record_image(record) + if isinstance(image, ImageTooLarge): + st.warning( + f"Image not downloaded — {image.describe()} exceeds the fetch limit." + ) + st.markdown(f"[Open original image]({image.url})") + elif image is not None: + st.image(image, width=280) st.markdown(f"**UUID:** `{selected_uuid}`") @@ -276,3 +218,86 @@ 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, first URL per record) in parallel. Deeper fallbacks (later + # candidates, or a record's alternate URL columns) 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): + # Prefetched URLs hit the process cache inside get_record_image; broken + # URLs fall back to the record's next URL column, and anything not + # prefetched falls back to a single synchronous fetch (also cached). + return get_record_image(df_plot.iloc[idx]) + + 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 + + def _details(idx): + row = df_plot.iloc[idx] + return {"id": str(row["uuid"]), "url": resolve_record_image_url(row)} + + render_representative_images( + representatives, + resolve_image=_resolve, + n_per_cluster=n_per_cluster, + caption_fn=_caption, + details_fn=_details, + ) diff --git a/docs/REPRESENTATIVE_IMAGES.md b/docs/REPRESENTATIVE_IMAGES.md new file mode 100644 index 0000000..ff4cfa8 --- /dev/null +++ b/docs/REPRESENTATIVE_IMAGES.md @@ -0,0 +1,107 @@ +# Representative images per cluster + +After a KMeans run, both apps show a **Representative Images** panel: for every cluster, up to three images of the records that sit closest to the cluster's center. This page explains how those records are chosen, how their images are loaded, what happens when an image cannot be shown. + +## 1. Overview: What the panel shows + +Each cluster is one card. Inside it, every tile has a short header above the image: + +- the record's name (`scientific_name`, else `species`, `common_name`, or the uuid; in Embed & Explore the tile shows a single `File path` label instead), +- a **short id** that copies the full uuid (or file path) to the clipboard when clicked; hovering previews the full value, +- a **source** link that opens the fetched image URL in a new tab. + +The example card below mixes two rendered images with one that was too large to download. The third tile is the app's own placeholder, drawn by the same function the app uses, and its "source" link still opens the original. + + + + + + + + + + + + + + + + + +
Cluster 2
Sombrero Galaxy
3f1c9a0b · source		Horsehead Nebula
9a77e2d1 · source		Cosmic Reef
51be0c3e · source
rendered imagerendered imageplaceholder for an image over the download cap
+ +Example images: ESA/Hubble (CC BY 4.0). The third source is a 97 MiB file, above the app's 25 MiB download cap, so the app shows the placeholder instead of fetching it. + +A cluster whose records all fail to load shows the caption *"No images available for this cluster."* instead of tiles. + +## 2. How representatives are chosen + +For each cluster, the app takes the mean of its members' embeddings (the **centroid**, computed in the full embedding space, not in the 2D projection) and ranks the members by squared Euclidean distance to it. The closest three are the representatives. + +In symbols: let $x_i \in \mathbb{R}^D$ be the embedding of record $i$ (for BioCLIP 2, $D = 768$) and $C_k = \{\, i : \text{label}_i = k \,\}$ the members of cluster $k$. The centroid and each member's distance to it are + +$$ +\mu_k = \frac{1}{|C_k|} \sum_{i \in C_k} x_i, +\qquad +d_i = \lVert x_i - \mu_k \rVert^2 = \sum_{j=1}^{D} (x_{ij} - \mu_{kj})^2 . +$$ + +Sorting the members of $C_k$ by $d_i$ ascending gives the ranked list $r_1, r_2, \dots$; the app keeps the first + +$$ +K = \min\big(n \cdot o,\; |C_k|\big), \qquad n = 3,\; o = 4, +$$ + +as **candidates**, and the **representatives shown** are the first $n$ of those whose image can be displayed. With every image loadable they are simply $r_1, r_2, r_3$. + +The ranking is **oversampled** by the factor $o$: the app keeps 4 times as many candidates as it intends to show (12 for 3 slots, or the whole cluster if it is smaller). The extra candidates are backups, used only when an image cannot be shown. Representatives therefore never change because of a broken URL: the same three records are tried first every time; a failure moves on to the fourth-closest, and so on. + +Two details of the implementation: ties in $d_i$ are broken by record order, and $\mu_k$ is computed on the embeddings as stored in the dataset, whereas KMeans itself clusters L2-normalized copies, so for unnormalized embeddings a member's rank can differ slightly from its rank by distance to the normalized centroid. + +ranking of cluster members by distance to the centroid + +Two consequences worth knowing: + +- Representatives depend only on the embeddings and the KMeans labels, not on the projection method; PCA, t-SNE and UMAP views of the same KMeans run share the same representatives. +- A cluster with fewer than three loadable images shows fewer tiles rather than borrowing from another cluster. + +## 3. How images are loaded, and what happens when they cannot be + +```mermaid +flowchart TD + A[KMeans labels + embeddings] --> B[rank members per cluster
closest to centroid first, keep 12] + B --> C[prefetch the first URL of the first 6 candidates per cluster
8 threads, results cached per process] + C --> D{for each candidate, in rank order} + D --> E[walk the record's URL columns in order
identifier, image_url, url, img_url, image] + E --> F{fetch one URL} + F -->|HTTP error, not an image,
undecodable| G[try next column] + F -->|Content-Length or stream
exceeds 25 MiB| H[remember as too large,
try next column] + F -->|image decodes| I[image for this record] + G --> J{columns left?} + H --> J + J -->|yes| E + J -->|no, and a too-large URL was seen| K[too large for this record] + J -->|no| L[no image for this record:
skip, slot not consumed] + I --> M{3 images collected?} + K --> M + L --> M + M -->|not yet, candidates left| D + M -->|done| N[render: images first,
too-large placeholders fill empty slots,
nothing at all: No images available] +``` + +In words, there are two fallback layers: + +1. **Per record, across URL columns.** A record may hold image URLs in several columns (`identifier`, `image_url`, `url`, `img_url`, `image`; see [`shared/utils/images.py`](../shared/utils/images.py)). The app tries them in that order and uses the first one that actually loads. A broken column moves on to the next. A column whose image is over the size cap also moves on; it is remembered only if no column loads. +2. **Per cluster, across candidates.** The renderer walks the ranked candidates until it has three real images. A record with nothing loadable is skipped and consumes no slot. A record whose best outcome is "too large" is set aside; real images fill the slots first, and the set-aside records fill only the slots that would otherwise stay empty, as placeholder tiles. If there is nothing at all, the cluster shows the empty caption. + +how the renderer walks the ranked candidates + +### Why there is a size cap + +The cap originates from a [GitHub Copilot review comment on the feature PR](https://github.com/Imageomics/emb-explorer/pull/42#discussion_r3598642716), which pointed out that the first version read every response into memory with no limit, so a malicious or misconfigured URL returning a very large payload could drive memory use up and stall the app. The app now refuses any image larger than 25 MiB: it checks the declared Content-Length before reading a single byte, and it also counts bytes while streaming and stops as soon as the limit is passed, since a server can omit or misstate Content-Length. + +Every fetch result is cached for the lifetime of the app process, including failures and too-large verdicts, so switching views or re-running KMeans does not refetch, and a dead URL is tried at most once per session. Fetches time out after 5 seconds. + +## 4. Known limits + +HTTP 429 (rate limiting) and 503 are treated like any other failure, with no retry. Hosts that throttle the app's User-Agent, Wikimedia Commons for example, can make a cluster look empty. diff --git a/docs/images/readme_rep_c1_1.png b/docs/images/readme_rep_c1_1.png new file mode 100644 index 0000000..a5fc218 Binary files /dev/null and b/docs/images/readme_rep_c1_1.png differ diff --git a/docs/images/readme_rep_c1_2.png b/docs/images/readme_rep_c1_2.png new file mode 100644 index 0000000..3d3fd66 Binary files /dev/null and b/docs/images/readme_rep_c1_2.png differ diff --git a/docs/images/readme_rep_c1_3.png b/docs/images/readme_rep_c1_3.png new file mode 100644 index 0000000..91db436 Binary files /dev/null and b/docs/images/readme_rep_c1_3.png differ diff --git a/docs/images/readme_rep_c2_1.png b/docs/images/readme_rep_c2_1.png new file mode 100644 index 0000000..a596a51 Binary files /dev/null and b/docs/images/readme_rep_c2_1.png differ diff --git a/docs/images/readme_rep_c2_2.png b/docs/images/readme_rep_c2_2.png new file mode 100644 index 0000000..4abbefa Binary files /dev/null and b/docs/images/readme_rep_c2_2.png differ diff --git a/docs/images/representatives_candidate_walk.png b/docs/images/representatives_candidate_walk.png new file mode 100644 index 0000000..ef713b6 Binary files /dev/null and b/docs/images/representatives_candidate_walk.png differ diff --git a/docs/images/representatives_ranking.png b/docs/images/representatives_ranking.png new file mode 100644 index 0000000..4667bc0 Binary files /dev/null and b/docs/images/representatives_ranking.png differ diff --git a/docs/images/representatives_tile_1.png b/docs/images/representatives_tile_1.png new file mode 100644 index 0000000..c069374 Binary files /dev/null and b/docs/images/representatives_tile_1.png differ diff --git a/docs/images/representatives_tile_2.png b/docs/images/representatives_tile_2.png new file mode 100644 index 0000000..48c3f3f Binary files /dev/null and b/docs/images/representatives_tile_2.png differ diff --git a/docs/images/representatives_tile_too_large.png b/docs/images/representatives_tile_too_large.png new file mode 100644 index 0000000..40c9774 Binary files /dev/null and b/docs/images/representatives_tile_too_large.png differ diff --git a/pyproject.toml b/pyproject.toml index b19de98..922fbab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ dependencies = [ "pandas>=2.0.0", "pillow>=9.0.0", "pyarrow>=10.0.0", + "requests>=2.28.0", # Visualization "altair>=5.0.0", # Machine learning @@ -97,12 +98,6 @@ emb-embed-explore = "apps.embed_explore.app:main" emb-precalculated = "apps.precalculated.app:main" list-models = "shared.utils.models:print_available_models" -[tool.hatch.version] -path = "shared/__init__.py" - -[tool.hatch.metadata] -allow-direct-references = true - [tool.hatch.build.targets.wheel] packages = ["shared", "apps"] diff --git a/shared/__init__.py b/shared/__init__.py index ae13b8e..d0214a5 100644 --- a/shared/__init__.py +++ b/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" diff --git a/shared/components/representatives.py b/shared/components/representatives.py new file mode 100644 index 0000000..12f4002 --- /dev/null +++ b/shared/components/representatives.py @@ -0,0 +1,221 @@ +"""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. + +URL-backed resolvers may also return an `ImageTooLarge` sentinel (see +`shared.utils.images`): the record's image exists but exceeded the download +cap. Real images still take priority for the cluster's slots; too-large +candidates only fill slots that would otherwise stay empty, rendered as an +informative placeholder tile with a link to the original URL. +""" + +import html +import json +from typing import Any, Callable, Dict, List, Optional, Tuple + +import streamlit as st +import streamlit.components.v1 as components + +from shared.utils.images import ImageTooLarge, too_large_placeholder +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) + + +# Header rendered above each tile: an optional name line plus the id/source +# line. Heights are fixed so tiles with different image sizes still align. +_ID_LINE_HEIGHT = 26 +_HEADER_HEIGHT = 48 + + +def build_tile_header_html(details: Dict[str, Any], caption: Optional[str] = None) -> str: + """HTML for the block above a tile: the record name (``caption``) and a + short id that copies the full id on click, plus a "source" link that + opens the record's URL in a new tab. + + ``details`` keys: ``id`` (full text to copy; required), ``id_short`` + (display text, default the first 8 characters of ``id``; a label such as + "File path" also works), ``url`` (optional http(s) link). Hovering the + label shows the full value. Everything is escaped; the full value travels + in a data attribute, never in the visible text. + + Copying uses the async Clipboard API where the page is a secure context + (https, or http://localhost such as a port-forward) and falls back to + the legacy execCommand path elsewhere (e.g. a node's plain-http URL); + if both fail the full id is shown inline so it can be selected by hand. + """ + full = str(details.get("id", "")) + short = str(details.get("id_short") or full[:8]) + url = details.get("url") + link = "" + if isinstance(url, str) and url.startswith(("http://", "https://")): + link = (f'·' + f'source') + name = f'
{html.escape(caption)}
' if caption else "" + return f""" +{name}
{html.escape(short)}{link}
+""" + + +def _embed_html(markup: str, height: int) -> None: + """Embed our own (fully escaped) markup. st.iframe is the current API; + components.v1.html is the fallback below the Streamlit version that + introduced it (the project floor is 1.50).""" + if hasattr(st, "iframe"): + st.iframe(markup, height=height) + else: + components.html(markup, height=height) + + +def _render_tile_header(cell: Any, caption: Optional[str], details: Optional[Dict[str, Any]]) -> None: + if details and details.get("id"): + with cell: + _embed_html(build_tile_header_html(details, caption), + _HEADER_HEIGHT if caption else _ID_LINE_HEIGHT) + + +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], Optional[str]]] = None, + columns: int = 3, + details_fn: Optional[Callable[[int], Optional[Dict[str, Any]]]] = None, +) -> 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), an + `ImageTooLarge` sentinel (over-cap URL, rendered as a placeholder + tile), 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, or None for no caption. + columns: images per row. + details_fn: optional idx -> {"id", "id_short", "url"}. When given, each + tile gets a header above the image with the caption, a short id + that copies the full id on click, and a "source" link that opens + the URL in a new tab (see `build_tile_header_html`); the image + itself is then rendered without a caption. None keeps the plain + captioned-image layout. + """ + if details_fn is not None: + st.caption("Above each image: click the underlined label to copy its full " + "value (hover to preview it); \"source\" opens the image in a new tab.") + + for cluster_id in _sorted_cluster_ids(representatives): + candidates = representatives[cluster_id] + # One bordered card per cluster so the grid reads as separate modules. + card = st.container(border=True) + card.markdown(f"**Cluster {cluster_id}**") + + # Walk ranked candidates, collecting successful resolutions until we + # have n_per_cluster (or run out of candidates). Over-cap candidates + # are set aside: a later candidate with a loadable image is a better + # slot-filler than a placeholder. + shown: List[Tuple[Any, Optional[str], Optional[Dict[str, Any]]]] = [] # (displayable, caption, details) + too_large: List[Tuple[ImageTooLarge, Optional[str], Optional[Dict[str, Any]]]] = [] + 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 None: + continue + caption = caption_fn(idx) if caption_fn else None + details = details_fn(idx) if details_fn else None + if isinstance(img, ImageTooLarge): + too_large.append((img, caption, details)) + else: + shown.append((img, caption, details)) + + # Fill remaining slots with informative too-large placeholders. + placeholders = too_large[: n_per_cluster - len(shown)] + + # Nothing renderable in this cluster: every candidate (up to + # n_per_cluster * oversample, or the whole cluster if smaller) failed + # on all of its URL columns, and none was merely over the size cap, + # so there is not even a placeholder to show. + if not shown and not placeholders: + card.caption("No images available for this cluster.") + continue + + entries = shown + placeholders + cols = card.columns(min(columns, len(entries))) + for i, (img, caption, details) in enumerate(entries): + cell = cols[i % len(cols)] + has_header = bool(details and details.get("id")) + if has_header: + # Name, id and source above the image: tops stay aligned even + # when images have different heights. + _render_tile_header(cell, caption, details) + tile_caption = None if has_header else caption + if isinstance(img, ImageTooLarge): + # The tile itself can't carry a link; point at where it is. + hint = 'click "source" above to open' if has_header else "see the link below" + cell.image(too_large_placeholder(img, hint=hint), caption=tile_caption, width="stretch") + if not has_header: # with a header, "source" already links the URL + cell.caption(f"[Open original image]({img.url})") + else: + cell.image(img, caption=tile_caption, width="stretch") diff --git a/shared/components/summary.py b/shared/components/summary.py index 1a74847..668446e 100644 --- a/shared/components/summary.py +++ b/shared/components/summary.py @@ -6,6 +6,7 @@ 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__) @@ -197,18 +198,29 @@ def render_clustering_summary(show_taxonomy=False): 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), - ) + + 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_details(idx): + # Local images have no uuid or URL, and the path is the record id: + # show one "File path" label that previews the full path on hover + # and copies it on click (file names alone can be long and repeat). + path = df_plot.iloc[idx]["image_path"] + if not isinstance(path, str): + return None + return {"id": path, "id_short": "File path"} + + render_representative_images( + representatives, + resolve_image=_resolve_local_image, + n_per_cluster=3, + details_fn=_local_details, + ) else: # Precalculated app: show taxonomy tree (works with or without KMeans) if show_taxonomy: diff --git a/shared/services/clustering_service.py b/shared/services/clustering_service.py index 6847b9f..0f9d736 100644 --- a/shared/services/clustering_service.py +++ b/shared/services/clustering_service.py @@ -186,7 +186,7 @@ def generate_clustering_summary( df_plot: pd.DataFrame ) -> Tuple[pd.DataFrame, Dict[int, List[int]]]: """ - Generate clustering summary statistics and representative images. + Generate clustering summary statistics and representative candidates. Args: embeddings: Original embeddings @@ -194,27 +194,27 @@ def generate_clustering_summary( df_plot: Clustering dataframe Returns: - Tuple of (summary dataframe, representatives dict) + Tuple of (summary dataframe, representatives dict). The dict maps + each cluster label to a ranked candidate list (closest-to-centroid + first), oversampled beyond the display count so renderers can skip + entries whose image fails to load — see + `find_cluster_representatives`. Renderers cap how many they show. """ + 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), diff --git a/shared/utils/images.py b/shared/utils/images.py new file mode 100644 index 0000000..0eb6335 --- /dev/null +++ b/shared/utils/images.py @@ -0,0 +1,351 @@ +"""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) may hold image URLs in several of ``IMAGE_URL_COLUMNS``; +``resolve_record_image_urls`` lists the candidates in column order and +``get_record_image`` walks them until one actually loads (a URL can look valid +but be broken/unreachable). Two call paths consume these: + +1. Cluster representatives (bulk, eager). + ``render_cluster_representatives`` warms the cache up front by feeding each + candidate's first URL to ``fetch_images_concurrent`` (thread pool, 8 + workers). Worker threads only run ``download_image_bytes``; the caller + thread decodes each result with ``bytes_to_image`` as it completes and + stores the PIL image (or ``None`` on failure) in ``_IMAGE_CACHE``. The + renderer then resolves via ``get_record_image``; broken URLs fall back to + the record's next URL column, and records with no loadable image fall back + to the cluster's next candidate. + +2. Click preview (single, lazy). + ``render_data_preview`` calls ``get_record_image``, which serves cached + images if present and otherwise does a single synchronous + ``download_image_bytes`` -> ``bytes_to_image`` per URL 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. +- ``ImageTooLarge`` is cached the same way for URLs whose payload exceeds + ``MAX_IMAGE_BYTES``: the fetch is abandoned early (Content-Length when + declared, else while streaming), and renderers surface the sentinel as an + informative placeholder with the original URL instead of a silent miss. + +Each worker thread gets its own ``requests.Session`` (``requests.Session`` is +not thread-safe); every session carries the project User-Agent so data hosts +can identify / allowlist us. +""" + +import concurrent.futures +import threading +import time +from io import BytesIO +from typing import Dict, Iterable, List, Optional, Union + +import requests +from PIL import Image, ImageDraw, ImageFont + +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)" +) + +# Refuse to download image payloads over this size: a malicious or +# misconfigured URL could otherwise stall fetches and spike memory. +MAX_IMAGE_BYTES = 25 * 1024 * 1024 +_STREAM_CHUNK_BYTES = 64 * 1024 + + +class ImageTooLarge: + """Fetch outcome for a URL whose payload exceeded the download cap. + + Cached in ``_IMAGE_CACHE`` like a known miss (so the URL is not retried), + but distinguishable from one: renderers show an informative placeholder + and keep ``url`` clickable so users can still open the image themselves. + ``size_bytes`` is the declared Content-Length, or None when the server + didn't declare one and the cap was hit mid-stream. + """ + + __slots__ = ("url", "size_bytes", "max_bytes") + + def __init__(self, url: str, size_bytes: Optional[int], max_bytes: int) -> None: + self.url = url + self.size_bytes = size_bytes + self.max_bytes = max_bytes + + def describe(self) -> str: + """Human-readable size vs. cap, e.g. '38.5 MB (cap: 25 MB)'.""" + cap_mb = self.max_bytes / (1024 * 1024) + if self.size_bytes is None: + return f"over the {cap_mb:.0f} MB cap" + return f"{self.size_bytes / (1024 * 1024):.1f} MB (cap: {cap_mb:.0f} MB)" + + +# requests.Session is not thread-safe, and fetch_images_concurrent calls +# download_image_bytes from a thread pool — so keep one session per thread. +_thread_local = threading.local() + + +def _get_session() -> requests.Session: + """Lazily build a per-thread requests.Session carrying our User-Agent.""" + session = getattr(_thread_local, "session", None) + if session is None: + session = requests.Session() + session.headers.update({"User-Agent": USER_AGENT}) + _thread_local.session = session + return session + + +def download_image_bytes( + url: str, timeout: int = 5, max_bytes: Optional[int] = None +) -> Union[bytes, ImageTooLarge, None]: + """Fetch raw image bytes via the per-thread session. + + Returns the bytes, an `ImageTooLarge` sentinel when the payload exceeds + `max_bytes` (checked against Content-Length when declared, and enforced + while streaming regardless — the header can be absent or lie), or None + on any other failure. `max_bytes` defaults to the module-level + `MAX_IMAGE_BYTES`, resolved at call time. + + 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 + if max_bytes is None: + max_bytes = MAX_IMAGE_BYTES + try: + with _get_session().get(url, timeout=timeout, stream=True) as resp: + resp.raise_for_status() + if not resp.headers.get('content-type', '').lower().startswith('image/'): + return None + + declared: Optional[int] = None + try: + declared = int(resp.headers['content-length']) + except (KeyError, ValueError): + pass + if declared is not None and declared > max_bytes: + logger.warning( + f"[Image] Skipping oversized payload " + f"({declared / (1024 * 1024):.1f} MB declared): {url[:80]}" + ) + return ImageTooLarge(url, declared, max_bytes) + + chunks: List[bytes] = [] + received = 0 + for chunk in resp.iter_content(chunk_size=_STREAM_CHUNK_BYTES): + received += len(chunk) + if received > max_bytes: + logger.warning( + f"[Image] Aborting oversized download " + f"(>{max_bytes / (1024 * 1024):.0f} MB): {url[:80]}" + ) + return ImageTooLarge(url, declared, max_bytes) + chunks.append(chunk) + return b"".join(chunks) + 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: + img = Image.open(BytesIO(data)) + # Force the actual decode now: Image.open is lazy, so a truncated or + # corrupt payload would otherwise only raise later inside st.image, + # outside this handler, and never be cached as a known miss. + img.load() + return img + 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, an ImageTooLarge sentinel (over-cap URL, +# not retried), or None (known miss). +_IMAGE_CACHE: Dict[str, Union[Image.Image, ImageTooLarge, None]] = {} +_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, Union[Image.Image, ImageTooLarge, None]]: + """Fetch many image URLs concurrently with a thread pool. + + Returns {url: PIL image, ImageTooLarge, or None}. Per-URL results are + cached in a process-level dict so reruns and overlapping clusters don't + refetch. Worker threads only do HTTP (no st.* calls, no cache writes); + PIL decode and caching happen on the caller thread as downloads complete. + """ + 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: + data = fut.result() + if isinstance(data, ImageTooLarge): + _IMAGE_CACHE[u] = data + else: + _IMAGE_CACHE[u] = bytes_to_image(data) + 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 +) -> Union[Image.Image, ImageTooLarge, None]: + """Get a single image from a URL, using the process cache. + + Logs the request; results (including misses and over-cap sentinels) 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() + data = download_image_bytes(url, timeout) + image = data if isinstance(data, ImageTooLarge) else bytes_to_image(data) + elapsed = time.time() - start_time + if isinstance(image, Image.Image): + logger.info(f"[Image] Loaded in {elapsed:.3f}s") + elif image is None: + logger.warning(f"[Image] Failed to load: {url[:50]}...") + + _IMAGE_CACHE[url] = image + _trim_cache() + return image + + +def resolve_record_image_urls(row) -> List[str]: + """Return all candidate HTTP(S) image URLs from a record/row, in + `IMAGE_URL_COLUMNS` order (deduplicated). + + `row` is anything supporting `col in row` membership and `row[col]` + indexing (e.g. a pandas Series or a dict). + """ + urls: List[str] = [] + 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://')) + and val not in urls + ): + urls.append(val) + return urls + + +def resolve_record_image_url(row) -> Optional[str]: + """Return the first candidate HTTP(S) image URL from a record/row, else None.""" + urls = resolve_record_image_urls(row) + return urls[0] if urls else None + + +def get_record_image( + row, timeout: int = 5 +) -> Union[Image.Image, ImageTooLarge, None]: + """Fetch a record's image, falling back across candidate URL columns. + + A URL that looks valid can still be broken/unreachable; walk the record's + candidate URLs (cached fetches) until one actually loads. An over-cap URL + does not stop the walk — a smaller alternate column still wins — but if + nothing loads, the first `ImageTooLarge` seen is returned (rather than + None) so renderers can show an informative placeholder with the URL. + """ + too_large: Optional[ImageTooLarge] = None + for url in resolve_record_image_urls(row): + image = get_image_from_url(url, timeout) + if isinstance(image, ImageTooLarge): + too_large = too_large or image + elif image is not None: + return image + return too_large + + +def too_large_placeholder( + info: ImageTooLarge, size: int = 280, hint: str = "not downloaded" +) -> Image.Image: + """Generate a placeholder tile for an over-cap image. + + Streamlit-free (plain PIL) so any renderer can use it; the clickable URL + is the renderer's job since an image can't carry a link. ``hint`` is the + third line and should tell the user where that link is. + """ + img = Image.new("RGB", (size, size), (233, 236, 239)) + draw = ImageDraw.Draw(img) + try: + font = ImageFont.load_default(size=max(12, size // 18)) + except TypeError: # Pillow < 10.1: no size parameter + font = ImageFont.load_default() + lines = ["Image too large", info.describe(), hint] + line_heights = [] + widths = [] + for line in lines: + x0, y0, x1, y1 = draw.textbbox((0, 0), line, font=font) + widths.append(x1 - x0) + line_heights.append(y1 - y0) + spacing = max(line_heights) // 2 + total_h = sum(line_heights) + spacing * (len(lines) - 1) + y = (size - total_h) // 2 + for line, w, h in zip(lines, widths, line_heights): + draw.text(((size - w) // 2, y), line, fill=(108, 117, 125), font=font) + y += h + spacing + return img diff --git a/shared/utils/representatives.py b/shared/utils/representatives.py new file mode 100644 index 0000000..329d078 --- /dev/null +++ b/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 diff --git a/tests/test_clustering_service.py b/tests/test_clustering_service.py index 9c3fb40..4fa8c34 100644 --- a/tests/test_clustering_service.py +++ b/tests/test_clustering_service.py @@ -38,11 +38,15 @@ def test_counts_sum_to_total(self): assert summary["Count"].sum() == 50 def test_representatives_per_cluster(self): + # Representatives are ranked candidate lists, oversampled (default + # 3 * 4 = 12) so renderers can skip broken images; capped at cluster size. emb, labels, df = self._make_inputs(n_samples=30, n_clusters=3) _, reps = ClusteringService.generate_clustering_summary(emb, labels, df) for cluster_id, indices in reps.items(): cluster_size = (labels == cluster_id).sum() - assert len(indices) <= min(3, cluster_size) + assert len(indices) <= min(12, cluster_size) + # Candidates are valid member indices of their cluster. + assert all(labels[i] == cluster_id for i in indices) def test_single_sample_cluster(self): """Cluster with 1 sample should have variance 0.""" diff --git a/tests/test_representatives.py b/tests/test_representatives.py new file mode 100644 index 0000000..779d838 --- /dev/null +++ b/tests/test_representatives.py @@ -0,0 +1,61 @@ +"""Tests for shared/utils/representatives.py (find_cluster_representatives). + +Covers the pure ranking logic only! Centroid ordering, global-index +correctness, the oversample cap, and label-type handling. The Streamlit +renderer is intentionally not tested here. +""" + +import numpy as np + +from shared.utils.representatives import find_cluster_representatives + + +def test_ranks_closest_to_centroid_first(): + """Members are ordered by ascending distance to the cluster centroid.""" + # Single cluster; centroid_x = (0+3+100)/3 = 34.33, so idx 1 (x=3) is + # closest, then idx 0 (x=0), then idx 2 (x=100). + embeddings = np.array([[0.0, 0.0], [3.0, 0.0], [100.0, 0.0]]) + labels = [0, 0, 0] # All in one cluster + + reps = find_cluster_representatives(embeddings, labels, n_per_cluster=3) + + assert reps[0] == [1, 0, 2] + + +def test_indices_are_global_and_match_cluster(): + """Returned indices are global and only reference members of that cluster.""" + labels = [0, 1, 0, 1, 0] + embeddings = np.random.RandomState(0).rand(5, 4) + + reps = find_cluster_representatives(embeddings, labels, n_per_cluster=2) + + assert set(reps.keys()) == {0, 1} + labels_arr = np.asarray(labels) + for cluster_id, idxs in reps.items(): + assert all(labels_arr[i] == cluster_id for i in idxs) + + +def test_oversample_capped_at_cluster_size(): + """Candidates = min(n_per_cluster * oversample, cluster size).""" + # Cluster 0: 4 members, cluster 1: 10 members. + labels = [0] * 4 + [1] * 10 + embeddings = np.random.RandomState(1).rand(14, 3) + + reps = find_cluster_representatives( + embeddings, labels, n_per_cluster=2, oversample=3 + ) # n_candidates = 6 + + assert len(reps[0]) == 4 # capped at cluster size + assert len(reps[1]) == 6 # capped at n_candidates + + +def test_preserves_label_type(): + """String labels stay string keys; numpy int labels become Python ints.""" + embeddings = np.random.RandomState(2).rand(3, 2) + + str_reps = find_cluster_representatives(embeddings, ["a", "a", "b"]) + assert set(str_reps.keys()) == {"a", "b"} + + int_reps = find_cluster_representatives(embeddings, np.array([0, 0, 1])) + assert set(int_reps.keys()) == {0, 1} + assert all(isinstance(k, int) for k in int_reps.keys())