import time import json import sys import machine from machine import Pin, SPI, I2C, I2S try: import network has_network = True except ImportError: has_network = False import board_config class DummyMCP: def __init__(self): self.override_active = False self.active_led_mode = "off" def update(self): pass def start(self, port=80): pass class DummyVStream: def __init__(self): self.active = False def update(self): pass def start(self): pass # Import our utility classes from shtc3_util import SHTC3 from rtc_util import PCF85063 from battery_util import BatteryMonitor from button_util import BoardButtons from rgb_led_util import BoardLED from ble_util import BLEUART from mcp_server import MCPServer from video_stream import VideoStreamServer from audio_util import ES8311 import struct import urequests from websocket_client import WebSocketClient def create_wav_header(data_size): # Generates a 44-byte WAV header for 16kHz, 16-bit mono PCM riff = b'RIFF' file_size = data_size + 36 wave = b'WAVE' fmt = b'fmt ' chunk_size = 16 audio_format = 1 # PCM channels = 1 # Mono sample_rate = 16000 bits_per_sample = 16 byte_rate = sample_rate * channels * (bits_per_sample // 8) block_align = channels * (bits_per_sample // 8) data_label = b'data' return struct.pack('<4sI4s4sIHHIIHH4sI', riff, file_size, wave, fmt, chunk_size, audio_format, channels, sample_rate, byte_rate, block_align, bits_per_sample, data_label, data_size) def socket_readline(s): line = bytearray() while True: try: char = s.recv(1) except OSError: break if not char: break line.extend(char) if char == b'\n': break return line def socket_read_exactly(s, n): res = bytearray() while len(res) < n: try: chunk = s.recv(n - len(res)) except OSError: break if not chunk: break res.extend(chunk) return res def stream_hermes_request(url, headers, filename): # Parse URL proto, _, host_port_path = url.split('/', 2) host_port = host_port_path.split('/', 1)[0] path = '/' + host_port_path.split('/', 1)[1] if '/' in host_port_path else '/' if ':' in host_port: host, port = host_port.split(':') port = int(port) else: host, port = host_port, 80 import socket addr = socket.getaddrinfo(host, port)[0][-1] s = socket.socket() s.settimeout(30.0) s.connect(addr) # Calculate file size import os try: file_size = os.stat(filename)[6] except OSError: file_size = 0 content_length = file_size + 44 # WAV header + PCM # Send request headers s.write(f"POST {path} HTTP/1.1\r\n".encode()) s.write(f"Host: {host_port}\r\n".encode()) for k, v in headers.items(): s.write(f"{k}: {v}\r\n".encode()) s.write(f"Content-Length: {content_length}\r\n".encode()) s.write(b"\r\n") # Write WAV header s.write(create_wav_header(file_size)) # Stream audio file from flash if file_size > 0: buf = bytearray(2048) with open(filename, "rb") as f: while True: n = f.readinto(buf) if n == 0: break s.write(buf[:n]) # Read status line status_line = socket_readline(s).decode() parts = status_line.split(' ') status_code = int(parts[1]) if len(parts) >= 2 else 500 # Read headers resp_headers = {} while True: line = socket_readline(s) if line == b"\r\n" or not line: break p = line.decode().split(':', 1) if len(p) == 2: resp_headers[p[0].strip().lower()] = p[1].strip() return status_code, resp_headers, s def sync_ntp_time(rtc_chip): if rtc_chip is None: return False import ntptime import wifi_config tz_offset = getattr(wifi_config, 'TZ_OFFSET', 0) print(f"Syncing time from NTP server... (Timezone offset: {tz_offset} hours)") for attempt in range(3): try: utc_sec = ntptime.time() local_sec = utc_sec + int(tz_offset * 3600) t = time.localtime(local_sec) dt = (t[0], t[1], t[2], t[6], t[3], t[4], t[5]) rtc_chip.set_datetime(dt) rtc_chip.sync_to_system() t_str = f"{t[0]:04d}-{t[1]:02d}-{t[2]:02d} {t[3]:02d}:{t[4]:02d}:{t[5]:02d}" print(f"Successfully synced RTC with NTP. Local time: {t_str}") return True except Exception as e: print(f"NTP sync attempt {attempt+1} failed: {e}") time.sleep_ms(200) return False # LED mode options for manual cycling led_modes = [ ("Red (Breathing)", lambda led: led.set_color(40, 0, 0), "breath"), ("Green (Breathing)", lambda led: led.set_color(0, 40, 0), "breath"), ("Blue (Breathing)", lambda led: led.set_color(0, 0, 40), "breath"), ("Cyan (Breathing)", lambda led: led.set_color(0, 30, 30), "breath"), ("Magenta (Breathing)", lambda led: led.set_color(30, 0, 30), "breath"), ("Rainbow Cycle", lambda led: led.set_color(30, 30, 30), "rainbow"), ("LED Off", lambda led: led.off(), "off") ] local_led_mode_idx = 1 # Green breathing def main(): # Check if we should run the audio loopback test instead import os try: os.stat("run_loopback.txt") print("run_loopback.txt found! Starting local audio loopback test...") import demo_audio_loopback demo_audio_loopback.main() return except OSError: pass global local_led_mode_idx import board_config print("=== Starting MCP Server Main Boot (Type: {}) ===".format(board_config.BOARD_TYPE)) # 1. Use pre-initialized display and buses from board_config i2c = board_config.i2c_bus spi = board_config.spi_bus display = board_config.display_instance touch = board_config.touch # Configure Audio Amp control pin to save power if sys.platform != 'rp2' and board_config.audio_amp_pin is not None: # Turn OFF amplifier on boot (active-low vs active-high) off_val = 1 if board_config.audio_amp_active_level == 0 else 0 amp_pin = Pin(board_config.audio_amp_pin, Pin.OUT, value=off_val) # 2. Initialize utility objects sensor = None rtc_chip = None if i2c is not None: if board_config.has_sensor: try: sensor = SHTC3(i2c) except Exception as e: print("Failed to initialize SHTC3:", e) if board_config.has_rtc: try: rtc_chip = PCF85063(i2c) except Exception as e: print("Failed to initialize PCF85063:", e) battery = BatteryMonitor() led = BoardLED(board_config.led_pin) buttons = None if sys.platform != 'rp2': buttons = BoardButtons() ble_name = "ESP32-S3-" + ("RLCD" if board_config.BOARD_TYPE == "WAVESHARE_RLCD" else "Touch") ble_uart = BLEUART(name=ble_name) # Sync system clock from RTC chip if rtc_chip: try: rtc_chip.sync_to_system() except Exception as e: print("Failed to sync clock:", e) # 3. Connect to Wi-Fi status check ip_addr = "Offline (USB)" mcp = DummyMCP() vstream = DummyVStream() if has_network: try: wlan = network.WLAN(network.STA_IF) ip_addr = wlan.ifconfig()[0] if wlan.isconnected() else "Disconnected" # 4. Start background TCP/UDP Video Streaming Server from video_stream import VideoStreamServer vstream = VideoStreamServer(display, tcp_port=8081, udp_port=8082) vstream.start() # 4b. Start MCP Server from mcp_server import MCPServer mcp = MCPServer(display, led, battery, sensor, rtc_chip, ble_uart, vstream=vstream, touch=touch) mcp.start(port=80) if wlan.isconnected(): sync_ntp_time(rtc_chip) except Exception as ne: print("Failed to start network services:", ne) # Default to Green Breathing led_modes[local_led_mode_idx][1](led) mcp.active_led_mode = led_modes[local_led_mode_idx][2] # Last user actions last_action_str = "Boot finished." force_dashboard_redraw = True voice_assistant_active = False def is_talk_trigger_active(): if touch: return touch.is_touched() elif buttons and buttons.key: return buttons.key.is_pressed() return False # 5. Register Button Handlers def on_key_click(): if voice_assistant_active: return global local_led_mode_idx local_led_mode_idx = (local_led_mode_idx + 1) % len(led_modes) mode_name, color_fn, mode_type = led_modes[local_led_mode_idx] color_fn(led) mcp.active_led_mode = mode_type print(f"Local Button: Cycle LED -> {mode_name}") nonlocal last_action_str, force_dashboard_redraw last_action_str = f"Local Button: {mode_name}" # Disable override when manual button is pressed mcp.override_active = False force_dashboard_redraw = True def on_boot_click(): print("Local Button: Force Dashboard refresh.") nonlocal last_action_str, force_dashboard_redraw last_action_str = "Dashboard Refreshed" # Disable override when manual button is pressed mcp.override_active = False force_dashboard_redraw = True if buttons: buttons.key.on_click(on_key_click) buttons.boot.on_click(on_boot_click) # Loop state last_dashboard_update = 0 dashboard_update_interval_ms = 5000 last_led_update = 0 last_wifi_check = 0 print("ESP32 MCP loop running...") # 6. Main execution loop while True: now = time.ticks_ms() # A. Check Wi-Fi status periodically (every 10 seconds) and auto-reconnect if has_network and time.ticks_diff(now, last_wifi_check) >= 10000: last_wifi_check = now if not wlan.isconnected(): print("Wi-Fi connection lost. Attempting reconnect...") last_action_str = "Wi-Fi Disconnected" if ip_addr != "Disconnected": ip_addr = "Disconnected" force_dashboard_redraw = True try: wlan.connect(wifi_config.WIFI_SSID, wifi_config.WIFI_PASS) except Exception as e: print("Wi-Fi reconnect trigger failed:", e) elif ip_addr == "Disconnected" or ip_addr == "Offline (USB)": ip_addr = wlan.ifconfig()[0] print(f"Wi-Fi Connected! IP Address: {ip_addr}") last_action_str = f"Wi-Fi Connected: {ip_addr}" force_dashboard_redraw = True sync_ntp_time(rtc_chip) # Check voice assistant trigger (touch screen or physical key button) if is_talk_trigger_active(): print("Voice assistant trigger detected! Starting Hermes Voice Assistant...") voice_assistant_active = True def draw_status_bar(text): y_bar = display.height - 25 display.line(0, y_bar, display.width, y_bar, 1) display.fill_rect(0, y_bar + 1, display.width, 24, 0) display.text(text, 10, y_bar + 7, 1) display.show() def clear_status_bar(): y_bar = display.height - 25 display.fill_rect(0, y_bar, display.width, 25, 0) display.show() draw_status_bar("PTT Voice: Initializing...") # Start recording immediately! tap_started = True if tap_started: draw_status_bar("Recording: 10s...") # 1. Start MCLK PWM and configure mic path based on board config mclk_pwm = None if board_config.audio_mclk_pin is not None: mclk_pin = Pin(board_config.audio_mclk_pin, Pin.OUT) mclk_pwm = machine.PWM(mclk_pin) mclk_pwm.freq(board_config.audio_mclk_freq) mclk_pwm.duty_u16(32768) if board_config.audio_mic_codec == "ES7210": from audio_util import ES7210 codec = ES7210(i2c) codec.init(sample_rate=16000, bit_width=16) else: from audio_util import ES8311 codec = ES8311(i2c) if codec.init(sample_rate=16000): codec.set_volume(80) try: codec._write(0x14, 0x1A) # Enable analog mic input & PGA codec._write(0x16, 0x01) # Enable +6dB gain boost codec._write(0x17, 0xC8) # Set ADC digital volume except Exception as e: print("Failed to set mic gain:", e) # 2. Configure I2S RX for recording (Stereo 16kHz — ES7210 outputs stereo) i2s_rx = I2S(1, sck=Pin(board_config.audio_i2s_sck), ws=Pin(board_config.audio_i2s_ws), sd=Pin(board_config.audio_i2s_rx_sd), mode=I2S.RX, ibuf=16000, rate=16000, bits=16, format=I2S.STEREO) ws_connected = False ws = None try: # Determine dynamic device ID based on board configuration device_id = "esp32_screen" if board_config.BOARD_TYPE == 'WAVESHARE_RLCD' else "little32" headers = { "Authorization": "Bearer mcT1YA1vOr9wXSiHpCYalweEGGZKX-PIfZv2drp8BSg", "X-Device-ID": device_id } ws = WebSocketClient("ws://192.168.68.126:8642/api/esp32/voice/ws", headers=headers, timeout=30) ws.connect() ws.send_text(json.dumps({ "event": "start", "device_id": device_id, "sample_rate": 16000, "channels": 1, "sample_width": 2, "format": "pcm_s16le" })) ws.recv_frame() # ready ws.recv_frame() # listening ws_connected = True except Exception as wse: print("WebSocket connect error:", wse) draw_status_bar("Connection Error") time.sleep(2) if ws_connected and ws: draw_status_bar("Recording & streaming...") # Open I2S RX for recording (Stereo 16kHz) i2s_rx = I2S(1, sck=Pin(board_config.audio_i2s_sck), ws=Pin(board_config.audio_i2s_ws), sd=Pin(board_config.audio_i2s_rx_sd), mode=I2S.RX, ibuf=16000, rate=16000, bits=16, format=I2S.STEREO) total_data_bytes = 0 buffer = bytearray(2048) mono_buf = bytearray(1024) rec_start_time = time.ticks_ms() max_rec_duration_ms = 10000 # 10 seconds max duration try: # Stream chunks while talk trigger is active while is_talk_trigger_active(): elapsed = time.ticks_diff(time.ticks_ms(), rec_start_time) if elapsed >= max_rec_duration_ms: print("Recording stopped: maximum duration reached") break bytes_read = i2s_rx.readinto(buffer) if bytes_read > 0: mono_len = bytes_read // 2 j = 0 for i in range(0, bytes_read, 4): mono_buf[j] = buffer[i] mono_buf[j + 1] = buffer[i + 1] j += 2 ws.send_binary(mono_buf[:mono_len]) total_data_bytes += mono_len except Exception as e: print("Error during recording/streaming:", e) finally: i2s_rx.deinit() if board_config.audio_mic_codec == "ES8311": try: codec._write(0x16, 0x00) # Reset mic gain except: pass if total_data_bytes < 3200: print("Recording too short, cancelling.") try: ws.send_text(json.dumps({"event": "cancel"})) ws.close() except: pass draw_status_bar("Cancelled") time.sleep(1) else: draw_status_bar("Processing...") try: ws.send_text(json.dumps({"event": "stop"})) i2s_tx = None received_audio_bytes = 0 speaker_write_failed = False while True: opcode, payload = ws.recv_frame() if opcode is None: break if opcode == 0x1: # Text JSON event try: event_data = json.loads(payload.decode('utf-8')) evt = event_data.get("event") if evt == "transcript": txt = event_data.get("text", "") print(f"Heard: {txt}") draw_status_bar(f"Heard: {txt[:20]}...") elif evt == "thinking": draw_status_bar("Thinking...") elif evt == "response_text": txt = event_data.get("text", "") print(f"Response: {txt}") elif evt == "audio_start": draw_status_bar("Playing response...") on_val = 0 if board_config.audio_amp_active_level == 0 else 1 amp_pin.value(on_val) # Enable Amp # Initialize ES8311 Speaker DAC try: from audio_util import ES8311 dac = ES8311(i2c) dac.init(sample_rate=16000) dac.set_volume(85) except Exception as dace: print("Failed to initialize ES8311 DAC for playback:", dace) i2s_format = I2S.MONO # WebSocket audio response is mono i2s_tx = I2S(1, sck=Pin(board_config.audio_i2s_sck), ws=Pin(board_config.audio_i2s_ws), sd=Pin(board_config.audio_i2s_tx_sd), mode=I2S.TX, ibuf=4096, rate=16000, bits=16, format=i2s_format) elif evt == "audio_end": if i2s_tx: time.sleep_ms(150) off_val = 1 if board_config.audio_amp_active_level == 0 else 0 amp_pin.value(off_val) # Disable Amp i2s_tx.deinit() i2s_tx = None if speaker_write_failed: draw_status_bar("Speaker write failed") else: draw_status_bar(f"Recv {received_audio_bytes} bytes") elif evt == "done": break elif evt == "error": msg = event_data.get("message", "Unknown error") print(f"Server error: {msg}") draw_status_bar(f"Error: {msg[:20]}") time.sleep(2) break except Exception as e: print("Error parsing event text:", e) elif opcode == 0x2: # Binary frame (Audio WAV chunk) if i2s_tx: chunk = payload if chunk.startswith(b'RIFF') and len(chunk) > 44: chunk = chunk[44:] try: i2s_tx.write(chunk) received_audio_bytes += len(chunk) except Exception as e: speaker_write_failed = True print("Error writing to speaker:", e) except Exception as he: print("Hermes WS query failed:", he) draw_status_bar("Connection Error") time.sleep(2) finally: try: ws.close() except: pass # Deinit MCLK PWM if mclk_pwm: try: mclk_pwm.deinit() except: pass # Debounce release at the very end of wizard clear_status_bar() release_end = time.ticks_ms() while is_talk_trigger_active(): if time.ticks_diff(time.ticks_ms(), release_end) > 2000: break time.sleep_ms(30) time.sleep_ms(200) voice_assistant_active = False last_action_str = "Voice query finished" force_dashboard_redraw = True # A. Handle non-blocking MCP client connection updates mcp.update() # B. Handle non-blocking background video stream updates vstream.update() # C. Draw local dashboard (if not overridden by MCP draw text commands or active video stream) if not mcp.override_active and not vstream.active: if force_dashboard_redraw or time.ticks_diff(now, last_dashboard_update) >= dashboard_update_interval_ms: force_dashboard_redraw = False last_dashboard_update = now # Fetch sensor data t, h = sensor.read_sensor() if sensor else (None, None) t_str = f"{t} C" if t is not None else "Error" h_str = f"{h} %" if h is not None else "Error" # Fetch battery status bat_v = battery.read_voltage() bat_p = battery.read_percentage() bat_str = f"{bat_v:.2f}V ({bat_p}%)" if bat_v is not None else "Error" # Fetch current time dt = rtc_chip.get_datetime() if rtc_chip else None time_str = f"{dt[0]:04d}-{dt[1]:02d}-{dt[2]:02d} {dt[4]:02d}:{dt[5]:02d}:{dt[6]:02d}" if dt else "RTC Error" # Draw standard status dashboard layout display.clear(0) line_w = display.width - 10 if board_config.BOARD_TYPE == 'WAVESHARE_RLCD' or sys.platform == 'rp2': title_text = "RP2350-TFT MCP SERVER" if sys.platform == 'rp2' else "Waveshare ESP32-S3-RLCD Server" display.text(title_text, 10, 10, 1) display.line(10, 20, line_w, 20, 1) display.text_large("ENVIRONMENT", 15, 30, scale=2, c=1) display.text(f"Temp : {t_str}", 25, 55, 1) display.text(f"Humid : {h_str}", 25, 70, 1) display.line(10, 95, line_w, 95, 1) display.text_large("MCP NET CONNECTION", 15, 105, scale=2, c=1) display.text(f"IP Address : {ip_addr}", 25, 130, 1) display.text(f"Port / Path : 80 /api/mcp", 25, 145, 1) display.text(f"BLE Name : {ble_name}", 25, 160, 1) display.line(10, 185, line_w, 185, 1) display.text_large("SYSTEM STATUS", 15, 195, scale=2, c=1) display.text(f"Battery : {bat_str}", 25, 220, 1) display.text(f"Time : {time_str}", 25, 235, 1) display.line(10, 255, line_w, 255, 1) display.text(f"Status: {last_action_str}", 15, 265, 1) else: title_text = "Hosyond ESP32-S3 Server" display.text(title_text, 10, 8, 1) display.line(10, 18, line_w, 18, 1) # Left Column (System & Environment) display.text("SYSTEM & ENV", 10, 28, 1) display.line(10, 38, 150, 38, 1) display.text(f"Temp : {t_str}", 10, 46, 1) display.text(f"Hum : {h_str}", 10, 58, 1) display.text(f"Bat : {bat_str}", 10, 70, 1) display.text(f"Time : {time_str[11:19]}", 10, 82, 1) display.text(f"Date : {time_str[0:10]}", 10, 94, 1) # Right Column (Network) display.text("MCP NETWORK", 170, 28, 1) display.line(170, 38, line_w, 38, 1) display.text(f"IP : {ip_addr}", 170, 46, 1) display.text("Port: 80/api/mcp", 170, 58, 1) display.text(f"BLE : {ble_name[-6:]}", 170, 70, 1) # Bottom Status display.line(10, 115, line_w, 115, 1) display.text(f"Status: {last_action_str}", 10, 125, 1) display.show() # D. Update NeoPixel animation smoothly (runs every 50ms) if time.ticks_diff(now, last_led_update) >= 50: last_led_update = now mode = mcp.active_led_mode if mode == "breath": led.update_breathing(1.5) elif mode == "rainbow": led.update_rainbow(0.4) elif mode == "off": led.off() # Poll rapidly if stream is active, otherwise sleep 50ms to save power if vstream.active: time.sleep_ms(2) else: time.sleep_ms(50) if __name__ == "__main__": main()