File size: 2,729 Bytes
7ba87ed 9dfe153 7ba87ed 9dfe153 7ba87ed 2bf33f5 7ba87ed | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import json
import uvicorn
app = FastAPI()
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
# Serve static files from the "static" directory
app.mount("/static", StaticFiles(directory="static"), name="static")
# Set up Jinja2 templates
templates = Jinja2Templates(directory="templates")
with open("data.json", "r", encoding="utf8") as file:
maps_data = json.load(file)
@app.get("/", response_class=HTMLResponse)
async def read_index(request: Request):
return templates.TemplateResponse("index.html", {"request": request, "maps": maps_data})
@app.get("/map/{normalized_name}", response_class=HTMLResponse)
async def get_map_by_normalized_name(request: Request, normalized_name: str):
map_entry = maps_data.get(normalized_name)
if map_entry:
filtered_maps = [map_entry]
else:
filtered_maps = []
return templates.TemplateResponse("map.html", {"request": request, "maps": filtered_maps})
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
print("active_connections:", self.active_connections)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
print("active_connections:", self.active_connections)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
# print(f"Broadcast: {message}")
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(data) # Direktes Broadcasten der Daten
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(json.dumps({"type": "client_info", "data": f"Client #{client_id} left the chat"}))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
|