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

"""
Native adb client-to-server protocol implementation.

This is an async implementation, suitable for including in an event loop.
"""

import os
import sys
import stat
import errno
import struct
import signal
import typing

from devtest import logging
from devtest import timers
from devtest import ringbuffer
from devtest.io import socket
from devtest.io import streams
from devtest.os import process
from devtest.os import procutils
from devtest.os import exitstatus
from devtest.io.reactor import (get_kernel, spawn, block_in_thread, sleep,
                                SignalEvent, timeout_after, TaskTimeout)

from . import logcat


ADB = procutils.which("adb")
if ADB is None:
    raise ImportError("The adb program was not found in PATH. "
                      "This module will not work.")

ADB_PORT = 5037

MAX_PAYLOAD_V2 = 256 * 1024
MAX_PAYLOAD = MAX_PAYLOAD_V2


class Error(Exception):
    pass


class AdbProtocolError(Error):
    """An error in the protocol was detected."""


class AdbCommandFail(Error):
    """An error indicated by the server."""


def _run_adb(adbcommand):
    """Run the adb binary."""
    cmd = [ADB]
    cmd.extend(adbcommand.split())
    return process.check_output(cmd)


def start_server(port=ADB_PORT):
    return _run_adb("-P {} start-server".format(port))


def kill_server(port=ADB_PORT):
    return _run_adb("-P {} kill-server".format(port))


class AdbConnection:
    """Asynchronous adb socket wrapper."""

    def __init__(self, host="localhost", port=ADB_PORT):
        self.host = host
        self.port = port
        self.socket = None

    async def close(self):
        if self.socket is not None:
            s = self.socket
            self.socket = None
            await s.close()

    def isopen(self):
        return bool(self.socket)

    async def open(self):
        sock = socket.socket(family=socket.AF_INET,
                             type=socket.SOCK_STREAM)
        errno = await sock.connect_ex((self.host, self.port))
        if errno != 0:
            raise IOError(errno, "Could not connect to adb server")
        self.socket = sock

    async def message(self, msg, expect_response=True):
        msg = b"%04x%b" % (len(msg), msg)
        await self.socket.sendall(msg)
        stat = await self.socket.recv(4)
        if stat == b"OKAY":
            if expect_response:
                lenst = await self.socket.recv(4)
                length = int(lenst, 16)
                if length:
                    return await self.socket.recv(length)
                else:
                    return b""
        elif stat == b"FAIL":
            lenst = await self.socket.recv(4)
            length = int(lenst, 16)
            resp = await self.socket.recv(length)
            raise AdbCommandFail("message FAIL: {}".format(resp.decode("ascii")))
        else:
            raise AdbProtocolError(stat.decode("ascii"))

    async def read_protocol_string(self):
        lenst = await self.socket.recv(4)
        length = int(lenst, 16)
        if length:
            return await self.socket.recv(length)
        else:
            return b""


async def _transact(conn, msg):
    await conn.open()
    resp = await conn.message(msg)
    await conn.close()
    return resp


# On the host: 1st OKAY is connect, 2nd OKAY is status.
async def _command_transact(conn, msg):
    await conn.open()
    msg = b"%04x%b" % (len(msg), msg)
    await conn.socket.sendall(msg)
    connstat = await conn.socket.recv(4)
    if connstat == b"OKAY":
        stat = await conn.socket.recv(4)
        if stat != b"OKAY":
            raise AdbCommandFail("command not OKAY")
    elif connstat == b"FAIL":
        lenst = await conn.socket.recv(4)
        length = int(lenst, 16)
        resp = await conn.socket.recv(length)
        await conn.close()
        raise AdbCommandFail(resp.decode("ascii"))
    else:
        await conn.close()
        raise AdbProtocolError(stat.decode("ascii"))


