alex/esp-idf
alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2024-09-20 00:36:01 -04:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Merge branch 'feature/refactor-idf-monitor' into 'master'

Browse Source 
Decompose idf_monitor.py

Closes IDF-2432

See merge request espressif/esp-idf!11815
This commit is contained in:
Ivan Grokhotkov 2021-04-14 14:32:17 +00:00
commit e220e9b571
18 changed files with 1032 additions and 700 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
components/esptool_py/project_include.cmake
View File

@ -52,6 +52,12 @@ if(min_rev)
unset(min_rev)
endif()
if(CONFIG_ESP32_REV_MIN)
set(rev_min ${CONFIG_ESP32_REV_MIN})
else()
set(rev_min -1)
endif()
if(CONFIG_ESPTOOLPY_FLASHSIZE_DETECT)
# Set ESPFLASHSIZE to 'detect' *after* elf2image options are generated,
# as elf2image can't have 'detect' as an option...
@ -154,7 +160,7 @@ add_custom_target(monitor
COMMAND ${CMAKE_COMMAND}
-D IDF_PATH="${idf_path}"
-D SERIAL_TOOL="${ESPMONITOR}"
-D SERIAL_TOOL_ARGS="${elf_dir}/${elf}"
-D SERIAL_TOOL_ARGS="--target ${target} --revision ${rev_min} ${elf_dir}/${elf}"
-D WORKING_DIRECTORY="${build_dir}"
-P run_serial_tool.cmake
WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR}

2
requirements.txt
View File

@ -6,7 +6,7 @@ setuptools>=21
# Please keep it as the first item of this list. Version 21 is required to handle PEP 508 environment markers.
#
click>=7.0
pyserial>=3.0
pyserial>=3.3
future>=0.15.2
cryptography>=2.1.4
pyparsing>=2.0.3,<2.4.0

1
tools/ci/mypy_ignore_list.txt
View File

@ -228,7 +228,6 @@ tools/find_build_apps/make.py
tools/gdb_panic_server.py
tools/gen_esp_err_to_name.py
tools/idf.py
tools/idf_monitor.py
tools/idf_py_actions/constants.py
tools/idf_py_actions/core_ext.py
tools/idf_py_actions/create_ext.py

1
tools/cmake/project_description.json.in
View File

@ -8,6 +8,7 @@
"app_bin": "${PROJECT_BIN}",
"git_revision": "${IDF_VER}",
"target": "${CONFIG_IDF_TARGET}",
"rev": "${CONFIG_ESP32_REV_MIN}",
"phy_data_partition": "${CONFIG_ESP32_PHY_INIT_DATA_IN_PARTITION}",
"monitor_baud" : "${CONFIG_ESPTOOLPY_MONITOR_BAUD}",
"monitor_toolprefix": "${CONFIG_SDK_TOOLPREFIX}",

816
tools/idf_monitor.py
View File

File diff suppressed because it is too large Load Diff

35
tools/idf_monitor_base/__init__.py Normal file
View File

@ -0,0 +1,35 @@
import re
# regex matches an potential PC value (0x4xxxxxxx)
MATCH_PCADDR = re.compile(r'0x4[0-9a-f]{7}', re.IGNORECASE)
DEFAULT_TOOLCHAIN_PREFIX = 'xtensa-esp32-elf-'
DEFAULT_PRINT_FILTER = ''
# coredump related messages
COREDUMP_UART_START = b'================= CORE DUMP START ================='
COREDUMP_UART_END = b'================= CORE DUMP END ================='
COREDUMP_UART_PROMPT = b'Press Enter to print core dump to UART...'
# coredump states
COREDUMP_IDLE = 0
COREDUMP_READING = 1
COREDUMP_DONE = 2
# coredump decoding options
COREDUMP_DECODE_DISABLE = 'disable'
COREDUMP_DECODE_INFO = 'info'
# panic handler related messages
PANIC_START = r'Core \s*\d+ register dump:'
PANIC_END = b'ELF file SHA256:'
PANIC_STACK_DUMP = b'Stack memory:'
# panic handler decoding states
PANIC_IDLE = 0
PANIC_READING = 1
# panic handler decoding options
PANIC_DECODE_DISABLE = 'disable'
PANIC_DECODE_BACKTRACE = 'backtrace'

