Ce que vous allez apprendre
Dans ce chapitre, nous ajoutons un mode 2 joueurs en reseau via localhost. Un programme lance avec --server affiche la raquette du bas, l'autre avec --client la raquette du haut. La balle rebondit entre les deux joueurs.
Le concept
Le reseau utilise le module socket de Python avec le protocole TCP. Le serveur est autoritaire sur la balle : il calcule sa position et l'envoie au client. Chaque joueur envoie la position de sa raquette.
Le format de paquet est simple : 7 valeurs numeriques (position raquette, position et vitesse de la balle, scores) emballees avec struct.pack.
Etape par etape
1. La classe NetworkPeer
La classe gere la connexion de maniere non-bloquante. Le serveur ecoute et accepte un client. Le client se connecte au serveur. Les deux envoient et recoivent des paquets.
python
PACKET_FMT = "!fffffii" # paddle_x, ball_x, ball_y,
# ball_vx, ball_vy, score1, score2
PACKET_SIZE = struct.calcsize(PACKET_FMT)
# Envoyer
pkt = struct.pack(PACKET_FMT, paddle_x, ball_x,
ball_y, ball_vx, ball_vy,
score1, score2)
sock.sendall(pkt)
# Recevoir
data = sock.recv(4096)
vals = struct.unpack(PACKET_FMT, data[:PACKET_SIZE])2. Architecture serveur/client
Le serveur est autoritaire : il execute la physique de la balle et envoie les resultats. Le client affiche les donnees recues et envoie seulement la position de sa raquette.
3. Mode non-bloquant
Les sockets sont en mode non-bloquant avec select.select() pour ne jamais bloquer la boucle de jeu. L'attente de connexion se fait aussi de maniere non-bloquante.
4. Score et fin de partie
Quand la balle depasse le bas ou le haut de l'ecran, le joueur adverse marque un point. Le premier a 5 points gagne la partie.
Lancement
Ouvrez deux terminaux et lancez :
python
python main.py --server # Terminal 1
python main.py --client # Terminal 2Points cles a retenir
Astuce : en mode non-bloquant, gerez toujours les exceptions BlockingIOError et les deconnexions.
Piege courant : oublier de gerer les paquets partiels. Le buffer de reception peut contenir un paquet incomplet qu'il faut accumuler.
Code complet
python
"""Brick Breaker - Chapter 20: Networking (2-Player over localhost)
Usage:
python main.py --server # Start as server (bottom paddle)
python main.py --client # Start as client (top paddle)
python main.py # Single-player mode (default)
In 2-player mode, one ball bounces between the two paddles.
Each player tries to prevent the ball from passing their side.
"""
import pygame
import sys
import math
import array
import random
import socket
import select
import struct
import threading
pygame.init()
pygame.mixer.init(frequency=44100, size=-16, channels=1, buffer=512)
GAME_WIDTH = 800
GAME_HEIGHT = 600
FPS = 60
BG_COLOR = (20, 20, 40)
PADDLE_WIDTH = 100
PADDLE_HEIGHT = 15
PADDLE_SPEED = 8
BALL_RADIUS = 8
BALL_BASE_SPEED = 300
BALL_SPEED_INCREMENT = 5
MAX_SCORE = 5 # first to 5 wins in 2-player mode
NET_PORT = 5555
NET_HOST = "127.0.0.1"
# Packet format: "!fffffii" = paddle_x(f), ball_x(f), ball_y(f),
# ball_vx(f), ball_vy(f), score1(i), score2(i)
PACKET_FMT = "!fffffii"
PACKET_SIZE = struct.calcsize(PACKET_FMT)
# Animation constants
TRAIL_LENGTH = 8
TRAIL_INTERVAL = 0.016
# Game states
STATE_WAITING = 0
STATE_PLAYING = 1
STATE_GAME_OVER = 2
# -- Sound generation -------------------------------------------------------
def generate_beep(frequency=440, duration_ms=100, volume=0.3):
sample_rate = 44100
n_samples = int(sample_rate * duration_ms / 1000)
buf = array.array("h", [0] * n_samples)
max_amp = int(32767 * volume)
for i in range(n_samples):
t = i / sample_rate
value = int(max_amp * math.sin(2 * math.pi * frequency * t))
fade_samples = min(500, n_samples // 4)
if i >= n_samples - fade_samples:
value = int(value * (n_samples - i) / fade_samples)
buf[i] = value
return pygame.mixer.Sound(buffer=buf)
snd_paddle = generate_beep(frequency=500, duration_ms=60, volume=0.3)
snd_wall = generate_beep(frequency=300, duration_ms=50, volume=0.2)
snd_score = generate_beep(frequency=150, duration_ms=300, volume=0.4)
snd_win = generate_beep(frequency=880, duration_ms=400, volume=0.4)
sound_muted = False
master_volume = 0.5
def play_sound(sound):
if not sound_muted:
sound.set_volume(master_volume)
sound.play()
# -- BallTrail animation ---------------------------------------------------
class BallTrail:
def __init__(self):
self.positions = []
self.timer = 0.0
def update(self, dt, bx, by):
self.timer += dt
if self.timer >= TRAIL_INTERVAL:
self.timer -= TRAIL_INTERVAL
self.positions.append((bx, by))
if len(self.positions) > TRAIL_LENGTH:
self.positions.pop(0)
def draw(self, surface):
count = len(self.positions)
for i, (x, y) in enumerate(self.positions):
ratio = (i + 1) / count if count > 0 else 0
alpha = int(80 * ratio)
radius = max(1, int(BALL_RADIUS * ratio * 0.8))
ts = pygame.Surface((radius * 2, radius * 2), pygame.SRCALPHA)
pygame.draw.circle(ts, (100, 100, 255, alpha),
(radius, radius), radius)
surface.blit(ts, (int(x) - radius, int(y) - radius))
def reset(self):
self.positions.clear()
self.timer = 0.0
# -- Network helper ---------------------------------------------------------
class NetworkPeer:
"""Non-blocking network peer for 2-player communication.
The server listens and accepts one client.
The client connects to the server.
Both send/receive paddle positions and game state.
"""
def __init__(self, is_server):
self.is_server = is_server
self.connected = False
self.sock = None
self.peer_sock = None
self.recv_buffer = b""
# Latest received data
self.remote_paddle_x = GAME_WIDTH // 2
self.remote_ball_x = GAME_WIDTH // 2
self.remote_ball_y = GAME_HEIGHT // 2
self.remote_ball_vx = 0.0
self.remote_ball_vy = 0.0
self.remote_score1 = 0
self.remote_score2 = 0
def start(self):
"""Begin listening (server) or connecting (client) in background."""
if self.is_server:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind((NET_HOST, NET_PORT))
self.sock.listen(1)
self.sock.setblocking(False)
else:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setblocking(False)
try:
self.sock.connect((NET_HOST, NET_PORT))
except BlockingIOError:
pass # connection in progress
def update(self):
"""Non-blocking check for new connections or data."""
if not self.connected:
if self.is_server:
# Try to accept
try:
ready, _, _ = select.select([self.sock], [], [], 0)
if ready:
self.peer_sock, _ = self.sock.accept()
self.peer_sock.setblocking(False)
self.connected = True
except Exception:
pass
else:
# Check if connection completed
try:
_, writable, _ = select.select([], [self.sock], [], 0)
if writable:
err = self.sock.getsockopt(socket.SOL_SOCKET,
socket.SO_ERROR)
if err == 0:
self.peer_sock = self.sock
self.peer_sock.setblocking(False)
self.connected = True
except Exception:
pass
return
# Read available data
try:
ready, _, _ = select.select([self.peer_sock], [], [], 0)
if ready:
data = self.peer_sock.recv(4096)
if not data:
self.connected = False
return
self.recv_buffer += data
# Process complete packets
while len(self.recv_buffer) >= PACKET_SIZE:
pkt = self.recv_buffer[:PACKET_SIZE]
self.recv_buffer = self.recv_buffer[PACKET_SIZE:]
vals = struct.unpack(PACKET_FMT, pkt)
(self.remote_paddle_x, self.remote_ball_x,
self.remote_ball_y, self.remote_ball_vx,
self.remote_ball_vy, self.remote_score1,
self.remote_score2) = vals
except Exception:
pass
def send_state(self, paddle_x, ball_x, ball_y, ball_vx, ball_vy,
score1, score2):
if not self.connected:
return
pkt = struct.pack(PACKET_FMT, paddle_x, ball_x, ball_y,
ball_vx, ball_vy, score1, score2)
try:
self.peer_sock.sendall(pkt)
except Exception:
self.connected = False
def close(self):
try:
if self.peer_sock:
self.peer_sock.close()
if self.sock and self.sock != self.peer_sock:
self.sock.close()
except Exception:
pass
# -- Game classes -----------------------------------------------------------
class NetPaddle:
"""A paddle for the 2-player mode (not a sprite for simplicity)."""
def __init__(self, y, is_local=True):
self.x = float(GAME_WIDTH // 2)
self.y = y
self.width = PADDLE_WIDTH
self.height = PADDLE_HEIGHT
self.is_local = is_local
@property
def rect(self):
return pygame.Rect(int(self.x) - self.width // 2,
int(self.y),
self.width, self.height)
def update_local(self, mouse_x=None):
keys = pygame.key.get_pressed()
if keys[pygame.K_LEFT] or keys[pygame.K_a]:
self.x -= PADDLE_SPEED
if keys[pygame.K_RIGHT] or keys[pygame.K_d]:
self.x += PADDLE_SPEED
self.x = max(self.width // 2, min(GAME_WIDTH - self.width // 2,
self.x))
def draw(self, surface, color=(220, 220, 220)):
r = self.rect
pygame.draw.rect(surface, color, r)
# Highlight
pygame.draw.line(surface, (255, 255, 255),
(r.left + 2, r.top + 1), (r.right - 2, r.top + 1))
class NetBall:
"""Ball for 2-player mode."""
def __init__(self):
self.x = float(GAME_WIDTH // 2)
self.y = float(GAME_HEIGHT // 2)
self.vx = BALL_BASE_SPEED * 0.7
self.vy = BALL_BASE_SPEED * 0.7
self.speed = BALL_BASE_SPEED
def reset(self, direction=1):
"""Reset ball. direction: 1 = downward, -1 = upward."""
self.x = float(GAME_WIDTH // 2)
self.y = float(GAME_HEIGHT // 2)
self.speed = BALL_BASE_SPEED
angle = math.radians(60 * direction)
self.vx = self.speed * math.cos(angle)
self.vy = self.speed * math.sin(angle)
def update(self, dt):
self.x += self.vx * dt
self.y += self.vy * dt
# Wall bounces (left, right only)
if self.x - BALL_RADIUS <= 0:
self.x = BALL_RADIUS
self.vx = abs(self.vx)
play_sound(snd_wall)
if self.x + BALL_RADIUS >= GAME_WIDTH:
self.x = GAME_WIDTH - BALL_RADIUS
self.vx = -abs(self.vx)
play_sound(snd_wall)
def bounce_paddle(self, paddle_rect, going_down):
"""Check collision with a paddle. going_down: True if checking
bottom paddle, False for top paddle."""
ball_rect = pygame.Rect(int(self.x) - BALL_RADIUS,
int(self.y) - BALL_RADIUS,
BALL_RADIUS * 2, BALL_RADIUS * 2)
if not ball_rect.colliderect(paddle_rect):
return False
if going_down and self.vy > 0:
self.y = paddle_rect.top - BALL_RADIUS
hit_pos = (self.x - paddle_rect.left) / PADDLE_WIDTH
angle = 150 - hit_pos * 120
rad = math.radians(angle)
self.speed += BALL_SPEED_INCREMENT
self.vx = self.speed * math.cos(rad)
self.vy = -abs(self.speed * math.sin(rad))
return True
if not going_down and self.vy < 0:
self.y = paddle_rect.bottom + BALL_RADIUS
hit_pos = (self.x - paddle_rect.left) / PADDLE_WIDTH
angle = 150 - hit_pos * 120
rad = math.radians(angle)
self.speed += BALL_SPEED_INCREMENT
self.vx = self.speed * math.cos(rad)
self.vy = abs(self.speed * math.sin(rad))
return True
return False
def draw(self, surface):
pygame.draw.circle(surface, (255, 255, 255),
(int(self.x), int(self.y)), BALL_RADIUS)
# Glow
gs = pygame.Surface((BALL_RADIUS * 4, BALL_RADIUS * 4),
pygame.SRCALPHA)
pygame.draw.circle(gs, (100, 100, 255, 80),
(BALL_RADIUS * 2, BALL_RADIUS * 2),
BALL_RADIUS * 2)
surface.blit(gs, (int(self.x) - BALL_RADIUS * 2,
int(self.y) - BALL_RADIUS * 2))
# -- Parse command-line arguments -------------------------------------------
mode = "single" # default: single-player
for arg in sys.argv[1:]:
if arg == "--server":
mode = "server"
elif arg == "--client":
mode = "client"
# -- Game setup -------------------------------------------------------------
screen = pygame.display.set_mode((GAME_WIDTH, GAME_HEIGHT))
if mode == "single":
pygame.display.set_caption("Brick Breaker - Chapter 20 (Single Player)")
elif mode == "server":
pygame.display.set_caption("Brick Breaker - Ch20 (Server - Bottom)")
else:
pygame.display.set_caption("Brick Breaker - Ch20 (Client - Top)")
clock = pygame.time.Clock()
hud_font = pygame.font.SysFont("monospace", 18)
big_font = pygame.font.SysFont("monospace", 48, bold=True)
medium_font = pygame.font.SysFont("monospace", 24)
small_font = pygame.font.SysFont("monospace", 14)
# 2-player mode objects
if mode != "single":
net = NetworkPeer(is_server=(mode == "server"))
net.start()
bottom_paddle = NetPaddle(GAME_HEIGHT - 40, is_local=(mode == "server"))
top_paddle = NetPaddle(25, is_local=(mode == "client"))
net_ball = NetBall()
ball_trail = BallTrail()
score_bottom = 0 # server's score
score_top = 0 # client's score
net_state = STATE_WAITING
net_ball.reset(direction=1)
# Single-player mode uses the standard brick breaker (simplified here)
else:
# --- Single-player imports from ch19 pattern (simplified) ---
from collections import namedtuple
ball_trail = BallTrail()
class SPPaddle:
def __init__(self):
self.x = float(GAME_WIDTH // 2)
self.y = GAME_HEIGHT - 40
self.width = PADDLE_WIDTH
self.height = PADDLE_HEIGHT
@property
def rect(self):
return pygame.Rect(int(self.x) - self.width // 2,
int(self.y),
self.width, self.height)
def update(self):
keys = pygame.key.get_pressed()
if keys[pygame.K_LEFT] or keys[pygame.K_a]:
self.x -= PADDLE_SPEED
if keys[pygame.K_RIGHT] or keys[pygame.K_d]:
self.x += PADDLE_SPEED
self.x = max(self.width // 2,
min(GAME_WIDTH - self.width // 2, self.x))
def draw(self, surface):
r = self.rect
pygame.draw.rect(surface, (220, 220, 220), r)
pygame.draw.line(surface, (255, 255, 255),
(r.left + 2, r.top + 1),
(r.right - 2, r.top + 1))
sp_paddle = SPPaddle()
sp_ball = NetBall()
sp_ball.reset(direction=-1)
# Simple bricks
sp_bricks = []
for row in range(6):
hue = int(360 * row / 6)
c = pygame.Color(0)
c.hsva = (hue, 100, 100, 100)
color = (c.r, c.g, c.b)
for col in range(10):
bx = 30 + col * 75
by = 50 + row * 25
sp_bricks.append({"rect": pygame.Rect(bx, by, 70, 20),
"color": color, "alive": True})
sp_score = 0
sp_lives = 3
sp_state = STATE_PLAYING
# -- Main loop --------------------------------------------------------------
running = True
while running:
dt = clock.tick(FPS) / 1000.0
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
if event.type == pygame.KEYDOWN:
if event.key == pygame.K_ESCAPE:
running = False
if event.key == pygame.K_F1:
sound_muted = not sound_muted
if mode != "single":
# ===== 2-PLAYER MODE =====
net.update()
if net_state == STATE_WAITING:
if net.connected:
net_state = STATE_PLAYING
net_ball.reset(direction=1)
score_bottom = 0
score_top = 0
elif net_state == STATE_PLAYING:
# Update local paddle
if mode == "server":
bottom_paddle.update_local()
top_paddle.x = net.remote_paddle_x
else:
top_paddle.update_local()
bottom_paddle.x = net.remote_paddle_x
# Server is authoritative over the ball
if mode == "server":
net_ball.update(dt)
ball_trail.update(dt, net_ball.x, net_ball.y)
# Bounce off bottom paddle (server's)
if net_ball.bounce_paddle(bottom_paddle.rect, True):
play_sound(snd_paddle)
# Bounce off top paddle (client's)
if net_ball.bounce_paddle(top_paddle.rect, False):
play_sound(snd_paddle)
# Scoring
if net_ball.y - BALL_RADIUS > GAME_HEIGHT:
# Bottom missed -> top scores
score_top += 1
play_sound(snd_score)
net_ball.reset(direction=-1)
ball_trail.reset()
elif net_ball.y + BALL_RADIUS < 0:
# Top missed -> bottom scores
score_bottom += 1
play_sound(snd_score)
net_ball.reset(direction=1)
ball_trail.reset()
if score_bottom >= MAX_SCORE or score_top >= MAX_SCORE:
net_state = STATE_GAME_OVER
# Send state to client
net.send_state(bottom_paddle.x, net_ball.x, net_ball.y,
net_ball.vx, net_ball.vy,
score_bottom, score_top)
else:
# Client receives ball state from server
net_ball.x = net.remote_ball_x
net_ball.y = net.remote_ball_y
net_ball.vx = net.remote_ball_vx
net_ball.vy = net.remote_ball_vy
score_bottom = net.remote_score1
score_top = net.remote_score2
ball_trail.update(dt, net_ball.x, net_ball.y)
if score_bottom >= MAX_SCORE or score_top >= MAX_SCORE:
net_state = STATE_GAME_OVER
# Send paddle position to server
net.send_state(top_paddle.x, 0, 0, 0, 0, 0, 0)
if not net.connected:
net_state = STATE_GAME_OVER
# -- Draw 2-player --
screen.fill(BG_COLOR)
if net_state == STATE_WAITING:
wait_text = big_font.render("WAITING...", True, (255, 200, 50))
screen.blit(wait_text,
(GAME_WIDTH // 2 - wait_text.get_width() // 2,
GAME_HEIGHT // 2 - 50))
if mode == "server":
info = medium_font.render(
"Server listening on port {}".format(NET_PORT),
True, (150, 150, 150))
else:
info = medium_font.render(
"Connecting to {}:{}...".format(NET_HOST, NET_PORT),
True, (150, 150, 150))
screen.blit(info, (GAME_WIDTH // 2 - info.get_width() // 2,
GAME_HEIGHT // 2 + 20))
else:
# Draw centre line
for x in range(0, GAME_WIDTH, 20):
pygame.draw.rect(screen, (50, 50, 70),
(x, GAME_HEIGHT // 2 - 1, 10, 2))
# Paddles
bottom_paddle.draw(screen, (100, 200, 255))
top_paddle.draw(screen, (255, 150, 100))
# Ball
ball_trail.draw(screen)
net_ball.draw(screen)
# Scores
bs = big_font.render(str(score_bottom), True, (100, 200, 255))
screen.blit(bs, (GAME_WIDTH // 2 - bs.get_width() // 2,
GAME_HEIGHT // 2 + 30))
ts = big_font.render(str(score_top), True, (255, 150, 100))
screen.blit(ts, (GAME_WIDTH // 2 - ts.get_width() // 2,
GAME_HEIGHT // 2 - 70))
# Labels
bl = small_font.render("Server (bottom)", True, (100, 200, 255))
screen.blit(bl, (10, GAME_HEIGHT - 20))
tl = small_font.render("Client (top)", True, (255, 150, 100))
screen.blit(tl, (10, 5))
if net_state == STATE_GAME_OVER:
overlay = pygame.Surface((GAME_WIDTH, GAME_HEIGHT),
pygame.SRCALPHA)
overlay.fill((0, 0, 0, 150))
screen.blit(overlay, (0, 0))
if not net.connected:
msg = "DISCONNECTED"
color = (255, 150, 50)
elif score_bottom >= MAX_SCORE:
msg = "SERVER WINS!" if mode == "server" else "YOU LOSE!"
color = (100, 255, 100) if mode == "server" else (
255, 60, 60)
else:
msg = "CLIENT WINS!" if mode == "client" else "YOU LOSE!"
color = (100, 255, 100) if mode == "client" else (
255, 60, 60)
gt = big_font.render(msg, True, color)
screen.blit(gt,
(GAME_WIDTH // 2 - gt.get_width() // 2,
GAME_HEIGHT // 2 - 30))
hint = medium_font.render("Press ESC to quit", True,
(200, 200, 200))
screen.blit(hint,
(GAME_WIDTH // 2 - hint.get_width() // 2,
GAME_HEIGHT // 2 + 30))
else:
# ===== SINGLE-PLAYER MODE =====
if sp_state == STATE_PLAYING:
sp_paddle.update()
sp_ball.update(dt)
ball_trail.update(dt, sp_ball.x, sp_ball.y)
# Top wall bounce
if sp_ball.y - BALL_RADIUS <= 0:
sp_ball.y = BALL_RADIUS
sp_ball.vy = abs(sp_ball.vy)
play_sound(snd_wall)
# Paddle bounce
if sp_ball.bounce_paddle(sp_paddle.rect, True):
play_sound(snd_paddle)
# Brick collision
ball_rect = pygame.Rect(int(sp_ball.x) - BALL_RADIUS,
int(sp_ball.y) - BALL_RADIUS,
BALL_RADIUS * 2, BALL_RADIUS * 2)
for brick in sp_bricks:
if brick["alive"] and ball_rect.colliderect(brick["rect"]):
brick["alive"] = False
sp_ball.vy = -sp_ball.vy
sp_score += 10
play_sound(snd_paddle)
# Fall off bottom
if sp_ball.y - BALL_RADIUS > GAME_HEIGHT:
sp_lives -= 1
if sp_lives <= 0:
sp_state = STATE_GAME_OVER
else:
sp_ball.reset(direction=-1)
ball_trail.reset()
play_sound(snd_score)
# Win check
if not any(b["alive"] for b in sp_bricks):
sp_state = STATE_GAME_OVER
# Draw single-player
screen.fill(BG_COLOR)
for brick in sp_bricks:
if brick["alive"]:
pygame.draw.rect(screen, brick["color"], brick["rect"])
highlight = tuple(min(255, c + 50) for c in brick["color"])
pygame.draw.line(screen, highlight,
(brick["rect"].left + 1,
brick["rect"].top + 1),
(brick["rect"].right - 2,
brick["rect"].top + 1))
ball_trail.draw(screen)
sp_ball.draw(screen)
sp_paddle.draw(screen)
# HUD
sc = hud_font.render("Score: {}".format(sp_score), True,
(255, 255, 100))
screen.blit(sc, (10, 10))
lv = hud_font.render("Lives: {}".format(sp_lives), True,
(255, 100, 100))
screen.blit(lv, (GAME_WIDTH - lv.get_width() - 10, 10))
info = small_font.render(
"Run with --server or --client for 2-player mode",
True, (100, 100, 100))
screen.blit(info, (10, GAME_HEIGHT - 20))
if sp_state == STATE_GAME_OVER:
overlay = pygame.Surface((GAME_WIDTH, GAME_HEIGHT),
pygame.SRCALPHA)
overlay.fill((0, 0, 0, 150))
screen.blit(overlay, (0, 0))
if sp_lives <= 0:
msg = "GAME OVER"
color = (255, 60, 60)
else:
msg = "YOU WIN!"
color = (100, 255, 100)
gt = big_font.render(msg, True, color)
screen.blit(gt,
(GAME_WIDTH // 2 - gt.get_width() // 2,
GAME_HEIGHT // 2 - 30))
pygame.display.flip()
# Cleanup
if mode != "single":
net.close()
pygame.quit()
sys.exit()
Telechargement
Telecharger le code source de ce chapitre : brick-breaker-ch20.zip