Implement serial terminal

Author: rwalton-arm

news/20201111104222.feature

@@ -0,0 +1 @@
+Implement sterm command.

setup.py

@@ -48,6 +48,7 @@
         "pyudev; platform_system=='Linux'",
         "typing-extensions",
         "Jinja2",
+        "pyserial",
     ],
     license="Apache 2.0",
     long_description_content_type="text/markdown",

src/mbed_tools/sterm/__init__.py

@@ -0,0 +1,5 @@
+#
+# Copyright (C) 2020 Arm Mbed. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+"""Package containing sterm functionality."""

src/mbed_tools/sterm/terminal.py

@@ -0,0 +1,134 @@
+#
+# Copyright (C) 2020 Arm Mbed. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+"""Serial terminal implementation based on pyserial.tools.miniterm.
+
+The Mbed serial terminal makes the following modifications to the default Miniterm.
+* Custom help menu text
+* Constrained set of menu keys
+* CTRL-H to show help
+* CTRL-B sends serial break to the target
+
+To start the terminal clients should call the "run" function, this is the entry point to the module.
+"""
+from typing import Any
+
+from serial import Serial
+from serial.tools.miniterm import Miniterm
+
+
+def run(port: str, baud: int, echo: bool = True) -> None:
+    """Run the serial terminal.
+
+    This function is blocking as it waits for the terminal thread to finish executing before returning.
+
+    Args:
+        port: The serial port to open a terminal on.
+        baud: Serial baud rate.
+        echo: Echo user input back to the console.
+    """
+    term = SerialTerminal(Serial(port=port, baudrate=str(baud)), echo=echo)
+    term.start()
+
+    try:
+        term.join(True)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        term.join()
+        term.close()
+
+
+class SerialTerminal(Miniterm):
+    """An implementation of Miniterm that implements the additional Mbed terminal functionality.
+
+    Overrides the `writer` method to implement modified menu key handling behaviour.
+    Overrides the Miniterm::get_help_text method to return the Mbed custom help text.
+    Adds a `reset` method so users can send a reset signal to the device.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Set the rx/tx encoding and special characters."""
+        super().__init__(*args, **kwargs)
+        self.exit_character = CTRL_C
+        self.menu_character = CTRL_T
+        self.reset_character = CTRL_B
+        self.help_character = CTRL_H
+        self.set_rx_encoding("UTF-8")
+        self.set_tx_encoding("UTF-8")
+
+    def reset(self) -> None:
+        """Send a reset signal."""
+        self.serial.sendBreak()
+
+    def get_help_text(self) -> str:
+        """Return the text displayed when the user requests help."""
+        return HELP_TEXT
+
+    def writer(self) -> None:
+        """Implements terminal behaviour."""
+        menu_active = False
+        while self.alive:
+            try:
+                input_key = self.console.getkey()
+            except KeyboardInterrupt:
+                input_key = self.exit_character
+
+            if (menu_active and input_key in VALID_MENU_KEYS) or (input_key == self.help_character):
+                self.handle_menu_key(input_key)
+                menu_active = False
+
+            elif input_key == self.menu_character:
+                menu_active = True
+
+            elif input_key == self.reset_character:
+                self.reset()
+
+            elif input_key == self.exit_character:
+                self.stop()
+                break
+
+            else:
+                self._write_transformed_char(input_key)
+
+                if self.echo:
+                    self._echo_transformed_char(input_key)
+
+    def _write_transformed_char(self, text: str) -> None:
+        for transformation in self.tx_transformations:
+            text = transformation.tx(text)
+
+        self.serial.write(self.tx_encoder.encode(text))
+
+    def _echo_transformed_char(self, text: str) -> None:
+        for transformation in self.tx_transformations:
+            text = transformation.echo(text)
+
+        self.console.write(text)
+
+
+CTRL_B = "\x02"
+CTRL_C = "\x03"
+CTRL_H = "\x08"
+CTRL_T = "\x14"
+VALID_MENU_KEYS = ("p", "b", "\t", "\x01", "\x04", "\x05", "\x06", "\x0c", CTRL_C, CTRL_T)
+HELP_TEXT = """--- Mbed Serial Terminal
+--- Based on miniterm from pySerial
+---
+--- Ctrl+b     Send Break (reset target)
+--- Ctrl+c     Exit terminal
+--- Ctrl+h     Help
+--- Ctrl+t     Menu escape key, followed by:
+---    p       Change COM port
+---    b       Change baudrate
+---    Tab     Show detailed terminal info
+---    Ctrl+a  Change encoding (default UTF-8)
+---    Ctrl+f  Edit filters
+---    Ctrl+e  Toggle local echo
+---    Ctrl+l  Toggle EOL
+---    Ctrl+r  Toggle RTS
+---    Ctrl+d  Toggle DTR
+---    Ctrl+c  Send control character to remote
+---    Ctrl+t  Send control character to remote
+"""

tests/sterm/test_terminal.py

@@ -0,0 +1,225 @@
+#
+# Copyright (C) 2020 Arm Mbed. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+from unittest import mock
+
+import sys
+
+import pytest
+
+from mbed_tools.sterm import terminal
+
+
+@pytest.fixture
+def mock_serial():
+    with mock.patch("mbed_tools.sterm.terminal.Serial") as serial:
+        yield serial
+
+
+@pytest.fixture
+def mock_console():
+    with mock.patch("serial.tools.miniterm.Console") as console:
+        yield console
+
+
+@pytest.fixture
+def mock_sterm():
+    with mock.patch("mbed_tools.sterm.terminal.SerialTerminal") as sterm:
+        yield sterm
+
+
+def test_initialises_serial_port(mock_sterm, mock_serial):
+    port = "tty.1111"
+    baud = 9600
+
+    terminal.run(port, baud)
+
+    mock_serial.assert_called_once_with(port=port, baudrate=str(baud))
+
+
+def test_initialises_sterm(mock_sterm, mock_serial):
+    port = "tty.1111"
+    baud = "9600"
+
+    terminal.run(port, baud)
+
+    mock_sterm.assert_called_once_with(mock_serial(), echo=True)
+
+
+def test_starts_sterm_thread(mock_sterm, mock_serial):
+    terminal.run("tty.122", 9600)
+
+    mock_sterm().start.assert_called_once()
+
+
+def test_joins_tx_and_rx_threads(mock_sterm, mock_serial):
+    terminal.run("tty.122", 9600)
+
+    mock_sterm().join.assert_any_call(True)
+
+
+def test_joins_tx_thread_after_keyboard_interrupt(mock_sterm, mock_serial):
+    mock_sterm().join.side_effect = (KeyboardInterrupt(), None)
+
+    terminal.run("tty.122", 9600)
+
+    mock_sterm().join.assert_called_with()
+
+
+def test_closes_sterm(mock_sterm, mock_serial):
+    terminal.run("tty.122", 9600)
+
+    mock_sterm().close.assert_called_once()
+
+
+def test_closes_sterm_after_exception(mock_sterm, mock_serial):
+    mock_sterm().join.side_effect = (Exception(), None)
+    with pytest.raises(Exception):
+        terminal.run("tty.122", 9600)
+
+    mock_sterm().close.assert_called_once()
+
+
+def test_closes_sterm_after_keyboard_interrupt(mock_sterm, mock_serial):
+    mock_sterm().join.side_effect = (KeyboardInterrupt(), None)
+    terminal.run("tty.122", 9600)
+
+    mock_sterm().close.assert_called_once()
+
+
+def test_sets_terminal_special_chars(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+
+    assert term.exit_character == terminal.CTRL_C
+    assert term.menu_character == terminal.CTRL_T
+    assert term.reset_character == terminal.CTRL_B
+    assert term.help_character == terminal.CTRL_H
+
+
+def test_sets_terminal_rx_and_tx_encoding_to_utf8(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+
+    assert term.input_encoding == "UTF-8"
+    assert term.output_encoding == "UTF-8"
+
+
+def test_stops_terminal_when_ctrl_c_received(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    mock_console().getkey.return_value = terminal.CTRL_C
+
+    term.writer()
+
+    assert term.alive is False
+
+
+def test_stops_terminal_on_keyboard_interrupt(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    mock_console().getkey.side_effect = KeyboardInterrupt()
+
+    term.writer()
+
+    assert term.alive is False
+
+
+@pytest.mark.parametrize("menu_key", terminal.VALID_MENU_KEYS)
+def test_handles_valid_menu_key(menu_key, mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.handle_menu_key = mock.Mock()
+    term.alive = True
+    mock_console().getkey.side_effect = (terminal.CTRL_T, menu_key, terminal.CTRL_C)
+
+    term.writer()
+
+    term.handle_menu_key.assert_called_once_with(menu_key)
+
+
+INVALID_MENU_KEYS = tuple(set(chr(i) for i in range(0, 127)) - set(terminal.VALID_MENU_KEYS) - set([terminal.CTRL_H]))
+
+
+@pytest.mark.parametrize("menu_key", INVALID_MENU_KEYS)
+def test_ignores_invalid_menu_key(menu_key, mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.handle_menu_key = mock.Mock()
+    term.alive = True
+    mock_console().getkey.side_effect = (terminal.CTRL_T, menu_key)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    term.handle_menu_key.assert_not_called()
+
+
+def test_reset_sends_serial_break(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+
+    term.reset()
+
+    mock_serial().sendBreak.assert_called_once()
+
+
+def test_ctrl_b_sends_reset_to_serial_port(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    mock_console().getkey.side_effect = (terminal.CTRL_B,)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    mock_serial().sendBreak.assert_called_once()
+
+
+def test_ctrl_h_prints_help_text(mock_serial, mock_console):
+    sys.stderr.write = mock.Mock()
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    mock_console().getkey.side_effect = (terminal.CTRL_H,)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    sys.stderr.write.assert_called_once_with(term.get_help_text())
+
+
+def test_help_text_is_correct(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+
+    assert term.get_help_text() == terminal.HELP_TEXT
+
+
+def test_writes_normal_char_to_serial_output(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    normal_char = "h"
+    mock_console().getkey.side_effect = (normal_char,)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    mock_serial().write.assert_called_once_with(term.tx_encoder.encode(normal_char))
+
+
+def test_echo_to_console_is_default_disabled(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial())
+    term.alive = True
+    normal_char = "h"
+    mock_console().getkey.side_effect = (normal_char,)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    mock_console().write.assert_not_called()
+
+
+def test_echo_to_console_can_be_enabled(mock_serial, mock_console):
+    term = terminal.SerialTerminal(mock_serial(), echo=True)
+    term.alive = True
+    normal_char = "h"
+    mock_console().getkey.side_effect = (normal_char,)
+
+    with pytest.raises(StopIteration):
+        term.writer()
+
+    mock_console().write.assert_called_once_with(normal_char)