-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
66 lines (51 loc) · 2.09 KB
/
Copy pathserver.py
File metadata and controls
66 lines (51 loc) · 2.09 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
import asyncio
import json
import os
from physics import RocketSimulation
app = FastAPI()
# Mount static files (HTML, JS, CSS)
os.makedirs("web", exist_ok=True)
app.mount("/static", StaticFiles(directory="web"), name="static")
@app.get("/")
async def get():
with open("web/index.html", "r") as f:
return HTMLResponse(f.read())
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
sim = None
try:
while True:
# Wait for config from client to start
data = await websocket.receive_text()
config = json.loads(data)
print(f"Received mission config: {config}")
sim = RocketSimulation(config)
# Run simulation loop
# The python backend runs dt=0.05 (20hz). We can warp time easily here.
TIME_WARP = 2
while sim.current_phase != 5: # 5 = LANDED
try:
# Compute multiple steps per tick to achieve time warp
for _ in range(TIME_WARP):
telemetry = sim.step()
# Send latest state to browser
await websocket.send_json(telemetry)
except Exception as e:
print(f"Error during simulation step: {e}")
await websocket.send_json({"error": str(e)})
break
# Sleep to match real time (20Hz broadcast rate)
await asyncio.sleep(0.05)
# Send final landed state
await websocket.send_json(sim._get_telemetry())
print("Mission completed.")
except WebSocketDisconnect:
print("Client disconnected")
if __name__ == "__main__":
print("Starting server on http://localhost:8000")
uvicorn.run(app, host="0.0.0.0", port=8000)