PyQt6 + 소켓으로 만든 2인용 네트워크 오목게임
개요
Python과 PyQt6, 그리고 소켓 통신으로 두 플레이어가 네트워크를 통해 즐길 수 있는 오목 게임 입니다.
소스코드 전체를 구조별로 설명하니, 처음 보는 분도 흐름을 따라가는데 문제없으리라 생각합니다.
코드를 짜는 즐거움은 직접 😁, 일부 설명을 돕는 그림, 표는 클로드AI를 활용하였습니다.
전체 아키텍쳐 구조
서버에 2명의 클라이언트가 접속하면 오목게임이 시작되는 구조.
1. 서버 실행
2. 클라이언트 2개 실행
3. 서버가 2명의 클라이언트 접속을 감지하면 게임 시작
4. 먼저 접속한 순서에 따라 흑, 백 결정
5. 접속, 돌 놓기, 승부결과를 의미하는 프로토콜 구조를 상호간 전송
|
| [클로드AI 생성 이미지] |
개발환경
- VS Code, Python 3.12 64bit
- PyQt6 6.10.0
- Windows 11
1. 전체 파일 구조
총 6개의 파일로 구성, Github Link
전체 흐름 요약
서버를 실행하고, 두 클라이언트가 접속하면 서버가 각각에게 흑돌/백돌을 배정.
(클라이언트먼저 실행해도 무관)
이후 돌을 놓을 때마다 클라이언트 → 서버 → 전체 브로드캐스트 방식으로 동기화.
2. protocol.py, 패킷 설계
가장 먼저 살펴볼 파일입니다.
두 프로세스가 같은 패킷 구조를 공유해야 하므로, 이 모듈은 클라이언트와 서버 양쪽에서 공통으로 사용됩니다.
import struct
# type 정의
# 1 : 모든 클라이언트가 접속됨(흑돌)
# 2 : 모든 클라이언트가 접속됨(백돌)
# 3 : 흑돌을 놓았다
# 4 : 백돌을 놓았다
# 5 : 흑돌승
# 6 : 백돌승
# 7 : 무승부
# 8 : 흑재설정
# 9 : 백재설정
def makePt(t, r=0, c=0):
header = struct.pack("<HB", 5, t)
body = struct.pack('<BB', r, c)
packet = header + body
return packet
패킷은 총 5 byte 입니다.
struct.pack의 포맷 문자열
'<'는 리틀엔디안을 의미하고,
'H'는 unsigned short(2 bytes),
'B'는 unsigned char(1 byte)입니다.
타입(t) 값 정의
간단하지만 이 7가지 타입으로 게임의 모든 상태를 표현합니다.
코드 간단화를 위해 TCP패킷 단편화는 고려하지 않았습니다.
3. server_main.py, 서버 UI
서버는 PyQt6 GUI로 만들어졌습니다.
IP/PORT를 입력하고 Open 버튼을 누르면 소켓 서버가 시작되고, 접속한 클라이언트 목록이 테이블로 표시됩니다.
from PyQt6.QtWidgets import (QApplication, QWidget, QHBoxLayout,
QVBoxLayout, QPushButton, QLineEdit,
QLabel, QTableWidget, QMessageBox,
QTableWidgetItem)
from server_socket import Server
import sys
import socket as sk
import datetime
class Window(QWidget):
PORT = 5614
def __init__(self):
super().__init__()
self.initUi()
self.resize(640, 320)
self.server = Server(self)
def initUi(self):
hbox = QHBoxLayout()
lb_ip = QLabel('IP')
lb_port = QLabel('PORT')
self.le_ip = QLineEdit()
self.le_port = QLineEdit()
self.pb = QPushButton('Open')
hbox.addWidget(lb_ip)
hbox.addWidget(self.le_ip)
hbox.addWidget(lb_port)
hbox.addWidget(self.le_port)
hbox.addWidget(self.pb)
vbox = QVBoxLayout()
vbox.addLayout(hbox)
self.tw = QTableWidget()
label = ('IP', 'PORT', 'LOGIN TIME')
cols = len(label)
self.tw.setColumnCount(cols)
self.tw.setHorizontalHeaderLabels(label)
vbox.addWidget(self.tw)
self.setLayout(vbox)
# get ip
pc = sk.gethostname()
ip = sk.gethostbyname(pc)
self.le_ip.setText(ip)
self.le_port.setText(str(Window.PORT))
# toggle
self.pb.setCheckable(True)
# signal
self.pb.clicked.connect(self.onClick)
def onClick(self):
if self.pb.isChecked():
ip = self.le_ip.text()
port = int(self.le_port.text())
ok, msg = self.server.startServer(ip, port)
if ok:
self.pb.setText('Close')
else:
QMessageBox.critical(self, 'Error', msg, QMessageBox.StandardButton.Ok)
self.pb.setText('Open')
self.pb.setChecked(False)
else:
self.server.closeServer()
self.pb.setText('Open')
def onConnect(self, ip, port):
row = self.tw.rowCount()
self.tw.setRowCount(row+1)
item_ip = QTableWidgetItem(ip)
item_port = QTableWidgetItem(port)
self.tw.setItem(row, 0, item_ip)
self.tw.setItem(row, 1, item_port)
dt = datetime.datetime.now()
t = dt.strftime("%Y-%m-%d %H:%M:%S")
item_time = QTableWidgetItem(t)
self.tw.setItem(row, 2, item_time)
def onDisconnect(self, ip, port):
rows = self.tw.rowCount()
for i in range(rows):
cip = self.tw.item(i, 0).text()
cpt = self.tw.item(i, 1).text()
if ip==cip and port==cpt:
self.tw.removeRow(i)
break
def closeEvent(self, e):
self.server.closeServer()
if __name__ == '__main__':
app = QApplication(sys.argv)
w = Window()
w.show()
sys.exit(app.exec())
- Server 객체는 server_socket.py에서 실제 소켓 작업을 담당
- UI는 두 가지 신호(signal)를 수신해 접속/해제 이벤트를 테이블에 반영
4. server_socket.py, 서버 소켓 핵심 로직
서버의 핵심입니다.
클라이언트 접속 관리, 차례 검증, 승패 판정, 브로드캐스트가 모두 여기 있습니다.
import socket as sk
from PyQt6.QtCore import QObject, pyqtSignal
from threading import Thread
import os
curr_path = os.path.abspath('.')
#print(curr_path)
import sys
sys.path.append(curr_path)
from protocol import makePt
import struct
import time
class Server(QObject):
conn_signal = pyqtSignal(str, str)
disconn_signal = pyqtSignal(str, str)
def __init__(self, w):
super().__init__()
self.parent = w
self.clients = []
# 접속자수
self.maxc = 2
# 처음은 흑돌
self.dol = 1
# 전체 돌 정보 저장(19*19개), 0:돌없음, 1:흑돌, 2:백돌
self.line = 19
self.map = [ [0 for _ in range(self.line)] for _ in range(self.line) ]
#print(self.map)
# signal
self.conn_signal.connect(self.parent.onConnect)
self.disconn_signal.connect(self.parent.onDisconnect)
def startServer(self, ip, port):
self.sock = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
try:
self.sock.bind((ip, port))
except Exception as e:
print(e)
return False, str(e)
else:
self.listen_thread = Thread(target=self.listenThread, args=(self.sock, ip, port))
self.listen_thread.start()
return True, ''
def closeServer(self):
try:
for client, addr in self.clients:
client.close()
self.sock.close()
except Exception as e:
print(e)
def listenThread(self, sock, ip, port):
sock.listen(5)
while True:
try:
client, addr = sock.accept()
except Exception as e:
print(e)
break
else:
if self.maxc > len(self.clients):
self.clients.append((client, addr))
client_thread = Thread(target=self.clientThread, args=(client, addr))
client_thread.start()
self.conn_signal.emit(str(addr[0]), str(addr[1]))
print(f'connect client, {len(self.clients)} : {addr[0]}, {addr[1]}')
# 모든 클라이언트가 접속완료
if self.maxc == len(self.clients):
i = 1
for c, a in self.clients:
pk = makePt(i)
c.send(pk)
i+=1
else:
print('max client!')
print('terminated listen thread')
def clientThread(self, client, addr):
while True:
try:
data = client.recv(1024)
except Exception as e:
print(e)
break
else:
if not data:
break
# 정상 수신
size, t, r, c = struct.unpack('<HBBB', data)
print(size, t, r, c)
# 차례가 맞는지 확인
if self.dol == t and self.map[r][c]==0:
self.map[r][c] = t
# 흑,돌 순서변경
if t == 1:
self.dol = 2
else:
self.dol = 1
#print(self.map)
# 클라이언트로 전송
prot = makePt(t+2, r, c)
self.broadcast(prot)
# 판정
win = self.checkWinner()
if win!=0:
prot = makePt(win, 0, 0)
self.broadcast(prot)
# 판정 후 재설정
self.dol = 1
self.map = [ [0 for _ in range(self.line)] for _ in range(self.line) ]
# 지연
time.sleep(0.1)
# 모든 클라이언트 접속 끊기
for cs, ca in self.clients:
cs.close()
self.disconn_signal.emit( str(ca[0]), str(ca[1]) )
# 클라이언트 소켓 리스트 전체 삭제
self.clients.clear()
for cs, ca in self.clients:
if cs==client and ca==addr:
self.clients.remove((client, addr))
self.disconn_signal.emit( str(addr[0]), str(addr[1]) )
break
print(f'terminated client thread, {len(self.clients)}')
def broadcast(self, prot):
for c, a in self.clients:
c.send(prot)
def checkWinner(self):
# 0: 진행중, 5: 흑돌승, 6:백돌승, 7:무승부
zcnt = 0
for i in range(self.line):
for j in range(self.line):
# 0체크
if self.map[i][j]==0:
zcnt+=1
# 흑돌 가로판정
if j<=14:
if (self.map[i][j]==1
and self.map[i][j+1]==1
and self.map[i][j+2]==1
and self.map[i][j+3]==1
and self.map[i][j+4]==1):
return 5
# 흑돌 세로판정
if i<=14:
if (self.map[i][j]==1
and self.map[i+1][j]==1
and self.map[i+2][j]==1
and self.map[i+3][j]==1
and self.map[i+4][j]==1):
return 5
# 흑돌 좌->우 대각판정
if i<=14 and j<=14:
if (self.map[i][j]==1
and self.map[i+1][j+1]==1
and self.map[i+2][j+2]==1
and self.map[i+3][j+3]==1
and self.map[i+4][j+4]==1):
return 5
# 흑돌 우->좌 대각판정
if i<=14 and j>=4:
if (self.map[i][j]==1
and self.map[i+1][j-1]==1
and self.map[i+2][j-2]==1
and self.map[i+3][j-3]==1
and self.map[i+4][j-4]==1):
return 5
# 백돌 가로판정
if j<=14:
if (self.map[i][j]==2
and self.map[i][j+1]==2
and self.map[i][j+2]==2
and self.map[i][j+3]==2
and self.map[i][j+4]==2):
return 6
# 백돌 세로판정
if i<=14:
if (self.map[i][j]==2
and self.map[i+1][j]==2
and self.map[i+2][j]==2
and self.map[i+3][j]==2
and self.map[i+4][j]==2):
return 6
# 백돌 좌->우 대각판정
if i<=14 and j<=14:
if (self.map[i][j]==2
and self.map[i+1][j+1]==2
and self.map[i+2][j+2]==2
and self.map[i+3][j+3]==2
and self.map[i+4][j+4]==2):
return 6
# 백돌 우->좌 대각판정
if i<=14 and j>=4:
if (self.map[i][j]==2
and self.map[i+1][j-1]==2
and self.map[i+2][j-2]==2
and self.map[i+3][j-3]==2
and self.map[i+4][j-4]==2):
return 6
# 무승부 or 진행중
if zcnt == 0:
return 7
return 0
- 리슨 스레드 — 접속 대기
- sock.accept()는 블로킹 호출, 반드시 별도 스레드에서 실행해야 GUI 멈춤X
- 클라이언트 스레드 — 수신 및 처리
- 서버가 직접 차례와 유효성을 검증, 클라이언트가 임의로 상태 조작X
- 승패 판정 — checkWinner()
- 코드는 길지만 이해가 쉽도록 3중 for 사용X
- 가로, 세로, 우하향 대각, 우상향 대각 네 방향을 흑/백 각각 검사
- j <= 14 조건은 배열 범위를 초과하지 않게 막아주는 가드
5. client_socket.py, 클라이언트 소켓
클라이언트측 소켓 프로그래밍 구현부입니다.
주로 서버 접속, 연결끊기, 이벤트 송수신 패킷을 Send/Recv 처리하는 역할을 담당합니다.
import socket as sk
from threading import Thread
import struct
from PyQt6.QtCore import QObject, pyqtSignal
class Client(QObject):
IP = '192.168.0.2'
PORT = 5614
ready_sig = pyqtSignal(int)
dol_sig = pyqtSignal(int, int, int)
end_sig = pyqtSignal(int)
def __init__(self, g):
super().__init__()
self.parent = g
# 소켓 생성
self.sock = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
# 신호연결
self.ready_sig.connect(self.parent.onReady)
self.dol_sig.connect(self.parent.onDol)
self.end_sig.connect(self.parent.onEnd)
def connectSv(self):
try:
self.sock.connect( (Client.IP, Client.PORT) )
except Exception as e:
print(e)
return False
else:
# 데이터 수신 쓰레드
self.t = Thread(target=self.tf)
self.t.start()
return True
def disconnectSv(self):
try:
self.sock.close()
except Exception as e:
print(e)
def send(self, data):
try:
self.sock.send(data)
except Exception as e:
print(e)
def tf(self):
while True:
try:
data = self.sock.recv(1024)
except Exception as e:
print(e)
break
else:
if not data:
break
size, t, r, c = struct.unpack('<HBBB', data)
print(size, t, r, c)
if t==1 or t==2:
self.ready_sig.emit( t )
elif t==3 or t==4:
self.dol_sig.emit(t, r, c)
elif t==5 or t==6 or t==7:
self.end_sig.emit(t)
- PyQt6의 pyqtSignal을 사용하는 이유는, 소켓 수신이 별도 스레드에서 일어나기 때문
- Qt에서 GUI 업데이트는 반드시 메인 스레드에서
- 따라서 signal/slot 메커니즘으로 스레드 간 안전하게 통신
6. client_game.py, 게임 로직 + 그리기
오목게임의 핵심로직, 돌놓기, 승패 판정 등 게임의 진행과 화면 렌더링을 담당합니다.
from threading import Thread
from client_socket import Client
from PyQt6.QtCore import QRectF, QPointF, pyqtSignal, QObject
from PyQt6.QtGui import QPainter, QColor, QBrush, QColorConstants
import os
curr_path = os.path.abspath('.')
#print(curr_path)
import sys
sys.path.append(curr_path)
from protocol import makePt
class Game(QObject):
update_sig = pyqtSignal()
end_sig = pyqtSignal(int)
def __init__(self, w):
super().__init__()
self.parent = w
self.rect = w.rect()
# 화면 새로그리기 신호
self.update_sig.connect(w.update)
self.end_sig.connect(w.onEnd)
# 준비X:0, 흑:1, 백:2
self.dol = 0
self.line = 19
self.inrect = QRectF(self.rect)
gap = 20
self.inrect.adjust(gap, gap, -gap, -gap)
self.gap = gap
# 한 칸 크기
self.size = self.inrect.width() / (self.line-1)
# 소캣클래스 생성
self.client = Client(self)
# 전체 돌 정보 저장(19*19개), 0:돌없음, 1:흑돌, 2:백돌
self.map = [ [0 for _ in range(self.line)] for _ in range(self.line) ]
# 서버접속용 쓰레드 생성
self.run = True
self.t1 = Thread(target=self.cont)
self.t1.start()
def draw(self, qp):
qp.setRenderHint(QPainter.RenderHint.Antialiasing)
# 바둑판 그리기
if self.dol != 0:
c = QColor(158,115,35)
b = QBrush(c)
qp.setBrush(b)
qp.drawRect(self.inrect)
x1 = self.inrect.left()
y1 = self.inrect.top()
x2 = self.inrect.right()
y2 = self.inrect.top()
x3 = self.inrect.left()
y3 = self.inrect.bottom()
# 줄 긋기
for i in range(self.line):
qp.drawLine(QPointF(x1, y1+i*self.size), QPointF(x2, y2+i*self.size))
qp.drawLine(QPointF(x1+i*self.size, y1), QPointF(x3+i*self.size, y3))
# 바둑돌 그리기
bb = QBrush(QColorConstants.Black)
wb = QBrush(QColorConstants.White)
for i in range(self.line):
for j in range(self.line):
if self.map[i][j]==0:
continue # 다음 반복으로 바로 이동
elif self.map[i][j]==1:
qp.setBrush(bb)
elif self.map[i][j]==2:
qp.setBrush(wb)
y = (i*self.size)-(self.size/2) + self.gap
x = (j*self.size)-(self.size/2) + self.gap
rect = QRectF(x, y, self.size, self.size)
qp.drawEllipse( rect )
def cont(self):
while self.run:
# 접속 시도
if self.client.connectSv():
print('success!')
break
def onReady(self, t):
self.dol = t
self.update_sig.emit()
# 차례표시
txt = ('선공', '후공')
self.parent.setWindowTitle(f'차례 : {txt[t-1]}')
def mouseDown(self, pt):
#print(pt)
pt = QPointF(pt)
if self.inrect.contains(pt):
r = int(pt.y() // self.size)
c = int(pt.x() // self.size)
print(self.dol, r, c)
# 서버로 전송
prot = makePt(self.dol, r, c)
self.client.send(prot)
else:
print('out')
def onDol(self, t, r, c):
# 전체 돌 정보 저장(19*19개), 0:돌없음, 1:흑돌, 2:백돌
dol = t-2
self.map[r][c] = dol
self.update_sig.emit()
# 차례표시
txt = ('백돌', '흑돌')
self.parent.setWindowTitle(f'차례 : {txt[dol-1]}')
def onEnd(self, t):
self.end_sig.emit(t)
def onDisc(self):
self.run = False
self.client.disconnectSv()
def onReset(self):
self.dol = 0
self.map = [ [0 for _ in range(self.line)] for _ in range(self.line) ]
prot = makePt(self.dol+7, 0,0)
self.client.send(prot)
self.update_sig.emit()
- 바둑판 그리기
- 돌의 위치는 격자 교차점을 기준으로, 크기는 격자 한 칸(self.size)에 꽉 차도록 계산
- 마우스 클릭 처리
- 클릭 좌표를 격자 인덱스로 변환해 서버로 전송
- 유효성 검사는 서버가 담당하므로 클라이언트는 단순히 전송만 처리
- 서버 접속 재시도 루프
- 접속에 실패해도 루프를 돌며 재시도
- 서버가 아직 열리지 않은 상태에서 먼저 클라이언트를 실행해도 계속 시도하므로 실용적
7. client_main.py, 클라이언트 UI
클라이언트 화면 UI를 구성하는 메인위젯입니다.
from PyQt6.QtWidgets import QApplication, QWidget, QMessageBox, QInputDialog
from PyQt6.QtGui import QPainter
import sys
from client_game import Game
class Window(QWidget):
def __init__(self):
super().__init__()
self.setFixedSize(600,600)
# 게임생성
self.game = Game(self)
def paintEvent(self, e):
qp = QPainter()
qp.begin(self)
self.game.draw(qp)
qp.end()
def mousePressEvent(self, e):
pt = e.pos()
self.game.mouseDown(pt)
def onEnd(self, t):
txt = ('흑돌승', '백돌승', '무승부')
yesno = QMessageBox.information(self, txt[t-5], '\t\tRetry(Y), Exit(N)\t\t', QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No)
if yesno == QMessageBox.StandardButton.Yes:
self.game.onDisc()
del self.game
self.game = Game(self)
else:
self.game.onDisc()
self.close()
def closeEvent(self, e):
self.game.onDisc()
if __name__ == '__main__':
app = QApplication(sys.argv)
w = Window()
w.show()
sys.exit(app.exec())
- paintEvent가 호출될 때마다 game.draw()로 전체 화면을 다시 그림
- update_sig 신호가 발생하면 Qt가 자동으로 paintEvent를 재호출
코드 흐름 시퀀스
마무리
이 프로젝트는 목요일 심화반 수강생들과 함께 제작하였습니다 👏.PyQt6의 시그널/슬롯, Python 소켓, struct 직렬화가 어떻게 함께 작동하는지 배우기 좋은 예제입니다.
코드 규모는 작지만, 실제 멀티플레이어 게임의 핵심 패턴(서버 중재 방식, 스레드 분리, 신호 기반 UI 업데이트)이 모두 담겨 있습니다.
처음부터 어려운 것을 하기보다 작은 것부터 내것으로 만들고 큰 그림을 그려보세요😊.
궁금한 점은 댓글로.. 감사합니다.
댓글
댓글 쓰기