Add dataset index, async filters, fix slider bounds, add spinner

CLAUDE.md

@@ -36,10 +36,14 @@ python app.py
 
 ## Key Patterns
 
+- **Full dataset filtering**: All ~100K rows' metadata is streamed at startup (metadata-only, no spectrum arrays). Once loaded, filters operate on the entire dataset. Metadata is published incrementally every 10K rows.
+- **On-demand spectrum fetching**: When filters select spectra outside the initial pool, their spectrum data is fetched in a background thread and plots generated progressively.
+- **Sequential dataset-order indexing**: All indices (initial pool, full metadata, annotations) use consistent dataset-order indices (0..N-1). No shuffle is used to avoid index space collisions.
+- **Serialized background tasks**: Full metadata loads first (undisturbed ~8 min), then remaining initial pool spectra (~7 min). This avoids bandwidth competition between concurrent HF streams.
 - **Annotation state** is thread-safe (`threading.Lock`) and saved to `state/annotations.json` on every annotation, synced to HF Hub periodically
 - **Spectrum selection** picks random un-annotated spectra; prefers those with cached plots for instant display
 - **Annotation categories** are configurable in `dataset_config.yaml` (Galaxy, Star, QSO, Interesting, Bad Data)
-- **Spectrum filters** (redshift range, S/N, ZWARN) are configurable in `dataset_config.yaml`
+- **Spectrum filters** (redshift range, EBV, ZWARN) are configurable in `dataset_config.yaml`
 - **Spectrum plots** are generated with matplotlib and cached as PNGs on disk (`cache/plots/`)
 - **Background pre-fetch** thread downloads and renders remaining spectra after the app starts (first 4 are synchronous)
 - **Callbacks** use Dash `ctx.triggered_id` to determine which annotation button was clicked

src/annotations.py

