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.

AI On the Edge, Raspberry Pi

AI on the Edge LESSON 34: Project Combining TTS, STT, Face Recognition and Servos on Pi 5

July 18, 2026 admin

In our previous lessons, we built individual components. We got our machine to listen locally using Speech-to-Text (STT), we got it to talk back cleanly with Text-to-Speech (TTS), and we learned how to manipulate physical hardware using precision servos. Today, we bridge the gap between software and the physical world. We are building a multi-threaded, autonomous edge system that tracks humanoids in real time.

The Engineering Mindset: Concurrency & Threading

If you try to build a system like this sequentially—running camera capture, object detection, audio listening, and audio speaking in a single while True: loop—your project will fail completely. Why? Because tasks like waiting for a voice command (stt.listen()) or synthesizing a voice line are blocking operations. If your code is stuck waiting for you to finish speaking a command, your camera frames freeze, your servo adjustments drop to zero, and your humanoid target escapes.

To solve this, we architect our software using Multi-threading. We spawn independent execution paths that run concurrently, passing data back and forth safely using thread-safe Queues:

The Main Thread: Handles the high-speed Picamera2 video capture loop, executes Haar Cascade face/eye calculations, and updates the servo angles via mathematical error tracking.
The Speak Thread: Idles quietly in the background until a message lands in the speakQ, instantly triggering the local Piper TTS engine without stalling our camera frame rate.
The Command Thread: Keeps a continuous ear open via our local STT engine. When it parses a voice command like “track”, “release”, or “quit”, it safely pops that token into the commandQ for the main loop to execute on its next pass.

The Complete Fusion Architecture Code

Below is the complete Python pipeline developed in today’s lesson. Make sure your hardware connections for the pan-and-tilt servos match pins 2 and 3 on your expansion setup, and verify your fusion_hat software stack is completely updated.

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()
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)
    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)&gt;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

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()

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)

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')

Breaking Down the Math & Logic

Pro-Tip on State Debouncing: Notice how we handle messages using the cnt variable and the msgOld buffer. If we simply asked the system to say “Humanoid Detected” every single time the frame loops, the TTS engine would crash from queue overflow, stuttering continuously. By checking if msgOld != msg, we ensure the system speaks a notification exactly once upon transition. The 25-frame hysteresis buffer ensures that a single missed frame doesn’t cause a false “No Humanoid” trigger.

Proportional Error Tracking

Look carefully at how the servo angles are adjusted:

panAngle = panAngle - xError/50/2

1	panAngle = panAngle - xError/50/2

We are calculating the error vector—how many pixels the center of the bounding box (xBoxC) is from the absolute center of our camera stream (xFrameC). Instead of moving the servo by a fixed step, we move it proportionally to the size of the error. Big error? Fast movement. Small error? Gentle, microscopic correction. Dividing by 50 dampens our proportional gain loop so our servos don’t violently oscillate and throw our camera system completely out of alignment.

Your Homework Assignment

You have the system reading frames, listening asynchronously, and adjusting angles proportionally. Now it’s time to earn your stripes as an edge systems designer.

Your Assignment: Look at our msg tracking logic. Right now, if a humanoid is detected, it simply asks, “Shall I track?”. Your task is to tie our speech input directly to this state. Modify the control flow so that the tracking logic requires verification. If a humanoid is detected, the system must wait until the commandQ yields a verified verbal confirmation—like “execute” or “yes”—before the proportional servo tracking begins. If it hears “abort”, it must break off and look away.

Drop your custom implementation script in the comments section below, explain the logic behind your state modifications, and let’s see how you optimize your loop efficiency. Go get ’em!

Technology Tutorials

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

AI on the Edge LESSON 34: Project Combining TTS, STT, Face Recognition and Servos on Pi 5

The Engineering Mindset: Concurrency & Threading

The Complete Fusion Architecture Code

Breaking Down the Math & Logic

Proportional Error Tracking

Your Homework Assignment

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