使用pyaudio
库调用电脑的麦克风和声音播放设备,
使用socket
实现网络通讯
代码如下:
server:
import socket
import threading
class Server:
def __init__(self):
self.ip = socket.gethostbyname(socket.gethostname())
while True:
try:
self.port = 9808
self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.s.bind((self.ip, self.port))
break
except:
print("Couldn't bind to that port")
self.connections = []
self.accept_connections()
def accept_connections(self):
self.s.listen(100)
print('Running on IP: ' + self.ip)
print('Running on port: ' + str(self.port))
while True:
c, addr = self.s.accept()
self.connections.append(c)
threading.Thread(target=self.handle_client, args=(c, addr,)).start()
def broadcast(self, sock, data):
for client in self.connections:
if client != self.s and client != sock:
try:
client.send(data)
except:
pass
def handle_client(self, c, addr):
while 1:
try:
data = c.recv(1024)
print(data)
self.broadcast(c, data)
except socket.error:
c.close()
server = Server()
client:
import socket
import threading
import pyaudio
class Client:
def __init__(self):
self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.target_ip = input('连接ip:')
self.target_port = int(input('端口号:'))
print(self.target_port,type(self.target_port))
while 1:
try:
# self.target_ip = '192.168.141.1'
# self.target_port = 9808
self.s.connect((self.target_ip, self.target_port))
break
except Exception as e:
raise e
print("Couldn't connect to server")
chunk_size = 1024 # 512
audio_format = pyaudio.paInt16
channels = 1
rate = 20000
self.p = pyaudio.PyAudio()
self.playing_stream = self.p.open(format=audio_format, channels=channels, rate=rate, output=True,
frames_per_buffer=chunk_size)
self.recording_stream = self.p.open(format=audio_format, channels=channels, rate=rate, input=True,
frames_per_buffer=chunk_size)
print("Connected to Server")
# start threads
receive_thread = threading.Thread(target=self.receive_server_data).start()
self.send_data_to_server()
def receive_server_data(self):
while True:
try:
data = self.s.recv(1024)
self.playing_stream.write(data)
except:
pass
def send_data_to_server(self):
while True:
try:
data = self.recording_stream.read(1024)
self.s.sendall(data)
except:
pass
client = Client()
最后:可以把server
部署到服务器,然后将client
打包成exe
文件,就可以直接双击运行了
网友评论