What you will learn
In this chapter we add a 2-player networked mode over localhost. One instance launched with --server controls the bottom paddle, the other with --client controls the top paddle. The ball bounces between them.
The concept
Networking uses Python's socket module with TCP. The server is authoritative over the ball: it computes position and sends it to the client. Each player sends their paddle position.
The packet format is simple: 7 numeric values (paddle position, ball position and velocity, scores) packed with struct.pack.
Step by step
1. The NetworkPeer class
The class handles the connection in non-blocking mode. The server listens and accepts one client. The client connects to the server. Both send and receive packets.
python
PACKET_FMT = "!fffffii" # paddle_x, ball_x, ball_y,
# ball_vx, ball_vy, score1, score2
PACKET_SIZE = struct.calcsize(PACKET_FMT)
# Send
pkt = struct.pack(PACKET_FMT, paddle_x, ball_x,
ball_y, ball_vx, ball_vy,
score1, score2)
sock.sendall(pkt)
# Receive
data = sock.recv(4096)
vals = struct.unpack(PACKET_FMT, data[:PACKET_SIZE])2. Server/client architecture
The server is authoritative: it runs the ball physics and sends the results. The client displays the received data and only sends its paddle position.
3. Non-blocking mode
Sockets are set to non-blocking mode using select.select() so the game loop never stalls. Connection waiting is also non-blocking.
4. Scoring and game end
When the ball passes the bottom or top of the screen, the opposing player scores a point. First to 5 points wins the match.
How to launch
Open two terminals and run:
python
python main.py --server # Terminal 1
python main.py --client # Terminal 2Key takeaways
Tip: in non-blocking mode, always handle BlockingIOError exceptions and disconnections.
Common pitfall: forgetting to handle partial packets. The receive buffer may contain an incomplete packet that must be accumulated.
Complete code
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()
Download
Download the source code for this chapter: brick-breaker-ch20.zip