new comments

generate_data.py

@@ -1,19 +1,46 @@
-import asyncio
-import websockets
-import json
+"""
+generate_data.py - Standalone script to simulate plugin telemetry data.
+
+This script connects to the plugin WebSocket at /ws/position and sends
+fabricated TelemetrySnapshot payloads at regular intervals. Useful for:
+  - Functional testing of the telemetry ingestion pipeline
+  - Demonstrating real-time map updates without a live game client
+"""
+import asyncio       # Async event loop and sleep support
+import websockets    # WebSocket client for Python
+import json          # JSON serialization of payloads
 from datetime import datetime, timedelta, timezone
-from main import TelemetrySnapshot
+from main import TelemetrySnapshot  # Pydantic model matches plugin protocol
 
 
 async def main() -> None:
+    """
+    Continuously emit synthetic telemetry snapshots at fixed intervals.
+
+    Updates in-game coordinates (ew, ns) gradually and increments
+    an 'online_time' counter to mimic real gameplay progression.
+    Each iteration:
+      1. Build TelemetrySnapshot with current state
+      2. Serialize to JSON, set 'type' field
+      3. Send over WebSocket
+      4. Sleep for 'wait' seconds
+    """
+    # Interval between snapshots (seconds)
     wait = 10
-    online_time = 24 * 3600  # start at 1 day
+    # Simulated total online time in seconds (starting at 24h)
+    online_time = 24 * 3600
+    # Starting coordinates (E/W and N/S)
     ew = 0.0
     ns = 0.0
+    # WebSocket endpoint for plugin telemetry (include secret for auth)
     uri = "ws://localhost:8000/ws/position?secret=your_shared_secret"
+    # Connect to the plugin WebSocket endpoint with authentication
+    # Establish WebSocket connection to the server
     async with websockets.connect(uri) as websocket:
         print(f"Connected to {uri}")
+        # Loop indefinitely, sending telemetry at each interval
         while True:
+            # Construct a new TelemetrySnapshot dataclass instance
             snapshot = TelemetrySnapshot(
                 character_name="Test name",
                 char_tag="test_tag",
@@ -30,13 +57,18 @@ async def main() -> None:
                 prismatic_taper_count=0,
                 vt_state="test state",
             )
-            # wrap in envelope with message type
-            # Serialize telemetry snapshot without telemetry.kind 'rares_found'
+            # Prepare payload dictionary:
+            #  - Convert Pydantic model to dict
+            #  - Remove any extraneous fields (e.g., 'rares_found')
+            #  - Insert message 'type' for server routing
             payload = snapshot.model_dump()
             payload.pop("rares_found", None)
             payload["type"] = "telemetry"
+            # Send JSON-encoded payload over WebSocket
+            # Transmit JSON payload (datetime serialized via default=str)
             await websocket.send(json.dumps(payload, default=str))
             print(f"Sent snapshot: EW={ew:.2f}, NS={ns:.2f}")
+            # Wait before next update, then increment simulated state
             await asyncio.sleep(wait)
             ew += 0.1
             ns += 0.1