from tkinter import *
import pygame
class GameSound:
def __init__(self):
window = Tk() # 윈도우 생성
window.title("게임사운드") # 제목을 설정
window.geometry("640x480") # 윈도우 크기 설정
window.resizable(0,0)
self.canvas = Canvas(window, bg = "white")
self.canvas.pack(expand=True,fill=BOTH)
window.bind("<KeyPress>",self.keyPressHandler)
window.bind("<KeyRelease>",self.keyReleaseHandler)
#pygame 에서 music vs sound
# music: 배경음악 재생을 위해 사용
# sound: 효과음을 위해 사용
#BG sound.
pygame.init()
pygame.mixer.music.load("bgm.wav") #Loading File Into Mixer
pygame.mixer.music.play(-1) #Playing It In The Whole Device
self.canvas.create_text(320,400,font="Times 15 italic bold",text="Sound Example")
#Effect sound
self.sounds = pygame.mixer
self.sounds.init()
self.s_effect1 = self.sounds.Sound("gunsound.mp3")
while True:
#
window.after(33)
window.update()
def keyReleaseHandler(self, event):
if event.keycode in self.keys:
self.keys.remove(event.keycode)
def keyPressHandler(self, event):
self.s_effect1.play()
self.keys.add(event.keycode)
GameSound()
코드 설명
pygame 모듈을 사용하기 위해 pygame을 import 한다.(line 2)
pygame을 사용하기 위해서는 pygame.init()함수를 호출해야한다.(line 21)
위 코드에서 music.load(), music.play() 함수를 이용하여 배경음악을 무한반복 재생한다.(line 22, 23)
effect sound를 위해 sounds 를 초기화 한다.(line 28)
이펙트 사운드 'gunsound.mp3' 파일을 읽고 sound 객체를 반환한다. sound 객체는 이펙트 사운드는 게임중 재생 및 멈춤을 컨트롤할 수 있다.
line 41에서 이펙트 사운드를 재생한다.
위 코드만으로 간단히 게임에서 필요한 음악 재생은 가능하며 좀 더 다양한 기능을 구현하기 위해서는 아래의 링크를 참조하기 바랍니다.
import speech_recognition as sr
r = sr.Recognizer()
with sr.Microphone() as source:
r.adjust_for_ambient_noise(source, duration=0.2)
while True:
try:
print('listen...')
#listens for the user's input
user_audio = r.listen(source)
# Using google to recognize audio
#text = r.recognize_google(user_audio) # 영어
text = r.recognize_google(user_audio, language='ko-KR') # 한글
print("Did you say: ",text)
except sr.RequestError as e:
print("Could not request results; {0}".format(e))
except sr.UnknownValueError:
print("unknown error occurred")
음성인식 패키지는 모두 아래와 같이 13개의 음성인식 엔진 또는 API을 지원하고 4개가 오프라인으로 동작이 가능하다.
대용량 파일을 읽기 위해서는 다양한 방법들이 있는데, 그중에서 메모리 매핑을 통하여 빠르게 접급하는 방법에 대해서 알아 보자.
먼저 예제 코드는 아래와 같다.
import mmap
def process(data):
# 데이터 처리 로직
pass
with open('large_file.txt', 'r+b') as f:
mm = mmap.mmap(f.fileno(), 0)
while True:
line = mm.readline()
if not line:
break
# 각 줄에 대한 처리
process(line)
mm.close()
첫줄에 mmap 를 사용할 것이라고 선언(import) 을 하는데 mmap 는 메모리 매핑(memory mapping)을 수행하는 모듈입니다. 장점은 메모리에 매핑된 데이터를 디스크에서 직접 읽어오기 때문에 디스크에서 메모리로 데이터 복사가 없습니다. 그러므로 대용량 처리를 가능하게 합니다.
그리고 open() 함수를 이용하여 파일 열기를 수행합니다.
다음으로 mmap()와 fileno() 함수를 이용하는데 이는 파일이 가지고 있는 식별자 즉 id(또는 파일 디스크립터)를 메모리에 연결시킵니다.
** 디스크립터(식별자 또는 id)는 파일이 열 때 운영체제로 부터 자동으로 할당되어짐
이제 디스크에 저장된 파일과 메모리 연결이 완료되었으니 readline()을 이용하여 한줄씩 접근하고 처리합니다.
프로세스는 독립적으로 실행되는 프로그램의 인스턴스입니다. 각 프로세스는 별도의 메모리 공간과 리소스를 가진다.
장점:
GIL의 영향을 받지 않아, 진정한 병렬 처리가 가능합니다.( GIL은 Global Interpreter Lock의 약어로, 여러 개의 스레드가 파이썬 바이트코드를 한번에 하나만 사용할 수 있게 락을 거는 것을 의미한다, 바이트 코드(Byte code)는 고급 언어로 작성된 소스 코드를 가상머신 이 이해할 수 있는 중간 코드로 컴파일한 것을 말한다.)
프로세스 간의 격리로 인해 하나의 프로세스에서 발생한 오류가 다른 프로세스에 영향을 주지 않습니다.(상호독립적으로 구성)
단점:
프로세스 간 통신(IPC, Inter-Process Communication)이 상대적으로 복잡하고 느립니다.
메모리 사용량이 더 많습니다.
스레드 (Thread)
정의: 스레드는 동일한 프로세스 내에서 실행되는 가벼운 단위입니다. 스레드들은 같은 메모리 공간을 공유합니다.
장점:
메모리 공유로 인한 효율적인 자원 사용.
스레드 간 통신이 빠르고 간단합니다.
단점:
GIL(Global Interpreter Lock)로 인해, 진정한 병렬 처리가 어렵습니다. (특히 CPU 바운드 작업에서)
하나의 스레드에서 발생한 오류가 전체 프로세스에 영향을 줄 수 있습니다.
멀티스레딩 예지
import threading
import time
def print_numbers():
for i in range(1, 6):
print(f"Number: {i}")
time.sleep(1)
def print_letters():
for letter in ['A', 'B', 'C', 'D', 'E']:
print(f"Letter: {letter}")
time.sleep(1.5)
thread1 = threading.Thread(target=print_numbers)
thread2 = threading.Thread(target=print_letters)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print("All threads are done!")
멀티프로세싱 예제
import multiprocessing
import time
def print_numbers():
for i in range(1, 6):
print(f"Number: {i}")
time.sleep(1)
def print_letters():
for letter in ['A', 'B', 'C', 'D', 'E']:
print(f"Letter: {letter}")
time.sleep(1.5)
process1 = multiprocessing.Process(target=print_numbers)
process2 = multiprocessing.Process(target=print_letters)
process1.start()
process2.start()
process1.join()
process2.join()
print("All processes are done!")
주요 차이점 요약
메모리 공유: 스레드는 같은 메모리를 공유하지만, 프로세스는 별도의 메모리를 사용합니다.
GIL의 영향: 스레드는 GIL의 영향을 받으며, CPU 바운드 작업에서 성능이 제한될 수 있습니다. 반면, 프로세스는 GIL의 영향을 받지 않습니다.
안정성: 프로세스는 서로 격리되어 있어 하나의 프로세스 오류가 다른 프로세스에 영향을 주지 않습니다.
통신 방법: 스레드는 간단하고 빠른 통신이 가능하지만, 프로세스는 복잡하고 느린 통신을 사용합니다.
적절한 방법을 선택하는 것은 작업의 특성에 따라 다르며, I/O 바운드 작업에는 스레드를, CPU 바운드 작업에는 프로세스를 사용하는 것이 일반적입니다.
import threading
import time
# 스레드에서 실행할 함수 정의
def print_numbers():
for i in range(1, 6):
print(f"Number: {i}")
time.sleep(1)
def print_letters():
for letter in ['A', 'B', 'C', 'D', 'E']:
print(f"Letter: {letter}")
time.sleep(1.5)
# 스레드 생성
thread1 = threading.Thread(target=print_numbers)
thread2 = threading.Thread(target=print_letters)
# 스레드 시작
thread1.start()
thread2.start()
# 메인 스레드가 thread1, thread2가 종료될 때까지 대기
thread1.join()
thread2.join()
print("All threads are done!")
### 코드 설명 1. `print_numbers`와 `print_letters`라는 두 개의 함수를 정의합니다. 각 함수는 숫자와 문자를 순차적으로 출력하며, 각 출력 사이에 `time.sleep`을 통해 지연 시간을 둡니다. 2. `threading.Thread` 클래스를 사용하여 두 개의 스레드를 생성합니다. 각각의 스레드는 실행할 함수를 `target` 인자로 받습니다. 3. `start()` 메서드를 호출하여 스레드를 시작합니다. 이 메서드를 호출하면 스레드가 백그라운드에서 실행되기 시작합니다. 4. `join()` 메서드를 사용하여 메인 스레드가 각 스레드의 실행이 완료될 때까지 기다리도록 합니다. 이 메서드를 호출하지 않으면 메인 스레드는 스레드의 실행이 완료되기 전에 종료될 수 있습니다. 5. 모든 스레드가 완료되면 "All threads are done!"이라는 메시지를 출력합니다.
이 예제는 두 개의 스레드를 생성하고 실행하며, 각각의 스레드가 독립적으로 동작하면서 동시에 숫자와 문자를 출력합니다. `join()` 메서드를 통해 메인 스레드가 두 스레드의 종료를 기다리므로, 모든 작업이 완료된 후에 "All threads are done!" 메시지가 출력됩니다.
<Button-1> : 마우스 왼쪽 버튼 클릭 <Button-2> : 마우스 중간 버튼 클릭 <Button-3> : 마우스 오른쪽 버튼 클릭 <Button-4> : 마우스 휠을 스크롤바에서 위로 올릴 때(scroll up event) <Button-5> : 마우스 휠을 스크롤바에서 아래로 내릴 때(scroll down event) <ButtonPress> : 아무 마우스 버튼이라도 눌리면 호출 된다. 휠까지도 <Double-Button-1> : 마우스 왼쪽 버튼 더블 클릭 <Double-Button-2> : 마우스 중간 버튼 더블 클릭 <Double-Button-3> : 마우스 오른쪽 버튼 더블 클릭 <Return> : 엔터가 눌러짐 <Key> : 키가 눌러짐 <Enter> : 마우스 포인터가 위젯에 들어올 때 <Leave> : 마우스 포인터가 위젯을 떠날때 <Configure> : 위젯 사이즈에 변화가 있을때
import speech_recognition as sr
import pyttsx3
import openai
openai.api_key = "openai key"
# Initialize the recognizer
r = sr.Recognizer()
# Initialize the voice engine
voice_engine = pyttsx3.init()
#스피치 목소리 타입 0: K, 1: female, 2: male
voices = voice_engine.getProperty('voices')
voice_engine.setProperty('voice', voices[0].id)
voice_engine.setProperty('rate', 170)
# Function to convert text to speech
def SpeakText(user_text,command):
#음성 파일 저장
voice_engine.save_to_file(command, '/answer.mp3')
#TTS 실행
voice_engine.say(command)
voice_engine.runAndWait()
def main():
# 무한
with sr.Microphone() as source:
r.adjust_for_ambient_noise(source, duration=0.2)
while True:
try:
print('listen...')
#listens for the user's input
user_audio = r.listen(source)
# Using google to recognize audio
user_text = r.recognize_google(user_audio)
user_text = user_text.lower()
print("Did you say: ",user_text)
completion = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "user", "content": user_text}
]
)
#chatgpt 결과 출력
print(completion.choices[0].message.content)
SpeakText(user_text,completion.choices[0].message.content)
except sr.RequestError as e:
print("Could not request results; {0}".format(e))
except sr.UnknownValueError:
print("unknown error occurred")
if __name__ == "__main__":
main()
from tkinter import * # tkinter에서 모든 정의를 임포트한다.
import time
import pygame
import random
import time
import socket
import ast
import json
import threading
class GameClient:
def __init__(self,user,host,port):
#사용자 아이디에서 공백 제거
user = user.replace(" " , "")
#네트워크 초기화 및 서버 연결
#
self.port = port # 연결 포트설정
self.host = host #서버 IP
self.user = user #사용자 아이디 설정
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((self.host, self.port))
#서버 연결 확인을 위한 패킷 송수신
self.client_socket.sendall(self.user.encode())#서버에 사용자 아이디 전송
data = self.client_socket.recv(1024)#서버응답 받기
data = data.decode()#서버에서 받은 데이터를 문자열로 변환
#서버 접속 확인: 접속 실패시 프로그램 종료
if "Success" in data: print("서버에 접속하였습니다.")
elif "Fail" in data:
print(self.user, "동일한 사용자가 존재합니다. 사용자를 변경하기 바랍니다.")
exit()
else:
print("서버에 접속에 실패하였습니다.","프로그램을 종료합니다.")
exit()
#서버에서 전송된 사용자 데이터를 저장하는 변수
self.user_data={}
#
self.receive_thread = threading.Thread(target=self.handle_receive)
self.receive_thread.daemon = True
self.receive_thread.start()
#게임윈도우 만들기 시작...
print("게임윈도우를 생성합니다.")
window = Tk() # 윈도우 생성
window.title("네트워크게임예제") # 제목을 설정
window.geometry("1440x960") # 윈도우 크기 설정
window.resizable(0,0) # 윈도우 크기 조절 X
self.lightingTimer = time.time() # 일정 주기로 lighting effect 효과를 위한 타임 변수
self.keys=set() # 중복된 사용자 입력 키 값을 쉽게 제거하기 위해 set 타입으로 선언
self.canvas = Canvas(window, bg = "white")
self.canvas.pack(expand=True,fill=BOTH)
window.bind("<KeyPress>",self.keyPressHandler)
window.bind("<KeyRelease>",self.keyReleaseHandler)
self.img_dragon = [PhotoImage(file='image/dragon-animated-gif.gif',format = 'gif -index %i' %(i)).subsample(3) for i in range(40)]
self.img_light_effect = PhotoImage(file="image/lightning-effect-png2.png")
self.img_fire = PhotoImage(file = "image/fire_type2.png")
self.img_bg = PhotoImage(file="image/bgimage2.png")
self.img_enemy = PhotoImage(file='image/spaceship.png').subsample(6)
#배경 그리기.
self.canvas.create_image(0,0, image = self.img_bg,anchor = NW,tags="bg")
self.isDebugMode = False
#캔버스에 그려질 미사일에 대한 100 개의 더미 생성, 안보이게 설정
self.fire_obj_list = []
self.fire_obj_debug_list = []
for i in range(100):
self.fire_obj_list.append(self.canvas.create_image(0,0, image = self.img_fire,state='hidden', tags="fire"))
self.fire_obj_debug_list.append(self.canvas.create_rectangle(0,0,5,5, fill='white',state='hidden', tags="fire"))
#캔버스에 그려질 적에 대한 100 개의 더미 생성, 안보이게 설정
self.enemy_obj_list = []
self.enemy_obj_debug_list = []
for i in range(100):
self.enemy_obj_list.append(self.canvas.create_image(0,0, image = self.img_enemy,state='hidden', tags="enemy"))
self.enemy_obj_debug_list.append(self.canvas.create_rectangle(0,0,140,65,state='hidden', tags="enemy"))
#배경사운드
pygame.init()
pygame.mixer.music.load("sound/bgm.wav") #Loading File Into Mixer
pygame.mixer.music.play(-1) #Playing It In The Whole Device
#충돌 이펙트 사운드
self.sounds = pygame.mixer
self.sounds.init()
self.s_effect1 = self.sounds.Sound("sound/destruction.mp3")
self.canvas.create_text(150,60,fill="white",font="Times 15 italic bold",text="입력키: ↑ , ↓ , ← , → , space")
self.canvas.create_text(720,810,fill="white",font="Times 20 italic bold",text="Network Game Example")
self.canvas.create_text(720,850,fill="white",font="Times 20 italic bold",text="Major of computer software, Gyeongsang National Universityty")
#
user_list_in = set()
while True:
#사용자 입력이 있을 경우 서버로 데이터 전송
if(len(self.keys)>0):#각 키값을 ',' 구분된 아스키 코드 값으로 전송
self.client_socket.sendall(','.join(str(e) for e in self.keys).encode())
fire_obj_idx = 0
user_list_from_server = set()
for key, value in self.user_data.items():
if key != 'enemy':
try:
obj = self.canvas.find_withtag(key)
except:
print('프로그램을 종료합니다.')
exit()
img_dragon_idx = value[0]%len(self.img_dragon)
if len(obj)==0:
if self.user != key:
print(key,' 사용자가 입장하였습니다.')
self.canvas.create_image(value[1],value[2], image = self.img_dragon[img_dragon_idx],tags=key)
self.canvas.create_text(value[1]+self.img_dragon[img_dragon_idx].width()/2,value[2]+self.img_dragon[img_dragon_idx].height()/2+20, font="Times 20 italic bold",fill='#ff00ff',text=key+' '+str(value[4]),tag=key)
user_list_in.add(key)
user_list_from_server.add(key)
else:
self.canvas.itemconfig(obj[0], image = self.img_dragon[img_dragon_idx])
self.canvas.moveto(obj[0],value[1],value[2])
#itemcget(item_id,속성): canvas item 들의 속성 값 가져오기
pre_text = self.canvas.itemcget(obj[1],'text')
post_text = key+' '+str(value[4])
#값이 다르면 text 속성 업데이트, 충돌 사운드 재생
if pre_text != post_text:
self.canvas.itemconfig(obj[1], text=post_text)
self.s_effect1.play()
self.canvas.moveto(obj[1],value[1]+self.img_dragon[img_dragon_idx].width()/2,value[2]+self.img_dragon[img_dragon_idx].height()/2+20)
user_list_from_server.add(key)
for i in range(len(value[3])):
if(fire_obj_idx < len(self.fire_obj_list)):
self.canvas.itemconfig(self.fire_obj_list[fire_obj_idx],state='normal')
self.canvas.moveto(self.fire_obj_list[fire_obj_idx],value[3][i][0],value[3][i][1])
if self.isDebugMode:
self.canvas.itemconfig(self.fire_obj_debug_list[fire_obj_idx],state='normal')
else:
self.canvas.itemconfig(self.fire_obj_debug_list[fire_obj_idx],state='hidden')
self.canvas.moveto(self.fire_obj_debug_list[fire_obj_idx],value[3][i][0]+50,value[3][i][1])
fire_obj_idx += 1
else:#만약 보여되는 미사일 수가 더미로 미사일 수보다 많으면 추가로 만들기
self.fire_obj_list.append(self.canvas.create_image(value[3][i][0],value[3][i][1], image = self.img_fire, tags="fire"))
if self.isDebugMode:
self.fire_obj_debug_list.append(self.canvas.create_rectangle(value[3][i][0],value[3][i][1],value[3][i][0]+5,value[3][i][1]+5, fill='white', tags="fire"))
else:
self.fire_obj_debug_list.append(self.canvas.create_rectangle(value[3][i][0],value[3][i][1],value[3][i][0]+5,value[3][i][1]+5, state='hidden', fill='white', tags="fire"))
#서버에 접속이 끊긴 사용자들 제거
user_list_exit = user_list_in - user_list_from_server
for u in list(user_list_exit):
print(u,' 사용자가 퇴장하였습니다.')
user_list_in.remove(u)
objs = self.canvas.find_withtag(u)
for o in objs:
self.canvas.delete(o)
#사용되지 않은 미사일 이미지들은 안보이게 설정
for i in range(fire_obj_idx,len(self.fire_obj_list)):
self.canvas.itemconfig(self.fire_obj_list[i],state='hidden')
self.canvas.itemconfig(self.fire_obj_debug_list[i],state='hidden')
enemy_obj_idx = 0
#
for p in self.user_data['enemy']:
if(enemy_obj_idx < len(self.enemy_obj_list)):
self.canvas.itemconfig(self.enemy_obj_list[enemy_obj_idx],state='normal')
self.canvas.moveto(self.enemy_obj_list[enemy_obj_idx],p[0],p[1])
if self.isDebugMode:
self.canvas.itemconfig(self.enemy_obj_debug_list[enemy_obj_idx],state='normal')
else:
self.canvas.itemconfig(self.enemy_obj_debug_list[enemy_obj_idx],state='hidden')
self.canvas.moveto(self.enemy_obj_debug_list[enemy_obj_idx],p[0],p[1])
enemy_obj_idx += 1
else:#만약 보여되는 적 수가 더미 적 수보다 많으면 추가로 만들기
self.enemy_obj_list.append(self.canvas.create_image(p[0],p[1], image = self.img_enemy, tags="enemy"))
if self.isDebugMode:
self.enemy_obj_debug_list.append(self.canvas.create_rectangle(p[0],p[1],p[0]+5,p[1]+5, fill='red', tags="enemy"))
else:
self.enemy_obj_debug_list.append(self.canvas.create_rectangle(p[0],p[1],p[0]+5,p[1]+5, state='hidden', fill='red', tags="enemy"))
#사용되지 않은 적 이미지들은 안보이게 설정
for i in range(enemy_obj_idx,len(self.enemy_obj_list)):
self.canvas.itemconfig(self.enemy_obj_list[i],state='hidden')
self.canvas.itemconfig(self.enemy_obj_debug_list[i],state='hidden')
self.display_lighting_effect()
window.after(33)
window.update()
#
self.client_socket.close()
self.receive_thread.join()
def keyReleaseHandler(self, event):
if event.keycode==68:
self.isDebugMode = not self.isDebugMode
key = str(event.keycode)
if key in self.keys:
self.keys.remove(key)
def lighting_effect(self):
for i in range(0,random.randint(5,15)):
self.canvas.create_image(random.randint(0,self.canvas.winfo_width()),random.randint(0,self.canvas.winfo_height()), image = self.img_light_effect,anchor = NW,tags="lighting")
def display_lighting_effect(self):
if(self.lightingTimer == -1):
self.lightingTimer = time.time()
self.lighting_effect()
else:
now = time.time()
if(now - self.lightingTimer > 4.0):
self.lightingTimer = -1
elif(now - self.lightingTimer > 2.0):
lightings = self.canvas.find_withtag("lighting")
for light in lightings:
self.canvas.delete(light)
def keyPressHandler(self, event):
key = str(event.keycode)
self.keys.add(key)
def handle_receive(self):
while True:
try:
data = self.client_socket.recv(1024)#데이터가 수신될 때까지 대기함
except:
print("서버 연결 끊김")
break
#byte를 문자열로 디코딩
data = data.decode()
#디코딩된 문자열은 json 형식으로 되어 있음
#네트워크 딜레이로 인해 누적된 패킷중 마지막 패킷만 추출하기
data = data[:data.rfind('}')+1]
data = data[data.rfind('{'):]
#패킷의 형식 오류 확인
if len(data) > 0 and data[0] =='{' and data[-1] == '}':
#json 형식을 딕셔너리 타입으로 변환 후 저장
self.user_data = json.loads(data)
else:
pass#print('패킷오류')
if __name__ == '__main__':
#사용자 아이디, 서버 IP, 연결포트 설정
GameClient('CKang','127.0.0.1',4000) # GameClient 생성한다.
서버 코드입니다.
import socket
import argparse
import threading
import time
import random
import json
user_data = {'enemy':[]}
user_socket = {}
user_image_size = [280,249]
user_missile_size = [50,10]
user_enemy_size = [140, 65]
user_wnd_size = [1440,960]
#frame, 위치 x, 위치 y, 미사일, 점수
user_ini_data = [0,100,480-user_image_size[1]/2,[],0]
def handle_sendToUsers():
while 1:
#모든 미사일의 위치값 업데이트
for key in user_data.keys():
if key != 'enemy':
for i in range(len(user_data[key][3])):
user_data[key][3][i][0] += 9
#미사일, 적 충돌 체크
isChecked = False
for e in user_data['enemy']:
if e[0] < user_data[key][3][i][0]+user_missile_size[0] and user_data[key][3][i][0]+user_missile_size[0] < e[0] + user_enemy_size[0] and e[1] < user_data[key][3][i][1] and user_data[key][3][i][1] < e[1] + user_enemy_size[1]:
e[0] = -1440 #충돌된 적은 화면을 벗어나게 하기
user_data[key][4] += 1#사용자 점수 증가
#print(key,'명중:',user_data[key][4])
isChecked=True
if isChecked:
user_data[key][3][i][0] = 2880#충돌된 미사일은 화면을 벗어나게 하기
#화면을 벗어난 미사일 데이터 제거
user_data[key][3] = [e for e in user_data[key][3] if e[0]<user_wnd_size[0]]
#각 사용자의 프레임 번호 업데이트
for key in user_data.keys():
if key != 'enemy':
user_data[key][0] += 1
#적 생성
if (random.randint(0,100) == 0 and len(user_data.keys())>1):
user_data['enemy'].append([user_wnd_size[0],random.randint(10, user_wnd_size[1]-10)])
# 적 이동
for e in user_data['enemy']:
e[0] -= 4
# 화면 밖으로 벗어난 적 데이터 제거
user_data['enemy'] = [e for e in user_data['enemy'] if e[0]>-user_enemy_size[0]]
#각 서버에 저장된 사용자 데이터를 모든 클라이언트에 전송
for con in user_socket.values():
#딕셔너리 타입을 json 포맷 변환후 바이트형으로 인코딩
user_encode_data = json.dumps(user_data).encode()
#사용자에게 전송함
try:
con.sendall(user_encode_data)
except:
pass
time.sleep(0.03)
def handle_receive(client_socket, addr, user):
missile_time = time.time()
while 1:
try:
#클라이언트로부터 데이터 올때까지 대기함
data = client_socket.recv(1024)
#클라이언트로부터 받은 데이터를 문자열 변환후 입력 키 값들을 분리한다.
key_list = data.decode().split(',')
for key in key_list:
if key == '39' and user_data[user][1] < (user_wnd_size[0]-user_image_size[0]/2): # right direction key
user_data[user][1] = user_data[user][1] + 5
elif key == '37' and -user_image_size[0]/2 < user_data[user][1]: # left direction key
user_data[user][1] = user_data[user][1] - 5
elif key == '38' and -user_image_size[1]/2 < user_data[user][2]: # up direction key
user_data[user][2] = user_data[user][2] - 5
elif key == '40' and user_data[user][2] < (user_wnd_size[1]-user_image_size[1]/2): # down direction key
user_data[user][2] = user_data[user][2] + 5
elif key == '32':
now = time.time()
#미사일 연사속도 조절
if (now-missile_time) > 0.4:#0.4초
missile_time = now
#사용자의 현재 위치값을 이용하여 미사일 최초 위치 설정하고 미사일 위치 데이터 저장
user_data[user][3].append([user_data[user][1]+user_image_size[0]/2+95,user_data[user][2]+user_image_size[1]/2+12])
except:
print(user,': 연결 끊김')
break
del user_data[user]
del user_socket[user]
client_socket.close()
if __name__ == '__main__':
host = "127.0.0.1"
port = 4000
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((host, port))
server_socket.listen(5)# 클라이언트 최대 접속 수
send_thread = threading.Thread(target=handle_sendToUsers)
send_thread.daemon = True
send_thread.start()
print('게임서버를 시작합니다.')
while 1:
try:
client_socket, addr = server_socket.accept()#클라이언트 접속 까지 대기
except KeyboardInterrupt:
for user, con in user_socket:
con.close()
server_socket_tcp.close()
print("Keyboard interrupt")
break
user = client_socket.recv(1024)
user = user.decode()
if user in user_data:
print(user,'와 동일한 접속자가 있습니다.(접속실패)')
client_socket.sendall("Fail: 동일한 사용자가 존재합니다.".encode())
else:
print(user,'서버에 접속하였습니다.')
client_socket.sendall("Success".encode())
user_socket[user] = client_socket
user_data[user] = list(user_ini_data)
receive_thread = threading.Thread(target=handle_receive, args=(client_socket, addr,user))
receive_thread.daemon = True
receive_thread.start()
# 서버코드
import socket
import argparse
import threading
import time
user_list = {}
def handle_receive(client_socket, addr, user):
while 1:
data = client_socket.recv(1024)
data = data.decode()
if data == "/종료" or data == "/exit" : break
print(user,data)
#받은 데이터를 모든 클라이언트에게 전송한다.
for con in user_list.values():
try:
con.sendall(data.encode())
except:
print("클라이언트 소켓 에러")
del user_list[user]
client_socket.close()
if __name__ == '__main__':
host = "host_ip" #127.0.0.1
port = 4000
#IPv4 체계, TCP 타입 소켓 객체를 생성
# socket.AF_INET: 해당 소켓을 IP version 4 용으로 사용하겠다는 의미
# socket.SOCK_STREAM: 해당 소켓에 TCP 패킷을 받겠다는 의미
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#포트를 사용 중 일때 에러를 해결하기 위한 구문
#기본적으로 socket을 사용하고 나면 사용한 port는 close로 닫아주는게 맞다.
#그래야 해당 port를 다른데서도 사용할 수 있으니깐.
#근데 port를 닫았다고해도 일정시간 동안은 TIME_WAIT 상태로 대기한다고 한다.
#그리고 이 상태에서는 즉각적으로 다시 바로 사용할 수 없다.
#그래서 socket을 쓰고 close를 했는데도 already in use 에러가 발생하게된다.
#이런 에러를 막기위해서 setsockopt을 사용한다. 사용법은 아래와 같다.
#socket.SO_REUSEADDR: 방금 사용하고 close한 port를 즉시 다시 사용한다는 의미
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#ip주소와 port번호를 함께 socket에 바인드 한다.
#포트의 범위: 1~65535
server_socket.bind((host, port))
server_socket.listen(5)# 클라이언트 최대 접속 수
print('채팅서버를 시작합니다.')
while 1:
try:
#클라이언트 함수가 접속하면 새로운 소켓을 반환한다.
client_socket, addr = server_socket.accept()#클라이언트 접속까지 대기함
except:
#의도하지 않은 서버 종료에 대한 예외 처리
#프로그램 중단시 모든 클라이언트 소켓들 닫기.
for user, con in user_list:
con.close()
server_socket.close()
break
user = client_socket.recv(1024)
user = user.decode()
print(user,'서버에 접속하였습니다.')
user_list[user] = client_socket
#daemon 속성은 서브쓰레드가 데몬 쓰레드인지 아닌지를 지정하는 것인데,
#데몬 쓰레드란 백그라운드에서 실행되는 쓰레드로 메인 쓰레드가 종료되면 즉시 종료되는 쓰레드이다.
#반면 데몬 쓰레드가 아니면 해당 서브쓰레드는 메인 쓰레드가 종료할 지라도 자신의 작업이 끝날 때까지 계속 실행된다.
receive_thread = threading.Thread(target=handle_receive, args=(client_socket, addr,user))
receive_thread.daemon = True
receive_thread.start()
import socket
import threading
def handle_receive(client_socket, user):
while True:
try:
data = client_socket.recv(1024)#데이터가 수신될 때까지 대기함
except:
print("연결 끊김")
break
data = data.decode()#바이트 타입의 데이터를 decode()로 문자열로 변환
if not user in data:#자신의 데이터를 제외한 나머지 데이터를 출력함
print(data)
def handle_send(client_socket, user):
while True:
data = input()#사용자로부터 문자열을 입력받는 함수
#데이터는 encode()함수를 이용하여 바이트형으로 변환
client_socket.sendall(data.encode())#소켓을 통해 서버로 데이터 보내기
if data == "/종료" or data == "/exit": break
client_socket.close()
print('소켓 닫기')
if __name__ == '__main__':
port = 4000 # 포트넘버
host = 'server_ip' #서버 IP #127.0.0.1
user = 'user_id'
#IPv4 체계, TCP 타입 소켓 객체를 생성
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 지정한 host와 prot를 통해 서버에 접속합니다.
client_socket.connect((host, port))
# 문자열을 전송할 때 encode()을 이용한다. 파이썬 문자열의 encode() 메소드는 문자열을 byte로 변환해주는 메소드이기 때문이다.
# 파이썬 내부에서 다뤄지는 문자열은 파이썬에서 생성된 객체이고, 이를 바로 트랜스포트에 그대로 싣는 것은 불가능합니다.
# 그러므로 적절한 인코딩을 하여 보내야만 합니다.
# 만약 encode()을 사용하지 않을 경우 에러가 발생 한다.
client_socket.sendall(user.encode())
receive_thread = threading.Thread(target=handle_receive, args=(client_socket, user))
receive_thread.daemon = True
receive_thread.start()
send_thread = threading.Thread(target=handle_send, args=(client_socket, user))
send_thread.daemon = True
send_thread.start()
#join()은 부모쓰레드가 자식 쓰레드가 종료될 때까지 기다려준다.
#즉 전체 프로그램의 실행이 하나의 쓰레드이고 부모쓰레드가 되며,
#먼저 send_thread, receive_thread가 종료된후 프로그램이 종료하게 된다.
send_thread.join()
receive_thread.join()
