diff --git a/.gitignore b/.gitignore index a7237ac..d274230 100644 --- a/.gitignore +++ b/.gitignore @@ -27,4 +27,8 @@ vite.config.ts.timestamp-* .svelte-kit/ -build/ \ No newline at end of file +build/ + +venv/ + +__pycache__/ \ No newline at end of file diff --git a/README.md b/README.md index c52e69d..2b2fa87 100644 --- a/README.md +++ b/README.md @@ -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 sv@0.15.1 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`). \ No newline at end of file diff --git a/backend/Dockerfile b/backend/Dockerfile new file mode 100644 index 0000000..1c09e8c --- /dev/null +++ b/backend/Dockerfile @@ -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"] diff --git a/backend/README.md b/backend/README.md new file mode 100644 index 0000000..efe6378 --- /dev/null +++ b/backend/README.md @@ -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://: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" +``` diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..e0b9c8d --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,10 @@ +fastapi +uvicorn +websockets +faster-whisper +requests +python-dotenv +python-multipart +ultralytics +opencv-python-headless +scipy diff --git a/backend/server.py b/backend/server.py new file mode 100644 index 0000000..e3ec33a --- /dev/null +++ b/backend/server.py @@ -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(" 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 diff --git a/backend/vision.py b/backend/vision.py new file mode 100644 index 0000000..9f4d499 --- /dev/null +++ b/backend/vision.py @@ -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 + } + } diff --git a/docker-compose.yml b/docker-compose.yml index 2d3d2a9..2e103ca 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,3 +11,12 @@ services: - .env environment: - NODE_ENV=production + + ai_backend: + build: + context: ./backend + dockerfile: Dockerfile + ports: + - "8000:8000" + env_file: + - ./backend/.env diff --git a/m5go/main.py b/m5go/main.py index 6b8e13b..ecfd6fd 100644 --- a/m5go/main.py +++ b/m5go/main.py @@ -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) \ No newline at end of file + # 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)