328 lines
10 KiB
Python
328 lines
10 KiB
Python
import network
|
|
import time
|
|
import machine
|
|
import json
|
|
import math
|
|
import _thread
|
|
import websocket
|
|
from m5stack import *
|
|
from m5ui import *
|
|
from uiflow import *
|
|
|
|
setScreenColor(0x222222)
|
|
|
|
# --- CONFIGURATION ---
|
|
WIFI_SSID = "Blobby"
|
|
WIFI_PASS = "73556088"
|
|
# Update this to the IP address of your backend server
|
|
WS_URL = "ws://192.168.137.1:8000/ws/voice"
|
|
# ---------------------
|
|
|
|
def draw_status(status, color):
|
|
lcd.fillRect(0, 220, 320, 20, 0x222222)
|
|
lcd.print(status, int((320 - len(status) * 8) / 2), 220, 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)
|
|
time.sleep(1)
|
|
lcd.clear()
|
|
else:
|
|
lcd.clear()
|
|
lcd.print("WiFi Failed", 0, 0, 0xFF0000)
|
|
time.sleep(2)
|
|
lcd.clear()
|
|
|
|
try:
|
|
adc = machine.ADC(34)
|
|
adc.atten(machine.ADC.ATTN_11DB)
|
|
except:
|
|
try:
|
|
adc = machine.ADC(machine.Pin(34))
|
|
adc.atten(machine.ADC.ATTN_11DB)
|
|
except:
|
|
adc = None
|
|
|
|
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
|
|
|
|
# Use your specific calculation formula to properly scale to human DB
|
|
db = 20 * math.log10(amp) + 25
|
|
return db
|
|
|
|
def init_mic():
|
|
try:
|
|
if hasattr(machine.I2S, "RX"):
|
|
audio_in = machine.I2S(
|
|
0,
|
|
sck=machine.Pin(0),
|
|
ws=machine.Pin(0),
|
|
sd=machine.Pin(34),
|
|
mode=machine.I2S.RX,
|
|
bits=16,
|
|
format=machine.I2S.MONO,
|
|
rate=16000,
|
|
ibuf=4096
|
|
)
|
|
return audio_in
|
|
else:
|
|
# Fallback for old Micropython (e.g. M5Stack UIFlow)
|
|
mode = getattr(machine.I2S, "MODE_MASTER", 1) | getattr(machine.I2S, "MODE_RX", 2)
|
|
if hasattr(machine.I2S, "MODE_PDM"):
|
|
mode |= getattr(machine.I2S, "MODE_PDM", 0)
|
|
|
|
cfmt = getattr(machine.I2S, "CHANNEL_FMT_ALL_LEFT", 1)
|
|
dfmt = getattr(machine.I2S, "FORMAT_I2S", 1)
|
|
|
|
try_args = [
|
|
([getattr(machine.I2S, "NUM0", 0), mode, 16000, 16, cfmt, dfmt], {}),
|
|
([getattr(machine.I2S, "NUM0", 0), mode, 16000, 16], {}),
|
|
([getattr(machine.I2S, "NUM0", 0)], {"mode": mode, "sample_rate": 16000, "bits": 16, "channel_format": cfmt, "data_format": dfmt}),
|
|
([getattr(machine.I2S, "NUM0", 0)], {"mode": mode, "sample_rate": 16000, "bits": 16}),
|
|
([getattr(machine.I2S, "NUM0", 0)], {"mode": mode, "bck": 0, "ws": 0, "sd": 34, "sample_rate": 16000, "bits": 16}),
|
|
([], {"mode": mode, "sample_rate": 16000, "bits": 16}),
|
|
([getattr(machine.I2S, "NUM0", 0)], {"mode": mode}),
|
|
([], {"mode": mode}),
|
|
]
|
|
|
|
audio_in = None
|
|
last_e = None
|
|
for args, kwargs in try_args:
|
|
try:
|
|
audio_in = machine.I2S(*args, **kwargs)
|
|
break
|
|
except Exception as e:
|
|
last_e = e
|
|
if audio_in is None:
|
|
raise Exception("Mic fallback failed: " + str(last_e))
|
|
return audio_in
|
|
except Exception as e:
|
|
print("Mic I2S Error:", repr(e))
|
|
lcd.print("Mic Err: " + str(e)[:20], 0, 40, 0xFF0000)
|
|
return None
|
|
|
|
def init_spk():
|
|
try:
|
|
if hasattr(machine.I2S, "TX"):
|
|
audio_out = machine.I2S(
|
|
1,
|
|
sck=machine.Pin(12),
|
|
ws=machine.Pin(0),
|
|
sd=machine.Pin(25), # M5Stack Core/GO speaker is typically on Pin 25
|
|
mode=machine.I2S.TX,
|
|
bits=16,
|
|
format=machine.I2S.MONO,
|
|
rate=16000,
|
|
ibuf=8192
|
|
)
|
|
return audio_out
|
|
else:
|
|
mode = getattr(machine.I2S, "MODE_MASTER", 1) | getattr(machine.I2S, "MODE_TX", 2)
|
|
if hasattr(machine.I2S, "MODE_DAC_BUILT_IN"):
|
|
mode |= machine.I2S.MODE_DAC_BUILT_IN
|
|
|
|
cfmt = getattr(machine.I2S, "CHANNEL_FMT_RIGHT_LEFT", 1)
|
|
dfmt = getattr(machine.I2S, "FORMAT_I2S_MSB", 1)
|
|
|
|
try_args = [
|
|
([getattr(machine.I2S, "NUM1", 1), mode, 16000, 16, cfmt, dfmt], {}),
|
|
([getattr(machine.I2S, "NUM1", 1), mode, 16000, 16], {}),
|
|
([getattr(machine.I2S, "NUM1", 1)], {"mode": mode, "sample_rate": 16000, "bits": 16, "channel_format": cfmt, "data_format": dfmt}),
|
|
([getattr(machine.I2S, "NUM1", 1)], {"mode": mode, "sample_rate": 16000, "bits": 16}),
|
|
([getattr(machine.I2S, "NUM1", 1)], {"mode": mode, "bck": 12, "ws": 0, "sd": 25, "sample_rate": 16000, "bits": 16}),
|
|
([], {"mode": mode, "sample_rate": 16000, "bits": 16}),
|
|
([getattr(machine.I2S, "NUM1", 1)], {"mode": mode}),
|
|
([], {"mode": mode}),
|
|
]
|
|
|
|
audio_out = None
|
|
last_e = None
|
|
for args, kwargs in try_args:
|
|
try:
|
|
audio_out = machine.I2S(*args, **kwargs)
|
|
break
|
|
except Exception as e:
|
|
last_e = e
|
|
if audio_out is None:
|
|
raise Exception("Spk fallback failed: " + str(last_e))
|
|
return audio_out
|
|
except Exception as e:
|
|
print("Spk I2S Error:", repr(e))
|
|
lcd.print("Spk Err: " + str(e)[:20], 0, 60, 0xFF0000)
|
|
return None
|
|
|
|
def deinit_i2s(i2s_obj):
|
|
if i2s_obj:
|
|
try:
|
|
if hasattr(i2s_obj, 'deinit'):
|
|
i2s_obj.deinit()
|
|
except Exception as e:
|
|
print("I2S Deinit Error:", e)
|
|
|
|
ws = None
|
|
|
|
def connect_ws():
|
|
global ws
|
|
try:
|
|
ws = websocket.WebSocket()
|
|
ws.connect(WS_URL)
|
|
return True
|
|
except Exception as e:
|
|
draw_status("WS Connection Error", 0xFF0000)
|
|
return False
|
|
|
|
draw_status("Hold Button A to Talk", 0xFFFFFF)
|
|
|
|
buf = bytearray(1024)
|
|
|
|
# Noise monitoring variables
|
|
l_db_s = ""
|
|
s_db = 30.0
|
|
|
|
# Location settings - change this depending on where the M5GO is placed
|
|
CURRENT_ROOM_ID = "mckeldin"
|
|
CURRENT_LAT = 38.986021
|
|
CURRENT_LNG = -76.944949
|
|
|
|
last_db_post_time = 0
|
|
|
|
while True:
|
|
# m5stack core button check
|
|
if btnA.isPressed():
|
|
if not ws:
|
|
draw_status("Connecting...", 0xFFFF00)
|
|
if not connect_ws():
|
|
time.sleep(1)
|
|
continue
|
|
|
|
draw_status("Listening...", 0x0000FF)
|
|
|
|
# Read and send audio while button is held
|
|
audio_in = init_mic()
|
|
while btnA.isPressed():
|
|
try:
|
|
if audio_in:
|
|
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
|
|
deinit_i2s(audio_in)
|
|
audio_in = None
|
|
|
|
# Button released
|
|
draw_status("Thinking...", 0xFFFF00)
|
|
try:
|
|
if ws:
|
|
ws.send(json.dumps({"event": "stop_listening"}))
|
|
|
|
# Wait for response audio
|
|
draw_status("Speaking...", 0x00FF00)
|
|
audio_out = init_spk()
|
|
while True:
|
|
resp = ws.recv()
|
|
if resp and isinstance(resp, bytes):
|
|
if len(resp) == 0:
|
|
break # End of audio transmission
|
|
|
|
# Briefly pause face animation updates while speaker is playing
|
|
# to prevent dropping packets or stuttering
|
|
if audio_out:
|
|
audio_out.write(resp)
|
|
else:
|
|
break # Empty or non-bytes response means end
|
|
deinit_i2s(audio_out)
|
|
audio_out = None
|
|
except Exception as e:
|
|
draw_status("Error during playback", 0xFF0000)
|
|
ws = None # force reconnect next time
|
|
|
|
draw_status("Hold Button A to Talk", 0xFFFFFF)
|
|
|
|
# ---------------------------------------------
|
|
# NOISE MONITORING LOOP
|
|
# ---------------------------------------------
|
|
|
|
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
|
|
|
|
lcd.fillRect(0, 0, 320, 4, i_c)
|
|
db_s = "Noise: %d dB" % db
|
|
if db_s != l_db_s:
|
|
lcd.fillRect(0, 4, 150, 15, 0x222222)
|
|
lcd.print(db_s, 5, 4, i_c)
|
|
l_db_s = db_s
|
|
|
|
# ---------------------------------------------
|
|
# POST DB DATA PERIODICALLY
|
|
# ---------------------------------------------
|
|
try:
|
|
now_ms = time.ticks_ms()
|
|
if time.ticks_diff(now_ms, last_db_post_time) > 15000: # Every 15 seconds
|
|
last_db_post_time = now_ms
|
|
try:
|
|
payload = {
|
|
"room_id": CURRENT_ROOM_ID,
|
|
"location": {
|
|
"type": "Point",
|
|
"coordinates": [CURRENT_LNG, CURRENT_LAT]
|
|
},
|
|
"db": float(db)
|
|
}
|
|
# Convert ws url from ws://ip:port/ws/voice to http://ip:port/api/study-rooms
|
|
http_url = WS_URL.replace("ws://", "http://").replace("/ws/voice", "/api/study-rooms")
|
|
import urequests
|
|
res = urequests.post(http_url, json=payload)
|
|
res.close()
|
|
except Exception as e:
|
|
print("Failed to post db data:", e)
|
|
except Exception as main_e:
|
|
pass
|
|
|
|
time.sleep(0.02)
|