Spaces:
Sleeping
Sleeping
| """ | |
| Interactive Dash dashboard β faithful port from Android DashboardFragment. | |
| Features: | |
| - 3D scatter plot with clusters + Catmull-Rom spline path | |
| - Flow field particle animation (Turbo speed coloring, 20 FPS) | |
| - Ball probe following MDN flow field with trail | |
| - Bidirectional sliders: manual probe OR auto-track ball position | |
| - Mark control points + explain via LLM | |
| - Toggle: flow animation, ball flow, show points, show path | |
| - Dark theme (#1E1E1E matching Android glClearColor) | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import Optional, List | |
| import numpy as np | |
| try: | |
| import plotly.graph_objects as go | |
| from dash import Dash, html, dcc, Input, Output, State, callback_context, no_update | |
| except ImportError as _e: | |
| raise ImportError( | |
| "Visualization dependencies not installed. " | |
| "Install them with: pip install plotly dash\n" | |
| "Or install the full package: pip install tracescope" | |
| ) from _e | |
| from tracescope.models.analysis import AnalysisResult | |
| from tracescope.visualization.scatter3d import ( | |
| plot_clusters_3d, catmull_rom_spline, CLUSTER_COLORS, _apply_dark_theme, | |
| _score_to_plotly_colors, | |
| ) | |
| from tracescope.visualization.flow_field import FlowFieldSystem, turbo_colormap | |
| from tracescope.visualization.probe import probe_point | |
| logger = logging.getLogger(__name__) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Module-level state (not serialized across callbacks) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| _flow_system: Optional[FlowFieldSystem] = None | |
| _result: Optional[AnalysisResult] = None | |
| _explainer = None | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # DARK THEME STYLES | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| _BG = "#1E1E1E" | |
| _CARD_BG = "#2A2A2A" | |
| _BORDER = "#3A3A3A" | |
| _TEXT = "#E0E0E0" | |
| _ACCENT = "#FF6B35" | |
| _CARD_STYLE = { | |
| "backgroundColor": _CARD_BG, | |
| "border": f"1px solid {_BORDER}", | |
| "borderRadius": "8px", | |
| "padding": "12px", | |
| "marginBottom": "10px", | |
| } | |
| def _slider_to_raw(slider_val: float, axis_idx: int) -> float: | |
| """Convert slider [0..1000] to raw 3D coordinate.""" | |
| pts = _result.projected_3d | |
| axis_min = pts[:, axis_idx].min() | |
| axis_max = pts[:, axis_idx].max() | |
| return axis_min + (slider_val / 1000.0) * (axis_max - axis_min) | |
| def _raw_to_slider(raw_val: float, axis_idx: int) -> int: | |
| """Convert raw 3D coordinate to slider [0..1000].""" | |
| pts = _result.projected_3d | |
| axis_min = pts[:, axis_idx].min() | |
| axis_max = pts[:, axis_idx].max() | |
| span = axis_max - axis_min | |
| if span == 0: | |
| return 500 | |
| pct = (raw_val - axis_min) / span | |
| return int(np.clip(pct * 1000, 0, 1000)) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # LAYOUT | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _build_layout(result: AnalysisResult) -> html.Div: | |
| """Build the full Dash layout with all controls.""" | |
| axis_labels = result.axis_info.labels | |
| has_flow = result.velocity_grid is not None | |
| # Cluster legend items | |
| cluster_items = [] | |
| for c in range(result.clusters.n_clusters): | |
| color = CLUSTER_COLORS[c % len(CLUSTER_COLORS)] | |
| label = ( | |
| result.cluster_labels[c] | |
| if c < len(result.cluster_labels) | |
| else f"Cluster {c}" | |
| ) | |
| count = sum(1 for l in result.clusters.labels if l == c) | |
| cluster_items.append( | |
| html.Div([ | |
| html.Span("β ", style={"color": color, "fontSize": "16px"}), | |
| html.Strong(label, style={"color": _TEXT}), | |
| html.Span(f" ({count})", style={"color": "#888"}), | |
| ], style={"marginBottom": "3px"}) | |
| ) | |
| # Flow toggle options | |
| flow_options = [] | |
| if has_flow: | |
| flow_options = [ | |
| {"label": " Enable Flow Animation", "value": "flow_animation"}, | |
| {"label": " Ball Flow (follow MDN)", "value": "ball_flow"}, | |
| ] | |
| display_options = [ | |
| {"label": " Show Data Points", "value": "show_points"}, | |
| {"label": " Show Path", "value": "show_path"}, | |
| ] | |
| return html.Div([ | |
| # Hidden state stores | |
| dcc.Store(id="flow-state", data={ | |
| "ball_position": [0, 0, 0], | |
| "ball_trail": [], | |
| "control_points": [], | |
| "frame_index": 0, | |
| }), | |
| dcc.Store(id="control-points-store", data=[]), | |
| # Animation timer (50ms = 20 FPS) | |
| dcc.Interval( | |
| id="flow-interval", | |
| interval=50, | |
| n_intervals=0, | |
| disabled=True, | |
| ), | |
| # Header | |
| html.H2("TraceScope Dashboard", style={ | |
| "textAlign": "center", "color": _TEXT, "margin": "0 0 15px 0", | |
| "fontWeight": "300", "letterSpacing": "2px", | |
| }), | |
| # Main flex container | |
| html.Div([ | |
| # LEFT: 3D scatter plot | |
| html.Div([ | |
| dcc.Graph( | |
| id="scatter3d", | |
| figure=_build_base_figure(result), | |
| style={"height": "650px"}, | |
| config={"scrollZoom": True}, | |
| ), | |
| ], style={"flex": "3", "minWidth": "500px"}), | |
| # RIGHT: Controls panel | |
| html.Div([ | |
| # Flow controls | |
| html.Div([ | |
| html.H4("β‘ Flow Controls" if has_flow else "β‘ Flow (not available)", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| dcc.Checklist( | |
| id="flow-toggles", | |
| options=flow_options, | |
| value=[], | |
| style={"color": _TEXT, "fontSize": "13px"}, | |
| inputStyle={"marginRight": "6px"}, | |
| labelStyle={"display": "block", "marginBottom": "4px"}, | |
| ) if has_flow else html.Div( | |
| "Train with train_flow=True to enable", | |
| style={"color": "#666", "fontSize": "12px", "fontStyle": "italic"}, | |
| ), | |
| ], style=_CARD_STYLE), | |
| # Display toggles | |
| html.Div([ | |
| html.H4("π Display", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| dcc.Checklist( | |
| id="display-toggles", | |
| options=display_options, | |
| value=["show_points", "show_path"], | |
| style={"color": _TEXT, "fontSize": "13px"}, | |
| inputStyle={"marginRight": "6px"}, | |
| labelStyle={"display": "block", "marginBottom": "4px"}, | |
| ), | |
| ], style=_CARD_STYLE), | |
| # Score-based coloring (only if scores exist) | |
| *([html.Div([ | |
| html.H4("π¨ Color by Score", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| dcc.Dropdown( | |
| id="score-color-dropdown", | |
| options=[{"label": "(clusters)", "value": ""}] + [ | |
| {"label": ch, "value": ch} | |
| for ch in result.score_channels | |
| ], | |
| value="", | |
| clearable=False, | |
| style={"backgroundColor": "#333", "color": _TEXT, "fontSize": "13px"}, | |
| ), | |
| ], style=_CARD_STYLE)] if result.score_channels else [ | |
| html.Div(dcc.Dropdown(id="score-color-dropdown", options=[], value="", | |
| style={"display": "none"})) | |
| ]), | |
| # Probe sliders | |
| html.Div([ | |
| html.H4("π― Probe", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| # X slider | |
| html.Div(id="slider-x-label", | |
| children=f"{axis_labels[0]} 50%", | |
| style={"color": _TEXT, "fontSize": "12px", "marginBottom": "2px"}), | |
| dcc.Slider( | |
| id="slider-x", min=0, max=1000, value=500, step=1, | |
| marks={}, updatemode="drag", | |
| tooltip={"placement": "bottom", "always_visible": False}, | |
| ), | |
| # Y slider | |
| html.Div(id="slider-y-label", | |
| children=f"{axis_labels[1]} 50%", | |
| style={"color": _TEXT, "fontSize": "12px", "marginBottom": "2px"}), | |
| dcc.Slider( | |
| id="slider-y", min=0, max=1000, value=500, step=1, | |
| marks={}, updatemode="drag", | |
| tooltip={"placement": "bottom", "always_visible": False}, | |
| ), | |
| # Z slider | |
| html.Div(id="slider-z-label", | |
| children=f"{axis_labels[2]} 50%", | |
| style={"color": _TEXT, "fontSize": "12px", "marginBottom": "2px"}), | |
| dcc.Slider( | |
| id="slider-z", min=0, max=1000, value=500, step=1, | |
| marks={}, updatemode="drag", | |
| tooltip={"placement": "bottom", "always_visible": False}, | |
| ), | |
| # Buttons | |
| html.Div([ | |
| html.Button("Mark Point", id="btn-mark", n_clicks=0, | |
| style={"marginRight": "6px", "padding": "4px 12px", | |
| "backgroundColor": _ACCENT, "color": "white", | |
| "border": "none", "borderRadius": "4px", "cursor": "pointer"}), | |
| html.Button("Clear", id="btn-clear", n_clicks=0, | |
| style={"marginRight": "6px", "padding": "4px 12px", | |
| "backgroundColor": "#555", "color": "white", | |
| "border": "none", "borderRadius": "4px", "cursor": "pointer"}), | |
| html.Button("Explain", id="btn-explain", n_clicks=0, | |
| style={"padding": "4px 12px", | |
| "backgroundColor": "#4A90D9", "color": "white", | |
| "border": "none", "borderRadius": "4px", "cursor": "pointer"}), | |
| ], style={"marginTop": "8px"}), | |
| ], style=_CARD_STYLE), | |
| # Clusters | |
| html.Div([ | |
| html.H4("π΅ Clusters", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| html.Div(cluster_items), | |
| ], style=_CARD_STYLE), | |
| # Info panel | |
| html.Div([ | |
| html.H4("π Info", | |
| style={"color": _ACCENT, "margin": "0 0 8px 0", "fontSize": "14px"}), | |
| html.Div(id="info-panel", style={ | |
| "whiteSpace": "pre-wrap", "color": _TEXT, | |
| "fontSize": "12px", "maxHeight": "200px", "overflowY": "auto", | |
| }), | |
| ], style=_CARD_STYLE), | |
| ], style={ | |
| "flex": "1", "minWidth": "280px", "maxWidth": "350px", | |
| "padding": "0 0 0 10px", "overflowY": "auto", | |
| "maxHeight": "680px", | |
| }), | |
| ], style={"display": "flex", "gap": "10px"}), | |
| ], style={ | |
| "padding": "15px", "fontFamily": "'Segoe UI', sans-serif", | |
| "backgroundColor": _BG, "minHeight": "100vh", | |
| }) | |
| def _build_base_figure(result: AnalysisResult) -> go.Figure: | |
| """Build the initial figure with clusters + path.""" | |
| return plot_clusters_3d(result, show_path=True, use_spline=True) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # CALLBACKS | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _register_callbacks(app: Dash, result: AnalysisResult): | |
| """Register all Dash callbacks.""" | |
| # ββ Callback 1: Flow interval enable/disable βββββββββββββββββ | |
| def toggle_flow_interval(flow_toggles): | |
| flow_toggles = flow_toggles or [] | |
| # Enable interval when flow animation is checked | |
| return "flow_animation" not in flow_toggles | |
| # ββ Callback 2: Flow state update (animation tick) βββββββββββ | |
| def update_flow_state(n_intervals, state, flow_toggles): | |
| if _flow_system is None: | |
| return no_update | |
| flow_toggles = flow_toggles or [] | |
| # Step flow particles | |
| pos, colors, alphas, speeds = _flow_system.step() | |
| # Convert colors to Plotly rgba strings (subsample for performance) | |
| max_render = 4000 # Render at most 4000 particles | |
| step_size = max(1, len(pos) // max_render) | |
| render_pos = pos[::step_size] | |
| render_colors = colors[::step_size] | |
| render_alphas = alphas[::step_size] | |
| particle_colors = [ | |
| f"rgba({int(c[0]*255)},{int(c[1]*255)},{int(c[2]*255)},{a:.2f})" | |
| for c, a in zip(render_colors, render_alphas) | |
| ] | |
| state["particle_x"] = render_pos[:, 0].tolist() | |
| state["particle_y"] = render_pos[:, 1].tolist() | |
| state["particle_z"] = render_pos[:, 2].tolist() | |
| state["particle_colors"] = particle_colors | |
| state["frame_index"] = n_intervals | |
| # Ball flow | |
| if "ball_flow" in flow_toggles: | |
| _flow_system.advance_ball() | |
| bp = _flow_system.ball_pos.tolist() | |
| state["ball_position"] = bp | |
| # Trail: last 100 positions | |
| trail = state.get("ball_trail", []) | |
| trail.append(bp) | |
| if len(trail) > 100: | |
| trail = trail[-100:] | |
| state["ball_trail"] = trail | |
| else: | |
| # Ball stays where sliders put it | |
| state["ball_trail"] = [] | |
| return state | |
| # ββ Callback 3: Main figure render βββββββββββββββββββββββββββ | |
| def render_figure(flow_data, sx, sy, sz, click_data, | |
| display_toggles, control_points, score_channel, flow_toggles): | |
| display_toggles = display_toggles or [] | |
| flow_toggles = flow_toggles or [] | |
| score_channel = score_channel or "" | |
| fig = go.Figure() | |
| pts = result.projected_3d | |
| labels = result.clusters.labels | |
| axis_labels = result.axis_info.labels | |
| # ββ Scatter points ββββββββββββββββββββββββββββββββββ | |
| if "show_points" in display_toggles: | |
| if score_channel and score_channel in result.score_channels: | |
| # Score-based coloring | |
| entry_scores = result.get_entry_scores(score_channel) | |
| path_score_map = result.get_path_scores(score_channel) | |
| final_scores = [] | |
| for i, e in enumerate(result.session.entries): | |
| s = entry_scores[i] | |
| if s is None and e.path_id is not None: | |
| s = path_score_map.get(e.path_id) | |
| final_scores.append(s) | |
| point_colors = _score_to_plotly_colors(final_scores) | |
| hover = [ | |
| f"[{result.session.entries[i].role}] " | |
| f"{result.session.entries[i].text[:80]}... " | |
| f"({score_channel}: {final_scores[i]:.2f})" | |
| if final_scores[i] is not None | |
| else f"[{result.session.entries[i].role}] " | |
| f"{result.session.entries[i].text[:80]}..." | |
| for i in range(len(pts)) | |
| ] | |
| fig.add_trace(go.Scatter3d( | |
| x=pts[:, 0], y=pts[:, 1], z=pts[:, 2], | |
| mode="markers", | |
| marker=dict(size=6, color=point_colors, opacity=0.8), | |
| name=f"Score: {score_channel}", text=hover, hoverinfo="text", | |
| )) | |
| else: | |
| # Cluster-based coloring (default) | |
| for c in range(result.clusters.n_clusters): | |
| mask = [i for i, l in enumerate(labels) if l == c] | |
| if not mask: | |
| continue | |
| cluster_pts = pts[mask] | |
| cluster_texts = [result.session.entries[i].text[:80] for i in mask] | |
| cluster_roles = [result.session.entries[i].role for i in mask] | |
| hover = [f"[{r}] {t}..." for r, t in zip(cluster_roles, cluster_texts)] | |
| color = CLUSTER_COLORS[c % len(CLUSTER_COLORS)] | |
| label = ( | |
| result.cluster_labels[c] | |
| if c < len(result.cluster_labels) else f"Cluster {c}" | |
| ) | |
| fig.add_trace(go.Scatter3d( | |
| x=cluster_pts[:, 0], y=cluster_pts[:, 1], z=cluster_pts[:, 2], | |
| mode="markers", | |
| marker=dict(size=6, color=color, opacity=0.8), | |
| name=label, text=hover, hoverinfo="text", | |
| )) | |
| # ββ Catmull-Rom spline path βββββββββββββββββββββββββ | |
| if "show_path" in display_toggles and len(pts) >= 2: | |
| spline_pts = catmull_rom_spline(pts, samples_per_segment=20) | |
| fig.add_trace(go.Scatter3d( | |
| x=spline_pts[:, 0], y=spline_pts[:, 1], z=spline_pts[:, 2], | |
| mode="lines", | |
| line=dict(color="rgba(255,255,255,0.5)", width=4), | |
| name="Path", showlegend=True, hoverinfo="skip", | |
| )) | |
| # ββ Flow particles ββββββββββββββββββββββββββββββββββ | |
| if ("flow_animation" in flow_toggles and flow_data | |
| and flow_data.get("particle_x")): | |
| fig.add_trace(go.Scatter3d( | |
| x=flow_data["particle_x"], | |
| y=flow_data["particle_y"], | |
| z=flow_data["particle_z"], | |
| mode="markers", | |
| marker=dict(size=2, color=flow_data["particle_colors"], opacity=1.0), | |
| name="Flow particles", showlegend=False, hoverinfo="skip", | |
| )) | |
| # ββ Ball marker βββββββββββββββββββββββββββββββββββββ | |
| if flow_data and flow_data.get("ball_position"): | |
| bp = flow_data["ball_position"] | |
| fig.add_trace(go.Scatter3d( | |
| x=[bp[0]], y=[bp[1]], z=[bp[2]], | |
| mode="markers", | |
| marker=dict(size=12, color="rgb(255,100,50)", opacity=0.9, | |
| line=dict(width=1, color="white")), | |
| name="Ball", showlegend=False, | |
| )) | |
| # ββ Ball trail ββββββββββββββββββββββββββββββββββββββ | |
| if flow_data and flow_data.get("ball_trail") and len(flow_data["ball_trail"]) >= 2: | |
| trail = np.array(flow_data["ball_trail"]) | |
| fig.add_trace(go.Scatter3d( | |
| x=trail[:, 0], y=trail[:, 1], z=trail[:, 2], | |
| mode="lines", | |
| line=dict(color="yellow", width=3), | |
| name="Trail", showlegend=False, hoverinfo="skip", | |
| )) | |
| # ββ Probe marker (from sliders, when not in ball flow) ββ | |
| if "ball_flow" not in flow_toggles: | |
| px = _slider_to_raw(sx, 0) | |
| py = _slider_to_raw(sy, 1) | |
| pz = _slider_to_raw(sz, 2) | |
| fig.add_trace(go.Scatter3d( | |
| x=[px], y=[py], z=[pz], | |
| mode="markers", | |
| marker=dict(size=10, color="yellow", symbol="diamond", | |
| line=dict(width=1, color="white")), | |
| name="Probe", showlegend=False, | |
| )) | |
| # ββ Control points ββββββββββββββββββββββββββββββββββ | |
| if control_points and len(control_points) > 0: | |
| cp = np.array(control_points) | |
| fig.add_trace(go.Scatter3d( | |
| x=cp[:, 0], y=cp[:, 1], z=cp[:, 2], | |
| mode="markers", | |
| marker=dict(size=8, color="white", symbol="cross", | |
| line=dict(width=1, color=_ACCENT)), | |
| name="Marked", showlegend=False, | |
| )) | |
| # Connecting line between control points | |
| if len(cp) >= 2: | |
| spline = catmull_rom_spline(cp, samples_per_segment=10) | |
| fig.add_trace(go.Scatter3d( | |
| x=spline[:, 0], y=spline[:, 1], z=spline[:, 2], | |
| mode="lines", | |
| line=dict(color=_ACCENT, width=3, dash="dash"), | |
| name="Marked path", showlegend=False, hoverinfo="skip", | |
| )) | |
| # ββ Apply dark theme ββββββββββββββββββββββββββββββββ | |
| _apply_dark_theme(fig, axis_labels) | |
| fig.update_layout( | |
| showlegend=True, | |
| margin=dict(l=0, r=0, t=10, b=0), | |
| uirevision="constant", # Preserve camera angle across updates | |
| ) | |
| return fig | |
| # ββ Callback 4: Slider sync (bidirectional) + info panel βββββ | |
| def update_info_and_sliders(sx, sy, sz, click_data, | |
| explain_clicks, mark_clicks, clear_clicks, | |
| flow_data, flow_toggles, control_points): | |
| ctx = callback_context | |
| if not ctx.triggered: | |
| return (no_update,) * 8 | |
| triggered = ctx.triggered[0]["prop_id"] | |
| flow_toggles = flow_toggles or [] | |
| control_points = control_points or [] | |
| axis_labels = result.axis_info.labels | |
| # Default outputs (no update) | |
| info_text = no_update | |
| out_sx, out_sy, out_sz = no_update, no_update, no_update | |
| lbl_x = no_update | |
| lbl_y = no_update | |
| lbl_z = no_update | |
| out_cp = no_update | |
| # ββ Ball flow state β update sliders to track ball ββββ | |
| if "flow-state" in triggered and "ball_flow" in flow_toggles: | |
| if flow_data and flow_data.get("ball_position"): | |
| bp = flow_data["ball_position"] | |
| out_sx = _raw_to_slider(bp[0], 0) | |
| out_sy = _raw_to_slider(bp[1], 1) | |
| out_sz = _raw_to_slider(bp[2], 2) | |
| lbl_x = f"{axis_labels[0]} {round(out_sx / 10)}%" | |
| lbl_y = f"{axis_labels[1]} {round(out_sy / 10)}%" | |
| lbl_z = f"{axis_labels[2]} {round(out_sz / 10)}%" | |
| # Compute info for ball position | |
| info_text = _build_probe_info(bp[0], bp[1], bp[2]) | |
| return info_text, out_sx, out_sy, out_sz, lbl_x, lbl_y, lbl_z, out_cp | |
| # ββ Click on a scatter point ββββββββββββββββββββββββββ | |
| if "scatter3d.clickData" in triggered and click_data: | |
| point = click_data["points"][0] | |
| px_c, py_c, pz_c = point["x"], point["y"], point["z"] | |
| # Find nearest entry | |
| dists = np.linalg.norm( | |
| result.projected_3d - np.array([px_c, py_c, pz_c]), axis=1 | |
| ) | |
| idx = int(np.argmin(dists)) | |
| entry = result.session.entries[idx] | |
| cluster_id = result.clusters.labels[idx] | |
| cluster_label = ( | |
| result.cluster_labels[cluster_id] | |
| if cluster_id < len(result.cluster_labels) | |
| else f"Cluster {cluster_id}" | |
| ) | |
| info_text = ( | |
| f"Message #{entry.step_index} [{entry.role}]\n" | |
| f"Cluster: {cluster_label}\n\n" | |
| f"{entry.text}" | |
| ) | |
| # Jump sliders to clicked point | |
| out_sx = _raw_to_slider(px_c, 0) | |
| out_sy = _raw_to_slider(py_c, 1) | |
| out_sz = _raw_to_slider(pz_c, 2) | |
| lbl_x = f"{axis_labels[0]} {round(out_sx / 10)}%" | |
| lbl_y = f"{axis_labels[1]} {round(out_sy / 10)}%" | |
| lbl_z = f"{axis_labels[2]} {round(out_sz / 10)}%" | |
| # Set ball position if flow system exists | |
| if _flow_system is not None: | |
| _flow_system.set_ball_position(px_c, py_c, pz_c) | |
| return info_text, out_sx, out_sy, out_sz, lbl_x, lbl_y, lbl_z, out_cp | |
| # ββ Slider dragged (manual probe) βββββββββββββββββββββ | |
| if any(s in triggered for s in ["slider-x", "slider-y", "slider-z"]): | |
| px = _slider_to_raw(sx, 0) | |
| py = _slider_to_raw(sy, 1) | |
| pz = _slider_to_raw(sz, 2) | |
| # Update ball position if not flowing | |
| if "ball_flow" not in flow_toggles and _flow_system is not None: | |
| _flow_system.set_ball_position(px, py, pz) | |
| info_text = _build_probe_info(px, py, pz) | |
| lbl_x = f"{axis_labels[0]} {round(sx / 10)}%" | |
| lbl_y = f"{axis_labels[1]} {round(sy / 10)}%" | |
| lbl_z = f"{axis_labels[2]} {round(sz / 10)}%" | |
| return info_text, no_update, no_update, no_update, lbl_x, lbl_y, lbl_z, out_cp | |
| # ββ Mark Point button βββββββββββββββββββββββββββββββββ | |
| if "btn-mark" in triggered: | |
| px = _slider_to_raw(sx, 0) | |
| py = _slider_to_raw(sy, 1) | |
| pz = _slider_to_raw(sz, 2) | |
| control_points.append([float(px), float(py), float(pz)]) | |
| out_cp = control_points | |
| info_text = f"Marked point #{len(control_points)} at ({px:.2f}, {py:.2f}, {pz:.2f})" | |
| return info_text, no_update, no_update, no_update, no_update, no_update, no_update, out_cp | |
| # ββ Clear Points button βββββββββββββββββββββββββββββββ | |
| if "btn-clear" in triggered: | |
| out_cp = [] | |
| info_text = "Control points cleared." | |
| return info_text, no_update, no_update, no_update, no_update, no_update, no_update, out_cp | |
| # ββ Explain button ββββββββββββββββββββββββββββββββββββ | |
| if "btn-explain" in triggered: | |
| px = _slider_to_raw(sx, 0) | |
| py = _slider_to_raw(sy, 1) | |
| pz = _slider_to_raw(sz, 2) | |
| if _explainer and len(control_points) > 0: | |
| # Multi-point explanation | |
| try: | |
| info_text = _build_multi_explain(control_points) | |
| except Exception as e: | |
| info_text = f"Explain error: {e}" | |
| elif _explainer: | |
| # Single-point explanation | |
| try: | |
| from tracescope.visualization.probe import probe_with_explanation | |
| probe_info = probe_with_explanation(result, _explainer, px, py, pz) | |
| info_text = ( | |
| f"LLM Explanation:\n{probe_info['explanation']}\n\n" | |
| f"Nearest messages:\n" | |
| ) | |
| for item in probe_info["nearest_texts"][:3]: | |
| info_text += f" [{item['role']}] {item['text'][:100]}...\n" | |
| except Exception as e: | |
| info_text = f"Explain error: {e}" | |
| else: | |
| info_text = "No explainer configured. Pass explainer= to launch_dashboard()." | |
| return info_text, no_update, no_update, no_update, no_update, no_update, no_update, out_cp | |
| return (no_update,) * 8 | |
| def _build_probe_info(x: float, y: float, z: float) -> str: | |
| """Build probe info string with axis %, cluster distances, nearest texts.""" | |
| probe_info = probe_point(_result, x, y, z) | |
| lines = ["Probe Position:"] | |
| for axis_name, pct in probe_info["axis_percentages"].items(): | |
| lines.append(f" {axis_name}: {pct:.1f}%") | |
| lines.append("\nCluster Distances:") | |
| for name, pct in probe_info["cluster_distances"].items(): | |
| lines.append(f" {name}: {pct:.1f}% closeness") | |
| lines.append("\nNearest Messages:") | |
| for item in probe_info["nearest_texts"][:3]: | |
| lines.append(f" [{item['role']}] {item['text'][:80]}...") | |
| return "\n".join(lines) | |
| def _build_multi_explain(control_points: list) -> str: | |
| """Build multi-point path explanation via LLM.""" | |
| lines = ["Control Points Path:\n"] | |
| for i, cp in enumerate(control_points): | |
| probe_info = probe_point(_result, cp[0], cp[1], cp[2]) | |
| pcts = probe_info["axis_percentages"] | |
| dists = probe_info["cluster_distances"] | |
| pct_str = ", ".join(f"{k}: {v:.0f}%" for k, v in pcts.items()) | |
| dist_str = ", ".join(f"{k}: {v:.0f}%" for k, v in dists.items()) | |
| lines.append(f" Point {i}: {pct_str}") | |
| lines.append(f" Clusters: {dist_str}") | |
| if _explainer: | |
| try: | |
| # Use multi-point explanation | |
| axis_labels = list(probe_point(_result, *control_points[0])["axis_percentages"].keys()) | |
| all_pcts = [] | |
| all_dists = [] | |
| for cp in control_points: | |
| pi = probe_point(_result, cp[0], cp[1], cp[2]) | |
| all_pcts.append(list(pi["axis_percentages"].values())) | |
| all_dists.append(pi["cluster_distances"]) | |
| control_points_for_llm = [] | |
| for pcts_vals, dists_dict in zip(all_pcts, all_dists): | |
| control_points_for_llm.append({ | |
| "axis_pcts": [int(v) for v in pcts_vals], | |
| "cluster_distances": [(k, int(v)) for k, v in dists_dict.items()], | |
| }) | |
| explanation = _explainer.explain_probe_multi( | |
| axis_labels=axis_labels, | |
| control_points=control_points_for_llm, | |
| ) | |
| lines.append(f"\nLLM Analysis:\n{explanation}") | |
| except Exception as e: | |
| lines.append(f"\nExplain error: {e}") | |
| return "\n".join(lines) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # PUBLIC API | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def launch_dashboard( | |
| result: AnalysisResult, | |
| port: int = 8050, | |
| debug: bool = False, | |
| explainer=None, | |
| ) -> None: | |
| """Launch the interactive Dash dashboard. | |
| Args: | |
| result: AnalysisResult from the pipeline. | |
| port: Port to serve on. | |
| debug: Enable Dash debug mode. | |
| explainer: Optional SemanticExplainer for probe explanations. | |
| """ | |
| global _flow_system, _result, _explainer | |
| _result = result | |
| _explainer = explainer | |
| # Initialize FlowFieldSystem if velocity grid is available | |
| if result.velocity_grid is not None: | |
| logger.info("Initializing FlowFieldSystem with %dΒ³ velocity grid", | |
| result.velocity_grid.shape[0]) | |
| _flow_system = FlowFieldSystem( | |
| result.velocity_grid, | |
| result.axis_min, | |
| result.axis_max, | |
| particle_grid=20, # 8000 particles for web performance | |
| ) | |
| # Set ball to center | |
| center = (result.axis_min + result.axis_max) / 2 | |
| _flow_system.set_ball_position(*center) | |
| logger.info("FlowFieldSystem ready: %d particles", _flow_system.particle_count) | |
| else: | |
| _flow_system = None | |
| logger.info("No velocity grid available β flow animation disabled") | |
| app = Dash(__name__) | |
| app.layout = _build_layout(result) | |
| _register_callbacks(app, result) | |
| print(f"\n{'='*60}") | |
| print(f" TraceScope Dashboard") | |
| print(f" http://localhost:{port}") | |
| print(f" Entries: {len(result.session)}") | |
| print(f" Clusters: {result.clusters.n_clusters}") | |
| print(f" Flow: {'enabled' if _flow_system else 'disabled'}") | |
| print(f"{'='*60}\n") | |
| app.run(port=port, debug=debug) | |