class AdbClient:
    """An adb client, synchronous.

    For general host side operations.
    """
    def __init__(self, host="localhost", port=ADB_PORT):
        self._conn = AdbConnection(host=host, port=port)

    def _message(self, msg):
        if self._conn is None:
            raise Error("Operation on a closed client.")
        return get_kernel().run(_transact(self._conn, msg))

    def close(self):
        if self._conn is not None:
            get_kernel().run(self._conn.close())
            self._conn = None

    def get_device(self, serial):
        """Get a AndroidDeviceClient instance.
        """
        return AndroidDeviceClient(
            serial, host=self._conn.host, port=self._conn.port)

    def get_state(self, serial):
        """Get the current state of a device.

        Arguments:
            serial: str of the device serial number.

        Returns:
            one of {"device", "unauthorized", "bootloader"}
            or None if serial not found.
        """
        for dev in self.get_device_list():
            if dev.serial == serial:
                return dev.state

    def get_device_list(self):
        """Get list of attached devices.

        Returns:
            list of AndroidDevice instances.
        """
        dl = []
        resp = self._message(b"host:devices-l")
        for line in resp.splitlines():
            dl.append(_device_factory(line))
        return dl

    def forward(self, serial, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        msg = b"host-serial:%b:forward:tcp:%d;tcp:%d" % (serial.encode("ascii"),
                                                         hostport, devport)
        get_kernel().run(_command_transact(self._conn, msg))

    def list_forward(self):
        """Return a list of currently forwarded ports.

        Returns:
            Tuple of (serial, host_port, device_port).
        """
        resp = self._message(b"host:list-forward")
        fl = []
        for line in resp.splitlines():
            # <serial> " " <local> " " <remote> "\n"
            serno, host_port, remote_port = line.split()
            fl.append((serno.decode("ascii"),
                       int(host_port.split(b":")[1]),
                       int(remote_port.split(b":")[1])))
        return fl

    def reconnect_offline(self):
        self._message(b"host:reconnect-offline")

    def reconnect(self, serial):
        self._message(b"host-serial:%b:reconnect" % (serial.encode("ascii"),))
        while self.get_state(serial) is None:
            timers.nanosleep(1.1)

    @property
    def server_version(self):
        """The server's version number."""
        resp = self._message(b"host:version")
        return int(resp, 16)


def getAsyncAndroidDeviceClient(serial, host="localhost", port=ADB_PORT):
    """Get initialized _AsyncAndroidDeviceClient object from sync code.
    """
    return get_kernel().run(AsyncAndroidDeviceClient(serial, host, port))


async def AsyncAndroidDeviceClient(serial, host="localhost", port=ADB_PORT):
    """Get initialized _AsyncAndroidDeviceClient instance from async code."""
    ac = _AsyncAndroidDeviceClient(serial, host, port)
    await ac._init()
    return ac


class _AsyncAndroidDeviceClient:
    """An active adb client per device.

    For device specific operations.
    For use in asynchronous event loops.
    """
    def __init__(self, serial, host="localhost", port=ADB_PORT):
        self.serial = serial.encode("ascii")
        self._conn = AdbConnection(host=host, port=port)

    async def _init(self):
        await self._conn.open()
        await self.get_features()

    async def _message(self, msg):
        return await _transact(self._conn, msg)

    async def get_features(self):
        msg = b"host-serial:%b:features" % (self.serial,)
        resp = await self._message(msg)
        self.features = resp.decode("ascii")
        return self.features

    async def close(self):
        await self._conn.close()

    async def open(self):
        await self._conn.open()

    async def get_state(self):
        msg = b"host-serial:%b:get-state" % (self.serial,)
        resp = await self._message(msg)
        return resp.decode("ascii")

    async def reboot(self):
        await _connect_command(self.serial, self._conn, b"reboot:")

    async def remount(self):
        resp = await _special_command(self.serial, self._conn, b"remount:")
        return resp.decode("ascii")

    async def root(self):
        resp = await _special_command(self.serial, self._conn, b"root:")
        await sleep(3)  # adbd needs some time to initialize
        return resp.decode("ascii")

    async def unroot(self):
        resp = await _special_command(self.serial, self._conn, b"unroot:")
        await sleep(3)  # adbd needs some time to initialize
        return resp.decode("ascii")

    async def forward(self, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        msg = b"host-serial:%b:forward:tcp:%d;tcp:%d" % (self.serial,
                                                         hostport, devport)
        await _command_transact(self._conn, msg)

    async def kill_forward(self, hostport):
        """Tell server to remove forwarding TCP ports.
        """
        msg = b"host-serial:%b:killforward:tcp:%d" % (self.serial, hostport)
        await _command_transact(self._conn, msg)

    async def kill_forward_all(self):
        """Tell server to remove all forwarding TCP ports.
        """
        msg = b"host-serial:%b:killforward-all" % (self.serial,)
        await _command_transact(self._conn, msg)

    async def list_forward(self):
        """Return a list of currently forwarded ports.

        Returns:
            Tuple of (host_port, device_port).
        """
        msg = b"host-serial:%b:list-forward" % (self.serial,)
        resp = await self._message(msg)
        fl = []
        for line in resp.splitlines():
            # <serial> " " <local> " " <remote> "\n"
            serno, host_port, remote_port = line.split()
            if serno == self.serial:
                fl.append((int(host_port.split(b":")[1]),
                           int(remote_port.split(b":")[1])))
        return fl

    async def wait_for(self, state: str):
        """Wait for device to be in a particular state.

        State must be one of {"any", "bootloader", "device", "recovery", "sideload"}
        """
        if state not in {"any", "bootloader", "device", "recovery", "sideload"}:
            raise ValueError("Invalid state to wait for.")
        msg = b"host-serial:%b:wait-for-usb-%b" % (self.serial, state.encode("ascii"))
        await _command_transact(self._conn, msg)

    async def command(self, cmdline, usepty=False):
        """Run a non-interactive shell command.

        Uses ring buffers to collect outputs to avoid a runaway device command
        from filling host memory. However, this might possibly truncate output.

        Returns:
            stdout (string): output of command
            stderr (string): error output of command
            exitstatus (ExitStatus): the exit status of the command.
        """
        if self.features is None:
            await self.get_features()
        if "shell_v2" not in self.features:
            raise AdbCommandFail("Only shell v2 protocol currently supported.")
        cmdline, name = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, usepty, cmdline)
        sp = ShellProtocol(self._conn.socket)
        stdout = ringbuffer.RingBuffer(MAX_PAYLOAD)
        stderr = ringbuffer.RingBuffer(MAX_PAYLOAD)
        resp = await sp.run(None, stdout, stderr)
        await self._conn.close()
        rc = resp[0]
        if rc & 0x80:
            rc = -(rc & 0x7F)
        return (stdout.read().decode("utf8"),
                stderr.read().decode("utf8"),
                exitstatus.ExitStatus(
                    None,
                    name="{}<EMAIL>(name, self.serial.decode("ascii")),
                    returncode=rc)
                )

    async def list(self, name, coro_cb):
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        await sp.list(name, coro_cb)
        await sp.quit()
        await self._conn.close()

    async def stat(self, path):
        """stat a remote file or directory.

        Return os.stat_result with attributes from remote path.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            st = await sp.stat(path)
        finally:
            await sp.quit()
            await self._conn.close()
        return st

    async def push(self, localfiles: list, remotepath: str, sync: bool = False):
        """Push a list of local files to remote file or directory.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            resp = await sp.push(localfiles, remotepath, sync)
        finally:
            await sp.quit()
            await self._conn.close()
        return resp

    async def pull_file(self, remotepath: str, filelike: typing.BinaryIO):
        """Pull a remote file into given file-like object.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            text = await sp.pull_file(remotepath, filelike)
        finally:
            await sp.quit()
            await self._conn.close()
        return text

    async def start(self, cmdline, stdoutstream, stderrstream):
        """Start a process on device with the shell protocol.

        Returns a curio Task object wrapping the ShellProtocol run.
        """
        cmdline, name = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, False, cmdline)
        sp = ShellProtocol(self._conn.socket)
        return await spawn(sp.run(None, stdoutstream, stderrstream))

    async def spawn(self, cmdline):
        """Start a process on device in raw mode.

        Return:
            DeviceProcess with active connection.
        """
        cmdline, name = _fix_command_line(cmdline)
        logging.info("adb.spawn({})".format(cmdline))
        sock = await _start_exec(self.serial, self._conn, cmdline)
        return DeviceProcess(sock, name)

    async def install(self, apkfile, allow_test=True, installer=None,
                      onsdcard=False, onflash=False, allow_downgrade=True,
                      grant_all=True):
        """Install an APK.

        Performs a streaming installation.  Returns True if Success.
        """
        # TODO(dart) other options
        st = os.stat(apkfile)  # TODO(dart) fix potential long blocker
        if not stat.S_ISREG(st.st_mode):
            raise ValueError("The apkfile must be a regular file.")
        cmdline = ["cmd", "package", "install"]
        if allow_test:
            cmdline.append("-t")
        if installer is not None:
            cmdline.extend(["-i", str(installer)])
        if onsdcard:
            cmdline.append("-s")
        if onflash:
            cmdline.append("-f")
        if allow_downgrade:
            cmdline.append("-d")
        if grant_all:
            cmdline.append("-g")
        cmdline.extend(["-S", str(st.st_size)])
        cmdline, name = _fix_command_line(cmdline)
        sock = await _start_exec(self.serial, self._conn, cmdline)
        p = DeviceProcess(sock, name)
        del sock
        async with streams.aopen(apkfile, "rb") as afo:
            await p.copy_from(afo)
        status_response = await p.read(4096)
        await p.close()
        return b'Success' in status_response
    # TODO(dart) install sessions

    async def package(self, cmd, *args, user=None, **kwargs):
        """Manage packages.

        Equivalent of 'pm' command.
        """
        cmdline = ['cmd', 'package']
        if user is not None:
            cmdline.append("--user")
            cmdline.append(str(user))
        cmdline.append(cmd)
        cmdline.extend(str(arg) for arg in args)
        for opt, optarg in kwargs.items():
            cmdline.append("--" + opt)
            if optarg not in (None, True):
                cmdline.append(str(optarg))
        out, err, es = await self.command(cmdline)
        if not es:
            raise AdbCommandFail(err)
        return out

    async def reconnect(self):
        msg = b"host-serial:%b:reconnect" % (self.serial,)
        await _command_transact(self._conn, msg)
        await self._conn.close()
        await self._init()

    async def logcat_clear(self):
        """Clear logcat buffer."""
        stdout, stderr, es = await self.command(["logcat", "-c"])
        if not es:
            raise AdbCommandFail("Didn't clear logcat")

    async def logcat(self, stdoutstream, stderrstream, format="threadtime",
                     buffers="default", modifiers=None, binary=False,
                     regex=None, dump=False, logtags=""):
        """Coroutine for streaming logcat output to the provided file-like
        streams.

        Args:
            stdout, stderr: file-like object to write log events to.
            binary: bool output binary format if True.
            regex: A Perl compatible regular expression to match messages against.
            format: str of one of the following:
                    "brief", "long", "process", "raw", "tag", "thread",
                    "threadtime", "time".
            buffers: list or comma separated string of:
                     'main', 'system', 'radio', 'events', 'crash', 'default' or 'all'
            modifiers: str of one or more of:
                       epoch", "monotonic", "uid", "usec", "UTC", "year", "zone"
            logcats: str of space separated filter expressions.
        """
        logtags = os.environ.get("ANDROID_LOG_TAGS", logtags)
        cmdline = ['exec', 'logcat']
        # buffers
        if isinstance(buffers, str):
            buffers = buffers.split(",")
        for bufname in buffers:
            cmdline.extend(["-b", bufname])
        if binary:
            cmdline.append("-B")
        if regex:
            cmdline.extend(["-e", regex])
        if dump:
            cmdline.append("-d")
        # output format
        if format not in {"brief", "long", "process", "raw", "tag",
                          "thread", "threadtime", "time"}:
            raise ValueError("Bad format type.")
        if modifiers:
            if isinstance(modifiers, str):
                modifiers = modifiers.split(",")
            for modifier in modifiers:
                if modifier in {"epoch", "monotonic", "uid", "usec", "UTC",
                                "year", "zone"}:
                    format += ("," + modifier)
                else:
                    raise ValueError("Invalid logcat format modifier")
        cmdline.extend(["-v", format])
        # logtags
        if logtags:
            logtags = logtags.replace('"', '\\"')
            cmdline.extend(logtags.split())
        # go!
        cmdline, _ = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, False, cmdline)
        sp = ShellProtocol(self._conn.socket)
        await sp.run(None, stdoutstream, stderrstream)


class AndroidDeviceClient:
    """An active adb client per device.

    For synchronous (blocking) style code.
    """
    def __init__(self, serial, host="localhost", port=ADB_PORT):
        self._aadb = _AsyncAndroidDeviceClient(serial, host, port)
        self.open()
        self.features = self.get_features()

    def close(self):
        get_kernel().run(self._aadb.close())

    def open(self):
        get_kernel().run(self._aadb.open())

    @property
    def serial(self):
        return self._aadb.serial

    @property
    def async_client(self):
        return self._aadb

    def get_features(self):
        return get_kernel().run(self._aadb.get_features())

    def get_state(self):
        return get_kernel().run(self._aadb.get_state())

    def reboot(self):
        """Reboot the device."""
        return get_kernel().run(self._aadb.reboot())

    def remount(self):
        """Remount filesystem read-write."""
        return get_kernel().run(self._aadb.remount())

    def root(self):
        """Become root on the device."""
        return get_kernel().run(self._aadb.root())

    def unroot(self):
        """Become non-root on the device."""
        return get_kernel().run(self._aadb.unroot())

    def forward(self, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        return get_kernel().run(self._aadb.forward(hostport, devport))

    def kill_forward(self, hostport):
        """Tell server to remove forwarding TCP ports.
        """
        return get_kernel().run(self._aadb.kill_forward(hostport))

    def list_forward(self):
        """Get a list of currently forwarded ports."""
        return get_kernel().run(self._aadb.list_forward())

    def wait_for(self, state: str):
        """Wait for device to be in a particular state.

        State must be one of {"any", "bootloader", "device", "recovery", "sideload"}
        """
        return get_kernel().run(self._aadb.wait_for(state))

    def command(self, cmdline, usepty=False):
        """Run a non-interactive shell command.

        Uses ring buffers to collect outputs to avoid a runaway device command
        from filling host memory. However, this might possibly truncate output.

        Returns:
            stdout (string): output of command
            stderr (string): error output of command
            exitstatus (ExitStatus): the exit status of the command.
        """
        return get_kernel().run(self._aadb.command(cmdline, usepty))

    def spawn(self, cmdline):
        """Start a process on device in raw mode.

        Return:
            DeviceProcess with active connection.
        """
        return get_kernel().run(self._aadb.spawn(cmdline))

    def install(self, apkfile, **kwargs):
        """Install an APK.

        Default flags are best for testing, but you can override. See asyn
        method.
        """
        coro = self._aadb.install(apkfile, **kwargs)
        return get_kernel().run(coro)

    def package(self, cmd, *args, user=None, **kwargs):
        """Manage packages.

        Equivalent of 'pm' command.
        """
        coro = self._aadb.package(cmd, *args, user=user, **kwargs)
        return get_kernel().run(coro)

    def list(self, name, cb):
        """Perform a directory listing.

        Arguments:
            name: str, name of directory
            cb: callable with signature cb(os.stat_result, filename)
        """
        async def acb(stat, path):
            await block_in_thread(cb, stat, path)
        coro = self._aadb.list(name, acb)
        return get_kernel().run(coro)

    def stat(self, path: str):
        """stat a remote file or directory.

        Return os.stat_result with attributes from remote path.
        """
        coro = self._aadb.stat(path)
        return get_kernel().run(coro)

    def push(self, localfiles: list, remotepath: str, sync: bool = False):
        """Push a list of local files to remote file or directory.
        """
        coro = self._aadb.push(localfiles, remotepath, sync)
        return get_kernel().run(coro)

    def pull(self, remotepath, localpath):
        """Pull a single file from device to local file system.
        """
        coro = self._aadb.pull(remotepath, localpath)
        return get_kernel().run(coro)

    def pull_file(self, remotepath: str, filelike: typing.BinaryIO):
        """Pull a file into local memory, as bytes.

        A path to a file on the device.

        """
        coro = self._aadb.pull_file(remotepath, filelike)
        return get_kernel().run(coro)

    def reconnect(self):
        """Reconnect from device side."""
        return get_kernel().run(self._aadb.reconnect())

    def logcat_clear(self):
        """Clear logcat buffer."""
        return get_kernel().run(self._aadb.logcat_clear())

    async def logcat(self, stdoutstream, stderrstream, format="threadtime",
                     buffers="default", modifiers=None, binary=False,
                     regex=None, logtags=""):
        """Coroutine for streaming logcat output to the provided file-like
        streams.
        """
        await self._aadb.logcat(stdoutstream, stderrstream, format=format,
                                buffers=buffers, modifiers=modifiers,
                                binary=binary, regex=regex, logtags=logtags)


class DeviceProcess:
    """Represents an attached process on the device.
    """
    def __init__(self, asocket, name):
        self.socket = asocket
        self.name = name

    def __str__(self):
        return "DeviceProcess running {!r}".format(self.name)

    async def copy_to(self, otherfile):
        """Copy output from this process to another file stream."""
        while True:
            data = await self.socket.recv(MAX_PAYLOAD)
            if not data:
                break
            await otherfile.write(data)

    async def copy_from(self, otherfile):
        """Copy output from another file stream to this process."""
        while True:
            data = await otherfile.read(MAX_PAYLOAD)
            if not data:
                break
            await self.socket.sendall(data)

    async def read(self, amt):
        return await self.socket.recv(amt)

    async def write(self, data):
        return await self.socket.sendall(data)

    async def close(self):
        if self.socket is not None:
            await self.socket.close()
            self.socket = None

    def sync_close(self):
        return get_kernel().run(self.close)


def _fix_command_line(cmdline):
    """Fix the command.

    If a list, quote the components if required.
    Return encoded command line as bytes and the command base name.
    """
    if isinstance(cmdline, list):
        name = cmdline[0]
        cmdline = " ".join('"{}"'.format(s) if " " in s else str(s) for s in cmdline)
    else:
        name = cmdline.split()[0]
    return cmdline.encode("utf8"), name


# Perform shell request, connection stays open
async def _start_shell(serial, conn, usepty, cmdline):
    tpmsg = b"host:transport:%b" % serial
    msg = b"shell,%b:%b" % (b",".join([b"v2",
                                       b"TERM=%b" % os.environb[b"TERM"],
                                       b"pty" if usepty else b"raw"]),
                            cmdline)
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)


async def _start_exec(serial, conn, cmdline):
    tpmsg = b"host:transport:%b" % serial
    msg = b"exec:%b" % (cmdline,)
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)
    sock = conn.socket.dup()
    await conn.close()
    return sock


# Send command to specific device with orderly shutdown
async def _connect_command(serial, conn, msg):
    tpmsg = b"host:transport:%b" % serial
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)
    await conn.close()


# Root command transaction is special since device adbd restarts.
async def _special_command(serial, conn, cmd):
    tpmsg = b"host:transport:%b" % serial
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(cmd, expect_response=False)
    resp = await conn.socket.recv(4096)
    await conn.close()
    return resp


class ShellProtocol:
    """Implement the shell protocol v2."""
    IDSTDIN = 0
    IDSTDOUT = 1
    IDSTDERR = 2
    IDEXIT = 3
    CLOSESTDIN = 4
    # Window size change (an ASCII version of struct winsize).
    WINDOWSIZECHANGE = 5
    # Indicates an invalid or unknown packet.
    INVALID = 255

    def __init__(self, socket):
        self._sock = socket
        self._header = struct.Struct(b"<BI")
        self._winsize = None

    async def run(self, inbuf, outbuf, errbuf):
        while True:
            tl = await self._sock.recv(5)
            msgtype, msglen = self._header.unpack(tl)
            if msgtype == 0:
                pass
            elif msgtype == 1:
                data = await self._sock.recv(msglen)
                outbuf.write(data)
            elif msgtype == 2:
                data = await self._sock.recv(msglen)
                errbuf.write(data)
            elif msgtype == 3:
                data = await self._sock.recv(msglen)
                return data
            elif msgtype == 4:
                pass  # TODO
            elif msgtype == 5:
                data = await self._sock.recv(msglen)
                self._winsize = data
            else:
                raise AdbProtocolError("Unhandled shell protocol message type.")


class SyncProtocol:
    """Implementation of Android SYNC protocol.

    Only recent devices are supported.
    """

    mkid = lambda code: int.from_bytes(code, byteorder='little')  # noqa
    ID_LSTAT_V1 = mkid(b'STAT')
    ID_STAT_V2 = mkid(b'STA2')
    ID_LSTAT_V2 = mkid(b'LST2')
    ID_LIST = mkid(b'LIST')
    ID_SEND = mkid(b'SEND')
    ID_RECV = mkid(b'RECV')
    ID_DENT = mkid(b'DENT')
    ID_DONE = mkid(b'DONE')
    ID_DATA = mkid(b'DATA')
    ID_OKAY = mkid(b'OKAY')
    ID_FAIL = mkid(b'FAIL')
    ID_QUIT = mkid(b'QUIT')
    del mkid

    SYNC_DATA_MAX = 65536
    SYNCMSG_DATA = struct.Struct("<II")  # id, size
    # id; mode; size; time; namelen;
    SYNCMSG_DIRENT = struct.Struct("<IIIII")
    # id; error; dev; ino; mode; nlink; uid; gid; size; atime; mtime; ctime;
    SYNCMSG_STAT_V2 = struct.Struct("<IIQQIIIIQqqq")
    SYNCMSG_STATUS = struct.Struct("<II")  # id, msglen

    def __init__(self, serial):
        self.serial = serial
        self.socket = None

    async def connect_with(self, adbconnection):
        tpmsg = b"host:transport:%b" % self.serial
        msg = b"sync:"
        await adbconnection.open()
        await adbconnection.message(tpmsg, expect_response=False)
        await adbconnection.message(msg, expect_response=False)
        self.socket = adbconnection.socket

    async def quit(self):
        msg = SyncProtocol.SYNCMSG_STATUS.pack(SyncProtocol.ID_QUIT, 0)
        self.socket.sendall(msg)

    async def send_request(self, protoid: int, path_and_mode: str):
        path_and_mode = path_and_mode.encode("utf-8")
        length = len(path_and_mode)
        if length > 1024:
            raise ValueError("Can't send message > 1024")
        hdr = SyncProtocol.SYNCMSG_DATA.pack(protoid, length)
        await self.socket.sendall(hdr + path_and_mode)

    async def list(self, path, cb_coro):
        """List a directory on device."""
        await self.send_request(SyncProtocol.ID_LIST, path)
        while True:
            resp = await self.socket.recv(SyncProtocol.SYNCMSG_DIRENT.size)
            msgid, mode, size, time, namelen = SyncProtocol.SYNCMSG_DIRENT.unpack(resp)
            if msgid == SyncProtocol.ID_DONE:
                return True
            if msgid != SyncProtocol.ID_DENT:
                return False
            name = await self.socket.recv(namelen)
            stat = os.stat_result((mode, None, None, None, 0, 0, size, None, float(time), None))
            await cb_coro(stat, name.decode("utf-8"))

    async def stat(self, remotepath):
        """Stat a path."""
        s = SyncProtocol.SYNCMSG_STAT_V2
        await self.send_request(SyncProtocol.ID_STAT_V2, remotepath)
        resp = await self.socket.recv(s.size)
        stat_id, err, dev, ino, mode, nlink, uid, gid, size, atime, mtime, ctime = s.unpack(resp)
        if stat_id == SyncProtocol.ID_STAT_V2:
            if err != 0:
                raise OSError(err, errno.errorcode[err], remotepath)
            sr = os.stat_result((mode, ino, dev, nlink, uid, gid, size,
                                 float(atime), float(mtime), float(ctime),  # floats
                                 # int nanoseconds, but not really
                                 atime * 1e9, mtime * 1e9, ctime * 1e9))
            return sr
        else:
            raise AdbProtocolError("SyncProtocol: invalid response type.")

    async def push(self, localfiles, remotepath, sync=False):
        """Push files to device destination."""
        try:
            dest_st = await self.stat(remotepath)
        except FileNotFoundError:
            dst_exists = False
            dst_isdir = False
        else:
            dst_exists = True
            if stat.S_ISDIR(dest_st.st_mode):
                dst_isdir = True
            elif stat.S_ISREG(dest_st.st_mode):
                dst_isdir = False
            else:
                raise ValueError("push: destination is not a directory or "
                                 "regular file")
        if not dst_isdir:
            if len(localfiles) > 1:
                raise ValueError("push: destination is not a dir when copying "
                                 "multiple files.")
            if dst_exists:
                raise ValueError("push: destination exists")
        for localfile in localfiles:
            local_st = os.stat(localfile)

            if stat.S_ISDIR(local_st.st_mode):
                rpath = os.path.join(remotepath, os.path.basename(localfile))
                await self._copy_local_dir_remote(localfile, rpath, sync)

            if stat.S_ISREG(local_st.st_mode):
                if dst_isdir:
                    rpath = os.path.join(remotepath, os.path.basename(localfile))
                # If synchronize requested, just stat remote and return if size
                # and mtime are equal.
                if sync:
                    try:
                        dst_stat = await self.stat(rpath)
                    except OSError:
                        pass
                    else:
                        if (local_st.st_size == dst_stat.st_size and
                                local_st.st_mtime == dst_stat.st_mtime):
                            return
                await self._sync_send(localfile.encode("utf8"),
                                      rpath.encode("utf8"),
                                      local_st)

    async def pull(self, remotepath, localpath):
        with open(localpath, "wb") as fo:
            await self.pull_file(remotepath, fo)
            # TODO(dart) directories and symlinks

    async def pull_file(self, remotepath, filelike):
        smd = SyncProtocol.SYNCMSG_DATA
        src_st = await self.stat(remotepath)
        if stat.S_ISREG(src_st.st_mode):
            await self.send_request(SyncProtocol.ID_RECV, remotepath)
            while True:
                resp = await self.socket.recv(smd.size)
                msg_id, msg_len = smd.unpack(resp)
                if msg_id == SyncProtocol.ID_DATA:
                    data = await self.socket.recv(msg_len)
                    filelike.write(data)
                elif msg_id == SyncProtocol.ID_DONE:
                    break
                else:
                    raise AdbProtocolError(
                        "SyncProtocol: pull: can't handle message: {!r}.".format(msg_id))
        elif stat.S_ISDIR(src_st.st_mode):
            raise NotImplementedError("TODO(dart)")
        else:
            raise NotImplementedError("TODO(dart)")

    async def _sync_send(self, src_path, dst_path, local_st):
        if stat.S_ISLNK(local_st.st_mode):
            raise NotImplementedError("TODO(dart)")
        if local_st.st_size < SyncProtocol.SYNC_DATA_MAX:
            with open(src_path, "rb") as fo:
                data = fo.read()
            path_and_mode = b"%s,%d" % (dst_path, local_st.st_mode)
            return await self._send_small_file(path_and_mode, data, int(local_st.st_mtime))
        else:
            return await self._send_large_file(src_path, dst_path, local_st)

    async def _send_small_file(self, path_and_mode, data, mtime):
        sm = SyncProtocol.SYNCMSG_DATA
        buf = ringbuffer.RingBuffer(SyncProtocol.SYNC_DATA_MAX << 1)  # big enough to not wrap
        buf.write(sm.pack(SyncProtocol.ID_SEND, len(path_and_mode)))
        buf.write(path_and_mode)
        buf.write(sm.pack(SyncProtocol.ID_DATA, len(data)))
        buf.write(data)
        buf.write(sm.pack(SyncProtocol.ID_DONE, mtime))
        await self.socket.sendall(buf.read())
        return await self._copy_done()

    async def _send_large_file(self, src_path, dst_path, local_st):
        sm = SyncProtocol.SYNCMSG_DATA
        buf = ringbuffer.RingBuffer(SyncProtocol.SYNC_DATA_MAX << 1)
        path_and_mode = b"%s,%d" % (dst_path, local_st.st_mode)
        buf.write(sm.pack(SyncProtocol.ID_SEND, len(path_and_mode)))
        buf.write(path_and_mode)
        await self.socket.sendall(buf.read())

        buf.clear()
        chunksize = SyncProtocol.SYNC_DATA_MAX - sm.size
        with open(src_path, "rb") as fo:
            while True:
                chunk = fo.read(chunksize)
                if not chunk:
                    break
                buf.write(sm.pack(SyncProtocol.ID_DATA, len(chunk)))
                buf.write(chunk)
                await self.socket.sendall(buf.read())

        buf.clear()
        buf.write(sm.pack(SyncProtocol.ID_DONE, int(local_st.st_mtime)))
        await self.socket.sendall(buf.read())
        return await self._copy_done()

    async def _copy_done(self):
        sm = SyncProtocol.SYNCMSG_STATUS
        raw = await self.socket.recv(sm.size)
        msg_id, msg_len = sm.unpack(raw)
        if msg_id == SyncProtocol.ID_OKAY:
            return True
        if msg_id == SyncProtocol.ID_FAIL:
            msg = await self.socket.recv(msg_len)
            raise AdbCommandFail(
                "large file copy not OKAY: {!r}".format(msg.decode("utf8")))

    async def _copy_local_dir_remote(self, localfile, remotepath, sync):
        raise NotImplementedError("TODO(dart)")


class LogcatHandler:
    """Host side logcat handler that receives logcat messages in binary mode
    over raw connection.
    """

    LOGCAT_MESSAGE = struct.Struct("<HHiIIIII")  # logger_entry_v4
    #  uint16_t len;       length of the payload
    #  uint16_t hdr_size;  sizeof(struct logger_entry_v4)
    #  int32_t pid;        generating process's pid
    #  uint32_t tid;       generating process's tid
    #  uint32_t sec;       seconds since Epoch
    #  uint32_t nsec;      nanoseconds
    #  uint32_t lid;       log id of the payload, bottom 4 bits currently
    #  uint32_t uid;       generating process's uid
    #  char msg[0];        the entry's payload

    def __init__(self, aadb):
        self._aadb = aadb

    def clear(self):
        """Clear logcat buffers."""
        return get_kernel().run(self._aadb.logcat_clear())

    def dump(self):
        """Dump logs to stdout until interrupted."""
        return get_kernel().run(self._dump())

    async def _dump(self):
        ac = self._aadb
        signalset = SignalEvent(signal.SIGINT, signal.SIGTERM)
        await ac.logcat_clear()
        proc = await ac.spawn('logcat -B -b default')
        task = await spawn(self._read_and_dump(proc, streams.FileStream(sys.stdout.buffer)))
        await signalset.wait()
        await task.cancel()
        await proc.close()

    async def _read_and_dump(self, proc, out):
        while True:
            lm = await self._read_one(proc)
            await out.write(str(lm).encode("utf8"))
            await out.write(b'\n')

    async def _read_one(self, proc):
        s = self.LOGCAT_MESSAGE
        rawhdr = await proc.read(s.size)
        payload_len, hdr_size, pid, tid, sec, nsec, lid, uid = s.unpack(rawhdr)
        payload = await proc.read(payload_len)
        return logcat.LogcatMessage(pid, tid, sec, nsec, lid, uid, payload)

    def dump_to(self, localfile, logtags=None):
        """Dump all current logs to a file, in binary format."""
        return get_kernel().run(self._dump_to(localfile, logtags))

    async def _dump_to(self, localfile, logtags):
        cmdline = ['logcat', '-d', '-B', '-b', 'default']
        if logtags:
            cmdline.extend(logtags.split())
        proc = await self._aadb.spawn(cmdline)
        with open(localfile, "ab") as fo:
            await proc.copy_to(streams.FileStream(fo))
        await proc.close()

    def watch_for(self, tag=None, priority=None, text=None, timeout=90):
        """Watch for first occurence of a particular set of tag, priority, or
        message text.

        If tag is given watch for first of that tag.
        If tag and priority is given watch for tag only with that priority.
        if tag and text is given watch for tag with the given text in the
        message part.
        If text is given look for first occurence of text in message part.
        """
        if tag is None and text is None:
            raise ValueError("watch_for: must supply one or both of 'tag' or "
                             "'text' parameters.")
        return get_kernel().run(self._watch_for(tag, priority, text, timeout))

    async def _watch_for(self, tag, priority, text, timeout):
        lm = None
        ac = self._aadb
        await ac.logcat_clear()
        proc = await ac.spawn('logcat -B -b default')
        try:
            async with timeout_after(timeout):
                while True:
                    new = await self._read_one(proc)
                    if tag and tag == new.tag:
                        if priority is not None:
                            if new.priority == priority:
                                lm = new
                                break
                        else:
                            lm = new
                            break
                        if text is not None:
                            if text in new.message:
                                lm = new
                                break
                    if text and text in new.message:
                        lm = new
                        break
        except TaskTimeout:
            pass
        await proc.close()
        return lm


class AndroidDevice:
    """Information about attached Android device.

    No connection necessary.
    """

    def __init__(self, serial, product, model, device, state):
        self.type = "phone"
        self.serial = serial
        self.product = product
        self.model = model
        self.device = device
        self.state = state

    def __repr__(self):
        return ("AndroidDevice("
                "serial={!r}, product={!r}, model={!r}, device={!r}, state={!r})".format(
                    self.serial, self.product, self.model, self.device, self.state))


# b'HTxxxserial device usb:1-1.1 product:marlin model:Pixel_XL device:marlin\n'
def _device_factory(line):
    parts = line.decode("ascii").split()
    serial = parts[0]
    state = parts[1]
    if state == "device":
        product = parts[3].partition(":")[2]
        model = parts[4].partition(":")[2]
        device = parts[5].partition(":")[2]
        return AndroidDevice(serial, product, model, device, state)
    else:
        return AndroidDevice(serial, None, None, None, state)


if __name__ == "__main__":
    from devtest import debugger
    debugger.autodebug()
    start_server()
    print("Test AdbClient:")
    c = AdbClient()
    print("  Server version:", c.server_version)
    for devinfo in c.get_device_list():
        print("    ", devinfo)
    print("Forwards:", c.list_forward())
    c.close()
    del c

    print("Test AndroidDeviceClient:")
    ac = AndroidDeviceClient(devinfo.serial)
    ac.wait_for("device")
    print("  features:", ac.features)
    print("  state:", ac.get_state())
    print("  running 'ls /sdcard':")
    stdout, stderr, es = ac.command(["ls", "/sdcard"])
    print("    ", es)
    print("    stdout:", repr(stdout))
    print("    stderr:", repr(stderr))
    print("forward list:")
    print(repr(ac.list_forward()))
    ac.close()
    del ac

    # Test async with logcat. ^C to stop it.
    async def dostuff():
        ac = await AsyncAndroidDeviceClient(devinfo.serial)

        stdout, stderr, es = await ac.command(["ls", "/sdcard"])
        print("    ", es)
        print("    stdout:", repr(stdout))
        print("    stderr:", repr(stderr))

        fl = await ac.list_forward()
        print("Forward list:")
        print(repr(fl))
        print("Logcat:")
        signalset = SignalEvent(signal.SIGINT, signal.SIGTERM)
        await ac.wait_for("device")
        try:
            await ac.logcat_clear()
            task = await spawn(ac.logcat(sys.stdout.buffer, sys.stdout.buffer,
                                         buffers="kernel,main",
                                         format="long", modifiers="epoch,usec",
                                         binary=False,
                                         logtags=" ".join(sys.argv[1:])))
            await signalset.wait()
            await task.cancel()
        finally:
            await ac.close()
        await ac.close()

    kern = get_kernel()
    kern.run(dostuff)

# vim:ts=4:sw=4:softtabstop=4:smarttab:expandtab

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

"""
Native adb client-to-server protocol implementation.

This is an async implementation, suitable for including in an event loop.
"""

import os
import sys
import stat
import errno
import struct
import signal
import typing

from devtest import logging
from devtest import timers
from devtest import ringbuffer
from devtest.io import socket
from devtest.io import streams
from devtest.os import process
from devtest.os import procutils
from devtest.os import exitstatus
from devtest.io.reactor import (get_kernel, spawn, block_in_thread, sleep,
                                SignalEvent, timeout_after, TaskTimeout)

from . import logcat


ADB = procutils.which("adb")
if ADB is None:
    raise ImportError("The adb program was not found in PATH. "
                      "This module will not work.")

ADB_PORT = 5037

MAX_PAYLOAD_V2 = 256 * 1024
MAX_PAYLOAD = MAX_PAYLOAD_V2


class Error(Exception):
    pass


class AdbProtocolError(Error):
    """An error in the protocol was detected."""


class AdbCommandFail(Error):
    """An error indicated by the server."""


def _run_adb(adbcommand):
    """Run the adb binary."""
    cmd = [ADB]
    cmd.extend(adbcommand.split())
    return process.check_output(cmd)


def start_server(port=ADB_PORT):
    return _run_adb("-P {} start-server".format(port))


def kill_server(port=ADB_PORT):
    return _run_adb("-P {} kill-server".format(port))


class AdbConnection:
    """Asynchronous adb socket wrapper."""

    def __init__(self, host="localhost", port=ADB_PORT):
        self.host = host
        self.port = port
        self.socket = None

    async def close(self):
        if self.socket is not None:
            s = self.socket
            self.socket = None
            await s.close()

    def isopen(self):
        return bool(self.socket)

    async def open(self):
        sock = socket.socket(family=socket.AF_INET,
                             type=socket.SOCK_STREAM)
        errno = await sock.connect_ex((self.host, self.port))
        if errno != 0:
            raise IOError(errno, "Could not connect to adb server")
        self.socket = sock

    async def message(self, msg, expect_response=True):
        msg = b"%04x%b" % (len(msg), msg)
        await self.socket.sendall(msg)
        stat = await self.socket.recv(4)
        if stat == b"OKAY":
            if expect_response:
                lenst = await self.socket.recv(4)
                length = int(lenst, 16)
                if length:
                    return await self.socket.recv(length)
                else:
                    return b""
        elif stat == b"FAIL":
            lenst = await self.socket.recv(4)
            length = int(lenst, 16)
            resp = await self.socket.recv(length)
            raise AdbCommandFail("message FAIL: {}".format(resp.decode("ascii")))
        else:
            raise AdbProtocolError(stat.decode("ascii"))

    async def read_protocol_string(self):
        lenst = await self.socket.recv(4)
        length = int(lenst, 16)
        if length:
            return await self.socket.recv(length)
        else:
            return b""


async def _transact(conn, msg):
    await conn.open()
    resp = await conn.message(msg)
    await conn.close()
    return resp


# On the host: 1st OKAY is connect, 2nd OKAY is status.
async def _command_transact(conn, msg):
    await conn.open()
    msg = b"%04x%b" % (len(msg), msg)
    await conn.socket.sendall(msg)
    connstat = await conn.socket.recv(4)
    if connstat == b"OKAY":
        stat = await conn.socket.recv(4)
        if stat != b"OKAY":
            raise AdbCommandFail("command not OKAY")
    elif connstat == b"FAIL":
        lenst = await conn.socket.recv(4)
        length = int(lenst, 16)
        resp = await conn.socket.recv(length)
        await conn.close()
        raise AdbCommandFail(resp.decode("ascii"))
    else:
        await conn.close()
        raise AdbProtocolError(stat.decode("ascii"))


class AdbClient:
    """An adb client, synchronous.

    For general host side operations.
    """
    def __init__(self, host="localhost", port=ADB_PORT):
        self._conn = AdbConnection(host=host, port=port)

    def _message(self, msg):
        if self._conn is None:
            raise Error("Operation on a closed client.")
        return get_kernel().run(_transact(self._conn, msg))

    def close(self):
        if self._conn is not None:
            get_kernel().run(self._conn.close())
            self._conn = None

    def get_device(self, serial):
        """Get a AndroidDeviceClient instance.
        """
        return AndroidDeviceClient(
            serial, host=self._conn.host, port=self._conn.port)

    def get_state(self, serial):
        """Get the current state of a device.

        Arguments:
            serial: str of the device serial number.

        Returns:
            one of {"device", "unauthorized", "bootloader"}
            or None if serial not found.
        """
        for dev in self.get_device_list():
            if dev.serial == serial:
                return dev.state

    def get_device_list(self):
        """Get list of attached devices.

        Returns:
            list of AndroidDevice instances.
        """
        dl = []
        resp = self._message(b"host:devices-l")
        for line in resp.splitlines():
            dl.append(_device_factory(line))
        return dl

    def forward(self, serial, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        msg = b"host-serial:%b:forward:tcp:%d;tcp:%d" % (serial.encode("ascii"),
                                                         hostport, devport)
        get_kernel().run(_command_transact(self._conn, msg))

    def list_forward(self):
        """Return a list of currently forwarded ports.

        Returns:
            Tuple of (serial, host_port, device_port).
        """
        resp = self._message(b"host:list-forward")
        fl = []
        for line in resp.splitlines():
            # <serial> " " <local> " " <remote> "\n"
            serno, host_port, remote_port = line.split()
            fl.append((serno.decode("ascii"),
                       int(host_port.split(b":")[1]),
                       int(remote_port.split(b":")[1])))
        return fl

    def reconnect_offline(self):
        self._message(b"host:reconnect-offline")

    def reconnect(self, serial):
        self._message(b"host-serial:%b:reconnect" % (serial.encode("ascii"),))
        while self.get_state(serial) is None:
            timers.nanosleep(1.1)

    @property
    def server_version(self):
        """The server's version number."""
        resp = self._message(b"host:version")
        return int(resp, 16)


def getAsyncAndroidDeviceClient(serial, host="localhost", port=ADB_PORT):
    """Get initialized _AsyncAndroidDeviceClient object from sync code.
    """
    return get_kernel().run(AsyncAndroidDeviceClient(serial, host, port))


async def AsyncAndroidDeviceClient(serial, host="localhost", port=ADB_PORT):
    """Get initialized _AsyncAndroidDeviceClient instance from async code."""
    ac = _AsyncAndroidDeviceClient(serial, host, port)
    await ac._init()
    return ac


class _AsyncAndroidDeviceClient:
    """An active adb client per device.

    For device specific operations.
    For use in asynchronous event loops.
    """
    def __init__(self, serial, host="localhost", port=ADB_PORT):
        self.serial = serial.encode("ascii")
        self._conn = AdbConnection(host=host, port=port)

    async def _init(self):
        await self._conn.open()
        await self.get_features()

    async def _message(self, msg):
        return await _transact(self._conn, msg)

    async def get_features(self):
        msg = b"host-serial:%b:features" % (self.serial,)
        resp = await self._message(msg)
        self.features = resp.decode("ascii")
        return self.features

    async def close(self):
        await self._conn.close()

    async def open(self):
        await self._conn.open()

    async def get_state(self):
        msg = b"host-serial:%b:get-state" % (self.serial,)
        resp = await self._message(msg)
        return resp.decode("ascii")

    async def reboot(self):
        await _connect_command(self.serial, self._conn, b"reboot:")

    async def remount(self):
        resp = await _special_command(self.serial, self._conn, b"remount:")
        return resp.decode("ascii")

    async def root(self):
        resp = await _special_command(self.serial, self._conn, b"root:")
        await sleep(3)  # adbd needs some time to initialize
        return resp.decode("ascii")

    async def unroot(self):
        resp = await _special_command(self.serial, self._conn, b"unroot:")
        await sleep(3)  # adbd needs some time to initialize
        return resp.decode("ascii")

    async def forward(self, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        msg = b"host-serial:%b:forward:tcp:%d;tcp:%d" % (self.serial,
                                                         hostport, devport)
        await _command_transact(self._conn, msg)

    async def kill_forward(self, hostport):
        """Tell server to remove forwarding TCP ports.
        """
        msg = b"host-serial:%b:killforward:tcp:%d" % (self.serial, hostport)
        await _command_transact(self._conn, msg)

    async def kill_forward_all(self):
        """Tell server to remove all forwarding TCP ports.
        """
        msg = b"host-serial:%b:killforward-all" % (self.serial,)
        await _command_transact(self._conn, msg)

    async def list_forward(self):
        """Return a list of currently forwarded ports.

        Returns:
            Tuple of (host_port, device_port).
        """
        msg = b"host-serial:%b:list-forward" % (self.serial,)
        resp = await self._message(msg)
        fl = []
        for line in resp.splitlines():
            # <serial> " " <local> " " <remote> "\n"
            serno, host_port, remote_port = line.split()
            if serno == self.serial:
                fl.append((int(host_port.split(b":")[1]),
                           int(remote_port.split(b":")[1])))
        return fl

    async def wait_for(self, state: str):
        """Wait for device to be in a particular state.

        State must be one of {"any", "bootloader", "device", "recovery", "sideload"}
        """
        if state not in {"any", "bootloader", "device", "recovery", "sideload"}:
            raise ValueError("Invalid state to wait for.")
        msg = b"host-serial:%b:wait-for-usb-%b" % (self.serial, state.encode("ascii"))
        await _command_transact(self._conn, msg)

    async def command(self, cmdline, usepty=False):
        """Run a non-interactive shell command.

        Uses ring buffers to collect outputs to avoid a runaway device command
        from filling host memory. However, this might possibly truncate output.

        Returns:
            stdout (string): output of command
            stderr (string): error output of command
            exitstatus (ExitStatus): the exit status of the command.
        """
        if self.features is None:
            await self.get_features()
        if "shell_v2" not in self.features:
            raise AdbCommandFail("Only shell v2 protocol currently supported.")
        cmdline, name = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, usepty, cmdline)
        sp = ShellProtocol(self._conn.socket)
        stdout = ringbuffer.RingBuffer(MAX_PAYLOAD)
        stderr = ringbuffer.RingBuffer(MAX_PAYLOAD)
        resp = await sp.run(None, stdout, stderr)
        await self._conn.close()
        rc = resp[0]
        if rc & 0x80:
            rc = -(rc & 0x7F)
        return (stdout.read().decode("utf8"),
                stderr.read().decode("utf8"),
                exitstatus.ExitStatus(
                    None,
                    name="{}<EMAIL>(name, self.serial.decode("ascii")),
                    returncode=rc)
                )

    async def list(self, name, coro_cb):
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        await sp.list(name, coro_cb)
        await sp.quit()
        await self._conn.close()

    async def stat(self, path):
        """stat a remote file or directory.

        Return os.stat_result with attributes from remote path.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            st = await sp.stat(path)
        finally:
            await sp.quit()
            await self._conn.close()
        return st

    async def push(self, localfiles: list, remotepath: str, sync: bool = False):
        """Push a list of local files to remote file or directory.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            resp = await sp.push(localfiles, remotepath, sync)
        finally:
            await sp.quit()
            await self._conn.close()
        return resp

    async def pull_file(self, remotepath: str, filelike: typing.BinaryIO):
        """Pull a remote file into given file-like object.
        """
        sp = SyncProtocol(self.serial)
        await sp.connect_with(self._conn)
        try:
            text = await sp.pull_file(remotepath, filelike)
        finally:
            await sp.quit()
            await self._conn.close()
        return text

    async def start(self, cmdline, stdoutstream, stderrstream):
        """Start a process on device with the shell protocol.

        Returns a curio Task object wrapping the ShellProtocol run.
        """
        cmdline, name = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, False, cmdline)
        sp = ShellProtocol(self._conn.socket)
        return await spawn(sp.run(None, stdoutstream, stderrstream))

    async def spawn(self, cmdline):
        """Start a process on device in raw mode.

        Return:
            DeviceProcess with active connection.
        """
        cmdline, name = _fix_command_line(cmdline)
        logging.info("adb.spawn({})".format(cmdline))
        sock = await _start_exec(self.serial, self._conn, cmdline)
        return DeviceProcess(sock, name)

    async def install(self, apkfile, allow_test=True, installer=None,
                      onsdcard=False, onflash=False, allow_downgrade=True,
                      grant_all=True):
        """Install an APK.

        Performs a streaming installation.  Returns True if Success.
        """
        # TODO(dart) other options
        st = os.stat(apkfile)  # TODO(dart) fix potential long blocker
        if not stat.S_ISREG(st.st_mode):
            raise ValueError("The apkfile must be a regular file.")
        cmdline = ["cmd", "package", "install"]
        if allow_test:
            cmdline.append("-t")
        if installer is not None:
            cmdline.extend(["-i", str(installer)])
        if onsdcard:
            cmdline.append("-s")
        if onflash:
            cmdline.append("-f")
        if allow_downgrade:
            cmdline.append("-d")
        if grant_all:
            cmdline.append("-g")
        cmdline.extend(["-S", str(st.st_size)])
        cmdline, name = _fix_command_line(cmdline)
        sock = await _start_exec(self.serial, self._conn, cmdline)
        p = DeviceProcess(sock, name)
        del sock
        async with streams.aopen(apkfile, "rb") as afo:
            await p.copy_from(afo)
        status_response = await p.read(4096)
        await p.close()
        return b'Success' in status_response
    # TODO(dart) install sessions

    async def package(self, cmd, *args, user=None, **kwargs):
        """Manage packages.

        Equivalent of 'pm' command.
        """
        cmdline = ['cmd', 'package']
        if user is not None:
            cmdline.append("--user")
            cmdline.append(str(user))
        cmdline.append(cmd)
        cmdline.extend(str(arg) for arg in args)
        for opt, optarg in kwargs.items():
            cmdline.append("--" + opt)
            if optarg not in (None, True):
                cmdline.append(str(optarg))
        out, err, es = await self.command(cmdline)
        if not es:
            raise AdbCommandFail(err)
        return out

    async def reconnect(self):
        msg = b"host-serial:%b:reconnect" % (self.serial,)
        await _command_transact(self._conn, msg)
        await self._conn.close()
        await self._init()

    async def logcat_clear(self):
        """Clear logcat buffer."""
        stdout, stderr, es = await self.command(["logcat", "-c"])
        if not es:
            raise AdbCommandFail("Didn't clear logcat")

    async def logcat(self, stdoutstream, stderrstream, format="threadtime",
                     buffers="default", modifiers=None, binary=False,
                     regex=None, dump=False, logtags=""):
        """Coroutine for streaming logcat output to the provided file-like
        streams.

        Args:
            stdout, stderr: file-like object to write log events to.
            binary: bool output binary format if True.
            regex: A Perl compatible regular expression to match messages against.
            format: str of one of the following:
                    "brief", "long", "process", "raw", "tag", "thread",
                    "threadtime", "time".
            buffers: list or comma separated string of:
                     'main', 'system', 'radio', 'events', 'crash', 'default' or 'all'
            modifiers: str of one or more of:
                       epoch", "monotonic", "uid", "usec", "UTC", "year", "zone"
            logcats: str of space separated filter expressions.
        """
        logtags = os.environ.get("ANDROID_LOG_TAGS", logtags)
        cmdline = ['exec', 'logcat']
        # buffers
        if isinstance(buffers, str):
            buffers = buffers.split(",")
        for bufname in buffers:
            cmdline.extend(["-b", bufname])
        if binary:
            cmdline.append("-B")
        if regex:
            cmdline.extend(["-e", regex])
        if dump:
            cmdline.append("-d")
        # output format
        if format not in {"brief", "long", "process", "raw", "tag",
                          "thread", "threadtime", "time"}:
            raise ValueError("Bad format type.")
        if modifiers:
            if isinstance(modifiers, str):
                modifiers = modifiers.split(",")
            for modifier in modifiers:
                if modifier in {"epoch", "monotonic", "uid", "usec", "UTC",
                                "year", "zone"}:
                    format += ("," + modifier)
                else:
                    raise ValueError("Invalid logcat format modifier")
        cmdline.extend(["-v", format])
        # logtags
        if logtags:
            logtags = logtags.replace('"', '\\"')
            cmdline.extend(logtags.split())
        # go!
        cmdline, _ = _fix_command_line(cmdline)
        await _start_shell(self.serial, self._conn, False, cmdline)
        sp = ShellProtocol(self._conn.socket)
        await sp.run(None, stdoutstream, stderrstream)


