From 06e7edef6fcf7645acc510bdc2b515115fd5341c Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Thu, 15 Oct 2026 10:26:36 +0400
Subject: [PATCH 1/7] Speed up screencap via gzip-compressed framebuffer with
 host-side PNG encoding

Capture the raw Android framebuffer gzip-compressed over ADB and encode the PNG on the host with the standard library. The device no longer spends seconds encoding PNG itself. Falls back to screencap -p when gzip or base64 is unavailable or the pixel format is unsupported. Corrupt or truncated data raises ScreencapError.
---
 androidtv/basetv/basetv_async.py |  32 +++++-
 androidtv/basetv/basetv_sync.py  |  30 +++++-
 androidtv/framebuffer.py         | 163 ++++++++++++++++++++++++++++++
 tests/test_basetv_async.py       |   5 +-
 tests/test_basetv_sync.py        |   5 +-
 tests/test_framebuffer.py        | 166 +++++++++++++++++++++++++++++++
 6 files changed, 389 insertions(+), 12 deletions(-)
 create mode 100644 androidtv/framebuffer.py
 create mode 100644 tests/test_framebuffer.py

diff --git a/androidtv/basetv/basetv_async.py b/androidtv/basetv/basetv_async.py
index d901d10c..71cd4877 100644
--- a/androidtv/basetv/basetv_async.py
+++ b/androidtv/basetv/basetv_async.py
@@ -3,11 +3,13 @@
 ADB Debugging must be enabled.
 """
 
+import asyncio
 import logging
 
 from .basetv import BaseTV
 from .. import constants
 from ..adb_manager.adb_manager_async import ADBPythonAsync, ADBServerAsync
+from ..framebuffer import COMPRESSED_SCREENCAP_COMMAND, UnsupportedScreencapFormatError, decode_compressed_screencap
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -145,16 +147,36 @@ 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: the raw framebuffer is transferred gzip-compressed and the PNG is encoded on the host with the
+        standard library (the host-side decode runs in an executor), which is much faster than having the
+        device encode the PNG itself. If the device lacks ``gzip`` or ``base64``, or the framebuffer pixel
+        format is not supported, the device-side ``screencap -p`` path is used instead
+        (: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
         -------
-        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
+
+        Raises
+        ------
+        androidtv.framebuffer.ScreencapError
+            If the compressed screencap is corrupt or truncated
 
         """
