first version streaming via webrtc but quality sucks

charles_actor.py

@@ -22,6 +22,13 @@ class CharlesActor:
         print("000")
         from streamlit_av_queue import StreamlitAVQueue
         self._streamlit_av_queue = StreamlitAVQueue()
+        self._out_audio_queue = self._streamlit_av_queue.get_out_audio_queue()
+
+        print("001")
+        from ffmpeg_converter_actor import FFMpegConverterActor
+        self._ffmpeg_converter_actor = FFMpegConverterActor.remote(self._out_audio_queue)
+        await self._ffmpeg_converter_actor.start_process.remote()
+        self._ffmpeg_converter_actor.run.remote()
 
         print("002")
         from speech_to_text_vosk_actor import SpeechToTextVoskActor
@@ -29,7 +36,7 @@ class CharlesActor:
 
         print("003")
         from respond_to_prompt_actor import RespondToPromptActor
-        self._respond_to_prompt_actor = RespondToPromptActor.remote()
+        self._respond_to_prompt_actor = RespondToPromptActor.remote(self._ffmpeg_converter_actor)
         
         self._debug_queue = [
             # "hello, how are you today?",

debug_app.py

@@ -61,6 +61,7 @@ def process_frame(old_frame):
 
         if not converter_queue.empty():
             frame_as_bytes = converter_queue.get()
+            # print(f"frame_as_bytes: {len(frame_as_bytes)}")
             samples = np.frombuffer(frame_as_bytes, dtype=np.int16)
         else: 
             # create a byte array of zeros

ffmpeg_converter_actor.py

@@ -7,7 +7,7 @@ from ray.util.queue import Queue
 class FFMpegConverterActor:
     def __init__(self, queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
         self.queue = queue
-        self.buffer_size = 1920
+        self.buffer_size = buffer_size
         self.output_format = output_format
 
         self.input_pipe = None
@@ -17,7 +17,8 @@ class FFMpegConverterActor:
 
     async def run(self):
         while True:
-            chunk = await self.output_pipe.read(self.buffer_size)
+            chunk = await self.output_pipe.readexactly(self.buffer_size)
+            # print(f"FFMpegConverterActor: read {len(chunk)} bytes")
             await self.queue.put_async(chunk)       
 
     async def start_process(self):
@@ -41,7 +42,7 @@ class FFMpegConverterActor:
         self.output_pipe = self.process.stdout
 
         assert self.input_pipe is not None, "input_pipe was not initialized"
-        print (f"input_pipe: {self.input_pipe}")
+        # print (f"input_pipe: {self.input_pipe}")
 
     async def push_chunk(self, chunk):
         try:

respond_to_prompt_actor.py

@@ -71,15 +71,34 @@ class SpeechToSpeakerActor:
     async def run(self):
         while True:
             audio_chunk = await self.input_queue.get_async()
+            # print (f"Got audio chunk {len(audio_chunk)}")
             self.chat_service.enqueue_speech_bytes_to_play([audio_chunk])
             
     def cancel(self):
         while not self.input_queue.empty():
             self.input_queue.get()            
 
+@ray.remote
+class SpeechToConverterActor:
+    def __init__(self, input_queue, ffmpeg_converter_actor):
+        load_dotenv()
+        self.input_queue = input_queue
+        self.ffmpeg_converter_actor = ffmpeg_converter_actor
+
+    async def run(self):
+        while True:
+            audio_chunk = await self.input_queue.get_async()
+            # print (f"Got audio chunk {len(audio_chunk)}")
+            await self.ffmpeg_converter_actor.push_chunk.remote(audio_chunk)
+            
+    def cancel(self):
+        while not self.input_queue.empty():
+            self.input_queue.get()            
+
+
 @ray.remote
 class RespondToPromptActor:
-    def __init__(self):
+    def __init__(self, ffmpeg_converter_actor):
         voice_id="2OviOUQc1JsQRQgNkVBj"
         self.prompt_queue = Queue(maxsize=100)
         self.llm_sentence_queue = Queue(maxsize=100)
@@ -87,18 +106,19 @@ class RespondToPromptActor:
         
         self.prompt_to_llm = PromptToLLMActor.remote(self.prompt_queue, self.llm_sentence_queue, voice_id)
         self.llm_sentence_to_speech = LLMSentanceToSpeechActor.remote(self.llm_sentence_queue, self.speech_chunk_queue, voice_id)
-        self.speech_to_speaker = SpeechToSpeakerActor.remote(self.speech_chunk_queue, voice_id)
+        # self.speech_output = SpeechToSpeakerActor.remote(self.speech_chunk_queue, voice_id)
+        self.speech_output = SpeechToConverterActor.remote(self.speech_chunk_queue, ffmpeg_converter_actor)
 
         # Start the pipeline components.
         self.prompt_to_llm.run.remote()
         self.llm_sentence_to_speech.run.remote()
-        self.speech_to_speaker.run.remote()
+        self.speech_output.run.remote()
             
     def enqueue_prompt(self, prompt):
         print("flush anything queued")
         prompt_to_llm_future = self.prompt_to_llm.cancel.remote()
         llm_sentence_to_speech_future = self.llm_sentence_to_speech.cancel.remote()
-        speech_to_speaker_future = self.speech_to_speaker.cancel.remote()
+        speech_to_speaker_future = self.speech_output.cancel.remote()
         ray.get([
             prompt_to_llm_future,
             llm_sentence_to_speech_future,

streamlit_av_queue.py

@@ -12,6 +12,7 @@ import torch
 
 class StreamlitAVQueue:
     def __init__(self, audio_bit_rate=16000):
+        self._output_channels = 2
         self._audio_bit_rate = audio_bit_rate
         self.queue_actor = WebRtcAVQueueActor.options(
             name="WebRtcAVQueueActor", 
@@ -56,14 +57,28 @@ class StreamlitAVQueue:
             
         # return empty frames to avoid echo
         new_frames = []
-        for frame in frames:
-            input_array = frame.to_ndarray()
-            new_frame = av.AudioFrame.from_ndarray(
-                np.zeros(input_array.shape, dtype=input_array.dtype),
-                layout=frame.layout.name,
-            )
-            new_frame.sample_rate = frame.sample_rate
-            new_frames.append(new_frame)    
+        try:
+            for frame in frames:
+                required_samples = frame.samples
+                # print (f"frame: {frame.format.name}, {frame.layout.name}, {frame.sample_rate}, {frame.samples}")
+                assert frame.format.bytes == 2
+                assert frame.format.name == 's16'
+                frame_as_bytes = await self.queue_actor.get_out_audio_frame.remote()
+                if frame_as_bytes:
+                    # print(f"frame_as_bytes: {len(frame_as_bytes)}")
+                    assert len(frame_as_bytes) == frame.samples * frame.format.bytes
+                    samples = np.frombuffer(frame_as_bytes, dtype=np.int16)
+                else:
+                    samples = np.zeros((required_samples * 2 * 1), dtype=np.int16)
+                if self._output_channels == 2:
+                    samples = np.vstack((samples, samples)).reshape((-1,), order='F')
+                samples = samples.reshape(1, -1)
+                layout = 'stereo' if self._output_channels == 2 else 'mono'
+                new_frame = av.AudioFrame.from_ndarray(samples, format='s16', layout=layout)
+                new_frame.sample_rate = frame.sample_rate
+                new_frames.append(new_frame)
+        except Exception as e:
+            print (e)
         return new_frames
 
     async def get_in_audio_frames_async(self) -> List[av.AudioFrame]:
@@ -72,4 +87,10 @@ class StreamlitAVQueue:
 
     async def get_video_frames_async(self) -> List[av.AudioFrame]:
         shared_tensors = await self.queue_actor.get_in_video_frames.remote()
-        return shared_tensors
+        return shared_tensors
+    
+    def get_out_audio_queue(self):
+        return self.queue_actor.get_out_audio_queue.remote()
+    
+    # def get_out_audio_frame(self):
+    #     return self.queue_actor.get_out_audio_frame.remote()

webrtc_av_queue_actor.py

@@ -10,6 +10,8 @@ class WebRtcAVQueueActor:
     def __init__(self):
         self.in_audio_queue = Queue(maxsize=100)  # Adjust the size as needed
         self.in_video_queue = Queue(maxsize=100)  # Adjust the size as needed
+        self.out_audio_queue = Queue(maxsize=100)  # Adjust the size as needed
+
 
     def enqueue_in_video_frame(self, shared_tensor_ref):
         if self.in_video_queue.full():
@@ -41,3 +43,12 @@ class WebRtcAVQueueActor:
             shared_tensor_ref = self.in_video_queue.get()
             video_frames.append(shared_tensor_ref)
         return video_frames
+    
+    def get_out_audio_queue(self):
+        return self.out_audio_queue
+    
+    async def get_out_audio_frame(self):
+        if self.out_audio_queue.empty():
+            return None
+        audio_frame = await self.out_audio_queue.get_async()
+        return audio_frame