Speed up screencap via gzip-compressed framebuffer with host-side PNG encoding

.github/workflows/python-package.yml

@@ -15,7 +15,7 @@ env:
 jobs:
   build:
 
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
     strategy:
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12']

androidtv/adb_manager/adb_manager_async.py

@@ -439,6 +439,33 @@ async def shell(self, cmd):
             _LOGGER.debug("Sending command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
             return await self._adb.shell(cmd)
 
+    async def shell_binary(self, cmd):
+        """Send an ADB command using the Python ADB implementation and return raw bytes.
+
+        Parameters
+        ----------
+        cmd : str
+            The ADB command to be sent
+
+        Returns
+        -------
+        bytes, None
+            The raw response from the device, if there is a response
+
+        """
+        if not self.available:
+            _LOGGER.debug(
+                "ADB binary command not sent to %s:%d because adb-shell connection is not established: %s",
+                self.host,
+                self.port,
+                cmd,
+            )
+            return None
+
+        async with _acquire(self._adb_lock):
+            _LOGGER.debug("Sending binary command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
+            return await self._adb.shell(cmd, decode=False)
+
 
 class ADBServerAsync(object):
     """A manager for ADB connections that uses an ADB server.

androidtv/adb_manager/adb_manager_sync.py

@@ -321,6 +321,33 @@ def shell(self, cmd):
             _LOGGER.debug("Sending command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
             return self._adb.shell(cmd)
 
+    def shell_binary(self, cmd):
+        """Send an ADB command using the Python ADB implementation and return raw bytes.
+
+        Parameters
+        ----------
+        cmd : str
+            The ADB command to be sent
+
+        Returns
+        -------
+        bytes, None
+            The raw response from the device, if there is a response
+
+        """
+        if not self.available:
+            _LOGGER.debug(
+                "ADB binary command not sent to %s:%d because adb-shell connection is not established: %s",
+                self.host,
+                self.port,
+                cmd,
+            )
+            return None
+
+        with _acquire(self._adb_lock):
+            _LOGGER.debug("Sending binary command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
+            return self._adb.shell(cmd, decode=False)
+
 
 class ADBServerSync(object):
     """A manager for ADB connections that uses an ADB server.

androidtv/basetv/basetv_async.py

@@ -3,11 +3,20 @@
 ADB Debugging must be enabled.
 """
 
+import asyncio
 import logging
+from time import monotonic
 
 from .basetv import BaseTV
 from .. import constants
 from ..adb_manager.adb_manager_async import ADBPythonAsync, ADBServerAsync
+from ..framebuffer import (
+    GZIP_SCREENCAP_COMMAND,
+    RAW_SCREENCAP_COMMAND,
+    ScreencapError,
+    decode_framebuffer,
+    decode_gzip_screencap,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -85,6 +94,7 @@ def __init__(
             adb = ADBServerAsync(host, port, adb_server_ip, adb_server_port)
 
         BaseTV.__init__(self, adb, host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules)
+        self._screencap_method = None
 
     # ======================================================================= #
     #                                                                         #
@@ -145,16 +155,108 @@ async def adb_push(self, local_path, device_path):
     async def adb_screencap(self):
         """Take a screencap.
 
-        This calls :py:meth:`androidtv.adb_manager.adb_manager_async.ADBPythonAsync.screencap` or :py:meth:`androidtv.adb_manager.adb_manager_async.ADBServerAsync.screencap`,
-        depending on whether the Python ADB implementation or an ADB server is used for communicating with the device.
+        Returns the screen as a binary .png image, or ``None`` if the device did not return anything.
+
+        Note: there are three capture paths. Plain ``screencap`` and ``screencap | gzip -1`` are read
+        back as raw bytes (:py:meth:`androidtv.adb_manager.adb_manager_async.ADBPythonAsync.shell_binary`)
+        with the PNG encoded on the host (the host-side decode runs in an executor); or the device encodes
+        the PNG itself via device-side ``screencap -p``
+        (:py:meth:`androidtv.adb_manager.adb_manager_async.ADBPythonAsync.screencap` or
+        :py:meth:`androidtv.adb_manager.adb_manager_async.ADBServerAsync.screencap`, depending on whether the
+        Python ADB implementation or an ADB server is used for communicating with the device). The ADB
+        server path has no binary shell, so it uses only device-side ``screencap -p``. Neither path is
+        always faster. The first call after :meth:`adb_connect` times each binary-safe candidate back to
+        back, discards any candidate that fails validation, pins the fastest valid one, and returns its
+        image; later calls use only the pinned method. If the pinned binary method later fails, it falls
+        back to device-side ``screencap -p`` and pins that for good.
 
         Returns
         -------
-        bytes
-            The screencap as a binary .png image
+        bytes, None
+            The screencap as a binary .png image, or ``None`` if the device did not return anything
 
         """
-        return await self._adb.screencap()
+        method = getattr(self, "_screencap_method", None)
+        if method == "native":
+            return await self._adb.screencap()
+        if method == "raw":
+            return await self._raw_screencap()
+        if method == "gzip":
+            return await self._gzip_screencap()
+        return await self._calibrate_screencap()
+
+    async def _binary_shell(self, cmd):
+        """Run a shell command and return raw bytes, or ``None`` when the path has no binary shell."""
+        shell_binary = getattr(self._adb, "shell_binary", None)
+        if shell_binary is None:
+            return None
+        return await shell_binary(cmd)
+
+    async def _raw_screencap(self):
+        """Run the raw path, falling back to device-side ``screencap -p`` on any decode failure."""
+        data = await self._binary_shell(RAW_SCREENCAP_COMMAND)
+        if not isinstance(data, bytes) or not data:
+            self._screencap_method = "native"
+            return await self._adb.screencap()
+        try:
+            return await asyncio.get_running_loop().run_in_executor(None, decode_framebuffer, data)
+        except ScreencapError as err:
+            _LOGGER.debug("Falling back to screencap -p: %s", err)
+            self._screencap_method = "native"
+            return await self._adb.screencap()
+
+    async def _gzip_screencap(self):
+        """Run the gzip path, falling back to device-side ``screencap -p`` on any decode failure."""
+        data = await self._binary_shell(GZIP_SCREENCAP_COMMAND)
+        if not isinstance(data, bytes) or not data:
+            self._screencap_method = "native"
+            return await self._adb.screencap()
+        try:
+            return await asyncio.get_running_loop().run_in_executor(None, decode_gzip_screencap, data)
+        except ScreencapError as err:
+            _LOGGER.debug("Falling back to screencap -p: %s", err)
+            self._screencap_method = "native"
+            return await self._adb.screencap()
+
+    async def _calibrate_screencap(self):
+        """Time each binary-safe candidate back to back, pin the fastest valid one, return its image."""
+        loop = asyncio.get_running_loop()
+        results = {}
+        binary_shell = getattr(self._adb, "shell_binary", None)
+        if binary_shell is None:
+            _LOGGER.debug("ADB server has no binary shell; using screencap -p")
+            self._screencap_method = "native"
+            return await self._adb.screencap()
+        for name, cmd, decode in (
+            ("raw", RAW_SCREENCAP_COMMAND, decode_framebuffer),
+            ("gzip", GZIP_SCREENCAP_COMMAND, decode_gzip_screencap),
+        ):
+            start = monotonic()
+            data = await binary_shell(cmd)
+            if isinstance(data, bytes) and data:
+                try:
+                    image = await loop.run_in_executor(None, decode, data)
+                except ScreencapError as err:
+                    _LOGGER.debug("Screencap candidate %r failed validation: %s", name, err)
+                else:
+                    results[name] = (monotonic() - start, image)
+                    continue
+            _LOGGER.debug("Screencap candidate %r returned no data", name)
+        start = monotonic()
+        native_image = await self._adb.screencap()
+        native_dt = monotonic() - start
+        if native_image is not None:
+            results["native"] = (native_dt, native_image)
+        _LOGGER.debug(
+            "Screencap calibration: %s",
+            ", ".join(f"{name} {dt:.3f}s" for name, (dt, _) in sorted(results.items(), key=lambda kv: kv[1][0])),
+        )
+        if not results:
+            self._screencap_method = "native"
+            return None
+        name = min(results, key=lambda key: results[key][0])
+        self._screencap_method = name
+        return results[name][1]
 
     async def adb_connect(
         self,
@@ -179,6 +281,7 @@ async def adb_connect(
             Whether or not the connection was successfully established and the device is available
 
         """
+        self._screencap_method = None
         if isinstance(self._adb, ADBPythonAsync):
             return await self._adb.connect(log_errors, auth_timeout_s, transport_timeout_s)
         return await self._adb.connect(log_errors)

androidtv/basetv/basetv_sync.py

@@ -4,9 +4,17 @@
 """
 
 import logging
+from time import monotonic
 
 from .basetv import BaseTV
 from .. import constants
+from ..framebuffer import (
+    GZIP_SCREENCAP_COMMAND,
+    RAW_SCREENCAP_COMMAND,
+    ScreencapError,
+    decode_framebuffer,
+    decode_gzip_screencap,
+)
 from ..adb_manager.adb_manager_sync import ADBPythonSync, ADBServerSync
 
 _LOGGER = logging.getLogger(__name__)
@@ -85,6 +93,7 @@ def __init__(
             adb = ADBServerSync(host, port, adb_server_ip, adb_server_port)
 
         BaseTV.__init__(self, adb, host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules)
+        self._screencap_method = None
 
     # ======================================================================= #
     #                                                                         #
@@ -145,16 +154,106 @@ def adb_push(self, local_path, device_path):
     def adb_screencap(self):
         """Take a screencap.
 
-        This calls :py:meth:`androidtv.adb_manager.adb_manager_sync.ADBPythonSync.screencap` or :py:meth:`androidtv.adb_manager.adb_manager_sync.ADBServerSync.screencap`,
-        depending on whether the Python ADB implementation or an ADB server is used for communicating with the device.
+        Returns the screen as a binary .png image, or ``None`` if the device did not return anything.
+
+        Note: there are three capture paths. Plain ``screencap`` and ``screencap | gzip -1`` are read
+        back as raw bytes (:py:meth:`androidtv.adb_manager.adb_manager_sync.ADBPythonSync.shell_binary`)
+        with the PNG encoded on the host; or the device encodes the PNG itself via device-side
+        ``screencap -p`` (:py:meth:`androidtv.adb_manager.adb_manager_sync.ADBPythonSync.screencap`
+        or :py:meth:`androidtv.adb_manager.adb_manager_sync.ADBServerSync.screencap`, depending on whether the
+        Python ADB implementation or an ADB server is used for communicating with the device). The ADB
+        server path has no binary shell, so it uses only device-side ``screencap -p``. Neither path is
+        always faster. The first call after :meth:`adb_connect` times each binary-safe candidate back to
+        back, discards any candidate that fails validation, pins the fastest valid one, and returns its
+        image; later calls use only the pinned method. If the pinned binary method later fails, it falls
+        back to device-side ``screencap -p`` and pins that for good.
 
         Returns
         -------
-        bytes
-            The screencap as a binary .png image
+        bytes, None
+            The screencap as a binary .png image, or ``None`` if the device did not return anything
 
         """
-        return self._adb.screencap()
+        method = getattr(self, "_screencap_method", None)
+        if method == "native":
+            return self._adb.screencap()
+        if method == "raw":
+            return self._raw_screencap()
+        if method == "gzip":
+            return self._gzip_screencap()
+        return self._calibrate_screencap()
+
+    def _binary_shell(self, cmd):
+        """Run a shell command and return raw bytes, or ``None`` when the path has no binary shell."""
+        shell_binary = getattr(self._adb, "shell_binary", None)
+        if shell_binary is None:
+            return None
+        return shell_binary(cmd)
+
+    def _raw_screencap(self):
+        """Run the raw path, falling back to device-side ``screencap -p`` on any decode failure."""
+        data = self._binary_shell(RAW_SCREENCAP_COMMAND)
+        if not isinstance(data, bytes) or not data:
+            self._screencap_method = "native"
+            return self._adb.screencap()
+        try:
+            return decode_framebuffer(data)
+        except ScreencapError as err:
+            _LOGGER.debug("Falling back to screencap -p: %s", err)
+            self._screencap_method = "native"
+            return self._adb.screencap()
+
+    def _gzip_screencap(self):
+        """Run the gzip path, falling back to device-side ``screencap -p`` on any decode failure."""
+        data = self._binary_shell(GZIP_SCREENCAP_COMMAND)
+        if not isinstance(data, bytes) or not data:
+            self._screencap_method = "native"
+            return self._adb.screencap()
+        try:
+            return decode_gzip_screencap(data)
+        except ScreencapError as err:
+            _LOGGER.debug("Falling back to screencap -p: %s", err)
+            self._screencap_method = "native"
+            return self._adb.screencap()
+
+    def _calibrate_screencap(self):
+        """Time each binary-safe candidate back to back, pin the fastest valid one, return its image."""
+        results = {}
+        binary_shell = getattr(self._adb, "shell_binary", None)
+        if binary_shell is None:
+            _LOGGER.debug("ADB server has no binary shell; using screencap -p")
+            self._screencap_method = "native"
+            return self._adb.screencap()
+        for name, cmd, decode in (
+            ("raw", RAW_SCREENCAP_COMMAND, decode_framebuffer),
+            ("gzip", GZIP_SCREENCAP_COMMAND, decode_gzip_screencap),
+        ):
+            start = monotonic()
+            data = binary_shell(cmd)
+            if isinstance(data, bytes) and data:
+                try:
+                    image = decode(data)
+                except ScreencapError as err:
+                    _LOGGER.debug("Screencap candidate %r failed validation: %s", name, err)
+                else:
+                    results[name] = (monotonic() - start, image)
+                    continue
+            _LOGGER.debug("Screencap candidate %r returned no data", name)
+        start = monotonic()
+        native_image = self._adb.screencap()
+        native_dt = monotonic() - start
+        if native_image is not None:
+            results["native"] = (native_dt, native_image)
+        _LOGGER.debug(
+            "Screencap calibration: %s",
+            ", ".join(f"{name} {dt:.3f}s" for name, (dt, _) in sorted(results.items(), key=lambda kv: kv[1][0])),
+        )
+        if not results:
+            self._screencap_method = "native"
+            return None
+        name = min(results, key=lambda key: results[key][0])
+        self._screencap_method = name
+        return results[name][1]
 
     def adb_connect(
         self,
@@ -179,6 +278,7 @@ def adb_connect(
             Whether or not the connection was successfully established and the device is available
 
         """
+        self._screencap_method = None
         if isinstance(self._adb, ADBPythonSync):
             return self._adb.connect(log_errors, auth_timeout_s, transport_timeout_s)
         return self._adb.connect(log_errors)