"""
st.markdown(html, unsafe_allow_html=True)
def _image_to_data_uri(img):
"""Convert PIL Image to JPEG base64 data URI."""
buf = io.BytesIO()
img.convert("RGB").save(buf, format="JPEG", quality=85)
return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
def _render_magnifier_image(img, caption, uid):
"""Render an image with hover/touch magnifier lens using st.html (inline, no iframe)."""
display_img = _prep_for_display(img)
data_uri = _image_to_data_uri(display_img)
html = f"""
{caption}
"""
st.html(html, unsafe_allow_javascript=True)
def _render_images(original_rows_by_model, variant_rows_by_model, selected_models, variant_name, instruction=None):
"""Render task instruction and side-by-side annotated images (the hero content). Full width."""
st.markdown("---")
st.markdown("#### Model Prediction")
if instruction:
st.markdown(
f"
"
f"Task: "
f"{instruction}"
f"
",
unsafe_allow_html=True,
)
first_original = next(iter(original_rows_by_model.values()), None)
first_variant = next(iter(variant_rows_by_model.values()), None)
if "_ds_index" not in st.session_state or "_ds_parquet_paths" not in st.session_state:
with st.spinner("Loading image index..."):
_ensure_dataset_loaded()
img_original = get_image_for_row(first_original) if first_original is not None else None
img_variant = get_image_for_row(first_variant) if first_variant is not None else None
with st.container(key="image_comparison"):
col1, col2 = st.columns(2)
with col1:
if img_original is not None:
annotated = annotate_image_multi_model(img_original, original_rows_by_model, selected_models, draw_predictions=True)
_render_magnifier_image(annotated, "Original", "orig")
else:
st.info("Image not available")
with col2:
if img_variant is not None:
annotated = annotate_image_multi_model(img_variant, variant_rows_by_model, selected_models, draw_predictions=True)
_render_magnifier_image(annotated, f"Perturbed ({variant_name.replace('_', ' ').title()})", "pert")
else:
st.info("Image not available")
@st.fragment
def _render_model_results(original_rows_by_model, variant_rows_by_model, selected_models):
"""Model results using tabs: one tab per model, each with Original vs Perturbed columns.
Status badges are always visible; MSE/coords/prediction details toggled by a shared button.
Uses @st.fragment so toggling details only reruns this section, not the whole page.
"""
if not selected_models:
return
if "show_model_details" not in st.session_state:
st.session_state.show_model_details = False
show = st.session_state.show_model_details
btn_label = "Hide Details" if show else "Show Details"
if st.button(btn_label, key="toggle_model_details"):
st.session_state.show_model_details = not show
st.rerun(scope="fragment")
tab_labels = [_model_label(m) for m in selected_models]
tabs = st.tabs(tab_labels)
for tab, model in zip(tabs, selected_models):
with tab:
orig_row = original_rows_by_model.get(model)
var_row = variant_rows_by_model.get(model)
# Always-visible status row
col_o, col_p = st.columns(2)
with col_o:
st.markdown("**Original**")
if orig_row is not None:
_render_model_status(orig_row)
else:
st.caption("No original data")
with col_p:
st.markdown("**Perturbed**")
if var_row is not None:
_render_model_status(var_row)
else:
st.caption("No perturbed data")
# Collapsible details (shared toggle)
if st.session_state.show_model_details:
det_o, det_p = st.columns(2)
with det_o:
if orig_row is not None:
_render_model_card_details(orig_row)
with det_p:
if var_row is not None:
_render_model_card_details(var_row, orig_row)
def main():
def _lap(label, t0):
print(f"[result_viewer] {label}: {time.perf_counter() - t0:.3f}s", file=sys.stderr, flush=True)
return time.perf_counter()
_t0 = time.perf_counter()
_t = _t0
print(f"\n[result_viewer] --- rerun start ---", file=sys.stderr, flush=True)
# --- Compact header (constrained width) ---
with st.container(key="narrow_header"):
_render_compact_header()
# --- Load data ---
with st.spinner("Loading results..."):
df = load_data(_mtime=_csv_mtime())
_t = _lap("load_data", _t)
if df.empty:
st.error("No data found")
with st.expander("Triage: path resolution", expanded=True):
for path_str, exists in _debug_csv_paths():
if exists is None:
st.text(path_str)
else:
st.text(f"{'✓' if exists else '✗'} {path_str}")
return
# --- Compute state from session defaults (widgets rendered later read from previous rerun) ---
query_types = sorted(df['query_type'].unique().tolist())
use_reasoning_options = sorted(df['use_reasoning'].unique().tolist())
_default_query_type = "relational_query" if "relational_query" in query_types else (query_types[0] if query_types else None)
if "query_type_filter" not in st.session_state and _default_query_type is not None:
st.session_state.query_type_filter = _default_query_type
selected_query_type = st.session_state.get("query_type_filter", _default_query_type)
selected_use_reasoning = st.session_state.get("use_reasoning_filter", use_reasoning_options[0] if use_reasoning_options else None)
if selected_query_type not in query_types:
selected_query_type = query_types[0] if query_types else None
if selected_use_reasoning not in use_reasoning_options:
selected_use_reasoning = use_reasoning_options[0] if use_reasoning_options else None
df_filtered = df[
(df['query_type'] == selected_query_type) &
(df['use_reasoning'] == selected_use_reasoning)
]
all_models = sorted(df_filtered['model'].unique().tolist())
if 'selected_models' not in st.session_state:
st.session_state.selected_models = {model: True for model in all_models}
for model in all_models:
if model not in st.session_state.selected_models:
st.session_state.selected_models[model] = True
_t = _lap("filter extraction + base filter + session state init", _t)
perturbation_variants = PERTURBATION_VARIANTS
if "selected_variant" not in st.session_state:
st.session_state.selected_variant = perturbation_variants[0]
# Determine which models are selected — read widget state directly (key="model_{m}")
# to avoid 1-rerun lag vs the manually maintained selected_models dict
selected_models = [m for m in all_models if st.session_state.get(f"model_{m}", True)]
if 'failure_mode_pills' not in st.session_state:
st.session_state.failure_mode_pills = "Spatial Reasoning Error"
# Initialize navigation state
if 'current_sample_index' not in st.session_state:
st.session_state.current_sample_index = 0
if 'current_task_id' not in st.session_state:
st.session_state.current_task_id = None
if 'current_step_index' not in st.session_state:
st.session_state.current_step_index = None
if 'previous_variant' not in st.session_state:
st.session_state.previous_variant = st.session_state.selected_variant
if 'sample_nav_input' not in st.session_state:
st.session_state.sample_nav_input = st.session_state.current_sample_index + 1
# Build available samples
_active_failure_mode = st.session_state.get("failure_mode_pills", "All")
if _active_failure_mode is None:
_active_failure_mode = "All"
available_samples, available_samples_all, full_list_index_by_sample = _build_available_samples(
df_filtered,
st.session_state.selected_variant,
_active_failure_mode,
)
_t = _lap("_build_available_samples", _t)
if not available_samples:
st.error(
f"No samples found with both original and {st.session_state.selected_variant} perturbation "
f"for failure mode \"{_active_failure_mode}\""
)
return
# Preserve current sample when filters change
current_filter_signature = (
selected_query_type,
selected_use_reasoning,
st.session_state.selected_variant,
_active_failure_mode,
)
filters_changed = st.session_state.get("_filter_signature") != current_filter_signature
if filters_changed:
st.session_state._filter_signature = current_filter_signature
_apply_filter_preservation(
available_samples, available_samples_all, full_list_index_by_sample
)
_t = _lap("filter preservation", _t)
if st.session_state.previous_variant != st.session_state.selected_variant:
st.session_state.previous_variant = st.session_state.selected_variant
st.session_state.num_available_samples = len(available_samples)
if st.session_state.current_sample_index >= len(available_samples):
st.session_state.current_sample_index = 0
current_sample = available_samples[st.session_state.current_sample_index]
st.session_state.current_task_id = current_sample['task_id']
st.session_state.current_step_index = current_sample['step_index']
# Build rows by model for current sample — load ALL models so images always display
sample_data = df_filtered[
(df_filtered['task_id'] == current_sample['task_id']) &
(df_filtered['step_index'] == current_sample['step_index'])
]
_t = _lap("sample data filter", _t)
original_rows_by_model = {}
variant_rows_by_model = {}
for model in all_models:
model_data = sample_data[sample_data['model'] == model]
original_data = model_data[model_data['variant'] == 'original']
variant_data = model_data[model_data['variant'] == st.session_state.selected_variant]
if not original_data.empty:
original_rows_by_model[model] = original_data.iloc[0]
if not variant_data.empty:
variant_rows_by_model[model] = variant_data.iloc[0]
_t = _lap("build rows by model", _t)
# ==========================================
# RENDER: Images (hero content)
# ==========================================
_render_images(
original_rows_by_model,
variant_rows_by_model,
selected_models,
st.session_state.selected_variant,
instruction=current_sample["instruction"],
)
_t = _lap("render images", _t)
# ==========================================
# RENDER: Model display checkboxes with cursor legend
# ==========================================
with st.container(key="narrow_model_display"):
model_display_cols = st.columns(len(all_models)) if all_models else []
for i, model in enumerate(all_models):
with model_display_cols[i]:
style = MODEL_STYLES.get(model, {"color": (180, 180, 180), "label": model})
r, g, b = style["color"]
# Cursor color swatch + checkbox
swatch = (
f''
)
st.markdown(
f"
{swatch}"
f"cursor
",
unsafe_allow_html=True,
)
checked = st.checkbox(
_model_label(model),
value=st.session_state.selected_models.get(model, True),
key=f"model_{model}",
)
st.session_state.selected_models[model] = checked
# Recompute selected_models from current checkbox widget state
selected_models = [m for m in all_models if st.session_state.get(f"model_{m}", True)]
# ==========================================
# RENDER: Sample navigation
# ==========================================
def on_sample_change():
new_val = st.session_state.sample_nav_input
if new_val - 1 != st.session_state.current_sample_index:
st.session_state.current_sample_index = new_val - 1
with st.container(key="narrow_sample_nav"):
position_in_full_list = full_list_index_by_sample.get(
(current_sample['task_id'], current_sample['step_index'])
)
total_in_full_list = len(available_samples_all)
nav_label = f"Sample ({st.session_state.current_sample_index + 1} of {len(available_samples)}"
if position_in_full_list is not None and total_in_full_list != len(available_samples):
nav_label += f" | {position_in_full_list} of {total_in_full_list} total"
nav_label += ")"
st.number_input(
nav_label,
min_value=1,
max_value=len(available_samples),
key="sample_nav_input",
on_change=on_sample_change,
)
# ==========================================
# RENDER: Model results (status always visible, details collapsible per tab)
# ==========================================
with st.container(key="narrow_results"):
st.markdown("#### Model Results")
if selected_models:
_render_model_results(original_rows_by_model, variant_rows_by_model, selected_models)
else:
st.caption("Select a model above to view results.")
_t = _lap("render model results", _t)
# ==========================================
# RENDER: Filters & Search
# ==========================================
with st.container(key="narrow_controls"):
st.markdown("---")
st.markdown("#### GUI-Perturbed Eval Data Filters & Search")
# Failure mode pills (key pre-initialized in session state)
st.pills(
"Failure Mode Example",
FAILURE_MODE_OPTIONS,
key="failure_mode_pills",
help="Filter samples by failure mode to view examples of each type. Select \"All\" to see every sample. Note that some failure samples have not yet been assigned a failure mode.",
)
# Filter dropdowns
f = st.columns(3)
with f[0]:
new_variant = st.selectbox(
"Visual Variant",
perturbation_variants,
index=perturbation_variants.index(st.session_state.selected_variant),
format_func=lambda x: x.replace('_', ' ').title(),
key="perturbation_select_main",
help="Precision: viewport zoom. Style: visual randomization. Text Shrink: font size reduced.",
)
with f[1]:
st.selectbox(
"Instruction Variant",
query_types,
key="query_type_filter",
format_func=lambda x: x.replace('_', ' ').replace('query', 'instruction').replace('Query', 'Instruction').title(),
help="Direct Instruction vs Relational Instruction",
)
with f[2]:
st.selectbox(
"Reasoning",
use_reasoning_options,
key="use_reasoning_filter",
format_func=lambda x: "Yes" if x else "No",
help="Whether chain-of-thought reasoning was used",
)
# Search instructions
def on_search():
query = st.session_state.get("instruction_search", "").strip().lower()
if not query:
return
n = len(available_samples)
if n == 0:
return
start = (st.session_state.current_sample_index + 1) % n
for offset in range(n):
idx = (start + offset) % n
instr = available_samples[idx].get("instruction", "").lower()
if query in instr:
st.session_state.current_sample_index = idx
st.session_state.sample_nav_input = idx + 1
return
st.text_input(
"Search instructions",
key="instruction_search",
on_change=on_search,
placeholder="Type to search task instructions...",
)
_t = _lap("control panel widgets", _t)
# Handle variant change
if new_variant != st.session_state.selected_variant:
st.session_state.selected_variant = new_variant
st.rerun()
_lap("main() total", _t0)
if __name__ == "__main__":
try:
main()
except Exception as e:
import traceback
st.error("Dashboard failed to load")
st.exception(e)
st.code(traceback.format_exc(), language="text")