121
tools/idf_monitor_base/ansi_color_convertor.py Normal file
View File

@ -0,0 +1,121 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import ctypes
import os
import re
import sys
from io import TextIOBase
from typing import Union
from .output_helpers import ANSI_NORMAL
STD_OUTPUT_HANDLE = -11
STD_ERROR_HANDLE = -12
# wincon.h values
FOREGROUND_INTENSITY = 8
FOREGROUND_GREY = 7
# matches the ANSI color change sequences that IDF sends
RE_ANSI_COLOR = re.compile(b'\033\\[([01]);3([0-7])m')
# list mapping the 8 ANSI colors (the indexes) to Windows Console colors
ANSI_TO_WINDOWS_COLOR = [0, 4, 2, 6, 1, 5, 3, 7]
if os.name == 'nt':
GetStdHandle = ctypes.windll.kernel32.GetStdHandle # type: ignore
SetConsoleTextAttribute = ctypes.windll.kernel32.SetConsoleTextAttribute # type: ignore
class ANSIColorConverter(object):
"""Class to wrap a file-like output stream, intercept ANSI color codes,
and convert them into calls to Windows SetConsoleTextAttribute.
Doesn't support all ANSI terminal code escape sequences, only the sequences IDF uses.
Ironically, in Windows this console output is normally wrapped by winpty which will then detect the console text
color changes and convert these back to ANSI color codes for MSYS' terminal to display. However this is the
least-bad working solution, as winpty doesn't support any "passthrough" mode for raw output.
"""
def __new__(cls, output=None, decode_output=False): # type: ignore # noqa
if os.name == 'nt':
return cls
return output
def __init__(self, output=None, decode_output=False):
# type: (TextIOBase, bool) -> None
self.output = output
self.decode_output = decode_output
self.handle = GetStdHandle(STD_ERROR_HANDLE if self.output == sys.stderr else STD_OUTPUT_HANDLE)
self.matched = b''
def _output_write(self, data): # type: (Union[str, bytes]) -> None
try:
if self.decode_output:
self.output.write(data.decode()) # type: ignore
else:
self.output.write(data) # type: ignore
except (IOError, OSError):
# Windows 10 bug since the Fall Creators Update, sometimes writing to console randomly throws
# an exception (however, the character is still written to the screen)
# Ref https://github.com/espressif/esp-idf/issues/1163
#
# Also possible for Windows to throw an OSError error if the data is invalid for the console
# (garbage bytes, etc)
pass
except UnicodeDecodeError:
# In case of double byte Unicode characters display '?'
self.output.write('?') # type: ignore
def write(self, data): # type: ignore
if isinstance(data, bytes):
data = bytearray(data)
else:
data = bytearray(data, 'utf-8')
for b in data:
b = bytes([b])
length = len(self.matched)
if b == b'\033': # ESC
self.matched = b
elif (length == 1 and b == b'[') or (1 < length < 7):
self.matched += b
if self.matched == ANSI_NORMAL.encode('latin-1'): # reset console
# Flush is required only with Python3 - switching color before it is printed would mess up the console
self.flush()
SetConsoleTextAttribute(self.handle, FOREGROUND_GREY)
self.matched = b''
elif len(self.matched) == 7: # could be an ANSI sequence
m = re.match(RE_ANSI_COLOR, self.matched)
if m is not None:
color = ANSI_TO_WINDOWS_COLOR[int(m.group(2))]
if m.group(1) == b'1':
color |= FOREGROUND_INTENSITY
# Flush is required only with Python3 - switching color before it is printed would mess up the console
self.flush()
SetConsoleTextAttribute(self.handle, color)
else:
self._output_write(self.matched) # not an ANSI color code, display verbatim
self.matched = b''
else:
self._output_write(b)
self.matched = b''
def flush(self): # type: () -> None
try:
self.output.flush() # type: ignore
except OSError:
# Account for Windows Console refusing to accept garbage bytes (serial noise, etc)
pass

