13221824407

Committed 09 Feb 2025 03:09AM UTC coverage: 99.892%. Remained the same

Build # 13221824407

Build Type

Pull #364

github

Committed by

web-flow

Commit Message

Merge 858979397 into 343b74ea7

Pull Request Pull Request #364: Drop support for Python 3.8 as it is EOL (Stacked PR)

Run Details

1848 of 1850 relevant lines covered (99.89%)

4.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/androidtv/adb_manager/adb_manager_sync.py

"""Classes to manage ADB connections.

* :py:class:`ADBPythonSync` utilizes a Python implementation of the ADB protocol.
* :py:class:`ADBServerSync` utilizes an ADB server to communicate with the device.

"""

from contextlib import contextmanager
import logging
import sys
import threading

from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
from ppadb.client import Client

from ..constants import (
    DEFAULT_ADB_TIMEOUT_S,
    DEFAULT_AUTH_TIMEOUT_S,
    DEFAULT_LOCK_TIMEOUT_S,
    DEFAULT_TRANSPORT_TIMEOUT_S,
)
from ..exceptions import LockNotAcquiredException

_LOGGER = logging.getLogger(__name__)

#: Use a timeout for the ADB threading lock if it is supported
LOCK_KWARGS = {"timeout": DEFAULT_LOCK_TIMEOUT_S} if sys.version_info[0] > 2 and sys.version_info[1] > 1 else {}

if sys.version_info[0] == 2:  # pragma: no cover
    FileNotFoundError = IOError  # pylint: disable=redefined-builtin


@contextmanager
def _acquire(lock):
    """Handle acquisition and release of a ``threading.Lock`` object with ``LOCK_KWARGS`` keyword arguments.

    Parameters
    ----------
    lock : threading.Lock
        The lock that we will try to acquire

    Yields
    ------
    acquired : bool
        Whether or not the lock was acquired

    Raises
    ------
    LockNotAcquiredException
        Raised if the lock was not acquired

    """
    try:
        acquired = lock.acquire(**LOCK_KWARGS)
        if not acquired:
            raise LockNotAcquiredException
        yield acquired

    finally:
        if acquired:
            lock.release()


