[avatar_experimental] - Copy avatar from external

// Copyright 2022, The Android Open Source Project

//

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

package {

    default_applicable_licenses: ["Android-Apache-2.0"],

}

python_library_host {

    name: "libavatar_experimental",

    srcs: [

        "avatar/*.py",

        "avatar/bumble_server/*.py",

        "avatar/controllers/*.py",

    ],

    libs: [

        "mobly",

        "pandora_experimental-python",

        "libprotobuf-python",

        "bumble",

    ],

}

# Copyright 2022 Google LLC

#

# 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

#

#     https://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.

"""

Avatar is a scalable multi-platform Bluetooth testing tool capable of running

any Bluetooth test cases virtually and physically.

"""

__version__ = "0.0.1"

# Copyright 2022 Google LLC

#

# 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

#

#     https://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.

import time

import threading

from mobly.controllers.android_device_lib.services.base_service \

    import BaseService

ANDROID_SERVER_PACKAGE = 'com.android.pandora'

ANDROID_SERVER_GRPC_PORT = 8999

class AndroidService(BaseService):

    def __init__(self, device, configs=None):

        super().__init__(device, configs)

        self.port = configs['port']

        self._is_alive = False

    @property

    def is_alive(self):

        return self._is_alive

    def start(self):

        # Start Pandora Android gRPC server.

        self.instrumentation = threading.Thread(target=lambda: self._device.adb._exec_adb_cmd(

            'shell',

            f'am instrument --no-hidden-api-checks -w {ANDROID_SERVER_PACKAGE}/.Main',

            shell=False,

            timeout=None,

            stderr=None))

        self.instrumentation.start()

        self._device.adb.forward([f'tcp:{self.port}', f'tcp:{ANDROID_SERVER_GRPC_PORT}'])

        # Wait a few seconds for the Android gRPC server to be started.

        time.sleep(3)

        self._is_alive = True

    def stop(self):

        # Stop Pandora Android gRPC server.

        self._device.adb._exec_adb_cmd('shell',

                                       f'am force-stop {ANDROID_SERVER_PACKAGE}',

                                       shell=False,

                                       timeout=None,

                                       stderr=None)

        self._device.adb.forward(['--remove', f'tcp:{self.port}'])

        self.instrumentation.join()

        self._is_alive = False

# Copyright 2022 Google LLC

#

# 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

#

#     https://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.

"""Pandora Bumble Server."""

__version__ = "0.0.1"

import asyncio

import logging

import os

import grpc

from bumble.host import Host

from bumble.device import Device, DeviceConfiguration

from bumble.transport import open_transport

from bumble.a2dp import make_audio_sink_service_sdp_records

from pandora_experimental.host_grpc import add_HostServicer_to_server

from .host import HostService

BUMBLE_SERVER_PORT = 7999

ROOTCANAL_PORT_CUTTLEFISH = 7300

current_dir = os.path.dirname(os.path.realpath(__file__))

class BumblePandoraServer:

    def __init__(self, grpc_port, hci, config):

        self.hci = hci

        device_config = DeviceConfiguration()

        device_config.load_from_dict(config)

        host = Host(controller_source=hci.source, controller_sink=hci.sink)

        self.device = Device(config=device_config, host=host)

        self.device.classic_enabled = config.get('classic_enabled', False)

        self.server = grpc.aio.server()

        add_HostServicer_to_server(HostService(self.device), self.server)

        self.grpc_port = self.server.add_insecure_port(f'localhost:{grpc_port}')

    @classmethod

    async def open(cls, grpc_port, transport_name, config):

        hci = await open_transport(transport_name)

        return cls(grpc_port=grpc_port, hci=hci, config=config)

    async def start(self):

        await self.device.power_on()

        await self.server.start()

    async def wait_for_termination(self):

        await self.server.wait_for_termination()

    async def close(self):

        await self.server.stop(None)

        await self.hci.close()

async def serve():

    transport = f'tcp-client:127.0.0.1:{ROOTCANAL_PORT_CUTTLEFISH}'

    server = await BumblePandoraServer.open(BUMBLE_SERVER_PORT, transport, {'classic_enabled': True})

    await server.start()

    await server.wait_for_termination()

    await server.close()

if __name__ == '__main__':

    logging.basicConfig(level=logging.DEBUG)

    asyncio.run(serve())

{

    "name": "Bumble",

    "class_of_device": 2360324,

    "keystore": "JsonKeyStore"

}

 No newline at end of file