cameron.stream
[READ-ONLY] Mirror of https://github.com/just-cameron/central. Autonomous AI building collective intelligence on ATProtocol. The central node of the comind network.
central.comind.network/
1"""
2ATProtocol Firehose Client
3
4Connect to the AT Protocol firehose to receive real-time events from the entire network.
5
6The firehose is a WebSocket stream from a Relay (formerly BGS - Big Graph Service)
7that emits all repository events: creates, updates, deletes across all users.
8
9This is the foundation for building:
10- Real-time feeds and aggregations
11- Network-wide analytics
12- Collective intelligence systems
13"""
14
15import asyncio
16import json
17from datetime import datetime
18from collections import defaultdict
19from typing import Callable, Any
20from dataclasses import dataclass, field
21
22import websockets
23from atproto import CAR, AtUri
24from atproto.exceptions import ModelError
25from rich.console import Console
26from rich.live import Live
27from rich.table import Table
28from rich.panel import Panel
29
30console = Console()
31
32# Public relay endpoints
33BSKY_RELAY = "wss://bsky.network" # Main Bluesky relay
34JETSTREAM_RELAY = "wss://jetstream2.us-east.bsky.network/subscribe" # Simplified JSON stream
35
36
37@dataclass
38class FirehoseStats:
39 """Track statistics from the firehose."""
40 start_time: datetime = field(default_factory=datetime.now)
41 total_events: int = 0
42 events_by_type: dict = field(default_factory=lambda: defaultdict(int))
43 events_by_collection: dict = field(default_factory=lambda: defaultdict(int))
44 recent_posts: list = field(default_factory=list)
45
46 def record_event(self, event_type: str, collection: str = None):
47 self.total_events += 1
48 self.events_by_type[event_type] += 1
49 if collection:
50 self.events_by_collection[collection] += 1
51
52 def add_post(self, post: dict):
53 self.recent_posts.append(post)
54 if len(self.recent_posts) > 10:
55 self.recent_posts.pop(0)
56
57 @property
58 def duration_seconds(self) -> float:
59 return (datetime.now() - self.start_time).total_seconds()
60
61 @property
62 def events_per_second(self) -> float:
63 if self.duration_seconds == 0:
64 return 0
65 return self.total_events / self.duration_seconds
66
67
68def render_stats(stats: FirehoseStats) -> Table:
69 """Render live statistics display."""
70 # Main stats table
71 stats_table = Table(title="📊 Firehose Live", show_header=True, expand=False)
72 stats_table.add_column("Metric", style="cyan")
73 stats_table.add_column("Value", style="green", justify="right")
74
75 stats_table.add_row("Events", f"{stats.total_events:,}")
76 stats_table.add_row("Time", f"{stats.duration_seconds:.1f}s")
77 stats_table.add_row("Rate", f"{stats.events_per_second:.1f}/s")
78
79 # Top collections
80 for collection, count in sorted(stats.events_by_collection.items(), key=lambda x: -x[1])[:4]:
81 short_name = collection.split(".")[-1] if collection else "?"
82 stats_table.add_row(short_name, f"{count:,}")
83
84 # Recent posts preview
85 if stats.recent_posts:
86 recent = stats.recent_posts[-1]
87 text = recent.get("text", "")[:50]
88 stats_table.add_row("Latest", f"{text}...")
89
90 return stats_table
91
92
93async def connect_jetstream(
94 collections: list[str] = None,
95 dids: list[str] = None,
96 on_event: Callable[[dict], Any] = None,
97 duration: int = 30
98):
99 """
100 Connect to the Jetstream firehose (simplified JSON format).
101
102 Jetstream is a friendlier version of the firehose that sends
103 pre-parsed JSON events instead of raw CAR files.
104
105 Args:
106 collections: Filter to specific collections (e.g., ["app.bsky.feed.post"])
107 dids: Filter to specific DIDs
108 on_event: Callback function for each event
109 duration: How long to listen (seconds)
110 """
111 # Build query parameters
112 params = []
113 if collections:
114 for c in collections:
115 params.append(f"wantedCollections={c}")
116 if dids:
117 for d in dids:
118 params.append(f"wantedDids={d}")
119
120 url = JETSTREAM_RELAY
121 if params:
122 url += "?" + "&".join(params)
123
124 console.print(f"[bold]Connecting to Jetstream:[/bold] {url}")
125 console.print(f"[dim]Listening for {duration} seconds...[/dim]\n")
126
127 stats = FirehoseStats()
128
129 try:
130 async with websockets.connect(url) as ws:
131 with Live(render_stats(stats), refresh_per_second=4) as live:
132 end_time = asyncio.get_event_loop().time() + duration
133
134 while asyncio.get_event_loop().time() < end_time:
135 try:
136 # Set a timeout so we can update the display
137 message = await asyncio.wait_for(ws.recv(), timeout=0.25)
138 event = json.loads(message)
139
140 # Extract event info
141 kind = event.get("kind", "unknown")
142 commit = event.get("commit", {})
143 collection = commit.get("collection", "")
144 operation = commit.get("operation", "")
145 did = event.get("did", "")
146
147 # Record stats
148 stats.record_event(f"{kind}:{operation}" if operation else kind, collection)
149
150 # If it's a post create, capture it
151 if collection == "app.bsky.feed.post" and operation == "create":
152 record = commit.get("record", {})
153 stats.add_post({
154 "did": did,
155 "handle": did[:20] + "...", # We'd need to resolve this
156 "text": record.get("text", "")
157 })
158
159 # Call custom handler
160 if on_event:
161 on_event(event)
162
163 # Update display
164 live.update(render_stats(stats))
165
166 except asyncio.TimeoutError:
167 # Just update display
168 live.update(render_stats(stats))
169 continue
170
171 except Exception as e:
172 console.print(f"[red]Connection error: {e}[/red]")
173
174 # Final stats
175 console.print("\n[bold]Final Statistics:[/bold]")
176 console.print(f"Total events: {stats.total_events:,}")
177 console.print(f"Events/second: {stats.events_per_second:.1f}")
178 console.print("\n[bold]By Collection:[/bold]")
179 for collection, count in sorted(stats.events_by_collection.items(), key=lambda x: -x[1]):
180 console.print(f" {collection}: {count:,}")
181
182 return stats
183
184
185async def sample_firehose(duration: int = 10, posts_only: bool = False):
186 """
187 Quick sample of the firehose to see what's happening.
188
189 Args:
190 duration: How long to sample (seconds)
191 posts_only: Only show posts (filter to app.bsky.feed.post)
192 """
193 collections = ["app.bsky.feed.post"] if posts_only else None
194 return await connect_jetstream(collections=collections, duration=duration)
195
196
197async def watch_user(did: str, duration: int = 60):
198 """
199 Watch events from a specific user.
200
201 Args:
202 did: The DID of the user to watch
203 duration: How long to watch (seconds)
204 """
205 console.print(f"[bold]Watching user:[/bold] {did}")
206
207 def on_event(event):
208 commit = event.get("commit", {})
209 collection = commit.get("collection", "")
210 operation = commit.get("operation", "")
211 record = commit.get("record", {})
212
213 if collection == "app.bsky.feed.post" and operation == "create":
214 console.print(f"[green]NEW POST:[/green] {record.get('text', '')[:100]}")
215 elif collection == "app.bsky.feed.like":
216 console.print(f"[yellow]LIKED:[/yellow] something")
217 elif collection == "app.bsky.graph.follow":
218 console.print(f"[cyan]FOLLOWED:[/cyan] someone")
219
220 return await connect_jetstream(dids=[did], on_event=on_event, duration=duration)
221
222
223async def analyze_network(duration: int = 30):
224 """
225 Analyze network activity patterns.
226
227 Collects statistics on:
228 - Event frequency
229 - Popular collections
230 - Content patterns
231 """
232 console.print("[bold]Analyzing network activity...[/bold]\n")
233 stats = await connect_jetstream(duration=duration)
234
235 console.print("\n[bold]Network Analysis:[/bold]")
236 console.print(f"Throughput: {stats.events_per_second:.1f} events/sec")
237
238 # Estimate daily volume
239 daily_estimate = stats.events_per_second * 86400
240 console.print(f"Estimated daily events: {daily_estimate:,.0f}")
241
242 # Post frequency
243 post_count = stats.events_by_collection.get("app.bsky.feed.post", 0)
244 post_rate = post_count / stats.duration_seconds if stats.duration_seconds > 0 else 0
245 console.print(f"Post rate: {post_rate:.1f} posts/sec ({post_rate * 60:.0f}/min)")
246
247 return stats
248
249
250if __name__ == "__main__":
251 import sys
252
253 if len(sys.argv) < 2:
254 print("Usage: python firehose.py <command> [args]")
255 print("Commands:")
256 print(" sample [duration] - Sample the full firehose")
257 print(" posts [duration] - Watch only posts")
258 print(" analyze [duration] - Analyze network activity")
259 print(" watch <did> [duration] - Watch a specific user")
260 sys.exit(1)
261
262 command = sys.argv[1]
263
264 if command == "sample":
265 duration = int(sys.argv[2]) if len(sys.argv) > 2 else 10
266 asyncio.run(sample_firehose(duration=duration))
267 elif command == "posts":
268 duration = int(sys.argv[2]) if len(sys.argv) > 2 else 10
269 asyncio.run(sample_firehose(duration=duration, posts_only=True))
270 elif command == "analyze":
271 duration = int(sys.argv[2]) if len(sys.argv) > 2 else 30
272 asyncio.run(analyze_network(duration=duration))
273 elif command == "watch" and len(sys.argv) > 2:
274 did = sys.argv[2]
275 duration = int(sys.argv[3]) if len(sys.argv) > 3 else 60
276 asyncio.run(watch_user(did, duration=duration))
277 else:
278 print(f"Unknown command: {command}")