# Copyright 2014 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.

import copy
import math
import os
import os.path
import re
import subprocess
import sys
import tempfile
import threading
import time

import its.caps
import its.cv2image
import its.device
from its.device import ItsSession
import its.image

import numpy as np

# For sanity checking the installed APK's target SDK version
MIN_SUPPORTED_SDK_VERSION = 28  # P

CHART_DELAY = 1  # seconds
CHART_DISTANCE = 30.0  # cm
CHART_HEIGHT = 13.5  # cm
CHART_LEVEL = 96
CHART_SCALE_START = 0.65
CHART_SCALE_STOP = 1.35
CHART_SCALE_STEP = 0.025
FACING_EXTERNAL = 2
NUM_TRYS = 2
PROC_TIMEOUT_CODE = -101  # terminated process return -process_id
PROC_TIMEOUT_TIME = 900  # timeout in seconds for a process (15 minutes)
SCENE3_FILE = os.path.join(os.environ['CAMERA_ITS_TOP'], 'pymodules', 'its',
                           'test_images', 'ISO12233.png')
SKIP_RET_CODE = 101  # note this must be same as tests/scene*/test_*
VGA_HEIGHT = 480
VGA_WIDTH = 640

# Not yet mandated tests
NOT_YET_MANDATED = {
        'scene0': [
                'test_test_patterns',
                'test_tonemap_curve'
        ],
        'scene1': [
                'test_ae_precapture_trigger',
                'test_channel_saturation'
        ],
        'scene2': [
                'test_auto_per_frame_control'
        ],
        'scene2b': [],
        'scene2c': [],
        'scene3': [],
        'scene4': [],
        'scene5': [],
        'sensor_fusion': []
}

# Must match mHiddenPhysicalCameraSceneIds in ItsTestActivity.java
HIDDEN_PHYSICAL_CAMERA_TESTS = {
        'scene0': [
                'test_burst_capture',
                'test_metadata',
                'test_read_write',
                'test_sensor_events'
        ],
        'scene1': [
                'test_exposure',
                'test_dng_noise_model',
                'test_linearity',
                'test_raw_exposure',
                'test_raw_sensitivity'
        ],
        'scene2': [
                'test_faces',
                'test_num_faces'
        ],
        'scene2b': [],
        'scene2c': [],
        'scene3': [],
        'scene4': [
                'test_aspect_ratio_and_crop'
        ],
        'scene5': [],
        'sensor_fusion': [
                'test_sensor_fusion'
        ]
}

def run_subprocess_with_timeout(cmd, fout, ferr, outdir):
    """Run subprocess with a timeout.

    Args:
        cmd:    list containing python command
        fout:   stdout file for the test
        ferr:   stderr file for the test
        outdir: dir location for fout/ferr

    Returns:
        process status or PROC_TIMEOUT_CODE if timer maxes
    """

    proc = subprocess.Popen(
            cmd, stdout=fout, stderr=ferr, cwd=outdir)
    timer = threading.Timer(PROC_TIMEOUT_TIME, proc.kill)

    try:
        timer.start()
        proc.communicate()
        test_code = proc.returncode
    finally:
        timer.cancel()

    if test_code < 0:
        return PROC_TIMEOUT_CODE
    else:
        return test_code


def calc_camera_fov(camera_id, hidden_physical_id):
    """Determine the camera field of view from internal params."""
    with ItsSession(camera_id, hidden_physical_id) as cam:
        props = cam.get_camera_properties()
        props = cam.override_with_hidden_physical_camera_props(props)
        focal_ls = props['android.lens.info.availableFocalLengths']
        if len(focal_ls) > 1:
            print 'Doing capture to determine logical camera focal length'
            cap = cam.do_capture(its.objects.auto_capture_request())
            focal_l = cap['metadata']['android.lens.focalLength']
        else:
            focal_l = focal_ls[0]
    sensor_size = props['android.sensor.info.physicalSize']
    diag = math.sqrt(sensor_size['height'] ** 2 +
                     sensor_size['width'] ** 2)
    try:
        fov = str(round(2 * math.degrees(math.atan(diag / (2 * focal_l))), 2))
    except ValueError:
        fov = str(0)
    print 'Calculated FoV: %s' % fov
    return fov


def evaluate_socket_failure(err_file_path):
    """Determine if test fails due to socket FAIL."""
    socket_fail = False
    with open(err_file_path, 'r') as ferr:
        for line in ferr:
            if (line.find('socket.error') != -1 or
                line.find('socket.timeout') != -1 or
                line.find('Problem with socket') != -1):
                socket_fail = True
    return socket_fail


