79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
"""
|
|
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 # 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
|
|
# 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",
|
|
session_id="test_session_id",
|
|
timestamp=datetime.now(tz=timezone.utc),
|
|
ew=ew,
|
|
ns=ns,
|
|
z=0,
|
|
kills=0,
|
|
kills_per_hour="kph_str",
|
|
onlinetime=str(timedelta(seconds=online_time)),
|
|
deaths=0,
|
|
# rares_found removed from telemetry payload; tracked via rare events
|
|
prismatic_taper_count=0,
|
|
vt_state="test state",
|
|
)
|
|
# 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
|
|
online_time += wait
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|