-        return await self._adb.screencap()
+        encoded = await self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
+        if not isinstance(encoded, str):
+            return None
+        if not encoded.strip():
+            return await self._adb.screencap()
+        try:
+            return await asyncio.get_running_loop().run_in_executor(None, decode_compressed_screencap, encoded)
+        except UnsupportedScreencapFormatError:
+            return await self._adb.screencap()
 
     async def adb_connect(
         self,
diff --git a/androidtv/basetv/basetv_sync.py b/androidtv/basetv/basetv_sync.py
index 57ad0d84..62fd3cc8 100644
--- a/androidtv/basetv/basetv_sync.py
+++ b/androidtv/basetv/basetv_sync.py
@@ -7,6 +7,7 @@
 
 from .basetv import BaseTV
 from .. import constants
+from ..framebuffer import COMPRESSED_SCREENCAP_COMMAND, UnsupportedScreencapFormatError, decode_compressed_screencap
 from ..adb_manager.adb_manager_sync import ADBPythonSync, ADBServerSync
 
 _LOGGER = logging.getLogger(__name__)
@@ -145,16 +146,35 @@ 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: the raw framebuffer is transferred gzip-compressed and the PNG is encoded on the host with the
+        standard library, which is much faster than having the device encode the PNG itself. If the device
+        lacks ``gzip`` or ``base64``, or the framebuffer pixel format is not supported, the device-side
+        ``screencap -p`` path is used instead (: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
         -------
-        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
+
+        Raises
+        ------
+        androidtv.framebuffer.ScreencapError
+            If the compressed screencap is corrupt or truncated
 
         """
-        return self._adb.screencap()
+        encoded = self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
+        if not isinstance(encoded, str):
+            return None
+        if not encoded.strip():
+            return self._adb.screencap()
+        try:
+            return decode_compressed_screencap(encoded)
+        except UnsupportedScreencapFormatError:
+            return self._adb.screencap()
 
     def adb_connect(
         self,
diff --git a/androidtv/framebuffer.py b/androidtv/framebuffer.py
new file mode 100644
index 00000000..230258fa
--- /dev/null
+++ b/androidtv/framebuffer.py
@@ -0,0 +1,163 @@
+"""Capture Android framebuffers compressed over ADB and encode them as PNG.
+
+``screencap -p`` makes the device CPU encode the PNG, which is slow on many
+TVs. Piping the raw framebuffer through ``gzip`` and ``base64`` is much faster;
+the host then parses the Android framebuffer header and encodes the PNG itself
+using only the standard library.
+"""
+
+import base64
+import gzip
+import struct
+import sys
+import zlib
+from array import array
+
+COMPRESSED_SCREENCAP_COMMAND = (
+    "if command -v gzip >/dev/null && command -v base64 >/dev/null; "
+    "then screencap | gzip -1 | base64; fi"
+)
+
+PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
+
+# Android ``PixelFormat`` values we can decode: format -> bytes per pixel.
+_BYTES_PER_PIXEL = {1: 4, 2: 4, 3: 3, 4: 2, 5: 4}
+
+_EXPAND_5 = bytes((c << 3) | (c >> 2) for c in range(32))
+_EXPAND_6 = bytes((c << 2) | (c >> 4) for c in range(64))
+
+
+class ScreencapError(Exception):
+    """A compressed screencap could not be transferred or decoded."""
+
+
+class UnsupportedScreencapFormatError(ScreencapError):
+    """The framebuffer uses a pixel format this decoder does not handle."""
+
+
+def decode_compressed_screencap(encoded):
+    """Decode the base64-encoded gzip output of :data:`COMPRESSED_SCREENCAP_COMMAND`.
+
+    Parameters
+    ----------
+    encoded : str
+        The text printed by the device shell
+
+    Returns
+    -------
+    bytes
+        The screencap as a binary .png image
+
+    Raises
+    ------
+    ScreencapError
+        If the data is not valid base64/gzip or the framebuffer is truncated
+    UnsupportedScreencapFormatError
+        If the framebuffer pixel format is not supported
+
+    """
+    try:
+        raw = gzip.decompress(base64.b64decode(encoded))
+    except (ValueError, OSError, EOFError, zlib.error) as err:
+        raise ScreencapError(f"Corrupt compressed screencap: {err}") from err
+    return decode_framebuffer(raw)
+
+
+def decode_framebuffer(raw):
+    """Convert a raw Android framebuffer dump (``screencap`` without ``-p``) to PNG.
+
+    The dump starts with little-endian ``width``, ``height`` and ``format``
+    fields. Android 10+ adds a ``dataspace`` field, so the header is either 12
+    or 16 bytes long; the header length is inferred from the total length.
+
+    Parameters
+    ----------
+    raw : bytes
+        The raw framebuffer dump
+
+    Returns
+    -------
+    bytes
+        The screencap as a binary .png image
+
+    """
+    if len(raw) < 12:
+        raise ScreencapError("Truncated screencap header")
+    width, height, pixel_format = struct.unpack_from("<III", raw)
+    if pixel_format not in _BYTES_PER_PIXEL:
+        raise UnsupportedScreencapFormatError(f"Unsupported framebuffer pixel format {pixel_format}")
+    if width == 0 or height == 0:
+        raise ScreencapError(f"Invalid screencap dimensions {width}x{height}")
+
+    pixel_bytes = width * height * _BYTES_PER_PIXEL[pixel_format]
+    header_bytes = len(raw) - pixel_bytes
+    if header_bytes < 12:
+        raise ScreencapError("Truncated screencap pixel data")
+    if header_bytes not in (12, 16):
+        raise ScreencapError("Unexpected screencap header length")
+    return _encode_png(width, height, pixel_format, raw[header_bytes:])
+
+
+def _encode_png(width, height, pixel_format, pixels):
+    """Encode framebuffer pixels as PNG, keeping alpha only when the source has it."""
+    count = width * height
+    if pixel_format == 1:
+        color_type, rgb = 6, pixels
+    elif pixel_format == 5:
+        color_type, rgb = 6, _bgra_to_rgba(pixels)
+    elif pixel_format == 2:
+        color_type = 2
+        rgb = bytearray(count * 3)
+        rgb[0::3] = pixels[0::4]
+        rgb[1::3] = pixels[1::4]
+        rgb[2::3] = pixels[2::4]
+    elif pixel_format == 3:
+        color_type, rgb = 2, pixels
+    else:
+        color_type, rgb = 2, _rgb565_to_rgb(pixels, count)
+
+    channels = 4 if color_type == 6 else 3
+    stride = width * channels
+    # Each scanline is prefixed with filter type 0 (None), which bytearray zero-fills.
+    scanlines = bytearray((stride + 1) * height)
+    view = memoryview(rgb)
+    for row in range(height):
+        start = row * stride
+        scanlines[row * (stride + 1) + 1 : (row + 1) * (stride + 1)] = view[start : start + stride]
+
+    ihdr = struct.pack(">IIBBBBB", width, height, 8, color_type, 0, 0, 0)
+    return b"".join(
+        (
+            PNG_SIGNATURE,
+            _png_chunk(b"IHDR", ihdr),
+            _png_chunk(b"IDAT", zlib.compress(scanlines, 1)),
+            _png_chunk(b"IEND", b""),
+        )
+    )
+
+
+def _png_chunk(tag, data):
+    crc = zlib.crc32(data, zlib.crc32(tag)) & 0xFFFFFFFF
+    return struct.pack(">I", len(data)) + tag + data + struct.pack(">I", crc)
+
+
+def _bgra_to_rgba(pixels):
+    out = bytearray(len(pixels))
+    out[0::4] = pixels[2::4]
+    out[1::4] = pixels[1::4]
+    out[2::4] = pixels[0::4]
+    out[3::4] = pixels[3::4]
+    return out
+
+
+def _rgb565_to_rgb(pixels, count):
+    """Expand little-endian RGB_565 (red in the top five bits) to 8-bit RGB."""
+    values = array("H")
+    values.frombytes(pixels)
+    if sys.byteorder == "big":
+        values.byteswap()
+    out = bytearray(count * 3)
+    out[0::3] = bytes(_EXPAND_5[v >> 11] for v in values)
+    out[1::3] = bytes(_EXPAND_6[(v >> 5) & 0x3F] for v in values)
+    out[2::3] = bytes(_EXPAND_5[v & 0x1F] for v in values)
+    return out
diff --git a/tests/test_basetv_async.py b/tests/test_basetv_async.py
index 96f60f88..c74daea7 100644
--- a/tests/test_basetv_async.py
+++ b/tests/test_basetv_async.py
@@ -59,7 +59,10 @@ async def test_adb_push(self):
     @awaiter
     async def test_adb_screencap(self):
         """Test that the ``adb_screencap`` method works correctly."""
-        with patch.object(self.btv._adb, "screencap", return_value=PNG_IMAGE, new_callable=async_patchers.AsyncMock):
+        # adb_shell returning "" means the device cannot compress; adb_screencap then falls back to _adb.screencap
+        with patch.object(self.btv, "adb_shell", return_value="", new_callable=async_patchers.AsyncMock), patch.object(
+            self.btv._adb, "screencap", return_value=PNG_IMAGE, new_callable=async_patchers.AsyncMock
+        ):
             self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
 
     @awaiter
diff --git a/tests/test_basetv_sync.py b/tests/test_basetv_sync.py
index ce6e7104..3b980d40 100644
--- a/tests/test_basetv_sync.py
+++ b/tests/test_basetv_sync.py
@@ -230,7 +230,10 @@ def test_adb_push(self):
 
     def test_adb_screencap(self):
         """Test that the ``adb_screencap`` method works correctly."""
-        with patch.object(self.btv._adb, "screencap", return_value=PNG_IMAGE):
+        # adb_shell returning "" means the device cannot compress; adb_screencap then falls back to _adb.screencap
+        with patch.object(self.btv, "adb_shell", return_value=""), patch.object(
+            self.btv._adb, "screencap", return_value=PNG_IMAGE
+        ):
             self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
 
     def test_keys(self):
diff --git a/tests/test_framebuffer.py b/tests/test_framebuffer.py
new file mode 100644
index 00000000..2e34bba1
--- /dev/null
+++ b/tests/test_framebuffer.py
@@ -0,0 +1,166 @@
+import base64
+import gzip
+import struct
+import sys
+import unittest
+import zlib
+from unittest.mock import Mock, patch
+
+
+sys.path.insert(0, "..")
+
+from androidtv import framebuffer
+from androidtv.basetv.basetv_sync import BaseTVSync
+
+
+PNG_IMAGE = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\n\x00\x00\x00\n\x08\x06\x00\x00\x00\x8d2\xcf\xbd\x00\x00\x00\x04sBIT\x08\x08\x08\x08|\x08d\x88\x00\x00\x00\tpHYs\x00\x00\x0fa\x00\x00\x0fa\x01\xa8?\xa7i\x00\x00\x00\x0eIDAT\x18\x95c`\x18\x05\x83\x13\x00\x00\x01\x9a\x00\x01\x16\xca\xd3i\x00\x00\x00\x00IEND\xaeB`\x82"
+
+
+def _header(width, height, pixel_format, dataspace=None):
+    if dataspace is None:
+        return struct.pack("<III", width, height, pixel_format)
+    return struct.pack("<IIII", width, height, pixel_format, dataspace)
+
+
+def _parse_png(png):
+    """Return (width, height, color_type, scanlines) for a PNG written by framebuffer."""
+    assert png.startswith(framebuffer.PNG_SIGNATURE)
+    offset = len(framebuffer.PNG_SIGNATURE)
+    chunks = {}
+    while offset < len(png):
+        (length,) = struct.unpack_from(">I", png, offset)
+        tag = png[offset + 4 : offset + 8]
+        data = png[offset + 8 : offset + 8 + length]
+        (crc,) = struct.unpack_from(">I", png, offset + 8 + length)
+        assert crc == zlib.crc32(tag + data) & 0xFFFFFFFF
+        chunks[tag] = data
+        offset += 12 + length
+    width, height, depth, color_type, _, _, _ = struct.unpack(">IIBBBBB", chunks[b"IHDR"])
+    assert depth == 8
+    return width, height, color_type, zlib.decompress(chunks[b"IDAT"])
+
+
+def _rows(scanlines, height, stride):
+    """Strip the per-row filter byte and return the pixel bytes of each row."""
+    row_len = stride + 1
+    rows = []
+    for row in range(height):
+        line = scanlines[row * row_len : (row + 1) * row_len]
+        assert line[0] == 0
+        rows.append(line[1:])
+    return rows
+
+
+class TestDecodeFramebuffer(unittest.TestCase):
+    def test_rgba_header_12(self):
+        raw = _header(2, 1, 1) + bytes([1, 2, 3, 4, 5, 6, 7, 8])
+        width, height, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual((width, height, color_type), (2, 1, 6))
+        self.assertEqual(_rows(scanlines, 1, 8), [bytes([1, 2, 3, 4, 5, 6, 7, 8])])
+
+    def test_rgba_header_16_with_dataspace(self):
+        raw = _header(1, 2, 1, dataspace=0) + bytes(range(8))
+        width, height, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual((width, height, color_type), (1, 2, 6))
+        self.assertEqual(_rows(scanlines, 2, 4), [bytes(range(4)), bytes(range(4, 8))])
+
+    def test_bgra_is_swapped_to_rgba(self):
+        raw = _header(1, 1, 5) + bytes([10, 20, 30, 40])
+        _, _, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual(color_type, 6)
+        self.assertEqual(_rows(scanlines, 1, 4), [bytes([30, 20, 10, 40])])
+
+    def test_rgbx_drops_padding_byte(self):
+        raw = _header(2, 1, 2) + bytes([1, 2, 3, 0, 4, 5, 6, 0])
+        _, _, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual(color_type, 2)
+        self.assertEqual(_rows(scanlines, 1, 6), [bytes([1, 2, 3, 4, 5, 6])])
+
+    def test_rgb888(self):
+        raw = _header(1, 2, 3) + bytes([1, 2, 3, 4, 5, 6])
+        _, _, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual(color_type, 2)
+        self.assertEqual(_rows(scanlines, 2, 3), [bytes([1, 2, 3]), bytes([4, 5, 6])])
+
+    def test_rgb565_expands_to_8_bit(self):
+        white = struct.pack("<H", 0xFFFF)
+        red = struct.pack("<H", 0xF800)
+        raw = _header(2, 1, 4) + white + red
+        _, _, color_type, scanlines = _parse_png(framebuffer.decode_framebuffer(raw))
+        self.assertEqual(color_type, 2)
+        self.assertEqual(_rows(scanlines, 1, 6), [bytes([255, 255, 255, 255, 0, 0])])
+
+    def test_unsupported_format_raises(self):
+        raw = _header(1, 1, 43) + bytes(4)
+        with self.assertRaises(framebuffer.UnsupportedScreencapFormatError):
+            framebuffer.decode_framebuffer(raw)
+
+    def test_truncated_header_raises(self):
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_framebuffer(bytes(11))
+
+    def test_truncated_pixels_raise(self):
+        raw = _header(2, 2, 1) + bytes(4)
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_framebuffer(raw)
+
+    def test_zero_dimensions_raise(self):
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_framebuffer(_header(0, 1, 1) + bytes(4))
+
+
+class TestDecodeCompressedScreencap(unittest.TestCase):
+    def _encode(self, raw):
+        return base64.b64encode(gzip.compress(raw)).decode()
+
+    def test_round_trip(self):
+        raw = _header(1, 1, 1) + bytes([9, 8, 7, 6])
+        png = framebuffer.decode_compressed_screencap(self._encode(raw))
+        self.assertEqual(_parse_png(png)[3], b"\x00" + bytes([9, 8, 7, 6]))
+
+    def test_corrupt_gzip_raises_screencap_error(self):
+        encoded = base64.b64encode(gzip.compress(_header(1, 1, 1) + bytes(4))[:-6]).decode()
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_compressed_screencap(encoded)
+
+    def test_invalid_base64_raises_screencap_error(self):
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_compressed_screencap("not base64!")
+
+    def test_truncated_compressed_frame_raises(self):
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_compressed_screencap(self._encode(bytes(11)))
+
+
+class TestAdbScreencap(unittest.TestCase):
+    def setUp(self):
+        self.btv = BaseTVSync.__new__(BaseTVSync)
+        self.btv._adb = Mock()
+
+    def test_returns_none_when_shell_not_text(self):
+        with patch.object(self.btv, "adb_shell", return_value=None):
+            self.assertIsNone(self.btv.adb_screencap())
+
+    def test_falls_back_when_device_has_no_compression(self):
+        self.btv._adb.screencap.return_value = PNG_IMAGE
+        with patch.object(self.btv, "adb_shell", return_value=""):
+            self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+            self.btv._adb.screencap.assert_called_once_with()
+
+    def test_falls_back_for_unsupported_format(self):
+        encoded = base64.b64encode(gzip.compress(_header(1, 1, 43) + bytes(4))).decode()
+        self.btv._adb.screencap.return_value = PNG_IMAGE
+        with patch.object(self.btv, "adb_shell", return_value=encoded):
+            self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+
+    def test_decodes_compressed_frame(self):
+        raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        encoded = base64.b64encode(gzip.compress(raw)).decode()
+        with patch.object(self.btv, "adb_shell", return_value=encoded) as shell:
+            png = self.btv.adb_screencap()
+        shell.assert_called_once_with(framebuffer.COMPRESSED_SCREENCAP_COMMAND)
+        self.assertEqual(_parse_png(png)[0:2], (1, 1))
+
+
+if __name__ == "__main__":
+    unittest.main()

From 1c9e6b74246bb1b4b1947327735296c22f546493 Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Thu, 15 Oct 2026 17:56:38 +0400
Subject: [PATCH 2/7] Keep corrupt screenshots from escaping the capture API

---
 androidtv/basetv/basetv_async.py | 15 ++++++---
 androidtv/basetv/basetv_sync.py  | 15 ++++++---
 androidtv/framebuffer.py         |  3 +-
 tests/test_framebuffer.py        | 55 +++++++++++++++++++++++++++++---
 4 files changed, 72 insertions(+), 16 deletions(-)

diff --git a/androidtv/basetv/basetv_async.py b/androidtv/basetv/basetv_async.py
index 71cd4877..aebc141a 100644
--- a/androidtv/basetv/basetv_async.py
+++ b/androidtv/basetv/basetv_async.py
@@ -9,7 +9,12 @@
 from .basetv import BaseTV
 from .. import constants
 from ..adb_manager.adb_manager_async import ADBPythonAsync, ADBServerAsync
-from ..framebuffer import COMPRESSED_SCREENCAP_COMMAND, UnsupportedScreencapFormatError, decode_compressed_screencap
+from ..framebuffer import (
+    COMPRESSED_SCREENCAP_COMMAND,
+    ScreencapError,
+    UnsupportedScreencapFormatError,
+    decode_compressed_screencap,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -162,10 +167,7 @@ async def adb_screencap(self):
         bytes, None
             The screencap as a binary .png image, or ``None`` if the device did not return anything
 
-        Raises
-        ------
-        androidtv.framebuffer.ScreencapError
-            If the compressed screencap is corrupt or truncated
+        Corrupt or truncated captures return ``None`` without closing the ADB connection.
 
         """
         encoded = await self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
@@ -177,6 +179,9 @@ async def adb_screencap(self):
             return await asyncio.get_running_loop().run_in_executor(None, decode_compressed_screencap, encoded)
         except UnsupportedScreencapFormatError:
             return await self._adb.screencap()
+        except ScreencapError as err:
+            _LOGGER.warning("Unable to decode screencap: %s", err)
+            return None
 
     async def adb_connect(
         self,
diff --git a/androidtv/basetv/basetv_sync.py b/androidtv/basetv/basetv_sync.py
index 62fd3cc8..ba671066 100644
--- a/androidtv/basetv/basetv_sync.py
+++ b/androidtv/basetv/basetv_sync.py
@@ -7,7 +7,12 @@
 
 from .basetv import BaseTV
 from .. import constants
-from ..framebuffer import COMPRESSED_SCREENCAP_COMMAND, UnsupportedScreencapFormatError, decode_compressed_screencap
+from ..framebuffer import (
+    COMPRESSED_SCREENCAP_COMMAND,
+    ScreencapError,
+    UnsupportedScreencapFormatError,
+    decode_compressed_screencap,
+)
 from ..adb_manager.adb_manager_sync import ADBPythonSync, ADBServerSync
 
 _LOGGER = logging.getLogger(__name__)
@@ -160,10 +165,7 @@ def adb_screencap(self):
         bytes, None
             The screencap as a binary .png image, or ``None`` if the device did not return anything
 
-        Raises
-        ------
-        androidtv.framebuffer.ScreencapError
-            If the compressed screencap is corrupt or truncated
+        Corrupt or truncated captures return ``None`` without closing the ADB connection.
 
         """
         encoded = self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
@@ -175,6 +177,9 @@ def adb_screencap(self):
             return decode_compressed_screencap(encoded)
         except UnsupportedScreencapFormatError:
             return self._adb.screencap()
+        except ScreencapError as err:
+            _LOGGER.warning("Unable to decode screencap: %s", err)
+            return None
 
     def adb_connect(
         self,
diff --git a/androidtv/framebuffer.py b/androidtv/framebuffer.py
index 230258fa..f0f593f5 100644
--- a/androidtv/framebuffer.py
+++ b/androidtv/framebuffer.py
@@ -14,8 +14,7 @@
 from array import array
 
 COMPRESSED_SCREENCAP_COMMAND = (
-    "if command -v gzip >/dev/null && command -v base64 >/dev/null; "
-    "then screencap | gzip -1 | base64; fi"
+    "if command -v gzip >/dev/null && command -v base64 >/dev/null; " "then screencap | gzip -1 | base64; fi"
 )
 
 PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
diff --git a/tests/test_framebuffer.py b/tests/test_framebuffer.py
index 2e34bba1..3a0c72dd 100644
--- a/tests/test_framebuffer.py
+++ b/tests/test_framebuffer.py
@@ -1,16 +1,15 @@
 import base64
 import gzip
 import struct
-import sys
 import unittest
 import zlib
 from unittest.mock import Mock, patch
 
-
-sys.path.insert(0, "..")
-
 from androidtv import framebuffer
 from androidtv.basetv.basetv_sync import BaseTVSync
+from androidtv.basetv.basetv_async import BaseTVAsync
+from .async_patchers import AsyncMock
+from .async_wrapper import awaiter
 
 
 PNG_IMAGE = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\n\x00\x00\x00\n\x08\x06\x00\x00\x00\x8d2\xcf\xbd\x00\x00\x00\x04sBIT\x08\x08\x08\x08|\x08d\x88\x00\x00\x00\tpHYs\x00\x00\x0fa\x00\x00\x0fa\x01\xa8?\xa7i\x00\x00\x00\x0eIDAT\x18\x95c`\x18\x05\x83\x13\x00\x00\x01\x9a\x00\x01\x16\xca\xd3i\x00\x00\x00\x00IEND\xaeB`\x82"
@@ -162,5 +161,53 @@ def test_decodes_compressed_frame(self):
         self.assertEqual(_parse_png(png)[0:2], (1, 1))
 
 
+class TestScreencapFailure(unittest.TestCase):
+    def test_corrupt_frame_keeps_sync_transport(self):
+        btv = BaseTVSync.__new__(BaseTVSync)
+        btv._adb = Mock()
+        encoded = base64.b64encode(bytes.fromhex("1f8b0800000000000003070000000000000000")).decode()
+        with patch.object(btv, "adb_shell", return_value=encoded):
+            self.assertIsNone(btv.adb_screencap())
+        btv._adb.close.assert_not_called()
+        btv._adb.screencap.assert_not_called()
+
+
+class TestAdbScreencapAsync(unittest.TestCase):
+    def setUp(self):
+        self.btv = BaseTVAsync.__new__(BaseTVAsync)
+        self.btv._adb = Mock()
+        self.btv._adb.screencap = AsyncMock(return_value=PNG_IMAGE)
+
+    @awaiter
+    async def test_decodes_compressed_frame(self):
+        raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        encoded = base64.b64encode(gzip.compress(raw)).decode()
+        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
+            png = await self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.btv._adb.screencap.assert_not_called()
+
+    @awaiter
+    async def test_corrupt_frame_keeps_transport(self):
+        encoded = base64.b64encode(bytes.fromhex("1f8b0800000000000003070000000000000000")).decode()
+        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
+            self.assertIsNone(await self.btv.adb_screencap())
+        self.btv._adb.close.assert_not_called()
+        self.btv._adb.screencap.assert_not_called()
+
+    @awaiter
+    async def test_unsupported_format_keeps_native_capture(self):
+        encoded = base64.b64encode(gzip.compress(_header(1, 1, 43) + bytes(4))).decode()
+        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
+            self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.btv._adb.screencap.assert_called_once_with()
+
+    @awaiter
+    async def test_transport_failure_remains_an_error(self):
+        with patch.object(self.btv, "adb_shell", side_effect=ConnectionResetError, new_callable=AsyncMock):
+            with self.assertRaises(ConnectionResetError):
+                await self.btv.adb_screencap()
+
+
 if __name__ == "__main__":
     unittest.main()

From 1ab8db8f907d0c235563f2b00b2e58cde8639dfe Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Thu, 15 Oct 2026 18:06:08 +0400
Subject: [PATCH 3/7] Run legacy Python matrix on Ubuntu 22.04

---
 .github/workflows/python-package.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 5989973c..ae4d71d5 100644
--- a/.github/workflows/python-package.yml
+++ b/.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']

From 5e97622120ad8e9d90ef2d7b694428df08783f44 Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Thu, 15 Oct 2026 20:12:20 +0400
Subject: [PATCH 4/7] Pick the faster screencap path per device and fall back
 on decode failure

---
 androidtv/adb_manager/adb_manager_async.py |  27 +++
 androidtv/adb_manager/adb_manager_sync.py  |  27 +++
 androidtv/basetv/basetv_async.py           | 108 +++++++++--
 androidtv/basetv/basetv_sync.py            | 107 +++++++++--
 androidtv/framebuffer.py                   |  38 ++--
 tests/test_framebuffer.py                  | 214 +++++++++++++++------
 6 files changed, 416 insertions(+), 105 deletions(-)

diff --git a/androidtv/adb_manager/adb_manager_async.py b/androidtv/adb_manager/adb_manager_async.py
index e5325291..db84c596 100644
--- a/androidtv/adb_manager/adb_manager_async.py
+++ b/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.
diff --git a/androidtv/adb_manager/adb_manager_sync.py b/androidtv/adb_manager/adb_manager_sync.py
index e93283d3..b0c1d054 100644
--- a/androidtv/adb_manager/adb_manager_sync.py
+++ b/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.
diff --git a/androidtv/basetv/basetv_async.py b/androidtv/basetv/basetv_async.py
index aebc141a..100d741b 100644
--- a/androidtv/basetv/basetv_async.py
+++ b/androidtv/basetv/basetv_async.py
@@ -5,15 +5,17 @@
 
 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 (
-    COMPRESSED_SCREENCAP_COMMAND,
+    GZIP_SCREENCAP_COMMAND,
+    RAW_SCREENCAP_COMMAND,
     ScreencapError,
-    UnsupportedScreencapFormatError,
-    decode_compressed_screencap,
+    decode_framebuffer,
+    decode_gzip_screencap,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -92,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
 
     # ======================================================================= #
     #                                                                         #
@@ -154,34 +157,106 @@ async def adb_screencap(self):
 
         Returns the screen as a binary .png image, or ``None`` if the device did not return anything.
 
-        Note: the raw framebuffer is transferred gzip-compressed and the PNG is encoded on the host with the
-        standard library (the host-side decode runs in an executor), which is much faster than having the
-        device encode the PNG itself. If the device lacks ``gzip`` or ``base64``, or the framebuffer pixel
-        format is not supported, the device-side ``screencap -p`` path is used instead
+        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).
+        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, None
             The screencap as a binary .png image, or ``None`` if the device did not return anything
 
-        Corrupt or truncated captures return ``None`` without closing the ADB connection.
-
         """
-        encoded = await self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
-        if not isinstance(encoded, str):
+        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
-        if not encoded.strip():
+        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_compressed_screencap, encoded)
-        except UnsupportedScreencapFormatError:
+            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.warning("Unable to decode screencap: %s", 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,
@@ -206,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)
diff --git a/androidtv/basetv/basetv_sync.py b/androidtv/basetv/basetv_sync.py
index ba671066..9d29cbc2 100644
--- a/androidtv/basetv/basetv_sync.py
+++ b/androidtv/basetv/basetv_sync.py
@@ -4,14 +4,16 @@
 """
 
 import logging
+from time import monotonic
 
 from .basetv import BaseTV
 from .. import constants
 from ..framebuffer import (
-    COMPRESSED_SCREENCAP_COMMAND,
+    GZIP_SCREENCAP_COMMAND,
+    RAW_SCREENCAP_COMMAND,
     ScreencapError,
-    UnsupportedScreencapFormatError,
-    decode_compressed_screencap,
+    decode_framebuffer,
+    decode_gzip_screencap,
 )
 from ..adb_manager.adb_manager_sync import ADBPythonSync, ADBServerSync
 
@@ -91,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
 
     # ======================================================================= #
     #                                                                         #
@@ -153,33 +156,104 @@ def adb_screencap(self):
 
         Returns the screen as a binary .png image, or ``None`` if the device did not return anything.
 
-        Note: the raw framebuffer is transferred gzip-compressed and the PNG is encoded on the host with the
-        standard library, which is much faster than having the device encode the PNG itself. If the device
-        lacks ``gzip`` or ``base64``, or the framebuffer pixel format is not supported, the device-side
-        ``screencap -p`` path is used instead (:py:meth:`androidtv.adb_manager.adb_manager_sync.ADBPythonSync.screencap`
+        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).
+        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, None
             The screencap as a binary .png image, or ``None`` if the device did not return anything
 
-        Corrupt or truncated captures return ``None`` without closing the ADB connection.
-
         """
-        encoded = self.adb_shell(COMPRESSED_SCREENCAP_COMMAND)
-        if not isinstance(encoded, str):
+        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
-        if not encoded.strip():
+        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_compressed_screencap(encoded)
-        except UnsupportedScreencapFormatError:
+            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.warning("Unable to decode screencap: %s", 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,
@@ -204,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)
diff --git a/androidtv/framebuffer.py b/androidtv/framebuffer.py
index f0f593f5..868a4355 100644
--- a/androidtv/framebuffer.py
+++ b/androidtv/framebuffer.py
@@ -1,21 +1,25 @@
-"""Capture Android framebuffers compressed over ADB and encode them as PNG.
-
-``screencap -p`` makes the device CPU encode the PNG, which is slow on many
-TVs. Piping the raw framebuffer through ``gzip`` and ``base64`` is much faster;
-the host then parses the Android framebuffer header and encodes the PNG itself
-using only the standard library.
+"""Capture Android framebuffers over ADB and encode them as PNG.
+
+``screencap -p`` makes the device CPU encode the PNG. Two binary alternatives
+avoid that: plain ``screencap`` (raw pixels, decoded on the host) and
+``screencap | gzip -1`` (raw pixels gzipped on the device). Neither path is
+always faster: on one Hi3751V350 projector the medians were 7.4 s for
+``screencap -p``, about 2.5 s for raw ``screencap`` read binary (0.7 s grab
+plus transfer of 8.3 MB), and about 4.1 s for the gzip path. Callers time each
+path back to back on first use and keep the fastest valid one for that
+device. The host parses the Android framebuffer header and encodes the PNG
+itself using only the standard library.
 """
 
-import base64
 import gzip
 import struct
 import sys
 import zlib
 from array import array
 
-COMPRESSED_SCREENCAP_COMMAND = (
-    "if command -v gzip >/dev/null && command -v base64 >/dev/null; " "then screencap | gzip -1 | base64; fi"
-)
+RAW_SCREENCAP_COMMAND = "screencap"
+
+GZIP_SCREENCAP_COMMAND = "screencap | gzip -1"
 
 PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
 
@@ -34,13 +38,13 @@ class UnsupportedScreencapFormatError(ScreencapError):
     """The framebuffer uses a pixel format this decoder does not handle."""
 
 
-def decode_compressed_screencap(encoded):
-    """Decode the base64-encoded gzip output of :data:`COMPRESSED_SCREENCAP_COMMAND`.
+def decode_gzip_screencap(data):
+    """Decode the raw gzip output of :data:`GZIP_SCREENCAP_COMMAND`.
 
     Parameters
     ----------
-    encoded : str
-        The text printed by the device shell
+    data : bytes
+        The raw bytes printed by the device shell
 
     Returns
     -------
@@ -50,15 +54,15 @@ def decode_compressed_screencap(encoded):
     Raises
     ------
     ScreencapError
-        If the data is not valid base64/gzip or the framebuffer is truncated
+        If the data is not valid gzip or the framebuffer is truncated
     UnsupportedScreencapFormatError
         If the framebuffer pixel format is not supported
 
     """
     try:
-        raw = gzip.decompress(base64.b64decode(encoded))
+        raw = gzip.decompress(data)
     except (ValueError, OSError, EOFError, zlib.error) as err:
-        raise ScreencapError(f"Corrupt compressed screencap: {err}") from err
+        raise ScreencapError(f"Corrupt gzip screencap: {err}") from err
     return decode_framebuffer(raw)
 
 
diff --git a/tests/test_framebuffer.py b/tests/test_framebuffer.py
index 3a0c72dd..c6b3c006 100644
--- a/tests/test_framebuffer.py
+++ b/tests/test_framebuffer.py
@@ -108,68 +108,111 @@ def test_zero_dimensions_raise(self):
             framebuffer.decode_framebuffer(_header(0, 1, 1) + bytes(4))
 
 
-class TestDecodeCompressedScreencap(unittest.TestCase):
-    def _encode(self, raw):
-        return base64.b64encode(gzip.compress(raw)).decode()
-
+class TestDecodeGzipScreencap(unittest.TestCase):
     def test_round_trip(self):
         raw = _header(1, 1, 1) + bytes([9, 8, 7, 6])
-        png = framebuffer.decode_compressed_screencap(self._encode(raw))
+        png = framebuffer.decode_gzip_screencap(gzip.compress(raw))
         self.assertEqual(_parse_png(png)[3], b"\x00" + bytes([9, 8, 7, 6]))
 
     def test_corrupt_gzip_raises_screencap_error(self):
-        encoded = base64.b64encode(gzip.compress(_header(1, 1, 1) + bytes(4))[:-6]).decode()
         with self.assertRaises(framebuffer.ScreencapError):
-            framebuffer.decode_compressed_screencap(encoded)
+            framebuffer.decode_gzip_screencap(gzip.compress(_header(1, 1, 1) + bytes(4))[:-6])
 
-    def test_invalid_base64_raises_screencap_error(self):
+    def test_non_gzip_raises_screencap_error(self):
         with self.assertRaises(framebuffer.ScreencapError):
-            framebuffer.decode_compressed_screencap("not base64!")
+            framebuffer.decode_gzip_screencap(b"not gzip at all")
 
     def test_truncated_compressed_frame_raises(self):
         with self.assertRaises(framebuffer.ScreencapError):
-            framebuffer.decode_compressed_screencap(self._encode(bytes(11)))
+            framebuffer.decode_gzip_screencap(gzip.compress(bytes(11)))
+
+    def test_corrupted_gzip_bytes_fail_crc(self):
+        raw = _header(1, 1, 1) + bytes([9, 8, 7, 6])
+        mangled = bytearray(gzip.compress(raw))
+        mangled[10] ^= 0xFF
+        with self.assertRaises(framebuffer.ScreencapError):
+            framebuffer.decode_gzip_screencap(bytes(mangled))
 
 
 class TestAdbScreencap(unittest.TestCase):
     def setUp(self):
         self.btv = BaseTVSync.__new__(BaseTVSync)
         self.btv._adb = Mock()
+        self.btv._adb.screencap.return_value = PNG_IMAGE
+        self.raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        self.gzipped = gzip.compress(self.raw)
+        self.btv._adb.shell_binary.side_effect = lambda cmd: (
+            self.raw if cmd == framebuffer.RAW_SCREENCAP_COMMAND else self.gzipped
+        )
+
+    def test_calibration_picks_fastest_raw(self):
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
+            png = self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[0:2], (1, 1))
+        self.assertEqual(self.btv._screencap_method, "raw")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        png = self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[0:2], (1, 1))
+        self.btv._adb.shell_binary.assert_called_once_with(framebuffer.RAW_SCREENCAP_COMMAND)
+        self.btv._adb.screencap.assert_not_called()
 
