pygame联网飞机大战游戏实现

客户端

import pygame
import socket
import json
import threadingclass GameClient:def __init__(self):pygame.init()self.screen_width = 600  # 宽度减小self.screen_height = 800  # 高度增加self.screen = pygame.display.set_mode((self.screen_width, self.screen_height))pygame.display.set_caption("竖版飞机大战")self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)self.server_address = ('localhost', 5555)self.player_id = Noneself.game_state = {'players': {},'bullets': [],'enemies': []}self.input_state = {'up': False,'down': False,'left': False,'right': False,'shoot': False}# 连接服务器self.connect_to_server()self.running = True# 启动接收线程threading.Thread(target=self.receive_data).start()# 游戏主循环self.clock = pygame.time.Clock()self.game_loop()def connect_to_server(self):join_message = {'type': 'join'}self.client_socket.sendto(json.dumps(join_message).encode(), self.server_address)def receive_data(self):while self.running:try:data, _ = self.client_socket.recvfrom(4096)message = json.loads(data.decode())if message['type'] == 'welcome':self.player_id = message['id']elif message['type'] == 'update':self.game_state = messageexcept:passdef game_loop(self):while self.running:# 处理输入事件for event in pygame.event.get():if event.type == pygame.QUIT:self.running = Falseelif event.type == pygame.KEYDOWN:if event.key == pygame.K_UP:self.input_state['up'] = Trueelif event.key == pygame.K_DOWN:self.input_state['down'] = Trueelif event.key == pygame.K_LEFT:self.input_state['left'] = Trueelif event.key == pygame.K_RIGHT:self.input_state['right'] = Trueelif event.key == pygame.K_SPACE:self.input_state['shoot'] = Trueelif event.type == pygame.KEYUP:if event.key == pygame.K_UP:self.input_state['up'] = Falseelif event.key == pygame.K_DOWN:self.input_state['down'] = Falseelif event.key == pygame.K_LEFT:self.input_state['left'] = Falseelif event.key == pygame.K_RIGHT:self.input_state['right'] = Falseelif event.key == pygame.K_SPACE:self.input_state['shoot'] = False# 发送输入状态到服务器if self.player_id is not None:input_message = {'type': 'input','id': self.player_id,**self.input_state}self.client_socket.sendto(json.dumps(input_message).encode(), self.server_address)# 渲染游戏self.render()self.clock.tick(60)pygame.quit()def render(self):self.screen.fill((0, 0, 0))# 绘制玩家for player_id, player in self.game_state['players'].items():color = (0, 255, 0) if int(player_id) == self.player_id else (0, 0, 255)pygame.draw.rect(self.screen, color, (player['x'], player['y'], player['width'], player['height']))# 显示生命值if 'health' in player:font = pygame.font.SysFont(None, 24)health_text = font.render(f"HP: {player['health']}", True, (255, 255, 255))self.screen.blit(health_text, (player['x'], player['y'] - 20))# 显示分数if 'score' in player:font = pygame.font.SysFont(None, 24)score_text = font.render(f"Score: {player['score']}", True, (255, 255, 255))self.screen.blit(score_text, (player['x'], player['y'] - 40))# 绘制子弹for bullet in self.game_state['bullets']:pygame.draw.rect(self.screen, (255, 255, 0), (bullet['x'], bullet['y'], bullet['width'], bullet['height']))# 绘制敌机for enemy in self.game_state['enemies']:pygame.draw.circle(self.screen, (255, 0, 0), (enemy['x'], enemy['y']), enemy['radius'])pygame.display.flip()if __name__ == "__main__":client = GameClient()

服务器

