Developed an AI-powered smart glasses prototype to assist the visually impaired. Integrated a Raspberry Pi 5 and the TOKI AI voice assistant into a compact wearable. I specifically engineered the custom hardware power design and actively developed the device's Python-based AI voice assistant, utilizing speech-to-text, large language model query routing, and automated Geoapify location parsing. Secured 3rd place at Youth Tech Begin and won the Entrepreneurship Potential Award given by the Türkiye Technology Team Foundation (T3 Foundation).
Primary Supply: Dual 3.7V 18650 Li-ion cells connected in series to establish a 7.4V nominal bus.
Battery Management & Charging: Integrated a 2S BMS to handle cell balancing, overcharge/over-discharge protection, and thermal safety, paired with a dedicated 2S charging module.
Step-Down Regulation: Integrated an XL4016 DC-DC buck converter to step down the 7.4V bus to a stable 5V output capable of powering the high-current demands of the Raspberry Pi 5.
Delivery Infrastructure: Routed power through heavy-gauge 18 AWG wiring to minimize voltage drop, terminating into a USB-C breakout board and cable for stable delivery directly to the Pi.
Microphone Input: Integrated an INMP441 MEMS omnidirectional microphone using digital I2S audio protocol for direct, noise-resilient voice capture.
Digital-to-Analog Conversion: Since the Raspberry Pi 5 lacks an onboard analog DAC, integrated a PCM5102 I2S DAC module to decode high-fidelity audio streams for the voice assistant.
Shared I2S Bus Architecture: Mapped both input (INMP441) and output (PCM5102) onto shared clock lines from the Raspberry Pi 5 header.
Amplification & Output: Routed the analog output from the PCM5102 into a PAM8403 mini-amplifier board to drive an onboard 8Ω 3W speaker for real-time voice feedback.
Integrated an IMX219-83 Stereo Binocular Camera module via dual MIPI-CSI connectors to feed real-time visual data into the image processing pipeline.
import os
import re
import json
import subprocess
import time
import math
import requests
import speech_recognition as sr
from gtts import gTTS
from groq import Groq
from urllib.parse import quote
from ctypes import *
import threading
import pyaudio
from vosk import Model, KaldiRecognizer
# --- 1. HARDWARE LEVEL STDERR SUPPRESSION ---
class suppress_stderr:
def __enter__(self):
self.null_fd = os.open(os.devnull, os.O_RDWR)
self.save_fd = os.dup(2)
os.dup2(self.null_fd, 2)
def __exit__(self, *_):
os.dup2(self.save_fd, 2)
os.close(self.null_fd)
os.close(self.save_fd)
try:
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
def py_error_handler(filename, line, function, err, fmt):
pass
c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
asound = cdll.LoadLibrary('libasound.so')
asound.snd_lib_error_set_handler(c_error_handler)
except Exception:
passExplains the architectural resolution of low-level runtime stream pollution in embedded Linux audio pipelines.
Linux sound architectures (ALSA) and underlying hardware abstractions on embedded platforms like the Raspberry Pi persistently output unhandled, non-critical C-level driver warnings directly to standard error (stderr). During real-time voice assistant stream initializations, this extraneous text corrupts standard output logs and degrades clean process monitoring.
suppress_stderr Context Manager: I implemented a deterministic, resource-safe Python context manager utilizing low-level OS file descriptors.
Redirecting stderr to /dev/null: I duplicated file descriptor 2 (stderr) using os.dup() and re-routed output streams via os.dup2() into a null device stream during stream handshakes.
Restoring stderr Safely: I designed the system to automatically restore original file descriptor bindings upon exit execution to prevent global stream degradation.
ALSA Custom Error Handler using ctypes: I bypassed OS stream boundaries by directly interfacing with compiled shared object symbols in user space.
Loading libasound.so: I dynamically bound the native ALSA client library at runtime via CDLL.
snd_lib_error_set_handler(): I overwrote the default error callback hook with a silent Python no-op trampoline function, completely muting native hardware spam.
Suppressing raw driver-level C diagnostics isolates high-level Python application telemetry from hardware noise. This guarantees production-grade console hygiene for headless embedded edge devices without compromising exception visibility or masking genuine runtime faults.
libasound.so) can be safely intercepted using ctypes function signatures and custom error trampolines.__enter__ and __exit__) ensure deterministic resource acquisition and teardown for stream-level redirection.# --- 2. CONFIGURATION & API KEYS ---
# SECURITY: Removed compromised hardcoded keys. Use environment variables.
GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
GEOAPIFY_API_KEY = os.environ.get("GEOAPIFY_API_KEY")
GOOGLE_GEOLOCATION_API_KEY = os.environ.get("GOOGLE_GEOLOCATION_API_KEY")
ELEVENLABS_API_KEY = os.environ.get("ELEVENLABS_API_KEY")
# Telegram Bot Token
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
# --- 2.5 MEMORY CACHE ---
# --- 2.5 MEMORY CACHE & CIRCUIT BREAKERS ---
CACHED_LOCATION = None
LAST_LOCATION_TIME = 0
ELEVENLABS_DISABLED = False # The new circuit breaker
# Map spoken names to their specific Telegram Chat IDs
TELEGRAM_CONTACTS = {
"omar": os.environ.get("TELEGRAM_CHAT_ID_OMAR"),
"asadullah": os.environ.get("TELEGRAM_CHAT_ID_ASADULLAH"),
}
if not GROQ_API_KEY:
raise ValueError("GROQ_API_KEY environment variable is missing.")
llm_client = Groq(api_key=GROQ_API_KEY)
http_session = requests.Session() # Optimization: Connection pooling
r = sr.Recognizer()
print("Loading local STT model...")
with suppress_stderr():
vosk_model = Model("model")
category_map = {
"supermarket": "commercial.supermarket", "süpermarket": "commercial.supermarket", "market": "commercial.supermarket", "markete": "commercial.supermarket",
"cafe": "catering.cafe", "kafe": "catering.cafe", "kafeye": "catering.cafe",
"restaurant": "catering.restaurant", "restoran": "catering.restaurant", "restorana": "catering.restaurant",
"pharmacy": "healthcare.pharmacy", "eczane": "healthcare.pharmacy", "eczaneye": "healthcare.pharmacy",
"hospital": "healthcare.hospital", "hastane": "healthcare.hospital", "hastaneye": "healthcare.hospital",
"school": "education.school", "okul": "education.school", "okula": "education.school",
"university": "education.university", "üniversite": "education.university", "üniversiteye": "education.university",
"bank": "service.financial.bank", "banka": "service.financial.bank", "bankaya": "service.financial.bank",
"atm": "service.financial.atm", "atmye": "service.financial.atm",
"gas station": "service.vehicle.fuel", "benzinlik": "service.vehicle.fuel", "benzin istasyonu": "service.vehicle.fuel",
"hotel": "accommodation.hotel", "otel": "accommodation.hotel", "otele": "accommodation.hotel",
"bus stop": "public_transport.bus", "durak": "public_transport.bus", "durağa": "public_transport.bus",
"train station": "public_transport.train", "istasyon": "public_transport.train", "istasyona": "public_transport.train",
"park": "leisure.park", "parka": "leisure.park",
"store": "commercial", "mağaza": "commercial", "dükkan": "commercial", "dükkana": "commercial"
}Establishes secure credential management, optimizes network and memory states, and defines the semantic mapping necessary for routing natural language to geographic APIs.
Hardcoding API keys exposes sensitive credentials in public repositories. Furthermore, voice assistants suffer from latency if they re-initialize connections or repeatedly query GPS hardware for every command. Additionally, relying solely on cloud Text-to-Speech (TTS) services introduces a single point of failure if quotas are exceeded, and unstructured natural language inputs for navigation must be standardized into strict API-friendly parameters.
Environment Variable Security (os.environ.get): I migrated all API tokens and Telegram Chat IDs to environment variables, preventing credential leaks and decoupling configuration from the codebase for safe version control.
Circuit Breaker Pattern (ELEVENLABS_DISABLED): I implemented a state flag to instantly bypass the premium TTS API if a quota error or network timeout occurs, preventing cascading system lockups and ensuring rapid fallback to local TTS.
Memory Caching (CACHED_LOCATION): I introduced temporal caching for GPS coordinates to minimize redundant hardware polling, significantly reducing system latency during continuous conversation loops.
Connection Pooling (requests.Session()): I instantiated a persistent HTTP session to reuse TCP connections across the application, which eliminates the overhead of repeated TLS handshakes for high-frequency cloud API calls.
Semantic Category Mapping (category_map): I designed a bilingual (English/Turkish) dictionary structure to deterministically translate fuzzy spoken intents (e.g., "markete") into strict hierarchical tags (e.g., "commercial.supermarket") required by the Geoapify routing engine.
Designing robust configuration and state management is what transforms a prototype into a production-ready application. By implementing circuit breakers, connection pooling, and credential security, the system guarantees high availability, safe open-source distribution, and low latency—critical metrics for real-time voice interfaces running on edge devices.
HTTP Keep-Alive) is a crucial micro-optimization for latency-sensitive REST API integrations.def get_mic_index():
with suppress_stderr():
p = pyaudio.PyAudio()
target_index = None
for i in range(p.get_device_count()):
dev = p.get_device_info_by_index(i)
if "voicehat" in dev['name'].lower():
target_index = i
if "plug" in dev['name'].lower():
break
p.terminate()
return target_index
print("Probing I2S hardware...")
GLOBAL_MIC_IDX = get_mic_index()
# --- 3. CORE FUNCTIONS ---
def speak(text):
global ELEVENLABS_DISABLED
if not text:
return
print(f"Speaking: {text}")
clean_text = text.replace('"', '').replace("'", "")
# Only try ElevenLabs if we have a key AND the breaker hasn't tripped
if ELEVENLABS_API_KEY and not ELEVENLABS_DISABLED:
url = "https://api.elevenlabs.io/v1/text-to-speech/liF98vtVyeMPMLuaIUXC/stream"
headers = {
"Accept": "audio/mpeg",
"Content-Type": "application/json",
"xi-api-key": ELEVENLABS_API_KEY
}
payload = {
"text": clean_text,
"model_id": "eleven_turbo_v2_5",
"voice_settings": {
"stability": 0.5,
"similarity_boost": 0.75
}
}
try:
response = http_session.post(url, json=payload, headers=headers, stream=True, timeout=10)
if response.status_code == 200:
process = subprocess.Popen(
['mpg123', '-q', '-f', '6552', '-o', 'alsa', '-a', 'plughw:2,0', '-'],
stdin=subprocess.PIPE
)
for chunk in response.iter_content(chunk_size=4096):
if chunk:
process.stdin.write(chunk)
process.stdin.close()
process.wait(timeout=15)
return
elif response.status_code == 401 and "quota_exceeded" in response.text:
print("\n[SYSTEM] ElevenLabs Quota Exceeded. Tripping circuit breaker to ensure fast responses.")
ELEVENLABS_DISABLED = True # Breaker trips! Future calls skip ElevenLabs instantly.
else:
print(f"[WARNING] ElevenLabs API Error: {response.status_code}")
print("Switching to backup Google voice...")
except subprocess.TimeoutExpired:
print("[WARNING] Hardware audio playback timed out (ALSA lockup). Resetting...")
subprocess.run('killall -9 mpg123', shell=True)
return
except Exception as e:
print(f"Network streaming error: {e}")
print("Switching to backup Google voice...")
# --- FALLBACK TO FREE GOOGLE TTS ---
try:
tts_lang = 'tr' if any(c in clean_text for c in ['İ', 'ü', 'ş', 'ğ', 'ç', 'ö']) else 'en'
tts = gTTS(text=clean_text, lang=tts_lang, tld='us')
tts.save("response.mp3")
subprocess.run('mpg123 -q -f 6552 -o alsa -a plughw:2,0 response.mp3 >/dev/null 2>&1', shell=True, timeout=15)
except subprocess.TimeoutExpired:
subprocess.run('killall -9 mpg123', shell=True)
except Exception as e:
print(f"gTTS error: {e}")Handles dynamic hardware probing for the I2S microphone and orchestrates a highly resilient, streaming Text-to-Speech (TTS) pipeline to minimize latency.
In embedded Linux environments, ALSA audio device indices are unstable and can change randomly upon reboot, causing hardcoded applications to fail silently. Furthermore, traditional cloud TTS implementations download an entire audio file before playing it, creating unnatural delays in conversation. Finally, if the hardware audio decoder freezes or a premium API quota is reached, the entire application thread can lock up, requiring a hard manual reset of the edge device.
Dynamic Device Probing (get_mic_index): I utilized PyAudio to iterate through all system audio interfaces at runtime, searching specifically for the physical "voicehat" string. This guarantees the correct hardware is bound regardless of the OS boot order.
Audio Streaming via Popen: I bypassed local file saving by piping ElevenLabs API response chunks directly into the standard input of the mpg123 decoder via subprocess.Popen, significantly reducing the time-to-first-audio.
Hardware Interface Targeting: I explicitly routed playback to plughw:2,0 using ALSA command-line arguments to bypass default OS mixing layers and send the stream directly to the correct DAC.
Process Timeout & Cleanup: I engineered strict timeout parameters (process.wait(timeout=15)) and automated recovery routines (killall -9 mpg123) to aggressively terminate zombie audio threads and free locked hardware.
Automated Language Detection & Fallback: I built a character filter to dynamically detect Turkish-specific vowels (e.g., 'ş', 'ğ'), automatically switching the backup Google TTS engine (gTTS) to the correct local language profile when the primary stream fails.
In autonomous edge computing, hardware systems must be entirely self-healing. By implementing dynamic I2S probing and aggressive process termination, the application survives reboots and driver lockups without user intervention. Additionally, transitioning from file-based I/O to memory-piped streaming ensures the voice assistant achieves the near-instantaneous response times expected in modern AI architectures.