-    def test_returns_none_when_shell_not_text(self):
-        with patch.object(self.btv, "adb_shell", return_value=None):
-            self.assertIsNone(self.btv.adb_screencap())
+    def test_calibration_picks_fastest_gzip(self):
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 5, 5, 6, 6, 10]):
+            png = self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[0:2], (1, 1))
+        self.assertEqual(self.btv._screencap_method, "gzip")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        png = self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[0:2], (1, 1))
+        self.btv._adb.shell_binary.assert_called_once_with(framebuffer.GZIP_SCREENCAP_COMMAND)
+        self.btv._adb.screencap.assert_not_called()
 
-    def test_falls_back_when_device_has_no_compression(self):
-        self.btv._adb.screencap.return_value = PNG_IMAGE
-        with patch.object(self.btv, "adb_shell", return_value=""):
+    def test_calibration_picks_fastest_native(self):
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 5, 5, 9, 9, 10]):
             self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
-            self.btv._adb.screencap.assert_called_once_with()
+        self.assertEqual(self.btv._screencap_method, "native")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.btv._adb.shell_binary.assert_not_called()
+        self.btv._adb.screencap.assert_called_once_with()
 
-    def test_falls_back_for_unsupported_format(self):
-        encoded = base64.b64encode(gzip.compress(_header(1, 1, 43) + bytes(4))).decode()
-        self.btv._adb.screencap.return_value = PNG_IMAGE
-        with patch.object(self.btv, "adb_shell", return_value=encoded):
+    def test_invalid_candidates_are_discarded(self):
+        mangled_raw = self.raw.replace(b"\n", b"\r\n")
+        if len(mangled_raw) == len(self.raw):
+            mangled_raw = self.raw + b"\r"
+        self.btv._adb.shell_binary.side_effect = lambda cmd: (
+            mangled_raw if cmd == framebuffer.RAW_SCREENCAP_COMMAND else b"not gzip"
+        )
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 1, 1, 2, 2, 3]):
             self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
