#!/usr/bin/env python3 """ ESP32 Unified Deployment Tool Combines firmware flashing and device configuration with full control. Operation Modes: Default: Build + Flash + Configure --config-only: Configure only (no flashing) --flash-only: Build + Flash only (no configure) --flash-erase: Erase + Flash + Configure Examples: # Full deployment (default: flash + config) ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 # Config only (firmware already flashed) ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 --config-only # Flash only (preserve existing config) ./esp32_deploy.py --flash-only # Full erase + flash + config ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 --flash-erase # Limit concurrent flash for unpowered USB hub ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 --max-concurrent 2 # Retry specific failed devices (automatic sequential flashing) ./esp32_deploy.py --devices /dev/ttyUSB3,/dev/ttyUSB6,/dev/ttyUSB7 -s ClubHouse2G -P ez2remember --start-ip 192.168.1.63 # CSI-enabled devices ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.111 --csi # Monitor mode on channel 36 ./esp32_deploy.py --start-ip 192.168.1.90 -M MONITOR -mc 36 --config-only # 5GHz with 40MHz bandwidth ./esp32_deploy.py -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 -b 5G -B HT40 """ import asyncio import serial_asyncio import sys import os import argparse import ipaddress import re import time import logging from pathlib import Path # 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) # --- Configuration --- DEFAULT_MAX_CONCURRENT_FLASH = 4 # Conservative default for USB hub power limits class Colors: GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' RESET = '\033[0m' class DeviceLoggerAdapter(logging.LoggerAdapter): def process(self, msg, kwargs): return '[%s] %s' % (self.extra['connid'], msg), kwargs logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%H:%M:%S') logger = logging.getLogger("Deploy") class UnifiedDeployWorker: """Handles both flashing and configuration for a single ESP32 device""" def __init__(self, port, target_ip, args, build_dir, flash_sem): self.port = port self.target_ip = target_ip self.args = args self.build_dir = build_dir self.flash_sem = flash_sem self.log = DeviceLoggerAdapter(logger, {'connid': port}) # Regex Patterns self.regex_ready = re.compile(r'Initialization complete|GPS synced|GPS initialization aborted|No Config Found', re.IGNORECASE) self.regex_got_ip = re.compile(r'got ip:(\d+\.\d+\.\d+\.\d+)', re.IGNORECASE) self.regex_monitor_success = re.compile(r'Monitor mode active', re.IGNORECASE) self.regex_csi_saved = re.compile(r'CSI enable state saved', re.IGNORECASE) self.regex_status_connected = re.compile(r'WiFi connected: Yes', re.IGNORECASE) self.regex_error = re.compile(r'Error:|Failed|Disconnect', re.IGNORECASE) async def run(self): """Main execution workflow""" try: # Phase 1: Flash (if not config-only) if not self.args.config_only: async with self.flash_sem: if self.args.flash_erase: if not await self._erase_flash(): return False # HARD FAILURE: Flash Erase Failed if not await self._flash_firmware(): return False # HARD FAILURE: Flash Write Failed # Wait for port to stabilize after flash await asyncio.sleep(1.0) # Phase 2: Configure (if not flash-only) if not self.args.flash_only: if self.args.ssid and self.args.password: if not await self._configure_device(): # SOFT FAILURE: Config failed, but we treat it as success if flash passed self.log.warning(f"{Colors.YELLOW}Configuration verification failed. Marking as SUCCESS (Flash was OK).{Colors.RESET}") # We proceed to return True at the end else: self.log.warning("No SSID/Password provided, skipping config") if self.args.config_only: return False else: self.log.info(f"{Colors.GREEN}Flash Complete (Config Skipped){Colors.RESET}") return True except Exception as e: self.log.error(f"Worker Exception: {e}") return False async def _erase_flash(self): """Erase entire flash memory""" self.log.info(f"{Colors.YELLOW}Erasing flash...{Colors.RESET}") cmd = ['esptool.py', '-p', self.port, '-b', '115200', 'erase_flash'] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate() if proc.returncode == 0: self.log.info("Erase successful") return True self.log.error(f"Erase failed: {stderr.decode()}") return False async def _flash_firmware(self): """Flash firmware to device""" self.log.info("Flashing firmware...") cmd = [ 'esptool.py', '-p', self.port, '-b', str(self.args.baud), '--before', 'default_reset', '--after', 'hard_reset', 'write_flash', '@flash_args' ] proc = await asyncio.create_subprocess_exec( *cmd, cwd=self.build_dir, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300) except asyncio.TimeoutError: proc.kill() self.log.error("Flash timeout") return False if proc.returncode == 0: self.log.info("Flash successful") return True self.log.error(f"Flash failed: {stderr.decode()}") return False async def _configure_device(self): """Configure device via serial console""" self.log.info("Connecting to console...") try: reader, writer = await serial_asyncio.open_serial_connection( url=self.port, baudrate=115200 ) except Exception as e: self.log.error(f"Serial open failed: {e}") return False try: # Step 1: Hardware Reset (only if config-only mode) if self.args.config_only: self.log.info("Resetting device...") 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 # Step 2: Wait for Boot if not await self._wait_for_boot(reader): self.log.warning("Boot prompt missed, attempting config anyway...") # Step 3: Send Configuration await self._send_config(writer) # Step 4: Verify Success return await self._verify_configuration(reader) except Exception as e: self.log.error(f"Config error: {e}") return False finally: writer.close() await writer.wait_closed() async def _wait_for_boot(self, reader): """Wait for device boot completion""" self.log.info("Waiting for boot...") timeout = time.time() + 10 while time.time() < timeout: try: line_bytes = await asyncio.wait_for(reader.readline(), timeout=0.5) line = line_bytes.decode('utf-8', errors='ignore').strip() if self.regex_ready.search(line): return True except asyncio.TimeoutError: continue return False async def _send_config(self, writer): """Build and send configuration message""" csi_val = '1' if self.args.csi_enable else '0' self.log.info(f"Sending config for {self.target_ip} (Mode:{self.args.mode}, CSI:{csi_val})...") 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"CSI:{csi_val}\n" f"END\n" ) writer.write(config_str.encode('utf-8')) await writer.drain() async def _verify_configuration(self, reader): """Verify configuration success""" self.log.info("Verifying configuration...") timeout = time.time() + 20 csi_saved = False 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 CSI save confirmation if self.regex_csi_saved.search(line): csi_saved = True # Check for Station Mode Success (IP Address) m_ip = self.regex_got_ip.search(line) if m_ip: got_ip = m_ip.group(1) if got_ip == self.target_ip: csi_msg = f" {Colors.CYAN}(CSI saved){Colors.RESET}" if csi_saved else "" self.log.info(f"{Colors.GREEN}SUCCESS: Assigned {got_ip}{csi_msg}{Colors.RESET}") return True else: self.log.warning(f"IP MISMATCH: Wanted {self.target_ip}, got {got_ip}") # Check for Monitor Mode Success if self.regex_monitor_success.search(line): csi_msg = f" {Colors.CYAN}(CSI saved){Colors.RESET}" if csi_saved else "" self.log.info(f"{Colors.GREEN}SUCCESS: Monitor Mode Active{csi_msg}{Colors.RESET}") return True # Check for status command responses if self.regex_status_connected.search(line): csi_msg = f" {Colors.CYAN}(CSI saved){Colors.RESET}" if csi_saved else "" self.log.info(f"{Colors.GREEN}SUCCESS: Connected{csi_msg}{Colors.RESET}") return True # 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 confirm configuration") return False def parse_args(): parser = argparse.ArgumentParser( description='ESP32 Unified Deployment Tool', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Operation Modes: Default: Build + Flash + Configure --config-only: Configure only (no flashing) --flash-only: Build + Flash only (no configure) --flash-erase: Erase + Flash + Configure Examples: # Full deployment (flash + config) %(prog)s -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 # Config only (no flashing) %(prog)s -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 --config-only # Flash only (preserve config) %(prog)s --flash-only # Full erase + deploy %(prog)s -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 --flash-erase # CSI-enabled devices %(prog)s -s ClubHouse2G -P ez2remember --start-ip 192.168.1.111 --csi # Monitor mode %(prog)s --start-ip 192.168.1.90 -M MONITOR -mc 36 --config-only # 5GHz with 40MHz bandwidth %(prog)s -s ClubHouse2G -P ez2remember --start-ip 192.168.1.81 -b 5G -B HT40 """ ) # Operation Mode mode_group = parser.add_argument_group('Operation Mode') mode_group.add_argument('--config-only', action='store_true', help='Configure only (no flashing)') mode_group.add_argument('--flash-only', action='store_true', help='Flash only (no configure)') mode_group.add_argument('--flash-erase', action='store_true', help='Erase flash before flashing') # Build/Flash Options flash_group = parser.add_argument_group('Flash Options') flash_group.add_argument('-d', '--dir', default=os.getcwd(), help='Project directory (default: current)') flash_group.add_argument('-b', '--baud', type=int, default=460800, help='Flash baud rate (default: 460800)') flash_group.add_argument('--devices', type=str, help='Comma-separated list of devices (e.g., /dev/ttyUSB3,/dev/ttyUSB6,/dev/ttyUSB7) for selective deployment/retry. ' 'Automatically uses sequential flashing to avoid power issues.') flash_group.add_argument('--max-concurrent', type=int, default=None, help=f'Max concurrent flash operations (default: {DEFAULT_MAX_CONCURRENT_FLASH}). ' 'Defaults to 1 when using --devices, unless explicitly set. ' 'Lower this if you experience USB power issues. ' 'Try 2-3 for unpowered hubs, 4-8 for powered hubs.') # Network Configuration net_group = parser.add_argument_group('Network Configuration') net_group.add_argument('--start-ip', required=True, help='Starting static IP address') net_group.add_argument('-s', '--ssid', default='ClubHouse2G', help='WiFi SSID (default: ClubHouse2G)') net_group.add_argument('-P', '--password', default='ez2remember', help='WiFi password (default: ez2remember)') net_group.add_argument('-g', '--gateway', default='192.168.1.1', help='Gateway IP (default: 192.168.1.1)') net_group.add_argument('-m', '--netmask', default='255.255.255.0', help='Netmask (default: 255.255.255.0)') # WiFi Configuration wifi_group = parser.add_argument_group('WiFi Configuration') wifi_group.add_argument('--band', default='2.4G', choices=['2.4G', '5G'], help='WiFi band (default: 2.4G)') wifi_group.add_argument('-B', '--bandwidth', default='HT20', choices=['HT20', 'HT40', 'VHT80'], help='Channel bandwidth (default: HT20)') wifi_group.add_argument('-ps', '--powersave', default='NONE', help='Power save mode (default: NONE)') # Mode Configuration mode_config_group = parser.add_argument_group('Device Mode Configuration') mode_config_group.add_argument('-M', '--mode', default='STA', choices=['STA', 'MONITOR'], help='Operating mode (default: STA)') mode_config_group.add_argument('-mc', '--monitor-channel', type=int, default=36, help='Monitor mode channel (default: 36)') # Feature Flags feature_group = parser.add_argument_group('Feature Flags') feature_group.add_argument('--csi', dest='csi_enable', action='store_true', help='Enable CSI capture (default: disabled)') args = parser.parse_args() # Validation if args.config_only and args.flash_only: parser.error("Cannot use --config-only and --flash-only together") if args.flash_erase and args.config_only: parser.error("Cannot use --flash-erase with --config-only") if args.flash_erase and args.flash_only: parser.error("Cannot use --flash-erase with --flash-only (use default mode)") if not args.config_only and not args.flash_only: # Default mode or flash-erase mode if not args.ssid or not args.password: parser.error("SSID and password required for flash+config mode") return args def extract_device_number(device_path): """Extract numeric suffix from device path (e.g., /dev/ttyUSB3 -> 3)""" match = re.search(r'(\d+)$', device_path) if match: return int(match.group(1)) return 0 # Default to 0 if no number found async def run_deployment(args): """Main deployment orchestration""" # Determine operation mode if args.config_only: mode_str = f"{Colors.CYAN}CONFIG ONLY{Colors.RESET}" elif args.flash_only: mode_str = f"{Colors.YELLOW}FLASH ONLY{Colors.RESET}" elif args.flash_erase: mode_str = f"{Colors.RED}ERASE + FLASH + CONFIG{Colors.RESET}" else: mode_str = f"{Colors.GREEN}FLASH + CONFIG{Colors.RESET}" print(f"\n{Colors.BLUE}{'='*60}{Colors.RESET}") print(f" ESP32 Unified Deployment Tool") print(f" Operation Mode: {mode_str}") print(f"{Colors.BLUE}{'='*60}{Colors.RESET}\n") project_dir = Path(args.dir).resolve() build_dir = project_dir / 'build' # Phase 1: Build Firmware (if needed) if not args.config_only: print(f"{Colors.YELLOW}[1/3] Building Firmware...{Colors.RESET}") proc = await asyncio.create_subprocess_exec( 'idf.py', 'build', cwd=project_dir, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE ) _, stderr = await proc.communicate() if proc.returncode != 0: print(f"{Colors.RED}Build Failed:\n{stderr.decode()}{Colors.RESET}") return if not (build_dir / 'flash_args').exists(): print(f"{Colors.RED}Error: build/flash_args missing{Colors.RESET}") return print(f"{Colors.GREEN}Build Complete{Colors.RESET}") else: print(f"{Colors.CYAN}[1/3] Skipping Build (Config-Only Mode){Colors.RESET}") # Phase 2: Detect Devices step_num = 2 if not args.config_only else 1 print(f"{Colors.YELLOW}[{step_num}/3] Detecting Devices...{Colors.RESET}") # Get device list if args.devices: # User specified devices explicitly device_list = [d.strip() for d in args.devices.split(',')] print(f"{Colors.CYAN}Using specified devices: {', '.join(device_list)}{Colors.RESET}") # Create device objects for specified devices class SimpleDevice: def __init__(self, path): self.device = path devices = [SimpleDevice(d) for d in device_list] # Validate devices exist for dev in devices: if not os.path.exists(dev.device): print(f"{Colors.RED}Warning: Device {dev.device} not found{Colors.RESET}") else: # Auto-detect all devices devices = detect_esp32.detect_esp32_devices() if not devices: print(f"{Colors.RED}No devices found{Colors.RESET}") return # Sort devices naturally 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) print(f"{Colors.GREEN}Found {len(devices)} device{'s' if len(devices) != 1 else ''}{Colors.RESET}") # Validate start IP try: start_ip_obj = ipaddress.IPv4Address(args.start_ip) except: print(f"{Colors.RED}Invalid start IP: {args.start_ip}{Colors.RESET}") return # Phase 3: Deploy step_num = 3 if not args.config_only else 2 operation = "Configuring" if args.config_only else "Deploying to" print(f"{Colors.YELLOW}[{step_num}/3] {operation} {len(devices)} devices...{Colors.RESET}\n") # Show device-to-IP mapping when using --devices if args.devices and not args.flash_only: print(f"{Colors.CYAN}Device-to-IP Mapping:{Colors.RESET}") for dev in devices: dev_num = extract_device_number(dev.device) target_ip = str(start_ip_obj + dev_num) print(f" {dev.device} -> {target_ip}") print() # Determine flash concurrency if args.max_concurrent is not None: # Priority 1: User explicitly set a limit (honored always) max_concurrent = args.max_concurrent print(f"{Colors.CYAN}Flash Mode: {max_concurrent} concurrent operations (User Override){Colors.RESET}\n") elif args.devices and not args.config_only: # Priority 2: Retry/Specific mode defaults to sequential for safety max_concurrent = 1 print(f"{Colors.CYAN}Flash Mode: Sequential (retry mode default){Colors.RESET}\n") else: # Priority 3: Standard Bulk mode defaults to constant max_concurrent = DEFAULT_MAX_CONCURRENT_FLASH if not args.config_only: print(f"{Colors.CYAN}Flash Mode: {max_concurrent} concurrent operations{Colors.RESET}\n") flash_sem = asyncio.Semaphore(max_concurrent) tasks = [] # Map device to target IP # If --devices specified, use USB number as offset; otherwise use enumerate index device_ip_map = [] for i, dev in enumerate(devices): if args.devices: # Extract USB number and use as offset (e.g., ttyUSB3 -> offset 3) device_num = extract_device_number(dev.device) target_ip = str(start_ip_obj + device_num) device_ip_map.append((dev.device, target_ip, device_num)) else: # Use enumerate index as offset target_ip = str(start_ip_obj + i) device_ip_map.append((dev.device, target_ip, i)) worker = UnifiedDeployWorker(dev.device, target_ip, args, build_dir, flash_sem) tasks.append(worker.run()) # Execute all tasks concurrently results = await asyncio.gather(*tasks) # Phase 4: Summary success = results.count(True) failed = len(devices) - success # Track failed devices failed_devices = [] for i, (dev, result) in enumerate(zip(devices, results)): if not result: failed_devices.append(dev.device) # Determine feature status for summary features = [] # Add SSID (if configured) if not args.flash_only: features.append(f"SSID: {args.ssid}") if args.csi_enable: features.append(f"CSI: {Colors.GREEN}ENABLED{Colors.RESET}") else: features.append(f"CSI: DISABLED") if args.mode == 'MONITOR': features.append(f"Mode: MONITOR (Ch {args.monitor_channel})") else: features.append(f"Mode: STA") features.append(f"Band: {args.band}") features.append(f"BW: {args.bandwidth}") features.append(f"Power Save: {args.powersave}") print(f"\n{Colors.BLUE}{'='*60}{Colors.RESET}") print(f" Deployment Summary") print(f"{Colors.BLUE}{'='*60}{Colors.RESET}") print(f" Total Devices: {len(devices)}") print(f" Success: {Colors.GREEN}{success}{Colors.RESET}") print(f" Failed: {Colors.RED}{failed}{Colors.RESET}" if failed > 0 else f" Failed: {failed}") print(f"{Colors.BLUE}{'-'*60}{Colors.RESET}") for feature in features: print(f" {feature}") print(f"{Colors.BLUE}{'='*60}{Colors.RESET}") # Show failed devices and retry command if any failed if failed_devices: print(f"\n{Colors.RED}Failed Devices:{Colors.RESET}") for dev in failed_devices: # Show device and its intended IP dev_num = extract_device_number(dev) intended_ip = str(start_ip_obj + dev_num) print(f" {dev} -> {intended_ip}") # Generate retry command device_list = ','.join(failed_devices) retry_cmd = f"./esp32_deploy.py --devices {device_list}" # Add original arguments to retry command if args.ssid: retry_cmd += f" -s {args.ssid}" if args.password: retry_cmd += f" -P {args.password}" # Use original starting IP - device number extraction will handle the offset if not args.flash_only: retry_cmd += f" --start-ip {args.start_ip}" if args.band != '2.4G': retry_cmd += f" --band {args.band}" if args.bandwidth != 'HT20': retry_cmd += f" -B {args.bandwidth}" if args.powersave != 'NONE': retry_cmd += f" -ps {args.powersave}" if args.mode != 'STA': retry_cmd += f" -M {args.mode}" if args.monitor_channel != 36: retry_cmd += f" -mc {args.monitor_channel}" if args.csi_enable: retry_cmd += " --csi" if args.config_only: retry_cmd += " --config-only" elif args.flash_only: retry_cmd += " --flash-only" elif args.flash_erase: retry_cmd += " --flash-erase" if args.baud != 460800: retry_cmd += f" -b {args.baud}" if args.max_concurrent is not None: retry_cmd += f" --max-concurrent {args.max_concurrent}" print(f"\n{Colors.YELLOW}Retry Command:{Colors.RESET}") print(f" {retry_cmd}\n") else: print() # Extra newline for clean output def main(): args = parse_args() if os.name == 'nt': asyncio.set_event_loop(asyncio.ProactorEventLoop()) try: asyncio.run(run_deployment(args)) except KeyboardInterrupt: print(f"\n{Colors.YELLOW}Deployment cancelled by user{Colors.RESET}") sys.exit(1) if __name__ == '__main__': main()