@@ -218,7 +218,7 @@ def select_next_spectrum(filters: dict | None = None) -> int:
     Prefers un-annotated spectra; favours those whose plots are already cached.
     If filters is provided, restricts to matching spectra.
     """
-    from src.spectrum_data import get_filtered_pool
+    from src.spectrum_data import get_filtered_pool, is_plot_cached
 
     with _lock:
         if _state is None:
@@ -238,7 +238,11 @@ def select_next_spectrum(filters: dict | None = None) -> int:
     if not unannotated:
         unannotated = pool  # all done — cycle again
 
-    return random.choice(unannotated)
+    # Prefer spectra whose plots are already generated (avoids broken images
+    # while the background thread is still downloading data)
+    cached = [idx for idx in unannotated if is_plot_cached(idx)]
+    source = cached if cached else unannotated
+    return random.choice(source)
 
 
 def get_stats() -> dict:

src/callbacks.py

@@ -1,5 +1,7 @@
 """Dash callbacks for DESI Spectra Zoo."""
 
+import random
+import threading
 import uuid
 import logging
 
@@ -9,7 +11,11 @@ from dash.exceptions import PreventUpdate
 
 from src import annotations
 from src.hf_logging import log_annotation_event
-from src.spectrum_data import ensure_plot, get_filtered_pool, find_by_object_id
+from src.spectrum_data import (
+    ensure_plot, get_filtered_pool, find_by_object_id,
+    is_plot_cached, fetch_spectra_for_indices, cancel_fetch,
+    is_full_metadata_ready, get_full_metadata_count,
+)
 from src.components import (
     create_spectrum_display,
     create_annotation_buttons,
@@ -26,13 +32,34 @@ _CAT_BY_BTN_ID = {f"annotation-btn-{cat['id']}": cat["id"] for cat in ANNOTATION
 
 
 def _parse_filters(z_range, ebv_max, exclude_zwarn):
-    """Build a filters dict from UI control values."""
+    """Build a filters dict from UI control values.
+
+    Filters are only applied when the user has actually moved a slider
+    away from its default (extreme) position. This avoids excluding
+    spectra due to rounding of slider bounds.
+    """
+    from src.spectrum_data import get_metadata_ranges
+    import math
+
     filters = {}
+    ranges = get_metadata_ranges()
+
     if z_range and len(z_range) == 2:
-        filters["z_min"] = z_range[0]
-        filters["z_max"] = z_range[1]
+        real_z = ranges.get("Z", (0, 5))
+        slider_z_min = math.floor(real_z[0] * 100) / 100
+        slider_z_max = math.ceil(real_z[1] * 100) / 100
+        # Only apply if user moved the slider away from its default bound
+        if z_range[0] > slider_z_min:
+            filters["z_min"] = z_range[0]
+        if z_range[1] < slider_z_max:
+            filters["z_max"] = z_range[1]
+
     if ebv_max is not None:
-        filters["ebv_max"] = ebv_max
+        real_ebv = ranges.get("EBV", (0, 1))
+        slider_ebv_max = math.ceil(real_ebv[1] * 1000) / 1000
+        if ebv_max < slider_ebv_max:
+            filters["ebv_max"] = ebv_max
+
     if exclude_zwarn:
         filters["exclude_zwarn"] = True
     return filters
@@ -62,6 +89,11 @@ def register_callbacks(app):
         stats = annotations.get_stats()
         recent = annotations.get_recent()
         pool_count = len(get_filtered_pool({}))
+        total = get_full_metadata_count()
+        if is_full_metadata_ready() and total:
+            count_label = f"{pool_count} / {total} spectra"
+        else:
+            count_label = f"{pool_count} spectra"
         return (
             create_spectrum_display(idx),
             create_annotation_buttons(),
@@ -70,7 +102,7 @@ def register_callbacks(app):
             session_id,
             stats,
             create_recent_rows(recent),
-            f"{pool_count} spectra",
+            count_label,
         )
 
     # ---------- Annotation click ----------
@@ -135,6 +167,30 @@ def register_callbacks(app):
         stats = annotations.get_stats()
         recent = annotations.get_recent()
 
+        if not is_plot_cached(next_idx):
+            # Spectrum data not yet downloaded — trigger background fetch
+            threading.Thread(
+                target=fetch_spectra_for_indices,
+                args=([next_idx],),
+                daemon=True,
+            ).start()
+            return (
+                html.Div(
+                    "Fetching next spectrum... Click Skip to retry.",
+                    style={
+                        "textAlign": "center",
+                        "padding": "40px",
+                        "color": "rgba(255,213,79,0.6)",
+                        "fontFamily": "'Outfit', sans-serif",
+                    },
+                ),
+                next_idx,
+                ann_count + 1,
+                create_stats_panel(stats),
+                stats,
+                create_recent_rows(recent),
+            )
+
         return (
             create_spectrum_display(next_idx),
             next_idx,
@@ -162,6 +218,24 @@ def register_callbacks(app):
             raise PreventUpdate
         next_idx = annotations.select_next_spectrum(filters=active_filters or {})
         ensure_plot(next_idx)
+        if not is_plot_cached(next_idx):
+            threading.Thread(
+                target=fetch_spectra_for_indices,
+                args=([next_idx],),
+                daemon=True,
+            ).start()
+            return (
+                html.Div(
+                    "Fetching next spectrum... Click Skip to retry.",
+                    style={
+                        "textAlign": "center",
+                        "padding": "40px",
+                        "color": "rgba(255,213,79,0.6)",
+                        "fontFamily": "'Outfit', sans-serif",
+                    },
+                ),
+                next_idx,
+            )
         return create_spectrum_display(next_idx), next_idx
 
     # ---------- Apply filters ----------
@@ -188,10 +262,17 @@ def register_callbacks(app):
         matched = get_filtered_pool(filters)
         count = len(matched)
 
+        full_ready = is_full_metadata_ready()
+        total = get_full_metadata_count()
+        if full_ready and total:
+            source_label = f" / {total}"
+        else:
+            source_label = ""
+
         if count == 0:
             return (
                 filters,
-                "0 spectra — try relaxing filters",
+                f"0{source_label} spectra — try relaxing filters",
                 html.Div(
                     "No spectra match these filters. Try adjusting the ranges.",
                     style={
@@ -204,11 +285,43 @@ def register_callbacks(app):
                 dash.no_update,
             )
 
+        # Pick a random sample of up to 500 to fetch spectra for
+        MAX_FETCH = 500
+        to_fetch = random.sample(matched, min(MAX_FETCH, len(matched)))
+
+        # Start background download for spectra not yet cached
+        cancel_fetch()  # cancel any in-progress fetch
+        threading.Thread(
+            target=fetch_spectra_for_indices,
+            args=(to_fetch,),
+            daemon=True,
+        ).start()
+
+        # Show a spectrum that's already cached, or the first one to fetch
         next_idx = annotations.select_next_spectrum(filters=filters)
         ensure_plot(next_idx)
+
+        count_label = f"{count}{source_label} spectra"
+        if not is_plot_cached(next_idx):
+            return (
+                filters,
+                count_label + " (fetching...)",
+                html.Div(
+                    f"Fetching spectrum data... ({count} match). "
+                    "Try again in a few seconds.",
+                    style={
+                        "textAlign": "center",
+                        "padding": "40px",
+                        "color": "rgba(255,213,79,0.6)",
+                        "fontFamily": "'Outfit', sans-serif",
+                    },
+                ),
+                dash.no_update,
+            )
+
         return (
             filters,
-            f"{count} spectra",
+            count_label,
             create_spectrum_display(next_idx),
             next_idx,
         )
@@ -229,6 +342,25 @@ def register_callbacks(app):
         idx = find_by_object_id(object_id_value)
         if idx is None:
             raise PreventUpdate
+        # If not cached, fetch on-demand in background then show loading msg
+        if not is_plot_cached(idx):
+            threading.Thread(
+                target=fetch_spectra_for_indices,
+                args=([idx],),
+                daemon=True,
+            ).start()
+            return (
+                html.Div(
+                    f"Fetching spectrum {object_id_value}... Click Skip or retry in a few seconds.",
+                    style={
+                        "textAlign": "center",
+                        "padding": "40px",
+                        "color": "rgba(255,213,79,0.6)",
+                        "fontFamily": "'Outfit', sans-serif",
+                    },
+                ),
+                idx,
+            )
         ensure_plot(idx)
         return create_spectrum_display(idx), idx
 

src/components.py

@@ -567,10 +567,11 @@ def create_filter_panel():
     z_range = ranges.get("Z", (0, 5))
     ebv_range = ranges.get("EBV", (0, 1))
 
-    # Round for nicer slider bounds
-    z_min_val = max(0, round(z_range[0], 2))
-    z_max_val = round(z_range[1] + 0.01, 2)
-    ebv_max_val = round(ebv_range[1] + 0.01, 3)
+    import math
+    # Use floor/ceil so slider bounds always cover ALL data values
+    z_min_val = math.floor(z_range[0] * 100) / 100   # e.g. -0.003 → -0.01
+    z_max_val = math.ceil(z_range[1] * 100) / 100     # e.g. 6.437 → 6.44
+    ebv_max_val = math.ceil(ebv_range[1] * 1000) / 1000  # e.g. 0.876 → 0.876
 
     return html.Div([
         # Collapsible header
@@ -747,8 +748,34 @@ def create_layout():
             ),
 
             # Main arena: spectrum display + annotation buttons
-            html.Div(id="spectrum-container",
-                     style={"position": "relative", "zIndex": "10"}),
+            html.Div(
+                html.Div(
+                    [
+                        dbc.Spinner(
+                            color="warning",
+                            spinner_style={"width": "3rem", "height": "3rem"},
+                        ),
+                        html.Div(
+                            "Loading first spectrum...",
+                            style={
+                                "marginTop": "16px",
+                                "fontFamily": "'Outfit', sans-serif",
+                                "fontSize": "0.85rem",
+                                "color": "rgba(255,255,255,0.4)",
+                            },
+                        ),
+                    ],
+                    style={
+                        "display": "flex",
+                        "flexDirection": "column",
+                        "alignItems": "center",
+                        "justifyContent": "center",
+                        "padding": "60px 0",
+                    },
+                ),
+                id="spectrum-container",
+                style={"position": "relative", "zIndex": "10"},
+            ),
 
             html.Div(id="annotation-buttons-container",
                      style={

src/spectrum_data.py

@@ -1,8 +1,11 @@
 """DESI spectrum data: streaming loader, plot generation, and caching.
 
