在 PyGame 中实现 WebSocket 服务器以通过 HTML WebSocket 客户端控制对象
Implement a WebSocket Server in PyGame to control objects via a HTML WebSocket Client
总体思路
我成功配置了 raspberry pi as an access point，这样我就可以通过 WIFI 连接我的手机 phone 或笔记本电脑。现在，我想 运行 PyGame 在连接到屏幕的 raspberry pi 上，并通过连接的手机 phone 或笔记本电脑控制游戏中的对象到 raspberry pi.
在下文中，我先提供简单的工作示例，然后展示不工作的地方。
测试 WebSocket 服务器有效
为了给客户端提供一些内容，我在raspberry pi上安装了一个nginx服务器。当我用 raspberry pi (192.168.4.1) 的 IP 地址打开浏览器时，出现索引页面。
然后，为了测试 WebSocket 服务器，我基于 websockets package:
编写了一个简单的 python 脚本
import websockets
import asyncio
# handler processes the message and sends "Success" back to the client
async def handler(websocket, path):
async for message in websocket:
await processMsg(message)
await websocket.send("Success")
async def processMsg(message):
print(f"[Received]: {message}")
async def main():
async with websockets.serve(handler, "192.168.4.1", 6677):
await asyncio.Future() # run forever
if __name__ == "__main__":
asyncio.run(main())
我通过设置一个 HTML 页面来测试服务器，该页面通过 Javascript 文件连接到 WebSocket 服务器，并实现了一个向服务器发送字符串的按钮：
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>WebSocket Test</title>
<script type="text/javascript" src="static/client.js"></script>
</head>
<body>
<div>
<h2>WebSocket Test</h2>
<input type="button" name="send" value="Send Hello!" onClick="sendHello()">
</div>
</body>
</html>
和 client.js 文件：
let socket = new WebSocket("ws://192.168.4.1:6677/");
socket.onopen = function(e) {
console.log("[open] Connection established");
console.log("[send] Sending to server");
socket.send("Web connection established")
};
socket.onmessage = function(event) {
console.log(`[message] Data received from server: ${event.data}`);
};
socket.onclose = function(event) {
if (event.wasClean) {
console.log(`[close] Connection closed cleanly, code=${event.code} reason=${event.reason}`);
} else {
console.log('[close] Connection died!')
}
};
socket.onerror = function(error) {
console.log(`[error] ${error.message}`);
};
function sendHello() {
console.log("[send] Sending to server");
socket.send("Hello!");
};
通过这些简单的示例文件，我可以成功地建立持久连接并在服务器和客户端之间交换数据。
将 WebSocket 服务器添加到 PyGame 无效
为了用 PyGame 测试 WebSocket 服务器，我设置了一个只显示蓝色圆圈的简单游戏：
# import and init pygame library
import pygame
pygame.init()
# screen dimensions
HEIGHT = 320
WIDTH = 480
# set up the drawing window
screen = pygame.display.set_mode([WIDTH,HEIGHT])
# run until the user asks to quit
running = True
while running:
# did the user close the window
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
# fill the background with white
screen.fill((255,255,255))
# draw a solid blue circle in the center
pygame.draw.circle(screen, (0,0,255), (int(WIDTH/2),int(HEIGHT/2)), 30)
# flip the display
pygame.display.flip()
pygame.quit()
游戏 运行 正常，在白色背景上显示一个蓝色圆圈。
问题是，当我想将 WebSocket 服务器添加到游戏中时，其中一个正在阻止另一部分的执行，即，我只能 运行 在没有活动服务器的情况下进行游戏，或者我可以 运行 WebSocket 服务器，但游戏未显示。例如，像这样将 WebSocket 服务器放在游戏循环的前面
# WS server
async def echo(websocket, path):
async for message in websocket:
msg = message
print(f"[Received] {message}")
await websocket.send(msg)
async def server():
async with websockets.serve(echo, "192.168.4.1", 6677):
await asyncio.Future()
asyncio.ensure_future(server())
或在服务器内包含游戏循环：
async def server():
async with websockets.serve(echo, "192.168.4.1", 6677):
#await asyncio.Future()
# set up the drawing window
screen = pygame.display.set_mode([WIDTH,HEIGHT])
# run until the user asks to quit
running = True
while running:
#pygame code goes here
通过广泛的 Google 搜索，我发现 PyGame 和 asyncio 包（websockets 基于 asyncio）不能简单地一起工作，因为我必须以某种方式兼顾两者，手动异步循环和游戏循环。
希望有人能帮我解决这个问题...
多亏了我的一位同事，我找到了 Pygame 运行ning 的解决方案，包括 Websockets 服务器和响应移动客户端上按下的按钮。重要的一点是 运行 Websockets 服务器在不同的线程中并正确处理事件。
这里是server.py的代码。我将 IP 地址更改为本地主机 (127.0.0.1)，以便在我的笔记本电脑上测试代码。
import websockets
import asyncio
import pygame
IPADDRESS = "127.0.0.1"
PORT = 6677
EVENTTYPE = pygame.event.custom_type()
# handler processes the message and sends "Success" back to the client
async def handler(websocket, path):
async for message in websocket:
await processMsg(message)
await websocket.send("Success")
async def processMsg(message):
print(f"[Received]: {message}")
pygame.fastevent.post(pygame.event.Event(EVENTTYPE, message=message))
async def main(future):
async with websockets.serve(handler, IPADDRESS, PORT):
await future # run forever
if __name__ == "__main__":
asyncio.run(main())
这里的重要部分是 pygame.fastevent.post 触发 Pygame 事件的方法。
然后，在game.py中，服务器在不同的线程中启动：
# import and init pygame library
import threading
import asyncio
import pygame
import server
def start_server(loop, future):
loop.run_until_complete(server.main(future))
def stop_server(loop, future):
loop.call_soon_threadsafe(future.set_result, None)
loop = asyncio.get_event_loop()
future = loop.create_future()
thread = threading.Thread(target=start_server, args=(loop, future))
thread.start()
pygame.init()
pygame.fastevent.init()
# screen dimensions
HEIGHT = 320
WIDTH = 480
# set up the drawing window
screen = pygame.display.set_mode([WIDTH, HEIGHT])
color = pygame.Color('blue')
radius = 30
x = int(WIDTH/2)
# run until the user asks to quit
running = True
while running:
# did the user close the window
for event in pygame.fastevent.get():
if event.type == pygame.QUIT:
running = False
elif event.type == server.EVENTTYPE:
print(event.message)
color = pygame.Color('red')
x = (x + radius / 3) % (WIDTH - radius * 2) + radius
# fill the background with white
screen.fill((255,255,255))
# draw a solid blue circle in the center
pygame.draw.circle(screen, color, (x, int(HEIGHT/2)), radius)
# flip the display
pygame.display.flip()
print("Stoping event loop")
stop_server(loop, future)
print("Waiting for termination")
thread.join()
print("Shutdown pygame")
pygame.quit()
当您想退出 Pygame！
时，请注意正确关闭线程 运行ning Websockets 服务器
现在，当您使用包含我在问题中发布的 client.js 脚本的 HTML 文件时，每次按下“发送问候”按钮。
我希望这对所有尝试使用 Pygame 设置在线多人游戏的人有所帮助。
总体思路
我成功配置了 raspberry pi as an access point，这样我就可以通过 WIFI 连接我的手机 phone 或笔记本电脑。现在，我想 运行 PyGame 在连接到屏幕的 raspberry pi 上，并通过连接的手机 phone 或笔记本电脑控制游戏中的对象到 raspberry pi.
在下文中，我先提供简单的工作示例，然后展示不工作的地方。
测试 WebSocket 服务器有效
为了给客户端提供一些内容，我在raspberry pi上安装了一个nginx服务器。当我用 raspberry pi (192.168.4.1) 的 IP 地址打开浏览器时，出现索引页面。
然后，为了测试 WebSocket 服务器，我基于 websockets package:
编写了一个简单的 python 脚本import websockets
import asyncio
# handler processes the message and sends "Success" back to the client
async def handler(websocket, path):
async for message in websocket:
await processMsg(message)
await websocket.send("Success")
async def processMsg(message):
print(f"[Received]: {message}")
async def main():
async with websockets.serve(handler, "192.168.4.1", 6677):
await asyncio.Future() # run forever
if __name__ == "__main__":
asyncio.run(main())
我通过设置一个 HTML 页面来测试服务器，该页面通过 Javascript 文件连接到 WebSocket 服务器，并实现了一个向服务器发送字符串的按钮：
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>WebSocket Test</title>
<script type="text/javascript" src="static/client.js"></script>
</head>
<body>
<div>
<h2>WebSocket Test</h2>
<input type="button" name="send" value="Send Hello!" onClick="sendHello()">
</div>
</body>
</html>
和 client.js 文件：
let socket = new WebSocket("ws://192.168.4.1:6677/");
socket.onopen = function(e) {
console.log("[open] Connection established");
console.log("[send] Sending to server");
socket.send("Web connection established")
};
socket.onmessage = function(event) {
console.log(`[message] Data received from server: ${event.data}`);
};
socket.onclose = function(event) {
if (event.wasClean) {
console.log(`[close] Connection closed cleanly, code=${event.code} reason=${event.reason}`);
} else {
console.log('[close] Connection died!')
}
};
socket.onerror = function(error) {
console.log(`[error] ${error.message}`);
};
function sendHello() {
console.log("[send] Sending to server");
socket.send("Hello!");
};
通过这些简单的示例文件，我可以成功地建立持久连接并在服务器和客户端之间交换数据。
将 WebSocket 服务器添加到 PyGame 无效
为了用 PyGame 测试 WebSocket 服务器，我设置了一个只显示蓝色圆圈的简单游戏：
# import and init pygame library
import pygame
pygame.init()
# screen dimensions
HEIGHT = 320
WIDTH = 480
# set up the drawing window
screen = pygame.display.set_mode([WIDTH,HEIGHT])
# run until the user asks to quit
running = True
while running:
# did the user close the window
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
# fill the background with white
screen.fill((255,255,255))
# draw a solid blue circle in the center
pygame.draw.circle(screen, (0,0,255), (int(WIDTH/2),int(HEIGHT/2)), 30)
# flip the display
pygame.display.flip()
pygame.quit()
游戏 运行 正常，在白色背景上显示一个蓝色圆圈。
问题是，当我想将 WebSocket 服务器添加到游戏中时，其中一个正在阻止另一部分的执行，即，我只能 运行 在没有活动服务器的情况下进行游戏，或者我可以 运行 WebSocket 服务器，但游戏未显示。例如，像这样将 WebSocket 服务器放在游戏循环的前面
# WS server
async def echo(websocket, path):
async for message in websocket:
msg = message
print(f"[Received] {message}")
await websocket.send(msg)
async def server():
async with websockets.serve(echo, "192.168.4.1", 6677):
await asyncio.Future()
asyncio.ensure_future(server())
或在服务器内包含游戏循环：
async def server():
async with websockets.serve(echo, "192.168.4.1", 6677):
#await asyncio.Future()
# set up the drawing window
screen = pygame.display.set_mode([WIDTH,HEIGHT])
# run until the user asks to quit
running = True
while running:
#pygame code goes here
通过广泛的 Google 搜索，我发现 PyGame 和 asyncio 包（websockets 基于 asyncio）不能简单地一起工作，因为我必须以某种方式兼顾两者，手动异步循环和游戏循环。
希望有人能帮我解决这个问题...
多亏了我的一位同事，我找到了 Pygame 运行ning 的解决方案，包括 Websockets 服务器和响应移动客户端上按下的按钮。重要的一点是 运行 Websockets 服务器在不同的线程中并正确处理事件。
这里是server.py的代码。我将 IP 地址更改为本地主机 (127.0.0.1)，以便在我的笔记本电脑上测试代码。
import websockets
import asyncio
import pygame
IPADDRESS = "127.0.0.1"
PORT = 6677
EVENTTYPE = pygame.event.custom_type()
# handler processes the message and sends "Success" back to the client
async def handler(websocket, path):
async for message in websocket:
await processMsg(message)
await websocket.send("Success")
async def processMsg(message):
print(f"[Received]: {message}")
pygame.fastevent.post(pygame.event.Event(EVENTTYPE, message=message))
async def main(future):
async with websockets.serve(handler, IPADDRESS, PORT):
await future # run forever
if __name__ == "__main__":
asyncio.run(main())
这里的重要部分是 pygame.fastevent.post 触发 Pygame 事件的方法。
然后，在game.py中，服务器在不同的线程中启动：
# import and init pygame library
import threading
import asyncio
import pygame
import server
def start_server(loop, future):
loop.run_until_complete(server.main(future))
def stop_server(loop, future):
loop.call_soon_threadsafe(future.set_result, None)
loop = asyncio.get_event_loop()
future = loop.create_future()
thread = threading.Thread(target=start_server, args=(loop, future))
thread.start()
pygame.init()
pygame.fastevent.init()
# screen dimensions
HEIGHT = 320
WIDTH = 480
# set up the drawing window
screen = pygame.display.set_mode([WIDTH, HEIGHT])
color = pygame.Color('blue')
radius = 30
x = int(WIDTH/2)
# run until the user asks to quit
running = True
while running:
# did the user close the window
for event in pygame.fastevent.get():
if event.type == pygame.QUIT:
running = False
elif event.type == server.EVENTTYPE:
print(event.message)
color = pygame.Color('red')
x = (x + radius / 3) % (WIDTH - radius * 2) + radius
# fill the background with white
screen.fill((255,255,255))
# draw a solid blue circle in the center
pygame.draw.circle(screen, color, (x, int(HEIGHT/2)), radius)
# flip the display
pygame.display.flip()
print("Stoping event loop")
stop_server(loop, future)
print("Waiting for termination")
thread.join()
print("Shutdown pygame")
pygame.quit()
当您想退出 Pygame！
时，请注意正确关闭线程 运行ning Websockets 服务器现在，当您使用包含我在问题中发布的 client.js 脚本的 HTML 文件时，每次按下“发送问候”按钮。
我希望这对所有尝试使用 Pygame 设置在线多人游戏的人有所帮助。