class AndroidDeviceClient:
    """An active adb client per device.

    For synchronous (blocking) style code.
    """
    def __init__(self, serial, host="localhost", port=ADB_PORT):
        self._aadb = _AsyncAndroidDeviceClient(serial, host, port)
        self.open()
        self.features = self.get_features()

    def close(self):
        get_kernel().run(self._aadb.close())

    def open(self):
        get_kernel().run(self._aadb.open())

    @property
    def serial(self):
        return self._aadb.serial

    @property
    def async_client(self):
        return self._aadb

    def get_features(self):
        return get_kernel().run(self._aadb.get_features())

    def get_state(self):
        return get_kernel().run(self._aadb.get_state())

    def reboot(self):
        """Reboot the device."""
        return get_kernel().run(self._aadb.reboot())

    def remount(self):
        """Remount filesystem read-write."""
        return get_kernel().run(self._aadb.remount())

    def root(self):
        """Become root on the device."""
        return get_kernel().run(self._aadb.root())

    def unroot(self):
        """Become non-root on the device."""
        return get_kernel().run(self._aadb.unroot())

    def forward(self, hostport, devport):
        """Tell server to start forwarding TCP ports.
        """
        return get_kernel().run(self._aadb.forward(hostport, devport))

    def kill_forward(self, hostport):
        """Tell server to remove forwarding TCP ports.
        """
        return get_kernel().run(self._aadb.kill_forward(hostport))

    def list_forward(self):
        """Get a list of currently forwarded ports."""
        return get_kernel().run(self._aadb.list_forward())

    def wait_for(self, state: str):
        """Wait for device to be in a particular state.

        State must be one of {"any", "bootloader", "device", "recovery", "sideload"}
        """
        return get_kernel().run(self._aadb.wait_for(state))

    def command(self, cmdline, usepty=False):
        """Run a non-interactive shell command.

        Uses ring buffers to collect outputs to avoid a runaway device command
        from filling host memory. However, this might possibly truncate output.

        Returns:
            stdout (string): output of command
            stderr (string): error output of command
            exitstatus (ExitStatus): the exit status of the command.
        """
        return get_kernel().run(self._aadb.command(cmdline, usepty))

    def spawn(self, cmdline):
        """Start a process on device in raw mode.

        Return:
            DeviceProcess with active connection.
        """
        return get_kernel().run(self._aadb.spawn(cmdline))

    def install(self, apkfile, **kwargs):
        """Install an APK.

        Default flags are best for testing, but you can override. See asyn
        method.
        """
        coro = self._aadb.install(apkfile, **kwargs)
        return get_kernel().run(coro)

    def package(self, cmd, *args, user=None, **kwargs):
        """Manage packages.

        Equivalent of 'pm' command.
        """
        coro = self._aadb.package(cmd, *args, user=user, **kwargs)
        return get_kernel().run(coro)

    def list(self, name, cb):
        """Perform a directory listing.

        Arguments:
            name: str, name of directory
            cb: callable with signature cb(os.stat_result, filename)
        """
        async def acb(stat, path):
            await block_in_thread(cb, stat, path)
        coro = self._aadb.list(name, acb)
        return get_kernel().run(coro)

    def stat(self, path: str):
        """stat a remote file or directory.

        Return os.stat_result with attributes from remote path.
        """
        coro = self._aadb.stat(path)
        return get_kernel().run(coro)

    def push(self, localfiles: list, remotepath: str, sync: bool = False):
        """Push a list of local files to remote file or directory.
        """
        coro = self._aadb.push(localfiles, remotepath, sync)
        return get_kernel().run(coro)

    def pull(self, remotepath, localpath):
        """Pull a single file from device to local file system.
        """
        coro = self._aadb.pull(remotepath, localpath)
        return get_kernel().run(coro)

    def pull_file(self, remotepath: str, filelike: typing.BinaryIO):
        """Pull a file into local memory, as bytes.

        A path to a file on the device.

        """
        coro = self._aadb.pull_file(remotepath, filelike)
        return get_kernel().run(coro)

    def reconnect(self):
        """Reconnect from device side."""
        return get_kernel().run(self._aadb.reconnect())

    def logcat_clear(self):
        """Clear logcat buffer."""
        return get_kernel().run(self._aadb.logcat_clear())

    async def logcat(self, stdoutstream, stderrstream, format="threadtime",
                     buffers="default", modifiers=None, binary=False,
                     regex=None, logtags=""):
        """Coroutine for streaming logcat output to the provided file-like
        streams.
        """
        await self._aadb.logcat(stdoutstream, stderrstream, format=format,
                                buffers=buffers, modifiers=modifiers,
                                binary=binary, regex=regex, logtags=logtags)