-
-    def test_decodes_compressed_frame(self):
-        raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
-        encoded = base64.b64encode(gzip.compress(raw)).decode()
-        with patch.object(self.btv, "adb_shell", return_value=encoded) as shell:
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_pinned_failure_falls_back_to_native(self):
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
+            self.btv.adb_screencap()
+        self.assertEqual(self.btv._screencap_method, "raw")
+        self.btv._adb.shell_binary.side_effect = lambda cmd: b"garbage"
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_server_path_without_binary_shell_uses_native(self):
+        del self.btv._adb.shell_binary
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_reconnect_recalibrates(self):
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 5, 5, 9, 9, 10]):
+            self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+        self.btv.adb_connect()
+        self.assertIsNone(self.btv._screencap_method)
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
             png = self.btv.adb_screencap()
-        shell.assert_called_once_with(framebuffer.COMPRESSED_SCREENCAP_COMMAND)
         self.assertEqual(_parse_png(png)[0:2], (1, 1))
-
-
-class TestScreencapFailure(unittest.TestCase):
-    def test_corrupt_frame_keeps_sync_transport(self):
-        btv = BaseTVSync.__new__(BaseTVSync)
-        btv._adb = Mock()
-        encoded = base64.b64encode(bytes.fromhex("1f8b0800000000000003070000000000000000")).decode()
-        with patch.object(btv, "adb_shell", return_value=encoded):
-            self.assertIsNone(btv.adb_screencap())
-        btv._adb.close.assert_not_called()
-        btv._adb.screencap.assert_not_called()
+        self.assertEqual(self.btv._screencap_method, "raw")
 
 
 class TestAdbScreencapAsync(unittest.TestCase):
