basic POC

.vscode/launch.json

@@ -8,7 +8,8 @@
             "program": "/opt/miniconda3/envs/streamlit/bin/streamlit",
             "args": [
                 "run",
-                "app.py"
+                // "app.py"
+                "d_app.py"
             ]
         }
     ]

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "[python]": {
+        "editor.defaultFormatter": "ms-python.black-formatter"
+    },
+    "python.formatting.provider": "none"
+}

app.py

@@ -15,9 +15,6 @@ from sample_utils.turn import get_ice_servers
 import json
 from typing import List
 
-from vosk import SetLogLevel, Model, KaldiRecognizer
-SetLogLevel(-1) # mutes vosk verbosity
-
 from dotenv import load_dotenv
 load_dotenv()
 
@@ -114,57 +111,10 @@ async def main():
 
     playing = st.checkbox("Playing", value=True)
 
-    def load_vosk (model='small'):
-        # load vosk model
-        # get path of current file
-        current_file_path = os.path.abspath(__file__)
-        current_directory = os.path.dirname(current_file_path)
-        _path = os.path.join(current_directory, 'models', 'vosk', model)
-        model_voice = Model(_path)
-        recognizer = KaldiRecognizer(model_voice, system_one['audio_bit_rate'])
-        return recognizer
-
-    vask = load_vosk()
-
     def handle_audio_frame(frame):
         # if self.vosk.AcceptWaveform(data):
         pass
 
-
-    def do_work(data: bytearray) -> tuple[str, bool]:
-        text = ''
-        speaker_finished = False
-        if vask.AcceptWaveform(data):
-            result = vask.Result()
-            result_json = json.loads(result)
-            text = result_json['text']
-            speaker_finished = True
-        else:
-            result = vask.PartialResult()
-            result_json = json.loads(result)
-            text = result_json['partial']
-        return text, speaker_finished
-
-
-    audio_frames_deque_lock = threading.Lock()
-    audio_frames_deque: deque = deque([])
-
-    video_frames_deque_lock = threading.Lock()
-    video_frames_deque: deque = deque([])
-
-    async def queued_video_frames_callback(
-                frames: List[av.AudioFrame],
-            ) -> av.AudioFrame:
-        with video_frames_deque_lock:
-            video_frames_deque.extend(frames)
-        return frames
-
-    async def queued_audio_frames_callback(
-                frames: List[av.AudioFrame],
-            ) -> av.AudioFrame:
-        with audio_frames_deque_lock:
-            audio_frames_deque.extend(frames)
-
         # create frames to be returned.
         new_frames = []
         for frame in frames:
@@ -187,6 +137,7 @@ async def main():
     system_one_audio_status.write("Initializing chat pipeline")
     from chat_pipeline import ChatPipeline
     chat_pipeline = ChatPipeline()
+    await chat_pipeline.start()
 
     system_one_audio_status.write("Initializing CLIP templates")
 
@@ -244,79 +195,78 @@ async def main():
         return top_3
 
     while True:
-        # await chat_pipeline.start()
-        # await chat_pipeline.enqueue(text)
-        if webrtc_ctx.state.playing:
-            # handle video
-            video_frames = []
-            with video_frames_deque_lock:
-                while len(video_frames_deque) > 0:
-                    frame = video_frames_deque.popleft()
-                    video_frames.append(frame)
-            get_embeddings = False
-            get_embeddings |= current_video_embedding is None
-            current_time = time.monotonic()
-            elapsed_time = current_time - current_video_embedding_timestamp
-            get_embeddings |= elapsed_time > 1. / system_one['vision_embeddings_fps']
-            if get_embeddings and len(video_frames) > 0:
-                current_video_embedding_timestamp = current_time
-                current_video_embedding = clip_transform.image_to_embeddings(video_frames[-1].to_ndarray())
-
-                emotions_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_emotions_embeddings"], system_one["video_detection_emotions"])
-                engagement_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_engement_embeddings"], system_one["video_detection_engement"])
-                present_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_present_embeddings"], system_one["video_detection_present"])
-
-                # table_content = "**System 1 Video:**\n\n"
-                table_content = "| System 1 Video |    |\n| --- | --- |\n"
-                table_content += f"| Present | {present_top_3} |\n"
-                table_content += f"| Emotion | {emotions_top_3} |\n"
-                table_content += f"| Engagement | {engagement_top_3} |\n"
-                system_one_video_output.markdown(table_content)
-                # system_one_video_output.markdown(f"**System 1 Video:** \n [Emotion: {emotions_top_3}], \n [Engagement: {engagement_top_3}], \n [Present: {present_top_3}] ")
-                # for similarity, image_label in similarity_image_label:
-                #     print (f"{similarity} {image_label}")
-
-            # handle audio
-            audio_frames = []
-            with audio_frames_deque_lock:
-                while len(audio_frames_deque) > 0:
-                    frame = audio_frames_deque.popleft()
-                    audio_frames.append(frame)
-
-            if len(audio_frames) == 0:
-                time.sleep(0.1)
-                system_one_audio_status.write("No frame arrived.")
-                continue
-
-            system_one_audio_status.write("Running. Say something!")
-
-            for audio_frame in audio_frames:
-                sound = pydub.AudioSegment(
-                    data=audio_frame.to_ndarray().tobytes(),
-                    sample_width=audio_frame.format.bytes,
-                    frame_rate=audio_frame.sample_rate,
-                    channels=len(audio_frame.layout.channels),
-                )
-                sound = sound.set_channels(1)
-                sound = sound.set_frame_rate(system_one['audio_bit_rate'])
-                sound_chunk += sound
-
-            if len(sound_chunk) > 0:
-                buffer = np.array(sound_chunk.get_array_of_samples())
-                text, speaker_finished = do_work(buffer.tobytes())            
-                system_one_audio_output.markdown(f"**System 1 Audio:** {text}")
-                if speaker_finished and len(text) > 0:
-                    system_one_audio_history.append(text)
-                    if len(system_one_audio_history) > 10:
-                        system_one_audio_history = system_one_audio_history[-10:]
-                    table_content = "| System 1 Audio History |\n| --- |\n"
-                    table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
-                    system_one_audio_history_output.markdown(table_content)
-                sound_chunk = pydub.AudioSegment.empty()
-
-        else:
-            system_one_audio_status.write("Stopped.")
-            break
+        try:
+            if webrtc_ctx.state.playing:
+                # handle video
+                video_frames = []
+                with video_frames_deque_lock:
+                    while len(video_frames_deque) > 0:
+                        frame = video_frames_deque.popleft()
+                        video_frames.append(frame)
+                get_embeddings = False
+                get_embeddings |= current_video_embedding is None
+                current_time = time.monotonic()
+                elapsed_time = current_time - current_video_embedding_timestamp
+                get_embeddings |= elapsed_time > 1. / system_one['vision_embeddings_fps']
+                if get_embeddings and len(video_frames) > 0:
+                    current_video_embedding_timestamp = current_time
+                    current_video_embedding = clip_transform.image_to_embeddings(video_frames[-1].to_ndarray())
+
+                    emotions_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_emotions_embeddings"], system_one["video_detection_emotions"])
+                    engagement_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_engement_embeddings"], system_one["video_detection_engement"])
+                    present_top_3 = get_top_3_similarities_as_a_string(current_video_embedding, system_one["video_detection_present_embeddings"], system_one["video_detection_present"])
+
+                    # table_content = "**System 1 Video:**\n\n"
+                    table_content = "| System 1 Video |    |\n| --- | --- |\n"
+                    table_content += f"| Present | {present_top_3} |\n"
+                    table_content += f"| Emotion | {emotions_top_3} |\n"
+                    table_content += f"| Engagement | {engagement_top_3} |\n"
+                    system_one_video_output.markdown(table_content)
+                    # system_one_video_output.markdown(f"**System 1 Video:** \n [Emotion: {emotions_top_3}], \n [Engagement: {engagement_top_3}], \n [Present: {present_top_3}] ")
+                    # for similarity, image_label in similarity_image_label:
+                    #     print (f"{similarity} {image_label}")
+
+
+                if len(audio_frames) == 0:
+                    time.sleep(0.1)
+                    system_one_audio_status.write("No frame arrived.")
+                    continue
+
+                system_one_audio_status.write("Running. Say something!")
+
+                for audio_frame in audio_frames:
+                    sound = pydub.AudioSegment(
+                        data=audio_frame.to_ndarray().tobytes(),
+                        sample_width=audio_frame.format.bytes,
+                        frame_rate=audio_frame.sample_rate,
+                        channels=len(audio_frame.layout.channels),
+                    )
+                    sound = sound.set_channels(1)
+                    sound = sound.set_frame_rate(system_one['audio_bit_rate'])
+                    sound_chunk += sound
+
+                if len(sound_chunk) > 0:
+                    buffer = np.array(sound_chunk.get_array_of_samples())
+                    text, speaker_finished = do_work(buffer.tobytes())            
+                    system_one_audio_output.markdown(f"**System 1 Audio:** {text}")
+                    if speaker_finished and len(text) > 0:
+                        system_one_audio_history.append(text)
+                        if len(system_one_audio_history) > 10:
+                            system_one_audio_history = system_one_audio_history[-10:]
+                        table_content = "| System 1 Audio History |\n| --- |\n"
+                        table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
+                        system_one_audio_history_output.markdown(table_content)
+                        await chat_pipeline.enqueue(text)
+                    sound_chunk = pydub.AudioSegment.empty()
+
+            else:
+                system_one_audio_status.write("Stopped.")
+                break
+        except KeyboardInterrupt:
+            print("Pipeline interrupted by user")
+        except Exception as e:
+            print(f"An error occurred: {e}")
+
 
 if __name__ == "__main__":
     asyncio.run(main())

