refactor all the handlers - folder structure

connections/local_audio_streamer.py

@@ -0,0 +1,38 @@
+import threading
+import sounddevice as sd
+import numpy as np
+
+import time
+
+
+class LocalAudioStreamer:
+    def __init__(
+        self,
+        input_queue,
+        output_queue,
+        list_play_chunk_size=512,
+    ):
+        self.list_play_chunk_size = list_play_chunk_size
+
+        self.stop_event = threading.Event()
+        self.input_queue = input_queue
+        self.output_queue = output_queue
+
+    def run(self):
+        def callback(indata, outdata, frames, time, status):
+            if self.output_queue.empty():
+                self.input_queue.put(indata.copy())
+                outdata[:] = 0 * outdata
+            else:
+                outdata[:] = self.output_queue.get()[:, np.newaxis]
+
+        with sd.Stream(
+            samplerate=16000,
+            dtype="int16",
+            channels=1,
+            callback=callback,
+            blocksize=self.list_play_chunk_size,
+        ):
+            while not self.stop_event.is_set():
+                time.sleep(0.001)
+            print("Stopping recording")

connections/socket_receiver.py

@@ -0,0 +1,63 @@
+import socket
+from rich.console import Console
+import logging
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+console = Console()
+
+
+class SocketReceiver:
+    """
+    Handles reception of the audio packets from the client.
+    """
+
+    def __init__(
+        self,
+        stop_event,
+        queue_out,
+        should_listen,
+        host="0.0.0.0",
+        port=12345,
+        chunk_size=1024,
+    ):
+        self.stop_event = stop_event
+        self.queue_out = queue_out
+        self.should_listen = should_listen
+        self.chunk_size = chunk_size
+        self.host = host
+        self.port = port
+
+    def receive_full_chunk(self, conn, chunk_size):
+        data = b""
+        while len(data) < chunk_size:
+            packet = conn.recv(chunk_size - len(data))
+            if not packet:
+                # connection closed
+                return None
+            data += packet
+        return data
+
+    def run(self):
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.socket.bind((self.host, self.port))
+        self.socket.listen(1)
+        logger.info("Receiver waiting to be connected...")
+        self.conn, _ = self.socket.accept()
+        logger.info("receiver connected")
+
+        self.should_listen.set()
+        while not self.stop_event.is_set():
+            audio_chunk = self.receive_full_chunk(self.conn, self.chunk_size)
+            if audio_chunk is None:
+                # connection closed
+                self.queue_out.put(b"END")
+                break
+            if self.should_listen.is_set():
+                self.queue_out.put(audio_chunk)
+        self.conn.close()
+        logger.info("Receiver closed")

connections/socket_sender.py

@@ -0,0 +1,39 @@
+import socket
+from rich.console import Console
+import logging
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+console = Console()
+
+
+class SocketSender:
+    """
+    Handles sending generated audio packets to the clients.
+    """
+
+    def __init__(self, stop_event, queue_in, host="0.0.0.0", port=12346):
+        self.stop_event = stop_event
+        self.queue_in = queue_in
+        self.host = host
+        self.port = port
+
+    def run(self):
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.socket.bind((self.host, self.port))
+        self.socket.listen(1)
+        logger.info("Sender waiting to be connected...")
+        self.conn, _ = self.socket.accept()
+        logger.info("sender connected")
+
+        while not self.stop_event.is_set():
+            audio_chunk = self.queue_in.get()
+            self.conn.sendall(audio_chunk)
+            if isinstance(audio_chunk, bytes) and audio_chunk == b"END":
+                break
+        self.conn.close()
+        logger.info("Sender closed")