diff --git a/battery_util.py b/battery_util.py index 91ca8d7..79e1f4e 100644 --- a/battery_util.py +++ b/battery_util.py @@ -4,9 +4,10 @@ class BatteryMonitor: """Utility class for monitoring battery voltage and capacity on ESP32-S3-RLCD-4.2.""" def __init__(self, pin_num=4): - # Initialize ADC on GPIO 4 with 11dB attenuation + import sys self.adc = ADC(Pin(pin_num)) - self.adc.atten(ADC.ATTN_11DB) + if sys.platform == 'esp32': + self.adc.atten(ADC.ATTN_11DB) def read_voltage(self): """Reads the battery voltage in Volts using internal calibration. diff --git a/ble_util.py b/ble_util.py index 0ee0aab..af826e5 100644 --- a/ble_util.py +++ b/ble_util.py @@ -1,25 +1,40 @@ -import bluetooth +try: + import bluetooth + has_ble = True +except ImportError: + has_ble = False + import struct import time -# BLE UUID definitions (Nordic UART Service) -_UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E") -_UART_TX = ( - bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"), - bluetooth.FLAG_NOTIFY, -) -_UART_RX = ( - bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"), - bluetooth.FLAG_WRITE | bluetooth.FLAG_WRITE_NO_RESPONSE, -) -_UART_SERVICE = (_UART_UUID, (_UART_TX, _UART_RX)) +if not has_ble: + class BLEUART: + def __init__(self, ble=None, name="ESP32-S3-RLCD"): + print("BLE not supported on this platform.") + def on_rx(self, callback): pass + def write(self, data): return False + def is_connected(self): return False + def scan(self, duration_ms=3000): return {} + def close(self): pass +else: + # BLE UUID definitions (Nordic UART Service) + _UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E") + _UART_TX = ( + bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"), + bluetooth.FLAG_NOTIFY, + ) + _UART_RX = ( + bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"), + bluetooth.FLAG_WRITE | bluetooth.FLAG_WRITE_NO_RESPONSE, + ) + _UART_SERVICE = (_UART_UUID, (_UART_TX, _UART_RX)) -# BLE IRQ constants -_IRQ_CENTRAL_CONNECT = 1 -_IRQ_CENTRAL_DISCONNECT = 2 -_IRQ_GATTS_WRITE = 3 + # BLE IRQ constants + _IRQ_CENTRAL_CONNECT = 1 + _IRQ_CENTRAL_DISCONNECT = 2 + _IRQ_GATTS_WRITE = 3 -class BLEUART: +class _ActiveBLEUART: """A helper class to manage BLE UART (Serial Over BLE) for raw text data exchange.""" def __init__(self, ble=None, name="ESP32-S3-RLCD"): @@ -174,3 +189,6 @@ class BLEUART: self._ble.gap_disconnect(conn_handle) self._ble.active(False) print("BLE closed.") + +if has_ble: + BLEUART = _ActiveBLEUART diff --git a/boot.py b/boot.py index 0c89e31..8b5bbfc 100644 --- a/boot.py +++ b/boot.py @@ -1,17 +1,34 @@ # This file is executed on every boot (including wake-boot from deepsleep) -import network import time -import rlcd +import sys + +try: + import network + has_network = True +except ImportError: + has_network = False +if sys.platform == 'rp2': + import st7796 as display_module +else: + import rlcd as display_module from machine import Pin, SPI import wifi_config def connect_wifi(): + if sys.platform == 'rp2': + # Skip display and connection setup on RP2 (no network/Wi-Fi hardware) + # Keep the sleep window to make REPL interruption easy + time.sleep(1.5) + return + # Initialize display to show connection progress display = None try: + # ESP32-S3 configuration spi = SPI(1, baudrate=20000000, polarity=0, phase=0, sck=Pin(11), mosi=Pin(12)) - display = rlcd.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) + display = display_module.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) display.clear(0) + display.text("ESP32-S3-RLCD-4.2", 10, 10, 1) display.line(10, 20, 390, 20, 1) display.text("Connecting to Wi-Fi...", 10, 35, 1) @@ -21,8 +38,10 @@ def connect_wifi(): print("Display init failed in boot.py:", e) # Initialize Wi-Fi Station + if not has_network: + return + try: - import network network.hostname('esp32screen') print("Hostname set to: {}".format(network.hostname())) except Exception as he: diff --git a/fonts/CourierPrime.ttf b/fonts/CourierPrime.ttf new file mode 100644 index 0000000..2a0f8bd Binary files /dev/null and b/fonts/CourierPrime.ttf differ diff --git a/fonts/PatrickHand.ttf b/fonts/PatrickHand.ttf new file mode 100644 index 0000000..fb45ccd Binary files /dev/null and b/fonts/PatrickHand.ttf differ diff --git a/fonts/SpecialElite.ttf b/fonts/SpecialElite.ttf new file mode 100644 index 0000000..c842f35 --- /dev/null +++ b/fonts/SpecialElite.ttf @@ -0,0 +1,1483 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Page not found · GitHub · GitHub + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+ + +
+ Skip to content + + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + + + + +
+ +
+ + + + + + + + +
+ + + + + +
+ + + + + + + + + +
+
+ + + +
+
+ +
+
+ 404 “This is not the web page you are looking for” + + + + + + + + + + + + +
+
+ +
+
+ +
+ + +
+
+
+ Contact Support — + GitHub Status — + @githubstatus +
+
+ +
+ +
+ + + + + + + + + + + + + + + + + + + + + + +
+
+
+ + + diff --git a/ft6336u.py b/ft6336u.py new file mode 100644 index 0000000..9122bfd --- /dev/null +++ b/ft6336u.py @@ -0,0 +1,123 @@ +import time +from machine import Pin, I2C + +class FT6336U: + ADDR = 0x38 + + # Registers + REG_DEV_MODE = 0x00 + REG_TD_STATUS = 0x02 + REG_P1_XH = 0x03 + REG_P1_XL = 0x04 + REG_P1_YH = 0x05 + REG_P1_YL = 0x06 + REG_CTRL = 0x86 + REG_CHIPID = 0xA3 + REG_G_MODE = 0xA4 + + # Modes + CTRL_KEEP_ACTIVE = 0x00 + G_MODE_TRIGGER = 0x01 + + def __init__(self, i2c, rst_pin=28, int_pin=25, width=480, height=320, swap_xy=True, invert_x=True, invert_y=False): + self.i2c = i2c + self.rst = Pin(rst_pin, Pin.OUT) + self.int = Pin(int_pin, Pin.IN, Pin.PULL_UP) + self.width = width + self.height = height + self.swap_xy = swap_xy + self.invert_x = invert_x + self.invert_y = invert_y + self.initialized = False + + self.reset() + self.init_chip() + + def reset(self): + self.rst(0) + time.sleep_ms(10) + self.rst(1) + time.sleep_ms(300) # Wait for chip to wake up + + def read_reg(self, reg, n=1): + try: + return self.i2c.readfrom_mem(self.ADDR, reg, n) + except Exception as e: + print(f"I2C read failed at reg 0x{reg:02X}: {e}") + return None + + def write_reg(self, reg, val): + try: + self.i2c.writeto_mem(self.ADDR, reg, bytearray([val])) + return True + except Exception as e: + print(f"I2C write failed at reg 0x{reg:02X}: {e}") + return False + + def init_chip(self): + # 1. Read Chip ID + chip_id = self.read_reg(self.REG_CHIPID) + if chip_id is None or chip_id[0] != 0x64: + # Retry once + time.sleep_ms(100) + chip_id = self.read_reg(self.REG_CHIPID) + if chip_id is None or chip_id[0] != 0x64: + print(f"FT6336U error: Invalid Chip ID (got {chip_id[0] if chip_id else None}, expected 0x64)") + self.initialized = False + return False + + print(f"FT6336U touch controller detected (Chip ID: 0x{chip_id[0]:02X})") + + # 2. Configure operating mode (0 = Normal Mode) + self.write_reg(self.REG_DEV_MODE, 0x00) + + # 3. Configure CTRL mode (0 = Keep Active) + self.write_reg(self.REG_CTRL, self.CTRL_KEEP_ACTIVE) + + self.initialized = True + return True + + def is_touched(self): + """Returns True if screen is touched (the active-low INT pin is pulled LOW).""" + return self.int() == 0 + + def read_touch(self): + """Reads touch point coordinates. + Returns: + (x, y) coordinates of first touch point, or None if no touch. + """ + if not self.initialized: + return None + + # Read TD_STATUS to check if there is an active touch + td_status = self.read_reg(self.REG_TD_STATUS) + if td_status is None: + return None + + touch_count = td_status[0] & 0x0F + if touch_count == 0: + return None + + # Read P1 coordinates (6 bytes starting at P1_XH) + buf = self.read_reg(self.REG_P1_XH, 6) + if buf is None or len(buf) < 6: + return None + + raw_x = ((buf[0] & 0x0F) << 8) | buf[1] + raw_y = ((buf[2] & 0x0F) << 8) | buf[3] + + # Apply coordinate transformations to map touch coordinate space to screen space + if self.swap_xy: + raw_x, raw_y = raw_y, raw_x + + if self.invert_x: + raw_x = self.width - 1 - raw_x + + if self.invert_y: + raw_y = self.height - 1 - raw_y + + # Clamp coordinates to screen boundaries + x = max(0, min(self.width - 1, raw_x)) + y = max(0, min(self.height - 1, raw_y)) + + return (x, y) diff --git a/main.py b/main.py index b3d502a..1a3fcfc 100644 --- a/main.py +++ b/main.py @@ -1,8 +1,31 @@ import time -import network +import sys import machine from machine import Pin, SPI, I2C -import rlcd + +try: + import network + has_network = True +except ImportError: + has_network = False + +if sys.platform == 'rp2': + import st7796 as display_module +else: + import rlcd as display_module + +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 @@ -12,6 +35,7 @@ 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 # LED mode options for manual cycling led_modes = [ @@ -30,33 +54,83 @@ def main(): print("=== Starting ESP32-S3-RLCD-4.2 Main Boot ===") # 1. Initialize shared buses and peripherals - i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + i2c = None + if sys.platform != 'rp2': + try: + i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + except Exception as e: + print("Failed to initialize I2C0:", e) - spi = SPI(1, baudrate=20000000, polarity=0, phase=0, sck=Pin(11), mosi=Pin(12)) - display = rlcd.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) + if sys.platform == 'rp2': + spi = SPI(1, baudrate=32000000, polarity=0, phase=0, sck=Pin(10), mosi=Pin(11)) + display = display_module.ST7796(spi, cs=Pin(7), dc=Pin(4), rst=Pin(9), bl=Pin(6)) + + # Touch controller + touch = None + try: + touch_i2c = I2C(1, sda=Pin(2), scl=Pin(3), freq=400000) + from ft6336u import FT6336U + touch = FT6336U(touch_i2c, rst_pin=28, int_pin=25) + except Exception as te: + print("Failed to initialize touch:", te) + else: + spi = SPI(1, baudrate=20000000, polarity=0, phase=0, sck=Pin(11), mosi=Pin(12)) + display = display_module.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) + touch = None - # Configure Audio Amp control pin to save power - amp_pin = Pin(46, Pin.OUT, value=0) + if sys.platform != 'rp2': + # Configure Audio Amp control pin to save power + amp_pin = Pin(46, Pin.OUT, value=0) # 2. Initialize utility objects - sensor = SHTC3(i2c) - rtc_chip = PCF85063(i2c) + sensor = None + rtc_chip = None + if i2c is not None: + try: + sensor = SHTC3(i2c) + except Exception as e: + print("Failed to initialize SHTC3:", e) + try: + rtc_chip = PCF85063(i2c) + except Exception as e: + print("Failed to initialize PCF85063:", e) + battery = BatteryMonitor() - led = BoardLED(38) - buttons = BoardButtons() + led = BoardLED() + buttons = None + if sys.platform != 'rp2': + buttons = BoardButtons() ble_uart = BLEUART(name="ESP32-S3-RLCD") # Sync system clock from RTC chip - rtc_chip.sync_to_system() + 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 (boot.py already attempted connection) - wlan = network.WLAN(network.STA_IF) - ip_addr = wlan.ifconfig()[0] if wlan.isconnected() else "Disconnected" - - # 4. Start MCP Server - mcp = MCPServer(display, led, battery, sensor, rtc_chip, ble_uart) - mcp.start(port=80) + # 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) + mcp.start(port=80) + 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] @@ -87,12 +161,14 @@ def main(): mcp.override_active = False force_dashboard_redraw = True - buttons.key.on_click(on_key_click) - buttons.boot.on_click(on_boot_click) + 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 print("ESP32 MCP loop running...") @@ -100,18 +176,30 @@ def main(): while True: now = time.ticks_ms() + # Check touch interaction if available + if touch and touch.is_touched(): + pt = touch.read_touch() + if pt: + tx, ty = pt + print(f"Touch detected at: ({tx}, {ty})") + last_action_str = f"Touch: ({tx}, {ty})" + mcp.override_active = False + force_dashboard_redraw = True + # A. Handle non-blocking MCP client connection updates mcp.update() - - # C. Draw local dashboard (if not overridden by MCP draw text commands) - if not mcp.override_active: + # 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() + 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" @@ -121,45 +209,54 @@ def main(): 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() + 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) - display.text("ESP32-S3-RLCD MCP SERVER", 10, 10, 1) - display.line(10, 20, 390, 20, 1) + title_text = "RP2350-TFT MCP SERVER" if sys.platform == 'rp2' else "ESP32-S3-RLCD MCP SERVER" + line_w = 470 if sys.platform == 'rp2' else 390 + + 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, 390, 95, 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 : ESP32-S3-RLCD", 25, 160, 1) - display.line(10, 185, 390, 185, 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, 390, 255, 1) + display.line(10, 255, line_w, 255, 1) display.text(f"Status: {last_action_str}", 15, 265, 1) display.show() # D. Update NeoPixel animation smoothly (runs every 50ms) - 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() + 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() - time.sleep_ms(50) + # 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() diff --git a/mcp_server.py b/mcp_server.py index 4878c78..c852eee 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -5,13 +5,14 @@ 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): + def __init__(self, display, led, battery, sensor, rtc, ble, vstream=None): self.display = display self.led = led self.battery = battery self.sensor = sensor self.rtc = rtc self.ble = ble + self.vstream = vstream self.sock = None self.active_led_mode = "off" # static, breath, rainbow, off @@ -128,7 +129,7 @@ class MCPServer: "tools": [ { "name": "clear_screen", - "description": "Clear the 400x300 screen to white (0) or black (1).", + "description": "Clear the 480x320 screen to white (0) or black (1).", "inputSchema": { "type": "object", "properties": { @@ -144,8 +145,8 @@ class MCPServer: "type": "object", "properties": { "text": {"type": "string", "description": "The message to display"}, - "x": {"type": "integer", "description": "X coordinate (0-390)"}, - "y": {"type": "integer", "description": "Y coordinate (0-290)"}, + "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"] @@ -297,6 +298,21 @@ class MCPServer: }, "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": {}} } ] }, @@ -568,5 +584,45 @@ class MCPServer: 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}") diff --git a/note.txt b/note.txt new file mode 100644 index 0000000..69866d1 --- /dev/null +++ b/note.txt @@ -0,0 +1,17 @@ +Hey there This is now a note taking app. + +What do you think of this? + +This is going to be pretty cool if you ask me!!! + + +Now the themes are supper cool too!!!! + +What is going to happend if I write a line that is too long would it wrapped yes!!! + +It also scrolls for vertical overflow!! + + + + + diff --git a/note_2026-06-04.txt b/note_2026-06-04.txt new file mode 100644 index 0000000..cbebc7b --- /dev/null +++ b/note_2026-06-04.txt @@ -0,0 +1,5 @@ +Hey there buddy! +I think you are going to like this DEVICE! +I made it for you, so you can write your hearth into it!... +Love you... DAD + diff --git a/notetaker.py b/notetaker.py new file mode 100644 index 0000000..edda74d --- /dev/null +++ b/notetaker.py @@ -0,0 +1,490 @@ +#!/usr/bin/env python3 +import socket +import time +import sys +import os +import argparse +import curses +import datetime +from PIL import Image, ImageDraw, ImageFont + +# Screen Dimensions +WIDTH = 400 +HEIGHT = 300 + +# Margins and Layout +MARGIN_LEFT = 20 +MARGIN_RIGHT = 380 +TOP_LIMIT = 35 +BOTTOM_LIMIT = 265 + +# Precompute destination byte mapping for RLCD hardware buffer +DEST_BYTE_MAP = [] +for dst_idx in range(15000): + byte_x = dst_idx // 75 + block_y = dst_idx % 75 + x_base = 2 * byte_x + y_base = HEIGHT - 1 - 4 * block_y + + bits_map = [] + for local_y in range(4): + y = y_base - local_y + for local_x in range(2): + x = x_base + local_x + src_byte_idx = y * 50 + (x // 8) + src_bit_idx = 7 - (x % 8) + src_mask = 1 << src_bit_idx + dst_mask = 1 << (7 - (local_y * 2 + local_x)) + bits_map.append((src_byte_idx, src_mask, dst_mask)) + DEST_BYTE_MAP.append(tuple(bits_map)) + + +def map_to_rlcd_hw_buffer(img_1bit): + """Highly optimized byte-level 1-bit monochrome image mapping to Waveshare RLCD buffer.""" + img_bytes = img_1bit.tobytes() + hw_buffer = bytearray(15000) + for dst_idx in range(15000): + val = 0 + for src_idx, src_mask, dst_mask in DEST_BYTE_MAP[dst_idx]: + if img_bytes[src_idx] & src_mask: + val |= dst_mask + hw_buffer[dst_idx] = val + return hw_buffer + + +def wrap_text_by_width(text, cursor_pos, font, max_width=360): + """ + Wraps text to fit within max_width pixels. + Returns: + lines: list of strings (the wrapped lines) + line_starts: list of starting indices of each line in the raw text + cursor_line: index of line containing cursor + cursor_col: index of column in that line (in terms of character index in the line) + """ + lines = [] + line_starts = [] + cursor_line = 0 + cursor_col = 0 + + current_char_idx = 0 + paragraphs = text.split('\n') + + # Create a dummy draw context to measure text lengths + dummy_im = Image.new("1", (1, 1)) + dummy_draw = ImageDraw.Draw(dummy_im) + + for p_idx, p in enumerate(paragraphs): + if not p: + if current_char_idx <= cursor_pos <= current_char_idx: + cursor_line = len(lines) + cursor_col = 0 + lines.append("") + line_starts.append(current_char_idx) + current_char_idx += 1 # for the '\n' + continue + + words = p.split(' ') + curr_line = "" + curr_line_start = current_char_idx + + for w in words: + if not curr_line: + test_line = w + else: + test_line = curr_line + " " + w + + w_len = dummy_draw.textlength(test_line, font=font) + if w_len <= max_width: + curr_line = test_line + else: + # Wrap current line + if curr_line_start <= cursor_pos <= curr_line_start + len(curr_line): + cursor_line = len(lines) + cursor_col = cursor_pos - curr_line_start + lines.append(curr_line) + line_starts.append(curr_line_start) + + # Check for long words + w_start = curr_line_start + len(curr_line) + 1 # +1 for space/wrap boundary + while dummy_draw.textlength(w, font=font) > max_width: + part_len = 1 + while part_len <= len(w) and dummy_draw.textlength(w[:part_len], font=font) <= max_width: + part_len += 1 + part = w[:part_len - 1] + if not part: + part = w[0] + part_len = 2 + + if w_start <= cursor_pos <= w_start + len(part): + cursor_line = len(lines) + cursor_col = cursor_pos - w_start + lines.append(part) + line_starts.append(w_start) + w_start += len(part) + w = w[part_len - 1:] + + curr_line = w + curr_line_start = w_start + + if curr_line or p_idx == len(paragraphs) - 1: + if curr_line_start <= cursor_pos <= curr_line_start + len(curr_line): + cursor_line = len(lines) + cursor_col = cursor_pos - curr_line_start + lines.append(curr_line) + line_starts.append(curr_line_start) + current_char_idx = curr_line_start + len(curr_line) + + current_char_idx += 1 # for the '\n' + + return lines, line_starts, cursor_line, cursor_col + + +def render_screen(text, cursor_pos, scroll_top, font, char_h, line_spacing, max_lines, dark_mode, show_cursor, status_msg=""): + # Create 1-bit image (0 = Black background) + if dark_mode: + bg_color = 0 # Black background + text_color = 1 # White text + accent_color = 1 # White elements + rule_color = 1 # White dotted lines + else: + bg_color = 1 # White background + text_color = 0 # Black text + accent_color = 0 # Black elements + rule_color = 0 # Black dotted lines + + img = Image.new("1", (WIDTH, HEIGHT), bg_color) + draw = ImageDraw.Draw(img) + + # Load clean, non-clipping UI font for header/footer + try: + ui_font = ImageFont.load_default(size=12) + except TypeError: + ui_font = ImageFont.load_default() + + # 1. Wrap the text and get the cursor location (max_width = 360) + lines, line_starts, cursor_line, cursor_col = wrap_text_by_width(text, cursor_pos, font, max_width=360) + + # 2. Adjust scrolling + if cursor_line >= scroll_top + max_lines: + scroll_top = cursor_line - max_lines + 1 + elif cursor_line < scroll_top: + scroll_top = cursor_line + + # 3. Draw Header Title Bar + draw.rectangle([0, 0, WIDTH - 1, 28], fill=text_color) + + # Display note title + date + today_str = datetime.date.today().strftime("%a, %b %d, %Y").upper() + header_title = f"KID KEEPER - {today_str}" + draw.text((15, 7), header_title, fill=bg_color, font=ui_font) + + # Header stats (word and character count) + word_count = len(text.split()) + char_count = len(text) + stats_str = f"W:{word_count} C:{char_count}" + stats_bbox = draw.textbbox((0, 0), stats_str, font=ui_font) + stats_w = stats_bbox[2] - stats_bbox[0] + draw.text((WIDTH - stats_w - 15, 7), stats_str, fill=bg_color, font=ui_font) + + # 4. Draw Notebook Ruled Paper Lines (aligned with baseline of the font size) + for i in range(max_lines): + y_line = TOP_LIMIT + line_spacing * i + char_h + 5 + # Prevent drawing paper lines past bottom limit + if y_line < BOTTOM_LIMIT + 5: + for x in range(MARGIN_LEFT, MARGIN_RIGHT, 4): + draw.point((x, y_line), fill=rule_color) + + # 5. Draw Visible Text Lines + visible_lines = lines[scroll_top:scroll_top + max_lines] + for idx, line_text in enumerate(visible_lines): + y_pos = TOP_LIMIT + line_spacing * idx + 2 + draw.text((MARGIN_LEFT, y_pos), line_text, fill=text_color, font=font) + + # 6. Draw Blinking Text Cursor + if show_cursor: + cursor_line_text = lines[cursor_line] + text_before_cursor = cursor_line_text[:cursor_col] + cursor_x = MARGIN_LEFT + draw.textlength(text_before_cursor, font=font) + cursor_y = TOP_LIMIT + (cursor_line - scroll_top) * line_spacing + 2 + if MARGIN_LEFT <= cursor_x <= MARGIN_RIGHT: + # Draw a thick cursor bar (2px width) + draw.rectangle([cursor_x, cursor_y, cursor_x + 1, cursor_y + char_h + 2], fill=text_color) + + # 7. Draw Bottom Status Bar + draw.line((0, BOTTOM_LIMIT + 8, WIDTH - 1, BOTTOM_LIMIT + 8), fill=accent_color) + + help_text = "Ctrl+S: Save Ctrl+T: Theme Ctrl+F: Font Ctrl+X: Exit" + draw.text((15, BOTTOM_LIMIT + 12), help_text, fill=text_color, font=ui_font) + + if status_msg: + msg_bbox = draw.textbbox((0, 0), status_msg, font=ui_font) + msg_w = msg_bbox[2] - msg_bbox[0] + draw.rectangle([WIDTH - msg_w - 20, BOTTOM_LIMIT + 10, WIDTH - 5, HEIGHT - 2], fill=bg_color) + draw.text((WIDTH - msg_w - 15, BOTTOM_LIMIT + 12), status_msg, fill=text_color, font=ui_font) + + return img, scroll_top + + +def load_font(font_path, size): + """Safely loads a TTF font and computes its vertical metric using 'Hyg' ascenders/descenders.""" + try: + if font_path and os.path.exists(font_path): + font = ImageFont.truetype(font_path, size=size) + else: + font = ImageFont.load_default(size=size) + except Exception: + font = ImageFont.load_default() + + # Measure character height using ascenders and descenders + test_img = Image.new("1", (100, 100)) + test_draw = ImageDraw.Draw(test_img) + bbox = test_draw.textbbox((0, 0), "Hyg", font=font) + char_h = bbox[3] - bbox[1] + if char_h <= 0: + char_h = 12 + return font, char_h + + +def run_editor(stdscr, args): + # --- Curses Initialization --- + stdscr.nodelay(True) # Non-blocking input + curses.noecho() # Hide echoing + stdscr.keypad(True) # Handle arrow keys + + # Configure kid-friendly font choices + # Format: (path, size, name) + font_folder = "fonts" + if not os.path.exists(font_folder): + font_folder = os.path.expanduser("~/fonts") + + font_options = [ + (os.path.join(font_folder, "PatrickHand.ttf"), 22, "Handwritten"), + (os.path.join(font_folder, "CourierPrime.ttf"), 20, "Typewriter"), + (None, 14, "Default Monospace") + ] + font_idx = 0 + + # Load first font + f_path, f_size, f_name = font_options[font_idx] + font, char_h = load_font(f_path, f_size) + + # 2. Load note from file (use date-specific note by default) + note_text = "" + if os.path.exists(args.file): + try: + with open(args.file, "r") as f: + note_text = f.read() + except: + pass + + cursor_pos = len(note_text) + scroll_top = 0 + dark_mode = False + + # 3. Connect to the ESP32 TCP Stream Server + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(5.0) + try: + sock.connect((args.ip, args.port)) + except Exception as e: + curses.endwin() + print(f"\nConnection failed to {args.ip}:{args.port} -- {e}") + print("Please verify the ESP32 is powered on and connected to Wi-Fi.") + sys.exit(1) + + sock.setblocking(False) + + # State variables + last_blink_time = time.time() + show_cursor = True + status_msg = f"Font: {f_name}" + status_msg_expiry = time.time() + 2.0 + redraw = True + + while True: + now = time.time() + + # Calculate text layout dynamically based on current font height + line_spacing = char_h + 8 + max_lines_on_screen = (BOTTOM_LIMIT - TOP_LIMIT) // line_spacing + + # Cursor blink check (toggles every 500ms) + if now - last_blink_time >= 0.5: + show_cursor = not show_cursor + last_blink_time = now + redraw = True + + # Expiry check for status message + if status_msg and now > status_msg_expiry: + status_msg = "" + redraw = True + + # Read all pending characters from curses + keys = [] + while True: + try: + k = stdscr.get_wch() + keys.append(k) + except curses.error: + break + + if keys: + redraw = True + show_cursor = True + last_blink_time = now + + for key in keys: + if isinstance(key, str): + if key == '\x18': # Ctrl+X: Save and Exit + status_msg = "Saving..." + img, _ = render_screen(note_text, cursor_pos, scroll_top, font, char_h, line_spacing, max_lines_on_screen, dark_mode, False, status_msg) + try: + sock.sendall(map_to_rlcd_hw_buffer(img)) + except: + pass + try: + with open(args.file, "w") as f: + f.write(note_text) + except: + pass + sock.close() + return + + elif key == '\x13': # Ctrl+S: Save Note + try: + with open(args.file, "w") as f: + f.write(note_text) + status_msg = "Saved!" + except: + status_msg = "Err Save" + status_msg_expiry = now + 2.0 + + elif key == '\x14': # Ctrl+T: Toggle Theme + dark_mode = not dark_mode + status_msg = "Theme Dark" if dark_mode else "Theme Light" + status_msg_expiry = now + 2.0 + + elif key == '\x06': # Ctrl+F: Cycle Fonts + font_idx = (font_idx + 1) % len(font_options) + f_path, f_size, f_name = font_options[font_idx] + font, char_h = load_font(f_path, f_size) + status_msg = f"Font: {f_name}" + status_msg_expiry = now + 2.0 + + # Backspace + elif key in ('\x7f', '\x08'): + if cursor_pos > 0: + note_text = note_text[:cursor_pos - 1] + note_text[cursor_pos:] + cursor_pos -= 1 + + # Enter: Newline + elif key in ('\r', '\n'): + note_text = note_text[:cursor_pos] + "\n" + note_text[cursor_pos:] + cursor_pos += 1 + + # Normal typed characters + elif ord(key) >= 32: + note_text = note_text[:cursor_pos] + key + note_text[cursor_pos:] + cursor_pos += 1 + + elif isinstance(key, int): + lines, line_starts, cursor_line, cursor_col = wrap_text_by_width(note_text, cursor_pos, font, max_width=360) + + if key == curses.KEY_UP: + if cursor_line > 0: + target_line = cursor_line - 1 + target_col = min(cursor_col, len(lines[target_line])) + cursor_pos = line_starts[target_line] + target_col + + elif key == curses.KEY_DOWN: + if cursor_line < len(lines) - 1: + target_line = cursor_line + 1 + target_col = min(cursor_col, len(lines[target_line])) + cursor_pos = line_starts[target_line] + target_col + + elif key == curses.KEY_RIGHT: + if cursor_pos < len(note_text): + cursor_pos += 1 + + elif key == curses.KEY_LEFT: + if cursor_pos > 0: + cursor_pos -= 1 + + elif key == curses.KEY_BACKSPACE: + if cursor_pos > 0: + note_text = note_text[:cursor_pos - 1] + note_text[cursor_pos:] + cursor_pos -= 1 + + if redraw: + redraw = False + # Render to RLCD and stream + img, scroll_top = render_screen(note_text, cursor_pos, scroll_top, font, char_h, line_spacing, max_lines_on_screen, dark_mode, show_cursor, status_msg) + try: + raw_bytes = map_to_rlcd_hw_buffer(img) + sock.sendall(raw_bytes) + except BlockingIOError: + pass + except: + break + + # Draw Console UI on terminal screen + stdscr.erase() + h, w = stdscr.getmaxyx() + + # Header + stdscr.attron(curses.A_REVERSE) + today_header = datetime.date.today().strftime("%A, %B %d, %Y") + header_str = f" NOTE KEEPER | {today_header} " + stdscr.addstr(0, 0, header_str + " " * (w - len(header_str) - 1)) + stdscr.attroff(curses.A_REVERSE) + + # Note Content + stdscr.addstr(2, 0, f"--- Note File: {args.file} (Font: {f_name}) ---", curses.A_BOLD) + + # Wrap text to display in console + term_lines, _, term_cursor_line, term_cursor_col = wrap_text_by_width(note_text, cursor_pos, font, max_width=360) + + for idx, line in enumerate(term_lines): + if 4 + idx < h - 3: + stdscr.addstr(4 + idx, 2, line) + + # Footer + stdscr.attron(curses.A_REVERSE) + footer_str = " Ctrl+S: Save | Ctrl+T: Theme | Ctrl+F: Font | Ctrl+X: Save & Exit " + stdscr.addstr(h - 1, 0, footer_str + " " * (w - len(footer_str) - 1)) + stdscr.attroff(curses.A_REVERSE) + + # Corner stats + if status_msg: + stdscr.addstr(h - 2, w - len(status_msg) - 2, f"[{status_msg}]", curses.A_BOLD) + else: + stdscr.addstr(h - 2, w - 18, f"Chars: {len(note_text)}") + + # Position Terminal Cursor + try: + stdscr.move(4 + term_cursor_line, 2 + term_cursor_col) + except: + pass + + stdscr.refresh() + + time.sleep(0.01) + + +def main(): + # 1. Determine daily default filename: note_YYYY-MM-DD.txt + today_str = datetime.date.today().strftime("%Y-%m-%d") + default_filename = f"note_{today_str}.txt" + + parser = argparse.ArgumentParser(description="Curses Pi Zero Kid Note Taking Streaming App") + parser.add_argument("--ip", default="192.168.68.123", help="Target ESP32-S3 IP Address (default: 192.168.68.123)") + parser.add_argument("--port", type=int, default=8081, help="Streaming TCP Port (default: 8081)") + parser.add_argument("--file", default=default_filename, help=f"Filename to persist notes (default: {default_filename})") + args = parser.parse_args() + + curses.wrapper(run_editor, args) + print(f"\nExited. Note saved to {args.file}. Thank you for using Note Keeper!") + + +if __name__ == "__main__": + main() diff --git a/rgb_led_util.py b/rgb_led_util.py index 8365bf5..95447e9 100644 --- a/rgb_led_util.py +++ b/rgb_led_util.py @@ -6,7 +6,10 @@ import neopixel class BoardLED: """Utility class to control the onboard WS2812 (NeoPixel) RGB LED on GPIO 38.""" - def __init__(self, pin_num=38): + def __init__(self, pin_num=None): + import sys + if pin_num is None: + pin_num = 14 if sys.platform == 'rp2' else 38 self.np = neopixel.NeoPixel(Pin(pin_num), 1) self.base_color = (0, 0, 0) self.off() diff --git a/rlcd.py b/rlcd.py index 766551a..a0ad832 100644 --- a/rlcd.py +++ b/rlcd.py @@ -129,13 +129,19 @@ class RLCD: self.write_cmd(0x3A); self.write_data(0x11) self.write_cmd(0xB9); self.write_data(0x20) self.write_cmd(0xB8); self.write_data(0x29) - self.write_cmd(0x20) # Inversion OFF (White Background) + self.write_cmd(0x21) # Inversion ON (Black Background by default) self.write_cmd(0x2A); self.write_data([0x12, 0x2A]) self.write_cmd(0x2B); self.write_data([0x00, 0xC7]) self.write_cmd(0x35); self.write_data(0x00) self.write_cmd(0xD0); self.write_data(0xFF) self.write_cmd(0x38); self.write_cmd(0x29) + def invert(self, enable): + if enable: + self.write_cmd(0x21) # Inversion ON (Black Background) + else: + self.write_cmd(0x20) # Inversion OFF (White Background) + # --- THE HEAVY LIFTER (Optimized) --- @micropython.native def show(self): diff --git a/scratch/delete_all.py b/scratch/delete_all.py new file mode 100644 index 0000000..00a5ee1 --- /dev/null +++ b/scratch/delete_all.py @@ -0,0 +1,86 @@ +import serial +import time +import sys + +PORT = '/dev/cu.usbmodem101' +print("=== MicroPython Robust Flash Wipe Tool ===") +print(f"Waiting for a clean, active connection on {PORT}...") +print("Please press the physical RESET button on the board now...") + +ser = None +while True: + try: + # Open port + ser = serial.Serial(PORT, 115200, timeout=1.0) + # Try writing Ctrl-C to verify if the link is active + ser.write(b'\x03') + time.sleep(0.01) + ser.write(b'\x03') + + # If write succeeded, we have an active link! + print("\n[+] Active connection established!") + break + except (serial.SerialException, OSError, Exception) as e: + if ser is not None: + try: + ser.close() + except: + pass + ser = None + # Print dot to show progress, stay on same line + print(".", end="") + sys.stdout.flush() + time.sleep(0.1) + +print("[*] Sending remaining Ctrl-C storm to break into REPL...") +for _ in range(80): + try: + ser.write(b'\x03') + time.sleep(0.005) + except Exception as e: + print(f"\n[-] Write error during storm: {e}") + break + +time.sleep(0.1) +try: + ser.reset_input_buffer() +except Exception as e: + print(f"[-] Buffer reset failed: {e}") + +# Check if we have REPL access +print("[*] Verifying REPL responsiveness...") +try: + ser.write(b'\r\n') + time.sleep(0.2) + if ser.in_waiting: + resp = ser.read(ser.in_waiting) + print(f"REPL Response:\n{resp.decode('utf-8', errors='ignore')}") + else: + ser.write(b'\x03\r\n') + time.sleep(0.2) + resp = ser.read(ser.in_waiting) + print(f"REPL Response:\n{resp.decode('utf-8', errors='ignore')}") +except Exception as e: + print(f"[-] REPL verification failed: {e}") + ser.close() + sys.exit(1) + +print("[*] Deleting all files on the flash...") +wipe_code = b"import os; print('FILES_BEFORE:', os.listdir()); [os.remove(f) for f in os.listdir() if not (os.stat(f)[0] & 0x4000)]; print('FILES_AFTER:', os.listdir()); print('WIPE_SUCCESS')\r\n" + +try: + ser.write(wipe_code) + time.sleep(1.0) + if ser.in_waiting: + result = ser.read(ser.in_waiting).decode('utf-8', errors='ignore') + print(f"Execution Output:\n{result}") + if "WIPE_SUCCESS" in result: + print("[+] SUCCESS! Flash has been wiped clean.") + else: + print("[-] Wipe did not finish successfully.") + else: + print("[-] No response to wipe command.") +except Exception as e: + print(f"[-] Error during wipe command execution: {e}") + +ser.close() diff --git a/scratch/interact_repl.py b/scratch/interact_repl.py new file mode 100644 index 0000000..16a1e9c --- /dev/null +++ b/scratch/interact_repl.py @@ -0,0 +1,39 @@ +import serial +import time + +print("Opening serial port /dev/cu.usbmodem101...") +try: + ser = serial.Serial('/dev/cu.usbmodem101', 115200, timeout=1.0) + print("Opened successfully!") + + # Clear input buffer + ser.reset_input_buffer() + + # Send Ctrl-C to interrupt anything + print("Sending Ctrl-C...") + ser.write(b'\x03') + time.sleep(0.2) + + # Read response + if ser.in_waiting: + print(f"After Ctrl-C: {ser.read(ser.in_waiting)}") + + # Send newline + print("Sending newline...") + ser.write(b'\r\n') + time.sleep(0.2) + if ser.in_waiting: + print(f"After newline: {ser.read(ser.in_waiting)}") + + # Send test print command + print("Sending test print command...") + ser.write(b"print('REPL_ACTIVE')\r\n") + time.sleep(0.5) + if ser.in_waiting: + print(f"After command: {ser.read(ser.in_waiting)}") + else: + print("No response to command.") + + ser.close() +except Exception as e: + print(f"Error: {e}") diff --git a/scratch/listen_touch.py b/scratch/listen_touch.py new file mode 100644 index 0000000..88491dc --- /dev/null +++ b/scratch/listen_touch.py @@ -0,0 +1,24 @@ +import serial +import time +import sys + +PORT = '/dev/cu.usbmodem101' +print("=== MicroPython Serial Event Listener ===") +print(f"Opening {PORT} at 115200...") +try: + ser = serial.Serial(PORT, 115200, timeout=1.0) + print("Opened successfully! Press Ctrl+C on the host to stop listening.") + print("Listening for touch events or debug logs from the board...\n") + + # We do NOT send Ctrl-C to the board, we just read what it outputs + while True: + if ser.in_waiting: + data = ser.read(ser.in_waiting) + text = data.decode('utf-8', errors='ignore') + sys.stdout.write(text) + sys.stdout.flush() + time.sleep(0.05) +except KeyboardInterrupt: + print("\nStopped listening.") +except Exception as e: + print(f"\nError: {e}") diff --git a/scratch/nuke_boot.py b/scratch/nuke_boot.py new file mode 100644 index 0000000..3433ac8 --- /dev/null +++ b/scratch/nuke_boot.py @@ -0,0 +1,71 @@ +import serial +import time +import os +import sys + +PORT = '/dev/cu.usbmodem101' +print("=== MicroPython Boot Interrupter & Recovery Tool ===") +print(f"Waiting for {PORT} to appear. Please press the RESET button on the board...") + +while True: + try: + ser = serial.Serial(PORT, 115200, timeout=1.0) + print("\n[+] Port opened successfully!") + break + except Exception as e: + # Port not present or busy + time.sleep(0.01) + +print("[*] Sending Ctrl-C storm to halt boot sequence...") +# Send 100 Ctrl-C interrupts in rapid succession +for _ in range(100): + try: + ser.write(b'\x03') + time.sleep(0.005) + except Exception as e: + print(f"Write error during storm: {e}") + break + +time.sleep(0.1) +ser.reset_input_buffer() + +# Check if we have REPL access by sending a newline and expecting a prompt +print("[*] Verifying REPL responsiveness...") +ser.write(b'\r\n') +time.sleep(0.2) + +if ser.in_waiting: + resp = ser.read(ser.in_waiting) + print(f"REPL Response:\n{resp.decode('utf-8', errors='ignore')}") +else: + print("[-] No response to newline, trying to force it...") + ser.write(b'\x03\r\n') + time.sleep(0.2) + resp = ser.read(ser.in_waiting) + print(f"REPL Response:\n{resp.decode('utf-8', errors='ignore')}") + +# Try to rename main.py to stop it from running on subsequent boots +print("[*] Attempting to disable main.py...") +rename_code = b""" +import os +try: + os.rename('main.py', 'main_bad.py') + print('RENAME:OK') +except Exception as e: + print('RENAME:ERROR:', e) +""" +ser.write(rename_code + b'\r\n') +time.sleep(0.5) + +if ser.in_waiting: + result = ser.read(ser.in_waiting).decode('utf-8', errors='ignore') + print(f"Execution Output:\n{result}") + if "RENAME:OK" in result: + print("[+] SUCCESS! main.py has been disabled.") + print("[+] You can now safely upload files or reboot.") + else: + print("[-] Rename failed or returned unexpected output.") +else: + print("[-] No response to rename command.") + +ser.close() diff --git a/scratch/verify_actual_virtual_mcp.py b/scratch/verify_actual_virtual_mcp.py new file mode 100644 index 0000000..653b33c --- /dev/null +++ b/scratch/verify_actual_virtual_mcp.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +import sys +import os +import json +import time + +sys.path.append("/Users/adolforeyna/Projects/PiZeroNoteKeeper") +from kernel.system import SystemContext + +def run_test(): + context = SystemContext(target_ip="127.0.0.1", target_port=8081, pin_target=True) + # Start with wrong port (80 instead of 8080) to force fallback + context.mcp_port = 80 + + # Clear any initial status message from discovery + time.sleep(0.5) + context.status_msg = "" + + print("Sending play_beep command starting with port 80...") + context.play_beep(frequency=1200, duration_ms=40, volume=30) + + # Wait for fallback to execute (requires timeout retry) + time.sleep(2.0) + + # Verify fallback succeeded and updated port to 8080 + if context.mcp_port == 8080: + print("SUCCESS: Fallback automatically corrected mcp_port to 8080!") + else: + print("FAIL: Fallback did not correct port. Current port: {}, status: '{}'".format(context.mcp_port, context.status_msg)) + sys.exit(1) + +if __name__ == "__main__": + run_test() diff --git a/scratch/verify_launcher_beep.py b/scratch/verify_launcher_beep.py new file mode 100644 index 0000000..2fc66c2 --- /dev/null +++ b/scratch/verify_launcher_beep.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +import sys +import os +import json +import time +import http.server +import threading + +# Add PiZeroNoteKeeper to path so we can import kernel +sys.path.append("/Users/adolforeyna/Projects/PiZeroNoteKeeper") +from kernel.system import SystemContext + +# Mock HTTP Server to capture the POST request +class MockMCPServer(http.server.BaseHTTPRequestHandler): + received_request = None + + def do_POST(self): + if self.path == "/api/mcp": + content_length = int(self.headers['Content-Length']) + post_data = self.rfile.read(content_length) + MockMCPServer.received_request = json.loads(post_data.decode('utf-8')) + + # Send success response + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps({"jsonrpc": "2.0", "result": "OK", "id": "test"}).encode('utf-8')) + else: + self.send_response(404) + self.end_headers() + + def log_message(self, format, *args): + # Suppress logging to keep output clean + pass + +def run_test(): + # Start mock server + server = http.server.HTTPServer(('127.0.0.1', 9999), MockMCPServer) + server_thread = threading.Thread(target=server.serve_forever, daemon=True) + server_thread.start() + print("Mock MCP server running on http://127.0.0.1:9999") + + # Initialize SystemContext pointing to mock server + context = SystemContext(target_ip="127.0.0.1", target_port=8081, pin_target=True) + # Set the discovered/active mcp_port + context.mcp_port = 9999 + + print("Sending play_beep command...") + context.play_beep(frequency=1200, duration_ms=40, volume=30) + + # Wait for background thread to execute the HTTP request + timeout = 3.0 + start_time = time.time() + while MockMCPServer.received_request is None and (time.time() - start_time) < timeout: + time.sleep(0.1) + + # Shutdown server + server.shutdown() + server.server_close() + + # Assert and verify + req = MockMCPServer.received_request + if req is None: + print("FAIL: No request received by mock MCP server within timeout.") + sys.exit(1) + + print("Received request payload:") + print(json.dumps(req, indent=2)) + + try: + assert req["jsonrpc"] == "2.0" + assert req["method"] == "tools/call" + assert req["params"]["name"] == "play_tone" + args = req["params"]["arguments"] + assert args["frequency"] == 1200 + assert args["duration_ms"] == 40 + assert args["volume"] == 30 + print("\nSUCCESS: All request fields verified perfectly!") + except AssertionError as e: + print("\nFAIL: Request fields did not match expected values.", e) + sys.exit(1) + +if __name__ == "__main__": + run_test() diff --git a/st7796.py b/st7796.py new file mode 100644 index 0000000..c28ce01 --- /dev/null +++ b/st7796.py @@ -0,0 +1,249 @@ +import time +from machine import Pin, SPI +import framebuf +import micropython + +class ST7796: + def __init__(self, spi, cs, dc, rst, bl=None, width=480, height=320, invert_color=True): + self.spi = spi + self.cs = cs + self.dc = dc + self.rst = rst + self.bl = bl + self.width = width + self.height = height + self.invert_color = invert_color + + # 1. 1-bit Canvas Buffer (Standard MONO_HLSB for drawing) + self.hw_len = (self.width * self.height) // 8 + self.canvas_buffer = bytearray(self.hw_len) + self.canvas = framebuf.FrameBuffer(self.canvas_buffer, self.width, self.height, framebuf.MONO_HLSB) + + # Pre-allocate chunk buffer for conversion (16 rows: 480 * 16 * 2 = 15360 bytes) + self.chunk_rows = 16 + self.row_buffer = bytearray(self.width * self.chunk_rows * 2) + + # Initialize pins + self.cs.init(self.cs.OUT, value=1) + self.dc.init(self.dc.OUT, value=0) + self.rst.init(self.rst.OUT, value=1) + + if self.bl is not None: + # PWM or Pin backlight control + if isinstance(self.bl, Pin): + self.bl.init(self.bl.OUT, value=1) + + self.reset() + self.init_display() + self.clear(0) + self.show() + + # --- DRAWING WRAPPERS --- + def pixel(self, x, y, c): self.canvas.pixel(x, y, c) + def line(self, x1, y1, x2, y2, c): self.canvas.line(x1, y1, x2, y2, c) + def rect(self, x, y, w, h, c): self.canvas.rect(x, y, w, h, c) + def fill_rect(self, x, y, w, h, c): self.canvas.fill_rect(x, y, w, h, c) + def text(self, msg, x, y, c=1): self.canvas.text(msg, x, y, c) + def clear(self, c=0): self.canvas.fill(c) + + # --- SCALABLE TEXT --- + def text_large(self, msg, x, y, scale=2, c=1): + char_w = 8; char_h = 8 + tmp_buf = bytearray(char_w * char_h // 8) + tmp_fb = framebuf.FrameBuffer(tmp_buf, char_w, char_h, framebuf.MONO_HLSB) + for char in msg: + tmp_fb.fill(0); tmp_fb.text(char, 0, 0, 1) + for py in range(8): + for px in range(8): + if tmp_fb.pixel(px, py): + self.canvas.fill_rect(x + (px * scale), y + (py * scale), scale, scale, c) + x += (8 * scale) + + # --- RAW BITMAPS (1:1 scale) --- + def bitmap(self, x, y, w, h, pixel_data): + img = framebuf.FrameBuffer(pixel_data, w, h, framebuf.MONO_HLSB) + self.canvas.blit(img, x, y) + + # --- PBM FILE LOADER WITH SCALING --- + def draw_pbm(self, filename, x, y, scale=1): + try: + with open(filename, 'rb') as f: + line1 = f.readline() + if not line1.startswith(b'P4'): print("Err: Not P4 PBM"); return + while True: + line = f.readline() + if not line.startswith(b'#'): break + dims = line.split(); w = int(dims[0]); h = int(dims[1]) + data = bytearray(f.read()) + + src_fb = framebuf.FrameBuffer(data, w, h, framebuf.MONO_HLSB) + + if scale == 1: + self.canvas.blit(src_fb, x, y) + else: + for sy in range(h): + for sx in range(w): + if src_fb.pixel(sx, sy): + self.canvas.fill_rect(x + (sx * scale), y + (sy * scale), scale, scale, 1) + print(f"Loaded {filename} (scale {scale})") + except OSError: + print(f"Error: Could not open {filename}") + + # --- SCREENSHOT --- + def save_screenshot(self, filename): + print(f"Saving screenshot to {filename}...") + try: + with open(filename, 'wb') as f: + f.write(b'P4\n') + f.write(f"{self.width} {self.height}\n".encode()) + f.write(self.canvas_buffer) + print("Saved!") + except Exception as e: + print(f"Error saving screenshot: {e}") + + # --- HARDWARE LOGIC --- + def reset(self): + self.rst(1); time.sleep_ms(5); self.rst(0); time.sleep_ms(15); self.rst(1); time.sleep_ms(15) + + def write_cmd(self, cmd): + self.dc(0); self.cs(0); self.spi.write(bytearray([cmd])); self.cs(1) + + def write_data(self, data): + self.dc(1); self.cs(0) + if isinstance(data, int): self.spi.write(bytearray([data])) + elif isinstance(data, list): self.spi.write(bytearray(data)) + else: self.spi.write(data) + self.cs(1) + + def init_display(self): + # ST7796 Minimal Initialization Sequence (ported from working C++ codebase) + self.write_cmd(0x01) # SWRESET + time.sleep_ms(150) + + self.write_cmd(0x11) # SLPOUT + time.sleep_ms(120) + + # Pixel Format (COLMOD) = 0x55 (16-bit color / RGB565) + self.write_cmd(0x3A); self.write_data(0x55) + time.sleep_ms(10) + + # Memory Access Control (MADCTL) = 0xE0 (Landscape: MY=1, MX=1, MV=1, RGB order) + self.write_cmd(0x36); self.write_data(0xE0) + time.sleep_ms(10) + + # Display Inversion Control + if self.invert_color: + self.write_cmd(0x21) # INVON + else: + self.write_cmd(0x20) # INVOFF + time.sleep_ms(10) + + self.write_cmd(0x13) # NORON + time.sleep_ms(10) + + self.write_cmd(0x29) # DISPON + time.sleep_ms(120) + + def invert(self, enable): + if enable: + self.write_cmd(0x21) # INVON + else: + self.write_cmd(0x20) # INVOFF + + def set_window(self, x0, y0, x1, y1): + # Column Address Set (CASET) + self.write_cmd(0x2A) + self.write_data(bytearray([x0 >> 8, x0 & 0xFF, x1 >> 8, x1 & 0xFF])) + + # Row Address Set (RASET) + self.write_cmd(0x2B) + self.write_data(bytearray([y0 >> 8, y0 & 0xFF, y1 >> 8, y1 & 0xFF])) + + # Memory Write (RAMWR) + self.write_cmd(0x2C) + + @micropython.native + def _convert_rows(self, start_row, num_rows, row_buf): + """Converts 1-bit monochrome row segment to 16-bit RGB565 format. + Compiles block-wise bitwise operations at native speed. + """ + width = self.width + canvas_buf = self.canvas_buffer + idx = 0 + + for y in range(start_row, start_row + num_rows): + byte_offset = y * (width // 8) + for x_byte_idx in range(width // 8): + val = canvas_buf[byte_offset + x_byte_idx] + + # Unroll 8 bits for speed + # Bit 7 + if val & 0x80: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 6 + if val & 0x40: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 5 + if val & 0x20: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 4 + if val & 0x10: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 3 + if val & 0x08: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 2 + if val & 0x04: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 1 + if val & 0x02: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + # Bit 0 + if val & 0x01: + row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF + else: + row_buf[idx] = 0x00; row_buf[idx+1] = 0x00 + idx += 2 + + def show(self): + """Refreshes the screen by writing the frame buffer segment-by-segment.""" + self.set_window(0, 0, self.width - 1, self.height - 1) + + self.dc(1) + self.cs(0) + + num_chunks = self.height // self.chunk_rows + for chunk in range(num_chunks): + start_row = chunk * self.chunk_rows + self._convert_rows(start_row, self.chunk_rows, self.row_buffer) + self.spi.write(self.row_buffer) + + self.cs(1) diff --git a/test_local_notetaker.py b/test_local_notetaker.py new file mode 100644 index 0000000..ca02c3d --- /dev/null +++ b/test_local_notetaker.py @@ -0,0 +1,42 @@ +import sys +import os +import datetime +from PIL import Image, ImageDraw, ImageFont + +# Add workspace to path to import notetaker functions +sys.path.append("/Users/adolforeyna/Projects/MicroPython/test1/Screen") +import notetaker + +def test_rendering(): + output_dir = "/Users/adolforeyna/.gemini/antigravity/brain/95f1ff45-7d9b-4493-ad2c-fdf647d73805" + text = "Adolfo Reyna Hey\nThis is a bigger font for my son!\nIt supports handwriting and typewriter styles." + cursor_pos = len(text) + scroll_top = 0 + + # Font folder + font_folder = "/Users/adolforeyna/Projects/MicroPython/test1/Screen/fonts" + + # Test cases: (font_path, font_size, font_name, output_filename) + test_cases = [ + (os.path.join(font_folder, "PatrickHand.ttf"), 22, "Handwritten", "render_kids_handwritten.png"), + (os.path.join(font_folder, "CourierPrime.ttf"), 20, "Typewriter", "render_kids_typewriter.png") + ] + + for font_path, font_size, font_name, filename in test_cases: + font, char_h = notetaker.load_font(font_path, font_size) + line_spacing = char_h + 8 + max_lines = (notetaker.BOTTOM_LIMIT - notetaker.TOP_LIMIT) // line_spacing + + img, scroll_top = notetaker.render_screen( + text, cursor_pos, scroll_top, font, + char_h, line_spacing, max_lines, + dark_mode=False, show_cursor=True, status_msg=f"Font: {font_name}" + ) + + dest_path = os.path.join(output_dir, filename) + img.save(dest_path) + print(f"Saved {font_name} layout test to {dest_path}") + print(f" char_h: {char_h}px, line_spacing: {line_spacing}px, max_lines: {max_lines}") + +if __name__ == "__main__": + test_rendering() diff --git a/test_stream.py b/test_stream.py new file mode 100644 index 0000000..0cf31d1 --- /dev/null +++ b/test_stream.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +import socket +import time +import argparse +import sys +from PIL import Image, ImageDraw, ImageFont + +# Dimensions of the reflective LCD +WIDTH = 400 +HEIGHT = 300 + +def map_to_rlcd_hw_buffer(img_1bit): + """Converts a standard 1-bit monochrome PIL Image to the Waveshare RLCD hardware buffer layout.""" + px = img_1bit.load() + hw_buffer = bytearray(15000) + + for y in range(HEIGHT): + for x in range(WIDTH): + if px[x, y]: # True if pixel is white (1) + inv_y = HEIGHT - 1 - y + byte_x = x // 2 + block_y = inv_y // 4 + index = byte_x * 75 + block_y + local_x = x % 2 + local_y = inv_y % 4 + bit = 7 - (local_y * 2 + local_x) + hw_buffer[index] |= (1 << bit) + + return hw_buffer + +def send_udp_frame_chunked(sock, esp32_ip, port, frame_idx, raw_bytes): + """Sends a 15,000-byte frame split into 15 small UDP packets to bypass MTU and OS limits.""" + frame_id = frame_idx % 256 + for chunk_idx in range(15): + packet = bytearray(1002) + packet[0] = frame_id + packet[1] = chunk_idx + start = chunk_idx * 1000 + packet[2:1002] = raw_bytes[start : start + 1000] + sock.sendto(packet, (esp32_ip, port)) + time.sleep(0.010) # Add small pacing delay to prevent network buffer flooding + +def generate_animation_frame(frame_idx, mode_name): + """Generates a test pattern animation frame using Pillow.""" + # Create 1-bit image (0 = black background) + img = Image.new("1", (WIDTH, HEIGHT), 0) + draw = ImageDraw.Draw(img) + + # Draw header text + draw.text((10, 10), "ESP32-S3 VIDEO STREAM TEST", fill=1) + draw.line((10, 22, 390, 22), fill=1) + + # Display active settings + draw.text((15, 30), f"Protocol : {mode_name}", fill=1) + draw.text((15, 45), f"Frame No : {frame_idx}", fill=1) + + # 1. Draw a bouncing rectangle + rect_w, rect_h = 60, 40 + bounce_x = int((frame_idx * 5) % (WIDTH - rect_w - 20)) + 10 + bounce_y = int(120 + 20 * (frame_idx % 10 < 5 and (frame_idx % 5) or (5 - frame_idx % 5))) + draw.rectangle([bounce_x, bounce_y, bounce_x + rect_w, bounce_y + rect_h], outline=1, width=2) + draw.text((bounce_x + 10, bounce_y + 15), "TCP/UDP", fill=1) + + # 2. Draw a spinning needle / line + import math + angle = frame_idx * 0.1 + center_x, center_y = 300, 200 + radius = 40 + end_x = center_x + radius * math.cos(angle) + end_y = center_y + radius * math.sin(angle) + draw.ellipse([center_x - radius, center_y - radius, center_x + radius, center_y + radius], outline=1) + draw.line((center_x, center_y, end_x, end_y), fill=1) + + # 3. Dynamic scrolling text at the bottom + scrolling_text = "Waveshare RLCD 4.2inch -- Low Power -- High Framerate Streaming in MicroPython" + scroll_pos = (frame_idx * 3) % 400 + draw.text((10 - scroll_pos, 275), scrolling_text, fill=1) + draw.text((410 - scroll_pos, 275), scrolling_text, fill=1) + draw.line((10, 270, 390, 270), fill=1) + + return img + +def stream_camera(esp32_ip, port, protocol): + """Streams camera capture frames using OpenCV (requires opencv-python).""" + try: + import cv2 + except ImportError: + print("Error: opencv-python is not installed. Run 'pip install opencv-python' or use default animation mode.") + sys.exit(1) + + print(f"Opening camera stream...") + cap = cv2.VideoCapture(0) + if not cap.isOpened(): + print("Error: Could not open camera.") + sys.exit(1) + + sock = None + if protocol == "tcp": + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((esp32_ip, port)) + else: # udp + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + # Enable broadcast if destination is broadcast + if esp32_ip.endswith(".255") or esp32_ip == "255.255.255.255": + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + + print(f"Streaming webcam to {esp32_ip}:{port} via {protocol.upper()}...") + frame_count = 0 + start_time = time.time() + + try: + while cap.isOpened(): + ret, frame = cap.read() + if not ret: + break + + # Flip image, convert to RGB + frame = cv2.flip(frame, 1) + frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + pil_img = Image.fromarray(frame_rgb) + + # Resize and dither + pil_img.thumbnail((WIDTH, HEIGHT)) + bg = Image.new("1", (WIDTH, HEIGHT), 0) + offset = ((WIDTH - pil_img.width) // 2, (HEIGHT - pil_img.height) // 2) + bg.paste(pil_img.convert("1", dither=Image.FLOYDSTEINBERG), offset) + + # Map pixels to hardware buffer + raw_bytes = map_to_rlcd_hw_buffer(bg) + + # Transmit + if protocol == "tcp": + sock.sendall(raw_bytes) + else: + send_udp_frame_chunked(sock, esp32_ip, port, frame_count, raw_bytes) + + frame_count += 1 + elapsed = time.time() - start_time + if elapsed >= 2.0: + print(f"Streaming performance: {frame_count / elapsed:.1f} FPS") + frame_count = 0 + start_time = time.time() + + # Target ~30 FPS + time.sleep(0.033) + + finally: + cap.release() + if sock: + sock.close() + +def stream_animation(esp32_ip, port, protocol): + """Streams a generated Pillow vector animation.""" + sock = None + if protocol == "tcp": + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((esp32_ip, port)) + else: # udp + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + if esp32_ip.endswith(".255") or esp32_ip == "255.255.255.255": + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + + print(f"Streaming vector animation to {esp32_ip}:{port} via {protocol.upper()}...") + frame_idx = 0 + start_time = time.time() + + try: + while True: + # Generate frame + pil_img = generate_animation_frame(frame_idx, f"{protocol.upper()} Stream") + + # Map pixels to hardware buffer + raw_bytes = map_to_rlcd_hw_buffer(pil_img) + + # Transmit + if protocol == "tcp": + sock.sendall(raw_bytes) + else: + send_udp_frame_chunked(sock, esp32_ip, port, frame_idx, raw_bytes) + + frame_idx += 1 + + # FPS tracking printout + if frame_idx % 60 == 0: + elapsed = time.time() - start_time + print(f"Streaming performance: {60 / elapsed:.1f} FPS") + start_time = time.time() + + # Restrict rate to ~25 FPS to match update scan limits nicely + time.sleep(0.04) + + except KeyboardInterrupt: + print("\nStreaming stopped.") + finally: + if sock: + sock.close() + +def main(): + parser = argparse.ArgumentParser(description="Host Client for ESP32-S3 TCP/UDP Video Streaming") + parser.add_argument("--ip", default="192.168.68.123", help="Target ESP32-S3 IP Address (default: 192.168.68.123)") + parser.add_argument("--protocol", choices=["tcp", "udp"], default="tcp", help="Streaming protocol (default: tcp)") + parser.add_argument("--source", choices=["camera", "animation"], default="animation", help="Streaming source (default: animation)") + args = parser.parse_args() + + # Assign ports based on protocol selection + port = 8081 if args.protocol == "tcp" else 8082 + + if args.source == "camera": + stream_camera(args.ip, port, args.protocol) + else: + stream_animation(args.ip, port, args.protocol) + +if __name__ == "__main__": + main() diff --git a/upload_files.py b/upload_files.py new file mode 100644 index 0000000..7b7c5ee --- /dev/null +++ b/upload_files.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +import json +import urllib.request +import sys + +ESP32_IP = "192.168.68.123" +URL = f"http://{ESP32_IP}/api/mcp" + +def call_mcp_tool(tool_name, arguments): + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": arguments + } + } + + req_data = json.dumps(payload).encode('utf-8') + req = urllib.request.Request( + URL, + data=req_data, + headers={"Content-Type": "application/json"}, + method="POST" + ) + + try: + with urllib.request.urlopen(req, timeout=10.0) as response: + resp_body = response.read().decode('utf-8') + resp_json = json.loads(resp_body) + if "error" in resp_json: + print(f"Error calling {tool_name}: {resp_json['error']}") + return None + return resp_json.get("result", {}).get("content", [{}])[0].get("text", "") + except Exception as e: + print(f"Connection failed for {tool_name}: {e}") + return None + +def upload_file(local_path, remote_path): + print(f"Reading local file {local_path}...") + with open(local_path, 'r') as f: + content = f.read() + + print(f"Uploading to ESP32: {remote_path} ({len(content)} chars)...") + res = call_mcp_tool("write_file", {"path": remote_path, "content": content}) + if res: + print(f"Success: {res}") + return True + return False + +def main(): + print(f"--- RP2350/ESP32-S3 OTA File Uploader ({ESP32_IP}) ---") + + # Upload new drivers, utilities, and updated scripts + files_to_upload = [ + ("st7796.py", "st7796.py"), + ("ft6336u.py", "ft6336u.py"), + ("battery_util.py", "battery_util.py"), + ("rgb_led_util.py", "rgb_led_util.py"), + ("video_stream.py", "video_stream.py"), + ("mcp_server.py", "mcp_server.py"), + ("boot.py", "boot.py"), + ("main.py", "main.py") + ] + + for local, remote in files_to_upload: + if not upload_file(local, remote): + print(f"Failed to upload {local}. Stopping.") + sys.exit(1) + + # 4. Trigger remote reboot + print("Sending reboot command to board...") + reboot_code = "import machine\nmachine.reset()" + res = call_mcp_tool("execute_python", {"code": reboot_code}) + if res: + print("Reboot command acknowledged. The board is restarting!") + else: + print("Reboot request failed. You may need to manually reset the board.") + +if __name__ == "__main__": + main() diff --git a/verify_broadcast.py b/verify_broadcast.py new file mode 100644 index 0000000..708b93d --- /dev/null +++ b/verify_broadcast.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +import urllib.request +import json +import base64 +import time +import subprocess +import os +from PIL import Image + +ESP32_IP = "192.168.68.123" +BROADCAST_IP = "192.168.68.255" +URL = f"http://{ESP32_IP}/api/mcp" +ARTIFACTS_DIR = "/Users/adolforeyna/.gemini/antigravity/brain/8b421e8b-10a8-4d06-a7d0-27bd82b42804" + +def capture_screenshot(output_filename): + print(f"Requesting screenshot from ESP32 ({ESP32_IP})...") + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "get_screenshot", + "arguments": {} + } + } + + req_data = json.dumps(payload).encode('utf-8') + req = urllib.request.Request( + URL, + data=req_data, + headers={"Content-Type": "application/json"}, + method="POST" + ) + + try: + with urllib.request.urlopen(req, timeout=10.0) as response: + resp_body = response.read().decode('utf-8') + resp_json = json.loads(resp_body) + content = resp_json.get("result", {}).get("content", [{}])[0].get("text", "") + + if content.startswith("__PBM_BASE64__:"): + pbm_b64 = content.split(":", 1)[1] + pbm_bytes = base64.b64decode(pbm_b64) + + temp_pbm = "temp_capture_bcast.pbm" + with open(temp_pbm, "wb") as pf: + pf.write(pbm_bytes) + + # Convert to PNG using Pillow + img = Image.open(temp_pbm) + dest_path = os.path.join(ARTIFACTS_DIR, output_filename) + img.save(dest_path) + os.remove(temp_pbm) + print(f"Success: Screenshot saved to {dest_path}") + return True + else: + print("Error: Invalid screenshot response format.") + return False + except Exception as e: + print(f"Failed to capture screenshot: {e}") + return False + +def main(): + print(f"--- Starting UDP Broadcast Stream Verification ({BROADCAST_IP}) ---") + + # 1. Start test_stream.py targeting broadcast IP + cmd = ["python3", "test_stream.py", "--ip", BROADCAST_IP, "--protocol", "udp", "--source", "animation"] + proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + print(f"Broadcast stream started in background (PID: {proc.pid}).") + + # 2. Wait for packets to propagate + print("Broadcasting UDP packets... Waiting 3 seconds...") + time.sleep(3.0) + + # 3. Capture screen rendering on ESP32 + capture_screenshot("udp_broadcast_success.png") + + # 4. Terminate stream client + print(f"Terminating broadcast client process...") + proc.terminate() + proc.wait() + print(f"Broadcast stream client stopped.") + + # Let the screen return to dashboard + print("Waiting 4 seconds to let the screen return to dashboard...") + time.sleep(4.0) + print("Completed.") + +if __name__ == "__main__": + main() diff --git a/verify_broadcast_single_frame.py b/verify_broadcast_single_frame.py new file mode 100644 index 0000000..5547d61 --- /dev/null +++ b/verify_broadcast_single_frame.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +import socket +import time +import urllib.request +import json +import base64 +import os +from PIL import Image, ImageDraw + +ESP32_IP = "192.168.68.123" +BROADCAST_IP = "192.168.68.255" +PORT = 8082 +ARTIFACTS_DIR = "/Users/adolforeyna/.gemini/antigravity/brain/8b421e8b-10a8-4d06-a7d0-27bd82b42804" + +def main(): + print(f"--- Starting UDP Broadcast Single-Frame Test ({BROADCAST_IP}) ---") + + # 1. Generate a test frame with Pillow displaying "BROADCAST SUCCESS!" + img = Image.new("1", (400, 300), 0) + draw = ImageDraw.Draw(img) + draw.text((100, 140), "BROADCAST SUCCESS!", fill=1) + draw.rectangle([40, 110, 360, 180], outline=1, width=2) + + # Map to Waveshare RLCD buffer format + px = img.load() + hw_buffer = 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) + hw_buffer[idx] |= (1 << bit) + + # 2. Open UDP socket with broadcast permission + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + + # Send all 15 chunks representing the single frame + print("Broadcasting chunks...") + frame_id = 99 + for chunk_idx in range(15): + packet = bytearray(1002) + packet[0] = frame_id + packet[1] = chunk_idx + start = chunk_idx * 1000 + packet[2:1002] = hw_buffer[start : start + 1000] + sock.sendto(packet, (BROADCAST_IP, PORT)) + time.sleep(0.002) # Small pacing delay + + sock.close() + print("Frame broadcast sent.") + + # 3. Wait 1 second for ESP32 to render and display + time.sleep(1.0) + + # 4. Capture screenshot from ESP32 + print(f"Requesting screenshot from ESP32 ({ESP32_IP}) to verify broadcast reception...") + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "get_screenshot", + "arguments": {} + } + } + + req_data = json.dumps(payload).encode('utf-8') + req = urllib.request.Request( + f"http://{ESP32_IP}/api/mcp", + data=req_data, + headers={"Content-Type": "application/json"}, + method="POST" + ) + + try: + with urllib.request.urlopen(req, timeout=10.0) as response: + resp_body = response.read().decode('utf-8') + resp_json = json.loads(resp_body) + content = resp_json.get("result", {}).get("content", [{}])[0].get("text", "") + + if content.startswith("__PBM_BASE64__:"): + pbm_b64 = content.split(":", 1)[1] + pbm_bytes = base64.b64decode(pbm_b64) + + temp_pbm = "temp_capture_bcast.pbm" + with open(temp_pbm, "wb") as pf: + pf.write(pbm_bytes) + + # Convert to PNG using Pillow + img = Image.open(temp_pbm) + dest_path = os.path.join(ARTIFACTS_DIR, "udp_broadcast_success.png") + img.save(dest_path) + os.remove(temp_pbm) + print(f"Success: Broadcast screenshot saved to {dest_path}") + else: + print("Error: Invalid screenshot response format.") + except Exception as e: + print(f"Failed to capture screenshot: {e}") + +if __name__ == "__main__": + main() diff --git a/verify_streaming.py b/verify_streaming.py new file mode 100644 index 0000000..82762fc --- /dev/null +++ b/verify_streaming.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +import urllib.request +import json +import base64 +import time +import subprocess +import os +from PIL import Image + +ESP32_IP = "192.168.68.123" +URL = f"http://{ESP32_IP}/api/mcp" +ARTIFACTS_DIR = "/Users/adolforeyna/.gemini/antigravity/brain/8b421e8b-10a8-4d06-a7d0-27bd82b42804" + +def capture_screenshot(output_filename): + print(f"Requesting screenshot from ESP32 ({ESP32_IP})...") + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "get_screenshot", + "arguments": {} + } + } + + req_data = json.dumps(payload).encode('utf-8') + req = urllib.request.Request( + URL, + data=req_data, + headers={"Content-Type": "application/json"}, + method="POST" + ) + + try: + with urllib.request.urlopen(req, timeout=10.0) as response: + resp_body = response.read().decode('utf-8') + resp_json = json.loads(resp_body) + content = resp_json.get("result", {}).get("content", [{}])[0].get("text", "") + + if content.startswith("__PBM_BASE64__:"): + pbm_b64 = content.split(":", 1)[1] + pbm_bytes = base64.b64decode(pbm_b64) + + temp_pbm = "temp_capture.pbm" + with open(temp_pbm, "wb") as pf: + pf.write(pbm_bytes) + + # Convert to PNG using Pillow + img = Image.open(temp_pbm) + dest_path = os.path.join(ARTIFACTS_DIR, output_filename) + img.save(dest_path) + os.remove(temp_pbm) + print(f"Success: Screenshot saved to {dest_path}") + return True + else: + print("Error: Invalid screenshot response format.") + return False + except Exception as e: + print(f"Failed to capture screenshot: {e}") + return False + +def run_stream_test(protocol): + print(f"\n--- Starting {protocol.upper()} Stream Verification ---") + + # 1. Start test_stream.py in background + cmd = ["python3", "test_stream.py", "--ip", ESP32_IP, "--protocol", protocol, "--source", "animation"] + proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + print(f"Stream client process started in background (PID: {proc.pid}).") + + # 2. Wait for frames to stream and render + print("Streaming frames... Waiting 3 seconds...") + time.sleep(3.0) + + # 3. Capture screen rendering + filename = f"{protocol}_stream_success.png" + capture_screenshot(filename) + + # 4. Terminate stream client + print(f"Terminating stream client process...") + proc.terminate() + proc.wait() + print(f"Stream client stopped.") + +def main(): + if not os.path.exists(ARTIFACTS_DIR): + os.makedirs(ARTIFACTS_DIR) + + # Test TCP + run_stream_test("tcp") + + # Test Dashboard Auto-Timeout (stream timeout is 3.0s, let's wait 4.0s) + print("\n--- Verifying Stream Inactivity Timeout ---") + print("Waiting 4 seconds for stream to time out on ESP32...") + time.sleep(4.0) + capture_screenshot("dashboard_resumed.png") + + # Test UDP + run_stream_test("udp") + + # Clean up stream + print("\nWait 4 seconds to let the screen return to dashboard...") + time.sleep(4.0) + + print("\nVerification Completed successfully.") + +if __name__ == "__main__": + main() diff --git a/video_stream.py b/video_stream.py new file mode 100644 index 0000000..49258dc --- /dev/null +++ b/video_stream.py @@ -0,0 +1,285 @@ +import socket +import time +import select +import sys +import micropython + +class VideoStreamServer: + def __init__(self, display, tcp_port=8081, udp_port=8082): + self.display = display + self.tcp_port = tcp_port + self.udp_port = udp_port + + # Sockets + self.tcp_server = None + self.tcp_client = None + self.udp_sock = None + + # State + self.active = False + self.last_packet_time = 0 + self.timeout_ms = 3000 + + # Frame buffering (Pre-allocated to prevent memory allocations) + self.buffer = bytearray(15000) + self.view = memoryview(self.buffer) + self.tcp_bytes_received = 0 + + # UDP Reassembly (15 chunks of 1000 bytes each) + self.udp_temp_buffer = bytearray(1002) + self.udp_view = memoryview(self.udp_temp_buffer) + self.current_frame_id = -1 + self.chunks_received = 0 # Bitmask for chunks 0-14 + + # Performance Stats + self.debug = False + self.frames_drawn = 0 + self.udp_packets_received = 0 + self.udp_frames_complete = 0 + self.dropped_udp_frames = 0 + self.last_draw_ms = 0 + self.last_fps = 0.0 + + # FPS Calculation + self.fps_start_time = time.ticks_ms() + self.fps_frame_count = 0 + + @micropython.native + def rlcd_to_mono(self, rlcd_buf, canvas_buf): + # canvas_buf size: 19200 bytes (480x320) + # Clear canvas_buf to 0 + for i in range(19200): + canvas_buf[i] = 0 + + X_OFF = 40 + Y_OFF = 10 + + # Loop over 15000 bytes + for index in range(15000): + val = rlcd_buf[index] + if val == 0: + continue + + byte_x = index // 75 + block_y = index % 75 + x_base = 2 * byte_x + X_OFF + y_base = 299 - 4 * block_y + Y_OFF + + # bit 7: local_y = 0, local_x = 0 + if val & 0x80: + cx = x_base + cy = y_base + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 6: local_y = 0, local_x = 1 + if val & 0x40: + cx = x_base + 1 + cy = y_base + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 5: local_y = 1, local_x = 0 + if val & 0x20: + cx = x_base + cy = y_base - 1 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 4: local_y = 1, local_x = 1 + if val & 0x10: + cx = x_base + 1 + cy = y_base - 1 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 3: local_y = 2, local_x = 0 + if val & 0x08: + cx = x_base + cy = y_base - 2 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 2: local_y = 2, local_x = 1 + if val & 0x04: + cx = x_base + 1 + cy = y_base - 2 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 1: local_y = 3, local_x = 0 + if val & 0x02: + cx = x_base + cy = y_base - 3 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + # bit 0: local_y = 3, local_x = 1 + if val & 0x01: + cx = x_base + 1 + cy = y_base - 3 + canvas_buf[cy * 60 + (cx >> 3)] |= (1 << (7 - (cx & 7))) + + def start(self): + """Initialize and start the sockets.""" + # 1. Start TCP Server + try: + self.tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.tcp_server.bind(('', self.tcp_port)) + self.tcp_server.listen(1) + self.tcp_server.setblocking(False) + print(f"TCP Stream Server started on port {self.tcp_port}") + except Exception as e: + print(f"Failed to start TCP stream server: {e}") + + # 2. Start UDP Server + 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(('', self.udp_port)) + self.udp_sock.setblocking(False) + print(f"UDP Stream Server started on port {self.udp_port}") + except Exception as e: + print(f"Failed to start UDP stream server: {e}") + + def update(self): + """Updates connection states and checks for incoming video data. + Call this periodically in the main execution loop. + """ + now = time.ticks_ms() + + # Calculate dynamic FPS every 1 second + fps_elapsed = time.ticks_diff(now, self.fps_start_time) + if fps_elapsed >= 1000: + self.last_fps = (self.fps_frame_count * 1000.0) / fps_elapsed + self.fps_frame_count = 0 + self.fps_start_time = now + + # 1. Check for inactivity timeout + if self.active and time.ticks_diff(now, self.last_packet_time) > self.timeout_ms: + print("Video stream timed out. Returning to dashboard.") + self.active = False + self.tcp_bytes_received = 0 + + # 2. Handle incoming UDP packet (if UDP is active) + if self.udp_sock: + while True: + try: + n = self.udp_sock.readinto(self.udp_temp_buffer) + if n is None or n == 0: + break # Socket buffer queue is empty + + self.udp_packets_received += 1 + frame_id = self.udp_temp_buffer[0] + chunk_idx = self.udp_temp_buffer[1] + + if self.debug: + print(f"[UDP] Received packet size {n}: frame={frame_id}, chunk={chunk_idx}") + + if chunk_idx < 15: + # If a new frame sequence starts, reset tracking and check if old frame was incomplete + if frame_id != self.current_frame_id: + if self.chunks_received != 0: + self.dropped_udp_frames += 1 + self.current_frame_id = frame_id + self.chunks_received = 0 + + # Copy chunk data directly to frame buffer + start_offset = chunk_idx * 1000 + self.buffer[start_offset : start_offset + 1000] = self.udp_view[2:1002] + + # Mark chunk index as received + self.chunks_received |= (1 << chunk_idx) + + # Check if all 15 chunks (indices 0 to 14) are received (0x7FFF) + if self.chunks_received == 0x7FFF: + self.udp_frames_complete += 1 + if self.debug: + print(f"[UDP] Frame {frame_id} complete! Drawing to screen.") + self.active = True + self.last_packet_time = now + self._draw_frame() + self.chunks_received = 0 # Reset for next frame + except OSError as e: + break # EWOULDBLOCK (no data available) + + # 3. Handle TCP stream + if self.tcp_server: + # If no client is connected, try to accept one + if self.tcp_client is None: + try: + self.tcp_client, addr = self.tcp_server.accept() + self.tcp_client.setblocking(False) + self.tcp_bytes_received = 0 + self.active = True + self.last_packet_time = now + print(f"TCP Stream client connected from: {addr}") + except OSError: + pass + + # If client is connected, read data + if self.tcp_client is not None: + try: + remaining = 15000 - self.tcp_bytes_received + n = self.tcp_client.readinto(self.view[self.tcp_bytes_received : self.tcp_bytes_received + remaining]) + + if n is not None: + if n == 0: + # Client closed connection + print("TCP Stream client disconnected.") + self.close_tcp_client() + else: + self.tcp_bytes_received += n + if self.debug: + print(f"[TCP] Received {n} bytes (Total: {self.tcp_bytes_received}/15000)") + self.last_packet_time = now + self.active = True + + # If we have a full frame, render it + if self.tcp_bytes_received == 15000: + if self.debug: + print("[TCP] Frame complete! Drawing to screen.") + self.tcp_bytes_received = 0 + self._draw_frame() + except OSError as e: + # Connection reset or socket error + print(f"TCP Stream socket error: {e}") + self.close_tcp_client() + + def _draw_frame(self): + """Sends the frame buffer directly to the SPI display bus.""" + draw_start = time.ticks_ms() + + if sys.platform == 'rp2': + # Decode the 15,000-byte RLCD buffer and write it centered on the 480x320 canvas + self.rlcd_to_mono(self.buffer, self.display.canvas_buffer) + self.display.show() + else: + # Frame boundary command sequences for the RLCD display + self.display.write_cmd(0x2A); self.display.write_data(0x12); self.display.write_data(0x2A) + self.display.write_cmd(0x2B); self.display.write_data(0x00); self.display.write_data(0xC7) + self.display.write_cmd(0x2C) + + # Send raw 15,000 bytes directly + self.display.write_data(self.buffer) + + self.last_draw_ms = time.ticks_diff(time.ticks_ms(), draw_start) + self.frames_drawn += 1 + self.fps_frame_count += 1 + + def get_stats(self): + """Returns streaming performance counters.""" + return { + "frames_drawn": self.frames_drawn, + "tcp_bytes_received": self.tcp_bytes_received, + "udp_packets_received": self.udp_packets_received, + "udp_frames_complete": self.udp_frames_complete, + "dropped_udp_frames": self.dropped_udp_frames, + "last_draw_ms": self.last_draw_ms, + "last_fps": round(self.last_fps, 1), + "active": self.active + } + + def close_tcp_client(self): + """Safely disconnects the TCP client socket.""" + if self.tcp_client: + try: + self.tcp_client.close() + except: + pass + self.tcp_client = None + self.tcp_bytes_received = 0 diff --git a/virtual_screen_mcp/server.py b/virtual_screen_mcp/server.py index a07bd13..52c8f07 100644 --- a/virtual_screen_mcp/server.py +++ b/virtual_screen_mcp/server.py @@ -146,6 +146,76 @@ class MainHandler(tornado.web.RequestHandler): # Serve the index.html from the same directory self.render("index.html") +class APIHandler(tornado.web.RequestHandler): + def set_default_headers(self): + self.set_header("Access-Control-Allow-Origin", "*") + self.set_header("Access-Control-Allow-Headers", "x-requested-with, content-type") + self.set_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") + + def options(self): + self.set_status(204) + self.finish() + + async def post(self): + try: + req = json.loads(self.request.body.decode('utf-8')) + method = req.get("method") + rpc_id = req.get("id") + + if method == "tools/call": + params = req.get("params", {}) + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + result = await execute_mcp_tool(tool_name, arguments) + if isinstance(result, list): + content = result + else: + content = [{"type": "text", "text": str(result)}] + + resp = { + "jsonrpc": "2.0", + "id": rpc_id, + "result": { + "content": content + } + } + elif method == "initialize": + resp = { + "jsonrpc": "2.0", + "id": rpc_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": { + "tools": {} + }, + "serverInfo": { + "name": "virtual-screen-mcp", + "version": "1.0.0" + } + } + } + else: + resp = { + "jsonrpc": "2.0", + "id": rpc_id, + "error": { + "code": -32601, + "message": f"Method {method} not found" + } + } + self.write(json.dumps(resp)) + except Exception as e: + resp = { + "jsonrpc": "2.0", + "id": None, + "error": { + "code": -32000, + "message": str(e) + } + } + self.write(json.dumps(resp)) + class ClientWebSocketHandler(tornado.websocket.WebSocketHandler): clients = set() @@ -633,6 +703,7 @@ background_tasks = set() async def main_async(port): app = tornado.web.Application([ (r"/", MainHandler), + (r"/api/mcp", APIHandler), (r"/ws", ClientWebSocketHandler), ], template_path=os.path.dirname(os.path.abspath(__file__))) diff --git a/wifi_util.py b/wifi_util.py index 60bd2c4..388d93c 100644 --- a/wifi_util.py +++ b/wifi_util.py @@ -1,6 +1,10 @@ -import network +try: + import network + has_network = True +except ImportError: + has_network = False -class WiFiUtil: +class _ActiveWiFiUtil: def __init__(self): self.wlan = network.WLAN(network.STA_IF) self.wlan.active(True) @@ -11,7 +15,6 @@ class WiFiUtil: networks = self.wlan.scan() result = [] for net in networks: - # network tuple: (ssid, bssid, channel, RSSI, authmode, hidden) try: ssid = net[0].decode('utf-8') except Exception: @@ -26,3 +29,11 @@ class WiFiUtil: # Sort by signal strength (RSSI) descending result.sort(key=lambda x: x['rssi'], reverse=True) return result + +class _DummyWiFiUtil: + def __init__(self): + print("Wi-Fi not supported on this platform.") + def scan(self): + return [] + +WiFiUtil = _ActiveWiFiUtil if has_network else _DummyWiFiUtil