AI on the Edge LESSON 36: Select Active Camera in OpenCV With Voice Commands

July 24, 2026 admin

Welcome back, makers, engineers, and AI enthusiasts! In our previous lessons, we built out robust multi-camera setups, streaming feeds from both Raspberry Pi cameras and high-definition USB cameras. But watching four feeds in separate tiles is only half the battle. What happens when you want to interact with your system hands-free?

In this lesson, we are taking our edge vision projects to the next level by integrating voice commands. You will learn how to run a dedicated speech-to-text thread in the background, catch voice triggers safely using a thread-safe queue, and dynamically switch your active main camera feed in OpenCV on the fly—just by speaking.

What You Will Learn in This Lesson

Multi-Threaded Speech Recognition: How to run the stt listener in a separate daemon thread so it never blocks or stutters your high-FPS video processing loop.
Thread-Safe Communication: Using Python’s Queue module to pass voice commands seamlessly from the background listening thread into your main application loop.
Handling STT Variations: Accounting for common speech-to-text homophone variations (like “two”, “to”, and “too”, or “four” and “for”) to make your voice control robust and reliable.
Dynamic Frame Routing: Mapping spoken commands to specific camera streams (piCam1, piCam2, usbCam1, and usbCam2) and updating your primary display window instantly.
Organized Window Layouts: Positioning and sizing multiple OpenCV windows on your desktop workspace for a clean, professional multi-camera dashboard.

Step-by-Step Breakdown of the Script

1. Setting Up the Background Voice Thread

When working with real-time video processing in OpenCV, blocking functions are your worst enemy. If you call a speech recognition listening function directly inside your main while loop, your video frames will freeze while waiting for audio input.

To prevent this, we initialize a background worker function (getCamera) and launch it as a daemon thread:

The thread continuously listens for your voice commands using the Fusion Hat STT module.
Once a command is captured, it strips whitespace, checks for exit triggers (like saying 'quit'), and pushes valid commands directly into our commandQ.

2. Initializing Multiple Cameras

Our setup harnesses the full power of the hardware by combining native Raspberry Pi camera interfaces with standard USB webcams:

Pi Cameras: Configured via Picamera2 at a crisp 1280x720 resolution with RGB888 formatting running smoothly at 60 frames per second.
USB Cameras: Initialized through OpenCV’s VideoCapture class, set to target resolution and optimized for 30 FPS.

3. Processing and Routing Commands in the Main Loop

Inside the primary application loop, our script handles three core tasks simultaneously:

Calculate Performance: Continuously tracks and smooths out the frames-per-second (FPS) metric so you can monitor system load.
Check the Command Queue: Non-blockingly checks if the background thread has dropped a new camera selection into commandQ. If a new command is waiting, it updates the mainCam variable.
Route the Main Frame: Evaluates the active camera string—including clever fallback checks for common voice misinterpretations like 'camera to' or 'camera for'—and assigns the corresponding video stream to mainFrame.

4. Managing Your OpenCV Windows

To give you a complete command-center experience, the script generates a large primary display window for your active camera view, accompanied by a clean row of smaller preview tiles across the bottom of your screen for all four connected feeds.

import cv2
import time
from picamera2 import Picamera2
import threading
from queue import Queue
from fusion_hat.stt import STT
stt = STT('en-us')
commandQ = Queue()
mainCam = 'camera one'

W=1280
H=720
tStart = time.time()
fps = 0
RES = (W,H)

piCam1 = Picamera2(0)
piCam1.preview_configuration.main.size = RES
piCam1.preview_configuration.main.format = "RGB888"
piCam1.preview_configuration.controls.FrameRate=60
piCam1.preview_configuration.align()
piCam1.configure("preview")
piCam1.start()

piCam2 = Picamera2(1)
piCam2.preview_configuration.main.size = RES
piCam2.preview_configuration.main.format = "RGB888"
piCam2.preview_configuration.controls.FrameRate=60
piCam2.preview_configuration.align()
piCam2.configure("preview")
piCam2.start()

usbCam1 = cv2.VideoCapture(16)
usbCam1.set(cv2.CAP_PROP_FRAME_WIDTH,W)
usbCam1.set(cv2.CAP_PROP_FRAME_HEIGHT,H)
usbCam1.set(cv2.CAP_PROP_FPS, 30)

usbCam2 = cv2.VideoCapture(18)
usbCam2.set(cv2.CAP_PROP_FRAME_WIDTH,W)
usbCam2.set(cv2.CAP_PROP_FRAME_HEIGHT,H)
usbCam2.set(cv2.CAP_PROP_FPS, 30)

textLowerLeft = (int(W*.01),int(H*.05))
fontFace = cv2.FONT_HERSHEY_SIMPLEX
fontThickness = int(W/425)
fontScale = H*.0015
fontColor = (0,0,255)

running = True
def getCamera():
    print('Voice Thread Started')
    global running
    while running:
        myCamera = stt.listen(stream = False)
        myCamera = myCamera.strip()
        print(myCamera)
        if myCamera == 'quit':
            print('Security Camera Offline')
            running = False
            break
        commandQ.put(myCamera)
    print('Thread Terminated')

cameraThread = threading.Thread(target = getCamera,daemon=True)
cameraThread.start()

