#!/usr/bin/env python # coding=utf-8 # ------------------------------------------------------------------------------- # ESPurna OTA manager # xose.perez@gmail.com # # Requires PlatformIO Core # ------------------------------------------------------------------------------- from __future__ import print_function import argparse import os import re import shutil import socket import subprocess import sys import time from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf try: # noinspection PyUnresolvedReferences input = raw_input # Python2 *!! redefining build-in input. except NameError: pass # Python3 # ------------------------------------------------------------------------------- DISCOVER_TIMEOUT = 2 description = "ESPurna OTA Manager v0.3" devices = [] discover_last = 0 # ------------------------------------------------------------------------------- def on_service_state_change(zeroconf, service_type, name, state_change): """ Callback that adds discovered devices to "devices" list """ global discover_last if state_change is ServiceStateChange.Added: discover_last = time.time() info = zeroconf.get_service_info(service_type, name) if info: hostname = info.server.split(".")[0] device = { 'hostname': hostname.upper(), 'ip': socket.inet_ntoa(info.address), 'mac': '', 'app_name': '', 'app_version': '', 'target_board': '', 'mem_size': 0, 'sdk_size': 0, 'free_space': 0, } for key, item in info.properties.items(): device[key.decode('UTF-8')] = item.decode('UTF-8') # rename fields (needed for sorting by name) device['app'] = device['app_name'] device['device'] = device['target_board'] device['version'] = device['app_version'] devices.append(device) def list_devices(): """ Shows the list of discovered devices """ output_format = "{:>3} {:<14} {:<15} {:<17} {:<12} {:<12} {:<25} {:<8} {:<8} {:<10}" print(output_format.format( "#", "HOSTNAME", "IP", "MAC", "APP", "VERSION", "DEVICE", "MEM_SIZE", "SDK_SIZE", "FREE_SPACE" )) print("-" * 139) index = 0 for device in devices: index += 1 print(output_format.format( index, device.get('hostname', ''), device.get('ip', ''), device.get('mac', ''), device.get('app_name', ''), device.get('app_version', ''), device.get('target_board', ''), device.get('mem_size', 0), device.get('sdk_size', 0), device.get('free_space', 0), )) print() def get_boards(): """ Grabs board types fro hardware.h file """ boards = [] for line in open("espurna/config/hardware.h"): m = re.search(r'defined\((\w*)\)', line) if m: boards.append(m.group(1)) return sorted(boards) def get_device_size(device): if device.get('mem_size', 0) == device.get('sdk_size', 0): return int(device.get('mem_size', 0)) / 1024 return 0 def get_empty_board(): """ Returns the empty structure of a board to flash """ board = {'board': '', 'ip': '', 'size': 0, 'auth': '', 'flags': ''} return board def get_board_by_index(index): """ Returns the required data to flash a given board """ board = {} if 1 <= index <= len(devices): device = devices[index - 1] board['hostname'] = device.get('hostname') board['board'] = device.get('target_board', '') board['ip'] = device.get('ip', '') board['size'] = get_device_size(device) return board def get_board_by_mac(mac): """ Returns the required data to flash a given board """ for device in devices: if device.get('mac', '').lower() == mac: board = {} board['hostname'] = device.get('hostname') board['board'] = device.get('device') board['ip'] = device.get('ip') board['size'] = get_device_size(device) if not board['board'] or not board['ip'] or board['size'] == 0: return None return board return None def get_board_by_hostname(hostname): """ Returns the required data to flash a given board """ hostname = hostname.lower() for device in devices: if device.get('hostname', '').lower() == hostname: board = {} board['hostname'] = device.get('hostname') board['board'] = device.get('target_board') board['ip'] = device.get('ip') board['size'] = get_device_size(device) if not board['board'] or not board['ip'] or board['size'] == 0: return None return board return None def input_board(): """ Grabs info from the user about what device to flash """ # Choose the board try: index = int(input("Choose the board you want to flash (empty if none of these): ")) except ValueError: index = 0 if index < 0 or len(devices) < index: print("Board number must be between 1 and {}\n".format(str(len(devices)))) return None board = get_board_by_index(index) # Choose board type if none before if len(board.get('board', '')) == 0: print() count = 1 boards = get_boards() for name in boards: print("{:3d}\t{}".format(count, name)) count += 1 print() try: index = int(input("Choose the board type you want to flash: ")) except ValueError: index = 0 if index < 1 or len(boards) < index: print("Board number must be between 1 and {}\n".format(str(len(boards)))) return None board['board'] = boards[index - 1] # Choose board size of none before if board.get('size', 0) == 0: try: board['size'] = int(input("Board memory size (1 for 1M, 2 for 2M, 4 for 4M): ")) except ValueError: print("Wrong memory size") return None # Choose IP of none before if len(board.get('ip', '')) == 0: board['ip'] = input("IP of the device to flash (empty for 192.168.4.1): ") or "192.168.4.1" return board def boardname(board): return board.get('hostname', board['ip']) def store(device, env): source = ".pioenvs/{}/firmware.elf".format(env) destination = ".pioenvs/elfs/{}.elf".format(boardname(device).lower()) dst_dir = os.path.dirname(destination) if not os.path.exists(dst_dir): os.mkdir(dst_dir) shutil.move(source, destination) def run(device, env): print("Building and flashing image over-the-air...") environ = os.environ.copy() environ["ESPURNA_IP"] = device["ip"] environ["ESPURNA_BOARD"] = device["board"] environ["ESPURNA_AUTH"] = device["auth"] environ["ESPURNA_FLAGS"] = device["flags"] command = ("platformio", "run", "--silent", "--environment", env, "-t", "upload") subprocess.check_call(command, env=environ) store(device, env) # ------------------------------------------------------------------------------- if __name__ == '__main__': # Parse command line options parser = argparse.ArgumentParser(description=description) parser.add_argument("-c", "--core", help="flash ESPurna core", default=0, action='count') parser.add_argument("-f", "--flash", help="flash device", default=0, action='count') parser.add_argument("-o", "--flags", help="extra flags", default='') parser.add_argument("-p", "--password", help="auth password", default='') parser.add_argument("-s", "--sort", help="sort devices list by field", default='hostname') parser.add_argument("-y", "--yes", help="do not ask for confirmation", default=0, action='count') parser.add_argument("hostnames", nargs='*', help="Hostnames to update") args = parser.parse_args() print() print(description) print() # Look for services zeroconf = Zeroconf() browser = ServiceBrowser(zeroconf, "_arduino._tcp.local.", handlers=[on_service_state_change]) discover_last = time.time() while time.time() < discover_last + DISCOVER_TIMEOUT: pass # zeroconf.close() if len(devices) == 0: print("Nothing found!\n") sys.exit(0) # Sort list field = args.sort.lower() if field not in devices[0]: print("Unknown field '{}'\n".format(field)) sys.exit(1) devices = sorted(devices, key=lambda device: device.get(field, '')) # List devices list_devices() # Flash device if args.flash > 0: # Board(s) to flash queue = [] # Check if hostnames for hostname in args.hostnames: board = get_board_by_hostname(hostname) if board: board['auth'] = args.password board['flags'] = args.flags queue.append(board) # If no boards ask the user if len(queue) == 0: board = input_board() if board: board['auth'] = args.password or input("Authorization key of the device to flash: ") board['flags'] = args.flags or input("Extra flags for the build: ") queue.append(board) # If still no boards quit if len(queue) == 0: sys.exit(0) queue = sorted(queue, key=lambda device: device.get('board', '')) # Flash each board for board in queue: # Flash core version? if args.core > 0: board['flags'] = "-DESPURNA_CORE " + board['flags'] env = "esp8266-{:d}m-ota".format(board['size']) # Summary print() print("HOST = {}".format(boardname(board))) print("IP = {}".format(board['ip'])) print("BOARD = {}".format(board['board'])) print("AUTH = {}".format(board['auth'])) print("FLAGS = {}".format(board['flags'])) print("ENV = {}".format(env)) response = True if args.yes == 0: response = (input("\nAre these values right [y/N]: ") == "y") if response: print() run(board, env)