# Copyright (c) 2021 Jeff Irion and contributors
#
# This file is part of the adb-shell package. It incorporates work
# covered by the following license notice:
#
#
# Copyright 2014 Google Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Implement the :class:`AdbDeviceAsync` class, which can connect to a device and run ADB shell commands.
* :class:`_AdbIOManagerAsync`
* :meth:`_AdbIOManagerAsync._read_bytes_from_device`
* :meth:`_AdbIOManagerAsync._read_expected_packet_from_device`
* :meth:`_AdbIOManagerAsync._read_packet_from_device`
* :meth:`_AdbIOManagerAsync._send`
* :meth:`_AdbIOManagerAsync.close`
* :meth:`_AdbIOManagerAsync.connect`
* :meth:`_AdbIOManagerAsync.read`
* :meth:`_AdbIOManagerAsync.send`
* :class:`_AsyncBytesIO`
* :meth:`_AsyncBytesIO.read`
* :meth:`_AsyncBytesIO.write`
* :func:`_open_bytesio`
* :class:`AdbDeviceAsync`
* :meth:`AdbDeviceAsync._clse`
* :meth:`AdbDeviceAsync._filesync_flush`
* :meth:`AdbDeviceAsync._filesync_read`
* :meth:`AdbDeviceAsync._filesync_read_buffered`
* :meth:`AdbDeviceAsync._filesync_read_until`
* :meth:`AdbDeviceAsync._filesync_send`
* :meth:`AdbDeviceAsync._okay`
* :meth:`AdbDeviceAsync._open`
* :meth:`AdbDeviceAsync._pull`
* :meth:`AdbDeviceAsync._push`
* :meth:`AdbDeviceAsync._read_until`
* :meth:`AdbDeviceAsync._read_until_close`
* :meth:`AdbDeviceAsync._service`
* :meth:`AdbDeviceAsync._streaming_command`
* :meth:`AdbDeviceAsync._streaming_service`
* :attr:`AdbDeviceAsync.available`
* :meth:`AdbDeviceAsync.close`
* :meth:`AdbDeviceAsync.connect`
* :meth:`AdbDeviceAsync.list`
* :attr:`AdbDeviceAsync.max_chunk_size`
* :meth:`AdbDeviceAsync.pull`
* :meth:`AdbDeviceAsync.push`
* :meth:`AdbDeviceAsync.root`
* :meth:`AdbDeviceAsync.shell`
* :meth:`AdbDeviceAsync.stat`
* :meth:`AdbDeviceAsync.streaming_shell`
* :class:`AdbDeviceTcpAsync`
"""
from contextlib import asynccontextmanager
from io import BytesIO
from asyncio import Lock, get_running_loop
import logging
import os
import struct
import time
import aiofiles
from . import constants
from . import exceptions
from .adb_message import AdbMessage, checksum, int_to_cmd, unpack
from .transport.base_transport_async import BaseTransportAsync
from .transport.tcp_transport_async import TcpTransportAsync
from .hidden_helpers import DeviceFile, _AdbPacketStore, _AdbTransactionInfo, _FileSyncTransactionInfo, get_banner, get_files_to_push
_LOGGER = logging.getLogger(__name__)
[docs]
class _AsyncBytesIO:
"""An async wrapper for `BytesIO`.
Parameters
----------
bytesio : BytesIO
The BytesIO object that is wrapped
"""
def __init__(self, bytesio):
self._bytesio = bytesio
[docs]
async def read(self, size=-1):
"""Read data.
Parameters
----------
size : int
The size of the data to be read
Returns
-------
bytes
The data that was read
"""
return self._bytesio.read(size)
[docs]
async def write(self, data):
"""Write data.
Parameters
----------
data : bytes
The data to be written
"""
self._bytesio.write(data)
[docs]
@asynccontextmanager
async def _open_bytesio(stream, *args, **kwargs): # pylint: disable=unused-argument
"""An async context manager for a BytesIO object that does nothing.
Parameters
----------
stream : BytesIO
The BytesIO stream
args : list
Unused positional arguments
kwargs : dict
Unused keyword arguments
Yields
------
_AsyncBytesIO
The wrapped `stream` input parameter
"""
yield _AsyncBytesIO(stream)
[docs]
class _AdbIOManagerAsync(object):
"""A class for handling all ADB I/O.
Notes
-----
When the ``self._store_lock`` and ``self._transport_lock`` locks are held at the same time, it must always be the
case that the ``self._transport_lock`` is acquired first. This ensures that there is no potential for deadlock.
Parameters
----------
transport : BaseTransportAsync
A transport for communicating with the device; must be an instance of a subclass of :class:`~adb_shell.transport.base_transport_async.BaseTransportAsync`
Attributes
----------
_packet_store : _AdbPacketStore
A store for holding packets that correspond to different ADB streams
_store_lock : Lock
A lock for protecting ``self._packet_store`` (this lock is never held for long)
_transport : BaseTransportAsync
A transport for communicating with the device; must be an instance of a subclass of :class:`~adb_shell.transport.base_transport_async.BaseTransportAsync`
_transport_lock : Lock
A lock for protecting ``self._transport``
"""
def __init__(self, transport):
self._packet_store = _AdbPacketStore()
self._transport = transport
self._store_lock = Lock()
self._transport_lock = Lock()
[docs]
async def close(self):
"""Close the connection via the provided transport's ``close()`` method and clear the packet store.
"""
async with self._transport_lock:
await self._transport.close()
async with self._store_lock:
self._packet_store.clear_all()
[docs]
async def connect(self, banner, rsa_keys, auth_timeout_s, auth_callback, adb_info):
"""Establish an ADB connection to the device.
1. Use the transport to establish a connection
2. Send a ``b'CNXN'`` message
3. Read the response from the device
4. If ``cmd`` is not ``b'AUTH'``, then authentication is not necesary and so we are done
5. If no ``rsa_keys`` are provided, raise an exception
6. Loop through our keys, signing the last ``banner2`` that we received
1. If the last ``arg0`` was not :const:`adb_shell.constants.AUTH_TOKEN`, raise an exception
2. Sign the last ``banner2`` and send it in an ``b'AUTH'`` message
3. Read the response from the device
4. If ``cmd`` is ``b'CNXN'``, we are done
7. None of the keys worked, so send ``rsa_keys[0]``'s public key; if the response does not time out, we must have connected successfully
Parameters
----------
banner : bytearray, bytes
The hostname of the machine where the Python interpreter is currently running (:attr:`adb_shell.adb_device_async.AdbDeviceAsync._banner`)
rsa_keys : list, None
A list of signers of type :class:`~adb_shell.auth.sign_cryptography.CryptographySigner`,
:class:`~adb_shell.auth.sign_pycryptodome.PycryptodomeAuthSigner`, or :class:`~adb_shell.auth.sign_pythonrsa.PythonRSASigner`
auth_timeout_s : float, None
The time in seconds to wait for a ``b'CNXN'`` authentication response
auth_callback : function, None
Function callback invoked when the connection needs to be accepted on the device
adb_info : _AdbTransactionInfo
Info and settings for this connection attempt
Returns
-------
bool
Whether the connection was established
maxdata : int
Maximum amount of data in an ADB packet
Raises
------
adb_shell.exceptions.DeviceAuthError
Device authentication required, no keys available
adb_shell.exceptions.InvalidResponseError
Invalid auth response from the device
"""
async with self._transport_lock:
# 0. Close the connection and clear the store
await self._transport.close()
async with self._store_lock:
# We can release this lock because packets are only added to the store when the transport lock is held
self._packet_store.clear_all()
# 1. Use the transport to establish a connection
await self._transport.connect(adb_info.transport_timeout_s)
# 2. Send a ``b'CNXN'`` message
msg = AdbMessage(constants.CNXN, constants.VERSION, constants.MAX_ADB_DATA, b'host::%s\0' % banner)
await self._send(msg, adb_info)
# 3. Read the response from the device
cmd, arg0, maxdata, banner2 = await self._read_expected_packet_from_device([constants.AUTH, constants.CNXN], adb_info)
# 4. If ``cmd`` is not ``b'AUTH'``, then authentication is not necesary and so we are done
if cmd != constants.AUTH:
return True, maxdata
# 5. If no ``rsa_keys`` are provided, raise an exception
if not rsa_keys:
await self._transport.close()
raise exceptions.DeviceAuthError('Device authentication required, no keys available.')
# 6. Loop through our keys, signing the last ``banner2`` that we received
for rsa_key in rsa_keys:
# 6.1. If the last ``arg0`` was not :const:`adb_shell.constants.AUTH_TOKEN`, raise an exception
if arg0 != constants.AUTH_TOKEN:
await self._transport.close()
raise exceptions.InvalidResponseError('Unknown AUTH response: %s %s %s' % (arg0, maxdata, banner2))
# 6.2. Sign the last ``banner2`` and send it in an ``b'AUTH'`` message
signed_token = rsa_key.Sign(banner2)
msg = AdbMessage(constants.AUTH, constants.AUTH_SIGNATURE, 0, signed_token)
await self._send(msg, adb_info)
# 6.3. Read the response from the device
cmd, arg0, maxdata, banner2 = await self._read_expected_packet_from_device([constants.CNXN, constants.AUTH], adb_info)
# 6.4. If ``cmd`` is ``b'CNXN'``, we are done
if cmd == constants.CNXN:
return True, maxdata
# 7. None of the keys worked, so send ``rsa_keys[0]``'s public key; if the response does not time out, we must have connected successfully
pubkey = rsa_keys[0].GetPublicKey()
if not isinstance(pubkey, (bytes, bytearray)):
pubkey = bytearray(pubkey, 'utf-8')
if auth_callback is not None:
auth_callback(self)
msg = AdbMessage(constants.AUTH, constants.AUTH_RSAPUBLICKEY, 0, pubkey + b'\0')
await self._send(msg, adb_info)
adb_info.transport_timeout_s = auth_timeout_s
_, _, maxdata, _ = await self._read_expected_packet_from_device([constants.CNXN], adb_info)
return True, maxdata
[docs]
async def read(self, expected_cmds, adb_info, allow_zeros=False):
"""Read packets from the device until we get an expected packet type.
1. See if the expected packet is in the packet store
2. While the time limit has not been exceeded:
1. See if the expected packet is in the packet store
2. Read a packet from the device. If it matches what we are looking for, we are done. If it corresponds to a different stream, add it to the store.
3. Raise a timeout exception
Parameters
----------
expected_cmds : list[bytes]
We will read packets until we encounter one whose "command" field is in ``expected_cmds``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
allow_zeros : bool
Whether to allow the received ``arg0`` and ``arg1`` values to match with 0, in addition to ``adb_info.remote_id`` and ``adb_info.local_id``, respectively
Returns
-------
cmd : bytes
The received command, which is in :const:`adb_shell.constants.WIRE_TO_ID` and must be in ``expected_cmds``
arg0 : int
TODO
arg1 : int
TODO
data : bytes
The data that was read
Raises
------
adb_shell.exceptions.AdbTimeoutError
Never got one of the expected responses
"""
# First, try reading from the store. This way, you won't be waiting for the transport if it isn't needed
async with self._store_lock:
# Recall that `arg0` from the device corresponds to `adb_info.remote_id` and `arg1` from the device corresponds to `adb_info.local_id`
arg0_arg1 = self._packet_store.find(adb_info.remote_id, adb_info.local_id) if not allow_zeros else self._packet_store.find_allow_zeros(adb_info.remote_id, adb_info.local_id)
while arg0_arg1:
cmd, arg0, arg1, data = self._packet_store.get(arg0_arg1[0], arg0_arg1[1])
if cmd in expected_cmds:
return cmd, arg0, arg1, data
arg0_arg1 = self._packet_store.find(adb_info.remote_id, adb_info.local_id) if not allow_zeros else self._packet_store.find_allow_zeros(adb_info.remote_id, adb_info.local_id)
# Start the timer
start = time.time()
while True:
async with self._transport_lock:
# Try reading from the store (again) in case a packet got added while waiting to acquire the transport lock
async with self._store_lock:
# Recall that `arg0` from the device corresponds to `adb_info.remote_id` and `arg1` from the device corresponds to `adb_info.local_id`
arg0_arg1 = self._packet_store.find(adb_info.remote_id, adb_info.local_id) if not allow_zeros else self._packet_store.find_allow_zeros(adb_info.remote_id, adb_info.local_id)
while arg0_arg1:
cmd, arg0, arg1, data = self._packet_store.get(arg0_arg1[0], arg0_arg1[1])
if cmd in expected_cmds:
return cmd, arg0, arg1, data
arg0_arg1 = self._packet_store.find(adb_info.remote_id, adb_info.local_id) if not allow_zeros else self._packet_store.find_allow_zeros(adb_info.remote_id, adb_info.local_id)
# Read from the device
cmd, arg0, arg1, data = await self._read_packet_from_device(adb_info)
if not adb_info.args_match(arg0, arg1, allow_zeros):
# The packet is not a match -> put it in the store
async with self._store_lock:
self._packet_store.put(arg0, arg1, cmd, data)
else:
# The packet is a match for this `(adb_info.local_id, adb_info.remote_id)` pair
if cmd == constants.CLSE:
# Clear the entry in the store
async with self._store_lock:
self._packet_store.clear(arg0, arg1)
# If `cmd` is a match, then we are done
if cmd in expected_cmds:
return cmd, arg0, arg1, data
# Check if time is up
if time.time() - start > adb_info.read_timeout_s:
break
# Timeout
raise exceptions.AdbTimeoutError("Never got one of the expected responses: {} (transport_timeout_s = {}, read_timeout_s = {})".format(expected_cmds, adb_info.transport_timeout_s, adb_info.read_timeout_s))
[docs]
async def send(self, msg, adb_info):
"""Send a message to the device.
Parameters
----------
msg : AdbMessage
The data that will be sent
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
"""
async with self._transport_lock:
await self._send(msg, adb_info)
[docs]
async def _read_expected_packet_from_device(self, expected_cmds, adb_info):
"""Read packets from the device until we get an expected packet type.
Parameters
----------
expected_cmds : list[bytes]
We will read packets until we encounter one whose "command" field is in ``expected_cmds``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Returns
-------
cmd : bytes
The received command, which is in :const:`adb_shell.constants.WIRE_TO_ID` and must be in ``expected_cmds``
arg0 : int
TODO
arg1 : int
TODO
data : bytes
The data that was read
Raises
------
adb_shell.exceptions.AdbTimeoutError
Never got one of the expected responses
"""
start = time.time()
while True:
cmd, arg0, arg1, data = await self._read_packet_from_device(adb_info)
if cmd in expected_cmds:
return cmd, arg0, arg1, data
if time.time() - start > adb_info.read_timeout_s:
# Timeout
raise exceptions.AdbTimeoutError("Never got one of the expected responses: {} (transport_timeout_s = {}, read_timeout_s = {})".format(expected_cmds, adb_info.transport_timeout_s, adb_info.read_timeout_s))
[docs]
async def _read_bytes_from_device(self, length, adb_info):
"""Read ``length`` bytes from the device.
Parameters
----------
length : int
We will read packets until we get this length of data
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Returns
-------
bytes
The data that was read
Raises
------
adb_shell.exceptions.AdbTimeoutError
Did not read ``length`` bytes in time
"""
start = time.time()
data = bytearray()
while length > 0:
temp = await self._transport.bulk_read(length, adb_info.transport_timeout_s)
if temp:
# Only log if `temp` is not empty
_LOGGER.debug("bulk_read(%d): %.1000r", length, temp)
data += temp
length -= len(temp)
if length == 0:
break
if time.time() - start > adb_info.read_timeout_s:
# Timeout
raise exceptions.AdbTimeoutError("Timeout: read {} of {} bytes (transport_timeout_s = {}, read_timeout_s = {})".format(len(data), len(data) + length, adb_info.transport_timeout_s, adb_info.read_timeout_s))
return bytes(data)
[docs]
async def _read_packet_from_device(self, adb_info):
"""Read a complete ADB packet (header + data) from the device.
Parameters
----------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Returns
-------
cmd : bytes
The received command, which is in :const:`adb_shell.constants.WIRE_TO_ID` and must be in ``expected_cmds``
arg0 : int
TODO
arg1 : int
TODO
bytes
The data that was read
Raises
------
adb_shell.exceptions.InvalidCommandError
Unknown command
adb_shell.exceptions.InvalidChecksumError
Received checksum does not match the expected checksum
"""
msg = await self._read_bytes_from_device(constants.MESSAGE_SIZE, adb_info)
cmd, arg0, arg1, data_length, data_checksum = unpack(msg)
command = constants.WIRE_TO_ID.get(cmd)
if not command:
raise exceptions.InvalidCommandError("Unknown command: %d = '%s' (arg0 = %d, arg1 = %d, msg = '%s')" % (cmd, int_to_cmd(cmd), arg0, arg1, msg))
if data_length == 0:
return command, arg0, arg1, b""
data = await self._read_bytes_from_device(data_length, adb_info)
actual_checksum = checksum(data)
if actual_checksum != data_checksum:
raise exceptions.InvalidChecksumError("Received checksum {} != {}".format(actual_checksum, data_checksum))
return command, arg0, arg1, data
[docs]
async def _send(self, msg, adb_info):
"""Send a message to the device.
1. Send the message header (:meth:`adb_shell.adb_message.AdbMessage.pack <AdbMessage.pack>`)
2. Send the message data
Parameters
----------
msg : AdbMessage
The data that will be sent
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
"""
packed = msg.pack()
_LOGGER.debug("bulk_write(%d): %r", len(packed), packed)
await self._transport.bulk_write(packed, adb_info.transport_timeout_s)
if msg.data:
_LOGGER.debug("bulk_write(%d): %r", len(msg.data), msg.data)
await self._transport.bulk_write(msg.data, adb_info.transport_timeout_s)
[docs]
class AdbDeviceAsync(object):
"""A class with methods for connecting to a device and executing ADB commands.
Parameters
----------
transport : BaseTransportAsync
A user-provided transport for communicating with the device; must be an instance of a subclass of :class:`~adb_shell.transport.base_transport_async.BaseTransportAsync`
default_transport_timeout_s : float, None
Default timeout in seconds for transport packets, or ``None``
banner : str, bytes, None
The hostname of the machine where the Python interpreter is currently running; if
it is not provided, it will be determined via ``socket.gethostname()``
Raises
------
adb_shell.exceptions.InvalidTransportError
The passed ``transport`` is not an instance of a subclass of :class:`~adb_shell.transport.base_transport_async.BaseTransportAsync`
Attributes
----------
_available : bool
Whether an ADB connection to the device has been established
_banner : bytearray, bytes
The hostname of the machine where the Python interpreter is currently running
_default_transport_timeout_s : float, None
Default timeout in seconds for transport packets, or ``None``
_io_manager : _AdbIOManagerAsync
Used for handling all ADB I/O
_local_id : int
The local ID that is used for ADB transactions; the value is incremented each time and is always in the range ``[1, 2^32)``
_local_id_lock : Lock
A lock for protecting ``_local_id``; this is never held for long
_maxdata: int
Maximum amount of data in an ADB packet
_transport : BaseTransportAsync
The transport that is used to connect to the device; must be a subclass of :class:`~adb_shell.transport.base_transport_async.BaseTransportAsync`
"""
def __init__(self, transport, default_transport_timeout_s=None, banner=None):
if banner and not isinstance(banner, (bytes, bytearray)):
self._banner = bytearray(banner, 'utf-8')
else:
self._banner = banner
if not isinstance(transport, BaseTransportAsync):
raise exceptions.InvalidTransportError("`transport` must be an instance of a subclass of `BaseTransportAsync`")
self._io_manager = _AdbIOManagerAsync(transport)
self._available = False
self._default_transport_timeout_s = default_transport_timeout_s
self._local_id = 0
self._local_id_lock = Lock()
self._maxdata = constants.MAX_PUSH_DATA
# ======================================================================= #
# #
# Properties & simple methods #
# #
# ======================================================================= #
@property
def available(self):
"""Whether or not an ADB connection to the device has been established.
Returns
-------
bool
``self._available``
"""
return self._available
@property
def max_chunk_size(self):
"""Maximum chunk size for filesync operations
Returns
-------
int
Minimum value based on :const:`adb_shell.constants.MAX_CHUNK_SIZE` and ``_max_data / 2``, fallback to legacy :const:`adb_shell.constants.MAX_PUSH_DATA`
"""
return min(constants.MAX_CHUNK_SIZE, self._maxdata // 2) or constants.MAX_PUSH_DATA
[docs]
def _get_transport_timeout_s(self, transport_timeout_s):
"""Use the provided ``transport_timeout_s`` if it is not ``None``; otherwise, use ``self._default_transport_timeout_s``
Parameters
----------
transport_timeout_s : float, None
The potential transport timeout
Returns
-------
float
``transport_timeout_s`` if it is not ``None``; otherwise, ``self._default_transport_timeout_s``
"""
return transport_timeout_s if transport_timeout_s is not None else self._default_transport_timeout_s
# ======================================================================= #
# #
# Close & Connect #
# #
# ======================================================================= #
[docs]
async def close(self):
"""Close the connection via the provided transport's ``close()`` method.
"""
self._available = False
await self._io_manager.close()
[docs]
async def connect(self, rsa_keys=None, transport_timeout_s=None, auth_timeout_s=constants.DEFAULT_AUTH_TIMEOUT_S, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, auth_callback=None):
"""Establish an ADB connection to the device.
See :meth:`_AdbIOManagerAsync.connect`.
Parameters
----------
rsa_keys : list, None
A list of signers of type :class:`~adb_shell.auth.sign_cryptography.CryptographySigner`,
:class:`~adb_shell.auth.sign_pycryptodome.PycryptodomeAuthSigner`, or :class:`~adb_shell.auth.sign_pythonrsa.PythonRSASigner`
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
auth_timeout_s : float, None
The time in seconds to wait for a ``b'CNXN'`` authentication response
read_timeout_s : float
The total time in seconds to wait for expected commands in :meth:`_AdbIOManagerAsync._read_expected_packet_from_device`
auth_callback : function, None
Function callback invoked when the connection needs to be accepted on the device
Returns
-------
bool
Whether the connection was established (:attr:`AdbDeviceAsync.available`)
"""
# Get `self._banner` if it was not provided in the constructor
if not self._banner:
self._banner = await get_running_loop().run_in_executor(None, get_banner)
# Instantiate the `_AdbTransactionInfo`
adb_info = _AdbTransactionInfo(None, None, self._get_transport_timeout_s(transport_timeout_s), read_timeout_s, None)
# Mark the device as unavailable
self._available = False
# Use the IO manager to connect
self._available, self._maxdata = await self._io_manager.connect(self._banner, rsa_keys, auth_timeout_s, auth_callback, adb_info)
return self._available
# ======================================================================= #
# #
# Services #
# #
# ======================================================================= #
[docs]
async def _service(self, service, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
"""Send an ADB command to the device.
Parameters
----------
service : bytes
The ADB service to talk to (e.g., ``b'shell'``)
command : bytes
The command that will be sent
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
decode : bool
Whether to decode the output to utf8 before returning
Returns
-------
bytes, str
The output of the ADB command as a string if ``decode`` is True, otherwise as bytes.
"""
if decode:
return b''.join([x async for x in self._streaming_command(service, command, transport_timeout_s, read_timeout_s, timeout_s)]).decode('utf8', 'backslashreplace')
return b''.join([x async for x in self._streaming_command(service, command, transport_timeout_s, read_timeout_s, timeout_s)])
[docs]
async def _streaming_service(self, service, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, decode=True):
"""Send an ADB command to the device, yielding each line of output.
Parameters
----------
service : bytes
The ADB service to talk to (e.g., ``b'shell'``)
command : bytes
The command that will be sent
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
decode : bool
Whether to decode the output to utf8 before returning
Yields
-------
bytes, str
The line-by-line output of the ADB command as a string if ``decode`` is True, otherwise as bytes.
"""
stream = self._streaming_command(service, command, transport_timeout_s, read_timeout_s, None)
if decode:
async for line in (stream_line.decode('utf8', 'backslashreplace') async for stream_line in stream):
yield line
else:
async for line in stream:
yield line
[docs]
async def exec_out(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
"""Send an ADB ``exec-out`` command to the device.
https://www.linux-magazine.com/Issues/2017/195/Ask-Klaus
Parameters
----------
command : str
The exec-out command that will be sent
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
decode : bool
Whether to decode the output to utf8 before returning
Returns
-------
bytes, str
The output of the ADB exec-out command as a string if ``decode`` is True, otherwise as bytes.
"""
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
return await self._service(b'exec', command.encode('utf8'), transport_timeout_s, read_timeout_s, timeout_s, decode)
[docs]
async def reboot(self, fastboot=False, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None):
"""Reboot the device.
Parameters
----------
fastboot : bool
Whether to reboot the device into fastboot
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManager.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
"""
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
await self._open(b'reboot:bootloader' if fastboot else b'reboot:', transport_timeout_s, read_timeout_s, timeout_s)
[docs]
async def root(self, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None):
"""Gain root access.
The device must be rooted in order for this to work.
Parameters
----------
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
"""
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
await self._service(b'root', b'', transport_timeout_s, read_timeout_s, timeout_s, False)
[docs]
async def shell(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, timeout_s=None, decode=True):
"""Send an ADB shell command to the device.
Parameters
----------
command : str
The shell command that will be sent
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
decode : bool
Whether to decode the output to utf8 before returning
Returns
-------
bytes, str
The output of the ADB shell command as a string if ``decode`` is True, otherwise as bytes.
"""
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
return await self._service(b'shell', command.encode('utf8'), transport_timeout_s, read_timeout_s, timeout_s, decode)
[docs]
async def streaming_shell(self, command, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S, decode=True):
"""Send an ADB shell command to the device, yielding each line of output.
Parameters
----------
command : str
The shell command that will be sent
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
decode : bool
Whether to decode the output to utf8 before returning
Yields
-------
bytes, str
The line-by-line output of the ADB shell command as a string if ``decode`` is True, otherwise as bytes.
"""
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
async for line in self._streaming_service(b'shell', command.encode('utf8'), transport_timeout_s, read_timeout_s, decode):
yield line
# ======================================================================= #
# #
# FileSync #
# #
# ======================================================================= #
[docs]
async def list(self, device_path, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
"""Return a directory listing of the given path.
Parameters
----------
device_path : str
Directory to list.
transport_timeout_s : float, None
Expected timeout for any part of the pull.
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
Returns
-------
files : list[DeviceFile]
Filename, mode, size, and mtime info for the files in the directory
"""
if not device_path:
raise exceptions.DevicePathInvalidError("Cannot list an empty device path")
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
adb_info = await self._open(b'sync:', transport_timeout_s, read_timeout_s, None)
filesync_info = _FileSyncTransactionInfo(constants.FILESYNC_LIST_FORMAT, maxdata=self._maxdata)
await self._filesync_send(constants.LIST, adb_info, filesync_info, data=device_path)
files = []
async for cmd_id, header, filename in self._filesync_read_until([constants.DENT], [constants.DONE], adb_info, filesync_info):
if cmd_id == constants.DONE:
break
mode, size, mtime = header
files.append(DeviceFile(filename, mode, size, mtime))
await self._clse(adb_info)
return files
[docs]
async def pull(self, device_path, local_path, progress_callback=None, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
"""Pull a file from the device.
Parameters
----------
device_path : str
The file on the device that will be pulled
local_path : str, BytesIO
The path or BytesIO stream where the file will be downloaded
progress_callback : function, None
Callback method that accepts ``device_path``, ``bytes_written``, and ``total_bytes``
transport_timeout_s : float, None
Expected timeout for any part of the pull.
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
"""
if not device_path:
raise exceptions.DevicePathInvalidError("Cannot pull from an empty device path")
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
opener = _open_bytesio if isinstance(local_path, BytesIO) else aiofiles.open
async with opener(local_path, 'wb') as stream:
adb_info = await self._open(b'sync:', transport_timeout_s, read_timeout_s, None)
filesync_info = _FileSyncTransactionInfo(constants.FILESYNC_PULL_FORMAT, maxdata=self._maxdata)
try:
await self._pull(device_path, stream, progress_callback, adb_info, filesync_info)
finally:
await self._clse(adb_info)
[docs]
async def _pull(self, device_path, stream, progress_callback, adb_info, filesync_info):
"""Pull a file from the device into the file-like ``local_path``.
Parameters
----------
device_path : str
The file on the device that will be pulled
stream : AsyncBufferedIOBase, _AsyncBytesIO
File-like object for writing to
progress_callback : function, None
Callback method that accepts ``device_path``, ``bytes_written``, and ``total_bytes``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
"""
if progress_callback:
total_bytes = (await self.stat(device_path))[1]
await self._filesync_send(constants.RECV, adb_info, filesync_info, data=device_path)
async for cmd_id, _, data in self._filesync_read_until([constants.DATA], [constants.DONE], adb_info, filesync_info):
if cmd_id == constants.DONE:
break
await stream.write(data)
if progress_callback:
try:
await progress_callback(device_path, len(data), total_bytes)
except: # noqa pylint: disable=bare-except
pass
[docs]
async def push(self, local_path, device_path, st_mode=constants.DEFAULT_PUSH_MODE, mtime=0, progress_callback=None, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
"""Push a file or directory to the device.
Parameters
----------
local_path : str, BytesIO
A filename, directory, or BytesIO stream to push to the device
device_path : str
Destination on the device to write to
st_mode : int
Stat mode for ``local_path``
mtime : int
Modification time to set on the file
progress_callback : function, None
Callback method that accepts ``device_path``, ``bytes_written``, and ``total_bytes``
transport_timeout_s : float, None
Expected timeout for any part of the push
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
"""
if not device_path:
raise exceptions.DevicePathInvalidError("Cannot push to an empty device path")
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
local_path_is_dir, local_paths, device_paths = await get_running_loop().run_in_executor(None, get_files_to_push, local_path, device_path)
if local_path_is_dir:
await self.shell("mkdir " + device_path, transport_timeout_s, read_timeout_s)
for _local_path, _device_path in zip(local_paths, device_paths):
opener = _open_bytesio if isinstance(local_path, BytesIO) else aiofiles.open
async with opener(_local_path, 'rb') as stream:
adb_info = await self._open(b'sync:', transport_timeout_s, read_timeout_s, None)
filesync_info = _FileSyncTransactionInfo(constants.FILESYNC_PUSH_FORMAT, maxdata=self._maxdata)
await self._push(stream, _device_path, st_mode, mtime, progress_callback, adb_info, filesync_info)
await self._clse(adb_info)
[docs]
async def _push(self, stream, device_path, st_mode, mtime, progress_callback, adb_info, filesync_info):
"""Push a file-like object to the device.
Parameters
----------
stream : AsyncBufferedReader, _AsyncBytesIO
File-like object for reading from
device_path : str
Destination on the device to write to
st_mode : int
Stat mode for the file
mtime : int
Modification time
progress_callback : function, None
Callback method that accepts ``device_path``, ``bytes_written``, and ``total_bytes``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Raises
------
PushFailedError
Raised on push failure.
"""
fileinfo = ('{},{}'.format(device_path, int(st_mode))).encode('utf-8')
await self._filesync_send(constants.SEND, adb_info, filesync_info, data=fileinfo)
if progress_callback:
total_bytes = (await get_running_loop().run_in_executor(None, os.fstat, stream.fileno())).st_size
while True:
data = await stream.read(self.max_chunk_size)
if data:
await self._filesync_send(constants.DATA, adb_info, filesync_info, data=data)
if progress_callback:
try:
await progress_callback(device_path, len(data), total_bytes)
except: # noqa pylint: disable=bare-except
pass
else:
break
if mtime == 0:
mtime = int(time.time())
# DONE doesn't send data, but it hides the last bit of data in the size field.
await self._filesync_send(constants.DONE, adb_info, filesync_info, size=mtime)
async for cmd_id, _, data in self._filesync_read_until([], [constants.OKAY, constants.FAIL], adb_info, filesync_info):
if cmd_id == constants.OKAY:
return
raise exceptions.PushFailedError(data)
[docs]
async def stat(self, device_path, transport_timeout_s=None, read_timeout_s=constants.DEFAULT_READ_TIMEOUT_S):
"""Get a file's ``stat()`` information.
Parameters
----------
device_path : str
The file on the device for which we will get information.
transport_timeout_s : float, None
Expected timeout for any part of the pull.
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
Returns
-------
mode : int
The octal permissions for the file
size : int
The size of the file
mtime : int
The last modified time for the file
"""
if not device_path:
raise exceptions.DevicePathInvalidError("Cannot stat an empty device path")
if not self.available:
raise exceptions.AdbConnectionError("ADB command not sent because a connection to the device has not been established. (Did you call `AdbDeviceAsync.connect()`?)")
adb_info = await self._open(b'sync:', transport_timeout_s, read_timeout_s, None)
filesync_info = _FileSyncTransactionInfo(constants.FILESYNC_STAT_FORMAT, maxdata=self._maxdata)
await self._filesync_send(constants.STAT, adb_info, filesync_info, data=device_path)
_, (mode, size, mtime), _ = await self._filesync_read([constants.STAT], adb_info, filesync_info)
await self._clse(adb_info)
return mode, size, mtime
# ======================================================================= #
# #
# Hidden Methods: send packets #
# #
# ======================================================================= #
[docs]
async def _clse(self, adb_info):
"""Send a ``b'CLSE'`` message and then read a ``b'CLSE'`` message.
.. warning::
This is not to be confused with the :meth:`AdbDeviceAsync.close` method!
Parameters
----------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
"""
msg = AdbMessage(constants.CLSE, adb_info.local_id, adb_info.remote_id)
await self._io_manager.send(msg, adb_info)
await self._read_until([constants.CLSE], adb_info)
[docs]
async def _okay(self, adb_info):
"""Send an ``b'OKAY'`` mesage.
Parameters
----------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
"""
msg = AdbMessage(constants.OKAY, adb_info.local_id, adb_info.remote_id)
await self._io_manager.send(msg, adb_info)
# ======================================================================= #
# #
# Hidden Methods #
# #
# ======================================================================= #
[docs]
async def _open(self, destination, transport_timeout_s, read_timeout_s, timeout_s):
"""Opens a new connection to the device via an ``b'OPEN'`` message.
1. :meth:`~_AdbIOManagerAsync.send` an ``b'OPEN'`` command to the device that specifies the ``local_id``
2. :meth:`~_AdbIOManagerAsync.read` the response from the device and fill in the ``adb_info.remote_id`` attribute
Parameters
----------
destination : bytes
``b'SERVICE:COMMAND'``
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
Returns
-------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
"""
async with self._local_id_lock:
self._local_id += 1
if self._local_id == 2**32:
self._local_id = 1
adb_info = _AdbTransactionInfo(self._local_id, None, self._get_transport_timeout_s(transport_timeout_s), read_timeout_s, timeout_s)
msg = AdbMessage(constants.OPEN, adb_info.local_id, 0, destination + b'\0')
await self._io_manager.send(msg, adb_info)
_, adb_info.remote_id, _, _ = await self._io_manager.read([constants.OKAY], adb_info)
return adb_info
[docs]
async def _read_until(self, expected_cmds, adb_info):
"""Read a packet, acknowledging any write packets.
1. Read data via :meth:`_AdbIOManagerAsync.read`
2. If a ``b'WRTE'`` packet is received, send an ``b'OKAY'`` packet via :meth:`AdbDeviceAsync._okay`
3. Return the ``cmd`` and ``data`` that were read by :meth:`_AdbIOManagerAsync.read`
Parameters
----------
expected_cmds : list[bytes]
:meth:`_AdbIOManagerAsync.read` will look for a packet whose command is in ``expected_cmds``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Returns
-------
cmd : bytes
The command that was received by :meth:`_AdbIOManagerAsync.read`, which is in :const:`adb_shell.constants.WIRE_TO_ID` and must be in ``expected_cmds``
data : bytes
The data that was received by :meth:`_AdbIOManagerAsync.read`
"""
cmd, _, _, data = await self._io_manager.read(expected_cmds, adb_info, allow_zeros=True)
# Acknowledge write packets
if cmd == constants.WRTE:
await self._okay(adb_info)
return cmd, data
[docs]
async def _read_until_close(self, adb_info):
"""Yield packets until a ``b'CLSE'`` packet is received.
1. Read the ``cmd`` and ``data`` fields from a ``b'CLSE'`` or ``b'WRTE'`` packet via :meth:`AdbDeviceAsync._read_until`
2. If ``cmd`` is ``b'CLSE'``, then send a ``b'CLSE'`` message and stop
3. Yield ``data`` and repeat
Parameters
----------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
Yields
------
data : bytes
The data that was read by :meth:`AdbDeviceAsync._read_until`
"""
start = time.time()
while True:
cmd, data = await self._read_until([constants.CLSE, constants.WRTE], adb_info)
if cmd == constants.CLSE:
msg = AdbMessage(constants.CLSE, adb_info.local_id, adb_info.remote_id)
await self._io_manager.send(msg, adb_info)
break
yield data
# Make sure the ADB command has not timed out
if adb_info.timeout_s is not None and time.time() - start > adb_info.timeout_s:
raise exceptions.AdbTimeoutError("The command did not complete within {} seconds".format(adb_info.timeout_s))
[docs]
async def _streaming_command(self, service, command, transport_timeout_s, read_timeout_s, timeout_s):
"""One complete set of packets for a single command.
1. :meth:`~AdbDeviceAsync._open` a new connection to the device, where the ``destination`` parameter is ``service:command``
2. Read the response data via :meth:`AdbDeviceAsync._read_until_close`
.. note::
All the data is held in memory, and thus large responses will be slow and can fill up memory.
Parameters
----------
service : bytes
The ADB service (e.g., ``b'shell'``, as used by :meth:`AdbDeviceAsync.shell`)
command : bytes
The service command
transport_timeout_s : float, None
Timeout in seconds for sending and receiving data, or ``None``; see :meth:`BaseTransportAsync.bulk_read() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_read>`
and :meth:`BaseTransportAsync.bulk_write() <adb_shell.transport.base_transport_async.BaseTransportAsync.bulk_write>`
read_timeout_s : float
The total time in seconds to wait for a ``b'CLSE'`` or ``b'OKAY'`` command in :meth:`_AdbIOManagerAsync.read`
timeout_s : float, None
The total time in seconds to wait for the ADB command to finish
Yields
------
bytes
The responses from the service.
"""
adb_info = await self._open(b'%s:%s' % (service, command), transport_timeout_s, read_timeout_s, timeout_s)
async for data in self._read_until_close(adb_info):
yield data
# ======================================================================= #
# #
# FileSync Hidden Methods #
# #
# ======================================================================= #
[docs]
async def _filesync_flush(self, adb_info, filesync_info):
"""Write the data in the buffer up to ``filesync_info.send_idx``, then set ``filesync_info.send_idx`` to 0.
Parameters
----------
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
"""
# Send the buffer
msg = AdbMessage(constants.WRTE, adb_info.local_id, adb_info.remote_id, filesync_info.send_buffer[:filesync_info.send_idx])
await self._io_manager.send(msg, adb_info)
# Expect an 'OKAY' in response
await self._read_until([constants.OKAY], adb_info)
# Reset the send index
filesync_info.send_idx = 0
[docs]
async def _filesync_read(self, expected_ids, adb_info, filesync_info):
"""Read ADB messages and return FileSync packets.
Parameters
----------
expected_ids : tuple[bytes]
If the received header ID is not in ``expected_ids``, an exception will be raised
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
Returns
-------
command_id : bytes
The received header ID
tuple
The contents of the header
data : bytearray, None
The received data, or ``None`` if the command ID is :const:`adb_shell.constants.STAT`
Raises
------
adb_shell.exceptions.AdbCommandFailureException
Command failed
adb_shell.exceptions.InvalidResponseError
Received response was not in ``expected_ids``
"""
if filesync_info.send_idx:
await self._filesync_flush(adb_info, filesync_info)
# Read one filesync packet off the recv buffer.
header_data = await self._filesync_read_buffered(filesync_info.recv_message_size, adb_info, filesync_info)
header = struct.unpack(filesync_info.recv_message_format, header_data)
# Header is (ID, ...).
command_id = constants.FILESYNC_WIRE_TO_ID[header[0]]
# Whether there is data to read
read_data = command_id != constants.STAT
if read_data:
# Header is (ID, ..., size) --> read the data
data = await self._filesync_read_buffered(header[-1], adb_info, filesync_info)
else:
# No data to be read
data = bytearray()
if command_id not in expected_ids:
if command_id == constants.FAIL:
reason = data.decode('utf-8', errors='backslashreplace')
raise exceptions.AdbCommandFailureException('Command failed: {}'.format(reason))
raise exceptions.InvalidResponseError('Expected one of %s, got %s' % (expected_ids, command_id))
if not read_data:
return command_id, header[1:], None
return command_id, header[1:-1], data
[docs]
async def _filesync_read_buffered(self, size, adb_info, filesync_info):
"""Read ``size`` bytes of data from ``self.recv_buffer``.
Parameters
----------
size : int
The amount of data to read
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
Returns
-------
result : bytearray
The read data
"""
# Ensure recv buffer has enough data.
while len(filesync_info.recv_buffer) < size:
_, data = await self._read_until([constants.WRTE], adb_info)
filesync_info.recv_buffer += data
result = filesync_info.recv_buffer[:size]
filesync_info.recv_buffer = filesync_info.recv_buffer[size:]
return result
[docs]
async def _filesync_read_until(self, expected_ids, finish_ids, adb_info, filesync_info):
"""Useful wrapper around :meth:`AdbDeviceAsync._filesync_read`.
Parameters
----------
expected_ids : tuple[bytes]
If the received header ID is not in ``expected_ids``, an exception will be raised
finish_ids : tuple[bytes]
We will read until we find a header ID that is in ``finish_ids``
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
Yields
------
cmd_id : bytes
The received header ID
header : tuple
TODO
data : bytearray
The received data
"""
while True:
cmd_id, header, data = await self._filesync_read(expected_ids + finish_ids, adb_info, filesync_info)
yield cmd_id, header, data
# These lines are not reachable because whenever this method is called and `cmd_id` is in `finish_ids`, the code
# either breaks (`list` and `_pull`), returns (`_push`), or raises an exception (`_push`)
if cmd_id in finish_ids: # pragma: no cover
break
[docs]
async def _filesync_send(self, command_id, adb_info, filesync_info, data=b'', size=None):
"""Send/buffer FileSync packets.
Packets are buffered and only flushed when this connection is read from. All
messages have a response from the device, so this will always get flushed.
Parameters
----------
command_id : bytes
Command to send.
adb_info : _AdbTransactionInfo
Info and settings for this ADB transaction
filesync_info : _FileSyncTransactionInfo
Data and storage for this FileSync transaction
data : str, bytes
Optional data to send, must set data or size.
size : int, None
Optionally override size from len(data).
"""
if not isinstance(data, bytes):
data = data.encode('utf8')
if size is None:
size = len(data)
if not filesync_info.can_add_to_send_buffer(len(data)):
await self._filesync_flush(adb_info, filesync_info)
buf = struct.pack(b'<2I', constants.FILESYNC_ID_TO_WIRE[command_id], size) + data
filesync_info.send_buffer[filesync_info.send_idx:filesync_info.send_idx + len(buf)] = buf
filesync_info.send_idx += len(buf)
[docs]
class AdbDeviceTcpAsync(AdbDeviceAsync):
"""A class with methods for connecting to a device via TCP and executing ADB commands.
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)
default_transport_timeout_s : float, None
Default timeout in seconds for TCP packets, or ``None``
banner : str, bytes, None
The hostname of the machine where the Python interpreter is currently running; if
it is not provided, it will be determined via ``socket.gethostname()``
Attributes
----------
_available : bool
Whether an ADB connection to the device has been established
_banner : bytearray, bytes
The hostname of the machine where the Python interpreter is currently running
_default_transport_timeout_s : float, None
Default timeout in seconds for TCP packets, or ``None``
_local_id : int
The local ID that is used for ADB transactions; the value is incremented each time and is always in the range ``[1, 2^32)``
_maxdata : int
Maximum amount of data in an ADB packet
_transport : TcpTransportAsync
The transport that is used to connect to the device
"""
def __init__(self, host, port=5555, default_transport_timeout_s=None, banner=None):
transport = TcpTransportAsync(host, port)
super(AdbDeviceTcpAsync, self).__init__(transport, default_transport_timeout_s, banner)