src/python/binary_configurator.py

   1 #!/usr/bin/python
   2
   3 import os
   4 from random import randint
   5 import argparse
   6 import json
   7 from json import JSONEncoder
   8 import mmap
   9 import hashlib
  10 from enum import Enum
  11 import shutil
  12
  13 import firmware
  14 from firmware import DeviceType, FirmwareOptions, RadioType, MCUType, TXType
  15 import UnifiedConfiguration
  16 import binary_flash
  17 from binary_flash import UploadMethod
  18 from external import jmespath
  19
  20 class RegulatoryDomain(Enum):
  21     us_433 = 'us_433'
  22     us_433_wide = 'us_433_wide'
  23     eu_433 = 'eu_433'
  24     au_433 = 'au_433'
  25     in_866 = 'in_866'
  26     eu_868 = 'eu_868'
  27     au_915 = 'au_915'
  28     fcc_915 = 'fcc_915'
  29
  30     def __str__(self):
  31         return self.value
  32
  33 def generateUID(phrase):
  34     uid = [
  35         int(item) if item.isdigit() else -1
  36         for item in phrase.split(',')
  37     ]
  38     if (4 <= len(uid) <= 6) and all(ele >= 0 and ele < 256 for ele in uid):
  39         # Extend the UID to 6 bytes, as only 4 are needed to bind
  40         uid = [0] * (6 - len(uid)) + uid
  41         uid = bytes(uid)
  42     else:
  43         uid = hashlib.md5(("-DMY_BINDING_PHRASE=\""+phrase+"\"").encode()).digest()[0:6]
  44     return uid
  45
  46 def FREQ_HZ_TO_REG_VAL_SX127X(freq):
  47     return int(freq/61.03515625)
  48
  49 def FREQ_HZ_TO_REG_VAL_SX1280(freq):
  50     return int(freq/(52000000.0/pow(2,18)))
  51
  52 def domain_number(domain):
  53     if domain == RegulatoryDomain.au_915:
  54         return 0
  55     elif domain == RegulatoryDomain.fcc_915:
  56         return 1
  57     elif domain == RegulatoryDomain.eu_868:
  58         return 2
  59     elif domain == RegulatoryDomain.in_866:
  60         return 3
  61     elif domain == RegulatoryDomain.au_433:
  62         return 4
  63     elif domain == RegulatoryDomain.eu_433:
  64         return 5
  65     elif domain == RegulatoryDomain.us_433:
  66         return 6
  67     elif domain == RegulatoryDomain.us_433_wide:
  68         return 7
  69
  70 def patch_unified(args, options):
  71     json_flags = {}
  72     if args.phrase is not None:
  73         json_flags['uid'] = [x for x in generateUID(args.phrase)]
  74     if args.ssid is not None:
  75         json_flags['wifi-ssid'] = args.ssid
  76     if args.password is not None and args.ssid is not None:
  77         json_flags['wifi-password'] = args.password
  78     if args.auto_wifi is not None:
  79         json_flags['wifi-on-interval'] = args.auto_wifi
  80
  81     if args.tlm_report is not None:
  82         json_flags['tlm-interval'] = args.tlm_report
  83     if args.fan_min_runtime is not None:
  84         json_flags['fan-runtime'] = args.fan_min_runtime
  85
  86     if args.airport_baud is not None:
  87         json_flags['is-airport'] = True
  88         if options.deviceType is DeviceType.RX:
  89             json_flags['rcvr-uart-baud'] = args.airport_baud
  90         else:
  91             json_flags['airport-uart-baud'] = args.airport_baud
  92     elif args.rx_baud is not None:
  93         json_flags['rcvr-uart-baud'] = args.rx_baud
  94
  95     if args.lock_on_first_connection is not None:
  96         json_flags['lock-on-first-connection'] = args.lock_on_first_connection
  97
  98     if args.domain is not None:
  99         json_flags['domain'] = domain_number(args.domain)
 100
 101     json_flags['flash-discriminator'] = randint(1,2**32-1)
 102
 103     UnifiedConfiguration.doConfiguration(
 104         args.file,
 105         JSONEncoder().encode(json_flags),
 106         args.target,
 107         'tx' if options.deviceType is DeviceType.TX else 'rx',
 108         '2400' if options.radioChip is RadioType.SX1280 else '900' if options.radioChip is RadioType.SX127X else 'dual',
 109         '32' if options.mcuType is MCUType.ESP32 and options.deviceType is DeviceType.RX else '',
 110         options.luaName,
 111         args.rx_as_tx
 112     )
 113
 114 def length_check(l, f):
 115     def x(s):
 116         if (len(s) > l):
 117             raise argparse.ArgumentTypeError(f'too long, {l} chars max')
 118         else:
 119             return s
 120     return x
 121
 122 def ask_for_firmware(args):
 123     moduletype = 'tx' if args.tx else 'rx'
 124     with open('hardware/targets.json') as f:
 125         targets = json.load(f)
 126         products = []
 127         if args.target is not None:
 128             target = args.target
 129             config = jmespath.search('.'.join(map(lambda s: f'"{s}"', args.target.split('.'))), targets)
 130             if config is None and args.fdir is not None:
 131                 with open(os.path.join(args.fdir, 'hardware/targets.json')) as f:
 132                     targets = json.load(f)
 133                     config = jmespath.search('.'.join(map(lambda s: f'"{s}"', args.target.split('.'))), targets)
 134         else:
 135             i = 0
 136             for k in jmespath.search(f'*.["{moduletype}_2400","{moduletype}_900","{moduletype}_dual"][].*[]', targets):
 137                 i += 1
 138                 products.append(k)
 139                 print(f"{i}) {k['product_name']}")
 140             print('Choose a configuration to flash')
 141             choice = input()
 142             if choice != "":
 143                 config = products[int(choice)-1]
 144                 for v in targets:
 145                     for t in targets[v]:
 146                         if t != 'name':
 147                             for m in targets[v][t]:
 148                                 if targets[v][t][m]['product_name'] == config['product_name']:
 149                                     target = f'{v}.{t}.{m}'
 150     return target, config
 151
 152 class readable_dir(argparse.Action):
 153     def __call__(self, parser, namespace, values, option_string=None):
 154         prospective_dir=values
 155         if not os.path.isdir(prospective_dir):
 156             raise argparse.ArgumentTypeError("readable_dir:{0} is not a valid path".format(prospective_dir))
 157         if os.access(prospective_dir, os.R_OK):
 158             setattr(namespace,self.dest,prospective_dir)
 159         else:
 160             raise argparse.ArgumentTypeError("readable_dir:{0} is not a readable dir".format(prospective_dir))
 161
 162 class writeable_dir(argparse.Action):
 163     def __call__(self, parser, namespace, values, option_string=None):
 164         prospective_dir=values
 165         if not os.path.isdir(prospective_dir):
 166             raise argparse.ArgumentTypeError("readable_dir:{0} is not a valid path".format(prospective_dir))
 167         if os.access(prospective_dir, os.W_OK):
 168             setattr(namespace,self.dest,prospective_dir)
 169         else:
 170             raise argparse.ArgumentTypeError("readable_dir:{0} is not a writeable dir".format(prospective_dir))
 171
 172 class deprecate_action(argparse.Action):
 173     def __call__(self, parser, namespace, values, option_string=None):
 174         delattr(namespace, self.dest)
 175
 176 def main():
 177     parser = argparse.ArgumentParser(description="Configure Binary Firmware")
 178     # firmware/targets directory
 179     parser.add_argument('--dir', action=readable_dir, default=None, help='The directory that contains the "hardware" and other firmware directories')
 180     parser.add_argument('--fdir', action=readable_dir, default=None, help='If specified, then the firmware files are loaded from this directory')
 181     # Bind phrase
 182     parser.add_argument('--phrase', type=str, help='Your personal binding phrase')
 183     parser.add_argument('--flash-discriminator', type=int, default=randint(1,2**32-1), dest='flash_discriminator', help='Force a fixed flash-descriminator instead of random')
 184     # WiFi Params
 185     parser.add_argument('--ssid', type=length_check(32, "ssid"), required=False, help='Home network SSID')
 186     parser.add_argument('--password', type=length_check(64, "password"), required=False, help='Home network password')
 187     parser.add_argument('--auto-wifi', type=int, help='Interval (in seconds) before WiFi auto starts, if no connection is made')
 188     parser.add_argument('--no-auto-wifi', action='store_true', help='Disables WiFi auto start if no connection is made')
 189     # AirPort
 190     parser.add_argument('--airport-baud', type=int, const=None, nargs='?', action='store', help='If configured as an AirPort device then this is the baud rate to use')
 191     # RX Params
 192     parser.add_argument('--rx-baud', type=int, const=420000, nargs='?', action='store', help='The receiver baudrate talking to the flight controller')
 193     parser.add_argument('--lock-on-first-connection', dest='lock_on_first_connection', action='store_true', help='Lock RF mode on first connection')
 194     parser.add_argument('--no-lock-on-first-connection', dest='lock_on_first_connection', action='store_false', help='Do not lock RF mode on first connection')
 195     parser.set_defaults(lock_on_first_connection=None)
 196     # TX Params
 197     parser.add_argument('--tlm-report', type=int, const=240, nargs='?', action='store', help='The interval (in milliseconds) between telemetry packets')
 198     parser.add_argument('--fan-min-runtime', type=int, const=30, nargs='?', action='store', help='The minimum amount of time the fan should run for (in seconds) if it turns on')
 199     # Regulatory domain
 200     parser.add_argument('--domain', type=RegulatoryDomain, choices=list(RegulatoryDomain), default=None, help='For SX127X based devices, which regulatory domain is being used')
 201     # Unified target
 202     parser.add_argument('--target', type=str, help='Unified target JSON path')
 203     # Flashing options
 204     parser.add_argument("--flash", type=UploadMethod, choices=list(UploadMethod), help="Flashing Method")
 205     parser.add_argument("--erase", action='store_true', default=False, help="Full chip erase before flashing on ESP devices")
 206     parser.add_argument('--out', action=writeable_dir, default=None)
 207     parser.add_argument("--port", type=str, help="SerialPort or WiFi address to flash firmware to")
 208     parser.add_argument("--baud", type=int, default=0, help="Baud rate for serial communication")
 209     parser.add_argument("--force", action='store_true', default=False, help="Force upload even if target does not match")
 210     parser.add_argument("--confirm", action='store_true', default=False, help="Confirm upload if a mismatched target was previously uploaded")
 211     parser.add_argument("--tx", action='store_true', default=False, help="Flash a TX module, RX if not specified")
 212     parser.add_argument("--lbt", action='store_true', default=False, help="Use LBT firmware, default is FCC (only for 2.4GHz firmware)")
 213     parser.add_argument('--rx-as-tx', type=TXType, choices=list(TXType), required=False, default=None, help="Flash an RX module with TX firmware, either internal (full-duplex) or external (half-duplex)")
 214     # Deprecated options, left for backward compatibility
 215     parser.add_argument('--uart-inverted', action=deprecate_action, nargs=0, help='Deprecated')
 216     parser.add_argument('--no-uart-inverted', action=deprecate_action, nargs=0, help='Deprecated')
 217
 218     #
 219     # Firmware file to patch/configure
 220     parser.add_argument("file", nargs="?", type=argparse.FileType("r+b"))
 221
 222     args = parser.parse_args()
 223
 224     if args.dir is not None:
 225         os.chdir(args.dir)
 226
 227     if args.file is None:
 228         args.target, config = ask_for_firmware(args)
 229         try:
 230             file = config['firmware']
 231             if args.rx_as_tx is not None:
 232                 if config['platform'].startswith('esp32') or config['platform'].startswith('esp8285') and args.rx_as_tx == TXType.internal:
 233                     file = file.replace('_RX', '_TX')
 234                 else:
 235                     print("Selected device cannot operate as 'RX-as-TX' of this type.")
 236                     print("ESP8285 only supports full-duplex internal RX as TX.")
 237                     exit(1)
 238             firmware_dir = '' if args.fdir is None else args.fdir + '/'
 239             srcdir = firmware_dir + ('LBT/' if args.lbt else 'FCC/') + file
 240             dst = 'firmware.bin'
 241             shutil.copy2(srcdir + '/firmware.bin', ".")
 242             if os.path.exists(srcdir + '/bootloader.bin'): shutil.copy2(srcdir + '/bootloader.bin', ".")
 243             if os.path.exists(srcdir + '/partitions.bin'): shutil.copy2(srcdir + '/partitions.bin', ".")
 244             if os.path.exists(srcdir + '/boot_app0.bin'): shutil.copy2(srcdir + '/boot_app0.bin', ".")
 245             args.file = open(dst, 'r+b')
 246         except FileNotFoundError:
 247             print("Firmware files not found, did you download and unpack them in this directory?")
 248             exit(1)
 249     else:
 250         args.target, config = ask_for_firmware(args)
 251
 252     with args.file as f:
 253         mm = mmap.mmap(f.fileno(), 0)
 254
 255         pos = firmware.get_hardware(mm)
 256         options = FirmwareOptions(
 257             MCUType.ESP32 if config['platform'].startswith('esp32') else MCUType.ESP8266,
 258             DeviceType.RX if '.rx_' in args.target else DeviceType.TX,
 259             RadioType.SX127X if '_900.' in args.target else RadioType.SX1280 if '_2400.' in args.target else RadioType.LR1121,
 260             config['lua_name'] if 'lua_name' in config else '',
 261             config['stlink']['bootloader'] if 'stlink' in config else '',
 262             config['stlink']['offset'] if 'stlink' in config else 0,
 263             config['firmware']
 264         )
 265         patch_unified(args, options)
 266         args.file.close()
 267
 268         if options.mcuType == MCUType.ESP8266:
 269             import gzip
 270             with open(args.file.name, 'rb') as f_in:
 271                 with gzip.open('firmware.bin.gz', 'wb') as f_out:
 272                     shutil.copyfileobj(f_in, f_out)
 273
 274         if args.flash:
 275             args.target = config.get('firmware')
 276             args.accept = config.get('prior_target_name')
 277             args.platform = config.get('platform')
 278             return binary_flash.upload(options, args)
 279         elif 'upload_methods' in config and 'stock' in config['upload_methods']:
 280             shutil.copy(args.file.name, 'firmware.elrs')
 281     return 0
 282
 283 if __name__ == '__main__':
 284     try:
 285         exit(main())
 286     except AssertionError as e:
 287         print(e)
 288         exit(1)