Merge 08ef32e into fb5bfb3

androidtv/adb_manager.py

@@ -213,6 +213,23 @@ def push(self, local_path, device_path):
             self._adb.push(local_path, device_path)
             return
 
+    def screencap(self):
+        """Take a screenshot using the Python ADB implementation.
+
+        Returns
+        -------
+        bytes
+            The screencap as a binary .png image
+
+        """
+        if not self.available:
+            _LOGGER.debug("ADB screencap not taken from %s:%d because adb-shell connection is not established", self.host, self.port)
+            return None
+
+        with _acquire(self._adb_lock):
+            _LOGGER.debug("Taking screencap from %s:%d via adb-shell", self.host, self.port)
+            return self._adb.shell("screencap -p", decode=False)
+
     def shell(self, cmd):
         """Send an ADB command using the Python ADB implementation.
 
@@ -412,6 +429,23 @@ def push(self, local_path, device_path):
             self._adb_device.push(local_path, device_path)
             return
 
+    def screencap(self):
+        """Take a screenshot using an ADB server.
+
+        Returns
+        -------
+        bytes
+            The screencap as a binary .png image
+
+        """
+        if not self.available:
+            _LOGGER.debug("ADB screencap not taken from %s:%d via ADB server %s:%d because pure-python-adb connection is not established", self.host, self.port, self.adb_server_ip, self.adb_server_port)
+            return None
+
+        with _acquire(self._adb_lock):
+            _LOGGER.debug("Taking screencap from %s:%d via ADB server %s:%d", self.host, self.port, self.adb_server_ip, self.adb_server_port)
+            return self._adb_device.screencap()
+
     def shell(self, cmd):
         """Send an ADB command using an ADB server.
 

androidtv/basetv.py

@@ -155,6 +155,20 @@ def adb_push(self, local_path, device_path):
         """
         return self._adb.push(local_path, device_path)
 
+    def adb_screencap(self):
+        """Take a screencap.
+
+        This calls :py:meth:`androidtv.adb_manager.ADBPython.screencap` or :py:meth:`androidtv.adb_manager.ADBServer.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
+
+        """
+        return self._adb.screencap()
+
     def adb_connect(self, always_log_errors=True, auth_timeout_s=constants.DEFAULT_AUTH_TIMEOUT_S):
         """Connect to an Android TV / Fire TV device.
 

setup.py

@@ -14,7 +14,7 @@
     author='Jeff Irion',
     author_email='jefflirion@users.noreply.github.com',
     packages=['androidtv'],
-    install_requires=['adb-shell>=0.1.0', 'pure-python-adb>=0.2.2.dev0'],
+    install_requires=['adb-shell>=0.1.3', 'pure-python-adb>=0.2.2.dev0'],
     classifiers=[
         'License :: OSI Approved :: MIT License',
         'Operating System :: OS Independent',

tests/patchers.py

@@ -29,7 +29,7 @@ def push(self, *args, **kwargs):
     def pull(self, *args, **kwargs):
         """Pull a file from the device."""
 
-    def shell(self, cmd):
+    def shell(self, cmd, *args, **kwargs):
         """Send an ADB shell command."""
         return None
 
@@ -89,6 +89,10 @@ def shell(self, cmd):
         """Send an ADB shell command."""
         raise NotImplementedError
 
+    def screencap(self):
+        """Take a screencap."""
+        raise NotImplementedError
+
 
 def patch_connect(success):
     """Mock the `adb_shell.adb_device.AdbDeviceTcp` and `ppadb.client.Client` classes."""
@@ -109,12 +113,12 @@ def connect_fail_python(self, *args, **kwargs):
 def patch_shell(response=None, error=False):
     """Mock the `AdbDeviceTcpFake.shell` and `DeviceFake.shell` methods."""
 
-    def shell_success(self, cmd):
+    def shell_success(self, cmd, *args, **kwargs):
         """Mock the `AdbDeviceTcpFake.shell` and `DeviceFake.shell` methods when they are successful."""
         self.shell_cmd = cmd
         return response
 
-    def shell_fail_python(self, cmd):
+    def shell_fail_python(self, cmd, *args, **kwargs):
         """Mock the `AdbDeviceTcpFake.shell` method when it fails."""
         self.shell_cmd = cmd
         raise AttributeError

tests/test_adb_manager.py

@@ -30,6 +30,9 @@ def read(self):
         raise FileNotFoundError
 
 
+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'
+
+
 @contextmanager
 def open_priv(infile):
     """A patch that will read the private key but not the public key."""
@@ -242,6 +245,42 @@ def test_adb_pull_success(self):
                 self.adb.pull("TEST_LOCAL_PATH", "TEST_DEVICE_PATH")
                 self.assertEqual(patch_pull.call_count, 1)
 
+    def test_adb_screencap_fail_unavailable(self):
+        """Test when an ADB screencap command fails because the connection is unavailable.
+
+        """        
+        self.assertFalse(self.adb.available)
+        self.assertIsNone(self.adb.screencap())
+
+    def test_adb_screencap_lock_not_acquired(self):
+        """Test when an ADB screencap command fails because the ADB lock could not be acquired.
+
+        """
+        with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell("TEST")[self.PATCH_KEY]:
+            self.assertTrue(self.adb.connect())
+            self.assertEqual(self.adb.shell("TEST"), "TEST")
+
+        with patchers.patch_shell(PNG_IMAGE)[self.PATCH_KEY], patch.object(self.adb, '_adb_lock', LockedLock()):
+            with patch('{}.LockedLock.release'.format(__name__)) as release:
+                with self.assertRaises(LockNotAcquiredException):
+                    self.adb.screencap()
+
+                release.assert_not_called()
+
+    def test_adb_screencap_success(self):
+        """Test the `screencap` method.
+
+        """
+        with patchers.patch_connect(True)[self.PATCH_KEY], patchers.patch_shell(PNG_IMAGE)[self.PATCH_KEY]:
+            self.assertTrue(self.adb.connect())
+
+            if isinstance(self.adb, ADBPython):
+                self.assertEqual(self.adb.screencap(), PNG_IMAGE)
+
+            else:
+                with patch.object(self.adb._adb_device, 'screencap', return_value=PNG_IMAGE):
+                    self.assertEqual(self.adb.screencap(), PNG_IMAGE)
+
 
 class TestADBServer(TestADBPython):
     """Test the `ADBServer` class."""

tests/test_basetv.py

@@ -1,6 +1,11 @@
 import sys
 import unittest
 
+try:
+    from unittest.mock import patch
+except ImportError:
+    from mock import patch
+
 
 sys.path.insert(0, '..')
 
@@ -74,6 +79,8 @@
 STATE_DETECTION_RULES_INVALID10 = {'com.amazon.tv.launcher': [{'standby': {'media_session_state': 'INVALID'}}]}
 STATE_DETECTION_RULES_INVALID11 = {'com.amazon.tv.launcher': [{'standby': {'audio_state': 123}}]}
 
+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'
+
 
 class TestBaseTVPython(unittest.TestCase):
     PATCH_KEY = 'python'
@@ -115,6 +122,13 @@ def test_adb_push(self):
             self.btv.adb_push("TEST_LOCAL_PATCH", "TEST_DEVICE_PATH")
             self.assertEqual(patch_push.call_count, 1)
 
+    def test_adb_screencap(self):
+        """Test that the ``adb_screencap`` method works correctly.
+
+        """
+        with patch.object(self.btv._adb, 'screencap', return_value=PNG_IMAGE):
+            self.assertEqual(self.btv.adb_screencap(), PNG_IMAGE)
+
     def test_keys(self):
         """Test that the key methods send the correct commands.