def detect_chip(port=DEFAULT_PORT, baud=ESP_ROM_BAUD, connect_mode='default_reset'):
""" Use serial access to detect the chip type.
We use the UART's datecode register for this, it's mapped at
the same address on ESP8266 & ESP32 so we can use one
memory read and compare to the datecode register for each chip
type.
This routine automatically performs ESPLoader.connect() (passing
connect_mode parameter) as part of querying the chip.
"""
detect_port = ESPLoader(port, baud)
detect_port.connect(connect_mode)
print('Detecting chip type...', end='')
sys.stdout.flush()
date_reg = detect_port.read_reg(ESPLoader.UART_DATA_REG_ADDR)
for cls in [ESP8266ROM, ESP32ROM]:
if date_reg == cls.DATE_REG_VALUE:
# don't connect a second time
inst = cls(detect_port._port, baud)
print(' %s' % inst.CHIP_NAME)
return inst
print('')
raise FatalError("Unexpected UART datecode value 0x%08x. Failed to autodetect chip type." % date_reg)
python类read()的实例源码
def command(self, op=None, data=b"", chk=0, wait_response=True):
if op is not None:
pkt = struct.pack(b'<BBHI', 0x00, op, len(data), chk) + data
self.write(pkt)
if not wait_response:
return
# tries to get a response until that response has the
# same operation as the request or a retries limit has
# exceeded. This is needed for some esp8266s that
# reply with more sync responses than expected.
for retry in range(100):
p = self.read()
if len(p) < 8:
continue
(resp, op_ret, len_ret, val) = struct.unpack('<BBHI', p[:8])
if resp != 1:
continue
data = p[8:]
if op is None or op_ret == op:
return val, data
raise FatalError("Response doesn't match request")
def LoadFirmwareImage(chip, filename):
""" Load a firmware image. Can be for ESP8266 or ESP32. ESP8266 images will be examined to determine if they are
original ROM firmware images (ESPFirmwareImage) or "v2" OTA bootloader images.
Returns a BaseFirmwareImage subclass, either ESPFirmwareImage (v1) or OTAFirmwareImage (v2).
"""
with open(filename, 'rb') as f:
if chip == 'esp32':
return ESP32FirmwareImage(f)
else: # Otherwise, ESP8266 so look at magic to determine the image type
magic = ord(f.read(1))
f.seek(0)
if magic == ESPLoader.ESP_IMAGE_MAGIC:
return ESPFirmwareImage(f)
elif magic == ESPBOOTLOADER.IMAGE_V2_MAGIC:
return OTAFirmwareImage(f)
else:
raise FatalError("Invalid image magic number: %d" % magic)
def __init__(self, load_file=None):
super(ESP32FirmwareImage, self).__init__()
self.flash_mode = 0
self.flash_size_freq = 0
self.version = 1
if load_file is not None:
segments = self.load_common_header(load_file, ESPLoader.ESP_IMAGE_MAGIC)
additional_header = list(struct.unpack("B" * 16, load_file.read(16)))
# check these bytes are unused
if additional_header != [0] * 16:
print("WARNING: ESP32 image header contains unknown flags. Possibly this image is from a newer version of esptool.py")
for _ in range(segments):
self.load_segment(load_file)
self.checksum = self.read_checksum(load_file)
def _read_elf_file(self, f):
# read the ELF file header
LEN_FILE_HEADER = 0x34
try:
(ident,_type,machine,_version,
self.entrypoint,_phoff,shoff,_flags,
_ehsize, _phentsize,_phnum,_shentsize,
_shnum,shstrndx) = struct.unpack("<16sHHLLLLLHHHHHH", f.read(LEN_FILE_HEADER))
except struct.error as e:
raise FatalError("Failed to read a valid ELF header from %s: %s" % (self.name, e))
if byte(ident, 0) != 0x7f or ident[1:4] != b'ELF':
raise FatalError("%s has invalid ELF magic header" % self.name)
if machine != 0x5e:
raise FatalError("%s does not appear to be an Xtensa ELF file. e_machine=%04x" % (self.name, machine))
self._read_sections(f, shoff, shstrndx)
def expand_file_arguments():
""" Any argument starting with "@" gets replaced with all values read from a text file.
Text file arguments can be split by newline or by space.
Values are added "as-is", as if they were specified in this order on the command line.
"""
new_args = []
expanded = False
for arg in sys.argv:
if arg.startswith("@"):
expanded = True
with open(arg[1:],"r") as f:
for line in f.readlines():
new_args += shlex.split(line)
else:
new_args.append(arg)
if expanded:
print("esptool.py %s" % (" ".join(new_args[1:])))
sys.argv = new_args
def print_incoming_text(self):
"""
When starting the connection, print all the debug data until
we get to a line with the end sequence '$$$'.
"""
line = ''
#Wait for device to send data
time.sleep(1)
if self.ser.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = self.ser.read().decode('utf-8')
line += c
print(line);
else:
self.warn("No Message")
def openbci_id(self, serial):
"""
When automatically detecting port, parse the serial return for the "OpenBCI" ID.
"""
line = ''
#Wait for device to send data
time.sleep(2)
if serial.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = serial.read().decode('utf-8')
line += c
if "OpenBCI" in line:
return True
return False
def command(self, op=None, data=None, chk=0):
if op is not None:
pkt = struct.pack(b'<BBHI', 0x00, op, len(data), chk) + data
self.write(pkt)
# tries to get a response until that response has the
# same operation as the request or a retries limit has
# exceeded. This is needed for some esp8266s that
# reply with more sync responses than expected.
for retry in range(100):
p = self.read()
if len(p) < 8:
continue
(resp, op_ret, len_ret, val) = struct.unpack('<BBHI', p[:8])
if resp != 1:
continue
body = p[8:]
if op is None or op_ret == op:
return val, body # valid response received
raise FatalError("Response doesn't match request")
def __init__(self, load_file=None):
super(ESPFirmwareImage, self).__init__()
self.flash_mode = 0
self.flash_size_freq = 0
self.version = 1
if load_file is not None:
(magic, segments, self.flash_mode, self.flash_size_freq, self.entrypoint) = struct.unpack('<BBBBI', load_file.read(8))
# some sanity check
if magic != ESPROM.ESP_IMAGE_MAGIC or segments > 16:
raise FatalError('Invalid firmware image magic=%d segments=%d' % (magic, segments))
for i in range(segments):
self.load_segment(load_file)
self.checksum = self.read_checksum(load_file)
def flash_digest(self, addr, length, digest_block_size=0):
self._esp.write(struct.pack(b'<B', self.CMD_FLASH_DIGEST))
self._esp.write(struct.pack(b'<III', addr, length, digest_block_size))
digests = []
while True:
p = self._esp.read()
if len(p) == 16:
digests.append(p)
elif len(p) == 1:
status_code = struct.unpack('<B', p)[0]
if status_code != 0:
raise FatalError('Write failure, status: %x' % status_code)
break
else:
raise FatalError('Unexpected packet: %s' % hexify(p))
return digests[-1], digests[:-1]
def write_flash(esp, args):
flash_params = _get_flash_params(esp, args)
flasher = CesantaFlasher(esp, args.baud)
for address, argfile in args.addr_filename:
image = argfile.read()
argfile.seek(0) # rewind in case we need it again
if address + len(image) > int(args.flash_size.split('m')[0]) * (1 << 17):
print('WARNING: Unlikely to work as data goes beyond end of flash. Hint: Use --flash_size')
image = _update_image_flash_params(address, flash_params, image)
# Pad to sector size, which is the minimum unit of writing (erasing really).
if len(image) % esp.ESP_FLASH_SECTOR != 0:
image += b'\xff' * (esp.ESP_FLASH_SECTOR - (len(image) % esp.ESP_FLASH_SECTOR))
t = time.time()
flasher.flash_write(address, image, not args.no_progress)
t = time.time() - t
print('\rWrote %d bytes at 0x%x in %.1f seconds (%.1f kbit/s)...'
% (len(image), address, t, len(image) / t * 8 / 1000))
print('Leaving...')
if args.verify:
print('Verifying just-written flash...')
_verify_flash(esp, args, flasher)
flasher.boot_fw()
def detect_chip(port=DEFAULT_PORT, baud=ESP_ROM_BAUD, connect_mode='default_reset', trace_enabled=False):
""" Use serial access to detect the chip type.
We use the UART's datecode register for this, it's mapped at
the same address on ESP8266 & ESP32 so we can use one
memory read and compare to the datecode register for each chip
type.
This routine automatically performs ESPLoader.connect() (passing
connect_mode parameter) as part of querying the chip.
"""
detect_port = ESPLoader(port, baud, trace_enabled=trace_enabled)
detect_port.connect(connect_mode)
print('Detecting chip type...', end='')
sys.stdout.flush()
date_reg = detect_port.read_reg(ESPLoader.UART_DATA_REG_ADDR)
for cls in [ESP8266ROM, ESP32ROM]:
if date_reg == cls.DATE_REG_VALUE:
# don't connect a second time
inst = cls(detect_port._port, baud, trace_enabled=trace_enabled)
print(' %s' % inst.CHIP_NAME)
return inst
print('')
raise FatalError("Unexpected UART datecode value 0x%08x. Failed to autodetect chip type." % date_reg)
def LoadFirmwareImage(chip, filename):
""" Load a firmware image. Can be for ESP8266 or ESP32. ESP8266 images will be examined to determine if they are
original ROM firmware images (ESPFirmwareImage) or "v2" OTA bootloader images.
Returns a BaseFirmwareImage subclass, either ESPFirmwareImage (v1) or OTAFirmwareImage (v2).
"""
with open(filename, 'rb') as f:
if chip == 'esp32':
return ESP32FirmwareImage(f)
else: # Otherwise, ESP8266 so look at magic to determine the image type
magic = ord(f.read(1))
f.seek(0)
if magic == ESPLoader.ESP_IMAGE_MAGIC:
return ESPFirmwareImage(f)
elif magic == ESPBOOTLOADER.IMAGE_V2_MAGIC:
return OTAFirmwareImage(f)
else:
raise FatalError("Invalid image magic number: %d" % magic)
def _read_elf_file(self, f):
# read the ELF file header
LEN_FILE_HEADER = 0x34
try:
(ident,_type,machine,_version,
self.entrypoint,_phoff,shoff,_flags,
_ehsize, _phentsize,_phnum, shentsize,
shnum,shstrndx) = struct.unpack("<16sHHLLLLLHHHHHH", f.read(LEN_FILE_HEADER))
except struct.error as e:
raise FatalError("Failed to read a valid ELF header from %s: %s" % (self.name, e))
if byte(ident, 0) != 0x7f or ident[1:4] != b'ELF':
raise FatalError("%s has invalid ELF magic header" % self.name)
if machine != 0x5e:
raise FatalError("%s does not appear to be an Xtensa ELF file. e_machine=%04x" % (self.name, machine))
if shentsize != self.LEN_SEC_HEADER:
raise FatalError("%s has unexpected section header entry size 0x%x (not 0x28)" % (self.name, shentsize, self.LEN_SEC_HEADER))
if shnum == 0:
raise FatalError("%s has 0 section headers" % (self.name))
self._read_sections(f, shoff, shnum, shstrndx)
def expand_file_arguments():
""" Any argument starting with "@" gets replaced with all values read from a text file.
Text file arguments can be split by newline or by space.
Values are added "as-is", as if they were specified in this order on the command line.
"""
new_args = []
expanded = False
for arg in sys.argv:
if arg.startswith("@"):
expanded = True
with open(arg[1:],"r") as f:
for line in f.readlines():
new_args += shlex.split(line)
else:
new_args.append(arg)
if expanded:
print("esptool.py %s" % (" ".join(new_args[1:])))
sys.argv = new_args
def command(self, op=None, data=None, chk=0):
if op is not None:
pkt = struct.pack('<BBHI', 0x00, op, len(data), chk) + data
self.write(pkt)
# tries to get a response until that response has the
# same operation as the request or a retries limit has
# exceeded. This is needed for some esp8266s that
# reply with more sync responses than expected.
for retry in xrange(100):
p = self.read()
if len(p) < 8:
continue
(resp, op_ret, len_ret, val) = struct.unpack('<BBHI', p[:8])
if resp != 1:
continue
body = p[8:]
if op is None or op_ret == op:
return val, body # valid response received
raise FatalError("Response doesn't match request")
def __init__(self, load_file=None):
super(ESPFirmwareImage, self).__init__()
self.flash_mode = 0
self.flash_size_freq = 0
self.version = 1
if load_file is not None:
(magic, segments, self.flash_mode, self.flash_size_freq, self.entrypoint) = struct.unpack('<BBBBI', load_file.read(8))
# some sanity check
if magic != ESPROM.ESP_IMAGE_MAGIC or segments > 16:
raise FatalError('Invalid firmware image magic=%d segments=%d' % (magic, segments))
for i in xrange(segments):
self.load_segment(load_file)
self.checksum = self.read_checksum(load_file)
def flash_digest(self, addr, length, digest_block_size=0):
self._esp.write(struct.pack('<B', self.CMD_FLASH_DIGEST))
self._esp.write(struct.pack('<III', addr, length, digest_block_size))
digests = []
while True:
p = self._esp.read()
if len(p) == 16:
digests.append(p)
elif len(p) == 1:
status_code = struct.unpack('<B', p)[0]
if status_code != 0:
raise FatalError('Write failure, status: %x' % status_code)
break
else:
raise FatalError('Unexpected packet: %s' % hexify(p))
return digests[-1], digests[:-1]
def detect_chip(port=DEFAULT_PORT, baud=ESP_ROM_BAUD, connect_mode='default_reset'):
""" Use serial access to detect the chip type.
We use the UART's datecode register for this, it's mapped at
the same address on ESP8266 & ESP32 so we can use one
memory read and compare to the datecode register for each chip
type.
This routine automatically performs ESPLoader.connect() (passing
connect_mode parameter) as part of querying the chip.
"""
detect_port = ESPLoader(port, baud)
detect_port.connect(connect_mode)
print('Detecting chip type...', end='')
sys.stdout.flush()
date_reg = detect_port.read_reg(ESPLoader.UART_DATA_REG_ADDR)
for cls in [ESP8266ROM, ESP32ROM]:
if date_reg == cls.DATE_REG_VALUE:
# don't connect a second time
inst = cls(detect_port._port, baud)
print(' %s' % inst.CHIP_NAME)
return inst
print('')
raise FatalError("Unexpected UART datecode value 0x%08x. Failed to autodetect chip type." % date_reg)
def command(self, op=None, data=b"", chk=0, wait_response=True):
if op is not None:
pkt = struct.pack(b'<BBHI', 0x00, op, len(data), chk) + data
self.write(pkt)
if not wait_response:
return
# tries to get a response until that response has the
# same operation as the request or a retries limit has
# exceeded. This is needed for some esp8266s that
# reply with more sync responses than expected.
for retry in range(100):
p = self.read()
if len(p) < 8:
continue
(resp, op_ret, len_ret, val) = struct.unpack('<BBHI', p[:8])
if resp != 1:
continue
data = p[8:]
if op is None or op_ret == op:
return val, data
raise FatalError("Response doesn't match request")
def LoadFirmwareImage(chip, filename):
""" Load a firmware image. Can be for ESP8266 or ESP32. ESP8266 images will be examined to determine if they are
original ROM firmware images (ESPFirmwareImage) or "v2" OTA bootloader images.
Returns a BaseFirmwareImage subclass, either ESPFirmwareImage (v1) or OTAFirmwareImage (v2).
"""
with open(filename, 'rb') as f:
if chip == 'esp32':
return ESP32FirmwareImage(f)
else: # Otherwise, ESP8266 so look at magic to determine the image type
magic = ord(f.read(1))
f.seek(0)
if magic == ESPLoader.ESP_IMAGE_MAGIC:
return ESPFirmwareImage(f)
elif magic == ESPBOOTLOADER.IMAGE_V2_MAGIC:
return OTAFirmwareImage(f)
else:
raise FatalError("Invalid image magic number: %d" % magic)
def __init__(self, load_file=None):
super(ESP32FirmwareImage, self).__init__()
self.flash_mode = 0
self.flash_size_freq = 0
self.version = 1
if load_file is not None:
segments = self.load_common_header(load_file, ESPLoader.ESP_IMAGE_MAGIC)
additional_header = list(struct.unpack("B" * 16, load_file.read(16)))
# check these bytes are unused
if additional_header != [0] * 16:
print("WARNING: ESP32 image header contains unknown flags. Possibly this image is from a newer version of esptool.py")
for _ in range(segments):
self.load_segment(load_file)
self.checksum = self.read_checksum(load_file)
def _read_elf_file(self, f):
# read the ELF file header
LEN_FILE_HEADER = 0x34
try:
(ident,_type,machine,_version,
self.entrypoint,_phoff,shoff,_flags,
_ehsize, _phentsize,_phnum,_shentsize,
_shnum,shstrndx) = struct.unpack("<16sHHLLLLLHHHHHH", f.read(LEN_FILE_HEADER))
except struct.error as e:
raise FatalError("Failed to read a valid ELF header from %s: %s" % (self.name, e))
if byte(ident, 0) != 0x7f or ident[1:4] != b'ELF':
raise FatalError("%s has invalid ELF magic header" % self.name)
if machine != 0x5e:
raise FatalError("%s does not appear to be an Xtensa ELF file. e_machine=%04x" % (self.name, machine))
self._read_sections(f, shoff, shstrndx)
def expand_file_arguments():
""" Any argument starting with "@" gets replaced with all values read from a text file.
Text file arguments can be split by newline or by space.
Values are added "as-is", as if they were specified in this order on the command line.
"""
new_args = []
expanded = False
for arg in sys.argv:
if arg.startswith("@"):
expanded = True
with open(arg[1:],"r") as f:
for line in f.readlines():
new_args += shlex.split(line)
else:
new_args.append(arg)
if expanded:
print("esptool.py %s" % (" ".join(new_args[1:])))
sys.argv = new_args
def print_incoming_text(self):
"""
When starting the connection, print all the debug data until
we get to a line with the end sequence '$$$'.
"""
line = ''
#Wait for device to send data
time.sleep(1)
if self.ser.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = self.ser.read().decode('utf-8')
line += c
if "On Daisy" in line:
self.daisy = True
print(line);
else:
self.warn("No Message")
def openbci_id(self, serial):
"""
When automatically detecting port, parse the serial return for the "OpenBCI" ID.
Also auto-detects the daisy.
"""
board = False
line = ''
#Wait for device to send data
time.sleep(2)
if serial.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = serial.read().decode('utf-8')
line += c
if "OpenBCI" in line:
return True
# if "On Daisy" in line:
# self.daisy = True
return False
def daisy_id(self):
"""
When automatically detecting port, parse the serial return for the "OpenBCI" ID.
"""
line = ''
#Wait for device to send data
# time.sleep(2)
# if self.ser.inWaiting():
# line = ''
# c = ''
# #Look for end sequence $$$
# while '$$$' not in line:
# c = self.ser.read().decode('utf-8')
# line += c
# if "On Daisy" in line:
# return True
return False
def print_incoming_text(self,print_enable):
"""
When starting the connection, print all the debug data until
we get to a line with the end sequence '$$$'.
"""
line = ''
#Wait for device to send data
time.sleep(1)
if self.ser.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = self.ser.read().decode('utf-8', errors='replace')
line += c
if "On Daisy" in line:
self.daisy = True
if print_enable:
print(line);
# else:
# self.warn("No Message")
def openbci_id(self, serial):
"""
When automatically detecting port, parse the serial return for the "OpenBCI" ID.
Also auto-detects the daisy.
"""
board = False
line = ''
#Wait for device to send data
time.sleep(2)
if serial.inWaiting():
line = ''
c = ''
#Look for end sequence $$$
while '$$$' not in line:
c = serial.read().decode('utf-8', errors='replace')
line += c
if "OpenBCI" in line:
return True
return False
