Spaces:
Sleeping
Sleeping
| """Export all flow analysis data (clusters, axes, attractors, paths, explanations) | |
| to JSON and prompt-ready text files. | |
| This does NOT run any LLM calls — it only dumps data that the user has already | |
| generated (cached attractor explanations, saved paths, etc.). | |
| Usage: | |
| from tracescope.export import export_flow_data | |
| export_flow_data("cache/prm_demo_rbf", output_dir="docs/export") | |
| """ | |
| import json | |
| import os | |
| import numpy as np | |
| from tracescope.models.analysis import AnalysisResult | |
| from tracescope.visualization.probe import probe_point | |
| def export_flow_data(cache_path: str, output_dir: str = None, | |
| score_channels: list = None, | |
| sensitivity: float = 0.7) -> dict: | |
| """Export all cached flow data to a dict (and optionally to files). | |
| Args: | |
| cache_path: Base cache path (e.g. "cache/prm_demo_rbf"). | |
| output_dir: If provided, writes JSON + prompt .md files here. | |
| score_channels: Score channels to include in attractor data. | |
| sensitivity: Attractor detection sensitivity (0.0-1.0). | |
| Returns: | |
| dict with all exported data. | |
| """ | |
| result = AnalysisResult.load_result(cache_path) | |
| axis_labels = result.axis_info.labels | |
| cluster_labels = result.cluster_labels | |
| pts = result.projected_3d | |
| mins, maxs = pts.min(0), pts.max(0) | |
| ranges = maxs - mins | |
| ranges[ranges == 0] = 1.0 | |
| max_dist = float(np.linalg.norm(maxs - mins)) | |
| # Cluster info | |
| clusters = [] | |
| for c in range(result.clusters.n_clusters): | |
| labels_arr = np.array(result.clusters.labels) | |
| mask = labels_arr == c | |
| c_pts = pts[mask] | |
| name = cluster_labels[c] if c < len(cluster_labels) else f"Cluster {c}" | |
| clusters.append({ | |
| 'index': c, | |
| 'name': name, | |
| 'size': int(mask.sum()), | |
| 'centroid': c_pts.mean(0).tolist() if len(c_pts) > 0 else [0, 0, 0], | |
| }) | |
| # Attractors | |
| from tracescope.visualization.flow_field import FlowFieldSystem | |
| flow = FlowFieldSystem(result.velocity_grid, result.axis_min, result.axis_max) | |
| # Build score grids for each requested channel | |
| attractor_score_data = {} | |
| if score_channels: | |
| for ch in score_channels: | |
| raw_scores = result.get_entry_scores(ch) | |
| path_scores = result.get_path_scores(ch) | |
| vals = [] | |
| for i, e in enumerate(result.session.entries): | |
| s = raw_scores[i] | |
| if s is None and e.path_id is not None: | |
| s = path_scores.get(e.path_id) | |
| vals.append(s if s is not None else 0.5) | |
| attractor_score_data[ch] = np.array(vals, dtype=np.float32) | |
| attractors_list = [] | |
| raw_attractors = flow.find_attractors(sensitivity=sensitivity, score_grid=None) | |
| centroids_3d = result.cluster_centroids_3d | |
| for i, att in enumerate(raw_attractors): | |
| pos = att['position'] | |
| axis_pcts = [int(np.clip((pos[j] - mins[j]) / ranges[j] * 100, 0, 100)) | |
| for j in range(3)] | |
| cdists = [] | |
| if centroids_3d is not None: | |
| for ci in range(len(centroids_3d)): | |
| d = float(np.linalg.norm(pos - centroids_3d[ci])) | |
| closeness = max(0, int((1 - d / max_dist) * 100)) | |
| name = cluster_labels[ci] if ci < len(cluster_labels) else f"Cluster {ci}" | |
| cdists.append((name, closeness)) | |
| # Nearest texts | |
| dists = np.linalg.norm(pts - pos, axis=1) | |
| nearest_idx = np.argsort(dists)[:5] | |
| nearest_texts = [result.session.entries[int(j)].text for j in nearest_idx] | |
| # Score means per channel | |
| score_means = {} | |
| for ch, vals in attractor_score_data.items(): | |
| basin_vals = vals[np.linalg.norm(pts - pos, axis=1) < float(np.percentile(dists, 10))] | |
| if len(basin_vals) > 0: | |
| score_means[ch] = round(float(np.mean(basin_vals)), 3) | |
| att_data = { | |
| 'index': i, | |
| 'label': f'A{i + 1}', | |
| 'position': [float(x) for x in pos], | |
| 'strength': float(att['strength']), | |
| 'basin_fraction': float(att['basin_fraction']), | |
| 'divergence': float(att['divergence']), | |
| 'axis_percentages': dict(zip(axis_labels, axis_pcts)), | |
| 'cluster_distances': cdists, | |
| 'nearest_texts': nearest_texts, | |
| 'score_means': score_means, | |
| } | |
| attractors_list.append(att_data) | |
| # Load cached attractor explanations | |
| att_explain_path = f"{cache_path}_attractor_explanations.json" | |
| attractor_explanations = {} | |
| if os.path.exists(att_explain_path): | |
| with open(att_explain_path, 'r', encoding='utf-8') as f: | |
| attractor_explanations = json.load(f) | |
| # Attach explanations to attractors | |
| for att in attractors_list: | |
| key = str(att['index']) | |
| if key in attractor_explanations: | |
| att['explanation'] = attractor_explanations[key] | |
| # Load saved paths | |
| paths_file = f"{cache_path}_explained_paths.json" | |
| saved_paths = [] | |
| if os.path.exists(paths_file): | |
| with open(paths_file, 'r', encoding='utf-8') as f: | |
| saved_paths = json.load(f) | |
| data = { | |
| 'cache_path': cache_path, | |
| 'flow_mode': result.flow_mode, | |
| 'n_entries': len(result.session.entries), | |
| 'axis_labels': axis_labels, | |
| 'clusters': clusters, | |
| 'attractors': attractors_list, | |
| 'saved_paths': saved_paths, | |
| 'score_channels': result.score_channels, | |
| } | |
| if output_dir: | |
| os.makedirs(output_dir, exist_ok=True) | |
| basename = os.path.basename(cache_path) | |
| # JSON export | |
| json_path = os.path.join(output_dir, f"{basename}_export.json") | |
| with open(json_path, 'w', encoding='utf-8') as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| print(f"Exported JSON: {json_path}") | |
| # Prompt-ready markdown | |
| md_path = os.path.join(output_dir, f"{basename}_prompt.md") | |
| with open(md_path, 'w', encoding='utf-8') as f: | |
| f.write(_build_prompt_md(data)) | |
| print(f"Exported prompt: {md_path}") | |
| return data | |
| def _build_prompt_md(data: dict) -> str: | |
| """Build a prompt-ready markdown from exported data.""" | |
| lines = [] | |
| lines.append(f"# Flow Analysis Export: {data['cache_path']}") | |
| lines.append(f"Flow mode: {data['flow_mode']} | Entries: {data['n_entries']}") | |
| lines.append("") | |
| # Axes | |
| lines.append("## Semantic Axes") | |
| for i, label in enumerate(data['axis_labels']): | |
| lines.append(f"- Axis {i + 1}: {label}") | |
| lines.append("") | |
| # Clusters | |
| lines.append("## Clusters") | |
| for c in data['clusters']: | |
| lines.append(f"- **{c['name']}** ({c['size']} entries)") | |
| lines.append("") | |
| # Attractors | |
| lines.append("## Attractors") | |
| for att in data['attractors']: | |
| lines.append(f"### {att['label']} (strength {att['strength']:.0%})") | |
| lines.append(f"Basin: {att['basin_fraction']:.2%} | Divergence: {att['divergence']:.4f}") | |
| lines.append("") | |
| lines.append("Location:") | |
| for name, pct in att['axis_percentages'].items(): | |
| lines.append(f" - {name}: {pct}%") | |
| lines.append("") | |
| if att.get('score_means'): | |
| lines.append("Score means:") | |
| for ch, val in att['score_means'].items(): | |
| lines.append(f" - {ch}: {val}") | |
| lines.append("") | |
| if att.get('cluster_distances'): | |
| lines.append("Cluster proximity:") | |
| for name, closeness in att['cluster_distances']: | |
| lines.append(f" - {name}: {closeness}%") | |
| lines.append("") | |
| if att.get('explanation'): | |
| lines.append("**Explanation:**") | |
| lines.append(att['explanation']) | |
| lines.append("") | |
| # Paths | |
| if data['saved_paths']: | |
| lines.append("## Explained Paths") | |
| for i, sp in enumerate(data['saved_paths']): | |
| lines.append(f"### Path P{i + 1}") | |
| journey = sp.get('journey', {}) | |
| # Journey metadata | |
| trend = journey.get('score_trend', 'unknown') | |
| visited = journey.get('attractors_visited', []) | |
| n_steps = journey.get('n_steps', 0) | |
| dwelling = journey.get('dwelling_fraction', 0) | |
| visited_str = ' → '.join(f'A{v+1}' for v in visited) if visited else 'none' | |
| lines.append(f"Trend: **{trend}** | Steps: {n_steps} | " | |
| f"Basins visited: {visited_str} | " | |
| f"Dwelling: {dwelling:.0%}") | |
| # Score journey (from dense samples) | |
| samples = journey.get('samples', []) | |
| if samples: | |
| lines.append("Score journey:") | |
| for s in samples: | |
| frac = s.get('step_frac', 0) | |
| scores_str = ', '.join(f'{k}={v:.2f}' | |
| for k, v in s.get('scores', {}).items()) | |
| att = s.get('nearest_attractor') | |
| att_str = f'near A{att+1}' if att is not None else '' | |
| pct = int(frac * 100) | |
| lines.append(f" {pct:3d}%: {scores_str} {att_str}") | |
| lines.append("") | |
| if sp.get('explanation'): | |
| lines.append("**Explanation:**") | |
| lines.append(sp['explanation']) | |
| lines.append("") | |
| return "\n".join(lines) | |