70
tools/idf_monitor_base/chip_specific_config.py Normal file
View File

@ -0,0 +1,70 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# This file contains values (e.g. delay time, ...) that are different for each chip for a particular action.
# If adding a new device, set only values that are different from the default, e.g.:
# 'esp32s2': {
# 0: {
# 'reset': 0.35,
# }
# },
#
# for more information see the method "handle_commands" in idf_monitor.py
conf = {
# the default values were previously hardcoded in idf_monitor.py (taken from esptool.py)
'default': {
0: {
'reset': 0.2,
'enter_boot_set': 0.1,
'enter_boot_unset': 0.05,
}
},
'esp32': {
0: {
'reset': 0.2,
'enter_boot_set': 1.3,
'enter_boot_unset': 0.45,
},
1: {
'reset': 0.2,
'enter_boot_set': 0.1,
'enter_boot_unset': 0.05,
}
},
}
def get_chip_config(chip, revision=0):
# type: (str, int) -> dict
# If the config is not set in the `conf` dict for a specific chip, the `default` will be used.
# In case if only some values are specified, others are used from the `default`.
# If chip is set in `conf` but the specific revision R is missing,
# the values from highest revision lower than R are used.
# If some fields are missing, they will be taken from next lower revision or from the `default`.
default = dict(conf['default'][0])
rev_number = int(revision)
if chip not in conf.keys():
return default
chip_revisions = sorted(list(conf[chip].keys()), key=int)
for rev in chip_revisions:
if int(rev) > rev_number:
break
for key in conf[chip][rev].keys():
default[key] = conf[chip][rev][key]
return default

133
tools/idf_monitor_base/console_parser.py Normal file
View File

@ -0,0 +1,133 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import textwrap
from typing import Optional
import serial.tools.miniterm as miniterm
from .constants import (CMD_APP_FLASH, CMD_ENTER_BOOT, CMD_MAKE, CMD_OUTPUT_TOGGLE, CMD_RESET, CMD_STOP,
CMD_TOGGLE_LOGGING, CTRL_A, CTRL_F, CTRL_H, CTRL_L, CTRL_P, CTRL_R, CTRL_RBRACKET, CTRL_T,
CTRL_X, CTRL_Y, TAG_CMD, TAG_KEY, __version__)
from .output_helpers import red_print, yellow_print
key_description = miniterm.key_description
class ConsoleParser(object):
def __init__(self, eol='CRLF'): # type: (str) -> None
self.translate_eol = {
'CRLF': lambda c: c.replace('\n', '\r\n'),
'CR': lambda c: c.replace('\n', '\r'),
'LF': lambda c: c.replace('\r', '\n'),
}[eol]
self.menu_key = CTRL_T
self.exit_key = CTRL_RBRACKET
self._pressed_menu_key = False
def parse(self, key): # type: (str) -> Optional[tuple]
ret = None
if self._pressed_menu_key:
ret = self._handle_menu_key(key)
elif key == self.menu_key:
self._pressed_menu_key = True
elif key == self.exit_key:
ret = (TAG_CMD, CMD_STOP)
else:
key = self.translate_eol(key)
ret = (TAG_KEY, key)
return ret
def _handle_menu_key(self, c): # type: (str) -> Optional[tuple]
ret = None
if c == self.exit_key or c == self.menu_key: # send verbatim
ret = (TAG_KEY, c)
elif c in [CTRL_H, 'h', 'H', '?']:
red_print(self.get_help_text())
elif c == CTRL_R: # Reset device via RTS
ret = (TAG_CMD, CMD_RESET)
elif c == CTRL_F: # Recompile & upload
ret = (TAG_CMD, CMD_MAKE)
elif c in [CTRL_A, 'a', 'A']: # Recompile & upload app only
# "CTRL-A" cannot be captured with the default settings of the Windows command line, therefore, "A" can be used
# instead
ret = (TAG_CMD, CMD_APP_FLASH)
elif c == CTRL_Y: # Toggle output display
ret = (TAG_CMD, CMD_OUTPUT_TOGGLE)
elif c == CTRL_L: # Toggle saving output into file
ret = (TAG_CMD, CMD_TOGGLE_LOGGING)
elif c == CTRL_P:
yellow_print('Pause app (enter bootloader mode), press Ctrl-T Ctrl-R to restart')
# to fast trigger pause without press menu key
ret = (TAG_CMD, CMD_ENTER_BOOT)
elif c in [CTRL_X, 'x', 'X']: # Exiting from within the menu
ret = (TAG_CMD, CMD_STOP)
else:
red_print('--- unknown menu character {} --'.format(key_description(c)))
self._pressed_menu_key = False
return ret
def get_help_text(self): # type: () -> str
text = """\
--- idf_monitor ({version}) - ESP-IDF monitor tool
--- based on miniterm from pySerial
---
--- {exit:8} Exit program
--- {menu:8} Menu escape key, followed by:
--- Menu keys:
--- {menu:14} Send the menu character itself to remote
--- {exit:14} Send the exit character itself to remote
--- {reset:14} Reset target board via RTS line
--- {makecmd:14} Build & flash project
--- {appmake:14} Build & flash app only
--- {output:14} Toggle output display
--- {log:14} Toggle saving output into file
--- {pause:14} Reset target into bootloader to pause app via RTS line
--- {menuexit:14} Exit program
""".format(version=__version__,
exit=key_description(self.exit_key),
menu=key_description(self.menu_key),
reset=key_description(CTRL_R),
makecmd=key_description(CTRL_F),
appmake=key_description(CTRL_A) + ' (or A)',
output=key_description(CTRL_Y),
log=key_description(CTRL_L),
pause=key_description(CTRL_P),
menuexit=key_description(CTRL_X) + ' (or X)')
return textwrap.dedent(text)
def get_next_action_text(self): # type: () -> str
text = """\
--- Press {} to exit monitor.
--- Press {} to build & flash project.
--- Press {} to build & flash app.
--- Press any other key to resume monitor (resets target).
""".format(key_description(self.exit_key),
key_description(CTRL_F),
key_description(CTRL_A))
return textwrap.dedent(text)
def parse_next_action_key(self, c): # type: (str) -> Optional[tuple]
ret = None
if c == self.exit_key:
ret = (TAG_CMD, CMD_STOP)
elif c == CTRL_F: # Recompile & upload
ret = (TAG_CMD, CMD_MAKE)
elif c in [CTRL_A, 'a', 'A']: # Recompile & upload app only
# "CTRL-A" cannot be captured with the default settings of the Windows command line, therefore, "A" can be used
# instead
ret = (TAG_CMD, CMD_APP_FLASH)
return ret

101
tools/idf_monitor_base/console_reader.py Normal file
View File

@ -0,0 +1,101 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import time
from serial.tools.miniterm import Console
from .console_parser import ConsoleParser
from .constants import CMD_STOP, TAG_CMD
from .stoppable_thread import StoppableThread
try:
import queue
except ImportError:
import Queue as queue # type: ignore
class ConsoleReader(StoppableThread):
""" Read input keys from the console and push them to the queue,
until stopped.
"""
def __init__(self, console, event_queue, cmd_queue, parser, test_mode):
# type: (Console, queue.Queue, queue.Queue, ConsoleParser, bool) -> None
super(ConsoleReader, self).__init__()
self.console = console
self.event_queue = event_queue
self.cmd_queue = cmd_queue
self.parser = parser
self.test_mode = test_mode
def run(self):
# type: () -> None
self.console.setup()
try:
while self.alive:
try:
if os.name == 'nt':
# Windows kludge: because the console.cancel() method doesn't
# seem to work to unblock getkey() on the Windows implementation.
#
# So we only call getkey() if we know there's a key waiting for us.
import msvcrt
while not msvcrt.kbhit() and self.alive: # type: ignore
time.sleep(0.1)
if not self.alive:
break
elif self.test_mode:
# In testing mode the stdin is connected to PTY but is not used for input anything. For PTY
# the canceling by fcntl.ioctl isn't working and would hang in self.console.getkey().
# Therefore, we avoid calling it.
while self.alive:
time.sleep(0.1)
break
c = self.console.getkey()
except KeyboardInterrupt:
c = '\x03'
if c is not None:
ret = self.parser.parse(c)
if ret is not None:
(tag, cmd) = ret
# stop command should be executed last
if tag == TAG_CMD and cmd != CMD_STOP:
self.cmd_queue.put(ret)
else:
self.event_queue.put(ret)
finally:
self.console.cleanup()
def _cancel(self):
# type: () -> None
if os.name == 'posix' and not self.test_mode:
# this is the way cancel() is implemented in pyserial 3.3 or newer,
# older pyserial (3.1+) has cancellation implemented via 'select',
# which does not work when console sends an escape sequence response
#
# even older pyserial (<3.1) does not have this method
#
# on Windows there is a different (also hacky) fix, applied above.
#
# note that TIOCSTI is not implemented in WSL / bash-on-Windows.
# TODO: introduce some workaround to make it work there.
#
# Note: This would throw exception in testing mode when the stdin is connected to PTY.
import fcntl
import termios
fcntl.ioctl(self.console.fd, termios.TIOCSTI, b'\0')

