#!/usr/bin/env python3 """ ESP32 Async Batch Configuration Tool Configures 30+ ESP32 devices concurrently using non-blocking I/O. Features: - Concurrent execution (configure 30 devices in <20 seconds) - Regex-based state detection (Robust against noise) - Context-aware logging """ import asyncio import serial_asyncio import sys import os import argparse import ipaddress import re import time import logging # Ensure detection script is available sys.path.append(os.path.dirname(os.path.abspath(__file__))) try: import detect_esp32 except ImportError: print("Error: 'detect_esp32.py' not found.") sys.exit(1) # --- Logging Setup --- class DeviceLoggerAdapter(logging.LoggerAdapter): """Prefixes log messages with the device port name""" def process(self, msg, kwargs): return '[%s] %s' % (self.extra['connid'], msg), kwargs # Configure clean logging format logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%H:%M:%S') logger = logging.getLogger("BatchConfig") class AsyncConfigurator: """ Manages the lifecycle of configuring a single ESP32 device via Async Serial. """ def __init__(self, port, target_ip, args): self.port = port self.target_ip = target_ip self.args = args self.log = DeviceLoggerAdapter(logger, {'connid': port}) # Pre-compile Regex patterns for efficiency (Parsing logic) self.regex_got_ip = re.compile(r'got ip:(\d+\.\d+\.\d+\.\d+)', re.IGNORECASE) self.regex_config_saved = re.compile(r'Config saved|saved to NVS', re.IGNORECASE) # Prompts that indicate the device is alive and listening self.regex_ready = re.compile(r'Initialization complete|GPS synced|No WiFi config found', re.IGNORECASE) self.regex_error = re.compile(r'Error:|Failed|Disconnect', re.IGNORECASE) async def run(self): """Main execution workflow for this device""" try: reader, writer = await serial_asyncio.open_serial_connection(url=self.port, baudrate=115200) except Exception as e: self.log.error(f"Failed to open port: {e}") return False try: # 1. Hardware Reset self.log.info("Resetting...") writer.transport.serial.dtr = False writer.transport.serial.rts = True await asyncio.sleep(0.1) writer.transport.serial.rts = False await asyncio.sleep(0.1) writer.transport.serial.dtr = True # 2. Wait for Boot (Regex detection) if not await self._wait_for_boot(reader): self.log.warning("Boot prompt missed, sending config anyway...") # 3. Send Configuration await self._send_config(writer) # 4. Verify Success return await self._verify_configuration(reader) except Exception as e: self.log.error(f"Exception: {e}") return False finally: writer.close() await writer.wait_closed() async def _wait_for_boot(self, reader): """Reads stream until a known 'ready' prompt appears""" self.log.info("Waiting for boot...") timeout = time.time() + 5 # 5 second boot timeout while time.time() < timeout: try: # Read line with a short timeout to keep checking total time line_bytes = await asyncio.wait_for(reader.readline(), timeout=0.5) line = line_bytes.decode('utf-8', errors='ignore').strip() if not line: continue if self.regex_ready.search(line): return True except asyncio.TimeoutError: continue return False async def _send_config(self, writer): """Builds and transmits the configuration command""" self.log.info(f"Sending config for IP {self.target_ip}...") config_str = ( f"CFG\n" f"SSID:{self.args.ssid}\n" f"PASS:{self.args.password}\n" f"IP:{self.target_ip}\n" f"MASK:{self.args.netmask}\n" f"GW:{self.args.gateway}\n" f"DHCP:0\n" f"BAND:{self.args.band}\n" f"BW:{self.args.bandwidth}\n" f"POWERSAVE:{self.args.powersave}\n" f"MODE:{self.args.mode}\n" f"MON_CH:{self.args.monitor_channel}\n" f"END\n" ) writer.write(config_str.encode('utf-8')) await writer.drain() async def _verify_configuration(self, reader): """Monitors output for confirmation of IP assignment""" self.log.info("Verifying configuration...") timeout = time.time() + 15 # 15s connection timeout while time.time() < timeout: try: line_bytes = await asyncio.wait_for(reader.readline(), timeout=1.0) line = line_bytes.decode('utf-8', errors='ignore').strip() if not line: continue # Check for IP assignment m_ip = self.regex_got_ip.search(line) if m_ip: got_ip = m_ip.group(1) if got_ip == self.target_ip: self.log.info(f"SUCCESS: Assigned {got_ip}") return True else: self.log.warning(f"MISMATCH: Wanted {self.target_ip}, got {got_ip}") # Check for errors if self.regex_error.search(line): self.log.warning(f"Device Error: {line}") except asyncio.TimeoutError: continue self.log.error("Timeout: Device did not report IP address.") return False async def main_async(): parser = argparse.ArgumentParser(description='Async ESP32 Batch Config') # Arguments matching your existing tools parser.add_argument('--start-ip', required=True, help='Starting Static IP') parser.add_argument('-s', '--ssid', default='ClubHouse2G', help='WiFi SSID') parser.add_argument('-P', '--password', default='ez2remember', help='WiFi password') parser.add_argument('-g', '--gateway', default='192.168.1.1', help='Gateway IP') parser.add_argument('-m', '--netmask', default='255.255.255.0', help='Netmask') parser.add_argument('-b', '--band', default='2.4G', choices=['2.4G', '5G']) parser.add_argument('-B', '--bandwidth', default='HT20', choices=['HT20', 'HT40', 'VHT80']) parser.add_argument('-ps', '--powersave', default='NONE') parser.add_argument('-M', '--mode', default='STA') parser.add_argument('-mc', '--monitor-channel', type=int, default=36) args = parser.parse_args() # 1. Detect print("Step 1: Detecting Devices...") devices = detect_esp32.detect_esp32_devices() if not devices: print("No devices found.") return # Sort naturally (ttyUSB2 before ttyUSB10) def natural_keys(d): return [int(c) if c.isdigit() else c for c in re.split(r'(\d+)', d.device)] devices.sort(key=natural_keys) try: start_ip_obj = ipaddress.IPv4Address(args.start_ip) except: print(f"Invalid IP: {args.start_ip}") return # 2. Configure Concurrently print(f"Step 2: Configuring {len(devices)} devices concurrently...") tasks = [] for i, dev in enumerate(devices): current_ip = str(start_ip_obj + i) configurator = AsyncConfigurator(dev.device, current_ip, args) tasks.append(configurator.run()) # Run everything at once results = await asyncio.gather(*tasks) # 3. Report success_count = results.count(True) print("\n" + "="*40) print(f"Total Devices: {len(devices)}") print(f"Success: {success_count}") print(f"Failed: {len(devices) - success_count}") print("="*40) if __name__ == '__main__': try: if os.name == 'nt': asyncio.set_event_loop(asyncio.ProactorEventLoop()) asyncio.run(main_async()) except KeyboardInterrupt: print("\nAborted by user.")