• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

JeffLIrion / python-androidtv / 13221849111

09 Feb 2025 03:12AM CUT coverage: 99.892%. Remained the same
13221849111

Pull #365

github

web-flow
Merge ada048ea2 into 343b74ea7
Pull Request #365: Add Python 3.13 to CI

1848 of 1850 relevant lines covered (99.89%)

4.99 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
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
5✔
9
import logging
5✔
10
import sys
5✔
11
import threading
5✔
12

13
from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
5✔
14
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
5✔
15
from ppadb.client import Client
5✔
16

17
from ..constants import (
5✔
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
5✔
24

25
_LOGGER = logging.getLogger(__name__)
5✔
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 {}
5✔
29

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

33

34
@contextmanager
5✔
35
def _acquire(lock):
5✔
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:
5✔
55
        acquired = lock.acquire(**LOCK_KWARGS)
5✔
56
        if not acquired:
5✔
57
            raise LockNotAcquiredException
5✔
58
        yield acquired
5✔
59

60
    finally:
61
        if acquired:
5✔
62
            lock.release()
5✔
63

64

65
class ADBPythonSync(object):
5✔
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):
5✔
82
        self.host = host
5✔
83
        self.port = int(port)
5✔
84
        self.adbkey = adbkey
5✔
85

86
        if host:
5✔
87
            self._adb = AdbDeviceTcp(host=self.host, port=self.port, default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)
5✔
88
        else:
89
            self._adb = AdbDeviceUsb(default_transport_timeout_s=DEFAULT_ADB_TIMEOUT_S)
5✔
90

91
        self._signer = signer
5✔
92

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

96
    @property
5✔
97
    def available(self):
5✔
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
5✔
107

108
    def close(self):
5✔
109
        """Close the ADB socket connection."""
110
        self._adb.close()
5✔
111

