import os import queue import subprocess import sys import tempfile from contextlib import contextmanager from typing import Generator from .constants import COREDUMP_SCRIPT, TAG_KEY from .output_helpers import Logger, yellow_print from .web_socket_client import WebSocketClient # 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' class CoreDump: def __init__(self, decode_coredumps, event_queue, logger, websocket_client, elf_file): # type: (str, queue.Queue, Logger, WebSocketClient, str) -> None self._coredump_buffer = b'' self._decode_coredumps = decode_coredumps self.event_queue = event_queue self._reading_coredump = COREDUMP_IDLE self.logger = logger self.websocket_client = websocket_client self.elf_file = elf_file def _process_coredump(self): # type: () -> None if self._decode_coredumps != COREDUMP_DECODE_INFO: raise NotImplementedError('process_coredump: %s not implemented' % self._decode_coredumps) coredump_file = None try: # On Windows, the temporary file can't be read unless it is closed. # Set delete=False and delete the file manually later. with tempfile.NamedTemporaryFile(mode='wb', delete=False) as coredump_file: coredump_file.write(self._coredump_buffer) coredump_file.flush() if self.websocket_client: self.logger.output_enabled = True yellow_print('Communicating through WebSocket') self.websocket_client.send({'event': 'coredump', 'file': coredump_file.name, 'prog': self.elf_file}) yellow_print('Waiting for debug finished event') self.websocket_client.wait([('event', 'debug_finished')]) yellow_print('Communications through WebSocket is finished') else: cmd = [sys.executable, COREDUMP_SCRIPT, 'info_corefile', '--core', coredump_file.name, '--core-format', 'b64', self.elf_file ] output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) self.logger.output_enabled = True self.logger.print(output) # noqa: E999 self.logger.output_enabled = False # Will be reenabled in check_coredump_trigger_after_print except subprocess.CalledProcessError as e: yellow_print('Failed to run espcoredump script: {}\n{}\n\n'.format(e, e.output)) self.logger.output_enabled = True self.logger.print(COREDUMP_UART_START + b'\n') self.logger.print(self._coredump_buffer) # end line will be printed in handle_serial_input finally: if coredump_file is not None: try: os.unlink(coredump_file.name) except OSError as e: yellow_print('Couldn\'t remote temporary core dump file ({})'.format(e)) def _check_coredump_trigger_before_print(self, line): # type: (bytes) -> None if self._decode_coredumps == COREDUMP_DECODE_DISABLE: return if COREDUMP_UART_PROMPT in line: yellow_print('Initiating core dump!') self.event_queue.put((TAG_KEY, '\n')) return if COREDUMP_UART_START in line: yellow_print('Core dump started (further output muted)') self._reading_coredump = COREDUMP_READING self._coredump_buffer = b'' self.logger.output_enabled = False return if COREDUMP_UART_END in line: self._reading_coredump = COREDUMP_DONE yellow_print('\nCore dump finished!') self._process_coredump() return if self._reading_coredump == COREDUMP_READING: kb = 1024 buffer_len_kb = len(self._coredump_buffer) // kb self._coredump_buffer += line.replace(b'\r', b'') + b'\n' new_buffer_len_kb = len(self._coredump_buffer) // kb if new_buffer_len_kb > buffer_len_kb: yellow_print('Received %3d kB...' % new_buffer_len_kb, newline='\r') def _check_coredump_trigger_after_print(self): # type: () -> None if self._decode_coredumps == COREDUMP_DECODE_DISABLE: return # Re-enable output after the last line of core dump has been consumed if not self.logger.output_enabled and self._reading_coredump == COREDUMP_DONE: self._reading_coredump = COREDUMP_IDLE self.logger.output_enabled = True self._coredump_buffer = b'' @contextmanager def check(self, line): # type: (bytes) -> Generator self._check_coredump_trigger_before_print(line) try: yield finally: self._check_coredump_trigger_after_print()