class DeviceProcess:
    """Represents an attached process on the device.
    """
    def __init__(self, asocket, name):
        self.socket = asocket
        self.name = name

    def __str__(self):
        return "DeviceProcess running {!r}".format(self.name)

    async def copy_to(self, otherfile):
        """Copy output from this process to another file stream."""
        while True:
            data = await self.socket.recv(MAX_PAYLOAD)
            if not data:
                break
            await otherfile.write(data)

    async def copy_from(self, otherfile):
        """Copy output from another file stream to this process."""
        while True:
            data = await otherfile.read(MAX_PAYLOAD)
            if not data:
                break
            await self.socket.sendall(data)

    async def read(self, amt):
        return await self.socket.recv(amt)

    async def write(self, data):
        return await self.socket.sendall(data)

    async def close(self):
        if self.socket is not None:
            await self.socket.close()
            self.socket = None

    def sync_close(self):
        return get_kernel().run(self.close)


def _fix_command_line(cmdline):
    """Fix the command.

    If a list, quote the components if required.
    Return encoded command line as bytes and the command base name.
    """
    if isinstance(cmdline, list):
        name = cmdline[0]
        cmdline = " ".join('"{}"'.format(s) if " " in s else str(s) for s in cmdline)
    else:
        name = cmdline.split()[0]
    return cmdline.encode("utf8"), name


