[otci] allow callbacks on all read lines (#6851)

This allows for registering a callback on all lines read from a
device, making it possible to react to output from asynchronous cli
commands while concurrently executing new commands on the same
device. Normally, executing those new commands would result in the
async command output being lost.

Known caveats:

1. The same limitations as with wait apply to OtbrSshCommandRunner
(you can't get async output).

2. For OtCliCommandRunner, any commands sent to the device in the
callback need to be offloaded to another thread. Otherwise the read
routine thread will be blocked, and the command will timeout.

tools/otci/otci/command_handlers.py

@@ -32,7 +32,7 @@ import re
 import threading
 import time
 from abc import abstractmethod
-from typing import Union, List, Pattern
+from typing import Any, Callable, Optional, Union, List, Pattern
 
 from .connectors import OtCliHandler
 from .errors import ExpectLineTimeoutError, CommandError
@@ -66,6 +66,16 @@ class OTCommandHandler:
         """
         pass
 
+    @abstractmethod
+    def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
+        """Method set_line_read_callback should register a callback that will be called for every line
+        output by the OT CLI.
+
+        This is useful for handling asynchronous command output while still being able to execute
+        other commands.
+        """
+        pass
+
 
 class OtCliCommandRunner(OTCommandHandler):
     __PATTERN_COMMAND_DONE_OR_ERROR = re.compile(
@@ -82,6 +92,7 @@ class OtCliCommandRunner(OTCommandHandler):
         self.__otcli: OtCliHandler = otcli
         self.__is_spinel_cli = is_spinel_cli
         self.__expect_command_echoback = not self.__is_spinel_cli
+        self.__line_read_callback = None
 
         self.__pending_lines = queue.Queue()
         self.__should_close = threading.Event()
@@ -127,6 +138,9 @@ class OtCliCommandRunner(OTCommandHandler):
         self.__should_close.set()
         self.__otcli.close()
 
+    def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
+        self.__line_read_callback = callback
+
     #
     # Private methods
     #
@@ -169,6 +183,9 @@ class OtCliCommandRunner(OTCommandHandler):
             if line.startswith('> '):
                 line = line[2:]
 
+            if self.__line_read_callback is not None:
+                self.__line_read_callback(line)
+
             logging.debug('%s: %s', self.__otcli, line)
 
             if not OtCliCommandRunner.__PATTERN_LOG_LINE.match(line):
@@ -186,6 +203,8 @@ class OtbrSshCommandRunner(OTCommandHandler):
         self.__ssh = paramiko.SSHClient()
         self.__ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 
+        self.__line_read_callback = None
+
         try:
             self.__ssh.connect(host,
                                port=port,
@@ -214,6 +233,10 @@ class OtbrSshCommandRunner(OTCommandHandler):
 
         output = [l.rstrip('\r\n') for l in cmd_out.readlines()]
 
+        if self.__line_read_callback is not None:
+            for line in output:
+                self.__line_read_callback(line)
+
         if cmd in ('reset', 'factoryreset'):
             self.wait(3)
 
@@ -225,3 +248,6 @@ class OtbrSshCommandRunner(OTCommandHandler):
     def wait(self, duration: float) -> List[str]:
         time.sleep(duration)
         return []
+
+    def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
+        self.__line_read_callback = callback

tools/otci/otci/otci.py

@@ -30,7 +30,7 @@ import ipaddress
 import logging
 import re
 from collections import Counter
-from typing import List, Collection, Union, Tuple, Optional, Dict, Pattern, Any
+from typing import Callable, List, Collection, Union, Tuple, Optional, Dict, Pattern, Any
 
 from . import connectors
 from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner
@@ -118,6 +118,10 @@ class OTCI(object):
         """Set the logger for the OTCI instance, or None to disable logging."""
         self.__logger = logger
 
+    def set_line_read_callback(self, callback: Optional[Callable[[str], Any]]):
+        """Set the callback that will be called for each line output by the CLI."""
+        self.__otcmd.set_line_read_callback(callback)
+
     #
     # Constant properties
     #