43
tools/idf_monitor_base/constants.py Normal file
View File

@ -0,0 +1,43 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Control-key characters
CTRL_A = '\x01'
CTRL_B = '\x02'
CTRL_F = '\x06'
CTRL_H = '\x08'
CTRL_R = '\x12'
CTRL_T = '\x14'
CTRL_Y = '\x19'
CTRL_P = '\x10'
CTRL_X = '\x18'
CTRL_L = '\x0c'
CTRL_RBRACKET = '\x1d' # Ctrl+]
# Command parsed from console inputs
CMD_STOP = 1
CMD_RESET = 2
CMD_MAKE = 3
CMD_APP_FLASH = 4
CMD_OUTPUT_TOGGLE = 5
CMD_TOGGLE_LOGGING = 6
CMD_ENTER_BOOT = 7
# Tags for tuples in queues
TAG_KEY = 0
TAG_SERIAL = 1
TAG_SERIAL_FLUSH = 2
TAG_CMD = 3
__version__ = '1.1'

19
tools/idf_monitor_base/exceptions.py Normal file
View File

@ -0,0 +1,19 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
class SerialStopException(Exception):
"""
This exception is used for stopping the IDF monitor in testing mode.
"""
pass

71
tools/idf_monitor_base/line_matcher.py Normal file
View File

@ -0,0 +1,71 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re
class LineMatcher(object):
"""
Assembles a dictionary of filtering rules based on the --print_filter
argument of idf_monitor. Then later it is used to match lines and
determine whether they should be shown on screen or not.
"""
LEVEL_N = 0
LEVEL_E = 1
LEVEL_W = 2
LEVEL_I = 3
LEVEL_D = 4
LEVEL_V = 5
level = {'N': LEVEL_N, 'E': LEVEL_E, 'W': LEVEL_W, 'I': LEVEL_I, 'D': LEVEL_D,
'V': LEVEL_V, '*': LEVEL_V, '': LEVEL_V}
def __init__(self, print_filter):
# type: (str) -> None
self._dict = dict()
self._re = re.compile(r'^(?:\033\[[01];?[0-9]+m?)?([EWIDV]) \([0-9]+\) ([^:]+): ')
items = print_filter.split()
if len(items) == 0:
self._dict['*'] = self.LEVEL_V # default is to print everything
for f in items:
s = f.split(r':')
if len(s) == 1:
# specifying no warning level defaults to verbose level
lev = self.LEVEL_V
elif len(s) == 2:
if len(s[0]) == 0:
raise ValueError('No tag specified in filter ' + f)
try:
lev = self.level[s[1].upper()]
except KeyError:
raise ValueError('Unknown warning level in filter ' + f)
else:
raise ValueError('Missing ":" in filter ' + f)
self._dict[s[0]] = lev
def match(self, line):
# type: (str) -> bool
try:
m = self._re.search(line)
if m:
lev = self.level[m.group(1)]
if m.group(2) in self._dict:
return self._dict[m.group(2)] >= lev
return self._dict.get('*', self.LEVEL_N) >= lev
except (KeyError, IndexError):
# Regular line written with something else than ESP_LOG*
# or an empty line.
pass
# We need something more than "*.N" for printing.
return self._dict.get('*', self.LEVEL_N) > self.LEVEL_N