# Perform shell request, connection stays open
async def _start_shell(serial, conn, usepty, cmdline):
    tpmsg = b"host:transport:%b" % serial
    msg = b"shell,%b:%b" % (b",".join([b"v2",
                                       b"TERM=%b" % os.environb[b"TERM"],
                                       b"pty" if usepty else b"raw"]),
                            cmdline)
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)


async def _start_exec(serial, conn, cmdline):
    tpmsg = b"host:transport:%b" % serial
    msg = b"exec:%b" % (cmdline,)
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)
    sock = conn.socket.dup()
    await conn.close()
    return sock


# Send command to specific device with orderly shutdown
async def _connect_command(serial, conn, msg):
    tpmsg = b"host:transport:%b" % serial
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(msg, expect_response=False)
    await conn.close()


# Root command transaction is special since device adbd restarts.
async def _special_command(serial, conn, cmd):
    tpmsg = b"host:transport:%b" % serial
    await conn.open()
    await conn.message(tpmsg, expect_response=False)
    await conn.message(cmd, expect_response=False)
    resp = await conn.socket.recv(4096)
    await conn.close()
    return resp


class ShellProtocol:
    """Implement the shell protocol v2."""
    IDSTDIN = 0
    IDSTDOUT = 1
    IDSTDERR = 2
    IDEXIT = 3
    CLOSESTDIN = 4
    # Window size change (an ASCII version of struct winsize).
    WINDOWSIZECHANGE = 5
    # Indicates an invalid or unknown packet.
    INVALID = 255

    def __init__(self, socket):
        self._sock = socket
        self._header = struct.Struct(b"<BI")
        self._winsize = None

    async def run(self, inbuf, outbuf, errbuf):
        while True:
            tl = await self._sock.recv(5)
            msgtype, msglen = self._header.unpack(tl)
            if msgtype == 0:
                pass
            elif msgtype == 1:
                data = await self._sock.recv(msglen)
                outbuf.write(data)
            elif msgtype == 2:
                data = await self._sock.recv(msglen)
                errbuf.write(data)
            elif msgtype == 3:
                data = await self._sock.recv(msglen)
                return data
            elif msgtype == 4:
                pass  # TODO
            elif msgtype == 5:
                data = await self._sock.recv(msglen)
                self._winsize = data
            else:
                raise AdbProtocolError("Unhandled shell protocol message type.")


