# Copyright 2023 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.
"""Runs the webcam verification test and reports the result to CTSVerifier"""

import ast
import io
import logging
import platform
import re
import subprocess
import time

from typing import NamedTuple

from mobly import asserts
from mobly import base_test
from mobly import test_runner
from mobly.controllers import android_device


class SdkVersion(NamedTuple):
    """A class that represents the SDK version of the device."""

    major: int  # The major SDK version of the device
    minor: int  # The minor SDK version of the device


class DeviceAsWebcamTest(base_test.BaseTestClass):
    # Tests device as webcam functionality with Mobly base test class to run.

    _ACTION_WEBCAM_RESULT = (
        'com.android.cts.verifier.camera.webcam.ACTION_WEBCAM_RESULT'
    )
    _WEBCAM_RESULTS = 'camera.webcam.extra.RESULTS'
    _WEBCAM_TEST_ACTIVITY = (
        'com.android.cts.verifier/.camera.webcam.WebcamTestActivity'
    )
    # TODO(373791776): Find a way to discover PreviewActivity for vendors that
    # change the webcam service.
    # pylint: disable-next=line-too-long
    _DAC_PREVIEW_ACTIVITY = 'com.android.DeviceAsWebcam/com.android.deviceaswebcam.DeviceAsWebcamPreview'
    _ACTIVITY_START_WAIT = 1.5  # seconds
    _ADB_RESTART_WAIT = 9  # seconds
    _FPS_TOLERANCE = 0.15  # 15 percent
    _RESULT_PASS = 'PASS'
    _RESULT_FAIL = 'FAIL'
    _RESULT_FORCE_PASS = 'FORCE_PASS'
    _RESULT_NOT_EXECUTED = 'NOT_EXECUTED'
    _MANUAL_FRAME_CHECK_DURATION = 8  # seconds
    _WINDOWS_OS = 'Windows'
    _MAC_OS = 'Darwin'
    _LINUX_OS = 'Linux'
    _FORCE_PASS_SDK_VERSION = 36
    _FORCE_PASS_SDK_VERSION_FULL = 202504

    def get_full_sdk_version(self) -> SdkVersion:
        """Gets and parses the full SDK version of the dut.

        Returns:
          An SdkVersion namedtuple (major, minor).

        Raises:
          ValueError: If the SDK version string from the device property
            'ro.build.version.sdk_full' cannot be parsed.
        """
        sdk_full = self.dut.adb.getprop('ro.build.version.sdk_full')
        logging.debug('ro.build.version.sdk_full: %s', sdk_full)

        match = re.fullmatch(r'(\d+)\.(\d+)', sdk_full)
        if not match:
            raise ValueError(
                f'Failed to parse ro.build.version.sdk_full: {sdk_full}'
            )

        sdk_major = int(match.group(1))
        sdk_minor = int(match.group(2))
        logging.debug('sdk_major: %s; sdk_minor: %s', sdk_major, sdk_minor)

        return SdkVersion(sdk_major, sdk_minor)

    def run_os_specific_test(self):
        """Runs the os specific webcam test script.

        Returns:
          A result list of tuples (tested_fps, actual_fps)
        """
        results = []
        current_os = platform.system()

        if current_os == self._LINUX_OS:
            # pylint: disable-next=import-outside-toplevel
            import linux_webcam_test

            logging.info('Starting test on Linux')
            results = linux_webcam_test.main(self.dut.serial)
        elif current_os == self._WINDOWS_OS:
            logging.warning(
                'Webcam test on Windows is decrecated and will be removed '
                'in a future release. If the test fails, please try running on '
                'a Linux system before filing a bug or requesting exception.'
            )

            logging.info('Starting test on Windows')
            # Due to compatibility issues directly running the windows
            # main function, the results from the windows_webcam_test script
            # are printed to the stdout and retrieved
            output = subprocess.check_output(
                ['python', 'windows_webcam_test.py']
            )
            output_str = output.decode('utf-8')
            results = ast.literal_eval(output_str.strip())
        elif current_os == self._MAC_OS:
            logging.warning(
                'Webcam test on MacOS is deprecated and will be removed '
                'in a future release. If the test fails, please try running on '
                'a Linux system before filing a bug or requesting exception.'
            )
            # pylint: disable-next=import-outside-toplevel
            import mac_webcam_test

            logging.info('Starting test on Mac')
            results = mac_webcam_test.main()
        else:
            logging.error('Running on an unknown OS')

        return results

    def validate_fps(self, results):
        """Verifies the webcam FPS

        Verifies that the webcam FPS falls within the acceptable range of the
        tested FPS.

        Args:
            results: A result list of tuples (tested_fps, actual_fps)

        Returns:
            True if all FPS are within tolerance range, False otherwise
        """
        result = True

        for elem in results:
            tested_fps = elem[0]
            actual_fps = elem[1]

            max_diff = tested_fps * self._FPS_TOLERANCE

            if abs(tested_fps - actual_fps) > max_diff:
                logging.error(
                    'FPS is out of tolerance range!  Tested: %d Actual FPS: %d',
                    tested_fps,
                    actual_fps,
                )
                result = False

        return result

    def setup_class(self):
        # Registering android_device controller module declares the test
        # dependencies on Android device hardware. By default, we expect at
        # least one object is created from this.
        devices = self.register_controller(android_device, min_number=1)
        self.dut = devices[0]

        # Keep device on while testing since it requires a manual check on the
        # webcam frames
        # '7' is a combination of flags ORed together to keep the device on
        # in all cases
        self.dut.adb.shell(
            'settings put global stay_on_while_plugged_in 7'.split()
        )

    def teardown_class(self):
        self.dut.adb.shell(
            'settings put global stay_on_while_plugged_in 0'.split()
        )
        return super().teardown_class()

    def test_webcam(self):
        # This test was broken until 25Q4, and passed unconditionally on user
        # only builds. As we cannot break upgrading devices, this test
        # will pass unconditionally on devices launching with Android 2025Q4
        # or earlier.
        vendor_api = int(self.dut.adb.getprop('ro.vendor.api_level'))
        # vendor_api is either of the format YYYYMM or the Android SDK version
        # number.
        must_pass = vendor_api <= self._FORCE_PASS_SDK_VERSION or (
            vendor_api > 100000
            and vendor_api <= self._FORCE_PASS_SDK_VERSION_FULL
        )
        logging.debug('Vendor API: %s, Must Pass: %s', vendor_api, must_pass)
        test_status = self._RESULT_PASS
        try:
            cmd = (
                'am start'
                f' {self._WEBCAM_TEST_ACTIVITY} --activity-brought-to-front'
            )
            self.dut.adb.shell(cmd.split())

            # Set USB preference option to webcam
            # 'handle_usb_disconnect' reinitializes any mobly specific services that
            # may have been disrupted by the disconnection.
            with self.dut.handle_usb_disconnect():
                # Set USB preference option to webcam
                # This assumes that uvc is supported by the device which is safe
                # as the test should only be run if the device does indeed support
                # uvc.
                try:
                    logging.info('Setting USB preference option to webcam')
                    self.dut.adb.shell('svc usb setFunctions uvc'.split())
                except android_device.adb.AdbError as e:
                    # error code 255 may be returned because adb lost connection as
                    # part of switching to UVC. Other error codes are unexpected.
                    if e.ret_code != 255:
                        # unhandled exception. crash and burn
                        raise e
                finally:
                    # adb disconnects when changing usb function and reconnects
                    # after a while. Wait for device to come back. Will throw a
                    # AdbTimeoutError exception if adb does not recover in
                    # _ADB_RESTART_WAIT seconds.
                    self.dut.adb.wait_for_device(
                        timeout=DeviceAsWebcamTest._ADB_RESTART_WAIT
                    )

            # Check if device came back with uvc mode active.
            stderr = io.BytesIO()
            stdout = self.dut.adb.shell(
                'svc usb getFunctions'.split(), stderr=stderr
            )

            # For whatever reason, this call outputs to stderr instead of stdout
            # despite there being no error. This will likely change in the future.
            # For now, just check both stdout and stderr.
            stderr = stderr.getvalue().decode('utf-8')
            stdout = stdout.decode('utf-8')
            if 'uvc' not in stdout and 'uvc' not in stderr:
                logging.error('Could not switch to webcam mode')

                # Notify CTSVerifier test that setting webcam option was
                # unsuccessful
                cmd = (
                    f'am broadcast -a {self._ACTION_WEBCAM_RESULT} --es'
                    f' {self._WEBCAM_RESULTS} {self._RESULT_FAIL}'
                )
                self.dut.adb.shell(cmd.split())
                asserts.fail('Could not switch to webcam mode.')

            fps_results = self.run_os_specific_test()
            if not fps_results and not must_pass:
                # Notify CTSVerifier that no fps tests were executed.
                cmd = (
                    f'am broadcast -a {self._ACTION_WEBCAM_RESULT} --es'
                    f' {self._WEBCAM_RESULTS} {self._RESULT_FAIL}'
                )
                self.dut.adb.shell(cmd.split())
                asserts.fail('Could not run webcam test. See logs for errors.')

            logging.info('FPS test results (Expected, Actual): %s', fps_results)
            result = self.validate_fps(fps_results)

            if not result:
                logging.error('FPS testing failed')
                test_status = self._RESULT_FAIL

            if not result and must_pass:
                logging.warning(
                    'Test failed but ignoring due to vendor freeze testing'
                    ' guarantees. Please ensure that the webcam framerate is'
                    ' acceptable to your users.'
                )
                test_status = self._RESULT_FORCE_PASS

            # Send result to CTSVerifier test
            time.sleep(self._ACTIVITY_START_WAIT)
            cmd = (
                f'am broadcast -a {self._ACTION_WEBCAM_RESULT} --es'
                f' {self._WEBCAM_RESULTS} {test_status}'
            )
            self.dut.adb.shell(cmd.split())

            sdk_version = self.get_full_sdk_version()
            if sdk_version.major < 36 or (
                sdk_version.major == 36 and sdk_version.minor < 1
            ):
                # DeviceAsWebcam does not export the preview activity before
                # sdk 36.1. Attempting to start an unexported activity via ADB will
                # cause a SecurityException. To prevent the test from crashing on
                # older devices, skip pulling up the activity and instruct the
                # user to manually verify the functionality.
                logging.warning(
                    'Skipping pulling up the webcam activity as it causes a '
                    'SecurityException on older versions.'
                )
                logging.warning(
                    'Please manually verify that the Webcam Preview activity is'
                    ' working.'
                )
            else:
                # Enable the webcam service preview activity for a manual
                # check on webcam frames
                cmd = (
                    'am start'
                    f' {self._DAC_PREVIEW_ACTIVITY} --activity-no-history'
                )
                self.dut.adb.shell(cmd.split())
                time.sleep(self._MANUAL_FRAME_CHECK_DURATION)

                cmd = (
                    'am start'
                    f' {self._WEBCAM_TEST_ACTIVITY} --activity-brought-to-front'
                )
                self.dut.adb.shell(cmd.split())
        except Exception as e:
            if must_pass:
                logging.warning(
                    'Test failed but ignoring due to vendor freeze testing'
                    ' guarantees. Please ensure that the webcam framerate is'
                    ' acceptable to your users.',
                    exc_info=True,
                )
                test_status = self._RESULT_FORCE_PASS
                cmd = (
                    f'am broadcast -a {self._ACTION_WEBCAM_RESULT} --es'
                    f' {self._WEBCAM_RESULTS} {test_status}'
                )
                self.dut.adb.shell(cmd.split())
            else:
                raise e

        asserts.assert_true(
            test_status == self._RESULT_PASS
            or test_status == self._RESULT_FORCE_PASS,
            'Results: Failed',
        )


if __name__ == '__main__':
    test_runner.main()