xref: /cts/apps/CameraITS/tests/inprog/rolling_shutter_skew/test_rolling_shutter_skew.py

"""Experimentally determines a camera's rolling shutter skew.

See the accompanying PDF for instructions on how to use this test.
"""
from __future__ import division
from __future__ import print_function

import argparse
import glob
import math
import os
import sys
import tempfile

import cv2
import its.caps
import its.device
import its.image
import its.objects
import numpy as np

DEBUG = False

# Constants for which direction the camera is facing.
FACING_FRONT = 0
FACING_BACK = 1
FACING_EXTERNAL = 2

# Camera capture defaults.
FPS = 30
WIDTH = 640
HEIGHT = 480
TEST_LENGTH = 1

# Each circle in a cluster must be within this many pixels of some other circle
# in the cluster.
CLUSTER_DISTANCE = 50.0 / HEIGHT
# A cluster must consist of at least this percentage of the total contours for
# it to be allowed into the computation.
MAJORITY_THRESHOLD = 0.7

# Constants to make sure the slope of the fitted line is reasonable.
SLOPE_MIN_THRESHOLD = 0.5
SLOPE_MAX_THRESHOLD = 1.5

# To improve readability of unit conversions.
SEC_TO_NSEC = float(10**9)
MSEC_TO_NSEC = float(10**6)
NSEC_TO_MSEC = 1.0 / float(10**6)


class RollingShutterArgumentParser(object):
    """Parses command line arguments for the rolling shutter test."""

    def __init__(self):
        self.__parser = argparse.ArgumentParser(
                description='Run rolling shutter test')
        self.__parser.add_argument(
                '-d', '--debug',
                action='store_true',
                help='print and write data useful for debugging')
        self.__parser.add_argument(
                '-f', '--fps',
                type=int,
                help='FPS to capture with during the test (defaults to 30)')
        self.__parser.add_argument(
                '-i', '--img_size',
                help=('comma-separated dimensions of captured images (defaults '
                      'to 640x480). Example: --img_size=<width>,<height>'))
        self.__parser.add_argument(
                '-l', '--led_time',
                type=float,
                required=True,
                help=('how many milliseconds each column of the LED array is '
                      'lit for'))
        self.__parser.add_argument(
                '-p', '--panel_distance',
                type=float,
                help='how far the LED panel is from the camera (in meters)')
        self.__parser.add_argument(
                '-r', '--read_dir',
                help=('read existing test data from specified directory.  If '
                      'not specified, new test data is collected from the '
                      'device\'s camera)'))
        self.__parser.add_argument(
                '--device_id',
                help=('device ID for device being tested (can also use '
                      '\'device=<DEVICE ID>\')'))
        self.__parser.add_argument(
                '-t', '--test_length',
                type=int,
                help=('how many seconds the test should run for (defaults to 1 '
                      'second)'))
        self.__parser.add_argument(
                '-o', '--debug_dir',
                help=('write debugging information in a folder in the '
                      'specified directory.  Otherwise, the system\'s default '
                      'location for temporary folders is used.  --debug must '
                      'be specified along with this argument.'))

    def parse_args(self):
        """Returns object containing parsed values from the command line."""
        # Don't show argparse the 'device' flag, since it's in a different
        # format than the others (to maintain CameraITS conventions) and it will
        # complain.
        filtered_args = [arg for arg in sys.argv[1:] if 'device=' not in arg]
        args = self.__parser.parse_args(filtered_args)
        if args.device_id:
            # If argparse format is used, convert it to a format its.device can
            # use later on.
            sys.argv.append('device=%s' % args.device_id)
        return args


def main():
    global DEBUG
    global CLUSTER_DISTANCE

    parser = RollingShutterArgumentParser()
    args = parser.parse_args()

    DEBUG = args.debug
    if not DEBUG and args.debug_dir:
        print('argument --debug_dir requires --debug', file=sys.stderr)
        sys.exit()

    if args.read_dir is None:
        # Collect new data.
        raw_caps, reported_skew = collect_data(args)
        frames = [its.image.convert_capture_to_rgb_image(c) for c in raw_caps]
    else:
        # Load existing data.
        frames, reported_skew = load_data(args.read_dir)

    # Make the cluster distance relative to the height of the image.
    (frame_h, _, _) = frames[0].shape
    CLUSTER_DISTANCE = frame_h * CLUSTER_DISTANCE
    debug_print('Setting cluster distance to %spx.' % CLUSTER_DISTANCE)

    if DEBUG:
        debug_dir = setup_debug_dir(args.debug_dir)
        # Write raw frames.
        for i, img in enumerate(frames):
            its.image.write_image(img, '%s/raw/%03d.png' % (debug_dir, i))
    else:
        debug_dir = None

    avg_shutter_skew, num_frames_used = find_average_shutter_skew(
            frames, args.led_time, debug_dir)
    if debug_dir:
        # Write the reported skew with the raw images, so the directory can also
        # be used to read from.
        with open(debug_dir + '/raw/reported_skew.txt', 'w') as f:
            f.write('%sms\n' % reported_skew)

    if avg_shutter_skew is None:
        print('Could not find usable frames.')
    else:
        print('Device reported shutter skew of %sms.' % reported_skew)
        print('Measured shutter skew is %sms (averaged over %s frames).' %
              (avg_shutter_skew, num_frames_used))


def collect_data(args):
    """Capture a new set of frames from the device's camera.

    Args:
        args: Parsed command line arguments.

    Returns:
        A list of RGB images as numpy arrays.
    """
    fps = args.fps if args.fps else FPS
    if args.img_size:
        w, h = map(int, args.img_size.split(','))
    else:
        w, h = WIDTH, HEIGHT
    test_length = args.test_length if args.test_length else TEST_LENGTH

    with its.device.ItsSession() as cam:
        props = cam.get_camera_properties()
        its.caps.skip_unless(its.caps.manual_sensor(props))
        facing = props['android.lens.facing']
        if facing != FACING_FRONT and facing != FACING_BACK:
            print('Unknown lens facing %s' % facing)
            assert 0

        fmt = {'format': 'yuv', 'width': w, 'height': h}
        s, e, _, _, _ = cam.do_3a(get_results=True, do_af=False)
        req = its.objects.manual_capture_request(s, e)
        req['android.control.aeTargetFpsRange'] = [fps, fps]

        # Convert from milliseconds to nanoseconds.  We only want enough
        # exposure time to saturate approximately one column.
        exposure_time = (args.led_time / 2.0) * MSEC_TO_NSEC
        print('Using exposure time of %sns.' % exposure_time)
        req['android.sensor.exposureTime'] = exposure_time
        req["android.sensor.frameDuration"] = int(SEC_TO_NSEC / fps);

        if args.panel_distance is not None:
            # Convert meters to diopters and use that for the focus distance.
            req['android.lens.focusDistance'] = 1 / args.panel_distance
        print('Starting capture')
        raw_caps = cam.do_capture([req]*fps*test_length, fmt)
        print('Finished capture')

        # Convert from nanoseconds to milliseconds.
        shutter_skews = {c['metadata']['android.sensor.rollingShutterSkew'] *
                          NSEC_TO_MSEC for c in raw_caps}
        # All frames should have same rolling shutter skew.
        assert len(shutter_skews) == 1
        shutter_skew = list(shutter_skews)[0]

        return raw_caps, shutter_skew


def load_data(dir_name):
    """Reads camera frame data from an existing directory.

    Args:
        dir_name: Name of the directory to read data from.

    Returns:
        A list of RGB images as numpy arrays.
    """
    frame_files = glob.glob('%s/*.png' % dir_name)
    frames = []
    for frame_file in sorted(frame_files):
        frames.append(its.image.load_rgb_image(frame_file))
    with open('%s/reported_skew.txt' % dir_name, 'r') as f:
        reported_skew = f.readline()[:-2]  # Strip off 'ms' suffix
    return frames, reported_skew


