Fix keyword argument decoding issue on MicroPython and finalize raw image endpoint and clock sync changes

This commit is contained in:
Adolfo Reyna
2026-06-19 13:30:26 -04:00
parent c944d8c48f
commit c8853cc5df
7 changed files with 711 additions and 24 deletions
+179
View File
@@ -2,6 +2,7 @@ import time
from machine import Pin, SPI from machine import Pin, SPI
import framebuf import framebuf
import micropython import micropython
import struct
class ILI9341: class ILI9341:
def __init__(self, spi, cs, dc, rst=None, bl=None, width=320, height=240, invert_color=True): def __init__(self, spi, cs, dc, rst=None, bl=None, width=320, height=240, invert_color=True):
@@ -89,6 +90,184 @@ class ILI9341:
except OSError: except OSError:
print(f"Error: Could not open {filename}") print(f"Error: Could not open {filename}")
@micropython.native
def _convert_bgr24_to_rgb565(self, bgr_buf, rgb565_buf, width, src_offset, num_pixels):
idx = 0
for i in range(src_offset, src_offset + num_pixels):
b = bgr_buf[i * 3]
g = bgr_buf[i * 3 + 1]
r = bgr_buf[i * 3 + 2]
r_5 = r >> 3
g_6 = g >> 2
b_5 = b >> 3
rgb565_buf[idx] = (r_5 << 3) | (g_6 >> 3)
rgb565_buf[idx + 1] = ((g_6 & 0x07) << 5) | b_5
idx += 2
@micropython.native
def _convert_bgra32_to_rgb565(self, bgra_buf, rgb565_buf, width, src_offset, num_pixels):
idx = 0
for i in range(src_offset, src_offset + num_pixels):
b = bgra_buf[i * 4]
g = bgra_buf[i * 4 + 1]
r = bgra_buf[i * 4 + 2]
r_5 = r >> 3
g_6 = g >> 2
b_5 = b >> 3
rgb565_buf[idx] = (r_5 << 3) | (g_6 >> 3)
rgb565_buf[idx + 1] = ((g_6 & 0x07) << 5) | b_5
idx += 2
def draw_bmp(self, filename, x=0, y=0):
"""Draw a 24-bit or 32-bit uncompressed color BMP image at (x, y) coordinates."""
try:
with open(filename, 'rb') as f:
header = f.read(54)
if len(header) < 54 or header[0:2] != b'BM':
print("Err: Not a valid BMP file")
return False
pixel_offset = struct.unpack('<I', header[10:14])[0]
width, height = struct.unpack('<ii', header[18:26])
planes, bpp = struct.unpack('<HH', header[26:30])
compression = struct.unpack('<I', header[30:34])[0]
if bpp not in (24, 32):
print("Err: Only 24-bit and 32-bit BMP formats supported")
return False
if compression != 0:
print("Err: Only uncompressed BMP supported")
return False
f.seek(pixel_offset)
bottom_up = True
if height < 0:
height = -height
bottom_up = False
row_bytes = (width * bpp) // 8
row_padded = ((width * bpp + 31) // 32) * 4
read_buf = bytearray(row_padded)
rgb565_buf = bytearray(width * 2)
for row_idx in range(height):
n = f.readinto(read_buf)
if n < row_padded:
break
screen_y = y + (height - 1 - row_idx) if bottom_up else y + row_idx
if screen_y < 0 or screen_y >= self.height:
continue
x_start = x
x_end = x + width - 1
if x_start >= self.width or x_end < 0:
continue
win_x0 = max(0, x_start)
win_x1 = min(self.width - 1, x_end)
if win_x1 < win_x0:
continue
src_offset_pixels = win_x0 - x_start
win_w = win_x1 - win_x0 + 1
# Convert pixel data to RGB565 row buffer
if bpp == 24:
self._convert_bgr24_to_rgb565(read_buf, rgb565_buf, width, src_offset_pixels, win_w)
elif bpp == 32:
self._convert_bgra32_to_rgb565(read_buf, rgb565_buf, width, src_offset_pixels, win_w)
# Draw directly to the screen via SPI window
self.set_window(win_x0, screen_y, win_x1, screen_y)
self.dc(1)
self.cs(0)
self.spi.write(memoryview(rgb565_buf)[:win_w * 2])
self.cs(1)
# Also update internal 1-bit canvas buffer for screenshots/refresh consistency
for px in range(win_w):
screen_x = win_x0 + px
src_px = src_offset_pixels + px
if bpp == 24:
b = read_buf[src_px * 3]
g = read_buf[src_px * 3 + 1]
r = read_buf[src_px * 3 + 2]
else:
b = read_buf[src_px * 4]
g = read_buf[src_px * 4 + 1]
r = read_buf[src_px * 4 + 2]
# 0 = Black, 1 = White in conversion for MONO_HLSB canvas
lum = (r * 299 + g * 587 + b * 114) // 1000
mono_c = 1 if lum >= 128 else 0
self.canvas.pixel(screen_x, screen_y, mono_c)
return True
except Exception as e:
print("Error drawing BMP:", e)
return False
@micropython.native
def _update_mono_canvas_rgb565(self, x, y, w, h, data):
for cy in range(h):
screen_y = y + cy
if screen_y < 0 or screen_y >= self.height:
continue
for cx in range(w):
screen_x = x + cx
if screen_x < 0 or screen_x >= self.width:
continue
idx = (cy * w + cx) * 2
h_byte = data[idx]
l_byte = data[idx + 1]
# Extract RGB from RGB565
r = (h_byte & 0xF8)
g = ((h_byte & 0x07) << 5) | ((l_byte & 0xE0) >> 3)
b = (l_byte & 0x1F) << 3
# Convert to luminance
lum = (r * 299 + g * 587 + b * 114) // 1000
mono = 1 if lum >= 128 else 0
self.canvas.pixel(screen_x, screen_y, mono)
def draw_rgb565(self, x, y, w, h, data):
"""Draw raw RGB565 pixel data on the screen at specified (x,y) with width and height."""
# Clip coordinates
x_start = max(0, x)
x_end = min(self.width - 1, x + w - 1)
y_start = max(0, y)
y_end = min(self.height - 1, y + h - 1)
if x_start > x_end or y_start > y_end:
return True
# Fast path: if completely visible on screen, draw in one go
if x_start == x and x_end == x + w - 1 and y_start == y and y_end == y + h - 1:
self.set_window(x_start, y_start, x_end, y_end)
self.dc(1)
self.cs(0)
self.spi.write(data)
self.cs(1)
else:
# Slow path: row-by-row clipping
for cy in range(y_start, y_end + 1):
src_y = cy - y
src_row_offset = (src_y * w + (x_start - x)) * 2
row_len_bytes = (x_end - x_start + 1) * 2
self.set_window(x_start, cy, x_end, cy)
self.dc(1)
self.cs(0)
self.spi.write(memoryview(data)[src_row_offset : src_row_offset + row_len_bytes])
self.cs(1)
# Sync the internal 1-bit canvas buffer
self._update_mono_canvas_rgb565(x, y, w, h, data)
return True
# --- SCREENSHOT --- # --- SCREENSHOT ---
def save_screenshot(self, filename): def save_screenshot(self, filename):
print(f"Saving screenshot to {filename}...") print(f"Saving screenshot to {filename}...")
+92
View File
@@ -2,6 +2,7 @@ import time
from machine import Pin, SPI from machine import Pin, SPI
import framebuf import framebuf
import micropython import micropython
import struct
class RLCD: class RLCD:
def __init__(self, spi, cs, dc, rst, width=400, height=300): def __init__(self, spi, cs, dc, rst, width=400, height=300):
@@ -82,6 +83,97 @@ class RLCD:
except OSError: except OSError:
print(f"Error: Could not open {filename}") print(f"Error: Could not open {filename}")
def draw_bmp(self, filename, x=0, y=0):
"""Draw a 24-bit or 32-bit uncompressed color BMP image converted to 1-bit monochrome at (x, y) coordinates."""
try:
with open(filename, 'rb') as f:
header = f.read(54)
if len(header) < 54 or header[0:2] != b'BM':
print("Err: Not a valid BMP file")
return False
pixel_offset = struct.unpack('<I', header[10:14])[0]
width, height = struct.unpack('<ii', header[18:26])
planes, bpp = struct.unpack('<HH', header[26:30])
compression = struct.unpack('<I', header[30:34])[0]
if bpp not in (24, 32):
print("Err: Only 24-bit and 32-bit BMP formats supported")
return False
if compression != 0:
print("Err: Only uncompressed BMP supported")
return False
f.seek(pixel_offset)
bottom_up = True
if height < 0:
height = -height
bottom_up = False
row_bytes = (width * bpp) // 8
row_padded = ((width * bpp + 31) // 32) * 4
read_buf = bytearray(row_padded)
for row_idx in range(height):
n = f.readinto(read_buf)
if n < row_padded:
break
screen_y = y + (height - 1 - row_idx) if bottom_up else y + row_idx
if screen_y < 0 or screen_y >= self.height:
continue
for px in range(width):
screen_x = x + px
if screen_x < 0 or screen_x >= self.width:
continue
if bpp == 24:
b = read_buf[px * 3]
g = read_buf[px * 3 + 1]
r = read_buf[px * 3 + 2]
else: # 32-bit
b = read_buf[px * 4]
g = read_buf[px * 4 + 1]
r = read_buf[px * 4 + 2]
# Convert to monochrome (0 = White, 1 = Black)
lum = (r * 299 + g * 587 + b * 114) // 1000
c = 1 if lum < 128 else 0
self.canvas.pixel(screen_x, screen_y, c)
self.show()
return True
except Exception as e:
print("Error drawing BMP on RLCD:", e)
return False
def draw_rgb565(self, x, y, w, h, data):
"""Draw raw RGB565 pixel data converted to 1-bit monochrome on the RLCD."""
for cy in range(h):
screen_y = y + cy
if screen_y < 0 or screen_y >= self.height:
continue
for cx in range(w):
screen_x = x + cx
if screen_x < 0 or screen_x >= self.width:
continue
idx = (cy * w + cx) * 2
h_byte = data[idx]
l_byte = data[idx + 1]
# Extract RGB from RGB565
r = (h_byte & 0xF8)
g = ((h_byte & 0x07) << 5) | ((l_byte & 0xE0) >> 3)
b = (l_byte & 0x1F) << 3
# Convert to luminance (0 = White, 1 = Black in RLCD)
lum = (r * 299 + g * 587 + b * 114) // 1000
c = 1 if lum < 128 else 0
self.canvas.pixel(screen_x, screen_y, c)
self.show()
return True
# --- SCREENSHOT --- # --- SCREENSHOT ---
def save_screenshot(self, filename): def save_screenshot(self, filename):
print(f"Saving screenshot to {filename}...") print(f"Saving screenshot to {filename}...")
+179
View File
@@ -2,6 +2,7 @@ import time
from machine import Pin, SPI from machine import Pin, SPI
import framebuf import framebuf
import micropython import micropython
import struct
class ST7796: class ST7796:
def __init__(self, spi, cs, dc, rst, bl=None, width=480, height=320, invert_color=True): def __init__(self, spi, cs, dc, rst, bl=None, width=480, height=320, invert_color=True):
@@ -89,6 +90,184 @@ class ST7796:
except OSError: except OSError:
print(f"Error: Could not open {filename}") print(f"Error: Could not open {filename}")
@micropython.native
def _convert_bgr24_to_rgb565(self, bgr_buf, rgb565_buf, width, src_offset, num_pixels):
idx = 0
for i in range(src_offset, src_offset + num_pixels):
b = bgr_buf[i * 3]
g = bgr_buf[i * 3 + 1]
r = bgr_buf[i * 3 + 2]
r_5 = r >> 3
g_6 = g >> 2
b_5 = b >> 3
rgb565_buf[idx] = (r_5 << 3) | (g_6 >> 3)
rgb565_buf[idx + 1] = ((g_6 & 0x07) << 5) | b_5
idx += 2
@micropython.native
def _convert_bgra32_to_rgb565(self, bgra_buf, rgb565_buf, width, src_offset, num_pixels):
idx = 0
for i in range(src_offset, src_offset + num_pixels):
b = bgra_buf[i * 4]
g = bgra_buf[i * 4 + 1]
r = bgra_buf[i * 4 + 2]
r_5 = r >> 3
g_6 = g >> 2
b_5 = b >> 3
rgb565_buf[idx] = (r_5 << 3) | (g_6 >> 3)
rgb565_buf[idx + 1] = ((g_6 & 0x07) << 5) | b_5
idx += 2
def draw_bmp(self, filename, x=0, y=0):
"""Draw a 24-bit or 32-bit uncompressed color BMP image at (x, y) coordinates."""
try:
with open(filename, 'rb') as f:
header = f.read(54)
if len(header) < 54 or header[0:2] != b'BM':
print("Err: Not a valid BMP file")
return False
pixel_offset = struct.unpack('<I', header[10:14])[0]
width, height = struct.unpack('<ii', header[18:26])
planes, bpp = struct.unpack('<HH', header[26:30])
compression = struct.unpack('<I', header[30:34])[0]
if bpp not in (24, 32):
print("Err: Only 24-bit and 32-bit BMP formats supported")
return False
if compression != 0:
print("Err: Only uncompressed BMP supported")
return False
f.seek(pixel_offset)
bottom_up = True
if height < 0:
height = -height
bottom_up = False
row_bytes = (width * bpp) // 8
row_padded = ((width * bpp + 31) // 32) * 4
read_buf = bytearray(row_padded)
rgb565_buf = bytearray(width * 2)
for row_idx in range(height):
n = f.readinto(read_buf)
if n < row_padded:
break
screen_y = y + (height - 1 - row_idx) if bottom_up else y + row_idx
if screen_y < 0 or screen_y >= self.height:
continue
x_start = x
x_end = x + width - 1
if x_start >= self.width or x_end < 0:
continue
win_x0 = max(0, x_start)
win_x1 = min(self.width - 1, x_end)
if win_x1 < win_x0:
continue
src_offset_pixels = win_x0 - x_start
win_w = win_x1 - win_x0 + 1
# Convert pixel data to RGB565 row buffer
if bpp == 24:
self._convert_bgr24_to_rgb565(read_buf, rgb565_buf, width, src_offset_pixels, win_w)
elif bpp == 32:
self._convert_bgra32_to_rgb565(read_buf, rgb565_buf, width, src_offset_pixels, win_w)
# Draw directly to the screen via SPI window
self.set_window(win_x0, screen_y, win_x1, screen_y)
self.dc(1)
self.cs(0)
self.spi.write(memoryview(rgb565_buf)[:win_w * 2])
self.cs(1)
# Also update internal 1-bit canvas buffer for screenshots/refresh consistency
for px in range(win_w):
screen_x = win_x0 + px
src_px = src_offset_pixels + px
if bpp == 24:
b = read_buf[src_px * 3]
g = read_buf[src_px * 3 + 1]
r = read_buf[src_px * 3 + 2]
else:
b = read_buf[src_px * 4]
g = read_buf[src_px * 4 + 1]
r = read_buf[src_px * 4 + 2]
# 0 = Black, 1 = White in conversion for MONO_HLSB canvas
lum = (r * 299 + g * 587 + b * 114) // 1000
mono_c = 1 if lum >= 128 else 0
self.canvas.pixel(screen_x, screen_y, mono_c)
return True
except Exception as e:
print("Error drawing BMP:", e)
return False
@micropython.native
def _update_mono_canvas_rgb565(self, x, y, w, h, data):
for cy in range(h):
screen_y = y + cy
if screen_y < 0 or screen_y >= self.height:
continue
for cx in range(w):
screen_x = x + cx
if screen_x < 0 or screen_x >= self.width:
continue
idx = (cy * w + cx) * 2
h_byte = data[idx]
l_byte = data[idx + 1]
# Extract RGB from RGB565
r = (h_byte & 0xF8)
g = ((h_byte & 0x07) << 5) | ((l_byte & 0xE0) >> 3)
b = (l_byte & 0x1F) << 3
# Convert to luminance
lum = (r * 299 + g * 587 + b * 114) // 1000
mono = 1 if lum >= 128 else 0
self.canvas.pixel(screen_x, screen_y, mono)
def draw_rgb565(self, x, y, w, h, data):
"""Draw raw RGB565 pixel data on the screen at specified (x,y) with width and height."""
# Clip coordinates
x_start = max(0, x)
x_end = min(self.width - 1, x + w - 1)
y_start = max(0, y)
y_end = min(self.height - 1, y + h - 1)
if x_start > x_end or y_start > y_end:
return True
# Fast path: if completely visible on screen, draw in one go
if x_start == x and x_end == x + w - 1 and y_start == y and y_end == y + h - 1:
self.set_window(x_start, y_start, x_end, y_end)
self.dc(1)
self.cs(0)
self.spi.write(data)
self.cs(1)
else:
# Slow path: row-by-row clipping
for cy in range(y_start, y_end + 1):
src_y = cy - y
src_row_offset = (src_y * w + (x_start - x)) * 2
row_len_bytes = (x_end - x_start + 1) * 2
self.set_window(x_start, cy, x_end, cy)
self.dc(1)
self.cs(0)
self.spi.write(memoryview(data)[src_row_offset : src_row_offset + row_len_bytes])
self.cs(1)
# Sync the internal 1-bit canvas buffer
self._update_mono_canvas_rgb565(x, y, w, h, data)
return True
# --- SCREENSHOT --- # --- SCREENSHOT ---
def save_screenshot(self, filename): def save_screenshot(self, filename):
print(f"Saving screenshot to {filename}...") print(f"Saving screenshot to {filename}...")
+33 -2
View File
@@ -149,6 +149,33 @@ def stream_hermes_request(url, headers, filename):
return status_code, resp_headers, s 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 mode options for manual cycling
led_modes = [ led_modes = [
("Red (Breathing)", lambda led: led.set_color(40, 0, 0), "breath"), ("Red (Breathing)", lambda led: led.set_color(40, 0, 0), "breath"),
@@ -240,6 +267,9 @@ def main():
from mcp_server import MCPServer from mcp_server import MCPServer
mcp = MCPServer(display, led, battery, sensor, rtc_chip, ble_uart, vstream=vstream, touch=touch) mcp = MCPServer(display, led, battery, sensor, rtc_chip, ble_uart, vstream=vstream, touch=touch)
mcp.start(port=80) mcp.start(port=80)
if wlan.isconnected():
sync_ntp_time(rtc_chip)
except Exception as ne: except Exception as ne:
print("Failed to start network services:", ne) print("Failed to start network services:", ne)
@@ -317,6 +347,7 @@ def main():
print(f"Wi-Fi Connected! IP Address: {ip_addr}") print(f"Wi-Fi Connected! IP Address: {ip_addr}")
last_action_str = f"Wi-Fi Connected: {ip_addr}" last_action_str = f"Wi-Fi Connected: {ip_addr}"
force_dashboard_redraw = True force_dashboard_redraw = True
sync_ntp_time(rtc_chip)
# Check voice assistant trigger (touch screen or physical key button) # Check voice assistant trigger (touch screen or physical key button)
if is_talk_trigger_active(): if is_talk_trigger_active():
@@ -497,7 +528,7 @@ def main():
resp_text = "" resp_text = ""
if content_len > 0: if content_len > 0:
try: try:
resp_text = socket_read_exactly(s, content_len).decode('utf-8', errors='ignore') resp_text = socket_read_exactly(s, content_len).decode('utf-8', 'ignore')
except: except:
pass pass
print("Non-WAV response received:", resp_text) print("Non-WAV response received:", resp_text)
@@ -511,7 +542,7 @@ def main():
resp_text = "Unknown error" resp_text = "Unknown error"
if content_len > 0: if content_len > 0:
try: try:
resp_text = socket_read_exactly(s, content_len).decode('utf-8', errors='ignore') resp_text = socket_read_exactly(s, content_len).decode('utf-8', 'ignore')
except: except:
pass pass
print("Error response:", resp_text) print("Error response:", resp_text)
+221 -21
View File
@@ -96,11 +96,21 @@ class MCPServer:
try: try:
# Handle incoming connection # Handle incoming connection
client.settimeout(2.0) client.settimeout(2.0)
req = client.recv(2048).decode('utf-8') req_bytes = client.recv(2048)
if not req_bytes:
client.close()
return
# Simple HTTP parser header_end = req_bytes.find(b'\r\n\r\n')
lines = req.split('\r\n') if header_end == -1:
if len(lines) == 0: header_end = len(req_bytes)
body_start_idx = len(req_bytes)
else:
body_start_idx = header_end + 4
header_text = req_bytes[:header_end].decode('utf-8', 'ignore')
lines = header_text.split('\r\n')
if len(lines) == 0 or not lines[0]:
client.close() client.close()
return return
@@ -120,17 +130,15 @@ class MCPServer:
content_length = int(line.split(':')[1].strip()) content_length = int(line.split(':')[1].strip())
break break
# Locate start of JSON body body_bytes = bytearray(req_bytes[body_start_idx:])
body = "" while len(body_bytes) < content_length:
if '\r\n\r\n' in req: chunk = client.recv(min(1024, content_length - len(body_bytes)))
body = req.split('\r\n\r\n', 1)[1] if not chunk:
break
body_bytes.extend(chunk)
# Read remaining body if not fully received body_text = body_bytes.decode('utf-8', 'ignore')
while len(body) < content_length: rpc_req = json.loads(body_text)
body += client.recv(1024).decode('utf-8')
# Parse JSON-RPC 2.0 Request
rpc_req = json.loads(body)
rpc_resp = self._handle_rpc(rpc_req) rpc_resp = self._handle_rpc(rpc_req)
resp_body = json.dumps(rpc_resp) resp_body = json.dumps(rpc_resp)
@@ -140,6 +148,56 @@ class MCPServer:
resp += "Connection: close\r\n\r\n" resp += "Connection: close\r\n\r\n"
resp += resp_body resp += resp_body
data_to_send = resp.encode('utf-8')
total_sent = 0
while total_sent < len(data_to_send):
sent = client.write(data_to_send[total_sent:])
if sent is None or sent == 0:
time.sleep_ms(10)
continue
total_sent += sent
elif method == 'POST' and path.startswith('/api/screen/raw'):
# Read content length
content_length = 0
for line in lines:
if line.lower().startswith('content-length:'):
content_length = int(line.split(':')[1].strip())
break
body_bytes = bytearray(req_bytes[body_start_idx:])
while len(body_bytes) < content_length:
chunk = client.recv(min(1024, content_length - len(body_bytes)))
if not chunk:
break
body_bytes.extend(chunk)
# Parse query parameters from path (e.g. /api/screen/raw?x=0&y=0&w=320&h=240)
width = getattr(self.display, 'width', 320)
height = getattr(self.display, 'height', 240)
x, y, w, h = 0, 0, width, height
if '?' in path:
q_str = path.split('?', 1)[1]
for param in q_str.split('&'):
if '=' in param:
k, v = param.split('=', 1)
if k == 'x': x = int(v)
elif k == 'y': y = int(v)
elif k == 'w': w = int(v)
elif k == 'h': h = int(v)
# Draw directly using display raw RGB565 method
self.override_active = True
self.display.draw_rgb565(x, y, w, h, body_bytes)
resp_body = "OK"
resp = "HTTP/1.1 200 OK\r\n"
resp += "Content-Type: text/plain\r\n"
resp += f"Content-Length: {len(resp_body)}\r\n"
resp += "Connection: close\r\n\r\n"
resp += resp_body
data_to_send = resp.encode('utf-8') data_to_send = resp.encode('utf-8')
total_sent = 0 total_sent = 0
while total_sent < len(data_to_send): while total_sent < len(data_to_send):
@@ -151,7 +209,7 @@ class MCPServer:
else: else:
# Return 404 # Return 404
resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
client.send(resp.encode('utf-8')) client.write(resp.encode('utf-8'))
except Exception as e: except Exception as e:
print("Error handling client:", e) print("Error handling client:", e)
finally: finally:
@@ -162,6 +220,10 @@ class MCPServer:
method = req.get('method') method = req.get('method')
params = req.get('params', {}) params = req.get('params', {})
# Get active display dimensions dynamically to report correct screen resolution
width = getattr(self.display, 'width', 320)
height = getattr(self.display, 'height', 240)
if method == 'tools/list': if method == 'tools/list':
return { return {
"jsonrpc": "2.0", "jsonrpc": "2.0",
@@ -169,7 +231,7 @@ class MCPServer:
"tools": [ "tools": [
{ {
"name": "clear_screen", "name": "clear_screen",
"description": "Clear the 480x320 screen to white (0) or black (1).", "description": f"Clear the {width}x{height} screen to white (0) or black (1).",
"inputSchema": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
@@ -180,13 +242,13 @@ class MCPServer:
}, },
{ {
"name": "draw_text", "name": "draw_text",
"description": "Draw text on the screen at specified (x,y) coordinates.", "description": f"Draw text on the screen at specified (x,y) coordinates. Screen resolution is {width}x{height}.",
"inputSchema": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
"text": {"type": "string", "description": "The message to display"}, "text": {"type": "string", "description": "The message to display"},
"x": {"type": "integer", "description": "X coordinate (0-470)"}, "x": {"type": "integer", "description": f"X coordinate (0-{width-1})"},
"y": {"type": "integer", "description": "Y coordinate (0-310)"}, "y": {"type": "integer", "description": f"Y coordinate (0-{height-1})"},
"size": {"type": "integer", "enum": [1, 2], "description": "Text scale (1=normal, 2=large)"} "size": {"type": "integer", "enum": [1, 2], "description": "Text scale (1=normal, 2=large)"}
}, },
"required": ["text", "x", "y"] "required": ["text", "x", "y"]
@@ -233,12 +295,12 @@ class MCPServer:
}, },
{ {
"name": "get_screenshot", "name": "get_screenshot",
"description": "Capture the current reflective LCD screen rendering as a PNG image.", "description": "Capture the current screen rendering as a PNG image.",
"inputSchema": {"type": "object", "properties": {}} "inputSchema": {"type": "object", "properties": {}}
}, },
{ {
"name": "draw_image", "name": "draw_image",
"description": "Draw an image (PNG, JPEG, GIF, BMP, etc.) on the screen. The image will be converted to 1-bit monochrome and fit to display boundaries.", "description": f"Draw an image (PNG, JPEG, GIF, BMP, etc.) on the screen. The image will be converted to 1-bit monochrome and fit to display boundaries (max {width}x{height}).",
"inputSchema": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
@@ -250,6 +312,44 @@ class MCPServer:
"required": ["image_base64"] "required": ["image_base64"]
} }
}, },
{
"name": "draw_color_bmp",
"description": f"Draw a color BMP image on the {width}x{height} color screen at specified (x,y) coordinates.",
"inputSchema": {
"type": "object",
"properties": {
"bmp_base64": {"type": "string", "description": "Base64 encoded BMP image file (uncompressed 24-bit or 32-bit format)"},
"x": {"type": "integer", "description": "X coordinate to place the image (default 0)", "default": 0},
"y": {"type": "integer", "description": "Y coordinate to place the image (default 0)", "default": 0}
},
"required": ["bmp_base64"]
}
},
{
"name": "get_capabilities",
"description": "Get screen capabilities (resolution, color support, and supported formats).",
"inputSchema": {"type": "object", "properties": {}}
},
{
"name": "sync_time",
"description": "Synchronize the hardware and system clock with an internet NTP server.",
"inputSchema": {"type": "object", "properties": {}}
},
{
"name": "draw_raw_rgb565",
"description": f"Draw raw RGB565 pixel data on the {width}x{height} screen at specified (x,y) coordinates with width (w) and height (h).",
"inputSchema": {
"type": "object",
"properties": {
"rgb565_base64": {"type": "string", "description": "Base64 encoded raw RGB565 pixel data (Big-Endian, 2 bytes per pixel)"},
"x": {"type": "integer", "description": "X coordinate to place the image"},
"y": {"type": "integer", "description": "Y coordinate to place the image"},
"w": {"type": "integer", "description": "Width of the raw pixel block"},
"h": {"type": "integer", "description": "Height of the raw pixel block"}
},
"required": ["rgb565_base64", "x", "y", "w", "h"]
}
},
{ {
"name": "write_file", "name": "write_file",
"description": "Write a file (e.g., python script or config) to the board's flash storage.", "description": "Write a file (e.g., python script or config) to the board's flash storage.",
@@ -507,6 +607,106 @@ class MCPServer:
pass pass
return "Image displayed successfully." return "Image displayed successfully."
elif name == "draw_color_bmp":
self.override_active = True
bmp_b64 = args.get("bmp_base64")
x = int(args.get("x", 0))
y = int(args.get("y", 0))
if not bmp_b64:
raise ValueError("Missing bmp_base64 parameter")
import binascii
try:
bmp_bytes = binascii.a2b_base64(bmp_b64)
except Exception as e:
raise ValueError(f"Failed to decode base64 BMP: {e}")
filename = 'temp_recv.bmp'
with open(filename, 'wb') as f:
f.write(bmp_bytes)
try:
success = self.display.draw_bmp(filename, x, y)
if not success:
raise RuntimeError("Failed to parse or draw BMP image on screen.")
finally:
import os
try:
os.remove(filename)
except:
pass
return "Color BMP image displayed successfully."
elif name == "get_capabilities":
is_color = getattr(self.display, '__class__', None) is not None and self.display.__class__.__name__ != 'RLCD'
formats = ["rgb565_base64", "bmp_base64", "pbm_base64"] if is_color else ["pbm_base64", "bmp_base64", "rgb565_base64"]
return json.dumps({
"color": is_color,
"width": width,
"height": height,
"formats": formats
})
elif name == "sync_time":
import ntptime
import wifi_config
tz_offset = getattr(wifi_config, 'TZ_OFFSET', 0)
success = False
t = None
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])
if self.rtc:
self.rtc.set_datetime(dt)
self.rtc.sync_to_system()
success = True
break
except Exception as e:
time.sleep_ms(200)
if success and t is not None:
t_str = f"{t[0]:04d}-{t[1]:02d}-{t[2]:02d} {t[3]:02d}:{t[4]:02d}:{t[5]:02d}"
return f"Successfully synchronized board clock with NTP server. Local time: {t_str}"
else:
raise RuntimeError("Failed to sync clock with NTP server.")
elif name == "draw_raw_rgb565":
self.override_active = True
rgb_b64 = args.get("rgb565_base64")
x = int(args.get("x", 0))
y = int(args.get("y", 0))
w = int(args.get("w", 0))
h = int(args.get("h", 0))
if not rgb_b64:
raise ValueError("Missing rgb565_base64 parameter")
import binascii
try:
rgb_bytes = binascii.a2b_base64(rgb_b64)
except Exception as e:
raise ValueError(f"Failed to decode base64 RGB565: {e}")
if len(rgb_bytes) < w * h * 2:
raise ValueError(f"RGB565 data size too small (expected {w * h * 2} bytes, got {len(rgb_bytes)} bytes)")
try:
self.display.draw_rgb565(x, y, w, h, rgb_bytes)
except Exception as e:
raise RuntimeError(f"Failed to draw RGB565: {e}")
return "Raw RGB565 data displayed successfully."
elif name == "write_file": elif name == "write_file":
path = str(args.get("path")) path = str(args.get("path"))
content = str(args.get("content")) content = str(args.get("content"))
+2
View File
@@ -62,6 +62,7 @@ def main():
files_to_upload = [ files_to_upload = [
("lib/board_config.py", "lib/board_config.py"), ("lib/board_config.py", "lib/board_config.py"),
("lib/st7796.py", "lib/st7796.py"), ("lib/st7796.py", "lib/st7796.py"),
("lib/ili9341.py", "lib/ili9341.py"),
("lib/ft6336u.py", "lib/ft6336u.py"), ("lib/ft6336u.py", "lib/ft6336u.py"),
("lib/rlcd.py", "lib/rlcd.py"), ("lib/rlcd.py", "lib/rlcd.py"),
("lib/battery_util.py", "lib/battery_util.py"), ("lib/battery_util.py", "lib/battery_util.py"),
@@ -72,6 +73,7 @@ def main():
("demo_audio_loopback.py", "demo_audio_loopback.py"), ("demo_audio_loopback.py", "demo_audio_loopback.py"),
("video_stream.py", "video_stream.py"), ("video_stream.py", "video_stream.py"),
("mcp_server.py", "mcp_server.py"), ("mcp_server.py", "mcp_server.py"),
("wifi_config.py", "wifi_config.py"),
("boot.py", "boot.py"), ("boot.py", "boot.py"),
("main.py", "main.py") ("main.py", "main.py")
] ]
+4
View File
@@ -3,3 +3,7 @@
WIFI_SSID = "FamReynaMesh" WIFI_SSID = "FamReynaMesh"
WIFI_PASS = "Gloria2020" WIFI_PASS = "Gloria2020"
# Timezone offset in hours relative to UTC (e.g. -4 for EDT, -5 for EST)
TZ_OFFSET = -4