43
tools/idf_monitor_base/output_helpers.py Normal file
View File

@ -0,0 +1,43 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
from typing import Optional
# ANSI terminal codes (if changed, regular expressions in LineMatcher need to be udpated)
ANSI_RED = '\033[1;31m'
ANSI_YELLOW = '\033[0;33m'
ANSI_NORMAL = '\033[0m'
def color_print(message, color, newline='\n'):
# type: (str, str, Optional[str]) -> None
""" Print a message to stderr with colored highlighting """
sys.stderr.write('%s%s%s%s' % (color, message, ANSI_NORMAL, newline))
def normal_print(message):
# type: (str) -> None
sys.stderr.write(ANSI_NORMAL + message)
def yellow_print(message, newline='\n'):
# type: (str, Optional[str]) -> None
color_print(message, ANSI_YELLOW, newline)
def red_print(message, newline='\n'):
# type: (str, Optional[str]) -> None
color_print(message, ANSI_RED, newline)

85
tools/idf_monitor_base/serial_reader.py Normal file
View File

@ -0,0 +1,85 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import time
import serial
from .constants import TAG_SERIAL
from .output_helpers import red_print, yellow_print
from .stoppable_thread import StoppableThread
try:
import queue
except ImportError:
import Queue as queue # type: ignore # noqa
class SerialReader(StoppableThread):
""" Read serial data from the serial port and push to the
event queue, until stopped.
"""
def __init__(self, serial_instance, event_queue):
# type: (serial.Serial, queue.Queue) -> None
super(SerialReader, self).__init__()
self.baud = serial_instance.baudrate
self.serial = serial_instance
self.event_queue = event_queue
if not hasattr(self.serial, 'cancel_read'):
# enable timeout for checking alive flag,
# if cancel_read not available
self.serial.timeout = 0.25
def run(self):
# type: () -> None
if not self.serial.is_open:
self.serial.baudrate = self.baud
self.serial.rts = True # Force an RTS reset on open
self.serial.open()
self.serial.rts = False
self.serial.dtr = self.serial.dtr # usbser.sys workaround
try:
while self.alive:
try:
data = self.serial.read(self.serial.in_waiting or 1)
except (serial.serialutil.SerialException, IOError) as e:
data = b''
# self.serial.open() was successful before, therefore, this is an issue related to
# the disapperence of the device
red_print(e)
yellow_print('Waiting for the device to reconnect', newline='')
self.serial.close()
while self.alive: # so that exiting monitor works while waiting
try:
time.sleep(0.5)
self.serial.open()
break # device connected
except serial.serialutil.SerialException:
yellow_print('.', newline='')
sys.stderr.flush()
yellow_print('') # go to new line
if data:
self.event_queue.put((TAG_SERIAL, data), False)
finally:
self.serial.close()
def _cancel(self):
# type: () -> None
if hasattr(self.serial, 'cancel_read'):
try:
self.serial.cancel_read()
except Exception: # noqa
pass

68
tools/idf_monitor_base/stoppable_thread.py Normal file
View File

@ -0,0 +1,68 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import threading
from typing import Optional
class StoppableThread(object):
"""
Provide a Thread-like class which can be 'cancelled' via a subclass-provided
cancellation method.
Can be started and stopped multiple times.
Isn't an instance of type Thread because Python Thread objects can only be run once
"""
def __init__(self):
# type: () -> None
self._thread = None # type: Optional[threading.Thread]
@property
def alive(self):
# type: () -> bool
"""
Is 'alive' whenever the internal thread object exists
"""
return self._thread is not None
def start(self):
# type: () -> None
if self._thread is None:
self._thread = threading.Thread(target=self._run_outer)
self._thread.start()
def _cancel(self):
# type: () -> None
pass # override to provide cancellation functionality
def run(self):
# type: () -> None
pass # override for the main thread behaviour
def _run_outer(self):
# type: () -> None
try:
self.run()
finally:
self._thread = None
def stop(self):
# type: () -> None
if self._thread is not None:
old_thread = self._thread
self._thread = None
self._cancel()
old_thread.join()