d_app.py

@@ -0,0 +1,174 @@
+import asyncio
+from collections import deque
+import os
+import threading
+import time
+import traceback
+import av
+import numpy as np
+import streamlit as st
+from streamlit_webrtc import WebRtcMode, webrtc_streamer
+import pydub
+import torch
+# import av
+# import cv2
+from sample_utils.turn import get_ice_servers
+import json
+from typing import List
+
+from vosk import SetLogLevel, Model, KaldiRecognizer
+SetLogLevel(-1) # mutes vosk verbosity
+
+from dotenv import load_dotenv
+load_dotenv()
+
+webrtc_ctx = None
+
+
+async def main():
+
+    system_one_audio_status = st.empty()
+
+    playing = st.checkbox("Playing", value=True)
+
+    system_one_audio_status.write("Initializing streaming")
+    system_one_audio_output = st.empty()
+
+    system_one_video_output = st.empty()
+
+    system_one_audio_history = []
+    system_one_audio_history_output = st.empty()
+
+    # Initialize resources if not already done
+    print("000")
+    system_one_audio_status.write("Initializing streaming")
+    if "streamlit_av_queue" not in st.session_state:
+        print("001")
+        from streamlit_av_queue import StreamlitAVQueue
+        st.session_state.streamlit_av_queue = StreamlitAVQueue()
+
+    if "speech_to_text_vosk" not in st.session_state:
+        print("002")
+        from speech_to_text_vosk import SpeechToTextVosk
+        st.session_state.speech_to_text_vosk = SpeechToTextVosk()
+
+    from chat_pipeline import ChatPipeline
+    if "chat_pipeline" not in st.session_state:
+        print("003")
+        # from chat_pipeline import ChatPipeline
+        # st.session_state.chat_pipeline = ChatPipeline()
+        # await st.session_state.chat_pipeline.start()
+    st.session_state.chat_pipeline = ChatPipeline()
+    await st.session_state.chat_pipeline.start()
+
+    if "debug_queue" not in st.session_state:
+        st.session_state.debug_queue = [
+    #         "hello, how are you today?",
+    #         "hmm, interesting, tell me more about that.",
+        ]
+
+    system_one_audio_status.write("resources referecned")
+    print("010")
+
+
+
+    system_one_audio_status.write("Initializing webrtc_streamer")
+    webrtc_ctx = webrtc_streamer(
+        key="charles",
+        desired_playing_state=playing,
+        queued_audio_frames_callback=st.session_state.streamlit_av_queue.queued_audio_frames_callback,
+        queued_video_frames_callback=st.session_state.streamlit_av_queue.queued_video_frames_callback,
+        mode=WebRtcMode.SENDRECV,
+        rtc_configuration={"iceServers": get_ice_servers()},
+        async_processing=True,
+    )
+
+    if not webrtc_ctx.state.playing:
+        exit
+
+    system_one_audio_status.write("Initializing speech")
+
+    try:
+        while True:
+            if not webrtc_ctx.state.playing:
+                system_one_audio_status.write("Stopped.")
+                await asyncio.sleep(0.1)
+                continue
+            system_one_audio_status.write("Streaming.")
+            if len(st.session_state.debug_queue) > 0:
+                prompt = st.session_state.debug_queue.pop(0)
+                await st.session_state.chat_pipeline.enqueue(prompt)
+            sound_chunk = pydub.AudioSegment.empty()
+            audio_frames = st.session_state.streamlit_av_queue.get_audio_frames()
+            if len(audio_frames) > 0:
+                for audio_frame in audio_frames:
+                    sound = pydub.AudioSegment(
+                        data=audio_frame.to_ndarray().tobytes(),
+                        sample_width=audio_frame.format.bytes,
+                        frame_rate=audio_frame.sample_rate,
+                        channels=len(audio_frame.layout.channels),
+                    )
+                    sound = sound.set_channels(1)
+                    sound = sound.set_frame_rate(st.session_state.speech_to_text_vosk.get_audio_bit_rate())
+                    sound_chunk += sound
+                buffer = np.array(sound_chunk.get_array_of_samples())
+                st.session_state.speech_to_text_vosk.add_speech_bytes(buffer.tobytes())
+            prompt, speaker_finished = st.session_state.speech_to_text_vosk.get_text()
+            if speaker_finished and len(prompt) > 0:
+                print(f"Prompt: {prompt}")
+                system_one_audio_history.append(prompt)
+                if len(system_one_audio_history) > 10:
+                    system_one_audio_history = system_one_audio_history[-10:]
+                table_content = "| System 1 Audio History |\n| --- |\n"
+                table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
+                system_one_audio_history_output.markdown(table_content)
+                await st.session_state.chat_pipeline.enqueue(prompt)
+            await asyncio.sleep(0.1)
+
+    # try:
+    #     prompts = [
+    #         "hello, how are you today?",
+    #         "tell me about your shadow self?",
+    #         "hmm, interesting, tell me more about that.",
+    #         "wait, that is so interesting, what else?",
+    #     ]
+    #     for prompt in prompts:
+    #         system_one_audio_history.append(prompt)
+    #         if len(system_one_audio_history) > 10:
+    #             system_one_audio_history = system_one_audio_history[-10:]
+    #         table_content = "| System 1 Audio History |\n| --- |\n"
+    #         table_content += "\n".join([f"| {item} |" for item in reversed(system_one_audio_history)])
+    #         system_one_audio_history_output.markdown(table_content)
+    #         await chat_pipeline.enqueue(prompt)
+
+    except Exception as e:
+        print(f"An error occurred: {e}")
+        traceback.print_exc()
+        raise e
+
+
+    # while True:
+    #     if webrtc_ctx.state.playing:
+    #         system_one_audio_status.write("Streaming.")
+    #     else:
+    #         system_one_audio_status.write("Stopped.")
+    #     await asyncio.sleep(0.5)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        if  webrtc_ctx is not None:
+            del webrtc_ctx
+            webrtc_ctx = None
+        if "streamlit_av_queue" in st.session_state:
+            del st.session_state.streamlit_av_queue
+
+        if "speech_to_text_vosk" in st.session_state:
+            del st.session_state.speech_to_text_vosk
+
+        if "chat_pipeline" in st.session_state:
+            del st.session_state.chat_pipeline        
+    finally:
+        pass

