Adolfo Reyna
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
import json
|
|
import urllib.request
|
|
import time
|
|
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 multiple times for each call since the connection might reset
|
|
for attempt in range(1, 100):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=1.0) as response:
|
|
resp_body = response.read().decode('utf-8')
|
|
resp_json = json.loads(resp_body)
|
|
if "error" in resp_json:
|
|
return None
|
|
return resp_json.get("result", {}).get("content", [{}])[0].get("text", "")
|
|
except Exception as e:
|
|
time.sleep(0.01)
|
|
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}...")
|
|
for attempt in range(1, 20):
|
|
res = call_mcp_tool("write_file", {"path": remote_path, "content": content})
|
|
if res:
|
|
print(f"Success uploading {remote_path} on attempt {attempt}!")
|
|
return True
|
|
print(f"Retry {attempt} uploading {remote_path}...")
|
|
time.sleep(0.1)
|
|
return False
|
|
|
|
def main():
|
|
print("=== ESP32 OTA Recovery Tool ===")
|
|
|
|
# Upload mcp_server.py first
|
|
if not upload_file("mcp_server.py", "mcp_server.py"):
|
|
print("Failed to upload mcp_server.py. Exiting.")
|
|
sys.exit(1)
|
|
|
|
# Upload video_stream.py
|
|
if not upload_file("video_stream.py", "video_stream.py"):
|
|
print("Failed to upload video_stream.py. Exiting.")
|
|
sys.exit(1)
|
|
|
|
print("Rebooting the board...")
|
|
reboot_code = "import machine\nmachine.reset()"
|
|
call_mcp_tool("execute_python", {"code": reboot_code})
|
|
print("Reboot command sent! The board should recover now.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|