-Two-pass streaming:
-- Pass 1: metadata-only streaming (fast, all spectra)
-- Pass 2: full spectrum streaming for plot generation (background)
+Architecture:
+- Full metadata index: ALL rows' metadata loaded at startup (no spectrum arrays).
+  Enables filtering across the entire dataset.
+- On-demand spectrum fetch: when filters select new spectra, their spectrum data
+  is downloaded in background and plots generated progressively.
+- Initial quick pool: a small batch of spectra with full data for immediate display.
 """
 
 from __future__ import annotations
@@ -45,12 +48,26 @@ LINES = {
 _BG_COLOR = "#0a0a1e"
 _SHUFFLE_BUFFER = 200
 
+_META_COLUMNS = [
+    ID_COLUMN, "Z", "ZERR", "ZWARN", "EBV",
+    "FLUX_G", "FLUX_R", "FLUX_Z",
+    "FIBERFLUX_G", "FIBERFLUX_R", "FIBERFLUX_Z",
+]
+
 # ---------------------------------------------------------------------------
 # State
 # ---------------------------------------------------------------------------
 _lock = threading.Lock()
-_metadata_cache: dict[int, dict] = {}
-_row_cache: dict[int, dict] = {}  # full row data for on-demand plot generation
+_metadata_cache: dict[int, dict] = {}  # active pool metadata (initial + fetched)
+_row_cache: dict[int, dict] = {}       # full row data for plot generation
+
+# Full metadata index: ALL dataset rows (dataset-order index → metadata)
+_full_metadata: dict[int, dict] = {}
+_full_metadata_ready = threading.Event()
+_full_metadata_total: int = 0  # total rows in dataset
+
+# Background fetch control
+_fetch_cancel = threading.Event()
 
 _plot_dir = Path(PLOT_CACHE_DIR)
 
@@ -59,11 +76,10 @@ _plot_dir = Path(PLOT_CACHE_DIR)
 # Streaming dataset factory
 # ---------------------------------------------------------------------------
 
-def _make_dataset(seed: int, pool_size: int, with_spectrum: bool = True):
-    """Return a shuffled, length-limited streaming dataset iterator.
+def _make_streaming_dataset(with_spectrum: bool = True):
+    """Return a sequential streaming dataset iterator over ALL rows.
 
-    with_spectrum=False skips the spectrum column entirely at the Parquet level.
-    Both modes use the same seed+buffer so row i is always the same spectrum.
+    with_spectrum=False skips the spectrum column at the Parquet level.
     """
     from datasets import load_dataset
 