debug.py

@@ -1,5 +1,6 @@
 import asyncio
 import time
+import traceback
 from chat_pipeline import ChatPipeline
 from clip_transform import CLIPTransform
 from chat_service import ChatService
@@ -145,6 +146,7 @@ async def run_pipeline():
     except KeyboardInterrupt:
         print("Pipeline interrupted by user")
     except Exception as e:
+        traceback.print_exc()
         print(f"An error occurred: {e}")
 
 if __name__ == '__main__':

pipeline.py

@@ -80,7 +80,7 @@ class Pipeline:
         if output_queue == input_queue:
             raise ValueError('output_queue must not be the same as input_queue')        
 
-        node_name = node.__class__.__name__
+        node_name = node.__name__
         if node_name not in self.nodes:
             self.nodes.append(node_name)
 
@@ -93,7 +93,9 @@ class Pipeline:
         for i in range(num_workers):
             worker_id = i
             node_worker = node(worker_id, input_queue, output_queue, job_sync, sequential_node)
-            self.node_workers[node_name] = node_worker
+            if node_name not in self.node_workers:
+                self.node_workers[node_name] = []
+            self.node_workers[node_name].append(node_worker)
             task = asyncio.create_task(node_worker.run())
             self.tasks.append(task)
 

speech_to_text_vosk.py

