Add detect and save coredump

src/hardwario/chester/cli/app.py

@@ -91,17 +91,19 @@ def command_reset(ctx, halt, jlink_sn, jlink_speed):
 
 default_history_file = os.path.expanduser("~/.chester_history")
 default_console_file = os.path.expanduser("~/.chester_console")
+default_coredump_file = os.path.expanduser("~/.chester_coredump.bin")
 
 
 @cli.command('console')
 @click.option('--reset', is_flag=True, help='Reset application firmware.')
 @click.option('--latency', type=int, help='Latency for RTT readout in ms.', show_default=True, default=50)
 @click.option('--history-file', type=click.Path(writable=True), show_default=True, default=default_history_file)
 @click.option('--console-file', type=click.File('a', 'utf-8'), show_default=True, default=default_console_file)
+@click.option('--coredump-file', type=click.File('wb', 'utf-8', lazy=True), show_default=True, default=default_coredump_file)
 @click.option('--jlink-sn', '-n', type=int, metavar='SERIAL_NUMBER', help='JLink serial number')
 @click.option('--jlink-speed', type=int, metavar="SPEED", help='JLink clock speed in kHz', default=DEFAULT_JLINK_SPEED_KHZ, show_default=True)
 @click.pass_context
-def command_console(ctx, reset, latency, history_file, console_file, jlink_sn, jlink_speed):
+def command_console(ctx, reset, latency, history_file, console_file, coredump_file, jlink_sn, jlink_speed):
     '''Start interactive console for shell and logging.'''
     logger.remove(2)  # Remove stderr logger
 
@@ -112,7 +114,7 @@ def command_console(ctx, reset, latency, history_file, console_file, jlink_sn, j
         if reset:
             prog.reset()
             prog.go()
-        c = Console(prog, history_file, console_file, latency=latency)
+        c = Console(prog, history_file, console_file, coredump_file, latency=latency)
 
         click.echo('TIP: After J-Link connection, it is crucial to power cycle the target device; otherwise, the CPU debug mode results in a permanently increased power consumption.')
 

src/hardwario/chester/console.py

@@ -30,6 +30,7 @@
 from prompt_toolkit.validation import Validator, ValidationError
 from prompt_toolkit.layout.dimension import LayoutDimension
 from .nrfjprog import NRFJProg, NRFJProgRTTNoChannels, NRFJProgException
+from .utils import Coredump
 
 
 def get_time():
@@ -78,7 +79,7 @@ def validate(self, document: Document) -> None:
 
 class Console:
 
-    def __init__(self, prog: NRFJProg, history_file, console_file, latency=50):
+    def __init__(self, prog: NRFJProg, history_file, console_file, coredump_file, latency=50):
         self.exception = None
         self.show_status_bar = True
         self.scroll_to_end = True
@@ -292,6 +293,8 @@ def _(event):
             }, priority=Priority.MOST_PRECISE)
         )
 
+        coredump = Coredump()
+
         rtt_read_delay = latency / 1000.0
 
         if is_old:
@@ -363,7 +366,15 @@ async def task_rtt_read():
                                         continue
                                     console_file.write(get_time() + (' # ' if channel == 'Logger' else ' > '))
                                     console_file.write(sline)
-                                    console_file.write('\n')
+
+                                    coredump.feed_line(sline)
+                                    if coredump.has_end or coredump.has_error:
+                                        logger.info(f'Writing coredump to file, size: {len(coredump.data)}')
+                                        coredump_file.open()
+                                        coredump_file.write(coredump.data)
+                                        coredump_file.flush()
+                                        coredump.reset()
+
                                 console_file.flush()
 
                                 line = line.replace('\r', '')
@@ -373,7 +384,10 @@ async def task_rtt_read():
                                         buffer.cursor_position = len(buffer.text)
                                     buffer._text_changed()
 
-                            await asyncio.sleep(rtt_read_delay)
+                            if coredump.has_begin:
+                                await asyncio.sleep(0.01)
+                            else:
+                                await asyncio.sleep(rtt_read_delay)
 
         console_file.write(f'{ "*" * 80 }\n')
 

src/hardwario/chester/utils.py

@@ -3,8 +3,10 @@
 import hashlib
 import click
 import requests
+import binascii
 from loguru import logger
 
+
 DEFAULT_CACHE_PATH = expanduser("~/.hardwario/chester/cache")
 
 
@@ -57,3 +59,62 @@ def download_url(url, filename=None, cache_path=DEFAULT_CACHE_PATH):
                     f.write(data)
                     bar.update(dl)
     return filename
+
+
+COREDUMP_PREFIX_STR = "#CD:"
+COREDUMP_BEGIN_STR = COREDUMP_PREFIX_STR + "BEGIN#"
+COREDUMP_END_STR = COREDUMP_PREFIX_STR + "END#"
+COREDUMP_ERROR_STR = COREDUMP_PREFIX_STR + "ERROR CANNOT DUMP#"
+
+
+class Coredump:
+    def __init__(self):
+        self.has_begin = False
+        self.has_end = False
+        self.has_error = False
+        self.data = b''
+
+    def feed_line(self, line: str):
+        line = line.strip()
+        if not line:
+            return
+
+        if line.find(COREDUMP_BEGIN_STR) >= 0:
+            self.has_begin = True
+            self.data = b''
+            return
+
+        elif line.find(COREDUMP_END_STR) >= 0:
+            self.has_end = True
+            return
+
+        elif line.find(COREDUMP_ERROR_STR) >= 0:
+            self.has_error = True
+            return
+
+        if not self.has_begin:
+            return
+
+        prefix_idx = line.find(COREDUMP_PREFIX_STR)
+        if prefix_idx < 0:
+            self.has_end = True
+            self.has_error = True
+            return
+
+        if self.has_end:
+            raise Exception("Coredump already finished")
+
+        hex_str = line[prefix_idx + len(COREDUMP_PREFIX_STR):]
+
+        try:
+            self.data += binascii.unhexlify(hex_str)
+        except Exception as e:
+            logger.error("Cannot parse coredump hex_str: {}".format(hex_str))
+            self.has_error = True
+            self.has_end = True
+
+    def reset(self):
+        self.has_begin = False
+        self.has_end = False
+        self.has_error = False
+        self.data = b''