def find_average_shutter_skew(frames, led_time, debug_dir=None):
    """Finds the average shutter skew using the given frames.

    Frames without enough information will be discarded from the average to
    improve overall accuracy.

    Args:
        frames:    List of RGB images from the camera being tested.
        led_time:  How long a single LED column is lit for (in milliseconds).
        debug_dir: (optional) Directory to write debugging information to.

    Returns:
        The average calculated shutter skew and the number of frames used to
        calculate the average.
    """
    avg_shutter_skew = 0.0
    avg_slope = 0.0
    weight = 0.0
    num_frames_used = 0

    for i, frame in enumerate(frames):
        debug_print('------------------------')
        debug_print('| PROCESSING FRAME %03d |' % i)
        debug_print('------------------------')
        shutter_skew, confidence, slope = calculate_shutter_skew(
                frame, led_time, i, debug_dir=debug_dir)
        if shutter_skew is None:
            debug_print('Skipped frame.')
        else:
            debug_print('Shutter skew is %sms (confidence: %s).' %
                        (shutter_skew, confidence))
            # Use the confidence to weight the average.
            avg_shutter_skew += shutter_skew * confidence
            avg_slope += slope * confidence
            weight += confidence
            num_frames_used += 1

    debug_print('\n')
    if num_frames_used == 0:
        return None, None
    else:
        avg_shutter_skew /= weight
        avg_slope /= weight
        slope_err_str = ('The average slope of the fitted line was too %s '
                         'to get an accurate measurement (slope was %s).  '
                         'Try making the LED panel %s.')
        if avg_slope < SLOPE_MIN_THRESHOLD:
            print(slope_err_str % ('flat', avg_slope, 'slower'),
                  file=sys.stderr)
        elif avg_slope > SLOPE_MAX_THRESHOLD:
            print(slope_err_str % ('steep', avg_slope, 'faster'),
                  file=sys.stderr)
        return avg_shutter_skew, num_frames_used


def calculate_shutter_skew(frame, led_time, frame_num=None, debug_dir=None):
    """Calculates the shutter skew of the camera being used for this test.

    Args:
        frame:     A single RGB image captured by the camera being tested.
        led_time:  How long a single LED column is lit for (in milliseconds).
        frame_num: (optional) Number of the given frame.
        debug_dir: (optional) Directory to write debugging information to.

    Returns:
        The shutter skew (in milliseconds), the confidence in the accuracy of
        the measurement (useful for weighting averages), and the slope of the
        fitted line.
    """
    contours, scratch_img, contour_img, mono_img = find_contours(frame.copy())
    if debug_dir is not None:
        cv2.imwrite('%s/contour/%03d.png' % (debug_dir, frame_num), contour_img)
        cv2.imwrite('%s/mono/%03d.png' % (debug_dir, frame_num), mono_img)

    largest_cluster, cluster_percentage = find_largest_cluster(contours,
                                                               scratch_img)
    if largest_cluster is None:
        debug_print('No majority cluster found.')
        return None, None, None
    elif len(largest_cluster) <= 1:
        debug_print('Majority cluster was too small.')
        return None, None, None
    debug_print('%s points in the largest cluster.' % len(largest_cluster))

    np_cluster = np.array([[c.x, c.y] for c in largest_cluster])
    [vx], [vy], [x0], [y0] = cv2.fitLine(np_cluster, cv2.cv.CV_DIST_L2,
                                         0, 0.01, 0.01)
    slope = vy / vx
    debug_print('Slope is %s.' % slope)
    (frame_h, frame_w, _) = frame.shape
    # Draw line onto scratch frame.
    pt1 = tuple(map(int, (x0 - vx * 1000, y0 - vy * 1000)))
    pt2 = tuple(map(int, (x0 + vx * 1000, y0 + vy * 1000)))
    cv2.line(scratch_img, pt1, pt2, (0, 255, 255), thickness=3)

    # We only need the width of the cluster.
    _, _, cluster_w, _ = find_cluster_bounding_rect(largest_cluster,
                                                    scratch_img)

    num_columns = find_num_columns_spanned(largest_cluster)
    debug_print('%s columns spanned by cluster.' % num_columns)
    # How long it takes for a column to move from the left of the bounding
    # rectangle to the right.
    left_to_right_time = led_time * num_columns
    milliseconds_per_x_pixel = left_to_right_time / cluster_w
    # The distance between the line's intersection at the top of the frame and
    # the intersection at the bottom.
    x_range = frame_h / slope
    shutter_skew = milliseconds_per_x_pixel * x_range
    # If the aspect ratio is different from 4:3 (the aspect ratio of the actual
    # sensor), we need to correct, because it will be cropped.
    shutter_skew *= (float(frame_w) / float(frame_h)) / (4.0 / 3.0)

    if debug_dir is not None:
        cv2.imwrite('%s/scratch/%03d.png' % (debug_dir, frame_num),
                    scratch_img)

    return shutter_skew, cluster_percentage, slope


