• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15
16"""Interface for controller-specific Pandora server management."""
17
18import asyncio
19import avatar.aio
20import grpc
21import grpc.aio
22import portpicker
23import threading
24import types
25
26from avatar.controllers import bumble_device
27from avatar.controllers import pandora_device
28from avatar.controllers import usb_bumble_device
29from avatar.pandora_client import BumblePandoraClient
30from avatar.pandora_client import PandoraClient
31from bumble import pandora as bumble_server
32from bumble.pandora.device import PandoraDevice as BumblePandoraDevice
33from contextlib import suppress
34from mobly.controllers import android_device
35from mobly.controllers.android_device import AndroidDevice
36from typing import Generic, Optional, TypeVar
37
38ANDROID_SERVER_PACKAGE = 'com.android.pandora'
39ANDROID_SERVER_GRPC_PORT = 8999
40
41
42# Generic type for `PandoraServer`.
43TDevice = TypeVar('TDevice')
44
45
46class PandoraServer(Generic[TDevice]):
47    """Abstract interface to manage the Pandora gRPC server on the device."""
48
49    MOBLY_CONTROLLER_MODULE: types.ModuleType = pandora_device
50
51    device: TDevice
52
53    def __init__(self, device: TDevice) -> None:
54        """Creates a PandoraServer.
55
56        Args:
57            device: A Mobly controller instance.
58        """
59        self.device = device
60
61    def start(self) -> PandoraClient:
62        """Sets up and starts the Pandora server on the device."""
63        assert isinstance(self.device, PandoraClient)
64        return self.device
65
66    def stop(self) -> None:
67        """Stops and cleans up the Pandora server on the device."""
68
69
70class BumblePandoraServer(PandoraServer[BumblePandoraDevice]):
71    """Manages the Pandora gRPC server on a BumbleDevice."""
72
73    MOBLY_CONTROLLER_MODULE = bumble_device
74
75    _task: Optional[asyncio.Task[None]] = None
76
77    def start(self) -> BumblePandoraClient:
78        """Sets up and starts the Pandora server on the Bumble device."""
79        assert self._task is None
80
81        # set the event loop to make sure the gRPC server use the avatar one.
82        asyncio.set_event_loop(avatar.aio.loop)
83
84        # create gRPC server & port.
85        server = grpc.aio.server()
86        port = server.add_insecure_port(f'localhost:{0}')
87
88        config = bumble_server.Config()
89        self._task = avatar.aio.loop.create_task(
90            bumble_server.serve(self.device, config=config, grpc_server=server, port=port)
91        )
92
93        return BumblePandoraClient(f'localhost:{port}', self.device, config)
94
95    def stop(self) -> None:
96        """Stops and cleans up the Pandora server on the Bumble device."""
97
98        async def server_stop() -> None:
99            assert self._task is not None
100            if not self._task.done():
101                self._task.cancel()
102                with suppress(asyncio.CancelledError):
103                    await self._task
104            self._task = None
105
106        avatar.aio.run_until_complete(server_stop())
107
108
109class UsbBumblePandoraServer(BumblePandoraServer):
110    MOBLY_CONTROLLER_MODULE = usb_bumble_device
111
112
113class AndroidPandoraServer(PandoraServer[AndroidDevice]):
114    """Manages the Pandora gRPC server on an AndroidDevice."""
115
116    MOBLY_CONTROLLER_MODULE = android_device
117
118    _instrumentation: Optional[threading.Thread] = None
119    _port: int
120
121    def start(self) -> PandoraClient:
122        """Sets up and starts the Pandora server on the Android device."""
123        assert self._instrumentation is None
124
125        # start Pandora Android gRPC server.
126        self._port = portpicker.pick_unused_port()  # type: ignore
127        self._instrumentation = threading.Thread(
128            target=lambda: self.device.adb._exec_adb_cmd(  # type: ignore
129                'shell',
130                f'am instrument --no-hidden-api-checks -w {ANDROID_SERVER_PACKAGE}/.Main',
131                shell=False,
132                timeout=None,
133                stderr=None,
134            )
135        )
136
137        self._instrumentation.start()
138        self.device.adb.forward([f'tcp:{self._port}', f'tcp:{ANDROID_SERVER_GRPC_PORT}'])  # type: ignore
139
140        return PandoraClient(f'localhost:{self._port}', 'android')
141
142    def stop(self) -> None:
143        """Stops and cleans up the Pandora server on the Android device."""
144        assert self._instrumentation is not None
145
146        # Stop Pandora Android gRPC server.
147        self.device.adb._exec_adb_cmd(  # type: ignore
148            'shell', f'am force-stop {ANDROID_SERVER_PACKAGE}', shell=False, timeout=None, stderr=None
149        )
150
151        self.device.adb.forward(['--remove', f'tcp:{self._port}'])  # type: ignore
152        self._instrumentation.join()
153        self._instrumentation = None
154