112
    def connect(
5✔
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:
5✔
136
            with _acquire(self._adb_lock):
5✔
137
                # Catch exceptions
138
                try:
5✔
139
                    # Connect with authentication
140
                    if self.adbkey:
5✔
141
                        if not self._signer:
5✔
142
                            self._signer = self.load_adbkey(self.adbkey)
5✔
143

144
                        self._adb.connect(
5✔
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)
5✔
153

154
                    # ADB connection successfully established
155
                    _LOGGER.debug("ADB connection to %s:%d successfully established", self.host, self.port)
5✔
156
                    return True
5✔
157

158
                except OSError as exc:
5✔
159
                    if log_errors:
5✔
160
                        if exc.strerror is None:
5✔
161
                            exc.strerror = "Timed out trying to connect to ADB device."
5✔
162
                        _LOGGER.warning(
5✔
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()
5✔
172
                    return False
5✔
173

174
                except Exception as exc:  # pylint: disable=broad-except
5✔
175
                    if log_errors:
5✔
176
                        _LOGGER.warning(
5✔
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()
5✔
182
                    return False
5✔
183

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

189
    @staticmethod
5✔
190
    def load_adbkey(adbkey):
5✔
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:
5✔
206
            priv = f.read()
5✔
207

208
        # public key
209
        try:
5✔
210
            with open(adbkey + ".pub") as f:
5✔
211
                pub = f.read()
5✔
212
        except FileNotFoundError:
5✔
213
            pub = ""
5✔
214

215
        return PythonRSASigner(pub, priv)
5✔
216

217
    def pull(self, local_path, device_path):
5✔
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:
5✔
229
            _LOGGER.debug(
5✔
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
5✔
237

238
        with _acquire(self._adb_lock):
5✔
239
            _LOGGER.debug(
5✔
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)
5✔
243
            return
5✔
244

245
    def push(self, local_path, device_path):
5✔
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:
5✔
257
            _LOGGER.debug(
5✔
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
5✔
265

266
        with _acquire(self._adb_lock):
5✔
267
            _LOGGER.debug(
5✔
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)
5✔
271
            return
5✔
272

273
    def screencap(self):
5✔
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:
5✔
283
            _LOGGER.debug(
5✔
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
5✔
289

290
        with _acquire(self._adb_lock):
5✔
291
            _LOGGER.debug("Taking screencap from %s:%d via adb-shell", self.host, self.port)
5✔
292
            result = self._adb.shell("screencap -p", decode=False)
5✔
293
            if result and result[5:6] == b"\r":
5✔
294
                return result.replace(b"\r\n", b"\n")
5✔
295
            return result
5✔
296

297
    def shell(self, cmd):
5✔
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:
5✔
312
            _LOGGER.debug(
5✔
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
5✔
319

320
        with _acquire(self._adb_lock):
5✔
321
            _LOGGER.debug("Sending command to %s:%d via adb-shell: %s", self.host, self.port, cmd)
5✔
322
            return self._adb.shell(cmd)
5✔
323

324

325
class ADBServerSync(object):
5✔
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):
5✔
342
        self.host = host
5✔
343
        self.port = int(port)
5✔
344
        self.adb_server_ip = adb_server_ip
5✔
345
        self.adb_server_port = adb_server_port
5✔
346
        self._adb_client = None
5✔
347
        self._adb_device = None
5✔
348

349
        # keep track of whether the ADB connection is intact
350
        self._available = False
5✔
351

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

355
    @property
5✔
356
    def available(self):
5✔
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:
5✔
366
            return False
5✔
367

368
        return self._available
5✔
369

370
    def close(self):
5✔
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
5✔
377

378
    def connect(self, log_errors=True):
5✔
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:
5✔
393
            with _acquire(self._adb_lock):
5✔
394
                # Catch exceptions
395
                try:
5✔
396
                    self._adb_client = Client(host=self.adb_server_ip, port=self.adb_server_port)
5✔
397
                    self._adb_device = self._adb_client.device("{}:{}".format(self.host, self.port))
5✔
398

399
                    # ADB connection successfully established
400
                    if self._adb_device:
5✔
401
                        _LOGGER.debug(
5✔
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
5✔
409
                        return True
5✔
410

411
                    # ADB connection attempt failed (without an exception)
412
                    if log_errors:
5✔
413
                        _LOGGER.warning(
5✔
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()
5✔
422
                    self._available = False
5✔
423
                    return False
5✔
424

425
                # ADB connection attempt failed
426
                except Exception as exc:  # noqa pylint: disable=broad-except
5✔
427
                    if log_errors:
5✔
428
                        _LOGGER.warning(
5✔
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()
5✔
438
                    self._available = False
5✔
439
                    return False
5✔
440

441
        except LockNotAcquiredException:
5✔
442
            _LOGGER.warning(
5✔
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()
5✔
450
            self._available = False
5✔
451
            return False
5✔
452

453
    def pull(self, local_path, device_path):
5✔
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:
5✔
465
            _LOGGER.debug(
5✔
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
5✔
475

476
        with _acquire(self._adb_lock):
5✔
477
            _LOGGER.debug(
5✔
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)
5✔
487
            return
5✔
488

489
    def push(self, local_path, device_path):
5✔
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:
5✔
501
            _LOGGER.debug(
5✔
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
5✔
511

512
        with _acquire(self._adb_lock):
5✔
513
            _LOGGER.debug(
5✔
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)
5✔
523
            return
5✔
524

525
    def screencap(self):
5✔
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:
5✔
535
            _LOGGER.debug(
5✔
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
5✔
543

544
        with _acquire(self._adb_lock):
5✔
545
            _LOGGER.debug(
5✔
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()
5✔
553

554
    def shell(self, cmd):
5✔
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:
5✔
569
            _LOGGER.debug(
5✔
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
5✔
578

579
        with _acquire(self._adb_lock):
5✔
580
            _LOGGER.debug(
5✔
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)
5✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc