el-cli.py

#!/data/data/com.termux/files/usr/bin/python

import serial
import time
import sys
import urllib.request
import argparse
import subprocess

ips = subprocess.run(
    ["sh", "-c", "ip neigh | awk '{print $1}'"],
    capture_output=True,
    text=True
).stdout.splitlines()

for ip in ips:
    print(ip)
    
parser = argparse.ArgumentParser(description="A custom Linux-like CLI tool")

parser.add_argument(
    "-i", 
    "--input", 
    required=True, 
    help="path to the binary file"
)


# 4. Parse the command-line arguments
args = parser.parse_args()

# 5. Save them into separate variables
input_file = args.input

# 6. Use the variables
print(f"Reading from: {input_file}")

PORT = 23
BIN_PATH = args.input
START_ADDR = 0x08000000

def hard_reset_via_esp(ip):
    """toggles GPIO via esp-link REST API."""
    print(f"\033[93m sending hard reset command via esp-link...\033[0m")
    try:
        url = f"http://{ip}/console/reset"
        req = urllib.request.Request(url, method='POST')
        with urllib.request.urlopen(req, timeout=3) as response:
            if response.status == 200:
                print(f"reset signal sent via esp-link reset. waiting for STM32 to boot into loader...\033[0m")
                time.sleep(1.5) 
                return True
    except Exception as e:
        print(f"\033[91m️ ESP-link reset HTTP request failed: {e}\033[0m")
    return False

def get_checksum(data):
    res = 0
    for b in data: res ^= b
    return res

def send_command(ser, cmd):
    """sends command, returns True if ACK (0x79) received."""
    ser.write(bytes([cmd, cmd ^ 0xFF]))
    res = ser.read(1)
    return res == b'\x79'

def sync_stm32(ser):
    """clear buffers and sends 0x7F until ACK is received."""
    print(f"sending sync (0x7F)...\033[0m")
    ser.reset_input_buffer()
    ser.reset_output_buffer()
    for _ in range(15):
        ser.write(b'\x7f')
        res = ser.read(1)
        if res in [b'\x79', b'\x1f']: # ACK or NACK (NACK means already synced)
            return True
        time.sleep(0.1)
    return False

def run_go_command(ser):
    """executes the go command to run the application."""
    print(f"executing 'go' command...\033[0m")
    if send_command(ser, 0x21):
        addr_bytes = START_ADDR.to_bytes(4, 'big')
        ser.write(addr_bytes + bytes([get_checksum(addr_bytes)]))
        if ser.read(1) == b'\x79':
            print(f"\033[89m code execution starts\033[0m")
            return True
        else:
            print(f"\033[91m go address NACK.\033[0m")
    else:
        print(f"\033[91m go command rejected.\033[0m")
    return False

# --- main script ---
ser = None
try:
    synced = False
    max_attempts = 1

    for attempt in range(1, max_attempts + 1):
        print(f"\n\033[93m sync attempt {attempt}/{max_attempts}\033[0m")
        
        # 0. automate bootloader hardware jump via esp-link
        hard_reset_via_esp(ip)

        # 1. start/restart serial socket protocol
        if ser is not None:
            ser.close()
            
        print(f"connecting to {ip}:{PORT} (8E1 Parity)\033[0m")
        ser = serial.serial_for_url(f"socket://{ip}:{PORT}", timeout=3, parity='E')
        
        if sync_stm32(ser):
            print(f"\033[92m synced successfully\033[0m")
            synced = True
            break
        else:
            print(f"\033[91m️ sync attempt {attempt} failed.\033[0m")
            if attempt < max_attempts:
                print(f"waiting 2 seconds before retrying\033[0m")
                time.sleep(2)

    if not synced:
        print(f"\n\033[91m failed: all 3 sync trys are failed.\ncheck physical connections: BOOT0=1, wiring connections, and esp-link parity configuration.\033[0m")
        sys.exit(1)

    # 2. get ID to verify 
    if send_command(ser, 0x02):
        length = ord(ser.read(1))
        pid = ser.read(length + 1).hex().upper()
        ser.read(1) # Final ACK
        print(f"\033[92m connected to ID: 0x{pid}\033[0m")
    else:
        print(f"\033[91m️ command 0x02 rejected.\033[93m attempting Read-Unprotect...\033[0m")
        if send_command(ser, 0x92):
            print(f"Read-Unprotect accepted. Chip wiping... wait <5s to reset.\033[0m")
            time.sleep(5)
            if not sync_stm32(ser):
                print(f"\033[91m re-sync failed after unprotect reset.\033[0m")
                sys.exit(1)
        else:
            print(f"\033[91m failed: command rejected. check parity/wiring.\033[0m")
            sys.exit(1)

    # selection after sync successful 
    print(f"esp-link-cli tool options\n1. Flash + Go\n2. Go\n3. quit")
    

    choice = input("select an option (1-3): \033[0m").strip()

    if choice == '3' or choice.lower() == 'abort':
        print(f"\033[91m aborted by user.\033[0m")
        sys.exit(0)
    elif choice not in ['1', '2']:
        print(f"\033[91m no valid selection. exiting.\033[0m")
        sys.exit(1)

    # --- option 1: flash + go ---
    if choice == '1':
        with open(BIN_PATH, "rb") as f:
            content = f.read()

        # erasing 
        print(f"erasing flash...\033[0m")
        if send_command(ser, 0x43): 
            ser.write(b'\xff\x00') # Mass Erase
            if ser.read(1) == b'\x79':
                print(f"\033[92m erasing complete.\033[0m")
            else:
                print(f"\033[91m erasing failed.\033[0m")
                sys.exit(1)

        # writing 
        print(f"\033[89m writing {len(content)} bytes...\033[0m")
        for i in range(0, len(content), 256):
            chunk = content[i:i+256]
            curr_addr = START_ADDR + i
            
            if not send_command(ser, 0x31):
                print(f"\n\033[91m write command rejected at {hex(curr_addr)}\033[0m")
                sys.exit(1)
            
            # address + checksum
            addr_bytes = curr_addr.to_bytes(4, 'big')
            ser.write(addr_bytes + bytes([get_checksum(addr_bytes)]))
            if ser.read(1) != b'\x79':
                print(f"\n\033[91m address NACK at {hex(curr_addr)}\033[0m")
                sys.exit(1)
                
            # data length (N-1) + data + checksum
            d_len = len(chunk) - 1
            data_checksum = d_len ^ get_checksum(chunk)
            ser.write(bytes([d_len]) + chunk + bytes([data_checksum]))
            
            if ser.read(1) != b'\x79':
                print(f"\n\033[91m data NACK at {hex(curr_addr)}\033[0m")
                sys.exit(1)
                
            print(f"progress: {i + len(chunk)}/{len(content)} bytes", end='\r')

        print(f"\n\033[92m flash successful \033[0m")
        run_go_command(ser)

    # --- option 2: go only ---
    elif choice == '2':
     
        run_go_command(ser)

except Exception as e:
    print(f"\n\033[91m error: {e}\033[0m")
finally:
    if ser is not None: 
        ser.close()
        print(f"connection closed.\033[0m")