VoiceChat на несколько человек с блэкджеком и (Web?)UI над UDP socket python

Пару лет назад, когда все эти гитхабры для меня были птичьим пением, а делать мне было нечего — писал я значит небольшие проекты на python. Среди них был простенький голосовой чат на двоих через TCP. Но вот я нашел его и захотел допилить. Лучший способ допилить что‑то — это придумать заново.

Поскольку мы настроены серьёзно, наш проект будет солидно использовать venv.

Подготовка среды

mkdir pycord cd pycord python -m venv venv venv\scripts\activate  pip install numpy pip install pyaudio

Cоздаем папку, окружение, создаем один файлик PyCord.py.

Импортируем то, что точно понадобится.

import pyaudio import socket import numpy as np from threading import Thread

Подготавливаем IO для аудиоданных.

Pyaudio — самая изящная библиотека работы со звуковыми устройствами, которую я только встречал. Создаешь один объект с буферами ввода и вывода — пишешь, читаешь, все интуитивно понятно.

CHUNK=2048 FORMAT = pyaudio.paInt16 CHANNELS = 1 RATE = 44100 p = pyaudio.PyAudio()  stream=p.open(format=FORMAT, # Этот объект будет обеспечивать нам IO голоса             channels=CHANNELS,             rate=RATE,             input=True,                 output=True,             frames_per_buffer=CHUNK)

Sender (Audio stream -> IPs from list)

Псевдокод:

1. Читаем кусок данных (далее чанк) из аудиопотока.

2. Отправляем каждому из списка clients, полученному из get_clients() — об этом далее.

3. Повторяем до бесконечности пункты 1 и 2.

def Sender():     UDP_PORT = 6006      sock = socket.socket(socket.AF_INET,                         socket.SOCK_DGRAM)     while True:         clients = get_clients() #пока не реализовал         data = stream.read(CHUNK, exception_on_overflow = False)         for addr in clients:             sock.sendto(data, (addr, UDP_PORT))

Receiver (Internet -> Audio stream)

Есть один нюанс, recvfrom блокирует выполнение до получения чанка данных, в диалоге двух человек проблем не будет:

def Receiver():     me = "127.0.0.1"     UDP_PORT = 6006      sock = socket.socket(socket.AF_INET,                         socket.SOCK_DGRAM)     sock.bind((me, UDP_PORT))      while True:         num = get_num() #пока не реализовал         for i in range(num):             new = np.frombuffer(sock.recvfrom(CHUNK*10)[0],dtype=np.uint16)             if i == 0:                 data = new.copy()             else:                 data += new         if num != 0:             print(data.shape)             stream.write(data.tobytes())

• человек №1 записал чанк за 1/n секунды и отправил его.

• человек №2 получил его и воспроизводит в течение 1/n секунды.

Но, если принимать чанки сразу от двух человек:

• человек №1 записал чанк №1 за 1/n секунды и отправил его

• человек №2 записал чанк №2 за 1/n секунды и отправил его

• человек №3 получает их последовательно и воспроизводит в течение 2*1/n секунды

• Получается, что человек №3 тормоз! Его буфер получения растёт, он должен уметь воспроизводить чанки одновременно.

В случае с аудиоданными всё просто: суммируем их как массивы numpy.

Для решения этой проблемы отключаем блокировку recvfrom так: sock.setblocking(0)

1. Суммируем чанки пока не выпадает исключение о пустоте буфера.

2. Полученный наложенный звук с чистой совестью пишем в аудиопоток.

def Receiver():     me = "127.0.0.1"     UDP_PORT = 6006      sock = socket.socket(socket.AF_INET,                         socket.SOCK_DGRAM)     sock.setblocking(0)     sock.bind((me, UDP_PORT))       while True:         c = 0         first = True         while True:             blacklist = get_black() #пока не реализовал             try:                 d,addr = sock.recvfrom(CHUNK*10)                 if addr[0] in blacklist:                     continue                 new = np.frombuffer(d,dtype=np.uint16)                 c += 1                 if first:                     data = new.copy()                     first = False                 else:                     data += new             except:                 break         if not first:             #print(c)             stream.write(data.tobytes())

Как вы могли заметить, функции используют get_black() для ЧС и get_clients() для списка адресов отправки своего голоса

Getters & setters для динамического изменения clients, blacklist

black,clients=[],[]  def set_black(s):     global black     black = s       def get_black():     global black     return black       def set_clients(s):     global clients     clients = s       def get_clients():     global clients     return clients

Как способ общения с Reciver и Sender

sender = Thread(target=Sender, args=()) sender.start()  recv = Thread(target=Receiver, args=()) recv.start()

Обе функции запущены как потоки, и теперь работают автономно, как теперь контролировать их поведение?

set_black(["192.168.0.1"]) # Изменится количество принимаемых аудиопотоков на 2 set_clients(["127.0.0.1","192.168.0.15"]) #Теперь аудио отправляется на оба адреса

Подобная схема была выбрана, для удобства реализации взаимодействия с интерфейсом.

О нём сейчас и поговорим.

UI

Можно было бы идти по знакомой дорожке, именуемой Tk, или схожей, но чуть менее популярной Qt, можно было даже упороться в стиль с DearPyGui, но это всё заезжено и банально.

И тут я подумал: А почему бы не сделать веб-интерфейс?. Первое что пришло в голову — Gradio.

Просто было интересно сделать что-то с веб фронтом, а запариваться с отдельным фронтэндом на JS и даже api не уместно, если делаешь что-то десктопное вроде моего текущего клиента.

За следующие пару минут я не передумал — значит настрой серьёзен как никогда.

pip install gradio

import gradio as gr

Gradio позволяет конструировать webui для своего backendа прямо в его коде, при помощи интуитивно понятного конструктора.

WebUI будет иметь две вкладки:

• Отправлять

  • Слайдер выбора кол-ва адресов

  • Динамическое количество полей ввода под эти адреса

• Чёрный список

  • То же самое для конфигурации чёрного списка

import gradio as gr  def variable_outputs(k):     k = int(k)     return [gr.Textbox.update(visible=True)]*k + [gr.Textbox.update(visible=False)]*(max_textboxes-k)     max_textboxes = 10 with gr.Blocks() as demo:     gr.Markdown("PyCord")      with gr.Tab("Отправлять"):         snd = gr.Slider(0, max_textboxes, value=10, step=1, label="Адреса получателей")         textboxes = []         for i in range(max_textboxes):             t = gr.Textbox(f"127.0.0.1")             textboxes.append(t)         snd.change(variable_outputs, snd, textboxes)         button = gr.Button("Начать отправку")         button.click(lambda *s:set_clients(s[:s[-1]]),textboxes+[snd])      with gr.Tab("Чёрный список"):         snd2 = gr.Slider(0, max_textboxes, value=10, step=1, label="Блокировать входящие с адресов")         textboxes2 = []         for i in range(max_textboxes):             t = gr.Textbox(f"127.0.0.1")             textboxes2.append(t)         snd2.change(variable_outputs, snd2, textboxes2)         button2 = gr.Button("БАН")         button2.click(lambda *s:set_black(s[:s[-1]]),textboxes2+[snd2])  

«Gradio is the fastest way to demo your machine learning model with a friendly web interface so that anyone can use it, anywhere!» — я знал на что шёл, честно

Делать Webui к этому велосипеду была дикость, знаю. Зато можно управлять клиентом на пк из браузера на телефоне!

Напоминаю, что клиент всё еще работает локально на ПК, а это «интерфейс на вынос».

На телефоне интерфейс даже не выглядит пустым.

Но есть одна проблема, которая перекрывает всё удобство использования gradio интерфейсов на телефоне — это случайные нажатия.

Ты просто проводишь пальцем мимо слайдера/кнопки и они могут на это среагировать… Но ощущается это только тогда, когда интерфейс не вмещается в один экран телефона, и ты вынужден перематывать его свайпами.

Всё работает, но

Но кажется я понял — почему gradio не используют в подобных кейсах: оно очень, очень медленное

Оно долго обновляет фреймы, оно очень долго стартует. Этот интерфейс максимально неотзывчивый, он ловит случайные нажатия при свайпах на телефоне. При всём этом переделать под Tk не составит труда (изначальная версия была именно под него, но потом я решил сделать нечто необычное).

import pyaudio import socket import struct import random import numpy as np  from threading import Thread import gradio as gr   IP = socket.gethostbyname(socket.gethostname())    #Свой айпи   CHUNK=1024 FORMAT = pyaudio.paInt16 CHANNELS = 1 RATE = 44100 p = pyaudio.PyAudio()  stream=p.open(format=FORMAT,             channels=CHANNELS,             rate=RATE,             input=True,             output=True,             frames_per_buffer=CHUNK)     black,clients = [],[] def set_black(s):     global black     black = s       def get_black():     global black     return black     def set_clients(s):     global clients     clients = s       def get_clients():     global clients     return clients     def Sender():     UDP_PORT = 6006      sock = socket.socket(socket.AF_INET,                         socket.SOCK_DGRAM)     while True:         clients=get_clients()         data = stream.read(CHUNK, exception_on_overflow = False)         #print(len(data))         for addr in clients:             sock.sendto(data, (addr, UDP_PORT))    def Receiver():     me = "127.0.0.1"     UDP_PORT = 6006      sock = socket.socket(socket.AF_INET,                         socket.SOCK_DGRAM)     sock.setblocking(0)     sock.bind((me, UDP_PORT))       while True:         c = 0         first = True         while True:             blacklist = get_black()             try:                 d,addr = sock.recvfrom(CHUNK*10)                 if addr[0] in blacklist:                     continue                 new = np.frombuffer(d,dtype=np.uint16) # buffer size is 1024 bytes                 c += 1                 if first:                     data = new.copy()                     first = False                 else:                     data+=new             except:                 break         if not first:             #print(c)             stream.write(data.tobytes())     def variable_outputs(k):     k = int(k)     return [gr.Textbox.update(visible=True)]*k + [gr.Textbox.update(visible=False)]*(max_textboxes-k)   max_textboxes = 10 with gr.Blocks() as demo:     gr.Markdown("# PyCord")     with gr.Tab("Отправлять"):         snd = gr.Slider(0, max_textboxes, value=10, step=1, label="Адреса получателей")         textboxes = []         for i in range(max_textboxes):             t = gr.Textbox(f"127.0.0.1")             textboxes.append(t)         snd.change(variable_outputs, snd, textboxes)         button = gr.Button("Начать отправку")         button.click(lambda *s:set_clients(s[:s[-1]]),textboxes+[snd])      with gr.Tab("Чёрный список"):         snd2 = gr.Slider(0, max_textboxes, value=10, step=1, label="Блокировать входящие с адресов")         textboxes2 = []         for i in range(max_textboxes):             t = gr.Textbox(f"127.0.0.1")             textboxes2.append(t)         snd2.change(variable_outputs, snd2, textboxes2)         button2 = gr.Button("БАН")         button2.click(lambda *s:set_black(s[:s[-1]]),textboxes2+[snd2])        sender = Thread(target=Sender, args=()) sender.start() recv = Thread(target=Receiver, args=()) recv.start()  demo.launch(share=True) 

Finally

Никому не советую использовать gradio не по назначению — вот и весь вывод.

Спасибо за прочтение, надеюсь вы смогли найти что‑то интересное в этой статье, раз уж дочитали до сюда. Критика и идеи ожидаю в комментариях, на этом у меня всё, всем понятных сокетов и добра.