mirror of
https://github.com/espressif/esp-idf.git
synced 2024-10-05 20:47:46 -04:00
46c92405f2
A missing flush in the serial reader implementation for Linux target was causing input to idf.py monitor to not be forwarded to the application. This is fixed now.
379 lines
16 KiB
Python
Executable File
379 lines
16 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
# esp-idf serial output monitor tool. Does some helpful things:
|
|
# - Looks up hex addresses in ELF file with addr2line
|
|
# - Reset ESP32 via serial RTS line (Ctrl-T Ctrl-R)
|
|
# - Run flash build target to rebuild and flash entire project (Ctrl-T Ctrl-F)
|
|
# - Run app-flash build target to rebuild and flash app only (Ctrl-T Ctrl-A)
|
|
# - If gdbstub output is detected, gdb is automatically loaded
|
|
# - If core dump output is detected, it is converted to a human-readable report
|
|
# by espcoredump.py.
|
|
#
|
|
# SPDX-FileCopyrightText: 2015-2022 Espressif Systems (Shanghai) CO LTD
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
#
|
|
# Contains elements taken from miniterm "Very simple serial terminal" which
|
|
# is part of pySerial. https://github.com/pyserial/pyserial
|
|
# (C)2002-2015 Chris Liechti <cliechti@gmx.net>
|
|
#
|
|
# Originally released under BSD-3-Clause license.
|
|
#
|
|
|
|
import codecs
|
|
import io
|
|
import os
|
|
import queue
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from builtins import bytes
|
|
from typing import Any, List, Optional, Type, Union
|
|
|
|
import serial
|
|
import serial.tools.list_ports
|
|
# Windows console stuff
|
|
from idf_monitor_base.ansi_color_converter import get_converter
|
|
from idf_monitor_base.argument_parser import get_parser
|
|
from idf_monitor_base.console_parser import ConsoleParser
|
|
from idf_monitor_base.console_reader import ConsoleReader
|
|
from idf_monitor_base.constants import (CTRL_C, CTRL_H, DEFAULT_PRINT_FILTER, DEFAULT_TOOLCHAIN_PREFIX,
|
|
ESPPORT_ENVIRON, EVENT_QUEUE_TIMEOUT, GDB_EXIT_TIMEOUT,
|
|
GDB_UART_CONTINUE_COMMAND, LAST_LINE_THREAD_INTERVAL, MAKEFLAGS_ENVIRON,
|
|
PANIC_DECODE_DISABLE, PANIC_IDLE, TAG_CMD, TAG_KEY, TAG_SERIAL,
|
|
TAG_SERIAL_FLUSH)
|
|
from idf_monitor_base.coredump import COREDUMP_DECODE_INFO, CoreDump
|
|
from idf_monitor_base.exceptions import SerialStopException
|
|
from idf_monitor_base.gdbhelper import GDBHelper
|
|
from idf_monitor_base.line_matcher import LineMatcher
|
|
from idf_monitor_base.logger import Logger
|
|
from idf_monitor_base.output_helpers import normal_print, yellow_print
|
|
from idf_monitor_base.serial_handler import SerialHandler, SerialHandlerNoElf, run_make
|
|
from idf_monitor_base.serial_reader import LinuxReader, SerialReader
|
|
from idf_monitor_base.web_socket_client import WebSocketClient
|
|
from serial.tools import miniterm
|
|
|
|
key_description = miniterm.key_description
|
|
|
|
|
|
class Monitor:
|
|
"""
|
|
Monitor application base class.
|
|
|
|
This was originally derived from miniterm.Miniterm, but it turned out to be easier to write from scratch for this
|
|
purpose.
|
|
|
|
Main difference is that all event processing happens in the main thread, not the worker threads.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
serial_instance, # type: serial.Serial
|
|
elf_file, # type: str
|
|
print_filter, # type: str
|
|
make='make', # type: str
|
|
encrypted=False, # type: bool
|
|
reset=True, # type: bool
|
|
toolchain_prefix=DEFAULT_TOOLCHAIN_PREFIX, # type: str
|
|
eol='CRLF', # type: str
|
|
decode_coredumps=COREDUMP_DECODE_INFO, # type: str
|
|
decode_panic=PANIC_DECODE_DISABLE, # type: str
|
|
target='esp32', # type: str
|
|
websocket_client=None, # type: Optional[WebSocketClient]
|
|
enable_address_decoding=True, # type: bool
|
|
timestamps=False, # type: bool
|
|
timestamp_format='', # type: str
|
|
force_color=False # type: bool
|
|
):
|
|
self.event_queue = queue.Queue() # type: queue.Queue
|
|
self.cmd_queue = queue.Queue() # type: queue.Queue
|
|
self.console = miniterm.Console()
|
|
# if the variable is set ANSI will be printed even if we do not print to terminal
|
|
sys.stderr = get_converter(sys.stderr, decode_output=True, force_color=force_color)
|
|
self.console.output = get_converter(self.console.output, force_color=force_color)
|
|
self.console.byte_output = get_converter(self.console.byte_output, force_color=force_color)
|
|
|
|
self.elf_file = elf_file or ''
|
|
self.elf_exists = os.path.exists(self.elf_file)
|
|
self.logger = Logger(self.elf_file, self.console, timestamps, timestamp_format, b'', enable_address_decoding,
|
|
toolchain_prefix)
|
|
|
|
self.coredump = CoreDump(decode_coredumps, self.event_queue, self.logger, websocket_client,
|
|
self.elf_file) if self.elf_exists else None
|
|
|
|
# allow for possibility the "make" arg is a list of arguments (for idf.py)
|
|
self.make = make if os.path.exists(make) else shlex.split(make) # type: Any[Union[str, List[str]], str]
|
|
self.target = target
|
|
|
|
# testing hook - data from serial can make exit the monitor
|
|
if isinstance(self, SerialMonitor):
|
|
socket_mode = serial_instance.port.startswith('socket://')
|
|
self.serial = serial_instance
|
|
self.serial_reader = SerialReader(self.serial, self.event_queue, reset)
|
|
|
|
self.gdb_helper = GDBHelper(toolchain_prefix, websocket_client, self.elf_file, self.serial.port,
|
|
self.serial.baudrate) if self.elf_exists else None
|
|
|
|
else:
|
|
socket_mode = False
|
|
self.serial = subprocess.Popen([self.elf_file], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT)
|
|
self.serial_reader = LinuxReader(self.serial, self.event_queue)
|
|
|
|
self.gdb_helper = None
|
|
|
|
cls = SerialHandler if self.elf_exists else SerialHandlerNoElf
|
|
self.serial_handler = cls(b'', socket_mode, self.logger, decode_panic, PANIC_IDLE, b'', target,
|
|
False, False, self.serial, encrypted, reset, self.elf_file)
|
|
|
|
self.console_parser = ConsoleParser(eol)
|
|
self.console_reader = ConsoleReader(self.console, self.event_queue, self.cmd_queue, self.console_parser,
|
|
socket_mode)
|
|
|
|
self._line_matcher = LineMatcher(print_filter)
|
|
|
|
# internal state
|
|
self._invoke_processing_last_line_timer = None # type: Optional[threading.Timer]
|
|
|
|
def __enter__(self) -> None:
|
|
""" Use 'with self' to temporarily disable monitoring behaviour """
|
|
self.serial_reader.stop()
|
|
self.console_reader.stop()
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None: # type: ignore
|
|
raise NotImplementedError
|
|
|
|
def run_make(self, target: str) -> None:
|
|
with self:
|
|
run_make(target, self.make, self.console, self.console_parser, self.event_queue, self.cmd_queue,
|
|
self.logger)
|
|
|
|
def _pre_start(self) -> None:
|
|
self.console_reader.start()
|
|
self.serial_reader.start()
|
|
|
|
def main_loop(self) -> None:
|
|
self._pre_start()
|
|
|
|
try:
|
|
while self.console_reader.alive and self.serial_reader.alive:
|
|
try:
|
|
self._main_loop()
|
|
except KeyboardInterrupt:
|
|
yellow_print('To exit from IDF monitor please use \"Ctrl+]\". Alternatively, you can use Ctrl-T Ctrl-X to exit.')
|
|
self.serial_write(codecs.encode(CTRL_C))
|
|
except SerialStopException:
|
|
normal_print('Stopping condition has been received\n')
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
try:
|
|
self.console_reader.stop()
|
|
self.serial_reader.stop()
|
|
self.logger.stop_logging()
|
|
# Cancelling _invoke_processing_last_line_timer is not
|
|
# important here because receiving empty data doesn't matter.
|
|
self._invoke_processing_last_line_timer = None
|
|
except Exception: # noqa
|
|
pass
|
|
normal_print('\n')
|
|
|
|
def serial_write(self, *args: str, **kwargs: str) -> None:
|
|
raise NotImplementedError
|
|
|
|
def check_gdb_stub_and_run(self, line: bytes) -> None:
|
|
raise NotImplementedError
|
|
|
|
def invoke_processing_last_line(self) -> None:
|
|
self.event_queue.put((TAG_SERIAL_FLUSH, b''), False)
|
|
|
|
def _main_loop(self) -> None:
|
|
try:
|
|
item = self.cmd_queue.get_nowait()
|
|
except queue.Empty:
|
|
try:
|
|
item = self.event_queue.get(timeout=EVENT_QUEUE_TIMEOUT)
|
|
except queue.Empty:
|
|
return
|
|
|
|
event_tag, data = item
|
|
if event_tag == TAG_CMD:
|
|
self.serial_handler.handle_commands(data, self.target, self.run_make, self.console_reader,
|
|
self.serial_reader)
|
|
elif event_tag == TAG_KEY:
|
|
self.serial_write(codecs.encode(data))
|
|
elif event_tag == TAG_SERIAL:
|
|
self.serial_handler.handle_serial_input(data, self.console_parser, self.coredump,
|
|
self.gdb_helper, self._line_matcher,
|
|
self.check_gdb_stub_and_run)
|
|
if self._invoke_processing_last_line_timer is not None:
|
|
self._invoke_processing_last_line_timer.cancel()
|
|
self._invoke_processing_last_line_timer = threading.Timer(LAST_LINE_THREAD_INTERVAL,
|
|
self.invoke_processing_last_line)
|
|
self._invoke_processing_last_line_timer.start()
|
|
# If no further data is received in the next short period
|
|
# of time then the _invoke_processing_last_line_timer
|
|
# generates an event which will result in the finishing of
|
|
# the last line. This is fix for handling lines sent
|
|
# without EOL.
|
|
# finalizing the line when coredump is in progress causes decoding issues
|
|
# the espcoredump loader uses empty line as a sign for end-of-coredump
|
|
# line is finalized only for non coredump data
|
|
elif event_tag == TAG_SERIAL_FLUSH:
|
|
self.serial_handler.handle_serial_input(data, self.console_parser, self.coredump,
|
|
self.gdb_helper, self._line_matcher,
|
|
self.check_gdb_stub_and_run,
|
|
finalize_line=not self.coredump or not self.coredump.in_progress)
|
|
else:
|
|
raise RuntimeError('Bad event data %r' % ((event_tag, data),))
|
|
|
|
|
|
class SerialMonitor(Monitor):
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None: # type: ignore
|
|
""" Use 'with self' to temporarily disable monitoring behaviour """
|
|
self.console_reader.start()
|
|
if self.elf_exists:
|
|
self.serial_reader.gdb_exit = self.gdb_helper.gdb_exit # write gdb_exit flag
|
|
self.serial_reader.start()
|
|
|
|
def _pre_start(self) -> None:
|
|
super()._pre_start()
|
|
if self.elf_exists:
|
|
self.gdb_helper.gdb_exit = False
|
|
self.serial_handler.start_cmd_sent = False
|
|
|
|
def serial_write(self, *args: str, **kwargs: str) -> None:
|
|
self.serial: serial.Serial
|
|
try:
|
|
self.serial.write(*args, **kwargs)
|
|
except serial.SerialException:
|
|
pass # this shouldn't happen, but sometimes port has closed in serial thread
|
|
except UnicodeEncodeError:
|
|
pass # this can happen if a non-ascii character was passed, ignoring
|
|
|
|
def check_gdb_stub_and_run(self, line: bytes) -> None: # type: ignore # The base class one is a None value
|
|
if self.gdb_helper and self.gdb_helper.check_gdb_stub_trigger(line):
|
|
with self: # disable console control
|
|
self.gdb_helper.run_gdb()
|
|
|
|
def _main_loop(self) -> None:
|
|
if self.elf_exists and self.gdb_helper.gdb_exit:
|
|
self.gdb_helper.gdb_exit = False
|
|
time.sleep(GDB_EXIT_TIMEOUT)
|
|
# Continue the program after exit from the GDB
|
|
self.serial_write(codecs.encode(GDB_UART_CONTINUE_COMMAND))
|
|
self.serial_handler.start_cmd_sent = True
|
|
|
|
super()._main_loop()
|
|
|
|
|
|
class LinuxMonitor(Monitor):
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None: # type: ignore
|
|
""" Use 'with self' to temporarily disable monitoring behaviour """
|
|
self.console_reader.start()
|
|
self.serial_reader.start()
|
|
|
|
def serial_write(self, *args: str, **kwargs: str) -> None:
|
|
self.serial.stdin.write(*args, **kwargs)
|
|
self.serial.stdin.flush()
|
|
|
|
def check_gdb_stub_and_run(self, line: bytes) -> None:
|
|
return # fake function for linux target
|
|
|
|
|
|
def main() -> None:
|
|
parser = get_parser()
|
|
args = parser.parse_args()
|
|
|
|
# The port name is changed in cases described in the following lines. Use a local argument and
|
|
# avoid the modification of args.port.
|
|
port = args.port
|
|
|
|
# GDB uses CreateFile to open COM port, which requires the COM name to be r'\\.\COMx' if the COM
|
|
# number is larger than 10
|
|
if os.name == 'nt' and port.startswith('COM'):
|
|
port = port.replace('COM', r'\\.\COM')
|
|
yellow_print('--- WARNING: GDB cannot open serial ports accessed as COMx')
|
|
yellow_print('--- Using %s instead...' % port)
|
|
elif port.startswith('/dev/tty.') and sys.platform == 'darwin':
|
|
port = port.replace('/dev/tty.', '/dev/cu.')
|
|
yellow_print('--- WARNING: Serial ports accessed as /dev/tty.* will hang gdb if launched.')
|
|
yellow_print('--- Using %s instead...' % port)
|
|
|
|
if isinstance(args.elf_file, io.BufferedReader):
|
|
elf_file = args.elf_file.name
|
|
args.elf_file.close() # don't need this as a file
|
|
else:
|
|
elf_file = args.elf_file
|
|
|
|
# remove the parallel jobserver arguments from MAKEFLAGS, as any
|
|
# parent make is only running 1 job (monitor), so we can re-spawn
|
|
# all of the child makes we need (the -j argument remains part of
|
|
# MAKEFLAGS)
|
|
try:
|
|
makeflags = os.environ[MAKEFLAGS_ENVIRON]
|
|
makeflags = re.sub(r'--jobserver[^ =]*=[0-9,]+ ?', '', makeflags)
|
|
os.environ[MAKEFLAGS_ENVIRON] = makeflags
|
|
except KeyError:
|
|
pass # not running a make jobserver
|
|
|
|
ws = WebSocketClient(args.ws) if args.ws else None
|
|
try:
|
|
cls: Type[Monitor]
|
|
if args.target == 'linux':
|
|
serial_instance = None
|
|
cls = LinuxMonitor
|
|
yellow_print('--- idf_monitor on linux ---')
|
|
else:
|
|
serial_instance = serial.serial_for_url(port, args.baud, do_not_open=True)
|
|
serial_instance.dtr = False
|
|
serial_instance.rts = False
|
|
|
|
# Pass the actual used port to callee of idf_monitor (e.g. idf.py/cmake) through `ESPPORT` environment
|
|
# variable.
|
|
# Note that the port must be original port argument without any replacement done in IDF Monitor (idf.py
|
|
# has a check for this).
|
|
# To make sure the key as well as the value are str type, by the requirements of subprocess
|
|
espport_val = str(args.port)
|
|
os.environ.update({ESPPORT_ENVIRON: espport_val})
|
|
|
|
cls = SerialMonitor
|
|
yellow_print('--- idf_monitor on {p.name} {p.baudrate} ---'.format(p=serial_instance))
|
|
|
|
monitor = cls(serial_instance,
|
|
elf_file,
|
|
args.print_filter,
|
|
args.make,
|
|
args.encrypted,
|
|
not args.no_reset,
|
|
args.toolchain_prefix,
|
|
args.eol,
|
|
args.decode_coredumps,
|
|
args.decode_panic,
|
|
args.target,
|
|
ws,
|
|
not args.disable_address_decoding,
|
|
args.timestamps,
|
|
args.timestamp_format,
|
|
args.force_color)
|
|
|
|
yellow_print('--- Quit: {} | Menu: {} | Help: {} followed by {} ---'.format(
|
|
key_description(monitor.console_parser.exit_key),
|
|
key_description(monitor.console_parser.menu_key),
|
|
key_description(monitor.console_parser.menu_key),
|
|
key_description(CTRL_H)))
|
|
if args.print_filter != DEFAULT_PRINT_FILTER:
|
|
yellow_print('--- Print filter: {} ---'.format(args.print_filter))
|
|
monitor.main_loop()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
if ws:
|
|
ws.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|