gravity_topology.py
1 """ 2 Gravity Well Topology Visualization 3 4 Renders the topology of gravity wells - showing: 5 - WHERE wells are forming (source nodes in graph) 6 - SCOPE of each well (temporal, altitude, topic) 7 - CONNECTIONS between wells 8 - FORMATION history 9 10 Output: GRAVITY-TOPOLOGY.md in Roam-style outliner format 11 - Hierarchical bullets (Markov blankets within blankets) 12 - Shapes embedded with wiki-links (local coherence) 13 - Properties with :: syntax (queryable) 14 """ 15 16 from dataclasses import dataclass, field 17 from datetime import datetime, timedelta 18 from typing import List, Dict, Set, Optional, Any 19 from pathlib import Path 20 from enum import Enum 21 22 # Import from theory of mind 23 import sys 24 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) 25 26 try: 27 from core.theory_of_mind.cognitive_fingerprint import ( 28 GravityWell, TemporalScope, AltitudeScope, CognitiveFingerprint 29 ) 30 except ImportError: 31 # Define locally if import fails 32 GravityWell = None 33 34 try: 35 from core.graph.shape_registry import get_shape 36 except ImportError: 37 # Fallback if shape registry not available 38 def get_shape(concept: str) -> Optional[str]: 39 return None 40 41 42 @dataclass 43 class WellFormationEvent: 44 """Tracks when and where a gravity well formed.""" 45 well_concept: str 46 formed_at: datetime 47 source_nodes: List[str] 48 formation_trigger: str # 'edge_count', 'cross_session', 'signal_word', 'manual' 49 initial_mass: float 50 context: str 51 52 53 class GravityTopologyRenderer: 54 """ 55 Renders gravity well topology to markdown with Mermaid diagrams. 56 57 Shows: 58 1. Current active wells with their scopes 59 2. Formation locations in the graph 60 3. Inter-well connections 61 4. Temporal evolution 62 """ 63 64 def __init__(self, output_dir: Path): 65 self.output_dir = output_dir 66 self.output_dir.mkdir(parents=True, exist_ok=True) 67 68 def render( 69 self, 70 wells: List[GravityWell], 71 formation_events: List[WellFormationEvent] = None, 72 current_altitude: str = None, 73 current_topic: str = None 74 ) -> Path: 75 """Render gravity topology to GRAVITY-TOPOLOGY.md in Roam-style outliner format.""" 76 output_path = self.output_dir / "GRAVITY-TOPOLOGY.md" 77 78 lines = [ 79 "# Gravity Well Topology", 80 "", 81 f"*Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}*", 82 "", 83 "---", 84 "", 85 "- **related**", 86 " - [[DAILY-SYNTHESIS]]", 87 " - [[LIVE-COMPRESSION]]", 88 " - [[GRAPH-STATE]]", 89 "", 90 "---", 91 "", 92 ] 93 94 # Section 1: Active Wells (Roam-style) 95 lines.extend(self._render_wells_roam(wells, current_altitude, current_topic)) 96 97 # Section 2: Scope Distribution (Roam-style) 98 lines.extend(self._render_scope_roam(wells)) 99 100 # Section 3: Formation History (Roam-style) 101 if formation_events: 102 lines.extend(self._render_formation_roam(formation_events)) 103 104 # Section 4: By Altitude (Roam-style) 105 lines.extend(self._render_altitude_roam(wells)) 106 107 # Section 5: By Temporal Scope (Roam-style) 108 lines.extend(self._render_temporal_roam(wells)) 109 110 # Footer 111 lines.append("---") 112 lines.append("") 113 lines.append(f"*[[Gravity Topology]] | {datetime.now().strftime('%Y-%m-%d %H:%M')}*") 114 115 output_path.write_text('\n'.join(lines)) 116 return output_path 117 118 def _render_wells_roam( 119 self, 120 wells: List[GravityWell], 121 current_altitude: str = None, 122 current_topic: str = None 123 ) -> List[str]: 124 """Render active wells in Roam-style outliner format with shapes.""" 125 lines = [ 126 "## Active Gravity Wells", 127 "", 128 "- **wells**", 129 ] 130 131 # Sort by mass descending 132 sorted_wells = sorted(wells, key=lambda w: w.mass, reverse=True) 133 134 for well in sorted_wells[:20]: 135 # Get shape for this concept if available 136 shape = get_shape(well.concept) 137 138 lines.append(f" - [[{well.concept}]]") 139 if shape: 140 lines.append(f" - shape:: {shape}") 141 lines.append(f" - mass:: {well.mass:.2f}") 142 lines.append(f" - temporal:: #{well.temporal_scope.value}") 143 lines.append(f" - altitude:: #{well.altitude_scope.value}") 144 145 if well.topic_scope: 146 topics = ', '.join(list(well.topic_scope)[:3]) 147 lines.append(f" - topics:: {topics}") 148 149 effective_mass = well.get_effective_mass(current_altitude, current_topic) 150 lines.append(f" - active:: {'yes' if effective_mass > 0 else 'no'}") 151 152 lines.extend(["", "---", ""]) 153 return lines 154 155 def _render_scope_roam(self, wells: List[GravityWell]) -> List[str]: 156 """Render scope distribution in Roam-style.""" 157 lines = [ 158 "## Scope Distribution", 159 "", 160 ] 161 162 # Temporal distribution 163 temporal_counts = {} 164 for well in wells: 165 scope = well.temporal_scope.value 166 temporal_counts[scope] = temporal_counts.get(scope, 0) + 1 167 168 lines.append("- **temporal**") 169 for scope in ['ephemeral', 'contextual', 'seasonal', 'permanent']: 170 count = temporal_counts.get(scope, 0) 171 if count > 0: 172 wells_in_scope = [w.concept for w in wells if w.temporal_scope.value == scope][:3] 173 lines.append(f" - #{scope}:: {count} wells") 174 for w in wells_in_scope: 175 lines.append(f" - [[{w}]]") 176 else: 177 lines.append(f" - #{scope}:: 0 wells") 178 179 # Altitude distribution 180 altitude_counts = {} 181 for well in wells: 182 scope = well.altitude_scope.value 183 altitude_counts[scope] = altitude_counts.get(scope, 0) + 1 184 185 lines.append("- **altitude**") 186 for scope, count in sorted(altitude_counts.items(), key=lambda x: -x[1]): 187 lines.append(f" - #{scope}:: {count} wells") 188 189 lines.extend(["", "---", ""]) 190 return lines 191 192 def _render_formation_roam(self, events: List[WellFormationEvent]) -> List[str]: 193 """Render formation history in Roam-style.""" 194 lines = [ 195 "## Formation History", 196 "", 197 "- **question**", 198 " - Where are gravity wells forming?", 199 "- **events**", 200 ] 201 202 sorted_events = sorted(events, key=lambda e: e.formed_at, reverse=True) 203 204 for event in sorted_events[:15]: 205 time_str = event.formed_at.strftime('%m-%d %H:%M') 206 lines.append(f" - {time_str}") 207 lines.append(f" - well:: [[{event.well_concept}]]") 208 lines.append(f" - trigger:: {event.formation_trigger}") 209 lines.append(f" - source:: [[{event.source_nodes[0] if event.source_nodes else 'unknown'}]]") 210 lines.append(f" - context:: {event.context}") 211 212 lines.extend(["", "---", ""]) 213 return lines 214 215 def _render_altitude_roam(self, wells: List[GravityWell]) -> List[str]: 216 """Render wells by altitude in Roam-style.""" 217 lines = [ 218 "## Wells by Altitude", 219 "", 220 "- **question**", 221 " - Which wells are active at each thinking level?", 222 ] 223 224 altitudes = ['philosophical', 'strategic', 'tactical', 'operational'] 225 226 for altitude in altitudes: 227 active_wells = [w for w in wells if w.applies_at_altitude(altitude)] 228 active_wells.sort(key=lambda w: w.mass, reverse=True) 229 230 lines.append(f"- **#{altitude}** ({len(active_wells)} wells)") 231 for w in active_wells[:5]: 232 shape = get_shape(w.concept) 233 lines.append(f" - [[{w.concept}]] ({w.mass:.2f})") 234 if shape: 235 lines.append(f" - shape:: {shape[:80]}...") 236 237 lines.extend(["", "---", ""]) 238 return lines 239 240 def _render_temporal_roam(self, wells: List[GravityWell]) -> List[str]: 241 """Render wells by temporal scope in Roam-style.""" 242 lines = [ 243 "## Wells by Temporal Scope", 244 "", 245 ] 246 247 scope_info = [ 248 (TemporalScope.PERMANENT, "always resonate"), 249 (TemporalScope.SEASONAL, "weeks to months"), 250 (TemporalScope.CONTEXTUAL, "days to weeks"), 251 (TemporalScope.EPHEMERAL, "hours to days"), 252 ] 253 254 for scope, desc in scope_info: 255 scope_wells = [w for w in wells if w.temporal_scope == scope] 256 scope_wells.sort(key=lambda w: w.mass, reverse=True) 257 258 lines.append(f"- **#{scope.value}** - {desc}") 259 if scope_wells: 260 for w in scope_wells[:7]: 261 days_since = (datetime.now() - w.last_activated).days 262 age = f"{days_since}d ago" if days_since > 0 else "today" 263 lines.append(f" - [[{w.concept}]] ({w.mass:.2f})") 264 lines.append(f" - last:: {age}") 265 else: 266 lines.append(" - (none)") 267 268 lines.append("") 269 return lines 270 271 def _render_wells_table( 272 self, 273 wells: List[GravityWell], 274 current_altitude: str = None, 275 current_topic: str = None 276 ) -> List[str]: 277 """Render active wells as a table.""" 278 lines = [ 279 "## Active Gravity Wells", 280 "", 281 "| Concept | Mass | Temporal | Altitude | Topics | Active? |", 282 "|---------|------|----------|----------|--------|---------|", 283 ] 284 285 # Sort by mass descending 286 sorted_wells = sorted(wells, key=lambda w: w.mass, reverse=True) 287 288 for well in sorted_wells[:20]: # Top 20 289 effective_mass = well.get_effective_mass(current_altitude, current_topic) 290 active = "✓" if effective_mass > 0 else "○" 291 292 topics = ', '.join(list(well.topic_scope)[:3]) if well.topic_scope else "all" 293 if len(well.topic_scope) > 3: 294 topics += "..." 295 296 lines.append( 297 f"| **{well.concept}** | {well.mass:.2f} | " 298 f"{well.temporal_scope.value} | {well.altitude_scope.value} | " 299 f"{topics} | {active} |" 300 ) 301 302 lines.extend(["", "---", ""]) 303 return lines 304 305 def _render_scope_analysis(self, wells: List[GravityWell]) -> List[str]: 306 """Analyze distribution of scopes.""" 307 lines = [ 308 "## Scope Distribution", 309 "", 310 ] 311 312 # Temporal distribution 313 temporal_counts = {} 314 for well in wells: 315 scope = well.temporal_scope.value 316 temporal_counts[scope] = temporal_counts.get(scope, 0) + 1 317 318 lines.append("### Temporal Scope") 319 lines.append("```") 320 for scope, count in sorted(temporal_counts.items(), key=lambda x: -x[1]): 321 bar = "█" * count 322 lines.append(f"{scope:12} [{count:2}] {bar}") 323 lines.append("```") 324 lines.append("") 325 326 # Altitude distribution 327 altitude_counts = {} 328 for well in wells: 329 scope = well.altitude_scope.value 330 altitude_counts[scope] = altitude_counts.get(scope, 0) + 1 331 332 lines.append("### Altitude Scope") 333 lines.append("```") 334 for scope, count in sorted(altitude_counts.items(), key=lambda x: -x[1]): 335 bar = "█" * count 336 lines.append(f"{scope:12} [{count:2}] {bar}") 337 lines.append("```") 338 339 lines.extend(["", "---", ""]) 340 return lines 341 342 def _render_topology_graph(self, wells: List[GravityWell]) -> List[str]: 343 """Render Mermaid graph of well topology.""" 344 lines = [ 345 "## Well Topology", 346 "", 347 "```mermaid", 348 "graph TB", 349 "", 350 " %% Styling", 351 " classDef permanent fill:#4a9eff,stroke:#1a5fb4,stroke-width:3px", 352 " classDef seasonal fill:#57e389,stroke:#26a269,stroke-width:2px", 353 " classDef contextual fill:#f9f06b,stroke:#c64600,stroke-width:2px", 354 " classDef ephemeral fill:#ff7b63,stroke:#c01c28,stroke-width:1px", 355 "", 356 ] 357 358 # Group wells by temporal scope 359 for well in wells[:15]: # Top 15 by mass 360 node_id = well.concept.replace(' ', '_').replace('-', '_')[:20] 361 label = f"{well.concept}\\n({well.mass:.2f})" 362 363 # Add scope info 364 if well.altitude_scope != AltitudeScope.ALL: 365 label += f"\\n[{well.altitude_scope.value}]" 366 367 lines.append(f" {node_id}[\"{label}\"]") 368 369 # Apply class based on temporal scope 370 class_name = well.temporal_scope.value 371 lines.append(f" class {node_id} {class_name}") 372 373 # Draw edges between related wells 374 lines.append("") 375 lines.append(" %% Related concepts") 376 377 seen_edges = set() 378 for well in wells[:15]: 379 node_id = well.concept.replace(' ', '_').replace('-', '_')[:20] 380 381 for related in well.related_concepts[:3]: 382 # Check if related concept is also a well 383 related_wells = [w for w in wells if w.concept.lower() == related.lower()] 384 if related_wells: 385 related_id = related.replace(' ', '_').replace('-', '_')[:20] 386 edge_key = tuple(sorted([node_id, related_id])) 387 388 if edge_key not in seen_edges: 389 lines.append(f" {node_id} --- {related_id}") 390 seen_edges.add(edge_key) 391 392 lines.extend([ 393 "", 394 "```", 395 "", 396 "**Legend:**", 397 "- 🔵 Permanent (axioms, core principles)", 398 "- 🟢 Seasonal (projects, themes)", 399 "- 🟡 Contextual (current focus)", 400 "- 🔴 Ephemeral (immediate task)", 401 "", 402 "---", 403 "", 404 ]) 405 return lines 406 407 def _render_formation_history(self, events: List[WellFormationEvent]) -> List[str]: 408 """Render where wells formed.""" 409 lines = [ 410 "## Formation History", 411 "", 412 "*Where are gravity wells forming?*", 413 "", 414 "| When | Concept | Trigger | Source | Context |", 415 "|------|---------|---------|--------|---------|", 416 ] 417 418 # Sort by time descending 419 sorted_events = sorted(events, key=lambda e: e.formed_at, reverse=True) 420 421 for event in sorted_events[:15]: 422 time_str = event.formed_at.strftime('%m-%d %H:%M') 423 sources = ', '.join(event.source_nodes[:2]) 424 if len(event.source_nodes) > 2: 425 sources += "..." 426 context = event.context[:30] + "..." if len(event.context) > 30 else event.context 427 428 lines.append( 429 f"| {time_str} | **{event.well_concept}** | " 430 f"{event.formation_trigger} | {sources} | {context} |" 431 ) 432 433 lines.extend(["", "---", ""]) 434 return lines 435 436 def _render_altitude_distribution(self, wells: List[GravityWell]) -> List[str]: 437 """Render which wells apply at each altitude.""" 438 lines = [ 439 "## Wells by Altitude", 440 "", 441 "*Which wells are active at each thinking level?*", 442 "", 443 ] 444 445 altitudes = ['philosophical', 'strategic', 'tactical', 'operational'] 446 447 for altitude in altitudes: 448 active_wells = [w for w in wells if w.applies_at_altitude(altitude)] 449 active_wells.sort(key=lambda w: w.mass, reverse=True) 450 451 lines.append(f"### {altitude.title()} ({len(active_wells)} wells)") 452 if active_wells: 453 for w in active_wells[:5]: 454 lines.append(f"- **{w.concept}** ({w.mass:.2f})") 455 else: 456 lines.append("*No active wells*") 457 lines.append("") 458 459 lines.extend(["---", ""]) 460 return lines 461 462 def _render_temporal_distribution(self, wells: List[GravityWell]) -> List[str]: 463 """Render wells by temporal scope.""" 464 lines = [ 465 "## Wells by Temporal Scope", 466 "", 467 ] 468 469 for scope in TemporalScope: 470 scope_wells = [w for w in wells if w.temporal_scope == scope] 471 scope_wells.sort(key=lambda w: w.mass, reverse=True) 472 473 if scope == TemporalScope.PERMANENT: 474 emoji = "🔵" 475 desc = "Always resonate" 476 elif scope == TemporalScope.SEASONAL: 477 emoji = "🟢" 478 desc = "Weeks to months" 479 elif scope == TemporalScope.CONTEXTUAL: 480 emoji = "🟡" 481 desc = "Days to weeks" 482 else: 483 emoji = "🔴" 484 desc = "Hours to days" 485 486 lines.append(f"### {emoji} {scope.value.title()} - {desc} ({len(scope_wells)})") 487 488 if scope_wells: 489 for w in scope_wells[:7]: 490 days_since = (datetime.now() - w.last_activated).days 491 age = f"{days_since}d ago" if days_since > 0 else "today" 492 lines.append(f"- **{w.concept}** ({w.mass:.2f}) - last: {age}") 493 else: 494 lines.append("*None*") 495 lines.append("") 496 497 lines.extend([ 498 "---", 499 "", 500 f"*Gravity Topology | {datetime.now().strftime('%Y-%m-%d %H:%M')}*" 501 ]) 502 return lines 503 504 505 def render_gravity_topology( 506 fingerprint: 'CognitiveFingerprint', 507 output_dir: Path, 508 current_altitude: str = None, 509 current_topic: str = None 510 ) -> Path: 511 """ 512 Convenience function to render gravity topology from a CognitiveFingerprint. 513 """ 514 renderer = GravityTopologyRenderer(output_dir) 515 wells = list(fingerprint.gravity_wells.values()) 516 return renderer.render(wells, current_altitude=current_altitude, current_topic=current_topic) 517 518 519 if __name__ == "__main__": 520 # Demo with sample data 521 print("=== Gravity Topology Demo ===\n") 522 523 wells = [ 524 GravityWell( 525 concept="sovereign os", 526 mass=0.95, 527 temporal_scope=TemporalScope.SEASONAL, 528 altitude_scope=AltitudeScope.ALL, 529 related_concepts=["attention", "resonance", "metacognition"] 530 ), 531 GravityWell( 532 concept="attention", 533 mass=0.88, 534 temporal_scope=TemporalScope.PERMANENT, 535 altitude_scope=AltitudeScope.STRATEGIC, 536 related_concepts=["eeg", "biometric", "flow"] 537 ), 538 GravityWell( 539 concept="resonance", 540 mass=0.85, 541 temporal_scope=TemporalScope.PERMANENT, 542 altitude_scope=AltitudeScope.ALL, 543 related_concepts=["gravity well", "flow", "attention"] 544 ), 545 GravityWell( 546 concept="eeg integration", 547 mass=0.72, 548 temporal_scope=TemporalScope.CONTEXTUAL, 549 altitude_scope=AltitudeScope.TACTICAL, 550 topic_scope={"biometric", "pipeline"}, 551 related_concepts=["mindmonitor", "attention"] 552 ), 553 GravityWell( 554 concept="mission control", 555 mass=0.65, 556 temporal_scope=TemporalScope.EPHEMERAL, 557 altitude_scope=AltitudeScope.OPERATIONAL, 558 related_concepts=["first officer", "synthesis"] 559 ), 560 ] 561 562 renderer = GravityTopologyRenderer(Path("/tmp")) 563 output = renderer.render(wells, current_altitude="strategic") 564 565 print(f"Output written to: {output}") 566 print(output.read_text())