Adolfo Reyna
685 lines
31 KiB
Python
685 lines
31 KiB
Python
import socket
|
|
import json
|
|
import time
|
|
|
|
class MCPServer:
|
|
"""A lightweight JSON-RPC HTTP server implementing Model Context Protocol (MCP) endpoints."""
|
|
|
|
def __init__(self, display, led, battery, sensor, rtc, ble, vstream=None, touch=None):
|
|
self.display = display
|
|
self.led = led
|
|
self.battery = battery
|
|
self.sensor = sensor
|
|
self.rtc = rtc
|
|
self.ble = ble
|
|
self.vstream = vstream
|
|
self.touch = touch
|
|
|
|
self.sock = None
|
|
self.active_led_mode = "off" # static, breath, rainbow, off
|
|
self.override_active = False
|
|
|
|
def start(self, port=80):
|
|
"""Starts the TCP server non-blockingly."""
|
|
self.port = port
|
|
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.sock.bind(('', port))
|
|
self.sock.listen(3)
|
|
# Set socket to non-blocking so the main loop can run animations concurrently
|
|
self.sock.setblocking(False)
|
|
print(f"MCP server listening on port {port}...")
|
|
|
|
# Setup UDP socket for auto-discovery on port 5000
|
|
try:
|
|
self.udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.udp_sock.bind(('', 5000))
|
|
self.udp_sock.setblocking(False)
|
|
print("UDP Discovery responder listening on port 5000...")
|
|
except Exception as ue:
|
|
print("Failed to bind UDP Discovery socket: {}".format(ue))
|
|
self.udp_sock = None
|
|
|
|
def restart(self):
|
|
"""Re-initializes the listening sockets after a fatal error."""
|
|
print("Restarting MCP Server sockets...")
|
|
try:
|
|
if self.sock:
|
|
self.sock.close()
|
|
except:
|
|
pass
|
|
self.sock = None
|
|
try:
|
|
if hasattr(self, 'udp_sock') and self.udp_sock:
|
|
self.udp_sock.close()
|
|
except:
|
|
pass
|
|
self.udp_sock = None
|
|
|
|
# Add a short delay to allow socket reuse to settle
|
|
time.sleep_ms(100)
|
|
port = getattr(self, 'port', 80)
|
|
self.start(port)
|
|
|
|
def update(self):
|
|
"""Check for incoming HTTP requests and handle them non-blockingly."""
|
|
# 1. Handle UDP Discovery queries
|
|
if hasattr(self, 'udp_sock') and self.udp_sock is not None:
|
|
try:
|
|
data, addr = self.udp_sock.recvfrom(128)
|
|
if data == b"DISCOVER_SCREEN":
|
|
self.udp_sock.sendto(b"SCREEN_IP_80", addr)
|
|
except OSError:
|
|
pass
|
|
|
|
# 2. Check for incoming HTTP TCP connections
|
|
if self.sock is None:
|
|
return
|
|
|
|
try:
|
|
client, addr = self.sock.accept()
|
|
except OSError as e:
|
|
import errno
|
|
err = getattr(e, 'errno', None)
|
|
if err is None and e.args:
|
|
err = e.args[0]
|
|
ewouldblock = getattr(errno, 'EWOULDBLOCK', errno.EAGAIN)
|
|
if err in (errno.EAGAIN, ewouldblock) or err is None:
|
|
# No incoming connection
|
|
return
|
|
# Fatal socket error
|
|
print(f"Fatal socket error in MCP Server accept(): {e}. Re-initializing...")
|
|
self.restart()
|
|
return
|
|
|
|
try:
|
|
# Handle incoming connection
|
|
client.settimeout(2.0)
|
|
req = client.recv(2048).decode('utf-8')
|
|
|
|
# Simple HTTP parser
|
|
lines = req.split('\r\n')
|
|
if len(lines) == 0:
|
|
client.close()
|
|
return
|
|
|
|
first_line = lines[0]
|
|
parts = first_line.split(' ')
|
|
if len(parts) < 3:
|
|
client.close()
|
|
return
|
|
|
|
method, path = parts[0], parts[1]
|
|
|
|
if method == 'POST' and path == '/api/mcp':
|
|
# Read content length
|
|
content_length = 0
|
|
for line in lines:
|
|
if line.lower().startswith('content-length:'):
|
|
content_length = int(line.split(':')[1].strip())
|
|
break
|
|
|
|
# Locate start of JSON body
|
|
body = ""
|
|
if '\r\n\r\n' in req:
|
|
body = req.split('\r\n\r\n', 1)[1]
|
|
|
|
# Read remaining body if not fully received
|
|
while len(body) < content_length:
|
|
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)
|
|
|
|
resp_body = json.dumps(rpc_resp)
|
|
resp = "HTTP/1.1 200 OK\r\n"
|
|
resp += "Content-Type: application/json\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')
|
|
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
|
|
else:
|
|
# Return 404
|
|
resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
|
|
client.send(resp.encode('utf-8'))
|
|
except Exception as e:
|
|
print("Error handling client:", e)
|
|
finally:
|
|
client.close()
|
|
|
|
def _handle_rpc(self, req):
|
|
rpc_id = req.get('id')
|
|
method = req.get('method')
|
|
params = req.get('params', {})
|
|
|
|
if method == 'tools/list':
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"result": {
|
|
"tools": [
|
|
{
|
|
"name": "clear_screen",
|
|
"description": "Clear the 480x320 screen to white (0) or black (1).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"color": {"type": "integer", "enum": [0, 1], "description": "0 = White, 1 = Black"}
|
|
},
|
|
"required": ["color"]
|
|
}
|
|
},
|
|
{
|
|
"name": "draw_text",
|
|
"description": "Draw text on the screen at specified (x,y) coordinates.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"text": {"type": "string", "description": "The message to display"},
|
|
"x": {"type": "integer", "description": "X coordinate (0-470)"},
|
|
"y": {"type": "integer", "description": "Y coordinate (0-310)"},
|
|
"size": {"type": "integer", "enum": [1, 2], "description": "Text scale (1=normal, 2=large)"}
|
|
},
|
|
"required": ["text", "x", "y"]
|
|
}
|
|
},
|
|
{
|
|
"name": "set_led",
|
|
"description": "Control the onboard WS2812 NeoPixel RGB LED.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"r": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Red channel (0-255)"},
|
|
"g": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Green channel (0-255)"},
|
|
"b": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Blue channel (0-255)"},
|
|
"mode": {"type": "string", "enum": ["static", "breath", "rainbow", "off"], "description": "LED mode"}
|
|
},
|
|
"required": ["r", "g", "b", "mode"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_battery",
|
|
"description": "Read the current battery voltage and estimated capacity percentage.",
|
|
"inputSchema": {"type": "object", "properties": {}}
|
|
},
|
|
{
|
|
"name": "get_touch",
|
|
"description": "Read the current touch state and coordinates from the capacitive touchscreen.",
|
|
"inputSchema": {"type": "object", "properties": {}}
|
|
},
|
|
{
|
|
"name": "get_sensors",
|
|
"description": "Read onboard SHTC3 temperature and relative humidity.",
|
|
"inputSchema": {"type": "object", "properties": {}}
|
|
},
|
|
{
|
|
"name": "scan_ble",
|
|
"description": "Scan for nearby Bluetooth Low Energy devices / tracking tags.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"duration_ms": {"type": "integer", "description": "Scan duration in milliseconds (default 3000)"}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "get_screenshot",
|
|
"description": "Capture the current reflective LCD screen rendering as a PNG image.",
|
|
"inputSchema": {"type": "object", "properties": {}}
|
|
},
|
|
{
|
|
"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.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"image_base64": {"type": "string", "description": "Base64 encoded string of the source image file"},
|
|
"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},
|
|
"dither": {"type": "boolean", "description": "Whether to use Floyd-Steinberg dithering (default true)", "default": True}
|
|
},
|
|
"required": ["image_base64"]
|
|
}
|
|
},
|
|
{
|
|
"name": "write_file",
|
|
"description": "Write a file (e.g., python script or config) to the board's flash storage.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "The destination file path (e.g. 'my_script.py')"},
|
|
"content": {"type": "string", "description": "The text content of the file"}
|
|
},
|
|
"required": ["path", "content"]
|
|
}
|
|
},
|
|
{
|
|
"name": "read_file",
|
|
"description": "Read a text file from the board's flash storage.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "The file path to read (e.g. 'boot.py')"}
|
|
},
|
|
"required": ["path"]
|
|
}
|
|
},
|
|
{
|
|
"name": "execute_python",
|
|
"description": "Execute arbitrary Python code dynamically on the board. Standard output (prints) and errors will be captured and returned.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {"type": "string", "description": "The Python code to execute"}
|
|
},
|
|
"required": ["code"]
|
|
}
|
|
},
|
|
{
|
|
"name": "play_tone",
|
|
"description": "Play a pure sine wave tone/beep on the board speaker.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"frequency": {"type": "integer", "description": "Frequency of the tone in Hz (e.g. 440 for A4, default 440)", "default": 440},
|
|
"duration_ms": {"type": "integer", "description": "Duration of the tone in milliseconds (default 1000)", "default": 1000},
|
|
"volume": {"type": "integer", "description": "Volume of the tone from 0 to 100 (default 50)", "default": 50}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "play_audio",
|
|
"description": "Play a standard WAV audio file on the speaker.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"filename": {"type": "string", "description": "WAV file path to play (e.g. 'response.wav')"},
|
|
"volume": {"type": "integer", "description": "Volume from 0 to 100 (default 50)", "default": 50}
|
|
},
|
|
"required": ["filename"]
|
|
}
|
|
},
|
|
{
|
|
"name": "record_voice",
|
|
"description": "Record voice command from the microphone to a raw stereo PCM file.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"duration_sec": {"type": "integer", "description": "Duration in seconds (default 4)", "default": 4},
|
|
"filename": {"type": "string", "description": "Destination file path (e.g. 'recording.pcm')", "default": "recording.pcm"}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "play_audio_base64",
|
|
"description": "Decode a base64-encoded WAV file and play it directly on the speaker.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"wav_base64": {"type": "string", "description": "Base64 encoded string of the WAV audio file"},
|
|
"volume": {"type": "integer", "description": "Volume from 0 to 100 (default 50)", "default": 50}
|
|
},
|
|
"required": ["wav_base64"]
|
|
}
|
|
},
|
|
{
|
|
"name": "download_file",
|
|
"description": "Download a file from a URL over Wi-Fi directly to the board storage or microSD card.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"url": {"type": "string", "description": "The URL of the file to download"},
|
|
"filename": {"type": "string", "description": "The destination filename (e.g. 'podcast1.wav')"},
|
|
"use_sd": {"type": "boolean", "description": "Save to microSD card if true, or local flash if false (default true)", "default": True}
|
|
},
|
|
"required": ["url", "filename"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_video_streaming_instructions",
|
|
"description": "Get detailed instructions and sample Python code to stream video directly into the RLCD screen using TCP or UDP.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"protocol": {"type": "string", "enum": ["tcp", "udp", "both"], "description": "The streaming protocol to query (default 'both')"}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "get_stream_stats",
|
|
"description": "Get real-time device-side performance and frame-rate statistics for the TCP/UDP video stream.",
|
|
"inputSchema": {"type": "object", "properties": {}}
|
|
}
|
|
]
|
|
},
|
|
"id": rpc_id
|
|
}
|
|
|
|
elif method == 'tools/call':
|
|
tool_name = params.get('name')
|
|
args = params.get('arguments', {})
|
|
|
|
try:
|
|
content = self._call_tool(tool_name, args)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"result": {
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": content
|
|
}
|
|
]
|
|
},
|
|
"id": rpc_id
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"error": {
|
|
"code": -32000,
|
|
"message": str(e)
|
|
},
|
|
"id": rpc_id
|
|
}
|
|
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"error": {
|
|
"code": -32601,
|
|
"message": f"Method {method} not found"
|
|
},
|
|
"id": rpc_id
|
|
}
|
|
|
|
def _call_tool(self, name, args):
|
|
"""Executes hardware actions depending on the called tool name."""
|
|
if name == "clear_screen":
|
|
self.override_active = True
|
|
color = int(args.get("color", 0))
|
|
self.display.clear(color)
|
|
self.display.show()
|
|
return "Screen cleared."
|
|
|
|
elif name == "draw_text":
|
|
self.override_active = True
|
|
text = str(args.get("text", ""))
|
|
x = int(args.get("x", 10))
|
|
y = int(args.get("y", 10))
|
|
size = int(args.get("size", 1))
|
|
|
|
if size == 2:
|
|
self.display.text_large(text, x, y, scale=2, c=1)
|
|
else:
|
|
self.display.text(text, x, y, 1)
|
|
self.display.show()
|
|
return f"Successfully drew text '{text}' at ({x}, {y}) with size {size}."
|
|
|
|
elif name == "set_led":
|
|
r = int(args.get("r", 0))
|
|
g = int(args.get("g", 0))
|
|
b = int(args.get("b", 0))
|
|
mode = str(args.get("mode", "static"))
|
|
|
|
self.active_led_mode = mode
|
|
if mode == "static":
|
|
self.led.set_color(r, g, b)
|
|
elif mode == "off":
|
|
self.led.off()
|
|
else:
|
|
# Store base color for animations
|
|
self.led.base_color = (r, g, b)
|
|
return f"LED set to mode '{mode}' with base color ({r}, {g}, {b})."
|
|
|
|
elif name == "get_battery":
|
|
v = self.battery.read_voltage()
|
|
p = self.battery.read_percentage()
|
|
return json.dumps({"voltage_v": v, "percentage_pct": p})
|
|
|
|
elif name == "get_sensors":
|
|
t, h = self.sensor.read_sensor()
|
|
return json.dumps({"temperature_c": t, "humidity_pct": h})
|
|
|
|
elif name == "get_touch":
|
|
if self.touch is None:
|
|
return json.dumps({"error": "Touchscreen not configured on this device."})
|
|
is_t = self.touch.is_touched()
|
|
x, y = None, None
|
|
if is_t:
|
|
pt = self.touch.read_touch()
|
|
if pt:
|
|
x, y = pt
|
|
return json.dumps({"is_touched": is_t, "x": x, "y": y})
|
|
|
|
elif name == "scan_ble":
|
|
dur = int(args.get("duration_ms", 3000))
|
|
# Scan returns dictionary: {mac: {rssi: rssi, name: name}}
|
|
devices = self.ble.scan(dur)
|
|
return json.dumps(devices)
|
|
|
|
elif name == "get_screenshot":
|
|
import binascii
|
|
filename = 'screenshot_mcp.pbm'
|
|
try:
|
|
self.display.save_screenshot(filename)
|
|
with open(filename, 'rb') as f:
|
|
data = f.read()
|
|
# Encode to base64
|
|
b64 = binascii.b2a_base64(data).decode('utf-8').strip()
|
|
return f"__PBM_BASE64__:{b64}"
|
|
except Exception as e:
|
|
raise RuntimeError(f"Screenshot capture failed: {e}")
|
|
|
|
elif name == "draw_image":
|
|
self.override_active = True
|
|
|
|
pbm_b64 = args.get("pbm_base64")
|
|
x = int(args.get("x", 0))
|
|
y = int(args.get("y", 0))
|
|
|
|
if not pbm_b64:
|
|
raise ValueError("Missing pbm_base64 parameter (preprocessed by bridge)")
|
|
|
|
import binascii
|
|
pbm_bytes = binascii.a2b_base64(pbm_b64)
|
|
|
|
filename = 'temp_recv.pbm'
|
|
with open(filename, 'wb') as f:
|
|
f.write(pbm_bytes)
|
|
|
|
try:
|
|
self.display.draw_pbm(filename, x, y, scale=1)
|
|
self.display.show()
|
|
finally:
|
|
import os
|
|
try:
|
|
os.remove(filename)
|
|
except:
|
|
pass
|
|
return "Image displayed successfully."
|
|
|
|
elif name == "write_file":
|
|
path = str(args.get("path"))
|
|
content = str(args.get("content"))
|
|
with open(path, 'w') as f:
|
|
f.write(content)
|
|
return f"Successfully wrote {len(content)} characters to '{path}'."
|
|
|
|
elif name == "read_file":
|
|
path = str(args.get("path"))
|
|
with open(path, 'r') as f:
|
|
content = f.read()
|
|
return content
|
|
|
|
elif name == "execute_python":
|
|
code = str(args.get("code"))
|
|
|
|
import builtins
|
|
import sys
|
|
import io
|
|
|
|
output_buffer = []
|
|
old_print = builtins.print
|
|
|
|
def custom_print(*args, **kwargs):
|
|
sep = kwargs.get('sep', ' ')
|
|
end = kwargs.get('end', '\n')
|
|
str_args = [str(arg) for arg in args]
|
|
msg = sep.join(str_args) + end
|
|
output_buffer.append(msg)
|
|
|
|
builtins.print = custom_print
|
|
|
|
error = None
|
|
try:
|
|
exec(code, globals())
|
|
except Exception as e:
|
|
tb_file = io.StringIO()
|
|
sys.print_exception(e, tb_file)
|
|
error_msg = tb_file.getvalue()
|
|
output_buffer.append(f"\nExecution Failed:\n{error_msg}")
|
|
error = e
|
|
finally:
|
|
builtins.print = old_print
|
|
|
|
output = "".join(output_buffer)
|
|
if error:
|
|
return output
|
|
return f"Execution Succeeded. Console output:\n{output}"
|
|
|
|
elif name == "play_tone":
|
|
freq = int(args.get("frequency", 440))
|
|
duration = int(args.get("duration_ms", 1000))
|
|
vol = int(args.get("volume", 50))
|
|
|
|
# Limit duration to prevent blocking the main loop for too long
|
|
duration = max(50, min(5000, duration))
|
|
freq = max(50, min(10000, freq))
|
|
vol = max(0, min(100, vol))
|
|
|
|
from audio_util import play_tone
|
|
play_tone(frequency=freq, duration_ms=duration, volume=vol)
|
|
return f"Played tone of {freq}Hz for {duration}ms at volume {vol}."
|
|
|
|
elif name == "play_audio":
|
|
filename = str(args.get("filename"))
|
|
vol = int(args.get("volume", 50))
|
|
vol = max(0, min(100, vol))
|
|
|
|
from audio_util import play_wav
|
|
success = play_wav(filename, volume=vol)
|
|
if success:
|
|
return f"Successfully played audio file '{filename}'."
|
|
else:
|
|
raise RuntimeError(f"Failed to play audio file '{filename}'. Check format (16kHz 16-bit PCM WAV recommended).")
|
|
|
|
elif name == "record_voice":
|
|
duration = int(args.get("duration_sec", 4))
|
|
filename = str(args.get("filename", "recording.pcm"))
|
|
# Clamp duration to a reasonable range
|
|
duration = max(1, min(15, duration))
|
|
|
|
from audio_util import record_audio
|
|
success = record_audio(duration_seconds=duration, filename=filename)
|
|
if success:
|
|
return f"Successfully recorded {duration} seconds of audio to '{filename}'."
|
|
else:
|
|
raise RuntimeError("Audio recording failed.")
|
|
|
|
elif name == "play_audio_base64":
|
|
b64_data = args.get("wav_base64")
|
|
vol = int(args.get("volume", 50))
|
|
vol = max(0, min(100, vol))
|
|
|
|
if not b64_data:
|
|
raise ValueError("Missing wav_base64 parameter.")
|
|
|
|
import binascii
|
|
try:
|
|
audio_bytes = binascii.a2b_base64(b64_data)
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to decode base64 audio: {e}")
|
|
|
|
filename = "temp_play.wav"
|
|
with open(filename, "wb") as f:
|
|
f.write(audio_bytes)
|
|
|
|
try:
|
|
from audio_util import play_wav
|
|
success = play_wav(filename, volume=vol)
|
|
finally:
|
|
import os
|
|
try:
|
|
os.remove(filename)
|
|
except:
|
|
pass
|
|
|
|
if success:
|
|
return "Successfully played base64 audio stream."
|
|
else:
|
|
raise RuntimeError("Failed to play decoded audio stream. Check format (16kHz 16-bit PCM WAV recommended).")
|
|
|
|
elif name == "download_file":
|
|
url = str(args.get("url"))
|
|
filename = str(args.get("filename"))
|
|
use_sd = bool(args.get("use_sd", True))
|
|
|
|
from download_util import download_file
|
|
saved_path = download_file(url, filename, use_sd=use_sd)
|
|
if saved_path:
|
|
return f"Successfully downloaded file to '{saved_path}'."
|
|
else:
|
|
raise RuntimeError(f"Failed to download file from '{url}'.")
|
|
|
|
elif name == "get_video_streaming_instructions":
|
|
protocol = str(args.get("protocol", "both")).lower()
|
|
|
|
instructions = [
|
|
"### RP2350 TFT Video Streaming Instructions",
|
|
"Dimensions: 400x300 (decoded and centered automatically on the 480x320 screen), 1-bit monochrome (Floyd-Steinberg dithered).",
|
|
"Frame Buffer Size: 15,000 bytes. The host must convert and map standard pixels into the specific RLCD hardware buffer layout before sending.",
|
|
"Mapping logic (Python):",
|
|
" def map_to_rlcd(pil_img):",
|
|
" img_1bit = pil_img.convert('1', dither=1)",
|
|
" px = img_1bit.load()",
|
|
" buf = bytearray(15000)",
|
|
" for y in range(300):",
|
|
" for x in range(400):",
|
|
" if px[x, y]:",
|
|
" inv_y = 299 - y",
|
|
" bx = x // 2",
|
|
" by = inv_y // 4",
|
|
" idx = bx * 75 + by",
|
|
" lx, ly = x % 2, inv_y % 4",
|
|
" bit = 7 - (ly * 2 + lx)",
|
|
" buf[idx] |= (1 << bit)",
|
|
" return buf"
|
|
]
|
|
|
|
if protocol in ("tcp", "both"):
|
|
instructions.append("\n**TCP Streaming (Port 8081):**")
|
|
instructions.append("Open a TCP connection to the device's IP on port 8081 and send consecutive 15,000-byte frame blocks.")
|
|
|
|
if protocol in ("udp", "both"):
|
|
instructions.append("\n**UDP Streaming / Broadcast (Port 8082):**")
|
|
instructions.append("Split the 15,000-byte frame into 15 chunks of 1,000 bytes each. Send each chunk as a 1002-byte packet: byte 0 = frame_id (0-255), byte 1 = chunk_idx (0-14), bytes 2..1001 = chunk payload. Send to port 8082 (unicast or broadcast).")
|
|
|
|
return "\n".join(instructions)
|
|
|
|
elif name == "get_stream_stats":
|
|
if self.vstream is None:
|
|
return json.dumps({"error": "Video stream server not initialized."})
|
|
return json.dumps(self.vstream.get_stats())
|
|
|
|
else:
|
|
raise ValueError(f"Unknown tool: {name}")
|