@@ -177,37 +220,96 @@ def setUp(self):
         self.btv = BaseTVAsync.__new__(BaseTVAsync)
         self.btv._adb = Mock()
         self.btv._adb.screencap = AsyncMock(return_value=PNG_IMAGE)
+        self.btv._adb.connect = AsyncMock(return_value=True)
+        self.raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        self.gzipped = gzip.compress(self.raw)
+        self.btv._adb.shell_binary = AsyncMock(side_effect=self._shell_binary)
+
+    async def _shell_binary(self, cmd):
+        if cmd == framebuffer.RAW_SCREENCAP_COMMAND:
+            return self.raw
+        return self.gzipped
 
     @awaiter
-    async def test_decodes_compressed_frame(self):
-        raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
-        encoded = base64.b64encode(gzip.compress(raw)).decode()
-        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
+    async def test_calibration_picks_fastest_raw(self):
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
             png = await self.btv.adb_screencap()
         self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.assertEqual(self.btv._screencap_method, "raw")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        png = await self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.btv._adb.shell_binary.assert_called_once_with(framebuffer.RAW_SCREENCAP_COMMAND)
         self.btv._adb.screencap.assert_not_called()
 
     @awaiter
-    async def test_corrupt_frame_keeps_transport(self):
-        encoded = base64.b64encode(bytes.fromhex("1f8b0800000000000003070000000000000000")).decode()
-        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
-            self.assertIsNone(await self.btv.adb_screencap())
-        self.btv._adb.close.assert_not_called()
+    async def test_calibration_picks_fastest_gzip(self):
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 5, 5, 6, 6, 10]):
+            png = await self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.assertEqual(self.btv._screencap_method, "gzip")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        png = await self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.btv._adb.shell_binary.assert_called_once_with(framebuffer.GZIP_SCREENCAP_COMMAND)
         self.btv._adb.screencap.assert_not_called()
 
     @awaiter