@@ -78,17 +94,28 @@ def _make_dataset(seed: int, pool_size: int, with_spectrum: bool = True):
     features = getattr(ds, "features", None) or {}
 
     if not with_spectrum:
-        meta_cols = [c for c in [
-            ID_COLUMN, "Z", "ZERR", "ZWARN", "EBV",
-            "FLUX_G", "FLUX_R", "FLUX_Z",
-            "FIBERFLUX_G", "FIBERFLUX_R", "FIBERFLUX_Z",
-        ] if c in features]
+        meta_cols = [c for c in _META_COLUMNS if c in features]
         if meta_cols:
             ds = ds.select_columns(meta_cols)
 
-    ds = ds.shuffle(seed=seed, buffer_size=_SHUFFLE_BUFFER)
-    ds = ds.take(pool_size)
-    return iter(ds)
+    return ds
+
+
+def _row_to_meta(row: dict) -> dict:
+    """Extract metadata fields from a dataset row."""
+    return {
+        ID_COLUMN: row.get(ID_COLUMN),
+        "Z": row.get("Z", 0),
+        "ZERR": row.get("ZERR", 0),
+        "ZWARN": row.get("ZWARN", False),
+        "EBV": row.get("EBV", 0),
+        "FLUX_G": row.get("FLUX_G", 0),
+        "FLUX_R": row.get("FLUX_R", 0),
+        "FLUX_Z": row.get("FLUX_Z", 0),
+        "FIBERFLUX_G": row.get("FIBERFLUX_G", 0),
+        "FIBERFLUX_R": row.get("FIBERFLUX_R", 0),
+        "FIBERFLUX_Z": row.get("FIBERFLUX_Z", 0),
+    }
 
 
 # ---------------------------------------------------------------------------
@@ -101,10 +128,18 @@ def register_metadata(metadata_map: dict[int, dict]):
         _metadata_cache.update(metadata_map)
 
 
+def _get_meta(row_index: int) -> dict | None:
+    """Get metadata for a row, preferring full index, falling back to active cache."""
+    with _lock:
+        meta = _full_metadata.get(row_index)
+        if meta is None:
+            meta = _metadata_cache.get(row_index)
+    return meta
+
+
 def get_display_name(row_index: int) -> str:
     """Return a human-readable name for a spectrum."""
-    with _lock:
-        meta = _metadata_cache.get(row_index)
+    meta = _get_meta(row_index)
     if meta is not None:
         id_val = meta.get(ID_COLUMN)
         if id_val is not None:
@@ -114,8 +149,7 @@ def get_display_name(row_index: int) -> str:
 
 def get_display_info(row_index: int) -> dict:
     """Return full metadata for display (object_id, Z, ZERR, ZWARN, EBV)."""