def skip_sensor_fusion(camera_id):
    """Determine if sensor fusion test is skipped for this camera."""

    skip_code = SKIP_RET_CODE
    with ItsSession(camera_id) as cam:
        props = cam.get_camera_properties()
        if (its.caps.sensor_fusion(props) and its.caps.manual_sensor(props) and
                props['android.lens.facing'] is not FACING_EXTERNAL):
            skip_code = None
    return skip_code


def main():
    """Run all the automated tests, saving intermediate files, and producing
    a summary/report of the results.

    Script should be run from the top-level CameraITS directory.

    Command line arguments:
        camera:  the camera(s) to be tested. Use comma to separate multiple
                 camera Ids. Ex: "camera=0,1" or "camera=1"
        device:  device id for adb
        scenes:  the test scene(s) to be executed. Use comma to separate
                 multiple scenes. Ex: "scenes=scene0,scene1" or
                 "scenes=0,1,sensor_fusion" (sceneX can be abbreviated by X
                 where X is a integer)
        chart:   [Experimental] another android device served as test chart
                 display. When this argument presents, change of test scene
                 will be handled automatically. Note that this argument
                 requires special physical/hardware setup to work and may not
                 work on all android devices.
        result:  Device ID to forward results to (in addition to the device
                 that the tests are running on).
        rot_rig: [Experimental] ID of the rotation rig being used (formatted as
                 "<vendor ID>:<product ID>:<channel #>" or "default")
        tmp_dir: location of temp directory for output files
        skip_scene_validation: force skip scene validation. Used when test scene
                 is setup up front and don't require tester validation.
        dist:    [Experimental] chart distance in cm.
    """

    all_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4", "scene5",
                  "sensor_fusion"]

    auto_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4"]

    scene_req = {
        "scene0": None,
        "scene1": "A grey card covering at least the middle 30% of the scene",
        "scene2": "A picture containing human faces",
        "scene2b": "A picture containing human faces",
        "scene2c": "A picture containing human faces",
        "scene3": "The ISO 12233 chart",
        "scene4": "A specific test page of a circle covering at least the "
                  "middle 50% of the scene. See CameraITS.pdf section 2.3.4 "
                  "for more details",
        "scene5": "Capture images with a diffuser attached to the camera. See "
                  "CameraITS.pdf section 2.3.4 for more details",
        "sensor_fusion": "Rotating checkboard pattern. See "
                         "sensor_fusion/SensorFusion.pdf for detailed "
                         "instructions.\nNote that this test will be skipped "
                         "on devices not supporting REALTIME camera timestamp."
    }
    scene_extra_args = {
        "scene5": ["doAF=False"]
    }

    camera_id_combos = []
    scenes = []
    chart_host_id = None
    result_device_id = None
    rot_rig_id = None
    tmp_dir = None
    skip_scene_validation = False
    chart_distance = CHART_DISTANCE
    chart_level = CHART_LEVEL
    one_camera_argv = sys.argv[1:]

    for s in list(sys.argv[1:]):
        if s[:7] == "camera=" and len(s) > 7:
            camera_ids = s[7:].split(',')
            camera_id_combos = its.device.parse_camera_ids(camera_ids)
            one_camera_argv.remove(s)
        elif s[:7] == "scenes=" and len(s) > 7:
            scenes = s[7:].split(',')
        elif s[:6] == 'chart=' and len(s) > 6:
            chart_host_id = s[6:]
        elif s[:7] == 'result=' and len(s) > 7:
            result_device_id = s[7:]
        elif s[:8] == 'rot_rig=' and len(s) > 8:
            rot_rig_id = s[8:]  # valid values: 'default' or '$VID:$PID:$CH'
            # The default '$VID:$PID:$CH' is '04d8:fc73:1'
        elif s[:8] == 'tmp_dir=' and len(s) > 8:
            tmp_dir = s[8:]
        elif s == 'skip_scene_validation':
            skip_scene_validation = True
        elif s[:5] == 'dist=' and len(s) > 5:
            chart_distance = float(re.sub('cm', '', s[5:]))
        elif s[:11] == 'brightness=' and len(s) > 11:
            chart_level = s[11:]

    chart_dist_arg = 'dist= ' + str(chart_distance)
    chart_level_arg = 'brightness=' + str(chart_level)
    auto_scene_switch = chart_host_id is not None
    merge_result_switch = result_device_id is not None

    # Run through all scenes if user does not supply one
    possible_scenes = auto_scenes if auto_scene_switch else all_scenes
    if not scenes:
        scenes = possible_scenes
    else:
        # Validate user input scene names
        valid_scenes = True
        temp_scenes = []
        for s in scenes:
            if s in possible_scenes:
                temp_scenes.append(s)
            else:
                try:
                    # Try replace "X" to "sceneX"
                    scene_str = "scene" + s
                    if scene_str not in possible_scenes:
                        valid_scenes = False
                        break
                    temp_scenes.append(scene_str)
                except ValueError:
                    valid_scenes = False
                    break

        if not valid_scenes:
            print 'Unknown scene specified:', s
            assert False
        scenes = temp_scenes

    # Make output directories to hold the generated files.
    topdir = tempfile.mkdtemp(dir=tmp_dir)
    subprocess.call(['chmod', 'g+rx', topdir])
    print "Saving output files to:", topdir, "\n"

    device_id = its.device.get_device_id()
    device_id_arg = "device=" + device_id
    print "Testing device " + device_id

    # Sanity check CtsVerifier SDK level
    # Here we only do warning as there is no guarantee on pm dump output formt not changed
    # Also sometimes it's intentional to run mismatched versions
    cmd = "adb -s %s shell pm dump com.android.cts.verifier" % (device_id)
    dump_path = os.path.join(topdir, 'CtsVerifier.txt')
    with open(dump_path, 'w') as fout:
        fout.write('ITS minimum supported SDK version is %d\n--\n' % (MIN_SUPPORTED_SDK_VERSION))
        fout.flush()
        ret_code = subprocess.call(cmd.split(), stdout=fout)

    if ret_code != 0:
        print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?"

    ctsv_version = None
    ctsv_version_name = None
    with open(dump_path, 'r') as f:
        target_sdk_found = False
        version_name_found = False
        for line in f:
            match = re.search('targetSdk=([0-9]+)', line)
            if match:
                ctsv_version = int(match.group(1))
                target_sdk_found = True
            match = re.search('versionName=([\S]+)$', line)
            if match:
                ctsv_version_name = match.group(1)
                version_name_found = True
            if target_sdk_found and version_name_found:
                break

    if ctsv_version is None:
        print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?"
    elif ctsv_version < MIN_SUPPORTED_SDK_VERSION:
        print "Warning: CtsVerifier version (%d) < ITS version (%d), is this intentional?" % (
                ctsv_version, MIN_SUPPORTED_SDK_VERSION)
    else:
        print "CtsVerifier targetSdk is", ctsv_version
        if ctsv_version_name:
            print "CtsVerifier version name is", ctsv_version_name

    # Hard check on ItsService/host script version that should catch incompatible APK/script
    with ItsSession() as cam:
        cam.check_its_version_compatible()

    # Sanity Check for devices
    device_bfp = its.device.get_device_fingerprint(device_id)
    assert device_bfp is not None

    if auto_scene_switch:
        chart_host_bfp = its.device.get_device_fingerprint(chart_host_id)
        assert chart_host_bfp is not None

    if merge_result_switch:
        result_device_bfp = its.device.get_device_fingerprint(result_device_id)
        assert_err_msg = ('Cannot merge result to a different build, from '
                          '%s to %s' % (device_bfp, result_device_bfp))
        assert device_bfp == result_device_bfp, assert_err_msg

    # user doesn't specify camera id, run through all cameras
    if not camera_id_combos:
        with its.device.ItsSession() as cam:
            camera_ids = cam.get_camera_ids()
            camera_id_combos = its.device.parse_camera_ids(camera_ids);

    print "Running ITS on camera: %s, scene %s" % (camera_id_combos, scenes)

    if auto_scene_switch:
        # merge_result only supports run_parallel_tests
        if merge_result_switch and camera_ids[0] == "1":
            print "Skip chart screen"
            time.sleep(1)
        else:
            print "Waking up chart screen: ", chart_host_id
            screen_id_arg = ("screen=%s" % chart_host_id)
            cmd = ["python", os.path.join(os.environ["CAMERA_ITS_TOP"], "tools",
                                          "wake_up_screen.py"), screen_id_arg,
                   chart_level_arg]
            wake_code = subprocess.call(cmd)
            assert wake_code == 0

    for id_combo in camera_id_combos:
        # Initialize test results
        results = {}
        result_key = ItsSession.RESULT_KEY
        for s in all_scenes:
            results[s] = {result_key: ItsSession.RESULT_NOT_EXECUTED}

        camera_fov = calc_camera_fov(id_combo.id, id_combo.sub_id)
        id_combo_string = id_combo.id;
        has_hidden_sub_camera = id_combo.sub_id is not None
        if has_hidden_sub_camera:
            id_combo_string += ":" + id_combo.sub_id
            scenes = [scene for scene in scenes if HIDDEN_PHYSICAL_CAMERA_TESTS[scene]]
        # Loop capturing images until user confirm test scene is correct
        camera_id_arg = "camera=" + id_combo.id
        print "Preparing to run ITS on camera", id_combo_string, "for scenes ", scenes

        os.mkdir(os.path.join(topdir, id_combo_string))
        for d in scenes:
            os.mkdir(os.path.join(topdir, id_combo_string, d))

        tot_tests = []
        tot_pass = 0
        for scene in scenes:
            # unit is millisecond for execution time record in CtsVerifier
            scene_start_time = int(round(time.time() * 1000))
            skip_code = None
            tests = [(s[:-3], os.path.join("tests", scene, s))
                     for s in os.listdir(os.path.join("tests", scene))
                     if s[-3:] == ".py" and s[:4] == "test"]
            tests.sort()
            tot_tests.extend(tests)

            summary = "Cam" + id_combo_string + " " + scene + "\n"
            numpass = 0
            numskip = 0
            num_not_mandated_fail = 0
            numfail = 0
            validate_switch = True
            if scene_req[scene] is not None:
                out_path = os.path.join(topdir, id_combo_string, scene+".jpg")
                out_arg = "out=" + out_path
                if scene == 'sensor_fusion':
                    skip_code = skip_sensor_fusion(id_combo.id)
                    if rot_rig_id or skip_code == SKIP_RET_CODE:
                        validate_switch = False
                if skip_scene_validation:
                    validate_switch = False
                cmd = None
                if auto_scene_switch:
                    if (not merge_result_switch or
                            (merge_result_switch and id_combo_string == '0')):
                        scene_arg = 'scene=' + scene
                        fov_arg = 'fov=' + camera_fov
                        cmd = ['python',
                               os.path.join(os.getcwd(), 'tools/load_scene.py'),
                               scene_arg, chart_dist_arg, fov_arg, screen_id_arg]
                    else:
                        time.sleep(CHART_DELAY)
                else:
                    # Skip scene validation under certain conditions
                    if validate_switch and not merge_result_switch:
                        scene_arg = 'scene=' + scene_req[scene]
                        extra_args = scene_extra_args.get(scene, [])
                        cmd = ['python',
                               os.path.join(os.getcwd(),
                                            'tools/validate_scene.py'),
                               camera_id_arg, out_arg,
                               scene_arg, device_id_arg] + extra_args
                if cmd is not None:
                    valid_scene_code = subprocess.call(cmd, cwd=topdir)
                    assert valid_scene_code == 0
            print 'Start running ITS on camera %s, %s' % (id_combo_string, scene)
            # Extract chart from scene for scene3 once up front
            chart_loc_arg = ''
            chart_height = CHART_HEIGHT
            if scene == 'scene3':
                chart_height *= its.cv2image.calc_chart_scaling(
                        chart_distance, camera_fov)
                chart = its.cv2image.Chart(SCENE3_FILE, chart_height,
                                           chart_distance, CHART_SCALE_START,
                                           CHART_SCALE_STOP, CHART_SCALE_STEP,
                                           id_combo.id)
                chart_loc_arg = 'chart_loc=%.2f,%.2f,%.2f,%.2f,%.3f' % (
                        chart.xnorm, chart.ynorm, chart.wnorm, chart.hnorm,
                        chart.scale)
            # Run each test, capturing stdout and stderr.
            for (testname, testpath) in tests:
                # Only pick predefined tests for hidden physical camera
                if has_hidden_sub_camera and \
                        testname not in HIDDEN_PHYSICAL_CAMERA_TESTS[scene]:
                    numskip += 1
                    continue
                if auto_scene_switch:
                    if merge_result_switch and id_combo_string == '0':
                        # Send an input event to keep the screen not dimmed.
                        # Since we are not using camera of chart screen, FOCUS event
                        # should do nothing but keep the screen from dimming.
                        # The "sleep after x minutes of inactivity" display setting
                        # determines how long this command can keep screen bright.
                        # Setting it to something like 30 minutes should be enough.
                        cmd = ('adb -s %s shell input keyevent FOCUS'
                               % chart_host_id)
                        subprocess.call(cmd.split())
                t0 = time.time()
                for num_try in range(NUM_TRYS):
                    outdir = os.path.join(topdir, id_combo_string, scene)
                    outpath = os.path.join(outdir, testname+'_stdout.txt')
                    errpath = os.path.join(outdir, testname+'_stderr.txt')
                    if scene == 'sensor_fusion':
                        if skip_code is not SKIP_RET_CODE:
                            if rot_rig_id:
                                print 'Rotating phone w/ rig %s' % rot_rig_id
                                rig = ('python tools/rotation_rig.py rotator=%s' %
                                       rot_rig_id)
                                subprocess.Popen(rig.split())
                            else:
                                print 'Rotate phone 15s as shown in SensorFusion.pdf'
                        else:
                            test_code = skip_code
                    if skip_code is not SKIP_RET_CODE:
                        cmd = ['python', os.path.join(os.getcwd(), testpath)]
                        cmd += one_camera_argv + ["camera="+id_combo_string] + [chart_loc_arg]
                        cmd += [chart_dist_arg]
                        with open(outpath, 'w') as fout, open(errpath, 'w') as ferr:
                            test_code = run_subprocess_with_timeout(
                                cmd, fout, ferr, outdir)
                    if test_code == 0 or test_code == SKIP_RET_CODE:
                        break
                    else:
                        socket_fail = evaluate_socket_failure(errpath)
                        if socket_fail or test_code == PROC_TIMEOUT_CODE:
                            if num_try != NUM_TRYS-1:
                                print ' Retry %s/%s' % (scene, testname)
                            else:
                                break
                        else:
                            break
                t1 = time.time()

                test_failed = False
                if test_code == 0:
                    retstr = "PASS "
                    numpass += 1
                elif test_code == SKIP_RET_CODE:
                    retstr = "SKIP "
                    numskip += 1
                elif test_code != 0 and testname in NOT_YET_MANDATED[scene]:
                    retstr = "FAIL*"
                    num_not_mandated_fail += 1
                else:
                    retstr = "FAIL "
                    numfail += 1
                    test_failed = True

                msg = "%s %s/%s [%.1fs]" % (retstr, scene, testname, t1-t0)
                print msg
                its.device.adb_log(device_id, msg)
                msg_short = "%s %s [%.1fs]" % (retstr, testname, t1-t0)
                if test_failed:
                    summary += msg_short + "\n"

            # unit is millisecond for execution time record in CtsVerifier
            scene_end_time = int(round(time.time() * 1000))

            if numskip > 0:
                skipstr = ", %d test%s skipped" % (
                    numskip, "s" if numskip > 1 else "")
            else:
                skipstr = ""

            test_result = "\n%d / %d tests passed (%.1f%%)%s" % (
                numpass + num_not_mandated_fail, len(tests) - numskip,
                100.0 * float(numpass + num_not_mandated_fail) /
                (len(tests) - numskip)
                if len(tests) != numskip else 100.0, skipstr)
            print test_result

            if num_not_mandated_fail > 0:
                msg = "(*) tests are not yet mandated"
                print msg

            tot_pass += numpass
            print "%s compatibility score: %.f/100\n" % (
                    scene, 100.0 * numpass / len(tests))

            summary_path = os.path.join(topdir, id_combo_string, scene, "summary.txt")
            with open(summary_path, "w") as f:
                f.write(summary)

            passed = numfail == 0
            results[scene][result_key] = (ItsSession.RESULT_PASS if passed
                                          else ItsSession.RESULT_FAIL)
            results[scene][ItsSession.SUMMARY_KEY] = summary_path
            results[scene][ItsSession.START_TIME_KEY] = scene_start_time
            results[scene][ItsSession.END_TIME_KEY] = scene_end_time

        if tot_tests:
            print "Compatibility Score: %.f/100" % (100.0 * tot_pass / len(tot_tests))
        else:
            print "Compatibility Score: 0/100"

        msg = "Reporting ITS result to CtsVerifier"
        print msg
        its.device.adb_log(device_id, msg)
        if merge_result_switch:
            # results are modified by report_result
            results_backup = copy.deepcopy(results)
            its.device.report_result(result_device_id, id_combo_string, results_backup)

        # Report hidden_physical_id results as well.
        its.device.report_result(device_id, id_combo_string, results)

    if auto_scene_switch:
        if merge_result_switch:
            print 'Skip shutting down chart screen'
        else:
            print 'Shutting down chart screen: ', chart_host_id
            screen_id_arg = ('screen=%s' % chart_host_id)
            cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools',
                                          'turn_off_screen.py'), screen_id_arg]
            screen_off_code = subprocess.call(cmd)
            assert screen_off_code == 0

            print 'Shutting down DUT screen: ', device_id
            screen_id_arg = ('screen=%s' % device_id)
            cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools',
                                          'turn_off_screen.py'), screen_id_arg]
            screen_off_code = subprocess.call(cmd)
            assert screen_off_code == 0

    print "ITS tests finished. Please go back to CtsVerifier and proceed"

if __name__ == '__main__':
    main()