import socket
import threading
import json
import random
import timeclass GameServer:def __init__(self):self.players = {}self.bullets = []self.enemies = []self.last_enemy_spawn = 0self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)self.server_socket.bind(('0.0.0.0', 5555))self.running = True# 游戏参数self.screen_width = 600  # 宽度可以小一些self.screen_height = 800  # 高度增加self.player_speed = 5self.bullet_speed = 7self.enemy_speed = 3# 启动接收线程threading.Thread(target=self.receive_data).start()threading.Thread(target=self.game_loop).start()def receive_data(self):while self.running:try:data, addr = self.server_socket.recvfrom(1024)message = json.loads(data.decode())if message['type'] == 'join':player_id = len(self.players)# 玩家初始位置改为底部self.players[player_id] = {'x': 200 if player_id == 0 else 400,'y': self.screen_height - 100,'width': 40,'height': 40,'addr': addr,'last_shot': 0}response = {'type': 'welcome', 'id': player_id}self.server_socket.sendto(json.dumps(response).encode(), addr)elif message['type'] == 'input':player_id = message['id']if player_id in self.players:player = self.players[player_id]if 'up' in message and message['up']:player['y'] = max(0, player['y'] - self.player_speed)if 'down' in message and message['down']:player['y'] = min(self.screen_height - player['height'], player['y'] + self.player_speed)if 'left' in message and message['left']:player['x'] = max(0, player['x'] - self.player_speed)if 'right' in message and message['right']:player['x'] = min(self.screen_width - player['width'], player['x'] + self.player_speed)if 'shoot' in message and message['shoot']:current_time = time.time()if current_time - player['last_shot'] > 0.2:  # 射击冷却# 子弹改为向上发射self.bullets.append({'x': player['x'] + player['width'] // 2 - 5,'y': player['y'],'width': 5,'height': 10,'speed': -self.bullet_speed,  # 负值表示向上'owner': player_id})player['last_shot'] = current_timeexcept:passdef game_loop(self):while self.running:# 生成敌机(从顶部)current_time = time.time()if current_time - self.last_enemy_spawn > 2.0:  # 每2秒生成一个敌机self.enemies.append({'x': random.randint(0, self.screen_width - 20),'y': -20,  # 从屏幕外顶部生成'radius': 10,'speed': self.enemy_speed})self.last_enemy_spawn = current_time# 更新子弹位置for bullet in self.bullets[:]:bullet['y'] += bullet['speed']  # 速度可以是正或负if bullet['y'] < -10 or bullet['y'] > self.screen_height + 10:self.bullets.remove(bullet)# 更新敌机位置(向下移动)for enemy in self.enemies[:]:enemy['y'] += enemy['speed']if enemy['y'] > self.screen_height + 20:self.enemies.remove(enemy)# 碰撞检测self.check_collisions()# 广播游戏状态self.broadcast_state()time.sleep(0.016)  # 约60FPSdef check_collisions(self):# 子弹与敌机碰撞for bullet in self.bullets[:]:for enemy in self.enemies[:]:if (bullet['x'] < enemy['x'] + enemy['radius'] andbullet['x'] + bullet['width'] > enemy['x'] andbullet['y'] < enemy['y'] + enemy['radius'] andbullet['y'] + bullet['height'] > enemy['y']):# 增加玩家分数if bullet['owner'] in self.players:if 'score' not in self.players[bullet['owner']]:self.players[bullet['owner']]['score'] = 0self.players[bullet['owner']]['score'] += 1if bullet in self.bullets:self.bullets.remove(bullet)if enemy in self.enemies:self.enemies.remove(enemy)break# 玩家与敌机碰撞for player_id, player in self.players.items():for enemy in self.enemies[:]:if (player['x'] < enemy['x'] + enemy['radius'] andplayer['x'] + player['width'] > enemy['x'] andplayer['y'] < enemy['y'] + enemy['radius'] andplayer['y'] + player['height'] > enemy['y']):# 减少玩家生命值if 'health' not in player:player['health'] = 3player['health'] -= 1if enemy in self.enemies:self.enemies.remove(enemy)if player['health'] <= 0:# 玩家死亡，重置位置player['x'] = 200 if player_id == 0 else 400player['y'] = self.screen_height - 100player['health'] = 3def broadcast_state(self):game_state = {'type': 'update','players': self.players,'bullets': self.bullets,'enemies': self.enemies}for player_id, player in self.players.items():try:self.server_socket.sendto(json.dumps(game_state).encode(), player['addr'])except:passif __name__ == "__main__":server = GameServer()input("Press Enter to stop the server...\n")server.running = False