• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3# Copyright (C) 2016 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6# use this file except in compliance with the License. You may obtain a copy of
7# the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations under
15# the License.
16import inspect
17import time
18from acts import asserts
19from acts.controllers.buds_lib.dev_utils import apollo_sink_events
20from acts_contrib.test_utils.bt.bt_constants import bt_default_timeout
21
22
23
24def validate_controller(controller, abstract_device_class):
25    """Ensure controller has all methods in abstract_device_class.
26    Also checks method signatures to ensure parameters are satisfied.
27
28    Args:
29        controller: instance of a device controller.
30        abstract_device_class: class definition of an abstract_device interface.
31    Raises:
32         NotImplementedError: if controller is missing one or more methods.
33    """
34    ctlr_methods = inspect.getmembers(controller, predicate=callable)
35    reqd_methods = inspect.getmembers(
36        abstract_device_class, predicate=inspect.ismethod)
37    expected_func_names = {method[0] for method in reqd_methods}
38    controller_func_names = {method[0] for method in ctlr_methods}
39
40    if not controller_func_names.issuperset(expected_func_names):
41        raise NotImplementedError(
42            'Controller {} is missing the following functions: {}'.format(
43                controller.__class__.__name__,
44                repr(expected_func_names - controller_func_names)))
45
46    for func_name in expected_func_names:
47        controller_func = getattr(controller, func_name)
48        required_func = getattr(abstract_device_class, func_name)
49        required_signature = inspect.signature(required_func)
50        if inspect.signature(controller_func) != required_signature:
51            raise NotImplementedError(
52                'Method {} must have the signature {}{}.'.format(
53                    controller_func.__qualname__, controller_func.__name__,
54                    required_signature))
55
56
57class BluetoothHandsfreeAbstractDevice:
58    """Base class for all Bluetooth handsfree abstract devices.
59
60    Desired controller classes should have a corresponding Bluetooth handsfree
61    abstract device class defined in this module.
62    """
63
64    @property
65    def mac_address(self):
66        raise NotImplementedError
67
68    def accept_call(self):
69        raise NotImplementedError()
70
71    def end_call(self):
72        raise NotImplementedError()
73
74    def enter_pairing_mode(self):
75        raise NotImplementedError()
76
77    def next_track(self):
78        raise NotImplementedError()
79
80    def pause(self):
81        raise NotImplementedError()
82
83    def play(self):
84        raise NotImplementedError()
85
86    def power_off(self):
87        raise NotImplementedError()
88
89    def power_on(self):
90        raise NotImplementedError()
91
92    def previous_track(self):
93        raise NotImplementedError()
94
95    def reject_call(self):
96        raise NotImplementedError()
97
98    def volume_down(self):
99        raise NotImplementedError()
100
101    def volume_up(self):
102        raise NotImplementedError()
103
104
105class PixelBudsBluetoothHandsfreeAbstractDevice(
106        BluetoothHandsfreeAbstractDevice):
107
108    CMD_EVENT = 'EvtHex'
109
110    def __init__(self, pixel_buds_controller):
111        self.pixel_buds_controller = pixel_buds_controller
112
113    def format_cmd(self, cmd_name):
114        return self.CMD_EVENT + ' ' + apollo_sink_events.SINK_EVENTS[cmd_name]
115
116    @property
117    def mac_address(self):
118        return self.pixel_buds_controller.bluetooth_address
119
120    def accept_call(self):
121        return self.pixel_buds_controller.cmd(
122            self.format_cmd('EventUsrAnswer'))
123
124    def end_call(self):
125        return self.pixel_buds_controller.cmd(
126            self.format_cmd('EventUsrCancelEnd'))
127
128    def enter_pairing_mode(self):
129        return self.pixel_buds_controller.set_pairing_mode()
130
131    def next_track(self):
132        return self.pixel_buds_controller.cmd(
133            self.format_cmd('EventUsrAvrcpSkipForward'))
134
135    def pause(self):
136        return self.pixel_buds_controller.cmd(
137            self.format_cmd('EventUsrAvrcpPause'))
138
139    def play(self):
140        return self.pixel_buds_controller.cmd(
141            self.format_cmd('EventUsrAvrcpPlay'))
142
143    def power_off(self):
144        return self.pixel_buds_controller.power('Off')
145
146    def power_on(self):
147        return self.pixel_buds_controller.power('On')
148
149    def previous_track(self):
150        return self.pixel_buds_controller.cmd(
151            self.format_cmd('EventUsrAvrcpSkipBackward'))
152
153    def reject_call(self):
154        return self.pixel_buds_controller.cmd(
155            self.format_cmd('EventUsrReject'))
156
157    def volume_down(self):
158        return self.pixel_buds_controller.volume('Down')
159
160    def volume_up(self):
161        return self.pixel_buds_controller.volume('Up')
162
163
164class EarstudioReceiverBluetoothHandsfreeAbstractDevice(
165        BluetoothHandsfreeAbstractDevice):
166    def __init__(self, earstudio_controller):
167        self.earstudio_controller = earstudio_controller
168
169    @property
170    def mac_address(self):
171        return self.earstudio_controller.mac_address
172
173    def accept_call(self):
174        return self.earstudio_controller.press_accept_call()
175
176    def end_call(self):
177        return self.earstudio_controller.press_end_call()
178
179    def enter_pairing_mode(self):
180        return self.earstudio_controller.enter_pairing_mode()
181
182    def next_track(self):
183        return self.earstudio_controller.press_next()
184
185    def pause(self):
186        return self.earstudio_controller.press_play_pause()
187
188    def play(self):
189        return self.earstudio_controller.press_play_pause()
190
191    def power_off(self):
192        return self.earstudio_controller.power_off()
193
194    def power_on(self):
195        return self.earstudio_controller.power_on()
196
197    def previous_track(self):
198        return self.earstudio_controller.press_previous()
199
200    def reject_call(self):
201        return self.earstudio_controller.press_reject_call()
202
203    def volume_down(self):
204        return self.earstudio_controller.press_volume_down()
205
206    def volume_up(self):
207        return self.earstudio_controller.press_volume_up()
208
209
210class JaybirdX3EarbudsBluetoothHandsfreeAbstractDevice(
211        BluetoothHandsfreeAbstractDevice):
212    def __init__(self, jaybird_controller):
213        self.jaybird_controller = jaybird_controller
214
215    @property
216    def mac_address(self):
217        return self.jaybird_controller.mac_address
218
219    def accept_call(self):
220        return self.jaybird_controller.press_accept_call()
221
222    def end_call(self):
223        return self.jaybird_controller.press_reject_call()
224
225    def enter_pairing_mode(self):
226        return self.jaybird_controller.enter_pairing_mode()
227
228    def next_track(self):
229        return self.jaybird_controller.press_next()
230
231    def pause(self):
232        return self.jaybird_controller.press_play_pause()
233
234    def play(self):
235        return self.jaybird_controller.press_play_pause()
236
237    def power_off(self):
238        return self.jaybird_controller.power_off()
239
240    def power_on(self):
241        return self.jaybird_controller.power_on()
242
243    def previous_track(self):
244        return self.jaybird_controller.press_previous()
245
246    def reject_call(self):
247        return self.jaybird_controller.press_reject_call()
248
249    def volume_down(self):
250        return self.jaybird_controller.press_volume_down()
251
252    def volume_up(self):
253        return self.jaybird_controller.press_volume_up()
254
255
256class AndroidHeadsetBluetoothHandsfreeAbstractDevice(
257        BluetoothHandsfreeAbstractDevice):
258    def __init__(self, ad_controller):
259        self.ad_controller = ad_controller
260
261    @property
262    def mac_address(self):
263        """Getting device mac with more stability ensurance.
264
265        Sometime, getting mac address is flaky that it returns None. Adding a
266        loop to add more ensurance of getting correct mac address.
267        """
268        device_mac = None
269        start_time = time.time()
270        end_time = start_time + bt_default_timeout
271        while not device_mac and time.time() < end_time:
272            device_mac = self.ad_controller.droid.bluetoothGetLocalAddress()
273        asserts.assert_true(device_mac, 'Can not get the MAC address')
274        return device_mac
275
276    def accept_call(self):
277        return self.ad_controller.droid.telecomAcceptRingingCall(None)
278
279    def end_call(self):
280        return self.ad_controller.droid.telecomEndCall()
281
282    def enter_pairing_mode(self):
283        self.ad_controller.droid.bluetoothStartPairingHelper(True)
284        return self.ad_controller.droid.bluetoothMakeDiscoverable()
285
286    def next_track(self):
287        return (self.ad_controller.droid.bluetoothMediaPassthrough("skipNext"))
288
289    def pause(self):
290        return self.ad_controller.droid.bluetoothMediaPassthrough("pause")
291
292    def play(self):
293        return self.ad_controller.droid.bluetoothMediaPassthrough("play")
294
295    def power_off(self):
296        return self.ad_controller.droid.bluetoothToggleState(False)
297
298    def power_on(self):
299        return self.ad_controller.droid.bluetoothToggleState(True)
300
301    def previous_track(self):
302        return (self.ad_controller.droid.bluetoothMediaPassthrough("skipPrev"))
303
304    def reject_call(self):
305        return self.ad_controller.droid.telecomCallDisconnect(
306            self.ad_controller.droid.telecomCallGetCallIds()[0])
307
308    def reset(self):
309        return self.ad_controller.droid.bluetoothFactoryReset()
310
311    def volume_down(self):
312        target_step = self.ad_controller.droid.getMediaVolume() - 1
313        target_step = max(target_step, 0)
314        return self.ad_controller.droid.setMediaVolume(target_step)
315
316    def volume_up(self):
317        target_step = self.ad_controller.droid.getMediaVolume() + 1
318        max_step = self.ad_controller.droid.getMaxMediaVolume()
319        target_step = min(target_step, max_step)
320        return self.ad_controller.droid.setMediaVolume(target_step)
321
322
323class BluetoothHandsfreeAbstractDeviceFactory:
324    """Generates a BluetoothHandsfreeAbstractDevice for any device controller.
325    """
326
327    _controller_abstract_devices = {
328        'EarstudioReceiver': EarstudioReceiverBluetoothHandsfreeAbstractDevice,
329        'JaybirdX3Earbuds': JaybirdX3EarbudsBluetoothHandsfreeAbstractDevice,
330        'ParentDevice': PixelBudsBluetoothHandsfreeAbstractDevice,
331        'AndroidDevice': AndroidHeadsetBluetoothHandsfreeAbstractDevice
332    }
333
334    def generate(self, controller):
335        class_name = controller.__class__.__name__
336        if class_name in self._controller_abstract_devices:
337            return self._controller_abstract_devices[class_name](controller)
338        else:
339            validate_controller(controller, BluetoothHandsfreeAbstractDevice)
340            return controller
341