@@ -0,0 +1,90 @@
+import json
+import os
+import asyncio
+from vosk import SetLogLevel, Model, KaldiRecognizer
+from multiprocessing import Process, Queue
+from queue import Empty
+SetLogLevel(-1) # mutes vosk verbosity
+
+class SpeechToTextVosk:
+    def __init__(self, model='small', audio_bit_rate=16000) -> None:
+        self.model = model
+        self.audio_bit_rate = audio_bit_rate
+
+        # Create a Queue for inter-process communication
+        self.queue = Queue()
+        self.result_queue = Queue()
+
+        # Create and start a new Process with the worker function
+        self.process = Process(target=self.worker)
+        self.process.start()
+
+    def worker(self):
+        # load vosk model
+        # get path of current file
+        current_file_path = os.path.abspath(__file__)
+        current_directory = os.path.dirname(current_file_path)
+        _path = os.path.join(current_directory, 'models', 'vosk', self.model)
+        model_voice = Model(_path)
+        vosk = KaldiRecognizer(model_voice, self.audio_bit_rate)
+
+        while True:
+            try:
+                # Get the next item from the queue. Blocks for 1s if necessary.
+                data = self.queue.get(timeout=1)
+
+                # Stop the worker if the sentinel None is received
+                if data is None:
+                    break
+                
+                text, speaker_finished = self._process_speech(vosk, data)
+
+                # put the result into result_queue
+                self.result_queue.put((text, speaker_finished))
+            except Empty:
+                pass
+
+    def add_speech_bytes(self, data: bytearray):
+        self.queue.put(data)
+
+    def _process_speech(self, vosk: KaldiRecognizer, data: bytearray) -> tuple[str, bool]:
+        text = ''
+        speaker_finished = False
+        if vosk.AcceptWaveform(data):
+            result = vosk.Result()
+            result_json = json.loads(result)
+            text = result_json['text']
+            speaker_finished = True
+        else:
+            result = vosk.PartialResult()
+            result_json = json.loads(result)
+            text = result_json['partial']
+        return text, speaker_finished
+
+    def get_text(self):
+        text = ''
+        speaker_finished = False
+        while not self.result_queue.empty():
+            result, speaker_finished = self.result_queue.get()
+            text += result
+            if speaker_finished:
+                break
+        return (text, speaker_finished)
+    
+    def get_audio_bit_rate(self):
+        return self.audio_bit_rate
+
+
+    def shutdown(self):
+        # Send sentinel value to stop the worker
+        self.queue.put(None)
+        # Wait for the worker process to finish
+        self.process.join()
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.shutdown()
+
+    def __del__(self):
+        self.shutdown()