-    with _lock:
-        meta = _metadata_cache.get(row_index)
+    meta = _get_meta(row_index)
     if meta:
         return {
             "object_id": str(meta.get(ID_COLUMN, f"Spectrum #{row_index}")),
@@ -129,26 +163,19 @@ def get_display_info(row_index: int) -> dict:
 
 
 def find_by_object_id(search_id: str) -> int | None:
-    """Find the pool index for a given object ID string. Returns None if not found."""
+    """Find the dataset index for a given object ID string."""
     search_id = search_id.strip()
     if not search_id:
         return None
+    # Search full index first, then active cache
     with _lock:
-        for idx, meta in _metadata_cache.items():
+        source = _full_metadata if _full_metadata else _metadata_cache
+        for idx, meta in source.items():
             if str(meta.get(ID_COLUMN, "")) == search_id:
                 return idx
     return None
 
 
-def get_row_index_by_id(display_name: str) -> int | None:
-    """Reverse-lookup: return the row index for a given display name, or None."""
-    with _lock:
-        for row_index, meta in _metadata_cache.items():
-            if meta is not None and str(meta.get(ID_COLUMN)) == display_name:
-                return row_index
-    return None
-
-
 # ---------------------------------------------------------------------------
 # Plot cache (PNG files on disk)
 # ---------------------------------------------------------------------------
@@ -341,51 +368,52 @@ def sample_pool_streaming(
     seed: int | None = None,
     prefetch_plots: int = 4,
 ) -> tuple[list[int], dict[int, dict], int]:
-    """Build the spectrum pool, caching a small batch of plots before returning.
+    """Build the initial quick pool and start full metadata loading in background.
 
-    Two-pass streaming approach:
-    - Pass 1: metadata only (fast, all spectra) — no spectrum arrays downloaded
-    - Pass 2: full rows for plot generation — first N sync, rest background
+    Uses **sequential dataset-order** indexing (no shuffle) so that the initial
+    pool indices are consistent with the full metadata index loaded later.
+
+    1. Streams metadata for first `pool_size` rows (sequential, fast).
+    2. Streams spectrum data for the first `prefetch_plots` synchronously.
+    3. Continues spectrum data for the rest in background.
+    4. Starts a background thread to load ALL metadata (full dataset index).
 
     Returns:
-        ids: sequential ints 0..N-1
+        ids: dataset-order ints 0..pool_size-1
         metadata_map: {id -> metadata dict}
-        seed: seed used
+        seed: kept for API compat
     """
     if seed is None:
         seed = random.randint(0, 2**32 - 1)
 
     _plot_dir.mkdir(parents=True, exist_ok=True)
 
-    # Pass 1: metadata only — fast, no spectrum data downloaded
-    logger.info("Streaming metadata for %d spectra (seed=%d)...", pool_size, seed)
+    # --- Phase 1: metadata for first pool_size rows (sequential, fast) ---
+    logger.info("Streaming metadata for initial pool of %d spectra (sequential)...",
+                pool_size)
+
+    ds_meta = _make_streaming_dataset(with_spectrum=False)
+
     ids: list[int] = []
     metadata_map: dict[int, dict] = {}
-
-    for i, row in enumerate(_make_dataset(seed, pool_size, with_spectrum=False)):
-        metadata_map[i] = {
-            ID_COLUMN: row.get(ID_COLUMN),
-            "Z": row.get("Z", 0),
-            "ZERR": row.get("ZERR", 0),
-            "ZWARN": row.get("ZWARN", False),
-            "EBV": row.get("EBV", 0),
-            "FLUX_G": row.get("FLUX_G", 0),
-            "FLUX_R": row.get("FLUX_R", 0),
-            "FLUX_Z": row.get("FLUX_Z", 0),
-            "FIBERFLUX_G": row.get("FIBERFLUX_G", 0),
-            "FIBERFLUX_R": row.get("FIBERFLUX_R", 0),
-            "FIBERFLUX_Z": row.get("FIBERFLUX_Z", 0),
-        }
+    for i, row in enumerate(ds_meta):
+        if i >= pool_size:
+            break
+        metadata_map[i] = _row_to_meta(row)
         ids.append(i)
 
-    logger.info("All %d spectrum IDs ready", len(ids))
+    logger.info("All %d initial spectrum IDs ready", len(ids))
 
-    # Pass 2: full rows for plot generation — same seed => same row order
-    spec_it = _make_dataset(seed, pool_size, with_spectrum=True)
+    # --- Phase 2: spectrum data for first N plots (synchronous) ---
+    ds_full = _make_streaming_dataset(with_spectrum=True)
+    spec_it = iter(ds_full)
     sync_count = min(prefetch_plots, pool_size)
 
     for i in range(sync_count):
-        row = next(spec_it)
+        try:
+            row = next(spec_it)
+        except StopIteration:
+            break
         with _lock:
             _row_cache[i] = row
         if not is_plot_cached(i):
@@ -396,43 +424,164 @@ def sample_pool_streaming(
                 logger.warning("Plot generation failed for spectrum %d: %s", i, e)
         logger.info("Pre-fetched plot %d/%d", i + 1, sync_count)
 
-    logger.info("%d plots cached — app ready, %d remaining in background",
+    logger.info("%d plots cached — app ready, %d remaining after metadata load",
                 sync_count, pool_size - sync_count)
 
-    if sync_count < pool_size:
-        def _bg():
-            for i in range(sync_count, pool_size):
+    # --- Background: full metadata FIRST, then remaining initial pool spectra ---
+    # Serializing avoids bandwidth competition between two concurrent streams.
+    _pool_size_for_bg = pool_size
+    _sync_count_for_bg = sync_count
+
+    def _bg_metadata_then_spectra():
+        # 1) Load full metadata index (metadata-only, fast)
+        _load_full_metadata_index()
+
+        # 2) Load remaining initial pool spectra (with spectrum data)
+        logger.info("Now loading remaining initial pool spectra %d-%d...",
+                     _sync_count_for_bg, _pool_size_for_bg - 1)
+        ds = _make_streaming_dataset(with_spectrum=True)
+        for i, row in enumerate(ds):
+            if _fetch_cancel.is_set():
+                logger.info("Initial pool loading cancelled by fetch_cancel")
+                break
+            if i < _sync_count_for_bg:
+                continue
+            if i >= _pool_size_for_bg:
+                break
+            with _lock:
+                _row_cache[i] = row
+            if not is_plot_cached(i):
                 try:
-                    row = next(spec_it)
-                    with _lock:
-                        _row_cache[i] = row
-                    if not is_plot_cached(i):
-                        png_bytes = _generate_plot(i, row)
-                        (_plot_dir / f"{i}.png").write_bytes(png_bytes)
-                except StopIteration:
-                    break
+                    png_bytes = _generate_plot(i, row)
+                    (_plot_dir / f"{i}.png").write_bytes(png_bytes)
                 except Exception as e:
                     logger.warning("Background plot error at spectrum %d: %s", i, e)
-            logger.info("Background plot generation complete")
+        logger.info("Initial pool plot generation complete")
 
-        threading.Thread(target=_bg, daemon=True).start()
+    threading.Thread(target=_bg_metadata_then_spectra, daemon=True).start()
 
     return ids, metadata_map, seed
 
 
+# ---------------------------------------------------------------------------
+# Full metadata index (all dataset rows)
+# ---------------------------------------------------------------------------
+
+def _load_full_metadata_index():
+    """Stream ALL dataset rows (metadata-only) to build the full index.
+
+    Runs in background at startup. Updates _full_metadata incrementally
+    every 10K rows so filters can work on partial data during loading.
+    """
+    global _full_metadata, _full_metadata_total
+
+    logger.info("Loading full metadata index (all rows, metadata-only)...")
+
+    try:
+        ds = _make_streaming_dataset(with_spectrum=False)
+        temp: dict[int, dict] = {}
+        for i, row in enumerate(ds):
+            temp[i] = _row_to_meta(row)
+            if (i + 1) % 10000 == 0:
+                logger.info("Metadata indexed: %d rows...", i + 1)
+                # Publish partial index so filters work on available data
+                with _lock:
+                    _full_metadata = dict(temp)
+                    _full_metadata_total = len(temp)
+
+        with _lock:
+            _full_metadata = temp
+            _full_metadata_total = len(temp)
+        _full_metadata_ready.set()
+        logger.info("Full metadata index ready: %d rows", len(temp))
+
+    except Exception as e:
+        logger.error("Failed to load full metadata index: %s", e)
+        _full_metadata_ready.set()  # unblock waiters even on failure
+
+
+def is_full_metadata_ready() -> bool:
+    """Check if the full metadata index has finished loading."""
+    return _full_metadata_ready.is_set()
+
+
+def get_full_metadata_count() -> int:
+    """Return total number of rows in the full metadata index (0 if not ready)."""
+    with _lock:
+        return _full_metadata_total
+
+
+# ---------------------------------------------------------------------------
+# On-demand spectrum fetching
+# ---------------------------------------------------------------------------
+
+def fetch_spectra_for_indices(indices: list[int]) -> None:
+    """Download spectrum data + generate plots for given dataset-order indices.
+
+    Streams through the dataset sequentially, collecting only the needed rows.
+    Called in a background thread when filters select new spectra.
+    """
+    # Only fetch indices that aren't already cached
+    needed = [i for i in indices if not is_plot_cached(i) and i not in _row_cache]
+    if not needed:
+        return
+
+    _fetch_cancel.clear()
+    target_set = set(needed)
+    max_idx = max(needed)
+
+    logger.info("Fetching spectrum data for %d spectra (up to idx %d)...",
+                len(target_set), max_idx)
+
+    try:
+        ds = _make_streaming_dataset(with_spectrum=True)
+        for i, row in enumerate(ds):
+            if _fetch_cancel.is_set():
+                logger.info("Spectrum fetch cancelled")
+                return
+            if i in target_set:
+                with _lock:
+                    _row_cache[i] = row
+                    _metadata_cache[i] = _row_to_meta(row)
+                if not is_plot_cached(i):
+                    try:
+                        png_bytes = _generate_plot(i, row)
+                        _plot_dir.mkdir(parents=True, exist_ok=True)
+                        (_plot_dir / f"{i}.png").write_bytes(png_bytes)
+                    except Exception as e:
+                        logger.warning("Plot error for spectrum %d: %s", i, e)
+                target_set.discard(i)
+                if not target_set:
+                    break
+            if i > max_idx:
+                break
+
+        logger.info("Spectrum fetch complete (%d remaining)", len(target_set))
+
+    except Exception as e:
+        logger.error("Spectrum fetch failed: %s", e)
+
+
+def cancel_fetch():
+    """Signal the background fetch thread to stop."""
+    _fetch_cancel.set()
+
+
 def get_pool() -> list[int]:
-    """Return the current pool indices."""
+    """Return the current pool indices (all indices with metadata)."""
     with _lock:
+        if _full_metadata:
+            return sorted(_full_metadata.keys())
         return sorted(_metadata_cache.keys())
 
 
 def get_metadata_ranges() -> dict:
-    """Compute min/max for numeric metadata across the pool.
+    """Compute min/max for numeric metadata.
 
-    Used by the UI to set slider ranges.
+    Uses full index if available, else falls back to active pool.
     """
     with _lock:
-        metas = list(_metadata_cache.values())
+        metas = list(_full_metadata.values()) if _full_metadata else list(_metadata_cache.values())
     if not metas:
         return {}
 
@@ -452,16 +601,12 @@ def get_metadata_ranges() -> dict:
 
 
 def get_filtered_pool(filters: dict) -> list[int]:
-    """Return pool indices matching the given filters.
+    """Return dataset indices matching the given filters.
 
-    filters is a dict with optional keys:
-      z_min, z_max: redshift range
-      exclude_zwarn: bool
-      ebv_max: max EBV
-      snr_min: minimum S/N (approximated from FLUX_R if available)
+    Uses full metadata index if available, otherwise the active pool.
     """
     with _lock:
-        metas = dict(_metadata_cache)
+        metas = dict(_full_metadata) if _full_metadata else dict(_metadata_cache)
 
     z_min = filters.get("z_min")
     z_max = filters.get("z_max")

state/annotations.json

@@ -1,516 +0,0 @@
-{
-  "dataset_id": "MultimodalUniverse/desi",
-  "pool": [
-    0,
-    1,
-    2,
-    3,
-    4,
-    5,
-    6,
-    7,
-    8,
-    9,
-    10,
-    11,
-    12,
-    13,
-    14,
-    15,
-    16,
-    17,
-    18,
-    19,
-    20,
-    21,
-    22,
-    23,
-    24,
-    25,
-    26,
-    27,
-    28,
-    29,
-    30,
-    31,
-    32,
-    33,
-    34,
-    35,
-    36,
-    37,
-    38,
-    39,
-    40,
-    41,
-    42,
-    43,
-    44,
-    45,
-    46,
-    47,
-    48,
-    49,
-    50,
-    51,
-    52,
-    53,
-    54,
-    55,
-    56,
-    57,
-    58,
-    59,
-    60,
-    61,
-    62,
-    63,
-    64,
-    65,
-    66,
-    67,
-    68,
-    69,
-    70,
-    71,
-    72,
-    73,
-    74,
-    75,
-    76,
-    77,
-    78,
-    79,
-    80,
-    81,
-    82,
-    83,
-    84,
-    85,
-    86,
-    87,
-    88,
-    89,
-    90,
-    91,
-    92,
-    93,
-    94,
-    95,
-    96,
-    97,
-    98,
-    99,
-    100,
-    101,
-    102,
-    103,
-    104,
-    105,
-    106,
-    107,
-    108,
-    109,
-    110,
-    111,
-    112,
-    113,
-    114,
-    115,
-    116,
-    117,
-    118,
-    119,
-    120,
-    121,
-    122,
-    123,
-    124,
-    125,
-    126,
-    127,
-    128,
-    129,
-    130,
-    131,
-    132,
-    133,
-    134,
-    135,
-    136,
-    137,
-    138,
-    139,
-    140,
-    141,
-    142,
-    143,
-    144,
-    145,
-    146,
-    147,
-    148,
-    149,
-    150,
-    151,
-    152,
-    153,
-    154,
-    155,
-    156,
-    157,
-    158,
-    159,
-    160,
-    161,
-    162,
-    163,
-    164,
-    165,
-    166,
-    167,
-    168,
-    169,
-    170,
-    171,
-    172,
-    173,
-    174,
-    175,
-    176,
-    177,
-    178,
-    179,
-    180,
-    181,
-    182,
-    183,
-    184,
-    185,
-    186,
-    187,
-    188,
-    189,
-    190,
-    191,
-    192,
-    193,
-    194,
-    195,
-    196,
-    197,
-    198,
-    199,
-    200,
-    201,
-    202,
-    203,
-    204,
-    205,
-    206,
-    207,
-    208,
-    209,
-    210,
-    211,
-    212,
-    213,
-    214,
-    215,
-    216,
-    217,
-    218,
-    219,
-    220,
-    221,
-    222,
-    223,
-    224,
-    225,
-    226,
-    227,
-    228,
-    229,
-    230,
-    231,
-    232,
-    233,
-    234,
-    235,
-    236,
-    237,
-    238,
-    239,
-    240,
-    241,
-    242,
-    243,
-    244,
-    245,
-    246,
-    247,
-    248,
-    249,
-    250,
-    251,
-    252,
-    253,
-    254,
-    255,
-    256,
-    257,
-    258,
-    259,
-    260,
-    261,
-    262,
-    263,
-    264,
-    265,
-    266,
-    267,
-    268,
-    269,
-    270,
-    271,
-    272,
-    273,
-    274,
-    275,
-    276,
-    277,
-    278,
-    279,
-    280,
-    281,
-    282,
-    283,
-    284,
-    285,
-    286,
-    287,
-    288,
-    289,
-    290,
-    291,
-    292,
-    293,
-    294,
-    295,
-    296,
-    297,
-    298,
-    299,
-    300,
-    301,
-    302,
-    303,
-    304,
-    305,
-    306,
-    307,
-    308,
-    309,
-    310,
-    311,
-    312,
-    313,
-    314,
-    315,
-    316,
-    317,
-    318,
-    319,
-    320,
-    321,
-    322,
-    323,
-    324,
-    325,
-    326,
-    327,
-    328,
-    329,
-    330,
-    331,
-    332,
-    333,
-    334,
-    335,
-    336,
-    337,
-    338,
-    339,
-    340,
-    341,
-    342,
-    343,
-    344,
-    345,
-    346,
-    347,
-    348,
-    349,
-    350,
-    351,
-    352,
-    353,
-    354,
-    355,
-    356,
-    357,
-    358,
-    359,
-    360,
-    361,
-    362,
-    363,
-    364,
-    365,
-    366,
-    367,
-    368,
-    369,
-    370,
-    371,
-    372,
-    373,
-    374,
-    375,
-    376,
-    377,
-    378,
-    379,
-    380,
-    381,
-    382,
-    383,
-    384,
-    385,
-    386,
-    387,
-    388,
-    389,
-    390,
-    391,
-    392,
-    393,
-    394,
-    395,
-    396,
-    397,
-    398,
-    399,
-    400,
-    401,
-    402,
-    403,
-    404,
-    405,
-    406,
-    407,
-    408,
-    409,
-    410,
-    411,
-    412,
-    413,
-    414,
-    415,
-    416,
-    417,
-    418,
-    419,
-    420,
-    421,
-    422,
-    423,
-    424,
-    425,
-    426,
-    427,
-    428,
-    429,
-    430,
-    431,
-    432,
-    433,
-    434,
-    435,
-    436,
-    437,
-    438,
-    439,
-    440,
-    441,
-    442,
-    443,
-    444,
-    445,
-    446,
-    447,
-    448,
-    449,
-    450,
-    451,
-    452,
-    453,
-    454,
-    455,
-    456,
-    457,
-    458,
-    459,
-    460,
-    461,
-    462,
-    463,
-    464,
-    465,
-    466,
-    467,
-    468,
-    469,
-    470,
-    471,
-    472,
-    473,
-    474,
-    475,
-    476,
-    477,
-    478,
-    479,
-    480,
-    481,
-    482,
-    483,
-    484,
-    485,
-    486,
-    487,
-    488,
-    489,
-    490,
-    491,
-    492,
-    493,
-    494,
-    495,
-    496,
-    497,
-    498,
-    499
-  ],
-  "annotations": [
-    {
-      "spectrum_index": 435,
-      "spectrum_name": "39627752067827082",
-      "category": "galaxy",
-      "session_id": "77d21f0bf33941ce87f901e82471c840",
-      "comment": "",
-      "timestamp": "2026-03-31T10:01:54.189866"
-    }
-  ],
-  "total": 1
-}