class ADBPythonSync(object):
    """A manager for ADB connections that uses a Python implementation of the ADB protocol.

    Parameters
    ----------
    host : str
        The address of the device; may be an IP address or a host name
    port : int
        The device port to which we are connecting (default is 5555)
    adbkey : str
        The path to the ``adbkey`` file for ADB authentication
    signer : PythonRSASigner, None
        The signer for the ADB keys, as loaded by :meth:`ADBPythonSync.load_adbkey`

    """

    def __init__(self, host, port, adbkey="", signer=None):
        self.host = host
        self.port = int(port)
        self.adbkey = adbkey

        if host:
            self._adb = AdbDeviceTcp(host=self.host, port=self.port, default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)
        else:
            self._adb = AdbDeviceUsb(default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)

        self._signer = signer

        # use a lock to make sure that ADB commands don't overlap
        self._adb_lock = threading.Lock()

    @property
    def available(self):
        """Check whether the ADB connection is intact.

        Returns
        -------
        bool
            Whether or not the ADB connection is intact

        """
        return self._adb.available

    def close(self):
        """Close the ADB socket connection."""
        self._adb.close()

    def connect(
        self,
        log_errors=True,
        auth_timeout_s=DEFAULT_AUTH_TIMEOUT_S,
        transport_timeout_s=DEFAULT_TRANSPORT_TIMEOUT_S,
    ):
        """Connect to an Android TV / Fire TV device.

        Parameters
        ----------
        log_errors : bool
            Whether errors should be logged
        auth_timeout_s : float
            Authentication timeout (in seconds)
        transport_timeout_s : float
            Transport timeout (in seconds)

        Returns
        -------
        bool
            Whether or not the connection was successfully established and the device is available

        """
        try:
            with _acquire(self._adb_lock):
                # Catch exceptions
                try:
                    # Connect with authentication
                    if self.adbkey:
                        if not self._signer:
                            self._signer = self.load_adbkey(self.adbkey)

                        self._adb.connect(
                            rsa_keys=[self._signer],
                            transport_timeout_s=transport_timeout_s,
                            auth_timeout_s=auth_timeout_s,
                        )

                    # Connect without authentication
                    else:
                        self._adb.connect(transport_timeout_s=transport_timeout_s, auth_timeout_s=auth_timeout_s)

                    # ADB connection successfully established
                    _LOGGER.debug("ADB connection to %s:%d successfully established", self.host, self.port)
                    return True

                except OSError as exc:
                    if log_errors:
                        if exc.strerror is None:
                            exc.strerror = "Timed out trying to connect to ADB device."
                        _LOGGER.warning(
                            "Couldn't connect to %s:%d.  %s: %s",
                            self.host,
                            self.port,
                            exc.__class__.__name__,
                            exc.strerror,
                        )

                    # ADB connection attempt failed
                    self.close()
                    return False

                except Exception as exc:  # pylint: disable=broad-except
                    if log_errors:
                        _LOGGER.warning(
                            "Couldn't connect to %s:%d.  %s: %s", self.host, self.port, exc.__class__.__name__, exc
                        )

                    # ADB connection attempt failed
                    self.close()
                    return False

        except LockNotAcquiredException:
            _LOGGER.warning("Couldn't connect to %s:%d because adb-shell lock not acquired.", self.host, self.port)
            self.close()
            return False

    @staticmethod
    def load_adbkey(adbkey):
        """Load the ADB keys.

        Parameters
        ----------
        adbkey : str
            The path to the ``adbkey`` file for ADB authentication

        Returns
        -------
        PythonRSASigner
            The ``PythonRSASigner`` with the key files loaded

        """
        # private key
        with open(adbkey) as f:
            priv = f.read()

        # public key
        try:
            with open(adbkey + ".pub") as f:
                pub = f.read()
        except FileNotFoundError:
            pub = ""

        return PythonRSASigner(pub, priv)

    def pull(self, local_path, device_path):
        """Pull a file from the device using the Python ADB implementation.

        Parameters
        ----------
        local_path : str
            The path where the file will be saved
        device_path : str
            The file on the device that will be pulled

        """
        if not self.available:
            _LOGGER.debug(
                "ADB command not sent to %s:%d because adb-shell connection is not established: pull(%s, %s)",
                self.host,
                self.port,
                local_path,
                device_path,
            )
            return

        with _acquire(self._adb_lock):
            _LOGGER.debug(
                "Sending command to %s:%d via adb-shell: pull(%s, %s)", self.host, self.port, local_path, device_path
            )
            self._adb.pull(device_path, local_path)
            return

    def push(self, local_path, device_path):
        """Push a file to the device using the Python ADB implementation.

        Parameters
        ----------
        local_path : str
            The file that will be pushed to the device
        device_path : str
            The path where the file will be saved on the device

        """
        if not self.available:
            _LOGGER.debug(
                "ADB command not sent to %s:%d because adb-shell connection is not established: push(%s, %s)",
                self.host,
                self.port,
                local_path,
                device_path,
            )
            return

        with _acquire(self._adb_lock):
            _LOGGER.debug(
                "Sending command to %s:%d via adb-shell: push(%s, %s)", self.host, self.port, 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)
            result = self._adb.shell("screencap -p", decode=False)
            if result and result[5:6] == b"\r":
                return result.replace(b"\r\n", b"\n")
            return result

    def shell(self, cmd):
        """Send an ADB command using the Python ADB implementation.

        Parameters
        ----------
        cmd : str
            The ADB command to be sent

        Returns
        -------
        str, None
            The response from the device, if there is a response

        """
        if not self.available:
            _LOGGER.debug(
                "ADB 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 command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
            return self._adb.shell(cmd)


class ADBServerSync(object):
    """A manager for ADB connections that uses an ADB server.

    Parameters
    ----------
    host : str
        The address of the device; may be an IP address or a host name
    port : int
        The device port to which we are connecting (default is 5555)
    adb_server_ip : str
        The IP address of the ADB server
    adb_server_port : int
        The port for the ADB server

    """

    def __init__(self, host, port=5555, adb_server_ip="", adb_server_port=5037):
        self.host = host
        self.port = int(port)
        self.adb_server_ip = adb_server_ip
        self.adb_server_port = adb_server_port
        self._adb_client = None
        self._adb_device = None

        # keep track of whether the ADB connection is intact
        self._available = False

        # use a lock to make sure that ADB commands don't overlap
        self._adb_lock = threading.Lock()

    @property
    def available(self):
        """Check whether the ADB connection is intact.

        Returns
        -------
        bool
            Whether or not the ADB connection is intact

        """
        if not self._adb_client or not self._adb_device:
            return False

        return self._available

    def close(self):
        """Close the ADB server socket connection.

        Currently, this doesn't do anything except set ``self._available = False``.

        """
        self._available = False

    def connect(self, log_errors=True):
        """Connect to an Android TV / Fire TV device.

        Parameters
        ----------
        log_errors : bool
            Whether errors should be logged

        Returns
        -------
        bool
            Whether or not the connection was successfully established and the device is available

        """
        try:
            with _acquire(self._adb_lock):
                # Catch exceptions
                try:
                    self._adb_client = Client(host=self.adb_server_ip, port=self.adb_server_port)
                    self._adb_device = self._adb_client.device("{}:{}".format(self.host, self.port))

                    # ADB connection successfully established
                    if self._adb_device:
                        _LOGGER.debug(
                            "ADB connection to %s:%d via ADB server %s:%d successfully established",
                            self.host,
                            self.port,
                            self.adb_server_ip,
                            self.adb_server_port,
                        )
                        self._available = True
                        return True

                    # ADB connection attempt failed (without an exception)
                    if log_errors:
                        _LOGGER.warning(
                            "Couldn't connect to %s:%d via ADB server %s:%d because the server is not connected to the device",
                            self.host,
                            self.port,
                            self.adb_server_ip,
                            self.adb_server_port,
                        )

                    self.close()
                    self._available = False
                    return False

                # ADB connection attempt failed
                except Exception as exc:  # noqa pylint: disable=broad-except
                    if log_errors:
                        _LOGGER.warning(
                            "Couldn't connect to %s:%d via ADB server %s:%d, error: %s",
                            self.host,
                            self.port,
                            self.adb_server_ip,
                            self.adb_server_port,
                            exc,
                        )

                    self.close()
                    self._available = False
                    return False

        except LockNotAcquiredException:
            _LOGGER.warning(
                "Couldn't connect to %s:%d via ADB server %s:%d because pure-python-adb lock not acquired.",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
            )
            self.close()
            self._available = False
            return False

    def pull(self, local_path, device_path):
        """Pull a file from the device using an ADB server.

        Parameters
        ----------
        local_path : str
            The path where the file will be saved
        device_path : str
            The file on the device that will be pulled

        """
        if not self.available:
            _LOGGER.debug(
                "ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: pull(%s, %s)",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                local_path,
                device_path,
            )
            return

        with _acquire(self._adb_lock):
            _LOGGER.debug(
                "Sending command to %s:%d via ADB server %s:%d: pull(%s, %s)",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                local_path,
                device_path,
            )
            self._adb_device.pull(device_path, local_path)
            return

    def push(self, local_path, device_path):
        """Push a file to the device using an ADB server.

        Parameters
        ----------
        local_path : str
            The file that will be pushed to the device
        device_path : str
            The path where the file will be saved on the device

        """
        if not self.available:
            _LOGGER.debug(
                "ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: push(%s, %s)",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                local_path,
                device_path,
            )
            return

        with _acquire(self._adb_lock):
            _LOGGER.debug(
                "Sending command to %s:%d via ADB server %s:%d: push(%s, %s)",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                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, None
            The screencap as a binary .png image, or ``None`` if there was an ``IndexError`` exception

        """
        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.

        Parameters
        ----------
        cmd : str
            The ADB command to be sent

        Returns
        -------
        str, None
            The response from the device, if there is a response

        """
        if not self.available:
            _LOGGER.debug(
                "ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: %s",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                cmd,
            )
            return None

        with _acquire(self._adb_lock):
            _LOGGER.debug(
                "Sending command to %s:%d via ADB server %s:%d: %s",
                self.host,
                self.port,
                self.adb_server_ip,
                self.adb_server_port,
                cmd,
            )
            return self._adb_device.shell(cmd)

1	"""Classes to manage ADB connections.
2
3	* :py:class:`ADBPythonSync` utilizes a Python implementation of the ADB protocol.
4	* :py:class:`ADBServerSync` utilizes an ADB server to communicate with the device.
5
6	"""
7
8	from contextlib import contextmanager	4✔
9	import logging	4✔
10	import sys	4✔
11	import threading	4✔
12
13	from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb	4✔
14	from adb_shell.auth.sign_pythonrsa import PythonRSASigner	4✔
15	from ppadb.client import Client	4✔
16
17	from ..constants import (	4✔
18	DEFAULT_ADB_TIMEOUT_S,
19	DEFAULT_AUTH_TIMEOUT_S,
20	DEFAULT_LOCK_TIMEOUT_S,
21	DEFAULT_TRANSPORT_TIMEOUT_S,
22	)
23	from ..exceptions import LockNotAcquiredException	4✔
24
25	_LOGGER = logging.getLogger(__name__)	4✔
26
27	#: Use a timeout for the ADB threading lock if it is supported
28	LOCK_KWARGS = {"timeout": DEFAULT_LOCK_TIMEOUT_S} if sys.version_info[0] > 2 and sys.version_info[1] > 1 else {}	4✔
29
30	if sys.version_info[0] == 2: # pragma: no cover
31	FileNotFoundError = IOError # pylint: disable=redefined-builtin
32
33
34	@contextmanager	4✔
35	def _acquire(lock):	4✔
36	"""Handle acquisition and release of a ``threading.Lock`` object with ``LOCK_KWARGS`` keyword arguments.
37
38	Parameters
39	----------
40	lock : threading.Lock
41	The lock that we will try to acquire
42
43	Yields
44	------
45	acquired : bool
46	Whether or not the lock was acquired
47
48	Raises
49	------
50	LockNotAcquiredException
51	Raised if the lock was not acquired
52
53	"""
54	try:	4✔
55	acquired = lock.acquire(**LOCK_KWARGS)	4✔
56	if not acquired:	4✔
57	raise LockNotAcquiredException	4✔
58	yield acquired	4✔
59
60	finally:
61	if acquired:	4✔
62	lock.release()	4✔
63
64
65	class ADBPythonSync(object):	4✔
66	"""A manager for ADB connections that uses a Python implementation of the ADB protocol.
67
68	Parameters
69	----------
70	host : str
71	The address of the device; may be an IP address or a host name
72	port : int
73	The device port to which we are connecting (default is 5555)
74	adbkey : str
75	The path to the ``adbkey`` file for ADB authentication
76	signer : PythonRSASigner, None
77	The signer for the ADB keys, as loaded by :meth:`ADBPythonSync.load_adbkey`
78
79	"""
80
81	def __init__(self, host, port, adbkey="", signer=None):	4✔
82	self.host = host	4✔
83	self.port = int(port)	4✔
84	self.adbkey = adbkey	4✔
85
86	if host:	4✔
87	self._adb = AdbDeviceTcp(host=self.host, port=self.port, default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)	4✔
88	else:
89	self._adb = AdbDeviceUsb(default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)	4✔
90
91	self._signer = signer	4✔
92
93	# use a lock to make sure that ADB commands don't overlap
94	self._adb_lock = threading.Lock()	4✔
95
96	@property	4✔
97	def available(self):	4✔
98	"""Check whether the ADB connection is intact.
99
100	Returns
101	-------
102	bool
103	Whether or not the ADB connection is intact
104
105	"""
106	return self._adb.available	4✔
107
108	def close(self):	4✔
109	"""Close the ADB socket connection."""
110	self._adb.close()	4✔
111
112	def connect(	4✔
113	self,
114	log_errors=True,
115	auth_timeout_s=DEFAULT_AUTH_TIMEOUT_S,
116	transport_timeout_s=DEFAULT_TRANSPORT_TIMEOUT_S,
117	):
118	"""Connect to an Android TV / Fire TV device.
119
120	Parameters
121	----------
122	log_errors : bool
123	Whether errors should be logged
124	auth_timeout_s : float
125	Authentication timeout (in seconds)
126	transport_timeout_s : float
127	Transport timeout (in seconds)
128
129	Returns
130	-------
131	bool
132	Whether or not the connection was successfully established and the device is available
133
134	"""
135	try:	4✔
136	with _acquire(self._adb_lock):	4✔
137	# Catch exceptions
138	try:	4✔
139	# Connect with authentication
140	if self.adbkey:	4✔
141	if not self._signer:	4✔
142	self._signer = self.load_adbkey(self.adbkey)	4✔
143
144	self._adb.connect(	4✔
145	rsa_keys=[self._signer],
146	transport_timeout_s=transport_timeout_s,
147	auth_timeout_s=auth_timeout_s,
148	)
149
150	# Connect without authentication
151	else:
152	self._adb.connect(transport_timeout_s=transport_timeout_s, auth_timeout_s=auth_timeout_s)	4✔
153
154	# ADB connection successfully established
155	_LOGGER.debug("ADB connection to %s:%d successfully established", self.host, self.port)	4✔
156	return True	4✔
157
158	except OSError as exc:	4✔
159	if log_errors:	4✔
160	if exc.strerror is None:	4✔
161	exc.strerror = "Timed out trying to connect to ADB device."	4✔
162	_LOGGER.warning(	4✔
163	"Couldn't connect to %s:%d. %s: %s",
164	self.host,
165	self.port,
166	exc.__class__.__name__,
167	exc.strerror,
168	)
169
170	# ADB connection attempt failed
171	self.close()	4✔
172	return False	4✔
173
174	except Exception as exc: # pylint: disable=broad-except	4✔
175	if log_errors:	4✔
176	_LOGGER.warning(	4✔
177	"Couldn't connect to %s:%d. %s: %s", self.host, self.port, exc.__class__.__name__, exc
178	)
179
180	# ADB connection attempt failed
181	self.close()	4✔
182	return False	4✔
183
184	except LockNotAcquiredException:	4✔
185	_LOGGER.warning("Couldn't connect to %s:%d because adb-shell lock not acquired.", self.host, self.port)	4✔
186	self.close()	4✔
187	return False	4✔
188
189	@staticmethod	4✔
190	def load_adbkey(adbkey):	4✔
191	"""Load the ADB keys.
192
193	Parameters
194	----------
195	adbkey : str
196	The path to the ``adbkey`` file for ADB authentication
197
198	Returns
199	-------
200	PythonRSASigner
201	The ``PythonRSASigner`` with the key files loaded
202
203	"""
204	# private key
205	with open(adbkey) as f:	4✔
206	priv = f.read()	4✔
207
208	# public key
209	try:	4✔
210	with open(adbkey + ".pub") as f:	4✔
211	pub = f.read()	4✔
212	except FileNotFoundError:	4✔
213	pub = ""	4✔
214
215	return PythonRSASigner(pub, priv)	4✔
216
217	def pull(self, local_path, device_path):	4✔
218	"""Pull a file from the device using the Python ADB implementation.
219
220	Parameters
221	----------
222	local_path : str
223	The path where the file will be saved
224	device_path : str
225	The file on the device that will be pulled
226
227	"""
228	if not self.available:	4✔
229	_LOGGER.debug(	4✔
230	"ADB command not sent to %s:%d because adb-shell connection is not established: pull(%s, %s)",
231	self.host,
232	self.port,
233	local_path,
234	device_path,
235	)
236	return	4✔
237
238	with _acquire(self._adb_lock):	4✔
239	_LOGGER.debug(	4✔
240	"Sending command to %s:%d via adb-shell: pull(%s, %s)", self.host, self.port, local_path, device_path
241	)
242	self._adb.pull(device_path, local_path)	4✔
243	return	4✔
244
245	def push(self, local_path, device_path):	4✔
246	"""Push a file to the device using the Python ADB implementation.
247
248	Parameters
249	----------
250	local_path : str
251	The file that will be pushed to the device
252	device_path : str
253	The path where the file will be saved on the device
254
255	"""
256	if not self.available:	4✔
257	_LOGGER.debug(	4✔
258	"ADB command not sent to %s:%d because adb-shell connection is not established: push(%s, %s)",
259	self.host,
260	self.port,
261	local_path,
262	device_path,
263	)
264	return	4✔
265
266	with _acquire(self._adb_lock):	4✔
267	_LOGGER.debug(	4✔
268	"Sending command to %s:%d via adb-shell: push(%s, %s)", self.host, self.port, local_path, device_path
269	)
270	self._adb.push(local_path, device_path)	4✔
271	return	4✔
272
273	def screencap(self):	4✔
274	"""Take a screenshot using the Python ADB implementation.
275
276	Returns
277	-------
278	bytes
279	The screencap as a binary .png image
280
281	"""
282	if not self.available:	4✔
283	_LOGGER.debug(	4✔
284	"ADB screencap not taken from %s:%d because adb-shell connection is not established",
285	self.host,
286	self.port,
287	)
288	return None	4✔
289
290	with _acquire(self._adb_lock):	4✔
291	_LOGGER.debug("Taking screencap from %s:%d via adb-shell", self.host, self.port)	4✔
292	result = self._adb.shell("screencap -p", decode=False)	4✔
293	if result and result[5:6] == b"\r":	4✔
294	return result.replace(b"\r\n", b"\n")	4✔
295	return result	4✔
296
297	def shell(self, cmd):	4✔
298	"""Send an ADB command using the Python ADB implementation.
299
300	Parameters
301	----------
302	cmd : str
303	The ADB command to be sent
304
305	Returns
306	-------
307	str, None
308	The response from the device, if there is a response
309
310	"""
311	if not self.available:	4✔
312	_LOGGER.debug(	4✔
313	"ADB command not sent to %s:%d because adb-shell connection is not established: %s",
314	self.host,
315	self.port,
316	cmd,
317	)
318	return None	4✔
319
320	with _acquire(self._adb_lock):	4✔
321	_LOGGER.debug("Sending command to %s:%d via adb-shell: %s", self.host, self.port, cmd)	4✔
322	return self._adb.shell(cmd)	4✔
323
324
325	class ADBServerSync(object):	4✔
326	"""A manager for ADB connections that uses an ADB server.
327
328	Parameters
329	----------
330	host : str
331	The address of the device; may be an IP address or a host name
332	port : int
333	The device port to which we are connecting (default is 5555)
334	adb_server_ip : str
335	The IP address of the ADB server
336	adb_server_port : int
337	The port for the ADB server
338
339	"""
340
341	def __init__(self, host, port=5555, adb_server_ip="", adb_server_port=5037):	4✔
342	self.host = host	4✔
343	self.port = int(port)	4✔
344	self.adb_server_ip = adb_server_ip	4✔
345	self.adb_server_port = adb_server_port	4✔
346	self._adb_client = None	4✔
347	self._adb_device = None	4✔
348
349	# keep track of whether the ADB connection is intact
350	self._available = False	4✔
351
352	# use a lock to make sure that ADB commands don't overlap
353	self._adb_lock = threading.Lock()	4✔
354
355	@property	4✔
356	def available(self):	4✔
357	"""Check whether the ADB connection is intact.
358
359	Returns
360	-------
361	bool
362	Whether or not the ADB connection is intact
363
364	"""
365	if not self._adb_client or not self._adb_device:	4✔
366	return False	4✔
367
368	return self._available	4✔
369
370	def close(self):	4✔
371	"""Close the ADB server socket connection.
372
373	Currently, this doesn't do anything except set ``self._available = False``.
374
375	"""
376	self._available = False	4✔
377
378	def connect(self, log_errors=True):	4✔
379	"""Connect to an Android TV / Fire TV device.
380
381	Parameters
382	----------
383	log_errors : bool
384	Whether errors should be logged
385
386	Returns
387	-------
388	bool
389	Whether or not the connection was successfully established and the device is available
390
391	"""
392	try:	4✔
393	with _acquire(self._adb_lock):	4✔
394	# Catch exceptions
395	try:	4✔
396	self._adb_client = Client(host=self.adb_server_ip, port=self.adb_server_port)	4✔
397	self._adb_device = self._adb_client.device("{}:{}".format(self.host, self.port))	4✔
398
399	# ADB connection successfully established
400	if self._adb_device:	4✔
401	_LOGGER.debug(	4✔
402	"ADB connection to %s:%d via ADB server %s:%d successfully established",
403	self.host,
404	self.port,
405	self.adb_server_ip,
406	self.adb_server_port,
407	)
408	self._available = True	4✔
409	return True	4✔
410
411	# ADB connection attempt failed (without an exception)
412	if log_errors:	4✔
413	_LOGGER.warning(	4✔
414	"Couldn't connect to %s:%d via ADB server %s:%d because the server is not connected to the device",
415	self.host,
416	self.port,
417	self.adb_server_ip,
418	self.adb_server_port,
419	)
420
421	self.close()	4✔
422	self._available = False	4✔
423	return False	4✔
424
425	# ADB connection attempt failed
426	except Exception as exc: # noqa pylint: disable=broad-except	4✔
427	if log_errors:	4✔
428	_LOGGER.warning(	4✔
429	"Couldn't connect to %s:%d via ADB server %s:%d, error: %s",
430	self.host,
431	self.port,
432	self.adb_server_ip,
433	self.adb_server_port,
434	exc,
435	)
436
437	self.close()	4✔
438	self._available = False	4✔
439	return False	4✔
440
441	except LockNotAcquiredException:	4✔
442	_LOGGER.warning(	4✔
443	"Couldn't connect to %s:%d via ADB server %s:%d because pure-python-adb lock not acquired.",
444	self.host,
445	self.port,
446	self.adb_server_ip,
447	self.adb_server_port,
448	)
449	self.close()	4✔
450	self._available = False	4✔
451	return False	4✔
452
453	def pull(self, local_path, device_path):	4✔
454	"""Pull a file from the device using an ADB server.
455
456	Parameters
457	----------
458	local_path : str
459	The path where the file will be saved
460	device_path : str
461	The file on the device that will be pulled
462
463	"""
464	if not self.available:	4✔
465	_LOGGER.debug(	4✔
466	"ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: pull(%s, %s)",
467	self.host,
468	self.port,
469	self.adb_server_ip,
470	self.adb_server_port,
471	local_path,
472	device_path,
473	)
474	return	4✔
475
476	with _acquire(self._adb_lock):	4✔
477	_LOGGER.debug(	4✔
478	"Sending command to %s:%d via ADB server %s:%d: pull(%s, %s)",
479	self.host,
480	self.port,
481	self.adb_server_ip,
482	self.adb_server_port,
483	local_path,
484	device_path,
485	)
486	self._adb_device.pull(device_path, local_path)	4✔
487	return	4✔
488
489	def push(self, local_path, device_path):	4✔
490	"""Push a file to the device using an ADB server.
491
492	Parameters
493	----------
494	local_path : str
495	The file that will be pushed to the device
496	device_path : str
497	The path where the file will be saved on the device
498
499	"""
500	if not self.available:	4✔
501	_LOGGER.debug(	4✔
502	"ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: push(%s, %s)",
503	self.host,
504	self.port,
505	self.adb_server_ip,
506	self.adb_server_port,
507	local_path,
508	device_path,
509	)
510	return	4✔
511
512	with _acquire(self._adb_lock):	4✔
513	_LOGGER.debug(	4✔
514	"Sending command to %s:%d via ADB server %s:%d: push(%s, %s)",
515	self.host,
516	self.port,
517	self.adb_server_ip,
518	self.adb_server_port,
519	local_path,
520	device_path,
521	)
522	self._adb_device.push(local_path, device_path)	4✔
523	return	4✔
524
525	def screencap(self):	4✔
526	"""Take a screenshot using an ADB server.
527
528	Returns
529	-------
530	bytes, None
531	The screencap as a binary .png image, or ``None`` if there was an ``IndexError`` exception
532
533	"""
534	if not self.available:	4✔
535	_LOGGER.debug(	4✔
536	"ADB screencap not taken from %s:%d via ADB server %s:%d because pure-python-adb connection is not established",
537	self.host,
538	self.port,
539	self.adb_server_ip,
540	self.adb_server_port,
541	)
542	return None	4✔
543
544	with _acquire(self._adb_lock):	4✔
545	_LOGGER.debug(	4✔
546	"Taking screencap from %s:%d via ADB server %s:%d",
547	self.host,
548	self.port,
549	self.adb_server_ip,
550	self.adb_server_port,
551	)
552	return self._adb_device.screencap()	4✔
553
554	def shell(self, cmd):	4✔
555	"""Send an ADB command using an ADB server.
556
557	Parameters
558	----------
559	cmd : str
560	The ADB command to be sent
561
562	Returns
563	-------
564	str, None
565	The response from the device, if there is a response
566
567	"""
568	if not self.available:	4✔
569	_LOGGER.debug(	4✔
570	"ADB command not sent to %s:%d via ADB server %s:%d because pure-python-adb connection is not established: %s",
571	self.host,
572	self.port,
573	self.adb_server_ip,
574	self.adb_server_port,
575	cmd,
576	)
577	return None	4✔
578
579	with _acquire(self._adb_lock):	4✔
580	_LOGGER.debug(	4✔
581	"Sending command to %s:%d via ADB server %s:%d: %s",
582	self.host,
583	self.port,
584	self.adb_server_ip,
585	self.adb_server_port,
586	cmd,
587	)
588	return self._adb_device.shell(cmd)	4✔

JeffLIrion / python-androidtv / 13221824407

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous