Adolfo Reyna
692 lines
28 KiB
Python
692 lines
28 KiB
Python
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
|
|
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
|
|
|
|
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
|
|
|
|
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)
|
|
except Exception as e:
|
|
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()
|