1# Copyright 2023 The Android Open Source Project
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#      http://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
15import ast
16import logging
17import os
18import platform
19import subprocess
20import time
21
22from mobly import asserts
23from mobly import base_test
24from mobly import test_runner
25from mobly.controllers import android_device
26
27
28class DeviceAsWebcamTest(base_test.BaseTestClass):
29  # Tests device as webcam functionality with Mobly base test class to run.
30
31  _ACTION_WEBCAM_RESULT = 'com.android.cts.verifier.camera.webcam.ACTION_WEBCAM_RESULT'
32  _WEBCAM_RESULTS = 'camera.webcam.extra.RESULTS'
33  _WEBCAM_TEST_ACTIVITY = 'com.android.cts.verifier/.camera.webcam.WebcamTestActivity'
34  # TODO(373791776): Find a way to discover PreviewActivity for vendors that change
35  # the webcam service.
36  _DAC_PREVIEW_ACTIVITY = 'com.android.DeviceAsWebcam/com.android.deviceaswebcam.DeviceAsWebcamPreview'
37  _ACTIVITY_START_WAIT = 1.5  # seconds
38  _ADB_RESTART_WAIT = 9  # seconds
39  _FPS_TOLERANCE = 0.15 # 15 percent
40  _RESULT_PASS = 'PASS'
41  _RESULT_FAIL = 'FAIL'
42  _RESULT_NOT_EXECUTED = 'NOT_EXECUTED'
43  _MANUAL_FRAME_CHECK_DURATION = 8  # seconds
44  _WINDOWS_OS = 'Windows'
45  _MAC_OS = 'Darwin'
46  _LINUX_OS = 'Linux'
47
48  def run_os_specific_test(self):
49    """Runs the os specific webcam test script.
50
51    Returns:
52      A result list of tuples (tested_fps, actual_fps)
53    """
54    results = []
55    current_os = platform.system()
56
57    if current_os == self._WINDOWS_OS:
58      import windows_webcam_test
59      logging.info('Starting test on Windows')
60      # Due to compatibility issues directly running the windows
61      # main function, the results from the windows_webcam_test script
62      # are printed to the stdout and retrieved
63      output = subprocess.check_output(['python', 'windows_webcam_test.py'])
64      output_str = output.decode('utf-8')
65      results = ast.literal_eval(output_str.strip())
66    elif current_os == self._LINUX_OS:
67      import linux_webcam_test
68      logging.info('Starting test on Linux')
69      results = linux_webcam_test.main()
70    elif current_os == self._MAC_OS:
71      import mac_webcam_test
72      logging.info('Starting test on Mac')
73      results = mac_webcam_test.main()
74    else:
75      logging.info('Running on an unknown OS')
76
77    return results
78
79  def validate_fps(self, results):
80    """Verifies the webcam FPS falls within the acceptable range of the tested FPS.
81
82    Args:
83        results: A result list of tuples (tested_fps, actual_fps)
84
85    Returns:
86        True if all FPS are within tolerance range, False otherwise
87    """
88    result = True
89
90    for elem in results:
91      tested_fps = elem[0]
92      actual_fps = elem[1]
93
94      max_diff = tested_fps * self._FPS_TOLERANCE
95
96      if abs(tested_fps - actual_fps) > max_diff:
97        logging.error('FPS is out of tolerance range! '
98                      ' Tested: %d Actual FPS: %d', tested_fps, actual_fps)
99        result = False
100
101    return result
102
103  def run_cmd(self, cmd):
104    """Replaces os.system call, while hiding stdout+stderr messages."""
105    with open(os.devnull, 'wb') as devnull:
106      subprocess.check_call(cmd.split(), stdout=devnull,
107                            stderr=subprocess.STDOUT)
108
109  def setup_class(self):
110    # Registering android_device controller module declares the test
111    # dependencies on Android device hardware. By default, we expect at least
112    # one object is created from this.
113    devices = self.register_controller(android_device, min_number=1)
114    self.dut = devices[0]
115    self.dut.adb.root()
116
117  def test_webcam(self):
118
119    adb = f'adb -s {self.dut.serial}'
120
121    # Keep device on while testing since it requires a manual check on the
122    # webcam frames
123    # '7' is a combination of flags ORed together to keep the device on
124    # in all cases
125    self.dut.adb.shell(['settings', 'put', 'global',
126                        'stay_on_while_plugged_in', '7'])
127
128    cmd = f"""{adb} shell am start {self._WEBCAM_TEST_ACTIVITY}
129        --activity-brought-to-front"""
130    self.run_cmd(cmd)
131
132    # Check if webcam feature is enabled
133    dut_webcam_enabled = self.dut.adb.shell(['getprop', 'ro.usb.uvc.enabled'])
134    if 'true' in dut_webcam_enabled.decode('utf-8'):
135      logging.info('Webcam enabled, testing webcam')
136    else:
137      logging.info('Webcam not enabled, skipping webcam test')
138
139      # Notify CTSVerifier test that the webcam test was skipped,
140      # the test will be marked as PASSED for this case
141      cmd = (f"""{adb} shell am broadcast -a
142          {self._ACTION_WEBCAM_RESULT} --es {self._WEBCAM_RESULTS}
143          {self._RESULT_NOT_EXECUTED}""")
144      self.run_cmd(cmd)
145
146      return
147
148    # Set USB preference option to webcam
149    set_uvc = self.dut.adb.shell(['svc', 'usb', 'setFunctions', 'uvc'])
150    if not set_uvc:
151      logging.error('USB preference option to set webcam unsuccessful')
152
153      # Notify CTSVerifier test that setting webcam option was unsuccessful
154      cmd = (f"""{adb} shell am broadcast -a
155          {self._ACTION_WEBCAM_RESULT} --es {self._WEBCAM_RESULTS}
156          {self._RESULT_FAIL}""")
157      self.run_cmd(cmd)
158      return
159
160    # After resetting the USB preference, adb disconnects
161    # and reconnects so wait for device
162    time.sleep(self._ADB_RESTART_WAIT)
163
164    fps_results = self.run_os_specific_test()
165    logging.info('FPS test results (Expected, Actual): %s', fps_results)
166    result = self.validate_fps(fps_results)
167
168    test_status = self._RESULT_PASS
169    if not result or not fps_results:
170      logging.info('FPS testing failed')
171      test_status = self._RESULT_FAIL
172
173    # Send result to CTSVerifier test
174    time.sleep(self._ACTIVITY_START_WAIT)
175    cmd = (f"""{adb} shell am broadcast -a
176        {self._ACTION_WEBCAM_RESULT} --es {self._WEBCAM_RESULTS}
177        {test_status}""")
178    self.run_cmd(cmd)
179
180    # Enable the webcam service preview activity for a manual
181    # check on webcam frames
182    cmd = f"""{adb} shell am start {self._DAC_PREVIEW_ACTIVITY}
183        --activity-no-history"""
184    self.run_cmd(cmd)
185    time.sleep(self._MANUAL_FRAME_CHECK_DURATION)
186
187    cmd = f"""{adb} shell am start {self._WEBCAM_TEST_ACTIVITY}
188        --activity-brought-to-front"""
189    self.run_cmd(cmd)
190
191    asserts.assert_true(test_status == self._RESULT_PASS, 'Results: Failed')
192
193    self.dut.adb.shell(['settings', 'put',
194                        'global', 'stay_on_while_plugged_in', '0'])
195
196if __name__ == '__main__':
197  test_runner.main()
198