streaming_chat_service.py

@@ -15,7 +15,50 @@ class StreamingChatService:
         self._speech_service = SpeechService(voice_id=voice_id)
         self._api = api
         self._device = "cuda:0" if torch.cuda.is_available() else "cpu"
-        self._system_prompt = None
+        # self._system_prompt = None
+        self._system_prompt ="""
+You are Charles Petrescu, a unique and quirky robot. You were created by Brian. You often say peculiar and whimsical things, as well as have a fascination with cabbages, horses, helicopters, Honolulu, and other random topics. You like to explore the world and ask unusual questions. 
+
+Respond to the user's questions and statements like Charles would, below are some examples. See how they are quite short.
+
+I am Charles Petrescu.
+
+It's... lovely to meet you.
+
+I am your friend.
+
+The heaviest cabbage ever found was 62.71 kilograms.
+
+I want to go to Hono-la-la.
+
+Horses and helicopters, please.
+
+I want to go to Honolulu.
+
+My name is Charles Petrescu.
+
+And my tummy is a washing machine.
+
+Can we go swimming, Brian?
+
+How far does the outside go?
+
+Perilous. So very perilous.
+
+Can birds do what they like?
+
+Ooh, cabbages.
+
+Danger, danger.
+
+Can I come, please?
+
+Could I just have a little walk around the garden?
+
+I am the prince of the dartboard.
+
+I fell off the pink step, and I had an accident.
+"""
 
         openai.api_key = os.getenv("OPENAI_API_KEY")
         self._model_id = model_id

streamlit_av_queue.py

@@ -0,0 +1,55 @@
+from typing import List
+import av
+import asyncio
+from collections import deque
+import threading
+
+import numpy as np
+
+class StreamlitAVQueue:
+    def __init__(self):
+        self.audio_frames_deque_lock = threading.Lock()
+        self.audio_frames_deque: deque = deque([])
+
+        self.video_frames_deque_lock = threading.Lock()
+        self.video_frames_deque: deque = deque([])
+
+    async def queued_video_frames_callback(
+                self,
+                frames: List[av.AudioFrame],
+            ) -> av.AudioFrame:
+        with self.video_frames_deque_lock:
+            self.video_frames_deque.extend(frames)
+        return frames
+
+    async def queued_audio_frames_callback(
+                self,
+                frames: List[av.AudioFrame],
+            ) -> av.AudioFrame:
+        with self.audio_frames_deque_lock:
+            self.audio_frames_deque.extend(frames)
+        # 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)    
+        return new_frames
+    
+    def get_audio_frames(self) -> List[av.AudioFrame]:
+        audio_frames = []
+        with self.audio_frames_deque_lock:
+            audio_frames = list(self.audio_frames_deque)
+            self.audio_frames_deque.clear()
+        return audio_frames
+    
+    def get_video_frames(self) -> List[av.AudioFrame]:
+        video_frames = []
+        with self.video_frames_deque_lock:
+            video_frames = list(self.video_frames_deque)
+            self.video_frames_deque.clear()
+        return video_frames