Most AI agents live entirely on a screen. In this project, they reach into the real world: when the agents research, a blue LED pulses and a servo sweeps, and when they debate, a yellow LED takes over. You’ll build two projects that connect AI agents with Arduino hardware: J.A.R.V.I.S., a voice assistant with a physical status display, and an AI Pomodoro timer that fetches a research briefing during your break.
Get the code: the full source for both projects is on GitHub: J.A.R.V.I.S. and AI Pomodoro timer.
Common problem: Arduino ignoring new commands while it blinks or sweeps? See why delay() freezes your Arduino and how to use millis() instead.
The Hardware Basics
What is an Arduino?
An Arduino is a tiny programmable board that links software to the physical world. Unlike a PC, it doesn’t run an operating system. Instead, you write code on your computer, upload it over USB, and the board runs that code in a loop.
It takes inputs, such as a button press, a temperature change or a message from your computer, and turns them into outputs, such as a light, a motor or a message back. Because it’s open source and cheap, it’s the standard choice for prototypes. In this project, it’s the link that lets AI code move real parts.
What is a breadboard?
A breadboard is a plastic block full of small holes. You push wires and parts into the holes to connect them, so there’s no soldering and nothing is permanent.
Under the surface, metal strips join the holes. In the middle, each short column of five holes is connected. The long rows along the edges, marked red and blue, carry power and ground across the board. So when you wire something backwards, you just pull the wire out and move it.
What is an LED?
LED stands for Light Emitting Diode. It’s a small, efficient part that glows when current flows through it, and it stays cool and lasts for years.
Because it’s a diode, current only flows one way. So the longer leg (the anode) goes to positive, and the shorter leg (the cathode) goes to ground. LEDs are the simplest way to see what your code is doing, which is exactly how this project uses them.
Project 1: J.A.R.V.I.S., AI Agents with Arduino Status Lights
📦 Full project code: Jarvis-Hardware-Program on GitHub
J.A.R.V.I.S. is a local AI assistant with a web dashboard. A local model on Ollama writes the replies, OmniVoice Studio speaks them, and an Arduino shows what the AI is doing: blue for researching, yellow for debating, and a sweeping servo while it works.
Step 1: Wire the parts
You’ll need an Arduino Uno, a breadboard, a USB cable, a blue and a yellow LED, two 220–330Ω resistors, an SG90 micro servo and about 15 jumper wires.
Insert the blue LED. Then connect its long leg through a resistor to pin 12, and its short leg to the ground rail.
Do the same for the yellow LED, but use pin 13.
The servo has three wires. Connect the signal wire (orange or yellow) to pin 8, ground (brown or black) to the ground rail and power (red) to 5V.
Check that no stray wire bridges two rows, then plug the Arduino into your PC.
Finally, open Device Manager → Ports (COM & LPT) and note the COM port, for example COM3.
Power note: a servo bigger than an SG90 can pull too much current from the Arduino’s 5V pin and make the board reset. If the LEDs flicker when the servo moves, give the servo its own 5V supply and connect the grounds together.
Step 2: Upload the Arduino sketch
Install the Arduino IDE and open it.
Under Tools → Board, choose Arduino Uno. Then pick your COM port under Tools → Port.
Open
hardware/jarvis_hardware/jarvis_hardware.ino. It listens for single letters over USB:0for idle,Bfor researching,Yfor debating andWfor complete.Click Verify, then Upload, and wait for “Done uploading”.
Close the Serial Monitor before you start the Python app, because only one program can use the port at a time.
Step 3: Set up Python
Install Python 3.11 or newer. Then open a terminal in the project folder, create a virtual environment with python -m venv venv and activate it with venv\Scripts\activate. Next, install the packages:
pip install flask flask-socketio flask-cors python-dotenv requests pyserial ddgs
Then check that python -c "import serial; serial.Serial" runs without errors. If it says module 'serial' has no attribute 'Serial', two packages are fighting over the same name. Remove both and reinstall the right one:
pip uninstall -y serial pyserial
pip install pyserial
Step 4: Create the .env file
Settings go in a .env file in the project folder, so they stay out of the code:
ARDUINO_PORT=COM3, with your own port from Step 1.OLLAMA_URL=http://localhost:11434/api/chat, which is Ollama’s default address.OMNIVOICE_URL=http://localhost:3900/v1, which is OmniVoice Studio’s local API.CACHE_TTL=3600, so repeated replies are cached for an hour.SECRET_KEYis optional. Then addOMNIVOICE_VOICEin Step 6 and the CrewAI values in Step 7.
Add .env to .gitignore straight away, because it will soon hold a secret token.
Step 5: Install Ollama
Ollama runs AI models on your own computer. Install it from ollama.com, and it starts in the background. Then pick a model that fits your graphics card. Check its memory with nvidia-smi --query-gpu=memory.total --format=csv:
Up to 4 GB:
ollama pull qwen2.5-coder:1.5b8 GB:
ollama pull llama3.1:8b
Then test it with ollama run <model-name>. You can also pin a model with OVERRIDE_MODEL in .env.
Step 6: Install OmniVoice Studio
OmniVoice Studio gives Jarvis a voice, and it runs locally too. Download the Windows installer from its GitHub page, run it, and let the first-run wizard download the voice models. This can take several minutes.
If setup fails with “Backend process exited”, read the real error in
%APPDATA%\OmniVoice\omnivoice.log.If the log shows the voice engine running out of graphics memory, switch to KittenTTS under Settings → Engines → TTS. It’s small and runs on the CPU.
Then open
http://localhost:3900/v1/audio/voicesin a browser. If you see JSON, the API works.Pick a voice in Settings → Voice Gallery and put its id in
OMNIVOICE_VOICE. KittenTTS includes four male and four female voices.
You can close OmniVoice afterwards, because the app starts it automatically. A cold start takes 15 to 75 seconds, so the first reply may be slow.
Running OmniVoice from source instead
You can also run it from source. First check that Git and Python are installed:
git --version
python3 --version
pip3 --version
Then clone the project, create a virtual environment and install its requirements. On Windows, activate the environment with venv\Scripts\activate instead:
git clone https://github.com/debpalash/OmniVoice-Studio.git
cd OmniVoice-Studio
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
Check the project’s README for the start command and any .env settings. If you get “No module named X”, install that package with pip install X. If the port is already in use, stop the old process first.
Step 7: Set up the CrewAI agents
CrewAI lets several AI agents work together. In CrewAI Studio, create a blank project called Jarvis hardware program:
Then add two agents:
Lead Intelligence Gatherer. Its goal is to find accurate, up-to-date information about the topic. Give it the Brave Search tool.
Senior Executive Analyst (Jarvis). Its goal is to turn raw findings into a clear briefing. Tell it never to include raw commands or code, only clean, professional text.
Next, create two tasks. The first collects the facts the user’s question needs, as a bulleted list. The second turns that list into a short, conversational answer with no code blocks:
Then connect the agents to their tasks, test the crew and publish it:
When it’s deployed, copy its kickoff URL and bearer token into .env:
CREWAI_CREW_URL=your crew urlCREWAI_CREW_TOKEN=your crew token
To be clear about the current state: the app builds the CrewAI login header, but no route calls the crew yet. Connecting chat messages to a crew run is the next piece of work.
The J.A.R.V.I.S. Software
The HTML dashboard
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>J.A.R.V.I.S. Interface</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="app-container">
<!-- Left Sidebar: Hardware Status & TTS -->
<aside class="hardware-sidebar">
<div class="sidebar-header">
<h2>HARDWARE<br>TELEMETRY</h2>
</div>
<div class="orb-container">
<div id="main-orb" class="status-orb orb-idle"></div>
<p id="status-text" class="status-label">STANDBY</p>
<p id="sub-status-text" class="status-sub-label">System Idle</p>
</div>
<div class="hardware-map">
<p class="map-title">PIN STATES</p>
<div class="map-item">
<span class="pin-indicator" id="pin-12"></span>
<span>PIN 12 (Research)</span>
</div>
<div class="map-item">
<span class="pin-indicator" id="pin-13"></span>
<span>PIN 13 (Debate)</span>
</div>
<div class="map-item">
<span class="pin-indicator" id="relay-1"></span>
<span>RELAY 1 (Complete)</span>
</div>
</div>
<!-- NEW: EXPLICIT TEXT-TO-SPEECH MODULE -->
<div class="tts-panel">
<p class="map-title">VOICE SYNTHESIZER</p>
<textarea id="manual-tts-input" placeholder="Type text here to test the voice..."></textarea>
<button id="manual-speak-btn">SPEAK</button>
</div>
<div class="system-info">
<p>CREW AI: <span class="text-success">ONLINE</span></p>
<p>TTS ENGINE: <span class="text-success">ACTIVE</span></p>
</div>
</aside>
<!-- Main Area: Chat Interface -->
<main class="chat-main">
<header class="chat-header">
<h1>J.A.R.V.I.S.</h1>
<p>Conversational Task Deployment Interface</p>
</header>
<div class="chat-messages" id="chat-log">
<div class="message jarvis-message">
<div class="message-content">
{{ boot_greeting }}
<button class="inline-speak-btn">🔊</button>
</div>
</div>
</div>
<footer class="chat-input-area">
<div class="input-wrapper">
<input id="user-input" type="text" placeholder="Type a message to J.A.R.V.I.S..." autocomplete="off" />
<button id="send-btn" title="Send Message">
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><line x1="22" y1="2" x2="11" y2="13"></line><polygon points="22 2 15 22 11 13 2 9 22 2"></polygon></svg>
</button>
</div>
</footer>
</main>
</div>
<script src="https://cdn.socket.io/4.7.5/socket.io.min.js"></script>
<script src="/static/script.js" defer></script>
</body>
</html>
The status orb is the main indicator. JavaScript swaps its classes, so it looks idle, busy or speaking.
The ids
pin-12,pin-13andrelay-1mark the on-screen copies of the LEDs and servo, so the page can mirror the real hardware.{{ boot_greeting }}is filled in by Flask before the page reaches the browser.Finally, the Socket.IO scripts keep a live connection open, so updates arrive without reloading.
The CSS
/* --- Reset & Base --- */
:root {
--bg-dark: #050810;
--bg-sidebar: #0a0f1a;
--bg-chat: #080c15;
--primary-cyan: #00d4ff;
--text-main: #e2e8f0;
--text-dim: #64748b;
--blue-glow: #3b82f6;
--yellow-glow: #eab308;
--white-glow: #ffffff;
--user-bubble: #1e293b;
}
* { box-sizing: border-box; margin: 0; padding: 0; }
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
background-color: var(--bg-dark);
color: var(--text-main);
height: 100vh;
width: 100vw;
overflow: hidden;
}
.app-container { display: flex; height: 100vh; width: 100vw; }
/* --- Sidebar --- */
.hardware-sidebar {
width: 340px;
min-width: 340px;
background-color: var(--bg-sidebar);
border-right: 1px solid #1e293b;
display: flex;
flex-direction: column;
padding: 2rem;
box-shadow: 4px 0 15px rgba(0,0,0,0.5);
overflow-y: auto;
}
.sidebar-header h2 {
font-size: 1.2rem;
letter-spacing: 3px;
color: var(--primary-cyan);
margin-bottom: 2.5rem;
line-height: 1.5;
}
.orb-container {
display: flex;
flex-direction: column;
align-items: center;
margin-bottom: 3rem;
}
.status-orb {
width: 160px;
height: 160px;
border-radius: 50%;
transition: all 1s ease-in-out;
margin-bottom: 1.5rem;
}
.orb-idle { background: radial-gradient(circle at 30% 30%, #4b5563, #111827); box-shadow: 0 0 20px rgba(75, 85, 99, 0.2); }
.orb-researching { background: radial-gradient(circle at 30% 30%, #93c5fd, #1e3a8a); box-shadow: 0 0 60px var(--blue-glow), inset 0 0 20px rgba(59, 130, 246, 0.5); animation: pulse-blue 2s infinite; }
.orb-debating { background: radial-gradient(circle at 30% 30%, #fde047, #854d0e); box-shadow: 0 0 60px var(--yellow-glow), inset 0 0 20px rgba(234, 179, 8, 0.5); animation: pulse-yellow 1.5s infinite; }
.orb-complete { background: radial-gradient(circle at 30% 30%, #ffffff, #94a3b8); box-shadow: 0 0 80px var(--white-glow), inset 0 0 20px rgba(255, 255, 255, 0.8); }
@keyframes pulse-blue { 0%, 100% { transform: scale(1); } 50% { transform: scale(1.05); } }
@keyframes pulse-yellow { 0%, 100% { transform: scale(1); } 50% { transform: scale(1.08); } }
.status-label { font-size: 1.4rem; font-weight: bold; letter-spacing: 2px; }
.status-sub-label { font-size: 0.9rem; color: var(--text-dim); margin-top: 0.5rem; }
.hardware-map { margin-bottom: 2rem; }
.map-title { font-size: 0.8rem; color: var(--text-dim); letter-spacing: 2px; margin-bottom: 1.5rem; border-bottom: 1px solid #1e293b; padding-bottom: 0.5rem; }
.map-item { display: flex; align-items: center; gap: 1rem; margin-bottom: 1rem; font-size: 0.95rem; color: #94a3b8; }
.pin-indicator {
width: 12px; height: 12px; border-radius: 50%;
background-color: #334155;
transition: all 0.5s;
}
.pin-indicator.active-blue { background-color: var(--blue-glow); box-shadow: 0 0 10px var(--blue-glow); }
.pin-indicator.active-yellow { background-color: var(--yellow-glow); box-shadow: 0 0 10px var(--yellow-glow); }
.pin-indicator.active-white { background-color: var(--white-glow); box-shadow: 0 0 10px var(--white-glow); }
/* NEW: TTS Panel Styles */
.tts-panel {
background: rgba(15, 23, 42, 0.5);
border: 1px solid #1e293b;
border-radius: 8px;
padding: 1rem;
margin-bottom: 2rem;
}
.tts-panel textarea {
width: 100%;
height: 80px;
background: #0f172a;
border: 1px solid #334155;
border-radius: 4px;
color: var(--text-main);
padding: 0.5rem;
font-family: inherit;
font-size: 0.9rem;
resize: none;
margin-bottom: 0.75rem;
outline: none;
}
.tts-panel textarea:focus { border-color: var(--primary-cyan); }
.tts-panel button {
width: 100%;
background: transparent;
border: 1px solid var(--primary-cyan);
color: var(--primary-cyan);
padding: 0.6rem;
border-radius: 4px;
font-weight: bold;
letter-spacing: 1px;
cursor: pointer;
transition: all 0.3s;
}
.tts-panel button:hover { background: rgba(0, 212, 255, 0.15); }
.system-info { font-size: 0.85rem; color: var(--text-dim); margin-top: auto; }
.text-success { color: #22c55e; }
/* --- Main Chat Area --- */
.chat-main {
flex: 1;
display: flex;
flex-direction: column;
background-color: var(--bg-chat);
background-image:
linear-gradient(rgba(0, 212, 255, 0.03) 1px, transparent 1px),
linear-gradient(90deg, rgba(0, 212, 255, 0.03) 1px, transparent 1px);
background-size: 40px 40px;
}
.chat-header {
padding: 2rem 3rem;
border-bottom: 1px solid #1e293b;
background: rgba(8, 12, 21, 0.8);
backdrop-filter: blur(10px);
}
.chat-header h1 { font-size: 2rem; letter-spacing: 4px; color: var(--primary-cyan); }
.chat-header p { color: var(--text-dim); margin-top: 0.5rem; font-size: 1rem; }
.chat-messages {
flex: 1;
overflow-y: auto;
padding: 3rem;
display: flex;
flex-direction: column;
gap: 2rem;
}
.chat-messages::-webkit-scrollbar { width: 8px; }
.chat-messages::-webkit-scrollbar-track { background: transparent; }
.chat-messages::-webkit-scrollbar-thumb { background: #1e293b; border-radius: 4px; }
.message { display: flex; max-width: 80%; animation: fadeIn 0.3s ease-in; }
@keyframes fadeIn { from { opacity: 0; transform: translateY(10px); } to { opacity: 1; transform: translateY(0); } }
.jarvis-message { align-self: flex-start; }
.user-message { align-self: flex-end; }
.message-content {
padding: 1.5rem 2rem;
border-radius: 1.5rem;
font-size: 1.1rem;
line-height: 1.6;
position: relative;
}
.jarvis-message .message-content {
background: rgba(15, 23, 42, 0.6);
border: 1px solid rgba(0, 212, 255, 0.2);
border-bottom-left-radius: 0.3rem;
color: #e2e8f0;
}
.user-message .message-content {
background: var(--user-bubble);
border: 1px solid #334155;
border-bottom-right-radius: 0.3rem;
color: #f8fafc;
}
/* NEW: Inline Speak Button Style */
.inline-speak-btn {
position: absolute;
bottom: 10px;
right: 15px;
background: rgba(0, 0, 0, 0.3);
border: 1px solid rgba(255,255,255,0.1);
border-radius: 50%;
width: 35px;
height: 35px;
cursor: pointer;
font-size: 16px;
display: flex;
align-items: center;
justify-content: center;
opacity: 0.5;
transition: all 0.3s;
}
.inline-speak-btn:hover { opacity: 1; background: rgba(0, 212, 255, 0.2); border-color: var(--primary-cyan); }
/* Input Area */
.chat-input-area {
padding: 2rem 3rem;
background: rgba(8, 12, 21, 0.9);
border-top: 1px solid #1e293b;
}
.input-wrapper {
display: flex;
background: #0f172a;
border: 1px solid #334155;
border-radius: 1rem;
padding: 0.5rem 0.5rem 0.5rem 2rem;
transition: border-color 0.3s;
}
.input-wrapper:focus-within { border-color: var(--primary-cyan); box-shadow: 0 0 15px rgba(0, 212, 255, 0.15); }
.input-wrapper input {
flex: 1;
background: transparent;
border: none;
outline: none;
color: var(--text-main);
font-size: 1.2rem;
font-family: inherit;
}
.input-wrapper input::placeholder { color: #475569; }
.input-wrapper button {
background: var(--primary-cyan);
color: #000;
border: none;
border-radius: 0.7rem;
width: 50px; height: 50px;
cursor: pointer;
display: flex; align-items: center; justify-content: center;
transition: background 0.3s;
}
.input-wrapper button:hover { background: #33e0ff; }
At the top, colour variables keep the design consistent, because changing one colour updates the whole dashboard. Meanwhile, the orb animations pulse while the agents work, the pin styles glow blue, yellow or white to match the hardware, and the chat bubbles put your messages on the right and Jarvis’s on the left.
The JavaScript
// --- DOM Elements ---
const chatLog = document.getElementById('chat-log');
const userInput = document.getElementById('user-input');
const sendBtn = document.getElementById('send-btn');
const manualTtsInput = document.getElementById('manual-tts-input');
const manualSpeakBtn = document.getElementById('manual-speak-btn');
const orb = document.getElementById('main-orb');
const statusText = document.getElementById('status-text');
const subStatusText = document.getElementById('sub-status-text');
const pin12 = document.getElementById('pin-12');
const pin13 = document.getElementById('pin-13');
const relay1 = document.getElementById('relay-1');
// --- Global TTS Function ---
// Made global so the HTML onclick can call it directly for the first message.
// Voice synthesis runs through OmniVoice Studio (local, OpenAI-compatible TTS
// server) via the /tts backend route, instead of the browser's speechSynthesis.
let currentAudio = null;
let speakRequestId = 0;
async function speakText(text) {
const cleanText = text.replace(/\*/g, '');
if (!cleanText) return;
// Claim this call as the authoritative one and stop whatever's already
// playing. Generation takes several seconds, so if speakText() gets
// called again before this call's fetch resolves (e.g. auto-play on
// load overlapping a manual speaker click), the requestId check below
// makes the earlier call drop its result instead of also playing --
// otherwise both would end up playing at once, overlapping out of sync.
const requestId = ++speakRequestId;
if (currentAudio) {
currentAudio.pause();
currentAudio = null;
}
// Voice synthesis on this hardware can take several seconds -- show
// visible feedback so it doesn't look hung.
const previousSubStatus = subStatusText.textContent;
subStatusText.textContent = 'Synthesizing voice...';
try {
const resp = await fetch('/tts', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: cleanText }),
});
if (!resp.ok) {
const data = await resp.json().catch(() => ({}));
throw new Error(data.error || `TTS request failed (${resp.status})`);
}
const blob = await resp.blob();
if (requestId !== speakRequestId) return; // superseded by a newer call
currentAudio = new Audio(URL.createObjectURL(blob));
currentAudio.play().catch(() => {}); // autoplay may be blocked before user interaction
} catch (err) {
console.error('Voice synthesis failed:', err);
} finally {
if (requestId === speakRequestId) {
subStatusText.textContent = previousSubStatus;
}
}
}
// --- Manual TTS Sidebar Button ---
manualSpeakBtn.addEventListener('click', () => {
const text = manualTtsInput.value.trim();
if(text) {
speakText(text);
}
});
// --- Hardware State Management ---
function setHardwareState(state) {
orb.classList.remove('orb-idle', 'orb-researching', 'orb-debating', 'orb-complete');
pin12.classList.remove('active-blue');
pin13.classList.remove('active-yellow');
relay1.classList.remove('active-white');
switch(state) {
case 'researching':
orb.classList.add('orb-researching');
pin12.classList.add('active-blue');
statusText.textContent = 'RESEARCHING';
subStatusText.textContent = 'Agents gathering data...';
break;
case 'debating':
orb.classList.add('orb-debating');
pin13.classList.add('active-yellow');
statusText.textContent = 'DEBATING';
subStatusText.textContent = 'Agents evaluating options...';
break;
case 'complete':
orb.classList.add('orb-complete');
relay1.classList.add('active-white');
statusText.textContent = 'TASK COMPLETE';
subStatusText.textContent = 'Ready for next command.';
break;
default:
orb.classList.add('orb-idle');
statusText.textContent = 'STANDBY';
subStatusText.textContent = 'System Idle';
}
}
// --- Chat Logic ---
function addMessage(text, sender) {
const messageDiv = document.createElement('div');
messageDiv.classList.add('message', `${sender}-message`);
const contentDiv = document.createElement('div');
contentDiv.classList.add('message-content');
contentDiv.innerText = text;
// If it's from Jarvis, add the explicit Speak button
if(sender === 'jarvis') {
const speakBtn = document.createElement('button');
speakBtn.classList.add('inline-speak-btn');
speakBtn.innerText = '🔊';
speakBtn.addEventListener('click', (e) => {
e.stopPropagation();
speakText(text);
});
contentDiv.appendChild(speakBtn);
}
messageDiv.appendChild(contentDiv);
chatLog.appendChild(messageDiv);
chatLog.scrollTo({ top: chatLog.scrollHeight, behavior: "smooth" });
}
// --- SocketIO: live hardware state pushed from the backend ---
const socket = io();
socket.on('status_update', (data) => setHardwareState(data.state));
async function fetchJarvisResponse(userText) {
const resp = await fetch('/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: userText }),
});
const data = await resp.json();
if (!resp.ok) {
throw new Error(data.error || `Request failed (${resp.status})`);
}
return data.reply;
}
const defaultInputPlaceholder = userInput.placeholder;
async function handleSend() {
const text = userInput.value.trim();
if (!text) return;
addMessage(text, 'user');
userInput.value = '';
sendBtn.disabled = true;
userInput.disabled = true;
userInput.placeholder = 'Jarvis is responding...';
try {
const response = await fetchJarvisResponse(text);
addMessage(response, 'jarvis');
} catch (err) {
addMessage(`Apologies, I encountered an error: ${err.message}`, 'jarvis');
} finally {
sendBtn.disabled = false;
userInput.disabled = false;
userInput.placeholder = defaultInputPlaceholder;
userInput.focus();
}
}
// Event Listeners
sendBtn.addEventListener('click', handleSend);
userInput.addEventListener('keypress', (e) => {
if (e.key === 'Enter') handleSend();
});
setHardwareState('idle');
// --- Boot greeting: speak the initial message once the page loads.
// Best effort -- browsers may block audio autoplay until the user has
// interacted with the page, so a failure here is silently ignored.
function getBootGreetingText() {
const greetingEl = document.querySelector('#chat-log .message-content');
if (!greetingEl) return '';
const clone = greetingEl.cloneNode(true);
clone.querySelector('.inline-speak-btn')?.remove();
return clone.textContent.trim();
}
document.querySelector('#chat-log .inline-speak-btn')?.addEventListener('click', (e) => {
e.stopPropagation();
const greetingText = getBootGreetingText();
if (greetingText) speakText(greetingText);
});
window.addEventListener('load', () => {
const greetingText = getBootGreetingText();
if (greetingText) speakText(greetingText);
});
First, the voice code sends each reply to OmniVoice and plays the audio. It also stops any older audio first, so voices never overlap.
Next, the UI controller clears the current state and then lights the orb and pin indicators for the new one.
Then the Socket.IO listener hears state changes from Python and updates the screen at once.
Finally, the chat function sends your message to the server and shows the reply when it arrives.
The Python server
import os
import re
import time
import logging
import subprocess
import threading
import hashlib
import html
import requests
import serial
import serial.tools.list_ports
from collections import OrderedDict
from datetime import date, datetime, timezone
from dotenv import load_dotenv
from ddgs import DDGS
from flask import Flask, render_template, request, jsonify, abort, Response
from flask_socketio import SocketIO, emit # <-- ADDED
load_dotenv(override=True)
# ── LOGGING ───────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(funcName)s:%(lineno)d %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)
app = Flask(__name__, static_folder="static", template_folder="templates")
app.secret_key = os.getenv("SECRET_KEY") or os.urandom(32)
# ── ADDED: SOCKETIO SETUP ─────────────────────────────────────
socketio = SocketIO(app, cors_allowed_origins="*", async_mode="threading")
# ── CONFIGURATION ──────────────────────────────────────────────
CREWAI_CREW_URL = os.getenv("CREWAI_CREW_URL", "")
CREWAI_CREW_TOKEN = os.getenv("CREWAI_CREW_TOKEN", "")
CREWAI_KICKOFF_URL = f"{CREWAI_CREW_URL}/kickoff" if CREWAI_CREW_URL else ""
CREWAI_STATUS_URL = f"{CREWAI_CREW_URL}/status/{{kickoff_id}}" if CREWAI_CREW_URL else ""
# The input variable name your crew's task prompts expect (e.g. a task
# written as "Research {topic}..." needs CREWAI_INPUT_KEY=topic). Defaults
# to "query" -- adjust in .env to match your actual crew, not this code.
CREWAI_INPUT_KEY = os.getenv("CREWAI_INPUT_KEY", "query")
CREWAI_POLL_INTERVAL_SECONDS = 5
CREWAI_POLL_TIMEOUT_SECONDS = 240
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434/api/chat")
DEBUG_MODE = os.getenv("FLASK_DEBUG", "1") == "1"
# OmniVoice Studio -- local, OpenAI-compatible TTS server (no API key needed).
# https://github.com/debpalash/OmniVoice-Studio
OMNIVOICE_URL = os.getenv("OMNIVOICE_URL", "http://localhost:3900/v1")
OMNIVOICE_VOICE = os.getenv("OMNIVOICE_VOICE", "alloy")
# ── ARDUINO SERIAL SETUP ──────────────────────────────────────
def _find_arduino_port() -> str | None:
"""Try the env-specified port first, then scan all COM ports for an Arduino."""
env_port = os.getenv("ARDUINO_PORT", "")
candidates = []
if env_port:
candidates.append(env_port)
for p in serial.tools.list_ports.comports():
desc = (p.description or "").lower()
if any(kw in desc for kw in ("arduino", "ch340", "cp210", "usb serial")):
if p.device not in candidates:
candidates.append(p.device)
for p in serial.tools.list_ports.comports():
if p.device not in candidates:
candidates.append(p.device)
log.info(f"[HARDWARE] Port candidates to try: {candidates}")
for port in candidates:
try:
conn = serial.Serial(port=port, baudrate=9600, timeout=1)
time.sleep(2)
log.info(f"🔌 [HARDWARE] Connected successfully on {port}.")
return conn
except Exception as e:
log.debug(f"[HARDWARE] {port} failed: {e}")
return None
# Flask's debug reloader runs this module twice: once as a watcher process
# that never exits, and once as the actual server (marked by WERKZEUG_RUN_MAIN).
# Only open the serial port in the process that's actually serving requests,
# otherwise the watcher holds it open and the real server can never connect.
_should_connect_hardware = not DEBUG_MODE or os.environ.get("WERKZEUG_RUN_MAIN") == "true"
arduino = _find_arduino_port() if _should_connect_hardware else None
if arduino is None and _should_connect_hardware:
log.warning("⚠️ [HARDWARE] No Arduino found on any port. Software-only mode enabled.")
# ── OMNIVOICE AUTO-LAUNCH ──────────────────────────────────────
OMNIVOICE_APP_PATH = os.getenv(
"OMNIVOICE_APP_PATH", r"C:\Program Files\OmniVoice Studio\omnivoice-studio.exe"
)
def _ensure_omnivoice_running():
"""Launch OmniVoice Studio in the background if its API isn't already up.
Its own backend takes ~10-15s to finish loading, so this doesn't wait --
/tts just returns a clear error until it's ready."""
try:
requests.get(f"{OMNIVOICE_URL}/audio/voices", timeout=2)
log.info("🔊 [OMNIVOICE] Already running.")
return
except Exception:
pass
if not os.path.exists(OMNIVOICE_APP_PATH):
log.warning(
f"⚠️ [OMNIVOICE] Not running and no executable found at {OMNIVOICE_APP_PATH}. "
"Set OMNIVOICE_APP_PATH in .env if it's installed elsewhere."
)
return
try:
subprocess.Popen([OMNIVOICE_APP_PATH], close_fds=True)
log.info("🔊 [OMNIVOICE] Not running -- launching it in the background.")
except Exception as e:
log.error(f"❌ [OMNIVOICE] Failed to launch: {e}")
if _should_connect_hardware:
_ensure_omnivoice_running()
# ── MODIFIED: JARVIS HARDWARE STATES ──────────────────────────
def trigger_hardware_state(state: str):
"""Sends signals to Arduino based on JARVIS UI states and emits to Web UI."""
signals = {
"idle": "0",
"researching": "B", # Blue LED
"debating": "Y", # Yellow LED
"complete": "W" # White LED / Relay
}
sig = signals.get(state, "0")
# 1. Push visual update to the Web UI instantly via WebSocket
socketio.emit('status_update', {'state': state})
# 2. Push physical update to Arduino
if not arduino or not arduino.is_open:
return
try:
arduino.write(sig.encode('utf-8'))
log.info(f"🔌 [HARDWARE] Sent '{sig}' -> State: {state.upper()}")
except Exception as e:
log.error(f"❌ Failed writing to serial interface: {e}")
_reconnect_arduino()
def _reconnect_arduino():
"""A write failure usually means the board dropped off and Windows
re-enumerated it on a new COM port (often a brownout from something like
a servo drawing too much current off the board's own 5V rail during a
long-running state) -- the old handle is dead for the rest of the
process otherwise, so re-scan and pick up wherever it lands."""
global arduino
try:
if arduino:
arduino.close()
except Exception:
pass
arduino = _find_arduino_port()
if arduino is None:
log.warning("⚠️ [HARDWARE] Reconnect failed -- no Arduino found. Software-only mode until next successful signal.")
# ── RATE LIMITER ──────────────────────────────────────────────
class RateLimiter:
def __init__(self, max_requests: int = 10, window_seconds: int = 60):
self.max_requests = max_requests
self.window = window_seconds
self._requests: dict[str, list[float]] = {}
self._lock = threading.Lock()
def is_allowed(self, key: str) -> bool:
now = time.time()
with self._lock:
if key not in self._requests:
self._requests[key] = []
self._requests[key] = [t for t in self._requests[key] if now - t < self.window]
if len(self._requests[key]) >= self.max_requests:
return False
self._requests[key].append(now)
return True
rate_limiter = RateLimiter(max_requests=10, window_seconds=60)
# ── BOUNDED LRU CACHE ────────────────────────────────────────
class LRUCache:
def __init__(self, ttl: int = 3600, max_size: int = 128):
self.ttl = ttl
self.max_size = max_size
self._store: OrderedDict[str, tuple[float, dict]] = OrderedDict()
self._lock = threading.Lock()
def get(self, key: str):
with self._lock:
if key not in self._store: return None
ts, data = self._store[key]
if time.time() - ts > self.ttl:
del self._store[key]
return None
self._store.move_to_end(key)
return data
def set(self, key: str, data: dict):
with self._lock:
if key in self._store:
del self._store[key]
elif len(self._store) >= self.max_size:
self._store.popitem(last=False)
self._store[key] = (time.time(), data)
cache = LRUCache(ttl=int(os.getenv("CACHE_TTL", 3600)), max_size=128)
# OmniVoice's own synthesis takes several seconds regardless of engine choice
# (measured, not our overhead) -- cache generated audio so repeated lines
# (e.g. the boot greeting, replayed on every page load) are instant after
# the first generation.
tts_cache = LRUCache(ttl=int(os.getenv("CACHE_TTL", 3600)), max_size=64)
# ── GPU TIER SYSTEM ──────────────────────────────────────────
GPU_TIERS = [
{"min_vram": 48, "model": "qwen2.5:72b", "num_ctx": 32768, "num_batch": 2048, "search_results": 10},
{"min_vram": 40, "model": "llama3.3:70b", "num_ctx": 32768, "num_batch": 2048, "search_results": 10},
{"min_vram": 32, "model": "qwen2.5:32b", "num_ctx": 24576, "num_batch": 1024, "search_results": 8},
{"min_vram": 24, "model": "qwen2.5:32b", "num_ctx": 16384, "num_batch": 1024, "search_results": 8},
{"min_vram": 16, "model": "qwen2.5:14b", "num_ctx": 16384, "num_batch": 512, "search_results": 7},
{"min_vram": 12, "model": "qwen2.5:14b", "num_ctx": 4096, "num_batch": 512, "search_results": 5},
{"min_vram": 8, "model": "llama3.1:8b", "num_ctx": 8192, "num_batch": 256, "search_results": 5},
{"min_vram": 0, "model": "qwen2.5-coder:1.5b","num_ctx": 4096, "num_batch": 128, "search_results": 4},
]
OVERRIDE_MODEL = os.getenv("OVERRIDE_MODEL", "")
def detect_gpu_vram() -> float:
try:
result = subprocess.run(
["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits"],
capture_output=True, text=True, timeout=5,
)
if result.returncode == 0 and result.stdout.strip():
total_mb = sum(int(line.strip()) for line in result.stdout.strip().split("\n") if line.strip().isdigit())
return round(total_mb / 1024, 1)
except Exception: pass
return 0.0
def select_gpu_tier(vram_gb: float) -> dict:
for tier in GPU_TIERS:
if vram_gb >= tier["min_vram"]:
selected = dict(tier)
if OVERRIDE_MODEL: selected["model"] = OVERRIDE_MODEL
selected["num_thread"] = 4
selected["keep_alive"] = os.getenv("OLLAMA_KEEP_ALIVE", "5m")
selected["temperature"] = 0.7 # Slightly higher for conversational chat
return selected
return dict(GPU_TIERS[-1])
GPU_VRAM = detect_gpu_vram()
GPU_CONFIG = select_gpu_tier(GPU_VRAM)
log.info("═" * 50)
log.info(f" GPU VRAM: {GPU_VRAM}GB")
log.info(f" Selected Model: {GPU_CONFIG['model']}")
log.info(f" Context Window: {GPU_CONFIG['num_ctx']} tokens")
log.info("═" * 50)
# ── JARVIS CHAT PROMPT ────────────────────────────────────────
JARVIS_SYSTEM_PROMPT = """You are J.A.R.V.I.S., a highly advanced, professional AI assistant created by Tony Stark.
You are helpful, concise, and use modern technology terminology. You have access to live internet search results to answer questions accurately.
Do not use markdown headers like # or ##. Just provide well-structured plain text responses."""
BOOT_GREETING = (
"Good evening, sir. Diagnostics complete, all systems nominal, "
"and I've taken the liberty of standing by. How may I be of service?"
)
# ── HELPERS ────────────────────────────────────────────────────
def get_crew_headers() -> dict:
return {"Authorization": f"Bearer {CREWAI_CREW_TOKEN}", "Content-Type": "application/json"}
STOP_WORDS = {"and", "or", "the", "a", "an", "of", "in", "on", "at", "to", "for", "is", "are", "was", "were", "with", "how", "what", "why", "who"}
def topic_keywords(topic: str) -> list[str]:
words = re.findall(r"[a-zA-Z0-9]+", topic.lower())
return [w for w in words if w not in STOP_WORDS and len(w) > 2] or [topic.lower()]
def sanitize_topic(topic: str) -> str:
topic = re.sub(r"<[^>]*>", "", topic)
topic = html.escape(topic)
topic = re.sub(r"[\x00-\x1f\x7f]", "", topic)
return re.sub(r"\s+", " ", topic).strip()
def sanitize_speech_text(text: str) -> str:
text = re.sub(r"<[^>]*>", "", text)
text = re.sub(r"[\x00-\x1f\x7f]", "", text)
return re.sub(r"\s+", " ", text).strip()
def topic_specific_search(topic: str, max_results: int = 5) -> str:
keywords = topic_keywords(topic)
if not keywords: return "No specific keywords found."
query = " ".join(keywords[:5])
try:
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=max_results))
if not results: return f"No search results found for {topic}."
return "\n---\n".join([f"Title: {r.get('title', '')}\nSummary: {r.get('body', '')}" for r in results])
except Exception as e:
log.warning(f"Search failed: {e}")
return "Search engine unavailable."
def crewai_research(topic: str) -> str:
"""Kick off the configured Crew and poll for its result. Falls back to
the direct DDGS search if CrewAI isn't configured, or if the run errors
out or exceeds the poll timeout -- so /chat still works without a crew."""
if not CREWAI_CREW_URL:
return topic_specific_search(topic)
try:
kickoff = requests.post(
CREWAI_KICKOFF_URL,
headers=get_crew_headers(),
json={"inputs": {CREWAI_INPUT_KEY: topic}},
timeout=15,
)
kickoff.raise_for_status()
kickoff_data = kickoff.json()
kickoff_id = kickoff_data.get("kickoff_id") or kickoff_data.get("crew_id")
if not kickoff_id:
raise ValueError(f"No kickoff_id in kickoff response: {kickoff_data}")
deadline = time.time() + CREWAI_POLL_TIMEOUT_SECONDS
while time.time() < deadline:
time.sleep(CREWAI_POLL_INTERVAL_SECONDS)
status_resp = requests.get(
CREWAI_STATUS_URL.format(kickoff_id=kickoff_id),
headers=get_crew_headers(),
timeout=15,
)
status_resp.raise_for_status()
status_data = status_resp.json()
state = status_data.get("status")
if state == "completed":
result = status_data.get("result") or status_data.get("output") or ""
return str(result) if result else "The research crew completed but returned no result."
if state == "failed":
raise RuntimeError(status_data.get("error", "Crew run failed with no error detail."))
raise TimeoutError(f"Crew run did not complete within {CREWAI_POLL_TIMEOUT_SECONDS}s.")
except Exception as e:
log.warning(f"⚠️ [CREWAI] Research run failed, falling back to direct search: {e}")
return topic_specific_search(topic)
def ollama_chat(messages: list, retries: int = 2) -> str:
payload = {
"model": GPU_CONFIG["model"], "messages": messages, "stream": False,
"keep_alive": GPU_CONFIG["keep_alive"],
"options": {"temperature": GPU_CONFIG["temperature"], "num_ctx": GPU_CONFIG["num_ctx"], "num_batch": GPU_CONFIG["num_batch"]}
}
for attempt in range(retries + 1):
try:
timeout = 120 + (GPU_CONFIG["num_ctx"] // 1024) * 30
resp = requests.post(OLLAMA_URL, json=payload, timeout=timeout)
resp.raise_for_status()
return resp.json().get("message", {}).get("content", "").strip()
except Exception as e:
log.warning(f"Ollama failure: {e}")
time.sleep(3 ** attempt)
return "I apologize, but I'm having trouble reaching my reasoning engine right now. Please try again shortly."
# ── ROUTES ─────────────────────────────────────────────────────
@app.route("/")
def index():
return render_template("index.html", boot_greeting=BOOT_GREETING)
@app.route("/chat", methods=["POST"])
def chat():
if not rate_limiter.is_allowed(request.remote_addr or "unknown"):
return jsonify({"error": "Rate limit exceeded. Please slow down."}), 429
data = request.get_json(silent=True) or {}
user_message = sanitize_topic(str(data.get("message", "")).strip())
if not user_message:
return jsonify({"error": "Message cannot be empty."}), 400
cache_key = hashlib.sha256(user_message.lower().encode("utf-8")).hexdigest()
cached = cache.get(cache_key)
if cached:
return jsonify(cached)
trigger_hardware_state("researching")
try:
search_context = topic_specific_search(user_message)
messages = [
{"role": "system", "content": JARVIS_SYSTEM_PROMPT},
{"role": "system", "content": f"Live search results:\n{search_context}"},
{"role": "user", "content": user_message},
]
trigger_hardware_state("debating")
reply = ollama_chat(messages)
finally:
trigger_hardware_state("complete")
result = {"reply": reply}
cache.set(cache_key, result)
return jsonify(result)
def _tts_cache_key(text: str) -> str:
return hashlib.sha256(f"{OMNIVOICE_VOICE}:{text}".lower().encode("utf-8")).hexdigest()
def _synthesize_speech(text: str) -> bytes:
"""Get audio bytes for text, using the cache if present. Raises on failure.
OmniVoice takes ~15-60s to finish booting after _ensure_omnivoice_running()
launches it, so a freshly-started server routinely refuses the first
request or two -- retry through that startup window before giving up."""
cache_key = _tts_cache_key(text)
cached_audio = tts_cache.get(cache_key)
if cached_audio:
return cached_audio
connect_retries = 10
for attempt in range(connect_retries):
try:
resp = requests.post(
f"{OMNIVOICE_URL}/audio/speech",
json={"model": "tts-1", "voice": OMNIVOICE_VOICE, "input": text, "response_format": "wav"},
timeout=60,
)
resp.raise_for_status()
tts_cache.set(cache_key, resp.content)
return resp.content
except requests.exceptions.ConnectionError:
if attempt == connect_retries - 1:
raise
log.info(f"🔊 [OMNIVOICE] Not up yet (attempt {attempt + 1}/{connect_retries}), retrying...")
time.sleep(5)
@app.route("/tts", methods=["POST"])
def tts():
if not rate_limiter.is_allowed(request.remote_addr or "unknown"):
return jsonify({"error": "Rate limit exceeded. Please slow down."}), 429
data = request.get_json(silent=True) or {}
text = sanitize_speech_text(str(data.get("text", "")).strip())
if not text:
return jsonify({"error": "Text cannot be empty."}), 400
try:
audio = _synthesize_speech(text)
except requests.exceptions.ConnectionError as e:
log.error(f"❌ OmniVoice still unreachable: {e}")
return jsonify({"error": "OmniVoice Studio is still starting up. Try again in a few seconds."}), 502
except Exception as e:
log.error(f"❌ OmniVoice TTS failed: {e}")
return jsonify({"error": "Voice synthesis is unavailable. Is OmniVoice Studio running?"}), 502
return Response(audio, mimetype="audio/wav")
def _warm_boot_greeting_cache():
"""Pre-generate the boot greeting's audio in the background so it's
already cached by the time a browser actually loads the page, instead of
the user's first page load being the thing that eats the OmniVoice
startup + generation wait."""
try:
log.info("🔊 [OMNIVOICE] Pre-warming boot greeting audio in the background...")
_synthesize_speech(BOOT_GREETING)
log.info("🔊 [OMNIVOICE] Boot greeting cached and ready.")
except Exception as e:
log.warning(f"⚠️ [OMNIVOICE] Boot greeting pre-warm failed (will retry on first real request): {e}")
if _should_connect_hardware:
threading.Thread(target=_warm_boot_greeting_cache, daemon=True).start()
# ── ENTRYPOINT ─────────────────────────────────────────────────
if __name__ == "__main__":
socketio.run(app, host="0.0.0.0", port=5000, debug=DEBUG_MODE)
Socket.IO lets the server push hardware changes to the dashboard instantly.
Next, the trigger state function sends a letter to the Arduino over USB and tells the dashboard at the same time, so the real and on-screen lights always match.
Then the chat route checks the rate limit, switches to “researching”, gathers web context and asks Ollama for the answer.
Finally, the TTS route passes text to OmniVoice, caches the audio and copes with the voice server still starting up.
A hardware test script
This script checks the wiring without the rest of the app. It cycles through all four states:
"""Standalone Arduino connectivity check -- cycles through every hardware
state so you can confirm wiring/serial without starting the full Flask app.
Usage: python hardware/test_connection.py
"""
import os
import sys
import time
import serial
import serial.tools.list_ports
from dotenv import load_dotenv
load_dotenv(override=True)
STATES = [
("0", "IDLE", "Everything off, servo parked at 90 degrees."),
("B", "RESEARCHING", "Blue LED (pin 12) should pulse, servo should sweep."),
("Y", "DEBATING", "Yellow LED (pin 13) should pulse faster, servo should sweep."),
("W", "COMPLETE", "LEDs off, servo should park/stop."),
("0", "IDLE", "Back to everything off."),
]
SECONDS_PER_STATE = 4
def find_port() -> str | None:
env_port = os.getenv("ARDUINO_PORT", "")
if env_port:
return env_port
ports = list(serial.tools.list_ports.comports())
for p in ports:
desc = (p.description or "").lower()
if any(kw in desc for kw in ("arduino", "ch340", "cp210", "usb serial")):
return p.device
if ports:
print("No Arduino-looking device found. Available ports:")
for p in ports:
print(f" {p.device} - {p.description}")
return None
def main():
port = find_port()
if not port:
print("No serial port found. Is the Arduino plugged in?")
sys.exit(1)
print(f"Connecting to {port} ...")
try:
conn = serial.Serial(port=port, baudrate=9600, timeout=1)
except Exception as e:
print(f"FAILED to open {port}: {e}")
sys.exit(1)
time.sleep(2) # let the board reset after the connection opens
print(f"Connected on {port}.\n")
try:
for signal, name, expectation in STATES:
print(f"--> Sending '{signal}' ({name}). {expectation}")
conn.write(signal.encode("utf-8"))
time.sleep(SECONDS_PER_STATE)
print("\nDone. If you saw the LEDs/servo react as described, wiring and serial are good.")
finally:
conn.close()
if __name__ == "__main__":
main()
It finds the Arduino’s port by itself, then waits two seconds after connecting, because opening the port restarts the board. Each state lasts four seconds, so you can check the lights, and the port always closes at the end.
The Arduino sketch
// J.A.R.V.I.S. hardware state machine.
// Receives one-character state signals over serial (9600 baud) from app.py's
// trigger_hardware_state() and drives LEDs/servo to match:
// '0' idle -> everything off, servo parked/stopped
// 'B' researching -> blue LED pulses, loading servo sweeps
// 'Y' debating -> yellow LED pulses (faster), loading servo sweeps
// 'W' complete -> LEDs off, loading servo parks/stops (work is done)
#include <Servo.h>
const int PIN_RESEARCH_BLUE = 12;
const int PIN_DEBATE_YELLOW = 13;
const int PIN_LOADING_SERVO = 8;
const int SERVO_PARKED_ANGLE = 90; // "stopped" position
const int SERVO_MIN_ANGLE = 0;
const int SERVO_MAX_ANGLE = 180;
const int SERVO_STEP_DEGREES = 2;
const unsigned long SERVO_STEP_INTERVAL_MS = 15; // lower = faster sweep
Servo loadingServo;
char state = '0';
unsigned long lastLedToggle = 0;
bool pulseOn = false;
unsigned long lastServoStep = 0;
int servoAngle = SERVO_PARKED_ANGLE;
int servoDirection = 1;
bool servoParked = true;
void setLedsOff() {
digitalWrite(PIN_RESEARCH_BLUE, LOW);
digitalWrite(PIN_DEBATE_YELLOW, LOW);
}
void parkServo() {
if (!servoParked) {
loadingServo.write(SERVO_PARKED_ANGLE);
servoAngle = SERVO_PARKED_ANGLE;
servoParked = true;
}
}
// Continuously sweeps end-to-end while loading, like an old modem's
// activity indicator spinning to show something is happening.
void sweepServo() {
servoParked = false;
unsigned long now = millis();
if (now - lastServoStep < SERVO_STEP_INTERVAL_MS) return;
lastServoStep = now;
servoAngle += servoDirection * SERVO_STEP_DEGREES;
if (servoAngle >= SERVO_MAX_ANGLE) {
servoAngle = SERVO_MAX_ANGLE;
servoDirection = -1;
} else if (servoAngle <= SERVO_MIN_ANGLE) {
servoAngle = SERVO_MIN_ANGLE;
servoDirection = 1;
}
loadingServo.write(servoAngle);
}
void setup() {
pinMode(PIN_RESEARCH_BLUE, OUTPUT);
pinMode(PIN_DEBATE_YELLOW, OUTPUT);
setLedsOff();
loadingServo.attach(PIN_LOADING_SERVO);
loadingServo.write(SERVO_PARKED_ANGLE);
Serial.begin(9600);
}
void pulse(int pin, unsigned long intervalMs) {
unsigned long now = millis();
if (now - lastLedToggle >= intervalMs) {
pulseOn = !pulseOn;
lastLedToggle = now;
}
digitalWrite(pin, pulseOn ? HIGH : LOW);
}
void loop() {
if (Serial.available() > 0) {
char incoming = Serial.read();
if (incoming == '0' || incoming == 'B' || incoming == 'Y' || incoming == 'W') {
state = incoming;
setLedsOff();
lastLedToggle = millis();
pulseOn = true;
}
}
switch (state) {
case 'B':
pulse(PIN_RESEARCH_BLUE, 400);
sweepServo();
break;
case 'Y':
pulse(PIN_DEBATE_YELLOW, 200);
sweepServo();
break;
case 'W':
setLedsOff();
parkServo();
break;
default:
setLedsOff();
parkServo();
break;
}
}
The pin variables set where each part connects and how far and fast the servo sweeps.
setLedsOff()andparkServo()return everything to a neutral position when a task ends.The sweep and pulse code uses
millis()instead ofdelay(). As a result, the board keeps listening for commands while the lights and servo move.setup()sets the pins, attaches the servo to pin 8 and opens serial at 9600 baud.loop()reads incoming letters and runs the matching lights and movement.
Run and test J.A.R.V.I.S.
Activate the virtual environment and run
python app.py.Watch the log for Connected successfully on COM… and the chosen Ollama model.
Open
http://127.0.0.1:5000, and give OmniVoice up to 75 seconds for the greeting.Send a message. The status should cycle through researching, debating and complete, while the blue LED, yellow LED and servo follow along.
Send the same message again, and the voice should play almost instantly from the cache.
Troubleshooting
The Arduino never connects: check that
ARDUINO_PORTmatches Device Manager, because the port can change when you switch USB sockets.The app runs but the lights don’t: Flask’s debug reloader starts the code twice, so the first copy can grab the port. Restart the app fully. Also check the sketch was uploaded, not just verified.
Two voices at once: close extra browser tabs, because each one plays the greeting.
pip fails with SSL errors: antivirus HTTPS scanning (Kaspersky, for example) can cause this. Add an exception for python.exe.
The voice engine runs out of memory: switch OmniVoice to KittenTTS.
Quick health checks:
http://localhost:3900/v1/audio/voicesfor OmniVoice andhttp://localhost:11434/api/tagsfor Ollama.
Build the hardware
Here are the parts on the bench:
First, connect the Arduino’s 5V pin to the breadboard’s positive rail and GND to the negative rail. Then join the rails on both sides of the board:
Next, add the two LEDs with a 220Ω resistor each. Their short legs go to the negative rail, the blue LED’s long leg goes to pin 12 and the yellow one to pin 13:
Finally, connect the servo: black to ground, red to 5V and yellow to pin 8:
Project 2: An AI Pomodoro Timer with Arduino
📦 Full project code: AI Pomodoro Manager on GitHub
The second project is a Pomodoro timer on a 4-digit display. When the work session ends, or when you press a button, a CrewAI crew researches a topic and reads a short briefing aloud during your break.
Set up Python
Check that Python 3.12 is installed with
py -0p. CrewAI doesn’t support 3.14 yet.Create a virtual environment in the project folder with Python 3.12, then select it in VS Code with Python: Select Interpreter.
Install the packages:
pip install crewai pyserial python-dotenv requests kittentts soundfile.Then install eSpeak NG, which the voice needs:
winget install eSpeak-NG.eSpeak-NG.
Create the crew
In CrewAI, create a new project called “The Hardware-Integrated AI Pomodoro Manager”:
Senior AI Architect: finds the latest technical news about local AI models, agent frameworks and hardware. It ignores marketing and focuses on real changes.
Technical Briefing Writer: turns that research into a short, clear summary for busy developers on a break.
Then add two tasks. The research task looks for 3–5 recent technical updates about {topic}. The summary task condenses them into one paragraph that takes under a minute to read:
Give the architect Brave Search and Tavily, and the writer Brave Search. Both use gpt-4o-mini, and the process is sequential:
Then run it once to test, and publish it to get the API URL and bearer token:
The Pomodoro software
The .env file holds four settings:
CREWAI_CREW_URL= yout crewai url
CREWAI_CREW_TOKEN=your token
ARDUINO_PORT=COM9
OMNIVOICE_VOICE=expr-voice-2-m
CREWAI_CREW_URL and CREWAI_CREW_TOKEN connect to your deployed crew, so keep the token private. ARDUINO_PORT is your board’s port, while OMNIVOICE_VOICE picks the voice.
Here is the Arduino sketch:
// --- Pin Definitions ---
// Segments: A, B, C, D, E, F, G, DP
const int segmentPins[8] = {2, 3, 4, 5, 6, 7, 8, 9};
// Digits: 1, 2, 3, 4
const int digitPins[4] = {10, 11, 12, 13};
const int timerSwitch = A0; // Starts/Pauses Pomodoro
const int pomodoroLED = A1; // Solid when working
const int aiStatusLED = A2; // Pulses/Solid for AI status
const int manualAISwitch = A3; // Forces AI to run
// --- Variables ---
unsigned long previousMillis = 0;
const long interval = 1000; // 1 second update
// Set to 5*60 for the real 5-minute Pomodoro; lowered for fast bench testing.
const int POMODORO_SECONDS = 15;
int minutes = POMODORO_SECONDS / 60;
int seconds = POMODORO_SECONDS % 60;
bool isTimerRunning = false;
// --- AI research status (red LED blinks while Python is still working) ---
bool isResearching = false;
unsigned long lastBlinkMillis = 0;
const long blinkInterval = 250; // ms between blinks while researching
bool blinkState = false;
// 0-9 Segment Truth Table (Common Cathode)
const byte numbers[10][8] = {
{1,1,1,1,1,1,0,0}, // 0
{0,1,1,0,0,0,0,0}, // 1
{1,1,0,1,1,0,1,0}, // 2
{1,1,1,1,0,0,1,0}, // 3
{0,1,1,0,0,1,1,0}, // 4
{1,0,1,1,0,1,1,0}, // 5
{1,0,1,1,1,1,1,0}, // 6
{1,1,1,0,0,0,0,0}, // 7
{1,1,1,1,1,1,1,0}, // 8
{1,1,1,1,0,1,1,0} // 9
};
void setup() {
Serial.begin(9600); // Start serial for Python communication
for(int i = 0; i < 8; i++) pinMode(segmentPins[i], OUTPUT);
for(int i = 0; i < 4; i++) pinMode(digitPins[i], OUTPUT);
pinMode(timerSwitch, INPUT_PULLUP);
pinMode(manualAISwitch, INPUT_PULLUP);
pinMode(pomodoroLED, OUTPUT);
pinMode(aiStatusLED, OUTPUT);
}
void loop() {
unsigned long currentMillis = millis();
// 1. Check Timer Switch
if (digitalRead(timerSwitch) == LOW) {
delay(200); // Simple debounce
isTimerRunning = !isTimerRunning;
digitalWrite(pomodoroLED, isTimerRunning ? HIGH : LOW);
}
// 2. Check Manual AI Switch
if (digitalRead(manualAISwitch) == LOW) {
delay(200);
Serial.println("TRIGGER_AI"); // Send signal to Python
startResearching();
}
// 3. Handle Countdown
if (isTimerRunning && (currentMillis - previousMillis >= interval)) {
previousMillis = currentMillis;
if (seconds == 0) {
if (minutes == 0) {
// Timer Hit Zero!
isTimerRunning = false;
digitalWrite(pomodoroLED, LOW);
Serial.println("TRIGGER_AI"); // Send signal to Python
startResearching();
minutes = POMODORO_SECONDS / 60; // Reset for next time
seconds = POMODORO_SECONDS % 60;
} else {
minutes--;
seconds = 59;
}
} else {
seconds--;
}
}
// 4. Blink the AI status LED while Python is still working
if (isResearching && (currentMillis - lastBlinkMillis >= blinkInterval)) {
lastBlinkMillis = currentMillis;
blinkState = !blinkState;
digitalWrite(aiStatusLED, blinkState ? HIGH : LOW);
}
// 5. Listen for Python telling us the research is done
if (Serial.available() > 0) {
String incoming = Serial.readStringUntil('\n');
incoming.trim();
if (incoming == "AI_DONE") {
isResearching = false;
digitalWrite(aiStatusLED, LOW);
}
}
// 6. Multiplex the Display
displayNumber(minutes * 100 + seconds);
}
void displayNumber(int num) {
int displayDigits[4];
displayDigits[0] = num / 1000;
displayDigits[1] = (num / 100) % 10;
displayDigits[2] = (num / 10) % 10;
displayDigits[3] = num % 10;
for (int digit = 0; digit < 4; digit++) {
// Turn off all digits
for (int i = 0; i < 4; i++) digitalWrite(digitPins[i], HIGH);
// Set segments for current digit
for (int seg = 0; seg < 8; seg++) {
digitalWrite(segmentPins[seg], numbers[displayDigits[digit]][seg]);
}
// Turn on current digit
digitalWrite(digitPins[digit], LOW);
delay(5); // Wait a tiny bit so the eye sees it
}
}
void startResearching() {
// Kicks off the blinking state; the actual blinking happens non-blocking
// in loop() so the countdown display keeps running smoothly.
isResearching = true;
lastBlinkMillis = millis();
blinkState = true;
digitalWrite(aiStatusLED, HIGH);
}
It drives the display with 8 segment pins and 4 digit pins. Then it shows each digit very quickly in turn, which is called multiplexing.
The timer switch starts and pauses a 5-minute countdown. When it reaches zero, the board sends
TRIGGER_AIto Python and starts blinking the AI LED.The second button sends
TRIGGER_AIat any time.When Python replies with
AI_DONE, the blinking stops.
And the Python side:
import os
import re
import serial
import tempfile
import time
import winsound
import requests
import truststore
from dotenv import load_dotenv
# Use Windows' own certificate trust store instead of certifi's bundled one,
# so HTTPS requests still verify correctly when Kaspersky (or similar AV)
# intercepts TLS traffic with its own locally-trusted root certificate.
truststore.inject_into_ssl()
# KittenTTS needs eSpeak NG's DLL on PATH; point at the default winget install
# location so this works regardless of the shell's own PATH being stale.
_ESPEAK_DIR = r"C:\Program Files\eSpeak NG"
_ESPEAK_DLL = os.path.join(_ESPEAK_DIR, "libespeak-ng.dll")
if os.path.isfile(_ESPEAK_DLL):
os.environ.setdefault("PHONEMIZER_ESPEAK_LIBRARY", _ESPEAK_DLL)
if _ESPEAK_DIR not in os.environ.get("PATH", ""):
os.environ["PATH"] = _ESPEAK_DIR + os.pathsep + os.environ.get("PATH", "")
load_dotenv()
# Reads from .env; update ARDUINO_PORT there to match your exact Arduino port
SERIAL_PORT = os.getenv('ARDUINO_PORT', '/dev/ttyACM0')
BAUD_RATE = 9600
# Deployed CrewAI Enterprise crew (runs in the cloud, no local LLM key needed)
CREW_URL = os.getenv('CREWAI_CREW_URL')
CREW_TOKEN = os.getenv('CREWAI_CREW_TOKEN')
POLL_SECONDS = 2
POLL_TIMEOUT_SECONDS = 300
# Voice used to read the crew's result aloud during the break
OMNIVOICE_VOICE = os.getenv('OMNIVOICE_VOICE', 'expr-voice-2-m')
_tts_model = None
def speak(text):
global _tts_model
from kittentts import KittenTTS
import soundfile as sf
if _tts_model is None:
print("Loading voice model...")
_tts_model = KittenTTS()
# KittenTTS's nano model fails on long, multi-sentence text (onnxruntime
# "invalid expand shape" on /bert/Expand), so speak it one sentence at a time.
sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]
wav_path = os.path.join(tempfile.gettempdir(), "pomodoro_summary.wav")
for sentence in sentences:
audio = _tts_model.generate(sentence, voice=OMNIVOICE_VOICE)
sf.write(wav_path, audio, 24000)
winsound.PlaySound(wav_path, winsound.SND_FILENAME)
def run_crew(arduino):
# Tell the Arduino to start blinking the red AI-status LED, and make
# sure we always tell it to stop, however this run ends up finishing.
try:
headers = {
"Authorization": f"Bearer {CREW_TOKEN}",
"Content-Type": "application/json",
}
inputs = {
'topic': 'local LLM orchestration frameworks and physical hardware integration'
}
response = requests.post(f"{CREW_URL}/kickoff", json={"inputs": inputs}, headers=headers)
response.raise_for_status()
kickoff_id = response.json()["kickoff_id"]
print(f"Crew kicked off (id={kickoff_id}), waiting for it to finish...")
elapsed = 0
status = {}
while elapsed < POLL_TIMEOUT_SECONDS:
time.sleep(POLL_SECONDS)
elapsed += POLL_SECONDS
status_response = requests.get(f"{CREW_URL}/status/{kickoff_id}", headers=headers)
status_response.raise_for_status()
status = status_response.json()
state = str(status.get("state") or status.get("status") or "").upper()
print(f"Status: {status}")
if state in ("SUCCESS", "COMPLETED", "FAILED", "ERROR"):
break
else:
print("Timed out waiting for the crew to finish.")
return
result = status.get("result") or status.get("output") or status
print("\nResult:")
print(result)
if isinstance(result, str):
speak(result)
finally:
arduino.write(b"AI_DONE\n")
def listen_and_trigger():
print(f"Connecting to Arduino on {SERIAL_PORT}...")
try:
# Establish connection to the Arduino
arduino = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1)
time.sleep(2) # Give the Arduino a moment to reset upon connection
print("Connected! Listening for the Pomodoro timer or manual switch...")
while True:
# Check if the Arduino has sent any text
if arduino.in_waiting > 0:
raw = arduino.readline()
message = raw.decode('utf-8', errors='replace').strip()
if not message:
continue
# Print anything the Arduino sends so wiring/firmware issues
# are visible instead of silently swallowed.
if message != "TRIGGER_AI":
print(f"[Arduino] {message!r}")
continue
print("\nSignal received! Triggering the AI research crew...")
run_crew(arduino)
print("\nResearch complete and summarized!")
print("Listening for the next trigger...\n")
except serial.SerialException as e:
print(f"\nSerial Connection Error: {e}")
print("Check if the Arduino is plugged in and the SERIAL_PORT in .env is correct.")
except Exception as e:
print(f"\nAn unexpected error occurred: {e}")
if __name__ == "__main__":
listen_and_trigger()
It waits for
TRIGGER_AIfrom the Arduino, then starts the crew through its/kickoffendpoint.Every 5 seconds it checks
/status, for up to 5 minutes, until the crew finishes.Then it reads the result aloud. It splits the text into sentences first, because the small KittenTTS model struggles with long text.
A
finallyblock always sendsAI_DONE, so the LED never gets stuck blinking.
Build the Pomodoro hardware
ComponentQuantityBreadboard1Arduino Uno1USB Cable (for Arduino)14-Digit 7-Segment Display1Jumper Wires (short)24Jumper Wires (long)8Resistors (220Ω or 330Ω)11 (220Ω) or 9 (330Ω)Switches2LEDs2
First, power the breadboard: 5V to the positive rail and GND to the negative rail. If your board has two halves, join their rails too. Then test the power with an LED and resistor before you wire anything else:
Next, connect the LEDs and switches straight to the Arduino: the Pomodoro LED to A1, the AI LED to A2, the timer switch to A0 and the AI button to A3. The firmware uses INPUT_PULLUP, so the switches need no resistors. A 3-pin slide switch connects like this: the left pin to ground, the middle pin to the Arduino and the right pin to 5V.
Wire the 4-digit display
Pin layouts vary between displays, so check your datasheet. On mine, the pins are numbered like this:
(Top View)
________________
| |
| 12 11 10 9 |
| | | | | |
| |
| [ 8 8 8 8 ] | <-- The 4 digits
| |
| | | | | |
| 1 2 3 4 |
|________________|
Pin NumberConnects ToPin NumberConnects To*1Segment **E**12Digit **2* Common*2Segment **D**11Digit **1* Common*3Segment **DP* (Decimal)10Segment A4Segment C9Segment F5Segment G8Digit 3 Common*6Digit **4* Common*7Segment **B*
In total, the display needs 12 Arduino pins. Each of the 8 segment pins needs a 220Ω resistor, otherwise the segments burn out:
Arduino pinDisplay pinSegment210A37B44C52D61E79F85G93DP (decimal point)
Unlike the segments, the 4 digit pins choose which digit is lit, so they connect directly, without resistors. On a common cathode display, such as a 5461AS, the Arduino turns a digit on by setting its pin LOW:
Arduino Pin–>Display PinDigit*10–>11Digit 1 (Leftmost)11–>12Digit 212–>8Digit 313–>6*Digit 4 (Rightmost)
On a common anode display, such as a 5461BS, it’s the opposite: HIGH turns a digit on. For full brightness you’d add NPN transistors, but a direct connection is fine for testing.
If the numbers look scrambled, your pinout is different. In that case, map it with a coin cell battery: hold one pin to the battery, touch the other pins in turn and note which segment lights. When a whole digit lights up, you’ve found that digit’s common pin.
Where to Take AI Agents with Arduino Next
You’ve now connected AI agents with Arduino hardware twice: once to show what the agents are doing, and once to start them from a physical button. From here, you could add sensors that give the agents real data, or let a finished crew run trigger a motor or a buzzer.
Keep learning: Creating Agents: A Complete Guide · Part 2: Creating AI Agents with Python · How to make AI in Python
What would you connect to an AI agent: an LED, a motor or something else? Tell me in the comments.
Originally published on my blog. I post new C, Python and AI-agent guides there regularly.
More from my lab: IboChat — talk to AI characters by voice in 24+ languages (AI parodies, not real people) — and free web tools on emreguzel.ca. Tutorials like this one are published first on my Coding Blog.
Top comments (1)
What are the use-cases that we are looking at here? I'm sorry, I'm lost. Does the hardware integration serves a purpose?