Adolfo Reyna
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import urllib.request
|
|
import sys
|
|
|
|
# Default ESP32 IP
|
|
ESP32_IP = "192.168.68.124"
|
|
|
|
# Allow overriding IP via command-line argument
|
|
if len(sys.argv) > 1:
|
|
ESP32_IP = sys.argv[1]
|
|
|
|
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 = [
|
|
("lib/board_config.py", "lib/board_config.py"),
|
|
("lib/st7796.py", "lib/st7796.py"),
|
|
("lib/ft6336u.py", "lib/ft6336u.py"),
|
|
("lib/rlcd.py", "lib/rlcd.py"),
|
|
("lib/battery_util.py", "lib/battery_util.py"),
|
|
("lib/rgb_led_util.py", "lib/rgb_led_util.py"),
|
|
("lib/audio_util.py", "lib/audio_util.py"),
|
|
("lib/rtc_util.py", "lib/rtc_util.py"),
|
|
("lib/shtc3_util.py", "lib/shtc3_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()
|