In my last article, I started a project that used speech recognition to create a to-do list, while also storing it in a database. It used a microphone to record but saved the audio to a file to then be read by the system. This is not exactly real-time. So, this time, I decided to make the project work in real time. In case you didn’t read the last , here’s a recap of the project. article Eventually, my goal is to have a smart home. One device which I desire, in particular, is a smart mirror. However, I want to build it myself, so I’ve got projects on the go that will hopefully allow me to slowly learn how that could be done. For the smart mirror, I wanted the ability to say my to-do list out loud, and have that stored in a database to read back later. That way, I won’t forget everything later down the road. My first attempt recorded a file, then used that, but in this attempt I want the audio to be transcribed in real-time. I’m still using AssembyAI for the speech recognition transcription just like in my first article. However, as this will be in real-time, I had to upgrade to the pro plan to have that feature available. So, without any further delay, let’s talk about first what I needed to install for this project. If I already installed it in the last blog I will still include it in this one. That way if anyone missed the last article they can still follow along with what I did for my project. Installing a Few Things For this to be real-time, I will need to make continuous calls to the API. But that means I would need a connection that is continuous. That’s where WebSockets come in. Websockets in the Python library can establish this type of connection so I can feed the audio and also see the transcribed output. To install, use the following command: pip3 install websockets In the last speech recognition article I did, I used , so I already have that installed. However, there were a few other packages needed for this to work, so I used the following line: libportaudio2 sudo apt-get install libasound-dev portaudio19-dev libportaudio2 libportaudiocpp0 Instead of using in this part, I decided to use . This is needed for more input/output controls for the audio. sounddevice PyAudio pip3 install PyAudio To run this, it needed to be done asynchronously. For that I chose asyncio. That’s a provisional package from Python, so it may not work if you’re using a version of Python earlier than 3.4. I can’t remember if this package was something that came automatically, but should you need to install it you can use this command: pip3 install asyncio For this part, I used TinyDB this time as well to store my to-do list. That was installed in my last part, but in case you missed that you use this simple command: pip3 install tinydb Now that everything is installed, let’s jump straight into the code. The Real-Time Class File I wrote this code in two files. The first is the class with a few different methods for real-time speech, the WebSocket connection, and the sending and receiving of the audio. The second file is more of just the driver that calls the first method to get the ball rolling, and also provides the API key. The first thing is first though, let’s start with the file I called . stt_real_time.py Let’s go over all the imports I needed for that. So, we already know you need the asyncio, WebSockets, and tinydb, but for pyaudio, plus a few others. asyncio websockets pyaudio base64 json tinydb TinyDB, Query import import import import import from import Next, I’m ready to declare the class. Inside the class, I’ll also include the initializing function. There, I will need to create the endpoint, the database, set my recording rate like in the last part, and also open a stream with PyAudio using those frame rates. I also use the opening empty string at the beginning of the class. “”” “”” self.endpoint = “wss://api.assemblyai.com/v2/realtime/ws?sample_rate= ” self.key = key self.rate = self.frames_per_buffer = self.db = TinyDB(real_time_db.json) p = pyaudio.PyAudio() self.stream = p.open( frames_per_buffer=self.frames_per_buffer, rate=self.rate, format=pyaudio.paInt16, channels= , input= ) : class stt_real_time : def __init__ (self, key) 16000 16000 3200 1 True With the class initialized, we can move on to the next method. This will be where most of the work is done. The will be called , and within I created two functions. One to send and one to receive. So, the next few code segments will be inside this real_time method, but I will explain as I go. real_time real_time To begin with the method, I started by opening the WebSocket connection. While this could remain open until I terminate the project, I decided to add a timeout after 20 seconds. aysnc websockets.connect( self.endpoint, ping_interval= , ping_timeout= , extra_headers=((“Authorization”, self.key), ) ) connection: asyncio.sleep( ) async : def real_time (self) with 5 20 as await 0.1 With the connection now created, the rest of the code in the real_time method will be inside of that connection, so make sure it is indented within the connection. We can now begin the session: session_begins = connection.recv() print(session_begins) await Next, I created my send function, which listens to the audio from the microphone input, then sends a jsonified dump of the data through the connection. Here is also where we need a little error handling in case something goes wrong. : : data = self.stream.read(self.frames_per_buffer) data = base64.b64encode(data).decode(“utf ”) connection.send(json.dumps({“audio_data”: str(data)})) Exception ex: print(f”An error occurred: {ex}”) asyncio.sleep( ) async : def send () while True try -8 await except as break await 0.01 return True With the send function out of the way, I moved on to the receive function. In this function, I had to wait until the message was finished, which created the “Final Transcript”. It’s that final one that will be inserted into the database. For example, when I begin talking, the system will start displaying the text to what I am saying, but it will try to correct itself into a sentence structure as I go. Although it won’t always look perfect, once I finish talking it creates the final version of what I said transcribed. Here’s just an example of the transcript being read in and finalized: So, this means nothing but the very last line, the Final Transcript, will be inserted into the database. We will also need error handling on this step, as we did in the send function. : : message = connection.recv() json.loads(message)[“message_type”] == “FinalTranscript”: self.db.insert({“todo:” json.loads(message)[“text”]}) self.__list_todos() Exception ex: print(f”An error occurred: {ex}”) async : def receive () while True try await if except as break With the receive function written, we need to allow asyncio to gather both. So this next piece of code will be outside of the receive function but still inside of our real_time method. sent, received = asyncio.gather(send(), receive()) await That will conclude the real_time method. At this point, if you’re following my code you likely have an error on self.__list_todos() because we have not created that method yet. But we’ll do that now and it will be the last code in the file. All we’re going to do there is print each item from the database like so: print(“Todo List:”) item self.db: print(item[“todo”]) : def __list_todos (self) for in That will be all for this file, so let’s get started on the file to test and call the methods. Just for continuity, here is all the code for this file in one place, just for something easier to read: asyncio websockets pyaudio base64 json tinydb TinyDB, Query self.endpoint = self.key = key self.rate = self.frames_per_buffer = self.db = TinyDB( ) p = pyaudio.PyAudio() self.stream = p.open( frames_per_buffer=self.frames_per_buffer, rate=self.rate, format=pyaudio.paInt16, channels= , input= ) websockets.connect( self.endpoint, ping_interval= , ping_timeout= , extra_headers=(( , self.key), ) ) connection: asyncio.sleep( ) session_begins = connection.recv() print(session_begins) : : data = self.stream.read(self.frames_per_buffer) data = base64.b64encode(data).decode( ) connection.send(json.dumps({ : str(data)})) Exception ex: print( ) asyncio.sleep( ) : : message = connection.recv() json.loads(message)[ ] == : self.db.insert({ : json.loads(message)[ ]}) self.__list_todos() Exception ex: print( ) sent, received = asyncio.gather(send(), receive()) print( ) item self.db: print(item[ ]) import import import import import from import : class stt_real_time """ """ : def __init__ (self, key) "wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000" 16000 3200 'real_time_db.json' 1 True async : def real_time (self) async with 5 20 "Authorization" as await 0.1 await async : def send () while True try "utf-8" await "audio_data" except as f"An error occurred: " {ex} break await 0.01 return True async : def receive () while True try await # print(json.loads(message)["text"]) if "message_type" "FinalTranscript" "todo" "text" except as f"An error occurred: " {ex} break await : def __list_todos (self) "Todo List:" for in "todo" Writing the Test File I called this file . To start with the imports, I only needed and the class from the other file: test.py asyncio asyncio stt_real_time stt_real_time import from import The next step was simply to initialize the class, where I needed to also send my API key. real_time_stt = stt_real_time(YOURAPIKEY) There’s only one last step, and that’s to create the loop that will run asynchronously. All we have to do there is call the real_time method, which will call the others once things kick-off. I finished my code with these lines: : asyncio.run(real_time_stt.real_time()) while True The only thing left to do is testing. Now, again, I want to mention that the timeout was set for 20 seconds for the WebSocket connection, which you can always have longer for your project depending on what you need. But for testing and a simple to-do list generator, I didn’t need much time. As far as printing results, I also only print the final list of all records in the database, so as I add more to the list I will see the original input first, and the most recent last. You could always filter to view only the most recent for your project, but I wanted to see everything added to my to-do list. The results for the project running will print the session being created, as I displayed for the SessionBegins in my first screenshot, but then the output will be the list I say into the microphone in real-time. And with that, we’ve successfully run the code. The transcription was done quickly, and as long as I spoke clearly the speech recognition was very accurate. Conclusion As I work closer to building my own smart home devices, my smart mirror needed a way to handle speech recognition. Using the AssemblyAi API, I was able to build my to-do list generator using real-time speech recognition. The real-time recognition was built using a combination of the API, asyncio, and WebSocket. Overall, I thought following the tutorials available on AssemblyAi’s website was useful, but it was also highly adaptable for my to-do list project. Hopefully, you found my project helpful in building your own, and feel free to leave a comment on what you used to build your project. Until next time, cheers! References https://www.assemblyai.com/blog/real-time-speech-recognition-with-python/?_ga=2.201721049.918789168.1640546390-81735733.1640546390 https://www.assemblyai.com/blog/real-time-speech-recognition-with-assemblyai/ https://docs.assemblyai.com/walkthroughs#realtime-streaming-transcription https://websockets.readthedocs.io/en/stable/
