Add AdbDevice._streaming_service and AdbDevice.streaming_shell

adb_shell/adb_device.py

@@ -46,6 +46,7 @@
     * :meth:`AdbDevice._send`
     * :meth:`AdbDevice._service`
     * :meth:`AdbDevice._streaming_command`
+    * :meth:`AdbDevice._streaming_service`
     * :meth:`AdbDevice._write`
     * :attr:`AdbDevice.available`
     * :meth:`AdbDevice.close`
@@ -55,6 +56,7 @@
     * :meth:`AdbDevice.push`
     * :meth:`AdbDevice.shell`
     * :meth:`AdbDevice.stat`
+    * :meth:`AdbDevice.streaming_shell`
 
 * :class:`AdbDeviceTcp`
 
@@ -402,6 +404,38 @@ def _service(self, service, command, timeout_s=None, total_timeout_s=constants.D
             return b''.join(self._streaming_command(service, command, adb_info)).decode('utf8')
         return b''.join(self._streaming_command(service, command, adb_info))
 
+    def _streaming_service(self, service, command, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL_TIMEOUT_S, decode=True):
+        """Send an ADB command to the device, yielding each line of output.
+
+        Parameters
+        ----------
+        service : bytes
+            The ADB service to talk to (e.g., ``b'shell'``)
+        command : bytes
+            The command that will be sent
+        timeout_s : float, None
+            Timeout in seconds for sending and receiving packets, or ``None``; see :meth:`BaseHandle.bulk_read() <adb_shell.handle.base_handle.BaseHandle.bulk_read>`
+            and :meth:`BaseHandle.bulk_write() <adb_shell.handle.base_handle.BaseHandle.bulk_write>`
+        total_timeout_s : float
+            The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`AdbDevice._read`
+        decode : bool
+            Whether to decode the output to utf8 before returning
+
+        Yields
+        -------
+        bytes, str
+            The line-by-line output of the ADB command as a string if ``decode`` is True, otherwise as bytes.
+
+        """
+        adb_info = _AdbTransactionInfo(None, None, timeout_s, total_timeout_s)
+        stream = self._streaming_command(service, command, adb_info)
+        if decode:
+            for line in (stream_line.decode('utf8') for stream_line in stream):
+                yield line
+        else:
+            for line in stream:
+                yield line
+
     def shell(self, command, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL_TIMEOUT_S, decode=True):
         """Send an ADB shell command to the device.
 
@@ -425,6 +459,30 @@ def shell(self, command, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL
         """
         return self._service(b'shell', command.encode('utf8'), timeout_s, total_timeout_s, decode)
 
+    def streaming_shell(self, command, timeout_s=None, total_timeout_s=constants.DEFAULT_TOTAL_TIMEOUT_S, decode=True):
+        """Send an ADB shell command to the device, yielding each line of output.
+
+        Parameters
+        ----------
+        command : str
+            The shell command that will be sent
+        timeout_s : float, None
+            Timeout in seconds for sending and receiving packets, or ``None``; see :meth:`BaseHandle.bulk_read() <adb_shell.handle.base_handle.BaseHandle.bulk_read>`
+            and :meth:`BaseHandle.bulk_write() <adb_shell.handle.base_handle.BaseHandle.bulk_write>`
+        total_timeout_s : float
+            The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`AdbDevice._read`
+        decode : bool
+            Whether to decode the output to utf8 before returning
+
+        Yields
+        -------
+        bytes, str
+            The line-by-line output of the ADB shell command as a string if ``decode`` is True, otherwise as bytes.
+
+        """
+        for line in self._streaming_service(b'shell', command.encode('utf8'), timeout_s, total_timeout_s, decode):
+            yield line
+
     # ======================================================================= #
     #                                                                         #
     #                                 FileSync                                #

tests/test_adb_device.py

@@ -397,6 +397,40 @@ def test_issue29(self):
         self.device.shell('Android TV update command')
         self.device.shell('Android TV update command')
 
+    # ======================================================================= #
+    #                                                                         #
+    #                      `streaming_shell` tests                            #
+    #                                                                         #
+    # ======================================================================= #
+    def test_streaming_shell_decode(self):
+        self.assertTrue(self.device.connect())
+
+        # Provide the `bulk_read` return values
+        self.device._handle._bulk_read = join_messages(
+            AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
+            AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=b'ABC'),
+            AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=b'123'),
+        )
+
+        generator = self.device.streaming_shell('TEST', decode=True)
+        self.assertEqual('ABC', next(generator))
+        self.assertEqual('123', next(generator))
+
+    def test_streaming_shell_dont_decode(self):
+        self.assertTrue(self.device.connect())
+
+        # Provide the `bulk_read` return values
+        self.device._handle._bulk_read = join_messages(
+            AdbMessage(command=constants.OKAY, arg0=1, arg1=1, data=b'\x00'),
+            AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=b'ABC'),
+            AdbMessage(command=constants.WRTE, arg0=1, arg1=1, data=b'123'),
+        )
+
+        generator = self.device.streaming_shell('TEST', decode=False)
+        self.assertEqual(b'ABC', next(generator))
+        self.assertEqual(b'123', next(generator))
+
+
     # ======================================================================= #
     #                                                                         #
     #                         `filesync` tests                                #