Hackathons/HushMap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backend Vision Update

Browse Source
This commit is contained in:
SirBlob
2026-04-12 00:06:40 +00:00
parent 1f99b1278c
commit a45d46260e
9 changed files with 682 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5 -1
View File
@@ -27,4 +27,8 @@ vite.config.ts.timestamp-*
.svelte-kit/
build/
build/
venv/
__pycache__/
README.md
+47 -28
View File
@@ -1,42 +1,61 @@
# sv
# BitCamp 2026 - AI Study Buddy & Room Monitor
Everything you need to build a Svelte project, powered by [`sv`](https://github.com/sveltejs/cli).
This project is a comprehensive solution featuring an M5GO smart device integration, an AI Voice and Vision Backend, and a Svelte frontend dashboard. It connects physical hardware to advanced AI models (Terp AI, ElevenLabs, YOLOv8) to provide a real-time study buddy experience and a study room occupancy monitor.
## Creating a project
## Project Architecture
If you're seeing this, you've probably already done this step. Congrats!
### 1. Website Frontend (`/website` & Root)
A SvelteKit application providing the user interface for our system.
- Powered by `sv` (Svelte CLI) and Bun.
- Configured for production deployment via Docker.
```sh
# create a new project
npx sv create my-app
**Developing:**
```bash
cd website
bun install
bun run dev --open
```
To recreate this project with the same configuration:
### 2. AI Backend Services (`/backend`)
A FastAPI backend providing two core capabilities:
- **Real-time Voice WebSockets (`/ws/voice`)**: Connects the M5GO device to STT (faster-whisper), an LLM (Terp AI), and TTS (ElevenLabs). It streams audio bytes natively over WebSockets.
- **Vision Occupancy API (`/api/vision/room-status`)**: Uses YOLOv8 object detection to identify people and chairs in a room image, determining if a study room is fully occupied and pairing the closest person to an available chair.
```sh
# recreate this project
bun x [email protected] create --template minimal --types ts --add tailwindcss="plugins:typography,forms" --install bun ./
**Developing:**
```bash
cd backend
pip install -r requirements.txt
uvicorn server:app --host 0.0.0.0 --port 8000
```
*(Requires `ffmpeg`, `libgl1-mesa-glx`, and `libglib2.0-0` installed on your system)*
### 3. M5GO Device (`/m5go`)
MicroPython scripts for the M5Stack M5GO device.
- Uses `uwebsockets` to connect to the backend.
- High-quality audio I2S configuration for the internal microphone and speaker.
- Push-to-talk integration: Hold Button A to talk to the AI, release to get an audio response back.
## Docker Setup
The entire stack can be run via Docker Compose, which builds both the Svelte website and the Python AI Backend.
```bash
docker-compose up --build
```
## Developing
- **Web Frontend**: Runs on port `3000`
- **AI Backend**: Runs on port `8000`
Once you've created a project and installed dependencies with `npm install` (or `pnpm install` or `yarn`), start a development server:
## Configuration
```sh
npm run dev
Make sure you set up your `.env` variables before running the Docker containers or local servers.
# or start the server and open the app in a new browser tab
npm run dev -- --open
Create a `.env` in the `/backend` folder:
```ini
ELEVENLABS_API_KEY=your_elevenlabs_api_key_here
ELEVENLABS_VOICE_ID=JBFqnCBsd6RMkjVDRZzb
TERP_AI_BEARER_TOKEN=your_jwt_token_here
TERP_AI_CONVERSATION_ID=5e752e56-06c6-ec73-1f13-456029ce1299
```
## Building
To create a production version of your app:
```sh
npm run build
```
You can preview the production build with `npm run preview`.
> To deploy your app, you may need to install an [adapter](https://svelte.dev/docs/kit/adapters) for your target environment.
Update the `/m5go/main.py` file to include your Wi-Fi credentials and the correct local IP for the WebSocket (`WS_URL`).
backend/Dockerfile
+18
View File
@@ -0,0 +1,18 @@
FROM python:3.9-slim
# Install ffmpeg and other necessary packages
RUN apt-get update && \
apt-get install -y ffmpeg libgl1-mesa-glx libglib2.0-0 && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]
backend/README.md
+110
View File
@@ -0,0 +1,110 @@
# AI Voice Services Backend
This directory contains the FastAPI backend for the AI Voice Agent, facilitating communication between the M5GO device, Terp AI, and ElevenLabs.
## Setup Instructions
### Prerequisites
1. **Python 3.9+** is recommended.
2. **FFmpeg** must be installed on the system to handle audio format conversions (MP3 to 16-bit 16kHz PCM).
- On Ubuntu/Debian: `sudo apt install ffmpeg`
- On macOS: `brew install ffmpeg`
- On Windows: Download from the [FFmpeg website](https://ffmpeg.org/download.html) and add to PATH.
### Installation
1. Navigate to the `ai_services` directory.
2. (Optional but recommended) Create a virtual environment:
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
3. Install the required Python packages:
```bash
pip install -r requirements.txt
```
### Configuration
Update the `.env` file in this directory with your ElevenLabs credentials:
```ini
ELEVENLABS_API_KEY=your_elevenlabs_api_key_here
ELEVENLABS_VOICE_ID=JBFqnCBsd6RMkjVDRZzb
```
## Running the Server
Start the FastAPI application using Uvicorn:
```bash
uvicorn server:app --host 0.0.0.0 --port 8000
```
This will start the server and make it accessible on your local network on port 8000.
## WebSocket Endpoints
### `/ws/voice`
This is the primary WebSocket endpoint used by the M5GO device for real-time voice communication.
**Protocol Flow:**
1. **Connection:** The client establishes a WebSocket connection to `ws://<server_ip>:8000/ws/voice`.
2. **Streaming Audio (Client -> Server):** While the user holds the record button, the client continuously sends binary frames containing raw audio data.
- **Expected Format:** 16-bit signed integer, 16 kHz, Mono PCM.
3. **End of Audio Signal (Client -> Server):** When the user releases the button, the client sends a JSON text frame to signal the end of the recording:
```json
{
"event": "stop_listening"
}
```
4. **Processing (Server):** Upon receiving the `stop_listening` event, the server executes the AI pipeline:
- Transcribes the accumulated PCM audio using `faster-whisper`.
- Sends the transcribed text to the Terp AI conversational endpoint and waits for the full response.
- Sends the Terp AI response text to ElevenLabs TTS.
- Converts the received TTS audio to 16-bit 16kHz Mono PCM.
5. **Streaming Response (Server -> Client):** The server sends the converted PCM audio back to the client as binary frames.
6. **End of Response (Server -> Client):** The server sends an empty binary frame (`b""`) to signal that playback is complete.
## REST Endpoints
### `/api/vision/room-status` (POST)
This endpoint uses a YOLO object detection model to detect people and chairs in a room image, determining if the room is full and pairing the closest chairs to people.
**Request:**
- `file`: (Required) The image file to analyze (e.g., JPEG, PNG) sent as multipart form-data.
**Response:**
Returns a JSON object detailing the room status, counts, and pairings.
```json
{
"room_status": "full",
"counts": {
"people": 2,
"chairs": 2
},
"pairs": [
{
"person_index": 0,
"chair_index": 1,
"distance": 150.5
}
],
"details": {
"people": [ ... ],
"chairs": [ ... ]
}
}
```
## Client Integration Notes
For the ESP32/M5GO client (`m5go/main.py`), ensure you update the `WS_URL` variable to point to the correct local IP address of the machine running this backend server.
```python
# In m5go/main.py
WS_URL = "ws://192.168.1.100:8000/ws/voice"
```
backend/requirements.txt
+10
View File
@@ -0,0 +1,10 @@
fastapi
uvicorn
websockets
faster-whisper
requests
python-dotenv
python-multipart
ultralytics
opencv-python-headless
scipy
backend/server.py
+255
View File
@@ -0,0 +1,255 @@
import os
import io
import struct
import tempfile
import subprocess
import requests
import json
import base64
import urllib.request
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, File, UploadFile
from dotenv import load_dotenv
from vision import analyze_room_image
load_dotenv()
app = FastAPI()
SAMPLE_RATE = 16000
BITS_PER_SAMPLE = 16
NUM_CHANNELS = 1
CONVERSATION_ID = os.getenv("TERP_AI_CONVERSATION_ID", "5e752e56-06c6-ec73-1f13-456029ce1299")
HEADERS = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": f"Bearer {os.getenv('TERP_AI_BEARER_TOKEN', '')}",
"content-type": "application/json",
"origin": "https://patriotai.gmu.edu",
"referer": f"https://patriotai.gmu.edu/chat/8c3fc7f0-7c8b-4f2f-849c-5e2a45915066/{CONVERSATION_ID}",
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
"x-timezone": "America/New_York",
}
def _write_wav_to_buffer(pcm_data: bytes) -> bytes:
"""Wrap raw PCM data in a WAV header and return the full WAV bytes."""
data_size = len(pcm_data)
byte_rate = SAMPLE_RATE * NUM_CHANNELS * (BITS_PER_SAMPLE // 8)
block_align = NUM_CHANNELS * (BITS_PER_SAMPLE // 8)
buf = io.BytesIO()
buf.write(b"RIFF")
buf.write(struct.pack("<I", 36 + data_size))
buf.write(b"WAVE")
buf.write(b"fmt ")
buf.write(struct.pack("<I", 16))
buf.write(struct.pack("<H", 1)) # PCM
buf.write(struct.pack("<H", NUM_CHANNELS))
buf.write(struct.pack("<I", SAMPLE_RATE))
buf.write(struct.pack("<I", byte_rate))
buf.write(struct.pack("<H", block_align))
buf.write(struct.pack("<H", BITS_PER_SAMPLE))
buf.write(b"data")
buf.write(struct.pack("<I", data_size))
buf.write(pcm_data)
return buf.getvalue()
def _transcribe_pcm(pcm_data: bytes) -> str:
"""Transcribe raw PCM audio using faster-whisper via a temp WAV file."""
from faster_whisper import WhisperModel
wav_data = _write_wav_to_buffer(pcm_data)
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".wav")
try:
with os.fdopen(tmp_fd, "wb") as f:
f.write(wav_data)
# Initialize the model (using base model for speed)
model = WhisperModel("base", device="cpu", compute_type="int8")
segments, _ = model.transcribe(tmp_path, beam_size=5)
text = " ".join([segment.text for segment in segments])
return text.strip()
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
def get_terp_ai_response(message: str) -> str:
"""Send text to Terp AI and return the full response."""
url = f"https://patriotai.gmu.edu/api/internal/userConversations/{CONVERSATION_ID}/segments"
data = json.dumps({
"question": message,
"visionImageIds": [],
"attachmentIds": [],
"segmentTraceLogLevel": "NonPersisted"
}).encode("utf-8")
req = urllib.request.Request(url, data=data, method="POST")
for key, value in HEADERS.items():
req.add_header(key, value)
full_response = ""
event = None
try:
with urllib.request.urlopen(req) as response:
while True:
line = response.readline()
if not line:
break
line = line.decode("utf-8").strip()
if line.startswith("event: "):
event = line[7:]
elif line.startswith("data: "):
data = line[6:]
decoded = base64.b64decode(data).decode("utf-8")
if event == "response-updated":
full_response += decoded
except Exception as e:
print(f"Terp AI error: {e}")
return "I am sorry, there was an error connecting to Terp AI."
return full_response
def _convert_to_pcm(audio_data: bytes, input_format: str = "mp3") -> bytes | None:
"""Convert audio data to 16-bit 16 kHz mono PCM using ffmpeg."""
try:
result = subprocess.run(
[
"ffmpeg", "-y",
"-f", input_format, "-i", "pipe:0",
"-f", "s16le",
"-acodec", "pcm_s16le",
"-ar", str(SAMPLE_RATE),
"-ac", str(NUM_CHANNELS),
"pipe:1",
],
input=audio_data,
capture_output=True,
timeout=15,
)
if result.returncode != 0:
print(f"ffmpeg conversion failed: {result.stderr.decode()[:200]}")
return None
return result.stdout
except FileNotFoundError:
print("ffmpeg not installed")
return None
except subprocess.TimeoutExpired:
print("ffmpeg conversion timed out")
return None
def _generate_tts(text: str) -> bytes | None:
"""Generate speech audio from text using ElevenLabs TTS API."""
api_key = os.getenv("ELEVENLABS_API_KEY")
voice_id = os.getenv("ELEVENLABS_VOICE_ID", "JBFqnCBsd6RMkjVDRZzb")
if not api_key or api_key == "your_elevenlabs_api_key_here":
print("ElevenLabs API key not configured")
return None
url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}"
headers = {
"xi-api-key": api_key,
"Content-Type": "application/json",
"Accept": "audio/mpeg",
}
payload = {
"text": text,
"model_id": "eleven_flash_v2_5",
"voice_settings": {
"stability": 0.5,
"similarity_boost": 0.75,
"style": 0.0,
"use_speaker_boost": True,
},
}
try:
resp = requests.post(url, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
mp3_data = resp.content
if not mp3_data:
return None
# Convert MP3 to 16-bit 16 kHz mono PCM
return _convert_to_pcm(mp3_data, input_format="mp3")
except requests.exceptions.RequestException as e:
print(f"ElevenLabs TTS error: {e}")
return None
@app.websocket("/ws/voice")
async def websocket_voice(websocket: WebSocket):
await websocket.accept()
print("Device connected to WebSocket.")
audio_buffer = bytearray()
try:
while True:
data = await websocket.receive()
if "bytes" in data:
audio_buffer.extend(data["bytes"])
elif "text" in data:
try:
msg = json.loads(data["text"])
if msg.get("event") == "stop_listening":
pcm_data = bytes(audio_buffer)
audio_buffer = bytearray() # Reset for next time
print(f"Received stop_listening event. Buffer size: {len(pcm_data)} bytes.")
if len(pcm_data) < 3200:
print("Audio too short, ignoring.")
await websocket.send_bytes(b"")
continue
# Step 1: Speech to Text
print("Transcribing...")
user_text = _transcribe_pcm(pcm_data)
if not user_text:
print("Transcription failed or empty.")
await websocket.send_bytes(b"")
continue
print(f"User said: {user_text}")
# Step 2: Terp AI
print("Sending to Terp AI...")
ai_response_text = get_terp_ai_response(user_text)
if not ai_response_text:
print("No response from Terp AI.")
await websocket.send_bytes(b"")
continue
print(f"Terp AI response: {ai_response_text}")
# Step 3: Text to Speech
print("Generating TTS...")
tts_pcm = _generate_tts(ai_response_text)
if tts_pcm:
print(f"Sending {len(tts_pcm)} bytes of PCM back to device.")
await websocket.send_bytes(tts_pcm)
else:
print("TTS failed.")
await websocket.send_bytes(b"")
except json.JSONDecodeError:
pass
except Exception as e:
print(f"Error processing message: {e}")
await websocket.send_bytes(b"")
except WebSocketDisconnect:
print("Device disconnected.")
@app.post("/api/vision/room-status")
async def check_room_status(file: UploadFile = File(...)):
contents = await file.read()
result = analyze_room_image(contents)
return result
backend/vision.py
+102
View File
@@ -0,0 +1,102 @@
import cv2
import numpy as np
from ultralytics import YOLO
from scipy.spatial.distance import cdist
from scipy.optimize import linear_sum_assignment
# Load YOLOv8 nano model (downloads automatically if not found)
# 'yolov8n.pt' is lightweight and fast for this purpose
try:
model = YOLO("yolov8n.pt")
except Exception as e:
print(f"Error loading YOLO model: {e}")
model = None
# COCO Class IDs
PERSON_CLASS_ID = 0
CHAIR_CLASS_ID = 56
def analyze_room_image(image_bytes: bytes):
if not model:
return {"error": "Vision model is not loaded"}
# Convert bytes to numpy array then to cv2 image
nparr = np.frombuffer(image_bytes, np.uint8)
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if img is None:
return {"error": "Invalid image format"}
# Run inference
results = model(img)
people = []
chairs = []
for result in results:
boxes = result.boxes
for box in boxes:
cls_id = int(box.cls[0])
conf = float(box.conf[0])
# Extract center of bounding box
x1, y1, x2, y2 = box.xyxy[0]
cx = (x1 + x2) / 2.0
cy = (y1 + y2) / 2.0
centroid = [float(cx), float(cy)]
# Only consider detections with confidence > 0.3
if conf > 0.3:
if cls_id == PERSON_CLASS_ID:
people.append({
"centroid": centroid,
"box": [float(x1), float(y1), float(x2), float(y2)],
"conf": conf
})
elif cls_id == CHAIR_CLASS_ID:
chairs.append({
"centroid": centroid,
"box": [float(x1), float(y1), float(x2), float(y2)],
"conf": conf
})
num_people = len(people)
num_chairs = len(chairs)
pairs = []
# Bipartite matching if both people and chairs exist
if num_people > 0 and num_chairs > 0:
people_coords = [p["centroid"] for p in people]
chairs_coords = [c["centroid"] for c in chairs]
# Distance matrix (Euclidean distances)
dist_matrix = cdist(people_coords, chairs_coords, metric='euclidean')
# Hungarian algorithm to minimize total distance for pairings
row_ind, col_ind = linear_sum_assignment(dist_matrix)
for person_idx, chair_idx in zip(row_ind, col_ind):
distance = float(dist_matrix[person_idx, chair_idx])
pairs.append({
"person_index": int(person_idx),
"chair_index": int(chair_idx),
"distance": distance
})
# Basic logic: room is full if there are at least as many people as chairs.
# Can be adjusted based on specific room definitions
is_full = num_people >= num_chairs if num_chairs > 0 else False
return {
"room_status": "full" if is_full else "available",
"counts": {
"people": num_people,
"chairs": num_chairs
},
"pairs": pairs,
"details": {
"people": people,
"chairs": chairs
}
}
docker-compose.yml
+9
View File
@@ -11,3 +11,12 @@ services:
- .env
environment:
- NODE_ENV=production
ai_backend:
build:
context: ./backend
dockerfile: Dockerfile
ports:
- "8000:8000"
env_file:
- ./backend/.env
m5go/main.py
+126 -324
View File
@@ -1,339 +1,141 @@
import network
import time
import machine
import json
from m5stack import *
from m5ui import *
from uiflow import *
import time
import machine
import math
# Attempt to import websocket client (standard on some micropython builds like M5Stack)
try:
import uwebsockets.client as websockets
except ImportError:
websockets = None
lcd.print("websockets module missing", 0, 0, 0xFF0000)
setScreenColor(0x222222)
# --- CONFIGURATION ---
WIFI_SSID = "YOUR_SSID"
WIFI_PASS = "YOUR_PASSWORD"
# Update this to the IP address of your backend server
WS_URL = "ws://192.168.1.100:8000/ws/voice"
# ---------------------
def draw_status(status, color):
lcd.fillRect(0, 50, 320, 50, 0x222222)
lcd.print(status, int((320 - len(status) * 12) / 2), 65, color)
lcd.print("Connecting to WiFi...", 0, 0, 0xFFFFFF)
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(WIFI_SSID, WIFI_PASS)
# Simple connection loop
attempts = 0
while not wlan.isconnected() and attempts < 20:
time.sleep(0.5)
attempts += 1
if wlan.isconnected():
lcd.clear()
lcd.print("WiFi Connected!", 0, 0, 0x00FF00)
lcd.print(wlan.ifconfig()[0], 0, 20, 0x00FF00)
else:
lcd.clear()
lcd.print("WiFi Failed", 0, 0, 0xFF0000)
# Initialize I2S for Microphone (PDM on M5GO/Fire)
try:
adc = machine.ADC(34)
adc.atten(machine.ADC.ATTN_11DB)
except:
audio_in = machine.I2S(
0,
sck=machine.Pin(12),
ws=machine.Pin(0),
sd=machine.Pin(34),
mode=machine.I2S.RX,
bits=16,
format=machine.I2S.MONO,
rate=16000,
ibuf=4096
)
except Exception as e:
lcd.print("Mic I2S Error", 0, 40, 0xFF0000)
# Initialize I2S for Speaker
try:
audio_out = machine.I2S(
1,
sck=machine.Pin(12),
ws=machine.Pin(0),
sd=machine.Pin(2),
mode=machine.I2S.TX,
bits=16,
format=machine.I2S.MONO,
rate=16000,
ibuf=8192
)
except Exception as e:
lcd.print("Speaker I2S Error", 0, 60, 0xFF0000)
ws = None
def connect_ws():
global ws
if not websockets:
draw_status("WS Lib Missing", 0xFF0000)
return False
try:
adc = machine.ADC(machine.Pin(34))
adc.atten(machine.ADC.ATTN_11DB)
except:
adc = None
if ws:
ws.close()
ws = websockets.connect(WS_URL)
return True
except Exception as e:
draw_status("WS Connection Error", 0xFF0000)
return False
def get_db():
if not adc: return 30
sum_v = 0
sum_sq = 0
count = 0
end_t = time.ticks_ms() + 40
while time.ticks_ms() < end_t:
try:
v = adc.read()
sum_v += v
sum_sq += v * v
count += 1
except:
pass
if count == 0: return 30
mean = sum_v / count
variance = (sum_sq / count) - (mean * mean)
if variance <= 1: return 30
amp = math.sqrt(variance)
if amp <= 1: return 30
# +25 scales the RMS amplitude into a natural dB range
db = 20 * math.log10(amp) + 25
return db
draw_status("Hold Button A to Talk", 0xFFFFFF)
C = {
'0': 0x222222,
'Y': 0xFFFF00,
'R': 0xFF0000,
'W': 0xFFFFFF,
'B': 0x000000,
'P': 0xFF8888,
'D': 0x555555
}
f_s_o = [
"0000000000000000",
"0000000000000000",
"000WWW0000WWW000",
"00WWBW0000WWBW00",
"00WWBW0000WWBW00",
"000WWW0000WWW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"YY000000000000YY",
"0YY0000000000YY0",
"00YY00000000YY00",
"000YYYYYYYYYY000",
"00000YYYYYY00000",
"0000000000000000",
"0000000000000000"
]
f_s_h = [
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000WWW0000WWW000",
"00WWBW0000WWBW00",
"000WWW0000WWW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"YY000000000000YY",
"0YY0000000000YY0",
"00YY00000000YY00",
"000YYYYYYYYYY000",
"00000YYYYYY00000",
"0000000000000000",
"0000000000000000"
]
f_s_c = [
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"00WWWW0000WWWW00",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"YY000000000000YY",
"0YY0000000000YY0",
"00YY00000000YY00",
"000YYYYYYYYYY000",
"00000YYYYYY00000",
"0000000000000000",
"0000000000000000"
]
f_a_o = [
"0000000000000000",
"0DDDD000000DDDD0",
"00DDDD0000DDDD00",
"000DDDD00DDDD000",
"000WWW0000WWW000",
"00WWBB0000BBWW00",
"000WWW0000WWW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000000RRRR000000",
"0000RRRRRRRR0000",
"00RRRR0000RRRR00",
"0RRR00000000RRR0",
"0000000000000000",
"0000000000000000"
]
f_a_h = [
"0000000000000000",
"0DDDD000000DDDD0",
"00DDDD0000DDDD00",
"000DDDD00DDDD000",
"0000000000000000",
"000WWW0000WWW000",
"00WWBB0000BBWW00",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000000RRRR000000",
"0000RRRRRRRR0000",
"00RRRR0000RRRR00",
"0RRR00000000RRR0",
"0000000000000000",
"0000000000000000"
]
f_a_c = [
"0000000000000000",
"0DDDD000000DDDD0",
"00DDDD0000DDDD00",
"000DDDD00DDDD000",
"0000000000000000",
"0000000000000000",
"000WW000000WW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000000RRRR000000",
"0000RRRRRRRR0000",
"00RRRR0000RRRR00",
"0RRR00000000RRR0",
"0000000000000000",
"0000000000000000"
]
f_t_1 = [
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000WWW0000WWW000",
"00WWWW0000WWWW00",
"000WWW0000WWW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"000YYYYYYYYYY000",
"0000000000000000",
"0000000000000000",
"0000000000000000"
]
f_t_2 = [
"0000000000000000",
"0000000000000000",
"00DD00000000DD00",
"000DD000000DD000",
"0000000000000000",
"000WWW0000WWW000",
"00WWBW0000WWBW00",
"000WWW0000WWW000",
"0000000000000000",
"0000000000000000",
"0000000000000000",
"0000RRRRRRRR0000",
"000RR000000RR000",
"0000000000000000",
"0000000000000000",
"0000000000000000"
]
def d_s(f, s_x, s_y, p_s):
for r in range(16):
c = 0
while c < 16:
s_c = c
v = f[r][c]
while c < 16 and f[r][c] == v:
c += 1
w = c - s_c
x_p = s_x + (s_c * p_s)
y_p = s_y + (r * p_s)
lcd.fillRect(x_p, y_p, w * p_s, p_s, C[v])
s = 14
x = 48
y = 16
a_f = 's'
b_s = 'o'
t_s = 0
c_t = 0
l_f = None
b_p = False
al_t = 0
l_db_s = ""
s_db = 30.0
l_is_angry = False
buf = bytearray(1024)
while True:
r_db = get_db()
s_db = (s_db * 0.8) + (r_db * 0.2)
db = int(s_db)
if db < 40:
i_c = 0x89b4fa
elif db < 55:
i_c = 0x94e2d5
elif db < 65:
i_c = 0xf9e2af
elif db < 80:
i_c = 0xfab387
else:
i_c = 0xf38ba8
# m5stack core button check
if btnA.isPressed():
if not ws:
draw_status("Connecting...", 0xFFFF00)
if not connect_ws():
time.sleep(1)
continue
lcd.fillRect(0, 0, 320, 4, i_c)
db_s = "Noise: %d dB" % db
if db_s != l_db_s:
lcd.fillRect(0, 4, 120, 12, 0x222222)
lcd.print(db_s, 5, 4, i_c)
l_db_s = db_s
i_a = btnA.isPressed() or btnB.isPressed() or btnC.isPressed() or db >= 65
t_f = 'a' if i_a else 's'
b_n = btnB.isPressed()
is_angry = (t_f == 'a')
if b_n or is_angry:
al_t += 1
if al_t % 4 < 2:
try:
if b_n: speaker.tone(1200, 50)
rgb.setColorAll(0xFF0000)
except:
pass
else:
try:
if b_n: speaker.tone(800, 50)
rgb.setColorAll(0x0000FF)
except:
pass
elif b_p or l_is_angry:
try:
rgb.setColorAll(0x000000)
except:
pass
al_t = 0
b_p = b_n
l_is_angry = is_angry
c_t += 1
if a_f != t_f and t_s == 0:
t_s = 1
c_t = 0
draw_status("Listening...", 0x0000FF)
if t_s > 0:
if c_t >= 2:
t_s += 1
c_t = 0
if t_s == 3:
a_f = t_f
t_s = 0
b_s = 'o'
else:
if b_s == 'o' and c_t >= 50:
b_s = 'h'
c_t = 0
elif b_s == 'h' and c_t >= 1:
b_s = 'c'
c_t = 0
elif b_s == 'c' and c_t >= 2:
b_s = 'h_o'
c_t = 0
elif b_s == 'h_o' and c_t >= 1:
b_s = 'o'
c_t = 0
c_d = (a_f, b_s, t_s)
if c_d != l_f:
if t_s == 1:
f = f_t_1 if a_f == 's' else f_t_2
elif t_s == 2:
f = f_t_2 if a_f == 's' else f_t_1
else:
if a_f == 'a':
if b_s == 'o': f = f_a_o
elif b_s in ['h', 'h_o']: f = f_a_h
else: f = f_a_c
else:
if b_s == 'o': f = f_s_o
elif b_s in ['h', 'h_o']: f = f_s_h
else: f = f_s_c
# Read and send audio while button is held
while btnA.isPressed():
try:
num_read = audio_in.readinto(buf)
if num_read and num_read > 0 and ws:
ws.send(buf[:num_read])
except Exception as e:
pass
d_s(f, x, y, s)
l_f = c_d
time.sleep(0.02)
# Button released
draw_status("Thinking...", 0xFFFF00)
try:
if ws:
ws.send(json.dumps({"event": "stop_listening"}))
# Wait for response audio
draw_status("Speaking...", 0x00FF00)
while True:
resp = ws.recv()
if resp and isinstance(resp, bytes):
if len(resp) == 0:
break # End of audio transmission
audio_out.write(resp)
else:
break # Empty or non-bytes response means end
except Exception as e:
draw_status("Error during playback", 0xFF0000)
ws = None # force reconnect next time
draw_status("Hold Button A to Talk", 0xFFFFFF)
time.sleep(0.05)