-    async def test_unsupported_format_keeps_native_capture(self):
-        encoded = base64.b64encode(gzip.compress(_header(1, 1, 43) + bytes(4))).decode()
-        with patch.object(self.btv, "adb_shell", return_value=encoded, new_callable=AsyncMock):
+    async def test_calibration_picks_fastest_native(self):
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 5, 5, 9, 9, 10]):
             self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+        self.btv._adb.shell_binary.reset_mock()
+        self.btv._adb.screencap.reset_mock()
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.btv._adb.shell_binary.assert_not_called()
         self.btv._adb.screencap.assert_called_once_with()
 
     @awaiter
-    async def test_transport_failure_remains_an_error(self):
-        with patch.object(self.btv, "adb_shell", side_effect=ConnectionResetError, new_callable=AsyncMock):
-            with self.assertRaises(ConnectionResetError):
-                await self.btv.adb_screencap()
+    async def test_invalid_candidates_are_discarded(self):
+        mangled_raw = self.raw.replace(b"\n", b"\r\n")
+        if len(mangled_raw) == len(self.raw):
+            mangled_raw = self.raw + b"\r"
+
+        async def bad_shell(cmd):
+            return mangled_raw if cmd == framebuffer.RAW_SCREENCAP_COMMAND else b"not gzip"
 
