adolforeyna/mcp_screen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement notetaking app, ST7796 display with touch support, audio beep navigation, and video streaming

Browse Source
This commit is contained in:
Adolfo Reyna
2026-06-08 15:04:36 -04:00
parent 070390355e
commit 3356e5d4a2
30 changed files with 3983 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scratch/delete_all.py
+86
View File
@@ -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()
scratch/interact_repl.py
+39
View File
@@ -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}")
scratch/listen_touch.py
+24
View File
@@ -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}")
scratch/nuke_boot.py
+71
View File
@@ -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()
scratch/verify_actual_virtual_mcp.py
+33
View File
@@ -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()
scratch/verify_launcher_beep.py
+84
View File
@@ -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()