class SyncProtocol:
    """Implementation of Android SYNC protocol.

    Only recent devices are supported.
    """

    mkid = lambda code: int.from_bytes(code, byteorder='little')  # noqa
    ID_LSTAT_V1 = mkid(b'STAT')
    ID_STAT_V2 = mkid(b'STA2')
    ID_LSTAT_V2 = mkid(b'LST2')
    ID_LIST = mkid(b'LIST')
    ID_SEND = mkid(b'SEND')
    ID_RECV = mkid(b'RECV')
    ID_DENT = mkid(b'DENT')
    ID_DONE = mkid(b'DONE')
    ID_DATA = mkid(b'DATA')
    ID_OKAY = mkid(b'OKAY')
    ID_FAIL = mkid(b'FAIL')
    ID_QUIT = mkid(b'QUIT')
    del mkid

    SYNC_DATA_MAX = 65536
    SYNCMSG_DATA = struct.Struct("<II")  # id, size
    # id; mode; size; time; namelen;
    SYNCMSG_DIRENT = struct.Struct("<IIIII")
    # id; error; dev; ino; mode; nlink; uid; gid; size; atime; mtime; ctime;
    SYNCMSG_STAT_V2 = struct.Struct("<IIQQIIIIQqqq")
    SYNCMSG_STATUS = struct.Struct("<II")  # id, msglen

    def __init__(self, serial):
        self.serial = serial
        self.socket = None

    async def connect_with(self, adbconnection):
        tpmsg = b"host:transport:%b" % self.serial
        msg = b"sync:"
        await adbconnection.open()
        await adbconnection.message(tpmsg, expect_response=False)
        await adbconnection.message(msg, expect_response=False)
        self.socket = adbconnection.socket

    async def quit(self):
        msg = SyncProtocol.SYNCMSG_STATUS.pack(SyncProtocol.ID_QUIT, 0)
        self.socket.sendall(msg)

    async def send_request(self, protoid: int, path_and_mode: str):
        path_and_mode = path_and_mode.encode("utf-8")
        length = len(path_and_mode)
        if length > 1024:
            raise ValueError("Can't send message > 1024")
        hdr = SyncProtocol.SYNCMSG_DATA.pack(protoid, length)
        await self.socket.sendall(hdr + path_and_mode)

    async def list(self, path, cb_coro):
        """List a directory on device."""
        await self.send_request(SyncProtocol.ID_LIST, path)
        while True:
            resp = await self.socket.recv(SyncProtocol.SYNCMSG_DIRENT.size)
            msgid, mode, size, time, namelen = SyncProtocol.SYNCMSG_DIRENT.unpack(resp)
            if msgid == SyncProtocol.ID_DONE:
                return True
            if msgid != SyncProtocol.ID_DENT:
                return False
            name = await self.socket.recv(namelen)
            stat = os.stat_result((mode, None, None, None, 0, 0, size, None, float(time), None))
            await cb_coro(stat, name.decode("utf-8"))

    async def stat(self, remotepath):
        """Stat a path."""
        s = SyncProtocol.SYNCMSG_STAT_V2
        await self.send_request(SyncProtocol.ID_STAT_V2, remotepath)
        resp = await self.socket.recv(s.size)
        stat_id, err, dev, ino, mode, nlink, uid, gid, size, atime, mtime, ctime = s.unpack(resp)
        if stat_id == SyncProtocol.ID_STAT_V2:
            if err != 0:
                raise OSError(err, errno.errorcode[err], remotepath)
            sr = os.stat_result((mode, ino, dev, nlink, uid, gid, size,
                                 float(atime), float(mtime), float(ctime),  # floats
                                 # int nanoseconds, but not really
                                 atime * 1e9, mtime * 1e9, ctime * 1e9))
            return sr
        else:
            raise AdbProtocolError("SyncProtocol: invalid response type.")

    async def push(self, localfiles, remotepath, sync=False):
        """Push files to device destination."""
        try:
            dest_st = await self.stat(remotepath)
        except FileNotFoundError:
            dst_exists = False
            dst_isdir = False
        else:
            dst_exists = True
            if stat.S_ISDIR(dest_st.st_mode):
                dst_isdir = True
            elif stat.S_ISREG(dest_st.st_mode):
                dst_isdir = False
            else:
                raise ValueError("push: destination is not a directory or "
                                 "regular file")
        if not dst_isdir:
            if len(localfiles) > 1:
                raise ValueError("push: destination is not a dir when copying "
                                 "multiple files.")
            if dst_exists:
                raise ValueError("push: destination exists")
        for localfile in localfiles:
            local_st = os.stat(localfile)

            if stat.S_ISDIR(local_st.st_mode):
                rpath = os.path.join(remotepath, os.path.basename(localfile))
                await self._copy_local_dir_remote(localfile, rpath, sync)

            if stat.S_ISREG(local_st.st_mode):
                if dst_isdir:
                    rpath = os.path.join(remotepath, os.path.basename(localfile))
                # If synchronize requested, just stat remote and return if size
                # and mtime are equal.
                if sync:
                    try:
                        dst_stat = await self.stat(rpath)
                    except OSError:
                        pass
                    else:
                        if (local_st.st_size == dst_stat.st_size and
                                local_st.st_mtime == dst_stat.st_mtime):
                            return
                await self._sync_send(localfile.encode("utf8"),
                                      rpath.encode("utf8"),
                                      local_st)

    async def pull(self, remotepath, localpath):
        with open(localpath, "wb") as fo:
            await self.pull_file(remotepath, fo)
            # TODO(dart) directories and symlinks

    async def pull_file(self, remotepath, filelike):
        smd = SyncProtocol.SYNCMSG_DATA
        src_st = await self.stat(remotepath)
        if stat.S_ISREG(src_st.st_mode):
            await self.send_request(SyncProtocol.ID_RECV, remotepath)
            while True:
                resp = await self.socket.recv(smd.size)
                msg_id, msg_len = smd.unpack(resp)
                if msg_id == SyncProtocol.ID_DATA:
                    data = await self.socket.recv(msg_len)
                    filelike.write(data)
                elif msg_id == SyncProtocol.ID_DONE:
                    break
                else:
                    raise AdbProtocolError(
                        "SyncProtocol: pull: can't handle message: {!r}.".format(msg_id))
        elif stat.S_ISDIR(src_st.st_mode):
            raise NotImplementedError("TODO(dart)")
        else:
            raise NotImplementedError("TODO(dart)")

    async def _sync_send(self, src_path, dst_path, local_st):
        if stat.S_ISLNK(local_st.st_mode):
            raise NotImplementedError("TODO(dart)")
        if local_st.st_size < SyncProtocol.SYNC_DATA_MAX:
            with open(src_path, "rb") as fo:
                data = fo.read()
            path_and_mode = b"%s,%d" % (dst_path, local_st.st_mode)
            return await self._send_small_file(path_and_mode, data, int(local_st.st_mtime))
        else:
            return await self._send_large_file(src_path, dst_path, local_st)

    async def _send_small_file(self, path_and_mode, data, mtime):
        sm = SyncProtocol.SYNCMSG_DATA
        buf = ringbuffer.RingBuffer(SyncProtocol.SYNC_DATA_MAX << 1)  # big enough to not wrap
        buf.write(sm.pack(SyncProtocol.ID_SEND, len(path_and_mode)))
        buf.write(path_and_mode)
        buf.write(sm.pack(SyncProtocol.ID_DATA, len(data)))
        buf.write(data)
        buf.write(sm.pack(SyncProtocol.ID_DONE, mtime))
        await self.socket.sendall(buf.read())
        return await self._copy_done()

    async def _send_large_file(self, src_path, dst_path, local_st):
        sm = SyncProtocol.SYNCMSG_DATA
        buf = ringbuffer.RingBuffer(SyncProtocol.SYNC_DATA_MAX << 1)
        path_and_mode = b"%s,%d" % (dst_path, local_st.st_mode)
        buf.write(sm.pack(SyncProtocol.ID_SEND, len(path_and_mode)))
        buf.write(path_and_mode)
        await self.socket.sendall(buf.read())

        buf.clear()
        chunksize = SyncProtocol.SYNC_DATA_MAX - sm.size
        with open(src_path, "rb") as fo:
            while True:
                chunk = fo.read(chunksize)
                if not chunk:
                    break
                buf.write(sm.pack(SyncProtocol.ID_DATA, len(chunk)))
                buf.write(chunk)
                await self.socket.sendall(buf.read())

        buf.clear()
        buf.write(sm.pack(SyncProtocol.ID_DONE, int(local_st.st_mtime)))
        await self.socket.sendall(buf.read())
        return await self._copy_done()

    async def _copy_done(self):
        sm = SyncProtocol.SYNCMSG_STATUS
        raw = await self.socket.recv(sm.size)
        msg_id, msg_len = sm.unpack(raw)
        if msg_id == SyncProtocol.ID_OKAY:
            return True
        if msg_id == SyncProtocol.ID_FAIL:
            msg = await self.socket.recv(msg_len)
            raise AdbCommandFail(
                "large file copy not OKAY: {!r}".format(msg.decode("utf8")))

    async def _copy_local_dir_remote(self, localfile, remotepath, sync):
        raise NotImplementedError("TODO(dart)")


