122 lines
3.4 KiB
Python
122 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""POST-script: auto-detect upload_port by querying connected ESP32s.
|
|
|
|
Each ESP32 running SylDa firmware responds to the serial command "ID?"
|
|
with either "SylDa_Server" or "SylDa_Client". This script scans
|
|
/dev/serial/by-id/ for Espressif devices, queries each one, and sets
|
|
UPLOAD_PORT for the matching environment.
|
|
|
|
If no device responds (e.g. first flash), the user is prompted to
|
|
select a port manually.
|
|
"""
|
|
Import("env")
|
|
|
|
if env.IsIntegrationDump():
|
|
Return()
|
|
|
|
ENV_TO_ID = {
|
|
"s3lite-server": "SylDa_Server",
|
|
"s3lite-client": "SylDa_Client",
|
|
}
|
|
|
|
import glob
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
try:
|
|
import serial
|
|
except ImportError:
|
|
print(" WARNING: pyserial not installed, skipping auto-detect")
|
|
print(" Install: pip install pyserial")
|
|
Return()
|
|
|
|
|
|
def find_esp32_ports():
|
|
by_id = "/dev/serial/by-id"
|
|
if not os.path.isdir(by_id):
|
|
return []
|
|
ports = []
|
|
for entry in sorted(os.listdir(by_id)):
|
|
if "Espressif" in entry or "espressif" in entry:
|
|
ports.append(os.path.join(by_id, entry))
|
|
return ports
|
|
|
|
|
|
def query_id(port, timeout=5):
|
|
try:
|
|
ser = serial.Serial(port, 115200, timeout=0.5)
|
|
time.sleep(0.1)
|
|
ser.reset_input_buffer()
|
|
ser.write(b"ID?\n")
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
line = ser.readline().decode("utf-8", errors="ignore").strip()
|
|
if line:
|
|
ser.close()
|
|
return line
|
|
ser.close()
|
|
except (serial.SerialException, OSError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def detect_port(source, target, env):
|
|
if env.subst("$UPLOAD_PORT"):
|
|
print(f" Upload port already set to {env.subst('$UPLOAD_PORT')}, skipping auto-detect")
|
|
return
|
|
|
|
env_name = env.subst("$PIOENV")
|
|
expected_id = ENV_TO_ID.get(env_name)
|
|
|
|
if not expected_id:
|
|
print(f" Unknown environment '{env_name}', skipping auto-detect")
|
|
return
|
|
|
|
ports = find_esp32_ports()
|
|
|
|
if not ports:
|
|
print(f" No ESP32 serial ports found in /dev/serial/by-id/")
|
|
print(f" Connect an ESP32 and try again")
|
|
sys.exit(1)
|
|
|
|
if len(ports) == 1:
|
|
port = ports[0]
|
|
resp = query_id(port, timeout=5)
|
|
if resp and expected_id in resp:
|
|
print(f" Auto-detected {expected_id} on {port}")
|
|
else:
|
|
if resp:
|
|
print(f" Found '{resp}' on {port} (expected {expected_id})")
|
|
print(f" Only one ESP32 connected, using {port}")
|
|
env["UPLOAD_PORT"] = port
|
|
return
|
|
|
|
for port in ports:
|
|
resp = query_id(port, timeout=5)
|
|
if resp and expected_id in resp:
|
|
print(f" Found {expected_id} on {port}")
|
|
env["UPLOAD_PORT"] = port
|
|
return
|
|
|
|
print(f"\n Cannot auto-detect {expected_id} (no ID? response matched)")
|
|
print(f" Available ESP32 ports:")
|
|
for i, port in enumerate(ports, 1):
|
|
resp = query_id(port, timeout=3)
|
|
label = f" ({resp})" if resp else ""
|
|
print(f" {i}: {port}{label}")
|
|
|
|
try:
|
|
choice = input(f" Select port for {env_name} [{expected_id}] (1-{len(ports)}): ")
|
|
idx = int(choice) - 1
|
|
if 0 <= idx < len(ports):
|
|
env["UPLOAD_PORT"] = ports[idx]
|
|
return
|
|
except (ValueError, EOFError):
|
|
pass
|
|
|
|
print(" Invalid choice, aborting upload")
|
|
sys.exit(1)
|
|
|
|
|
|
env.AddPreAction("upload", detect_port)
|