+        self.btv._adb.shell_binary = AsyncMock(side_effect=bad_shell)
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 1, 1, 2, 2, 3]):
+            self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_pinned_failure_falls_back_to_native(self):
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
+            await self.btv.adb_screencap()
+        self.assertEqual(self.btv._screencap_method, "raw")
+        self.btv._adb.shell_binary = AsyncMock(return_value=b"garbage")
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
 
-if __name__ == "__main__":
-    unittest.main()
+    @awaiter
+    async def test_server_path_without_binary_shell_uses_native(self):
+        del self.btv._adb.shell_binary
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_reconnect_recalibrates(self):
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 5, 5, 9, 9, 10]):
+            self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+        await self.btv.adb_connect()
+        self.assertIsNone(self.btv._screencap_method)
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 1, 1, 5, 5, 6]):
+            png = await self.btv.adb_screencap()
+        self.assertEqual(_parse_png(png)[3], b"\x00\x01\x02\x03\x04")
+        self.assertEqual(self.btv._screencap_method, "raw")
+
+    @awaiter
+    async def test_transport_failure_remains_an_error(self):
+        self.btv._adb.shell_binary = AsyncMock(side_effect=ConnectionResetError)
+        with self.assertRaises(ConnectionResetError):
+            await self.btv.adb_screencap()

From b17622ed725162094607862053e4cbccd9d92d75 Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Fri, 16 Oct 2026 18:19:42 +0400
Subject: [PATCH 5/7] Test screencap fallbacks and shell_binary to restore
 coverage

The new framebuffer screencap calibration and ADBPythonSync.shell_binary
paths were uncovered, dropping total coverage on the PR branch below the
master baseline. Add tests for pinned-method fallbacks, calibration with
no valid candidate, big-endian RGB565 expansion, and shell_binary.
---
 tests/test_adb_manager_sync.py |  30 ++++++++
 tests/test_framebuffer.py      | 133 +++++++++++++++++++++++++++++++++
 2 files changed, 163 insertions(+)

diff --git a/tests/test_adb_manager_sync.py b/tests/test_adb_manager_sync.py
index 482df913..89da9b26 100644
--- a/tests/test_adb_manager_sync.py
+++ b/tests/test_adb_manager_sync.py
@@ -332,3 +332,33 @@ def test_close(self):
 
             self.adb.close()
             self.assertFalse(self.adb.available)
+
+
+class TestADBPythonSyncShellBinary(unittest.TestCase):
+    """Test the `ADBPythonSync.shell_binary` method."""
+
+    PATCH_KEY = "python"
+
+    def setUp(self):
+        """Create an `ADBPythonSync` instance."""
+        with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[self.PATCH_KEY]:
+            self.adb = ADBPythonSync("HOST", 5555)
+
+    def test_adb_shell_binary_returns_raw_bytes(self):
+        """Test that ``shell_binary`` returns the undecoded device output when connected."""
+        with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell(b"\x00\x01\x02")[self.PATCH_KEY]:
+            self.assertTrue(self.adb.connect())
+            self.assertEqual(self.adb.shell_binary("screencap"), b"\x00\x01\x02")
+
+    def test_adb_shell_binary_unavailable_returns_none(self):
+        """Test that ``shell_binary`` returns ``None`` when the adb-shell connection is not established."""
+        self.assertFalse(self.adb.available)
+        self.assertIsNone(self.adb.shell_binary("screencap"))
+
+    def test_adb_shell_binary_lock_not_acquired(self):
+        """Test that ``shell_binary`` raises when the ADB lock is held by another caller."""
+        with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell(b"\x00")[self.PATCH_KEY]:
+            self.assertTrue(self.adb.connect())
+            with patch.object(self.adb, "_adb_lock", LockedLock()):
+                with self.assertRaises(LockNotAcquiredException):
+                    self.adb.shell_binary("screencap")
diff --git a/tests/test_framebuffer.py b/tests/test_framebuffer.py
index c6b3c006..cf8c3ef0 100644
--- a/tests/test_framebuffer.py
+++ b/tests/test_framebuffer.py
@@ -215,6 +215,139 @@ def test_reconnect_recalibrates(self):
         self.assertEqual(self.btv._screencap_method, "raw")
 
 