class LogcatHandler:
    """Host side logcat handler that receives logcat messages in binary mode
    over raw connection.
    """

    LOGCAT_MESSAGE = struct.Struct("<HHiIIIII")  # logger_entry_v4
    #  uint16_t len;       length of the payload
    #  uint16_t hdr_size;  sizeof(struct logger_entry_v4)
    #  int32_t pid;        generating process's pid
    #  uint32_t tid;       generating process's tid
    #  uint32_t sec;       seconds since Epoch
    #  uint32_t nsec;      nanoseconds
    #  uint32_t lid;       log id of the payload, bottom 4 bits currently
    #  uint32_t uid;       generating process's uid
    #  char msg[0];        the entry's payload

    def __init__(self, aadb):
        self._aadb = aadb

    def clear(self):
        """Clear logcat buffers."""
        return get_kernel().run(self._aadb.logcat_clear())

    def dump(self):
        """Dump logs to stdout until interrupted."""
        return get_kernel().run(self._dump())

    async def _dump(self):
        ac = self._aadb
        signalset = SignalEvent(signal.SIGINT, signal.SIGTERM)
        await ac.logcat_clear()
        proc = await ac.spawn('logcat -B -b default')
        task = await spawn(self._read_and_dump(proc, streams.FileStream(sys.stdout.buffer)))
        await signalset.wait()
        await task.cancel()
        await proc.close()

    async def _read_and_dump(self, proc, out):
        while True:
            lm = await self._read_one(proc)
            await out.write(str(lm).encode("utf8"))
            await out.write(b'\n')

    async def _read_one(self, proc):
        s = self.LOGCAT_MESSAGE
        rawhdr = await proc.read(s.size)
        payload_len, hdr_size, pid, tid, sec, nsec, lid, uid = s.unpack(rawhdr)
        payload = await proc.read(payload_len)
        return logcat.LogcatMessage(pid, tid, sec, nsec, lid, uid, payload)

    def dump_to(self, localfile, logtags=None):
        """Dump all current logs to a file, in binary format."""
        return get_kernel().run(self._dump_to(localfile, logtags))

    async def _dump_to(self, localfile, logtags):
        cmdline = ['logcat', '-d', '-B', '-b', 'default']
        if logtags:
            cmdline.extend(logtags.split())
        proc = await self._aadb.spawn(cmdline)
        with open(localfile, "ab") as fo:
            await proc.copy_to(streams.FileStream(fo))
        await proc.close()

    def watch_for(self, tag=None, priority=None, text=None, timeout=90):
        """Watch for first occurence of a particular set of tag, priority, or
        message text.

        If tag is given watch for first of that tag.
        If tag and priority is given watch for tag only with that priority.
        if tag and text is given watch for tag with the given text in the
        message part.
        If text is given look for first occurence of text in message part.
        """
        if tag is None and text is None:
            raise ValueError("watch_for: must supply one or both of 'tag' or "
                             "'text' parameters.")
        return get_kernel().run(self._watch_for(tag, priority, text, timeout))

    async def _watch_for(self, tag, priority, text, timeout):
        lm = None
        ac = self._aadb
        await ac.logcat_clear()
        proc = await ac.spawn('logcat -B -b default')
        try:
            async with timeout_after(timeout):
                while True:
                    new = await self._read_one(proc)
                    if tag and tag == new.tag:
                        if priority is not None:
                            if new.priority == priority:
                                lm = new
                                break
                        else:
                            lm = new
                            break
                        if text is not None:
                            if text in new.message:
                                lm = new
                                break
                    if text and text in new.message:
                        lm = new
                        break
        except TaskTimeout:
            pass
        await proc.close()
        return lm