def find_contours(img):
    """Finds contours in the given image.

    Args:
        img: Image in Android camera RGB format.

    Returns:
        OpenCV-formatted contours, the original image in OpenCV format, a
        thresholded image with the contours drawn on, and a grayscale version of
        the image.
    """
    # Convert to format OpenCV can work with (BGR ordering with byte-ranged
    # values).
    img *= 255
    img = img.astype(np.uint8)
    img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)

    # Since the LED colors for the panel we're using are red, we can get better
    # contours for the LEDs if we ignore the green and blue channels.  This also
    # makes it so we don't pick up the blue control screen of the LED panel.
    red_img = img[:, :, 2]
    _, thresh = cv2.threshold(red_img, 0, 255, cv2.THRESH_BINARY +
                              cv2.THRESH_OTSU)

    # Remove noise before finding contours by eroding the thresholded image and
    # then re-dilating it.  The size of the kernel represents how many
    # neighboring pixels to consider for the result of a single pixel.
    kernel = np.ones((3, 3), np.uint8)
    opening = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel, iterations=2)

    if DEBUG:
        # Need to convert it back to BGR if we want to draw colored contours.
        contour_img = cv2.cvtColor(opening, cv2.COLOR_GRAY2BGR)
    else:
        contour_img = None
    cv2_version = cv2.__version__
    if cv2_version.startswith('3.'): # OpenCV 3.x
        _, contours, _ = cv2.findContours(
                opening, cv2.cv.CV_RETR_EXTERNAL, cv2.cv.CV_CHAIN_APPROX_NONE)
    else: # OpenCV 2.x and 4.x
        contours, _ = cv2.findContours(
                opening, cv2.cv.CV_RETR_EXTERNAL, cv2.cv.CV_CHAIN_APPROX_NONE)
    if DEBUG:
        cv2.drawContours(contour_img, contours, -1, (0, 0, 255), thickness=2)
    return contours, img, contour_img, red_img


def convert_to_circles(contours):
    """Converts given contours into circle objects.

    Args:
        contours: Contours generated by OpenCV.

    Returns:
        A list of circles.
    """

    class Circle(object):
        """Holds data to uniquely define a circle."""

        def __init__(self, contour):
            self.x = int(np.mean(contour[:, 0, 0]))
            self.y = int(np.mean(contour[:, 0, 1]))
            # Get diameters of each axis then half it.
            x_r = (np.max(contour[:, 0, 0]) - np.min(contour[:, 0, 0])) / 2.0
            y_r = (np.max(contour[:, 0, 1]) - np.min(contour[:, 0, 1])) / 2.0
            # Average x radius and y radius to get the approximate radius for
            # the given contour.
            self.r = (x_r + y_r) / 2.0
            assert self.r > 0.0

        def distance_to(self, other):
            return (math.sqrt((other.x - self.x)**2 + (other.y - self.y)**2) -
                    self.r - other.r)

        def intersects(self, other):
            return self.distance_to(other) <= 0.0

    return list(map(Circle, contours))