cv2.namedWindow('Main Camera',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Main Camera',0,65)
cv2.resizeWindow('Main Camera',W,H)

cv2.namedWindow('Camera1',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera1',0,H)
cv2.resizeWindow('Camera1',int(W/4),int(H/4))

cv2.namedWindow('Camera2',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera2',int(W/4),H)
cv2.resizeWindow('Camera2',int(W/4),int(H/4))

cv2.namedWindow('Camera3',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera3',2*int(W/4),H)
cv2.resizeWindow('Camera3',int(W/4),int(H/4))

cv2.namedWindow('Camera4',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera4',3*int(W/4),H)
cv2.resizeWindow('Camera4',int(W/4),int(H/4))

while running:
    deltaT = time.time() - tStart
    tStart=time.time()
    fps = fps*.95 + (1/deltaT)*.05
    frame1= piCam1.capture_array()
    #frame1=cv2.flip(frame1,-1)
    
    frame2= piCam2.capture_array()
    frame2=cv2.flip(frame2,-1)
    
    _ , frame3 = usbCam1.read()
    _ , frame4 = usbCam2.read()
    mainFrame = frame1
    
    if commandQ.empty() == False:
        mainCam = commandQ.get()
        
    if mainCam == 'camera one':
        mainFrame = frame1
    if mainCam == 'camera two' or mainCam == 'camera to' or mainCam == 'camera too':
        mainFrame = frame2
    if mainCam == 'camera three':
        mainFrame = frame3
    if mainCam == 'camera four' or mainCam == 'camera for':
        mainFrame = frame4
    myText = "FPS: "+str(round(fps,1))
    cv2.putText(frame1,myText,textLowerLeft,fontFace,fontScale,fontColor,fontThickness)
    cv2.imshow("Main Camera", mainFrame)
    cv2.imshow("Camera1", frame1)
    cv2.imshow("Camera2", frame2)
    cv2.imshow("Camera3", frame3)
    cv2.imshow("Camera4", frame4)    
    if cv2.waitKey(1)==ord('q'):
        break
cv2.destroyAllWindows()
print('Program Terminated')

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

import cv2

import time

from picamera2 import Picamera2

import threading

from queue import Queue

from fusion_hat.stt import STT

stt = STT('en-us')

commandQ = Queue()

mainCam = 'camera one'

W=1280

H=720

tStart = time.time()

fps = 0

RES = (W,H)

piCam1 = Picamera2(0)

piCam1.preview_configuration.main.size = RES

piCam1.preview_configuration.main.format = "RGB888"

piCam1.preview_configuration.controls.FrameRate=60

piCam1.preview_configuration.align()

piCam1.configure("preview")

piCam1.start()

piCam2 = Picamera2(1)

piCam2.preview_configuration.main.size = RES

piCam2.preview_configuration.main.format = "RGB888"

piCam2.preview_configuration.controls.FrameRate=60

piCam2.preview_configuration.align()

piCam2.configure("preview")

piCam2.start()

usbCam1 = cv2.VideoCapture(16)

usbCam1.set(cv2.CAP_PROP_FRAME_WIDTH,W)

usbCam1.set(cv2.CAP_PROP_FRAME_HEIGHT,H)

usbCam1.set(cv2.CAP_PROP_FPS, 30)

usbCam2 = cv2.VideoCapture(18)

usbCam2.set(cv2.CAP_PROP_FRAME_WIDTH,W)

usbCam2.set(cv2.CAP_PROP_FRAME_HEIGHT,H)

usbCam2.set(cv2.CAP_PROP_FPS, 30)

textLowerLeft = (int(W*.01),int(H*.05))

fontFace = cv2.FONT_HERSHEY_SIMPLEX

fontThickness = int(W/425)

fontScale = H*.0015

fontColor = (0,0,255)

running = True

def getCamera():

print('Voice Thread Started')

global running

while running:

myCamera = stt.listen(stream = False)

myCamera = myCamera.strip()

print(myCamera)

if myCamera == 'quit':

print('Security Camera Offline')

running = False

break

commandQ.put(myCamera)

print('Thread Terminated')

cameraThread = threading.Thread(target = getCamera,daemon=True)

cameraThread.start()

cv2.namedWindow('Main Camera',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Main Camera',0,65)

cv2.resizeWindow('Main Camera',W,H)

cv2.namedWindow('Camera1',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera1',0,H)

cv2.resizeWindow('Camera1',int(W/4),int(H/4))

cv2.namedWindow('Camera2',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera2',int(W/4),H)

cv2.resizeWindow('Camera2',int(W/4),int(H/4))

cv2.namedWindow('Camera3',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera3',2*int(W/4),H)

cv2.resizeWindow('Camera3',int(W/4),int(H/4))

cv2.namedWindow('Camera4',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera4',3*int(W/4),H)

cv2.resizeWindow('Camera4',int(W/4),int(H/4))

while running:

deltaT = time.time() - tStart

tStart=time.time()

fps = fps*.95 + (1/deltaT)*.05

frame1= piCam1.capture_array()

#frame1=cv2.flip(frame1,-1)

frame2= piCam2.capture_array()

frame2=cv2.flip(frame2,-1)

_ , frame3 = usbCam1.read()

_ , frame4 = usbCam2.read()

mainFrame = frame1

if commandQ.empty() == False:

mainCam = commandQ.get()

if mainCam == 'camera one':

mainFrame = frame1

if mainCam == 'camera two' or mainCam == 'camera to' or mainCam == 'camera too':

mainFrame = frame2

if mainCam == 'camera three':

mainFrame = frame3

if mainCam == 'camera four' or mainCam == 'camera for':

mainFrame = frame4

myText = "FPS: "+str(round(fps,1))

cv2.putText(frame1,myText,textLowerLeft,fontFace,fontScale,fontColor,fontThickness)

cv2.imshow("Main Camera", mainFrame)

cv2.imshow("Camera1", frame1)

cv2.imshow("Camera2", frame2)

cv2.imshow("Camera3", frame3)

cv2.imshow("Camera4", frame4)

if cv2.waitKey(1)==ord('q'):

break

cv2.destroyAllWindows()

print('Program Terminated')

NVIDIA, NVIDIA Jetson Orin Nano

Local Voice Control of NVIDIA Jetson Orin Nano with STT: Getting Started with Vosk

July 22, 2026 admin

Engineering Your Own Local Voice Assistant: No Cloud, No Compromise

Most “smart” voice assistants are just glorified remote controls for someone else’s server. Today, we’re changing that. We are going to build a local, offline voice command pipeline. This isn’t just about saving data; it’s about ownership. When you can control your hardware—like opening and closing a farm gate—without an internet connection, you have built a system that is robust, private, and yours to control forever. Today you are going to get Speech to Text up and running on your NVIDIA Jetson Orin Nano running under Jetpack 7.2.

The “Why” Behind the Setup

You might ask, “Why not just use a cloud API?” Because cloud APIs are fragile. They rely on internet stability, external servers, and privacy-invasive data logging. By running Vosk locally, we keep the processing on your hardware (like the NVIDIA Jetson). It’s faster, it works in the middle of a power-isolated homestead, and it’s 100% secure.

Part 1: Preparing the Environment

Before we can make the machine listen, we have to prepare the battlefield. We aren’t just downloading files; we are setting up a stable environment where your dependencies won’t conflict with your OS.

# Create a dedicated directory so our work stays organized
mkdir -p ~/STT
cd ~/STT

# We need the PortAudio headers. Why? Because Python’s 'pyaudio' is just a wrapper.
# Under the hood, it talks to the C-based PortAudio library. If the headers aren't there,
# the library won't compile, and your microphone will never wake up.
sudo apt update
sudo apt install -y portaudio19-dev wget unzip

# Download the lightweight, low-latency English model weights from Vosk
wget https://alphacephei.com/vosk/models/vosk-model-small-en-us-0.15.zip

# Unzip the archive and rename the folder cleanly to 'model' so the Python script finds it
unzip vosk-model-small-en-us-0.15.zip
mv vosk-model-small-en-us-0.15 model

# Clean up the downloaded zip file to keep the folder pristine
rm vosk-model-small-en-us-0.15.zip

# Setting up a Virtual Environment (venv) is the hallmark of a pro.
# It traps our project dependencies inside this folder so we don't break our system Python.
python3 -m venv sttVenv
source sttVenv/bin/activate
pip install --upgrade pip

# Vosk does the heavy lifting for speech recognition, and Pyaudio grabs the raw data.
pip install vosk pyaudio

# Create a dedicated directory so our work stays organized

mkdir -p ~/STT

cd ~/STT

# We need the PortAudio headers. Why? Because Python’s 'pyaudio' is just a wrapper.

# Under the hood, it talks to the C-based PortAudio library. If the headers aren't there,

# the library won't compile, and your microphone will never wake up.

sudo apt update

sudo apt install -y portaudio19-dev wget unzip

# Download the lightweight, low-latency English model weights from Vosk

wget https://alphacephei.com/vosk/models/vosk-model-small-en-us-0.15.zip

# Unzip the archive and rename the folder cleanly to 'model' so the Python script finds it

unzip vosk-model-small-en-us-0.15.zip

mv vosk-model-small-en-us-0.15 model

# Clean up the downloaded zip file to keep the folder pristine

rm vosk-model-small-en-us-0.15.zip

# Setting up a Virtual Environment (venv) is the hallmark of a pro.

# It traps our project dependencies inside this folder so we don't break our system Python.

python3 -m venv sttVenv

source sttVenv/bin/activate

pip install --upgrade pip

# Vosk does the heavy lifting for speech recognition, and Pyaudio grabs the raw data.

pip install vosk pyaudio

This is IMPORTANT!

Now you can post the code below. You also have to point Thonny to run in the virtual environment. Open Thonny, and under run – select interpreter. Then you must point it to /home/yourUserName/STT/ttsVenv/bin/python3. For me, my username is pjm, but you put in your user name in path above. Here is what mine looked like:

Part 2: Solving the PipeWire Challenge

The biggest headache in modern Linux audio is PipeWire. If you try to open a microphone stream using a hardcoded sample rate that doesn’t match your hardware, your program won’t just fail—it will segfault. We use the validation script below to programmatically query the hardware, asking it: “What sample rate are you running at?” before we even try to open a stream.

# Import the PyAudio library to handle real-time microphone recording and audio stream manipulation.
import pyaudio
# Import the built-in JSON module to parse the structured text results returned by the speech recognition engine.
import json
# Import the audioop library to perform real-time mathematical operations and downsampling on raw audio byte data.
import audioop
# Import the core Vosk objects: Model loads the AI brain/weights, and KaldiRecognizer handles the acoustic processing.
from vosk import Model, KaldiRecognizer
# Import the standard OS module to reliably build absolute system file paths across different user directories.
import os
state=None

# Define a function to discover the specific hardware index and sample rate of the active system sound server.
def get_pipewire_index():
    # Instantiate a PyAudio session object to query the operating system's hardware configuration list.
    p = pyaudio.PyAudio()
    # Loop through every single available audio device reported by the operating system kernel.
    for i in range(p.get_device_count()):
        # Retrieve a descriptive configuration dictionary for the current audio device index.
        dev = p.get_device_info_by_index(i)
        # Check if the name of the device contains 'pipewire' to target the modern Linux audio subsystem.
        if "pipewire" in dev.get('name', '').lower():
            # Return the valid device index integer and its hardware default sample rate, converted cleanly to an integer.
            return i, int(dev.get('defaultSampleRate'))
    # Return empty values if the loop completes without finding an active PipeWire sound server profile.
    return None, None

# Execute the discovery function to capture both the correct system hardware device index and its native sample rate.
dev_index, native_rate = get_pipewire_index()

# Check if the hardware discovery handshake failed to locate an active PipeWire audio interface.
if dev_index is None:
    # Alert the user on screen that the script cannot communicate with the system's active audio server layer.
    print("Error: Could not find 'pipewire' device.")
    # Stop the script execution immediately since it is impossible to record audio without a valid device hook.
    exit()

# Instantiate a fresh PyAudio session specifically dedicated to opening our incoming recording channel.
p = pyaudio.PyAudio()
# Open the active system recording stream channel using explicit audio hardware parameter configurations.
# PARAMETER MANUAL:
# - format=pyaudio.paInt16: Sets 16-bit Signed Integer PCM format. (LEAVE ALONE - required by the Vosk engine).
# - channels=1: Sets recording to Mono channel audio. (LEAVE ALONE - Vosk processes single-channel input).
# - rate=native_rate: Sets the sample rate to match the mic exactly. (LEAVE ALONE - prevents driver hardware failure).
# - input=True: Configures this stream as an audio capture device. (LEAVE ALONE - required to wake up the microphone).
# - input_device_index=dev_index: Points to our discovered target device. (LEAVE ALONE - ensures we grab the right mic).
# - frames_per_buffer=4096: Controls audio chunk size. (PLAY WITH: Lower values like 2048 reduce latency but risk buffer overflows; higher values like 8192 increase safety but introduce slight latency).
stream = p.open(format=pyaudio.paInt16, channels=1, rate=native_rate, 
                input=True, input_device_index=dev_index, frames_per_buffer=4096)

# Generate an absolute directory path targeting the exact folder containing your extracted acoustic model weights.
# PARAMETER MANUAL:
# - "STT", "model": Represents folder names. (PLAY WITH: Change these only if your physical folder layout uses different names).
model_path = os.path.join(os.path.expanduser("~"), "STT", "model")
# Load the heavy machine learning weights into system memory to initialize the Vosk voice processing brain.
model = Model(model_path)
# Initialize the primary speech recognizer object using the loaded model weights and target processing frequency.
# PARAMETER MANUAL:
# - model: The initialized AI brain variable. (LEAVE ALONE).
# - 16000: Specifies the target processing sample rate in Hertz. (LEAVE ALONE - Vosk models are mathematically trained for exactly 16000Hz).
recognizer = KaldiRecognizer(model, 16000)

# Print a status message to the terminal showing that the handshake succeeded and revealing the mic's operating frequency.
print(f"Listening on PipeWire at {native_rate}Hz...")
# Turn on the hardware recording valve to begin filling up the system input data buffers.
stream.start_stream()

# Set up an exception safety block to intercept manual user exits gracefully without leaving system resources locked up.
try:
    # Launch the infinite real-time audio capture loop to continuously monitor the microphone stream.
    while True:
        # Pull raw binary audio frame bytes directly out of the incoming hardware stream buffer memory block.
        # PARAMETER MANUAL:
        # - 4096: The chunk size to read. (LEAVE ALONE - must perfectly match your stream's frames_per_buffer setting).
        # - exception_on_overflow=False: Prevents Python crashes if the Orin momentarily falls behind. (LEAVE ALONE).
        data = stream.read(4096, exception_on_overflow=False)
        # Downsample the raw audio bytes mathematically to 16000Hz using digital filtering algorithms.
        # PARAMETER MANUAL:
        # - data: Raw input audio byte string. (LEAVE ALONE).
        # - 2: Size of sample width in bytes for 16-bit sound. (LEAVE ALONE).
        # - 1: Input audio channel count. (LEAVE ALONE).
        # - native_rate: The native frequency of your mic. (LEAVE ALONE).
        # - 16000: The required downsampled output frequency. (LEAVE ALONE).
        # - state: The digital filtering conversion state history object. (LEAVE ALONE).
        # CRITICAL NOTE: The 'state' variable must be initialized as 'state = None' before entering this 'try:' block, 
        # otherwise Python will throw a NameError on the very first loop iteration because 'state' does not exist yet.
        data_16k, state = audioop.ratecv(data, 2, 1, native_rate, 16000, state)
        
        # Feed the processed 16kHz audio data block into Vosk to evaluate if a complete spoken phrase has finished.
        if recognizer.AcceptWaveform(data_16k):
            # Parse the structured JSON text string returned by the acoustic model into a standard Python dictionary.
            res = json.loads(recognizer.Result())
            # Check the parsed result dictionary to ensure that the detected text sequence is not completely blank.
            if res.get("text"):
                # Cleanly output the decoded text command straight to the standard system terminal display screen.
                print("Command: "+res['text'])
# Intercept the user pressing Ctrl+C in the terminal to stop the continuous script loop cleanly.
except KeyboardInterrupt:
    # Print a clean closing status notification on a new line to confirm the signal was intercepted.
    print("\nStopped.")
# Execute mandatory clean up instructions that are guaranteed to run no matter how the script terminates or crashes.
finally:
    # Turn off the active recording stream data valve to stop pulling raw information from the audio driver layer.
    stream.stop_stream()
    # Close the recording stream channel completely to drop the active hardware pointer binding handles.
    stream.close()
    # Terminate the parent PyAudio engine instance to release all allocated audio device resources back to the OS.
    p.terminate()

100

101

102

103

104

105

106

107

108

# Import the PyAudio library to handle real-time microphone recording and audio stream manipulation.

import pyaudio

# Import the built-in JSON module to parse the structured text results returned by the speech recognition engine.

import json

# Import the audioop library to perform real-time mathematical operations and downsampling on raw audio byte data.

import audioop

# Import the core Vosk objects: Model loads the AI brain/weights, and KaldiRecognizer handles the acoustic processing.

from vosk import Model, KaldiRecognizer

# Import the standard OS module to reliably build absolute system file paths across different user directories.

import os

state=None

# Define a function to discover the specific hardware index and sample rate of the active system sound server.

def get_pipewire_index():

# Instantiate a PyAudio session object to query the operating system's hardware configuration list.

p = pyaudio.PyAudio()

# Loop through every single available audio device reported by the operating system kernel.

for i in range(p.get_device_count()):

# Retrieve a descriptive configuration dictionary for the current audio device index.

dev = p.get_device_info_by_index(i)

# Check if the name of the device contains 'pipewire' to target the modern Linux audio subsystem.

if "pipewire" in dev.get('name', '').lower():

# Return the valid device index integer and its hardware default sample rate, converted cleanly to an integer.

return i, int(dev.get('defaultSampleRate'))

# Return empty values if the loop completes without finding an active PipeWire sound server profile.

return None, None

# Execute the discovery function to capture both the correct system hardware device index and its native sample rate.

dev_index, native_rate = get_pipewire_index()

# Check if the hardware discovery handshake failed to locate an active PipeWire audio interface.

if dev_index is None:

# Alert the user on screen that the script cannot communicate with the system's active audio server layer.

print("Error: Could not find 'pipewire' device.")

# Stop the script execution immediately since it is impossible to record audio without a valid device hook.

exit()

# Instantiate a fresh PyAudio session specifically dedicated to opening our incoming recording channel.

p = pyaudio.PyAudio()

# Open the active system recording stream channel using explicit audio hardware parameter configurations.

# PARAMETER MANUAL:

# - format=pyaudio.paInt16: Sets 16-bit Signed Integer PCM format. (LEAVE ALONE - required by the Vosk engine).

# - channels=1: Sets recording to Mono channel audio. (LEAVE ALONE - Vosk processes single-channel input).

# - rate=native_rate: Sets the sample rate to match the mic exactly. (LEAVE ALONE - prevents driver hardware failure).

# - input=True: Configures this stream as an audio capture device. (LEAVE ALONE - required to wake up the microphone).

# - input_device_index=dev_index: Points to our discovered target device. (LEAVE ALONE - ensures we grab the right mic).

# - frames_per_buffer=4096: Controls audio chunk size. (PLAY WITH: Lower values like 2048 reduce latency but risk buffer overflows; higher values like 8192 increase safety but introduce slight latency).

stream = p.open(format=pyaudio.paInt16, channels=1, rate=native_rate,

input=True, input_device_index=dev_index, frames_per_buffer=4096)

# Generate an absolute directory path targeting the exact folder containing your extracted acoustic model weights.

# PARAMETER MANUAL:

# - "STT", "model": Represents folder names. (PLAY WITH: Change these only if your physical folder layout uses different names).

model_path = os.path.join(os.path.expanduser("~"), "STT", "model")

# Load the heavy machine learning weights into system memory to initialize the Vosk voice processing brain.

model = Model(model_path)

# Initialize the primary speech recognizer object using the loaded model weights and target processing frequency.

# PARAMETER MANUAL:

# - model: The initialized AI brain variable. (LEAVE ALONE).

# - 16000: Specifies the target processing sample rate in Hertz. (LEAVE ALONE - Vosk models are mathematically trained for exactly 16000Hz).

recognizer = KaldiRecognizer(model, 16000)

# Print a status message to the terminal showing that the handshake succeeded and revealing the mic's operating frequency.

print(f"Listening on PipeWire at {native_rate}Hz...")

# Turn on the hardware recording valve to begin filling up the system input data buffers.

stream.start_stream()

# Set up an exception safety block to intercept manual user exits gracefully without leaving system resources locked up.

try:

# Launch the infinite real-time audio capture loop to continuously monitor the microphone stream.

while True:

# Pull raw binary audio frame bytes directly out of the incoming hardware stream buffer memory block.

# PARAMETER MANUAL:

# - 4096: The chunk size to read. (LEAVE ALONE - must perfectly match your stream's frames_per_buffer setting).

# - exception_on_overflow=False: Prevents Python crashes if the Orin momentarily falls behind. (LEAVE ALONE).

data = stream.read(4096, exception_on_overflow=False)

# Downsample the raw audio bytes mathematically to 16000Hz using digital filtering algorithms.

# PARAMETER MANUAL:

# - data: Raw input audio byte string. (LEAVE ALONE).

# - 2: Size of sample width in bytes for 16-bit sound. (LEAVE ALONE).

# - 1: Input audio channel count. (LEAVE ALONE).

# - native_rate: The native frequency of your mic. (LEAVE ALONE).

# - 16000: The required downsampled output frequency. (LEAVE ALONE).

# - state: The digital filtering conversion state history object. (LEAVE ALONE).

# CRITICAL NOTE: The 'state' variable must be initialized as 'state = None' before entering this 'try:' block,

# otherwise Python will throw a NameError on the very first loop iteration because 'state' does not exist yet.

data_16k, state = audioop.ratecv(data, 2, 1, native_rate, 16000, state)

# Feed the processed 16kHz audio data block into Vosk to evaluate if a complete spoken phrase has finished.

if recognizer.AcceptWaveform(data_16k):

# Parse the structured JSON text string returned by the acoustic model into a standard Python dictionary.

res = json.loads(recognizer.Result())

# Check the parsed result dictionary to ensure that the detected text sequence is not completely blank.

if res.get("text"):

# Cleanly output the decoded text command straight to the standard system terminal display screen.

print("Command: "+res['text'])

# Intercept the user pressing Ctrl+C in the terminal to stop the continuous script loop cleanly.

except KeyboardInterrupt:

# Print a clean closing status notification on a new line to confirm the signal was intercepted.

print("\nStopped.")

# Execute mandatory clean up instructions that are guaranteed to run no matter how the script terminates or crashes.

finally:

# Turn off the active recording stream data valve to stop pulling raw information from the audio driver layer.

stream.stop_stream()

# Close the recording stream channel completely to drop the active hardware pointer binding handles.

stream.close()

# Terminate the parent PyAudio engine instance to release all allocated audio device resources back to the OS.

p.terminate()

Homework: Your Gate Controller

You now have a system that identifies audio input, resamples it to 16kHz, and outputs text. Your assignment: Transform this text output into an action.

I want you to add a conditional statement to the main loop. If the recognized text is “open”, print an ASCII art representation of an open gate. If it’s “close”, print the closed version. This is the first step in closing the loop between your AI and the physical world. Go get ’em, and don’t just copy the code—understand how the data flows from the microphone to your decision logic!

AI On the Edge, Raspberry Pi

AI on the Edge LESSON 35: Running Multiple Pi Cameras and USB Cameras on the Pi 5

July 22, 2026 admin

AI on the Edge LESSON 35: Running Multiple Pi Cameras and USB Cameras on the Pi 5

Hello everybody! Paul McWhorter here from toptechboy.com, welcoming you back to another thrilling session of our AI on the Edge series. Today, we are taking our hardware vision capabilities to the next level. If you have ever wondered how to scale up your Raspberry Pi 5 setup beyond just a single lens, this lesson is for you. We are going to write a Python script that pulls live feeds from multiple Raspberry Pi cameras and multiple USB webcams simultaneously, displaying all four streams cleanly in real time with an on-screen frames-per-second (FPS) tracker.

Codex Knowledge: Multi-Camera Architecture on the Pi 5

When working with edge hardware like the Raspberry Pi 5, managing multiple high-bandwidth video streams requires understanding how the underlying libraries interact with the Linux kernel and system memory. In this lesson, we leverage two distinct hardware interfaces:

Picamera2 API: We initialize two separate native Pi camera instances using Picamera2(0) and Picamera2(1). By explicitly configuring the preview size to 640×360, setting the format to RGB888, locking the frame rate to 60 FPS, and calling align() before starting, we ensure the hardware pipelines are optimized for low-latency streaming without frame-drop bottlenecks.
OpenCV VideoCapture: For our USB webcams, we use OpenCV’s cv2.VideoCapture() mapped to specific device indices (in our setup, indices 16 and 18). We explicitly set the frame width, height, and target frame rate properties to keep the data flow synchronized with our Pi camera streams.
Window Layout Management: Using OpenCV highgui functions like cv2.namedWindow, cv2.moveWindow, and cv2.resizeWindow, we programmatically arrange all four camera feeds into a neat 2×2 grid on your desktop workspace, preventing windows from stacking blindly on top of each other.

General Knowledge: The Evolution of Multi-Stream Machine Vision

In industrial automation, robotics, and advanced edge AI deployments, relying on a single camera angle is rarely enough. Multi-camera systems are the gold standard for comprehensive spatial awareness, 3D depth estimation, object tracking across wide fields of view, and panoramic monitoring. Historically, running multiple high-resolution video streams required bulky, power-hungry desktop workstations equipped with expensive capture cards. Today, single-board computers like the Raspberry Pi 5—combined with optimized kernel drivers and efficient software wrappers like Picamera2—allow engineers and creators to build robust, multi-sensor vision arrays right at the edge at a fraction of the cost and power consumption.

Python Source Code

Here is the complete, production-ready script for Lesson 35. Make sure your cameras are securely connected and properly indexed before running the program.

import cv2
import time
from picamera2 import Picamera2

W = 640
H = 360
tStart = time.time()
fps = 0
RES = (W, H)

# Initialize Pi Camera 1
piCam1 = Picamera2(0)
piCam1.preview_configuration.main.size = RES
piCam1.preview_configuration.main.format = "RGB888"
piCam1.preview_configuration.controls.FrameRate = 60
piCam1.preview_configuration.align()
piCam1.configure("preview")
piCam1.start()

# Initialize Pi Camera 2
piCam2 = Picamera2(1)
piCam2.preview_configuration.main.size = RES
piCam2.preview_configuration.main.format = "RGB888"
piCam2.preview_configuration.controls.FrameRate = 60
piCam2.preview_configuration.align()
piCam2.configure("preview")
piCam2.start()

# Initialize USB Camera 1
usbCam1 = cv2.VideoCapture(16)
usbCam1.set(cv2.CAP_PROP_FRAME_WIDTH, W)
usbCam1.set(cv2.CAP_PROP_FRAME_HEIGHT, H)
usbCam1.set(cv2.CAP_PROP_FPS, 30)

# Initialize USB Camera 2
usbCam2 = cv2.VideoCapture(18)
usbCam2.set(cv2.CAP_PROP_FRAME_WIDTH, W)
usbCam2.set(cv2.CAP_PROP_FRAME_HEIGHT, H)
usbCam2.set(cv2.CAP_PROP_FPS, 30)

textLowerLeft = (int(W * 0.01), int(H * 0.05))
fontFace = cv2.FONT_HERSHEY_SIMPLEX
fontThickness = int(W / 425)
fontScale = H * 0.0015
fontColor = (0, 0, 255)

# Setup Window Grid Layout
cv2.namedWindow('Camera1', cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera1', 0, 65)
cv2.resizeWindow('Camera1', W, H)

cv2.namedWindow('Camera2', cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera2', W, 65)
cv2.resizeWindow('Camera2', W, H)

cv2.namedWindow('Camera3', cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera3', 0, 65 + H + 20)
cv2.resizeWindow('Camera3', W, H)

cv2.namedWindow('Camera4', cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera4', W, 65 + H + 20)
cv2.resizeWindow('Camera4', W, H)

while True:
    deltaT = time.time() - tStart
    tStart = time.time()
    fps = fps * 0.95 + (1 / deltaT) * 0.05
    
    frame1 = piCam1.capture_array()
    # frame1 = cv2.flip(frame1, -1)
    
    frame2 = piCam2.capture_array()
    frame2 = cv2.flip(frame2, -1)
    
    _ , frame3 = usbCam1.read()
    _ , frame4 = usbCam2.read()
    
    myText = "FPS: " + str(round(fps, 1))
    cv2.putText(frame1, myText, textLowerLeft, fontFace, fontScale, fontColor, fontThickness)
    
    cv2.imshow("Camera1", frame1)
    cv2.imshow("Camera2", frame2)
    cv2.imshow("Camera3", frame3)
    cv2.imshow("Camera4", frame4)
    
    if cv2.waitKey(1) == ord('q'):
        break

cv2.destroyAllWindows()
print('Program Terminated')

import cv2

import time

from picamera2 import Picamera2

W = 640

H = 360

tStart = time.time()

fps = 0

RES = (W, H)

# Initialize Pi Camera 1

piCam1 = Picamera2(0)

piCam1.preview_configuration.main.size = RES

piCam1.preview_configuration.main.format = "RGB888"

piCam1.preview_configuration.controls.FrameRate = 60

piCam1.preview_configuration.align()

piCam1.configure("preview")

piCam1.start()

# Initialize Pi Camera 2

piCam2 = Picamera2(1)

piCam2.preview_configuration.main.size = RES

piCam2.preview_configuration.main.format = "RGB888"

piCam2.preview_configuration.controls.FrameRate = 60

piCam2.preview_configuration.align()

piCam2.configure("preview")

piCam2.start()

# Initialize USB Camera 1

usbCam1 = cv2.VideoCapture(16)

usbCam1.set(cv2.CAP_PROP_FRAME_WIDTH, W)

usbCam1.set(cv2.CAP_PROP_FRAME_HEIGHT, H)

usbCam1.set(cv2.CAP_PROP_FPS, 30)

# Initialize USB Camera 2

usbCam2 = cv2.VideoCapture(18)

usbCam2.set(cv2.CAP_PROP_FRAME_WIDTH, W)

usbCam2.set(cv2.CAP_PROP_FRAME_HEIGHT, H)

usbCam2.set(cv2.CAP_PROP_FPS, 30)

textLowerLeft = (int(W * 0.01), int(H * 0.05))

fontFace = cv2.FONT_HERSHEY_SIMPLEX

fontThickness = int(W / 425)

fontScale = H * 0.0015

fontColor = (0, 0, 255)

# Setup Window Grid Layout

cv2.namedWindow('Camera1', cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera1', 0, 65)

cv2.resizeWindow('Camera1', W, H)

cv2.namedWindow('Camera2', cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera2', W, 65)

cv2.resizeWindow('Camera2', W, H)

cv2.namedWindow('Camera3', cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera3', 0, 65 + H + 20)

cv2.resizeWindow('Camera3', W, H)

cv2.namedWindow('Camera4', cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera4', W, 65 + H + 20)

cv2.resizeWindow('Camera4', W, H)

while True:

deltaT = time.time() - tStart

tStart = time.time()

fps = fps * 0.95 + (1 / deltaT) * 0.05

frame1 = piCam1.capture_array()

# frame1 = cv2.flip(frame1, -1)

frame2 = piCam2.capture_array()

frame2 = cv2.flip(frame2, -1)

_ , frame3 = usbCam1.read()

_ , frame4 = usbCam2.read()

myText = "FPS: " + str(round(fps, 1))

cv2.putText(frame1, myText, textLowerLeft, fontFace, fontScale, fontColor, fontThickness)

cv2.imshow("Camera1", frame1)

cv2.imshow("Camera2", frame2)

cv2.imshow("Camera3", frame3)

cv2.imshow("Camera4", frame4)

if cv2.waitKey(1) == ord('q'):

break

cv2.destroyAllWindows()

print('Program Terminated')

Conclusion

There you have it! You are now successfully driving a multi-camera computer vision array right off your Raspberry Pi 5. Play around with the window positioning, check your device indices if your USB cameras don’t immediately pop up, and get ready because in our next lesson we will start piping these multi-source frames directly into our neural network inference models. Keep tinkering, stay curious, and I will see you in the next lesson!

AI On the Edge, Raspberry Pi

AI on the Edge LESSON 34 SUPPLEMENT: Simple Improvement to FPS

July 21, 2026 admin

In this video I show you how to dramatically improve the FPS of our work in lesson 34. I also show the solution to the issue of the Fusion Hat microphone not working with the project.

import cv2
import time
from picamera2 import Picamera2
#########################################
import threading
from queue import Queue
from fusion_hat.tts import Piper
tts = Piper()
tts.set_model('en_US-kristin-medium')

from fusion_hat.stt import STT
stt = STT('en-us')

cnt = 0
running = True
commandQ = Queue()
speakQ = Queue()
###########################################

from fusion_hat.servo import Servo
panPin = 2
tiltPin = 3
panServo = Servo(panPin)
tiltServo = Servo(tiltPin)
panAngle = -10
tiltAngle =-20
panServo.angle(panAngle)
tiltServo.angle(tiltAngle)
piCam = Picamera2(1)
W=1280
H=720
tStart = time.time()
fps = 0
RES = (W,H)
piCam.preview_configuration.main.size = RES
piCam.preview_configuration.main.format = "RGB888"
piCam.preview_configuration.controls.FrameRate=60
piCam.preview_configuration.align()
piCam.configure("preview")
piCam.start()

textLowerLeft = (int(W*.01),int(H*.05))
fontFace = cv2.FONT_HERSHEY_SIMPLEX
fontThickness = int(W/425)
fontScale = H*.0015
fontColor = (0,0,255)

faceFinder = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
eyeFinder = cv2.CascadeClassifier(cv2.data.haarcascades + "haarcascade_eye.xml")

track = False
xFrameC = int((W-1)/2)
yFrameC = int((H-1)/2)
xBoxC = xFrameC
yBoxC = yFrameC
##########################################
def speak():
    global running
    print('Speak Thread Running')
    while running:
        if speakQ.empty() == False:
            txt = speakQ.get()
            print(txt)
            tts.say(txt)
        time.sleep(.1)
    print('Speak Thread is Terminated')
    print('')
def getCommand():
    print('Get Command Thread is Running')
    global running
    while running:
        myCommand = stt.listen(stream = False)
        myCommand = myCommand.strip()
        print(myCommand)
        if myCommand == 'quit':
            msg = 'Project Humanoid Elimination Terminated. Have a Nice Day.'
            speakQ.put(msg)
            running = False
            break
        commandQ.put(myCommand)
    print('Get Command Thread Terminated')
    print('')

speakThread = threading.Thread(target=speak,daemon=True)
speakThread.start()

commandThread = threading.Thread(target=getCommand,daemon=True)
commandThread.start()
print('Main Program Has Commenced')
msg = 'Hello, I am Ready to Eliminate Humanoids'
speakQ.put(msg)
msgOld = 'Hello, I am Ready to Eliminate Humanoids'

cv2.namedWindow('Camera',cv2.WINDOW_GUI_NORMAL)
cv2.moveWindow('Camera',0,65)
cv2.resizeWindow('Camera',W,H)
############################################
while running:
    deltaT = time.time() - tStart
    tStart=time.time()
    fps = fps*.95 + (1/deltaT)*.05
    frame= piCam.capture_array()
    frame=cv2.flip(frame,-1)
    
    if commandQ.empty() == False:
        myCommand = commandQ.get()
        if myCommand == 'track':
            track = True
        if myCommand == 'release':
            track = False

    gray = cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY)
    faces = faceFinder.detectMultiScale(gray,scaleFactor=1.1, minNeighbors =5, minSize=(100,100))
    if len(faces) != 0:
        cnt = 0 
        msg = "Humanoid Detected, Shall I track?"
    if len(faces) == 0:
        cnt = cnt + 1
        if cnt == 25:
            msg = "No Humanoid Detected, Continuing to Scan"
    if msgOld != msg:
        speakQ.put(msg)
    msgOld = msg
    for (x,y,w,h) in faces:
        cv2.rectangle(frame, (x,y),(x+w,y+h),(0,255,255),3)
        
        xBoxC = x + int(w/2)
        yBoxC = y + int(h/2)
        xError = xBoxC - xFrameC
        yError = yBoxC - yFrameC
        
        eyes =eyeFinder.detectMultiScale(gray[y:y+h,x:x+w],scaleFactor=1.1,minNeighbors=5,minSize=(15,15))
        for (i,j,w,h) in eyes:
            cv2.rectangle(frame,(x+i,y+j),(x+i+w,y+j+h),(255,0,0),3)
    
    if track == True and len(faces)>0:
        panAngle = panAngle - xError/50/2
        panServo.angle(int(panAngle))
        #time.sleep(.02)
        tiltAngle = tiltAngle + yError/50/2
        tiltServo.angle(int(tiltAngle))
        #time.sleep(.02)
                        
    
    myText = "FPS: "+str(round(fps,1))
    cv2.putText(frame,myText,textLowerLeft,fontFace,fontScale,fontColor,fontThickness)
    cv2.imshow("Camera", frame)
    cv2.moveWindow("Camera",0,60)
    if cv2.waitKey(1)==ord('q'):
        break
cv2.destroyAllWindows()
print('')
print('Program Terminated')

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

import cv2

import time

from picamera2 import Picamera2

#########################################

import threading

from queue import Queue

from fusion_hat.tts import Piper

tts = Piper()

tts.set_model('en_US-kristin-medium')

from fusion_hat.stt import STT

stt = STT('en-us')

cnt = 0

running = True

commandQ = Queue()

speakQ = Queue()

###########################################

from fusion_hat.servo import Servo

panPin = 2

tiltPin = 3

panServo = Servo(panPin)

tiltServo = Servo(tiltPin)

panAngle = -10

tiltAngle =-20

panServo.angle(panAngle)

tiltServo.angle(tiltAngle)

piCam = Picamera2(1)

W=1280

H=720

tStart = time.time()

fps = 0

RES = (W,H)

piCam.preview_configuration.main.size = RES

piCam.preview_configuration.main.format = "RGB888"

piCam.preview_configuration.controls.FrameRate=60

piCam.preview_configuration.align()

piCam.configure("preview")

piCam.start()

textLowerLeft = (int(W*.01),int(H*.05))

fontFace = cv2.FONT_HERSHEY_SIMPLEX

fontThickness = int(W/425)

fontScale = H*.0015

fontColor = (0,0,255)

faceFinder = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')

eyeFinder = cv2.CascadeClassifier(cv2.data.haarcascades + "haarcascade_eye.xml")

track = False

xFrameC = int((W-1)/2)

yFrameC = int((H-1)/2)

xBoxC = xFrameC

yBoxC = yFrameC

##########################################

def speak():

global running

print('Speak Thread Running')

while running:

if speakQ.empty() == False:

txt = speakQ.get()

print(txt)

tts.say(txt)

time.sleep(.1)

print('Speak Thread is Terminated')

print('')

def getCommand():

print('Get Command Thread is Running')

global running

while running:

myCommand = stt.listen(stream = False)

myCommand = myCommand.strip()

print(myCommand)

if myCommand == 'quit':

msg = 'Project Humanoid Elimination Terminated. Have a Nice Day.'

speakQ.put(msg)

running = False

break

commandQ.put(myCommand)

print('Get Command Thread Terminated')

print('')

speakThread = threading.Thread(target=speak,daemon=True)

speakThread.start()

commandThread = threading.Thread(target=getCommand,daemon=True)

commandThread.start()

print('Main Program Has Commenced')

msg = 'Hello, I am Ready to Eliminate Humanoids'

speakQ.put(msg)

msgOld = 'Hello, I am Ready to Eliminate Humanoids'

cv2.namedWindow('Camera',cv2.WINDOW_GUI_NORMAL)

cv2.moveWindow('Camera',0,65)

cv2.resizeWindow('Camera',W,H)

############################################

while running:

deltaT = time.time() - tStart

tStart=time.time()

fps = fps*.95 + (1/deltaT)*.05

frame= piCam.capture_array()

frame=cv2.flip(frame,-1)

if commandQ.empty() == False:

myCommand = commandQ.get()

if myCommand == 'track':

track = True

if myCommand == 'release':

track = False

gray = cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY)

faces = faceFinder.detectMultiScale(gray,scaleFactor=1.1, minNeighbors =5, minSize=(100,100))

if len(faces) != 0:

cnt = 0

msg = "Humanoid Detected, Shall I track?"

if len(faces) == 0:

cnt = cnt + 1

if cnt == 25:

msg = "No Humanoid Detected, Continuing to Scan"

if msgOld != msg:

speakQ.put(msg)

msgOld = msg

for (x,y,w,h) in faces:

cv2.rectangle(frame, (x,y),(x+w,y+h),(0,255,255),3)

xBoxC = x + int(w/2)

yBoxC = y + int(h/2)

xError = xBoxC - xFrameC

yError = yBoxC - yFrameC

eyes =eyeFinder.detectMultiScale(gray[y:y+h,x:x+w],scaleFactor=1.1,minNeighbors=5,minSize=(15,15))

for (i,j,w,h) in eyes:

cv2.rectangle(frame,(x+i,y+j),(x+i+w,y+j+h),(255,0,0),3)

if track == True and len(faces)>0:

panAngle = panAngle - xError/50/2

panServo.angle(int(panAngle))

#time.sleep(.02)

tiltAngle = tiltAngle + yError/50/2

tiltServo.angle(int(tiltAngle))

#time.sleep(.02)

myText = "FPS: "+str(round(fps,1))

cv2.putText(frame,myText,textLowerLeft,fontFace,fontScale,fontColor,fontThickness)

cv2.imshow("Camera", frame)

cv2.moveWindow("Camera",0,60)

if cv2.waitKey(1)==ord('q'):

break

cv2.destroyAllWindows()

print('')

print('Program Terminated')

NVIDIA, NVIDIA Jetson Orin Nano

AI on the NVIDIA Jetson Orin Nano: Adding Text-to-Speech (TTS) with Piper

July 20, 2026 admin

In this lesson, we are building a local, offline voice pipeline using Piper. This engine runs natively on our Jetson Orin Nano hardware, providing fast and natural speech without needing an internet connection. To keep this simple, we will install everything into one specific folder so the software can easily find its own files.

Step 1: System Prep & Piper Installation

Open your terminal and run these commands one by one to create the workspace and download the required files. We are placing everything into the same directory to ensure the AI engine can always find the voice model.

# 1. Update and install dependencies
sudo apt update
sudo apt install -y wget tar libasound2-dev

# 2. Create the main project folder
mkdir -p ~/voiceAssistant/piper
cd ~/voiceAssistant/piper

# 3. Download and extract the Piper binary
wget https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_aarch64.tar.gz
tar -xvzf piper_linux_aarch64.tar.gz

# 4. Enter the folder where Piper was extracted
cd piper

# 5. Download the voice model files into this same folder
wget -O en_US-lessac-medium.onnx 'https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx?download=true'
wget -O en_US-lessac-medium.onnx.json 'https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx.json?download=true'

# 6. Grant execution permissions to the Piper engine
chmod +x piper

# 1. Update and install dependencies

sudo apt update

sudo apt install -y wget tar libasound2-dev

# 2. Create the main project folder

mkdir -p ~/voiceAssistant/piper

cd ~/voiceAssistant/piper

# 3. Download and extract the Piper binary

wget https://github.com/rhasspy/piper/releases/download/2023.11.14-2/piper_linux_aarch64.tar.gz

tar -xvzf piper_linux_aarch64.tar.gz

# 4. Enter the folder where Piper was extracted

cd piper

# 5. Download the voice model files into this same folder

wget -O en_US-lessac-medium.onnx 'https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx?download=true'

wget -O en_US-lessac-medium.onnx.json 'https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx.json?download=true'

# 6. Grant execution permissions to the Piper engine

chmod +x piper

Step 2: Identify Your Audio Device

Because every setup is different, we need to tell the system which speaker to use. Run the following command in your terminal:

aplay -l

aplay -l

Look through the list for your speaker. You will see something like card 0 and device 0. If your card is 0 and device is 0, your identifier is plughw:0,0. You will use these numbers in the Python script below.

Step 3: The Python Pipeline Script

Create a new Python file and paste the code below. Because we installed everything into the same folder, this script will find the files immediately.

import subprocess

# --- CONFIGURATION ---
# piperPath: The location of the downloaded Piper executable file.
# modelPath: The path to the downloaded AI voice model file (.onnx).
# audioDevice: The specific hardware address of the speaker (e.g., plughw:0,0).
piperPath = "/home/pjm/voiceAssistant/piper/piper/piper"
modelPath = "/home/pjm/voiceAssistant/piper/piper/en_US-lessac-medium.onnx"
audioDevice = "plughw:0,0"

# --- BUILD THE COMMANDS ---
# piperCommand: This runs the AI engine. 
# --output-raw tells Piper to send out raw audio data instead of a .wav file.
piperCommand = piperPath + " --model " + modelPath + " --output-raw"

# aplayCommand: This is the Linux sound player that pipes audio to your speakers.
# -D: Specifies the device (hardware).
# -r 22050: Sets the sample rate. Piper requires 22050Hz for this model.
# -f S16_LE: Sets the format to 16-bit Signed Little Endian (the raw digital format).
# -t raw: Tells aplay that the incoming data has no header and is pure audio.
# -: The hyphen tells aplay to read input from 'stdin' (the data flowing from the pipe).
aplayCommand = "aplay -D " + audioDevice + " -r 22050 -f S16_LE -t raw -"

# fullCommand: We connect the two commands using a pipe '|'. 
# This takes the output of the AI (piperCommand) and feeds it directly 
# into the speaker input (aplayCommand).
fullCommand = piperCommand + " | " + aplayCommand

# --- RUN THE PIPELINE ---
# subprocess.Popen allows us to run the command in the background while
# letting our Python code continue to run.
speechProcess = subprocess.Popen(fullCommand, shell=True, stdin=subprocess.PIPE)

# speechMessage: The text we want the AI to read.
# We must encode it as bytes (b"...") to send it to the system.
speechMessage = "Hello students. The voice pipeline is verified and operational."
speakable=speechMessage.encode('utf-8')

# communicate: This sends the text message into the running Piper process.
speechProcess.communicate(input=speakable)

import subprocess

# --- CONFIGURATION ---

# piperPath: The location of the downloaded Piper executable file.

# modelPath: The path to the downloaded AI voice model file (.onnx).

# audioDevice: The specific hardware address of the speaker (e.g., plughw:0,0).

piperPath = "/home/pjm/voiceAssistant/piper/piper/piper"

modelPath = "/home/pjm/voiceAssistant/piper/piper/en_US-lessac-medium.onnx"

audioDevice = "plughw:0,0"

# --- BUILD THE COMMANDS ---

# piperCommand: This runs the AI engine.

# --output-raw tells Piper to send out raw audio data instead of a .wav file.

piperCommand = piperPath + " --model " + modelPath + " --output-raw"

# aplayCommand: This is the Linux sound player that pipes audio to your speakers.

# -D: Specifies the device (hardware).

# -r 22050: Sets the sample rate. Piper requires 22050Hz for this model.

# -f S16_LE: Sets the format to 16-bit Signed Little Endian (the raw digital format).

# -t raw: Tells aplay that the incoming data has no header and is pure audio.

# -: The hyphen tells aplay to read input from 'stdin' (the data flowing from the pipe).

aplayCommand = "aplay -D " + audioDevice + " -r 22050 -f S16_LE -t raw -"

# fullCommand: We connect the two commands using a pipe '|'.

# This takes the output of the AI (piperCommand) and feeds it directly

# into the speaker input (aplayCommand).

fullCommand = piperCommand + " | " + aplayCommand

# --- RUN THE PIPELINE ---

# subprocess.Popen allows us to run the command in the background while

# letting our Python code continue to run.

speechProcess = subprocess.Popen(fullCommand, shell=True, stdin=subprocess.PIPE)

# speechMessage: The text we want the AI to read.

# We must encode it as bytes (b"...") to send it to the system.

speechMessage = "Hello students. The voice pipeline is verified and operational."

speakable=speechMessage.encode('utf-8')

# communicate: This sends the text message into the running Piper process.

speechProcess.communicate(input=speakable)

Step 4: Customizing Your Voice

Piper has dozens of voices available. To see the full library, visit the Piper Voices Repository. Download the .onnx and .onnx.json files for your preferred voice, place them in the ~/voiceAssistant/piper/piper/ folder, and update the modelPath variable in your script.

Homework: The Talking Echo Bot

Your assignment is to play around with different voice models, and choose several that you like the best. Then modify the Python script so it becomes an interactive “Echo Bot.” Instead of hardcoding the message, use Python’s input() function to ask the user what to say. When the user types a sentence and presses Enter, your script should pipe that text into Piper and speak it back to you. Use a while True: loop to keep the program running so you can continue talking to your computer. Make a video of your working project, and the voices you chose. In the description in your video, make sure to leave a link back to the video above. That way users can easily click between my video lesson over to your video, and then back to the class video.

Technology Tutorials

AI on the Edge LESSON 36: Select Active Camera in OpenCV With Voice Commands

What You Will Learn in This Lesson

Step-by-Step Breakdown of the Script

1. Setting Up the Background Voice Thread

2. Initializing Multiple Cameras

3. Processing and Routing Commands in the Main Loop

4. Managing Your OpenCV Windows

Local Voice Control of NVIDIA Jetson Orin Nano with STT: Getting Started with Vosk

Engineering Your Own Local Voice Assistant: No Cloud, No Compromise

The “Why” Behind the Setup

Part 1: Preparing the Environment

This is IMPORTANT!

Part 2: Solving the PipeWire Challenge

Homework: Your Gate Controller

AI on the Edge LESSON 35: Running Multiple Pi Cameras and USB Cameras on the Pi 5

AI on the Edge LESSON 35: Running Multiple Pi Cameras and USB Cameras on the Pi 5

Codex Knowledge: Multi-Camera Architecture on the Pi 5

General Knowledge: The Evolution of Multi-Stream Machine Vision

Python Source Code

Conclusion

AI on the Edge LESSON 34 SUPPLEMENT: Simple Improvement to FPS

AI on the NVIDIA Jetson Orin Nano: Adding Text-to-Speech (TTS) with Piper

Step 1: System Prep & Piper Installation

Step 2: Identify Your Audio Device

Step 3: The Python Pipeline Script

Step 4: Customizing Your Voice

Homework: The Talking Echo Bot

Making The World a Better Place One High Tech Project at a Time. Enjoy!