class AndroidDevice:
    """Information about attached Android device.

    No connection necessary.
    """

    def __init__(self, serial, product, model, device, state):
        self.type = "phone"
        self.serial = serial
        self.product = product
        self.model = model
        self.device = device
        self.state = state

    def __repr__(self):
        return ("AndroidDevice("
                "serial={!r}, product={!r}, model={!r}, device={!r}, state={!r})".format(
                    self.serial, self.product, self.model, self.device, self.state))


# b'HTxxxserial device usb:1-1.1 product:marlin model:Pixel_XL device:marlin\n'
def _device_factory(line):
    parts = line.decode("ascii").split()
    serial = parts[0]
    state = parts[1]
    if state == "device":
        product = parts[3].partition(":")[2]
        model = parts[4].partition(":")[2]
        device = parts[5].partition(":")[2]
        return AndroidDevice(serial, product, model, device, state)
    else:
        return AndroidDevice(serial, None, None, None, state)


if __name__ == "__main__":
    from devtest import debugger
    debugger.autodebug()
    start_server()
    print("Test AdbClient:")
    c = AdbClient()
    print("  Server version:", c.server_version)
    for devinfo in c.get_device_list():
        print("    ", devinfo)
    print("Forwards:", c.list_forward())
    c.close()
    del c

    print("Test AndroidDeviceClient:")
    ac = AndroidDeviceClient(devinfo.serial)
    ac.wait_for("device")
    print("  features:", ac.features)
    print("  state:", ac.get_state())
    print("  running 'ls /sdcard':")
    stdout, stderr, es = ac.command(["ls", "/sdcard"])
    print("    ", es)
    print("    stdout:", repr(stdout))
    print("    stderr:", repr(stderr))
    print("forward list:")
    print(repr(ac.list_forward()))
    ac.close()
    del ac

    # Test async with logcat. ^C to stop it.
    async def dostuff():
        ac = await AsyncAndroidDeviceClient(devinfo.serial)

        stdout, stderr, es = await ac.command(["ls", "/sdcard"])
        print("    ", es)
        print("    stdout:", repr(stdout))
        print("    stderr:", repr(stderr))

        fl = await ac.list_forward()
        print("Forward list:")
        print(repr(fl))
        print("Logcat:")
        signalset = SignalEvent(signal.SIGINT, signal.SIGTERM)
        await ac.wait_for("device")
        try:
            await ac.logcat_clear()
            task = await spawn(ac.logcat(sys.stdout.buffer, sys.stdout.buffer,
                                         buffers="kernel,main",
                                         format="long", modifiers="epoch,usec",
                                         binary=False,
                                         logtags=" ".join(sys.argv[1:])))
            await signalset.wait()
            await task.cancel()
        finally:
            await ac.close()
        await ac.close()

    kern = get_kernel()
    kern.run(dostuff)

# vim:ts=4:sw=4:softtabstop=4:smarttab:expandtab

# 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. Native adb client-to-server protocol implementation.  This is an async implementation, suitable for including in an event loop. An error in the protocol was detected. An error indicated by the server. Run the adb binary. Asynchronous adb socket wrapper. # On the host: 1st OKAY is connect, 2nd OKAY is status. An adb client, synchronous.      For general host side operations. Get a AndroidDeviceClient instance. Get the current state of a device.          Arguments:             serial: str of the device serial number.          Returns:             one of {"device", "unauthorized", "bootloader"}             or None if serial not found. Get list of attached devices.          Returns:             list of AndroidDevice instances. Tell server to start forwarding TCP ports. Return a list of currently forwarded ports.          Returns:             Tuple of (serial, host_port, device_port). # <serial> " " <local> " " <remote> "\n" The server's version number. Get initialized _AsyncAndroidDeviceClient object from sync code. Get initialized _AsyncAndroidDeviceClient instance from async code. An active adb client per device.      For device specific operations.     For use in asynchronous event loops. # adbd needs some time to initialize # adbd needs some time to initialize Tell server to start forwarding TCP ports. Tell server to remove forwarding TCP ports. Tell server to remove all forwarding TCP ports. Return a list of currently forwarded ports.          Returns:             Tuple of (host_port, device_port). # <serial> " " <local> " " <remote> "\n" Wait for device to be in a particular state.          State must be one of {"any", "bootloader", "device", "recovery", "sideload"} Run a non-interactive shell command.          Uses ring buffers to collect outputs to avoid a runaway device command         from filling host memory. However, this might possibly truncate output.          Returns:             stdout (string): output of command             stderr (string): error output of command             exitstatus (ExitStatus): the exit status of the command. stat a remote file or directory.          Return os.stat_result with attributes from remote path. Push a list of local files to remote file or directory. Pull a remote file into given file-like object. Start a process on device with the shell protocol.          Returns a curio Task object wrapping the ShellProtocol run. Start a process on device in raw mode.          Return:             DeviceProcess with active connection. Install an APK.          Performs a streaming installation.  Returns True if Success. # TODO(dart) other options # TODO(dart) fix potential long blocker # TODO(dart) install sessions Manage packages.          Equivalent of 'pm' command. Clear logcat buffer. Coroutine for streaming logcat output to the provided file-like         streams.          Args:             stdout, stderr: file-like object to write log events to.             binary: bool output binary format if True.             regex: A Perl compatible regular expression to match messages against.             format: str of one of the following:                     "brief", "long", "process", "raw", "tag", "thread",                     "threadtime", "time".             buffers: list or comma separated string of:                      'main', 'system', 'radio', 'events', 'crash', 'default' or 'all'             modifiers: str of one or more of:                        epoch", "monotonic", "uid", "usec", "UTC", "year", "zone"             logcats: str of space separated filter expressions. # buffers # output format # logtags # go! An active adb client per device.      For synchronous (blocking) style code. Reboot the device. Remount filesystem read-write. Become root on the device. Become non-root on the device. Tell server to start forwarding TCP ports. Tell server to remove forwarding TCP ports. Get a list of currently forwarded ports. Wait for device to be in a particular state.          State must be one of {"any", "bootloader", "device", "recovery", "sideload"} Run a non-interactive shell command.          Uses ring buffers to collect outputs to avoid a runaway device command         from filling host memory. However, this might possibly truncate output.          Returns:             stdout (string): output of command             stderr (string): error output of command             exitstatus (ExitStatus): the exit status of the command. Start a process on device in raw mode.          Return:             DeviceProcess with active connection. Install an APK.          Default flags are best for testing, but you can override. See asyn         method. Manage packages.          Equivalent of 'pm' command. Perform a directory listing.          Arguments:             name: str, name of directory             cb: callable with signature cb(os.stat_result, filename) stat a remote file or directory.          Return os.stat_result with attributes from remote path. Push a list of local files to remote file or directory. Pull a single file from device to local file system. Pull a file into local memory, as bytes.          A path to a file on the device. Reconnect from device side. Clear logcat buffer. Coroutine for streaming logcat output to the provided file-like         streams. Represents an attached process on the device. Copy output from this process to another file stream. Copy output from another file stream to this process. Fix the command.      If a list, quote the components if required.     Return encoded command line as bytes and the command base name. # Perform shell request, connection stays open # Send command to specific device with orderly shutdown # Root command transaction is special since device adbd restarts. Implement the shell protocol v2. # Window size change (an ASCII version of struct winsize). # Indicates an invalid or unknown packet. # TODO Implementation of Android SYNC protocol.      Only recent devices are supported. # noqa # id, size # id; mode; size; time; namelen; # id; error; dev; ino; mode; nlink; uid; gid; size; atime; mtime; ctime; # id, msglen List a directory on device. Stat a path. # floats # int nanoseconds, but not really Push files to device destination. # If synchronize requested, just stat remote and return if size # and mtime are equal. # TODO(dart) directories and symlinks # big enough to not wrap Host side logcat handler that receives logcat messages in binary mode     over raw connection. # logger_entry_v4 #  uint16_t len;       length of the payload #  uint16_t hdr_size;  sizeof(struct logger_entry_v4) #  int32_t pid;        generating process's pid #  uint32_t tid;       generating process's tid #  uint32_t sec;       seconds since Epoch #  uint32_t nsec;      nanoseconds #  uint32_t lid;       log id of the payload, bottom 4 bits currently #  uint32_t uid;       generating process's uid #  char msg[0];        the entry's payload Clear logcat buffers. Dump logs to stdout until interrupted. Dump all current logs to a file, in binary format. Watch for first occurence of a particular set of tag, priority, or         message text.          If tag is given watch for first of that tag.         If tag and priority is given watch for tag only with that priority.         if tag and text is given watch for tag with the given text in the         message part.         If text is given look for first occurence of text in message part. Information about attached Android device.      No connection necessary. # b'HTxxxserial device usb:1-1.1 product:marlin model:Pixel_XL device:marlin\n' # Test async with logcat. ^C to stop it. # vim:ts=4:sw=4:softtabstop=4:smarttab:expandtab