west: Add west rtt command

scripts/west-commands.yml

@@ -46,6 +46,9 @@ west-commands:
       - name: attach
         class: Attach
         help: interactively debug a board
+      - name: rtt
+        class: Rtt
+        help: open an rtt shell
   - file: scripts/west_commands/export.py
     commands:
       - name: zephyr-export

scripts/west_commands/debug.py

@@ -74,3 +74,21 @@ def do_add_parser(self, parser_adder):
 
     def do_run(self, my_args, runner_args):
         do_run_common(self, my_args, runner_args)
+
+
+class Rtt(WestCommand):
+
+    def __init__(self):
+        super(Rtt, self).__init__(
+            'rtt',
+            # Keep this in sync with the string in west-commands.yml.
+            'open an rtt shell',
+            "",
+            accepts_unknown_args=True)
+        self.runner_key = 'rtt-runner'  # in runners.yaml
+
+    def do_add_parser(self, parser_adder):
+        return add_parser_common(self, parser_adder)
+
+    def do_run(self, my_args, runner_args):
+        do_run_common(self, my_args, runner_args)

scripts/west_commands/run_common.py

@@ -628,7 +628,8 @@ def filetype(attr):
                         filetype('file_type'),
                         config('gdb'),
                         config('openocd'),
-                        config('openocd_search', []))
+                        config('openocd_search', []),
+                        config('rtt_address'))
 
 def dump_traceback():
     # Save the current exception to a file and return its path.

scripts/west_commands/runners/core.py

@@ -17,18 +17,28 @@
 import logging
 import os
 import platform
+import re
+import selectors
 import shlex
 import shutil
 import signal
+import socket
 import subprocess
-import re
+import sys
 from dataclasses import dataclass, field
 from functools import partial
 from enum import Enum
 from inspect import isabstract
 from typing import Dict, List, NamedTuple, NoReturn, Optional, Set, Type, \
     Union
 
+try:
+    from elftools.elf.elffile import ELFFile
+    ELFTOOLS_MISSING = False
+except ImportError:
+    ELFTOOLS_MISSING = True
+
+
 # Turn on to enable just logging the commands that would be run (at
 # info rather than debug level), without actually running them. This
 # can break runners that are expecting output or if one command
@@ -37,6 +47,27 @@
 
 _logger = logging.getLogger('runners')
 
+# FIXME: I assume this code belongs somewhere else, but i couldn't figure out
+# a good location for it, so i put it here for now
+# We could potentially search for RTT blocks in hex or bin files as well,
+# but since the magic string is "SEGGER RTT", i thought it might be better
+# to avoid, at the risk of false positives.
+def find_rtt_block(elf_file: str) -> Optional[int]:
+    if ELFTOOLS_MISSING:
+        raise RuntimeError('the Python dependency elftools was missing; '
+                           'see the getting started guide for details on '
+                           'how to fix')
+
+    with open(elf_file, 'rb') as f:
+        elffile = ELFFile(f)
+        for sect in elffile.iter_sections('SHT_SYMTAB'):
+            symbols = sect.get_symbol_by_name('_SEGGER_RTT')
+            if symbols is None:
+                continue
+            for s in symbols:
+                return s.entry.get('st_value')
+    return None
+
 
 class _DebugDummyPopen:
 
@@ -219,7 +250,7 @@ def __init__(self, program):
         super().__init__(errno.ENOENT, os.strerror(errno.ENOENT), program)
 
 
-_RUNNERCAPS_COMMANDS = {'flash', 'debug', 'debugserver', 'attach', 'simulate', 'robot'}
+_RUNNERCAPS_COMMANDS = {'flash', 'debug', 'debugserver', 'attach', 'simulate', 'robot', 'rtt'}
 
 @dataclass
 class RunnerCaps:
@@ -231,7 +262,7 @@ class RunnerCaps:
     Available capabilities:
 
     - commands: set of supported commands; default is {'flash',
-      'debug', 'debugserver', 'attach', 'simulate', 'robot'}.
+      'debug', 'debugserver', 'attach', 'simulate', 'robot', 'rtt'}.
 
     - dev_id: whether the runner supports device identifiers, in the form of an
       -i, --dev-id option. This is useful when the user has multiple debuggers
