1#!/usr/bin/env python3
2#
3# Copyright (C) 2019 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.
16"""Perform base Avrcp command from headset to dut"""
17import time
18import os
19import queue
20
21from acts import asserts
22from acts.test_utils.abstract_devices.bluetooth_handsfree_abstract_device import BluetoothHandsfreeAbstractDeviceFactory as Factory
23from acts.test_utils.bt.simulated_carkit_device import SimulatedCarkitDevice
24from acts.test_utils.bt.bt_test_utils import connect_phone_to_headset
25from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
26from acts.test_utils.car.car_media_utils import EVENT_PLAY_RECEIVED
27from acts.test_utils.car.car_media_utils import EVENT_PAUSE_RECEIVED
28from acts.test_utils.car.car_media_utils import EVENT_SKIP_NEXT_RECEIVED
29from acts.test_utils.car.car_media_utils import EVENT_SKIP_PREV_RECEIVED
30from acts.test_utils.car.car_media_utils import CMD_MEDIA_PLAY
31from acts.test_utils.car.car_media_utils import CMD_MEDIA_PAUSE
32
33ADB_FILE_EXISTS = 'test -e %s && echo True'
34DEFAULT_TIMEOUT = 5
35EVENT_TIMEOUT = 1
36
37
38class AvrcpBaseTest(BluetoothBaseTest):
39    def __init__(self, configs):
40        super(AvrcpBaseTest, self).__init__(configs)
41        self.dut = self.android_devices[0]
42        serial = self.user_params['simulated_carkit_device']
43        controller = SimulatedCarkitDevice(serial)
44        self.controller = Factory().generate(controller)
45
46        self.phone_music_files = []
47        self.host_music_files = []
48        for music_file in self.user_params['music_file_names']:
49            self.phone_music_files.append(os.path.join(
50                self.user_params['phone_music_file_dir'], music_file))
51            self.host_music_files.append(os.path.join(
52                self.user_params['host_music_file_dir'], music_file))
53
54        self.ensure_phone_has_music_file()
55
56    def setup_class(self):
57        super().setup_class()
58        self.controller.power_on()
59        time.sleep(DEFAULT_TIMEOUT)
60
61    def teardown_class(self):
62        super().teardown_class()
63        self.dut.droid.mediaPlayStop()
64        self.controller.destroy()
65
66    def setup_test(self):
67        self.dut.droid.bluetoothMediaPhoneSL4AMBSStart()
68        time.sleep(DEFAULT_TIMEOUT)
69
70        self.dut.droid.bluetoothStartPairingHelper(True)
71        if not connect_phone_to_headset(self.dut, self.controller, 600):
72            asserts.fail('Not able to connect to hands-free device')
73
74        #make sure SL4AMBS is active MediaSession
75        self.dut.droid.bluetoothMediaHandleMediaCommandOnPhone(CMD_MEDIA_PLAY)
76        time.sleep(0.5)
77        self.dut.droid.bluetoothMediaHandleMediaCommandOnPhone(CMD_MEDIA_PAUSE)
78
79    def teardown_test(self):
80        self.dut.droid.bluetoothMediaPhoneSL4AMBSStop()
81
82    def ensure_phone_has_music_file(self):
83        """Make sure music file (based on config values) is on the phone."""
84        for host_file, phone_file in zip(self.host_music_files,
85                                         self.phone_music_files):
86            if self.dut.adb.shell(ADB_FILE_EXISTS % phone_file):
87                self.log.info(
88                    'Music file {} already on phone. Skipping file transfer.'
89                    .format(host_file))
90            else:
91                self.dut.adb.push(host_file, phone_file)
92                has_file = self.dut.adb.shell(
93                        ADB_FILE_EXISTS % phone_file)
94                if not has_file:
95                    self.log.error(
96                        'Audio file {} not pushed to phone.'.format(host_file))
97                self.log.info('Music file successfully pushed to phone.')
98
99    def play_from_controller(self):
100        self.dut.ed.clear_all_events()
101        self.controller.play()
102        try:
103            self.dut.ed.pop_event(EVENT_PLAY_RECEIVED, EVENT_TIMEOUT)
104        except queue.Empty as e:
105            asserts.fail('{} Event Not received'.format(EVENT_PLAY_RECEIVED))
106        self.log.info('Event Received : {}'.format(EVENT_PLAY_RECEIVED))
107
108    def pause_from_controller(self):
109        self.dut.ed.clear_all_events()
110        self.controller.pause()
111        try:
112            self.dut.ed.pop_event(EVENT_PAUSE_RECEIVED, EVENT_TIMEOUT)
113        except queue.Empty as e:
114            asserts.fail('{} Event Not received'.format(EVENT_PAUSE_RECEIVED))
115        self.log.info('Event Received : {}'.format(EVENT_PAUSE_RECEIVED))
116
117    def skip_next_from_controller(self):
118        self.dut.ed.clear_all_events()
119        self.controller.next_track()
120        try:
121            self.dut.ed.pop_event(EVENT_SKIP_NEXT_RECEIVED, EVENT_TIMEOUT)
122        except queue.Empty as e:
123            asserts.fail('{} Event Not '
124                         'received'.format(EVENT_SKIP_NEXT_RECEIVED))
125        self.log.info('Event Received : {}'.format(EVENT_SKIP_NEXT_RECEIVED))
126
127    def skip_prev_from_controller(self):
128        self.dut.ed.clear_all_events()
129        self.controller.previous_track()
130        try:
131            self.dut.ed.pop_event(EVENT_SKIP_PREV_RECEIVED, EVENT_TIMEOUT)
132        except queue.Empty as e:
133            asserts.fail('{} Event Not '
134                         'received'.format(EVENT_SKIP_PREV_RECEIVED))
135        self.log.info('Event Received : {}'.format(EVENT_SKIP_PREV_RECEIVED))
136