def find_largest_cluster(contours, frame):
    """Finds the largest cluster in the given contours.

    Args:
        contours: Contours generated by OpenCV.
        frame:    For drawing debugging information onto.

    Returns:
        The cluster with the most contours in it and the percentage of all
        contours that the cluster contains.
    """
    clusters = proximity_clusters(contours)

    if not clusters:
        return None, None  # No clusters found.

    largest_cluster = max(clusters, key=len)
    cluster_percentage = len(largest_cluster) / len(contours)

    if cluster_percentage < MAJORITY_THRESHOLD:
        return None, None

    if DEBUG:
        # Draw largest cluster on scratch frame.
        for circle in largest_cluster:
            cv2.circle(frame, (int(circle.x), int(circle.y)), int(circle.r),
                       (0, 255, 0), thickness=2)

    return largest_cluster, cluster_percentage


def proximity_clusters(contours):
    """Sorts the given contours into groups by distance.

    Converts every given contour to a circle and clusters by adding a circle to
    a cluster only if it is close to at least one other circle in the cluster.

    TODO: Make algorithm faster (currently O(n**2)).

    Args:
        contours: Contours generated by OpenCV.

    Returns:
        A list of clusters, where each cluster is a list of the circles
        contained in the cluster.
    """
    circles = convert_to_circles(contours)

    # Use disjoint-set data structure to store assignments.  Start every point
    # in their own cluster.
    cluster_assignments = [-1 for i in range(len(circles))]

    def get_canonical_index(i):
        if cluster_assignments[i] >= 0:
            index = get_canonical_index(cluster_assignments[i])
            # Collapse tree for better runtime.
            cluster_assignments[i] = index
            return index
        else:
            return i

    def get_cluster_size(i):
        return -cluster_assignments[get_canonical_index(i)]

    for i, curr in enumerate(circles):
        close_circles = [j for j, p in enumerate(circles) if i != j and
                         curr.distance_to(p) < CLUSTER_DISTANCE]
        if close_circles:
            # Note: largest_cluster is an index into cluster_assignments.
            largest_cluster = min(close_circles, key=get_cluster_size)
            largest_size = get_cluster_size(largest_cluster)
            curr_index = get_canonical_index(i)
            curr_size = get_cluster_size(i)
            if largest_size > curr_size:
                # largest_cluster is larger than us.
                target_index = get_canonical_index(largest_cluster)
                # Add our cluster size to the bigger one.
                cluster_assignments[target_index] -= curr_size
                # Reroute our group to the bigger one.
                cluster_assignments[curr_index] = target_index
            else:
                # We're the largest (or equal to the largest) cluster.  Reroute
                # all groups to us.
                for j in close_circles:
                    smaller_size = get_cluster_size(j)
                    smaller_index = get_canonical_index(j)
                    if smaller_index != curr_index:
                        # We only want to modify clusters that aren't already in
                        # the current one.

                        # Add the smaller cluster's size to ours.
                        cluster_assignments[curr_index] -= smaller_size
                        # Reroute their group to us.
                        cluster_assignments[smaller_index] = curr_index

    # Convert assignments list into list of clusters.
    clusters_dict = {}
    for i in range(len(cluster_assignments)):
        canonical_index = get_canonical_index(i)
        if canonical_index not in clusters_dict:
            clusters_dict[canonical_index] = []
        clusters_dict[canonical_index].append(circles[i])
    return clusters_dict.values()