@@ -268,6 +299,9 @@ class RunnerCaps:
       discovered in the build directory.
 
     - hide_load_files: whether the elf/hex/bin file arguments should be hidden.
+
+    - rtt: whether the runner supports SEGGER RTT. This adds a --rtt-address
+      option.
     '''
 
     commands: Set[str] = field(default_factory=lambda: set(_RUNNERCAPS_COMMANDS))
@@ -279,6 +313,8 @@ class RunnerCaps:
     tool_opt: bool = False
     file: bool = False
     hide_load_files: bool = False
+    rtt: bool = False  # This capability exists separately from the rtt command
+                       # to allow other commands to use the rtt address
 
     def __post_init__(self):
         if not self.commands.issubset(_RUNNERCAPS_COMMANDS):
@@ -319,6 +355,7 @@ class RunnerConfig(NamedTuple):
     gdb: Optional[str] = None       # path to a usable gdb
     openocd: Optional[str] = None   # path to a usable openocd
     openocd_search: List[str] = []  # add these paths to the openocd search path
+    rtt_address: Optional[int] = None # address of the rtt control block
 
 
 _YN_CHOICES = ['Y', 'y', 'N', 'n', 'yes', 'no', 'YES', 'NO']
@@ -572,6 +609,13 @@ def add_parser(cls, parser):
                             help=(cls.tool_opt_help() if caps.tool_opt
                                   else argparse.SUPPRESS))
 
+        if caps.rtt:
+            parser.add_argument('--rtt-address', dest='rtt_address',
+                                type=lambda x: int(x, 0),
+                                help="address of RTT control block. If not supplied, it will be autodetected if possible")
+        else:
+            parser.add_argument('--rtt-address', help=argparse.SUPPRESS)
+
         # Runner-specific options.
         cls.do_add_parser(parser)
 
@@ -607,6 +651,8 @@ def create(cls, cfg: RunnerConfig,
             raise ValueError("--file-type requires --file")
         if args.file_type and not caps.file:
             _missing_cap(cls, '--file-type')
+        if args.rtt_address and not caps.rtt:
+            _missing_cap(cls, '--rtt-address')
 
         ret = cls.do_create(cfg, args)
         if args.erase:
@@ -731,6 +777,19 @@ def require(program: str, path: Optional[str] = None) -> str:
             raise MissingProgram(program)
         return ret
 
+    def get_rtt_address(self) -> int | None:
+        '''Helper method for extracting a the RTT control block address.
+
+        If args.rtt_address was supplied, returns that.
+
+        Otherwise, attempt to locate an rtt block in the elf file.
+        If this is not found, None is returned'''
+        if self.cfg.rtt_address is not None:
+            return self.cfg.rtt_address
+        elif self.cfg.elf_file is not None:
+            return find_rtt_block(self.cfg.elf_file)
+        return None
+
     def run_server_and_client(self, server, client, **kwargs):
         '''Run a server that ignores SIGINT, and a client that handles it.
 
@@ -846,3 +905,34 @@ def ensure_output(self, output_type: str) -> None:
 
         # RuntimeError avoids a stack trace saved in run_common.
         raise RuntimeError(err)
+
+    def run_telnet_client(self, host: str, port: int) -> None:
+        '''
+        Run a telnet client for user interaction.
+        '''
+        # If a `nc` command is available, run it, as it will provide the best support for
+        # CONFIG_SHELL_VT100_COMMANDS etc.
+        if shutil.which('nc') is not None:
+            client_cmd = ['nc', host, str(port)]
+            self.run_client(client_cmd)
+            return
+
+        # Otherwise, use a pure python implementation. This will work well for logging,
+        # but input is line based only.
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.connect((host, port))
+        sel = selectors.DefaultSelector()
+        sel.register(sys.stdin, selectors.EVENT_READ)
+        sel.register(sock, selectors.EVENT_READ)
+        while True:
+            events = sel.select()
+            for key, _ in events:
+                if key.fileobj == sys.stdin:
+                    text = sys.stdin.readline()
+                    if text:
+                        sock.send(text.encode())
+
+                elif key.fileobj == sock:
+                    resp = sock.recv(2048)
+                    if resp:
+                        print(resp.decode())