109
tools/idf_monitor_base/web_socket_client.py Normal file
View File

@ -0,0 +1,109 @@
# Copyright 2015-2021 Espressif Systems (Shanghai) CO LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import time
from .output_helpers import red_print, yellow_print
try:
import websocket
except ImportError:
# This is needed for IDE integration only.
pass
class WebSocketClient(object):
"""
WebSocket client used to advertise debug events to WebSocket server by sending and receiving JSON-serialized
dictionaries.
Advertisement of debug event:
{'event': 'gdb_stub', 'port': '/dev/ttyUSB1', 'prog': 'build/elf_file'} for GDB Stub, or
{'event': 'coredump', 'file': '/tmp/xy', 'prog': 'build/elf_file'} for coredump,
where 'port' is the port for the connected device, 'prog' is the full path to the ELF file and 'file' is the
generated coredump file.
Expected end of external debugging:
{'event': 'debug_finished'}
"""
RETRIES = 3
CONNECTION_RETRY_DELAY = 1
def __init__(self, url): # type: (str) -> None
self.url = url
self._connect()
def _connect(self): # type: () -> None
"""
Connect to WebSocket server at url
"""
self.close()
for _ in range(self.RETRIES):
try:
self.ws = websocket.create_connection(self.url)
break # success
except NameError:
raise RuntimeError('Please install the websocket_client package for IDE integration!')
except Exception as e: # noqa
red_print('WebSocket connection error: {}'.format(e))
time.sleep(self.CONNECTION_RETRY_DELAY)
else:
raise RuntimeError('Cannot connect to WebSocket server')
def close(self): # type: () -> None
try:
self.ws.close()
except AttributeError:
# Not yet connected
pass
except Exception as e: # noqa
red_print('WebSocket close error: {}'.format(e))
def send(self, payload_dict): # type: (dict) -> None
"""
Serialize payload_dict in JSON format and send it to the server
"""
for _ in range(self.RETRIES):
try:
self.ws.send(json.dumps(payload_dict))
yellow_print('WebSocket sent: {}'.format(payload_dict))
break
except Exception as e: # noqa
red_print('WebSocket send error: {}'.format(e))
self._connect()
else:
raise RuntimeError('Cannot send to WebSocket server')
def wait(self, expect_iterable): # type: (list) -> None
"""
Wait until a dictionary in JSON format is received from the server with all (key, value) tuples from
expect_iterable.
"""
for _ in range(self.RETRIES):
try:
r = self.ws.recv()
except Exception as e:
red_print('WebSocket receive error: {}'.format(e))
self._connect()
continue
obj = json.loads(r)
if all([k in obj and obj[k] == v for k, v in expect_iterable]):
yellow_print('WebSocket received: {}'.format(obj))
break
red_print('WebSocket expected: {}, received: {}'.format(dict(expect_iterable), obj))
else:
raise RuntimeError('Cannot receive from WebSocket server')

6
tools/idf_py_actions/serial_ext.py
View File

@ -99,8 +99,11 @@ def action_extensions(base_actions, project_path):
monitor_args += ['--decode-coredumps', coredump_decode]
target_arch_riscv = get_sdkconfig_value(project_desc['config_file'], 'CONFIG_IDF_TARGET_ARCH_RISCV')
monitor_args += ['--target', project_desc['target']]
monitor_args += ['--revision', project_desc.get('rev', -1)]
if target_arch_riscv:
monitor_args += ['--decode-panic', 'backtrace', '--target', project_desc['target']]
monitor_args += ['--decode-panic', 'backtrace']
if print_filter is not None:
monitor_args += ['--print_filter', print_filter]
@ -114,6 +117,7 @@ def action_extensions(base_actions, project_path):
if 'MSYSTEM' in os.environ:
monitor_args = ['winpty'] + monitor_args
run_tool('idf_monitor', monitor_args, args.project_dir)
def flash(action, ctx, args):