def find_cluster_bounding_rect(cluster, scratch_frame):
    """Finds the minimum rectangle that bounds the given cluster.

    The bounding rectangle will always be axis-aligned.

    Args:
        cluster:       Cluster being used to find the bounding rectangle.
        scratch_frame: Image that rectangle is drawn onto for debugging
                       purposes.

    Returns:
        The leftmost and topmost x and y coordinates, respectively, along with
        the width and height of the rectangle.
    """
    avg_distance = find_average_neighbor_distance(cluster)
    debug_print('Average distance between points in largest cluster is %s '
                'pixels.' % avg_distance)

    c_x = min(cluster, key=lambda c: c.x - c.r)
    c_y = min(cluster, key=lambda c: c.y - c.r)
    c_w = max(cluster, key=lambda c: c.x + c.r)
    c_h = max(cluster, key=lambda c: c.y + c.r)

    x = c_x.x - c_x.r - avg_distance
    y = c_y.y - c_y.r - avg_distance
    w = (c_w.x + c_w.r + avg_distance) - x
    h = (c_h.y + c_h.r + avg_distance) - y

    if DEBUG:
        points = np.array([[x, y], [x + w, y], [x + w, y + h], [x, y + h]],
                          np.int32)
        cv2.polylines(scratch_frame, [points], True, (255, 0, 0), thickness=2)

    return x, y, w, h


def find_average_neighbor_distance(cluster):
    """Finds the average distance between every circle and its closest neighbor.

    Args:
        cluster: List of circles

    Returns:
        The average distance.
    """
    avg_distance = 0.0
    for a in cluster:
        closest_point = None
        closest_dist = None
        for b in cluster:
            if a is b:
                continue
            curr_dist = a.distance_to(b)
            if closest_point is None or curr_dist < closest_dist:
                closest_point = b
                closest_dist = curr_dist
        avg_distance += closest_dist
    avg_distance /= len(cluster)
    return avg_distance


def find_num_columns_spanned(circles):
    """Finds how many columns of the LED panel are spanned by the given circles.

    Args:
        circles: List of circles (assumed to be from the LED panel).

    Returns:
        The number of columns spanned.
    """
    if not circles:
        return 0

    def x_intersects(c_a, c_b):
        return abs(c_a.x - c_b.x) < (c_a.r + c_b.r)

    circles = sorted(circles, key=lambda c: c.x)
    last_circle = circles[0]
    num_columns = 1
    for circle in circles[1:]:
        if not x_intersects(circle, last_circle):
            last_circle = circle
            num_columns += 1

    return num_columns


def setup_debug_dir(dir_name=None):
    """Creates a debug directory and required subdirectories.

    Each subdirectory contains images from a different step in the process.

    Args:
        dir_name: The directory to create.  If none is specified, a temp
        directory is created.

    Returns:
        The name of the directory that is used.
    """
    if dir_name is None:
        dir_name = tempfile.mkdtemp()
    else:
        force_mkdir(dir_name)
    print('Saving debugging files to "%s"' % dir_name)
    # For original captured images.
    force_mkdir(dir_name + '/raw', clean=True)
    # For monochrome images.
    force_mkdir(dir_name + '/mono', clean=True)
    # For contours generated from monochrome images.
    force_mkdir(dir_name + '/contour', clean=True)
    # For post-contour debugging information.
    force_mkdir(dir_name + '/scratch', clean=True)
    return dir_name


def force_mkdir(dir_name, clean=False):
    """Creates a directory if it doesn't already exist.

    Args:
        dir_name: Name of the directory to create.
        clean:    (optional) If set to true, cleans image files from the
                  directory (if it already exists).
    """
    if os.path.exists(dir_name):
        if clean:
            for image in glob.glob('%s/*.png' % dir_name):
                os.remove(image)
    else:
        os.makedirs(dir_name)


def debug_print(s, *args, **kwargs):
    """Only prints if the test is running in debug mode."""
    if DEBUG:
        print(s, *args, **kwargs)


if __name__ == '__main__':
    main()