scripts/west_commands/runners/jlink.py

@@ -12,6 +12,8 @@
 import shlex
 import subprocess
 import sys
+import socket
+import time
 import tempfile
 
 from runners.core import ZephyrBinaryRunner, RunnerCaps, FileType
@@ -25,6 +27,7 @@
 
 DEFAULT_JLINK_EXE = 'JLink.exe' if sys.platform == 'win32' else 'JLinkExe'
 DEFAULT_JLINK_GDB_PORT = 2331
+DEFAULT_JLINK_RTT_PORT = 19021
 
 def is_ip(ip):
     try:
@@ -49,6 +52,7 @@ def __init__(self, cfg, device, dev_id=None,
                  gdbserver='JLinkGDBServer',
                  gdb_host='',
                  gdb_port=DEFAULT_JLINK_GDB_PORT,
+                 rtt_port=DEFAULT_JLINK_RTT_PORT,
                  tui=False, tool_opt=[]):
         super().__init__(cfg)
         self.file = cfg.file
@@ -70,6 +74,7 @@ def __init__(self, cfg, device, dev_id=None,
         self.gdb_port = gdb_port
         self.tui_arg = ['-tui'] if tui else []
         self.loader = loader
+        self.rtt_port = rtt_port
 
         self.tool_opt = []
         for opts in [shlex.split(opt) for opt in tool_opt]:
@@ -81,9 +86,9 @@ def name(cls):
 
     @classmethod
     def capabilities(cls):
-        return RunnerCaps(commands={'flash', 'debug', 'debugserver', 'attach'},
+        return RunnerCaps(commands={'flash', 'debug', 'debugserver', 'attach', 'rtt'},
                           dev_id=True, flash_addr=True, erase=True, reset=True,
-                          tool_opt=True, file=True)
+                          tool_opt=True, file=True, rtt=True)
 
     @classmethod
     def dev_id_help(cls) -> str:
@@ -126,6 +131,10 @@ def do_add_parser(cls, parser):
                             dest='reset', nargs=0,
                             action=ToggleAction,
                             help='obsolete synonym for --reset/--no-reset')
+        parser.add_argument('--rtt-client', default='JLinkRTTClient',
+                            help='RTT client, default is JLinkRTTClient')
+        parser.add_argument('--rtt-port', default=DEFAULT_JLINK_RTT_PORT,
+                            help=f'jlink rtt port, defaults to {DEFAULT_JLINK_RTT_PORT}')
 
         parser.set_defaults(reset=False)
 
@@ -142,6 +151,7 @@ def do_create(cls, cfg, args):
                                  loader=args.loader,
                                  gdb_host=args.gdb_host,
                                  gdb_port=args.gdb_port,
+                                 rtt_port=args.rtt_port,
                                  tui=args.tui, tool_opt=args.tool_opt)
 
     def print_gdbserver_message(self):
@@ -248,6 +258,7 @@ def do_run(self, command, **kwargs):
                        '-singlerun'] +
                       (['-nogui'] if self.supports_nogui else []) +
                       (['-rtos', plugin_dir] if rtos else []) +
+                      ['-rtttelnetport', str(self.rtt_port)] +
                       self.tool_opt)
 
         if command == 'flash':
@@ -258,6 +269,27 @@ def do_run(self, command, **kwargs):
             self.require(self.gdbserver)
             self.print_gdbserver_message()
             self.check_call(server_cmd)
+        elif command == 'rtt':
+            self.print_gdbserver_message()
+            server_cmd += ['-nohalt']
+            server_proc = self.popen_ignore_int(server_cmd)
+            try:
+                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                # wait for the port to be open
+                while server_proc.poll() is None:
+                    try:
+                        sock.connect(('localhost', self.rtt_port))
+                        break
+                    except ConnectionRefusedError:
+                        time.sleep(0.1)
+                sock.shutdown(socket.SHUT_RDWR)
+                time.sleep(0.1)
+                self.run_telnet_client('localhost', self.rtt_port)
+            except Exception as e:
+                self.logger.error(e)
+            finally:
+                server_proc.terminate()
+                server_proc.wait()
         else:
             if self.gdb_cmd is None:
                 raise ValueError('Cannot debug; gdb is missing')