+class TestAdbScreencapFallbacks(unittest.TestCase):
+    def setUp(self):
+        self.btv = BaseTVSync.__new__(BaseTVSync)
+        self.btv._adb = Mock()
+        self.btv._adb.screencap.return_value = PNG_IMAGE
+        self.raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        self.gzipped = gzip.compress(self.raw)
+
+    def test_pinned_raw_empty_output_uses_native(self):
+        self.btv._screencap_method = "raw"
+        self.btv._adb.shell_binary.return_value = b""
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_pinned_raw_none_output_uses_native(self):
+        self.btv._screencap_method = "raw"
+        self.btv._adb.shell_binary.return_value = None
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_pinned_gzip_empty_output_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        self.btv._adb.shell_binary.return_value = b""
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_pinned_gzip_decode_failure_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        self.btv._adb.shell_binary.return_value = b"not gzip"
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_pinned_gzip_without_binary_shell_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        del self.btv._adb.shell_binary
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_calibration_without_binary_shell_uses_native(self):
+        del self.btv._adb.shell_binary
+        self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_calibration_without_any_valid_candidate_returns_none(self):
+        self.btv._adb.shell_binary.return_value = b""
+        self.btv._adb.screencap.return_value = None
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=range(10)):
+            self.assertIsNone(self.btv.adb_screencap())
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    def test_calibration_skips_undecodable_candidate(self):
+        self.btv._adb.shell_binary.side_effect = lambda cmd: (
+            b"garbage" if cmd == framebuffer.RAW_SCREENCAP_COMMAND else self.gzipped
+        )
+        with patch("androidtv.basetv.basetv_sync.monotonic", side_effect=[0, 1, 5, 5, 6]):
+            self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+
+class TestAdbScreencapFallbacksAsync(unittest.TestCase):
+    def setUp(self):
+        self.btv = BaseTVAsync.__new__(BaseTVAsync)
+        self.btv._adb = Mock()
+        self.btv._adb.screencap = AsyncMock(return_value=PNG_IMAGE)
+        self.raw = _header(1, 1, 1) + bytes([1, 2, 3, 4])
+        self.gzipped = gzip.compress(self.raw)
+
+    @awaiter
+    async def test_pinned_raw_empty_output_uses_native(self):
+        self.btv._screencap_method = "raw"
+        self.btv._adb.shell_binary = AsyncMock(return_value=b"")
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_pinned_raw_none_output_uses_native(self):
+        self.btv._screencap_method = "raw"
+        self.btv._adb.shell_binary = AsyncMock(return_value=None)
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_pinned_gzip_empty_output_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        self.btv._adb.shell_binary = AsyncMock(return_value=b"")
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_pinned_gzip_decode_failure_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        self.btv._adb.shell_binary = AsyncMock(return_value=b"not gzip")
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_pinned_gzip_without_binary_shell_uses_native(self):
+        self.btv._screencap_method = "gzip"
+        del self.btv._adb.shell_binary
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_calibration_without_binary_shell_uses_native(self):
+        del self.btv._adb.shell_binary
+        self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_calibration_without_any_valid_candidate_returns_none(self):
+        self.btv._adb.shell_binary = AsyncMock(return_value=b"")
+        self.btv._adb.screencap = AsyncMock(return_value=None)
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=range(10)):
+            self.assertIsNone(await self.btv.adb_screencap())
+        self.assertEqual(self.btv._screencap_method, "native")
+
+    @awaiter
+    async def test_calibration_skips_undecodable_candidate(self):
+        async def mixed_shell(cmd):
+            return b"garbage" if cmd == framebuffer.RAW_SCREENCAP_COMMAND else self.gzipped
+
+        self.btv._adb.shell_binary = AsyncMock(side_effect=mixed_shell)
+        with patch("androidtv.basetv.basetv_async.monotonic", side_effect=[0, 1, 5, 5, 6]):
+            self.assertEqual(await self.btv.adb_screencap(), PNG_IMAGE)
+        self.assertEqual(self.btv._screencap_method, "native")
+
+
+class TestRgb565ByteOrder(unittest.TestCase):
+    def test_big_endian_host_byteswaps_pixels(self):
+        with patch.object(framebuffer.sys, "byteorder", "big"):
+            self.assertEqual(framebuffer._rgb565_to_rgb(b"\xf8\x00", 1), bytes([255, 0, 0]))
+
+
 class TestAdbScreencapAsync(unittest.TestCase):
     def setUp(self):
         self.btv = BaseTVAsync.__new__(BaseTVAsync)

From d14034ccec4e83b47377162d868dffab6ae0f9ee Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Sun, 18 Oct 2026 03:07:32 +0400
Subject: [PATCH 6/7] Pin black to 23.3.0 so the unpinned formatter stops
 drifting under the lint matrix

The requirement was a bare `black`, so each CI Python resolved a different
release (23.3.0 on 3.7 ... 26.10.1 on 3.10-3.12) and the newest normalizes
androidtv/exceptions.py's module docstring value, which the repo's own
formatting cannot absorb without a literal change. 23.3.0 is the newest
release that still supports the maintained Python 3.7 leg and it leaves
package, tests, and setup.py clean as authored.
---
 venv_requirements.txt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/venv_requirements.txt b/venv_requirements.txt
index 8f953308..a2c26b9d 100644
--- a/venv_requirements.txt
+++ b/venv_requirements.txt
@@ -1,5 +1,5 @@
 # Standard requirements
-black
+black==23.3.0
 coverage
 coveralls
 flake8

From 66c3972840f379382649b4e845046f7454a2b166 Mon Sep 17 00:00:00 2001
From: Darafei Praliaskouski <me@komzpa.net>
Date: Sun, 18 Oct 2026 04:04:39 +0400
Subject: [PATCH 7/7] Fix Python 3.7 AsyncMock await semantics in
 tests.async_patchers

On Python 3.7 unittest.mock.AsyncMock is unavailable, so the fallback
AsyncMock subclass returned the coroutine from __call__ without
awaiting it. Await the result when side_effect is a coroutine function,
matching the 3.8+ contract so async test patches resolve correctly.

Document the 3.7 semantics in venv_requirements.txt. No runtime code
changed; dependencies unchanged.
---
 tests/async_patchers.py | 8 +++++++-
 venv_requirements.txt   | 1 +
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/tests/async_patchers.py b/tests/async_patchers.py
index 3ac715f2..18551792 100644
--- a/tests/async_patchers.py
+++ b/tests/async_patchers.py
@@ -1,5 +1,6 @@
 """Define patches used for androidtv tests."""
 
+from inspect import iscoroutinefunction
 from unittest.mock import patch
 
 try:
@@ -8,8 +9,13 @@
     from unittest.mock import MagicMock
 
     class AsyncMock(MagicMock):
+        """Await async side effects like unittest.mock.AsyncMock on Python 3.8+."""
+
         async def __call__(self, *args, **kwargs):
-            return super(AsyncMock, self).__call__(*args, **kwargs)
+            result = super(AsyncMock, self).__call__(*args, **kwargs)
+            if iscoroutinefunction(self.side_effect):
+                return await result
+            return result
 
 
 KEY_PYTHON = "python"
diff --git a/venv_requirements.txt b/venv_requirements.txt
index a2c26b9d..4f827e1f 100644
--- a/venv_requirements.txt
+++ b/venv_requirements.txt
@@ -1,4 +1,5 @@
 # Standard requirements
+# Async tests support Python 3.7 via tests.async_patchers.AsyncMock, which awaits async side effects.
 black==23.3.0
 coverage
 coveralls