Diff two images to tell if they are the same

diff_image.py

#!/usr/bin/env python3
# Copyright 2018 Canonical Ltd.
# Written by:
#   Taihsiang Ho <[email protected]>
#
# Original source written by Sylvain Pineau <[email protected]>:
# https://git.launchpad.net/plainbox-provider-sru/tree/bin/screenshot_validation
#
# opencv-python >= 3.4.2.17
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
import argparse
import imghdr
import os

import cv2


parser = argparse.ArgumentParser(
    description='''
Automatically validates two images by using 
OpenCV ORB detection and a FLANN Matcher (Fast Approximate Nearest Neighbor
Search Library)

On success returns 0. Otherwise a non-zero value is returned and a
diagnostic message is printed on standard error.
''',
    formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument('base',
                    metavar='BASE',
                    help='Input file to use as base image')
parser.add_argument('target',
                    metavar='TARGET',
                    default=None,
                    help='Input file to use as target image')
# benchmark: the same picture is 1652
# 435 for lab monitoring - a half of human showing up in front of the camera (< 1m)
parser.add_argument('--min_matches',
                    type=int,
                    default=1200,
                    help='Minimum threshold value to validate a \
                          positive match')
args = parser.parse_args()

# Initiate ORB features detector
orb = cv2.ORB_create(nfeatures=100000)

if not imghdr.what(args.base):
    raise SystemExit(
        "ERROR: unable to read the base file: {}".format(args.base))
if not imghdr.what(args.target):
    raise SystemExit(
        "ERROR: unable to read the target file: {}".format(args.target))
image_base = cv2.imread(args.base, cv2.IMREAD_GRAYSCALE)
image_target = cv2.imread(args.target, cv2.IMREAD_GRAYSCALE)

# Find the keypoints and descriptors with ORB
kp1, des1 = orb.detectAndCompute(image_base, None)
if des1 is None:
    raise SystemExit(
        "ERROR: Not enough keypoints in base image, aborting...")
kp2, des2 = orb.detectAndCompute(image_target, None)
if des2 is None:
    raise SystemExit(
        "ERROR: Not enough keypoints in target image, aborting...")

# Use the FLANN Matcher (Fast Approximate Nearest Neighbor Search Library)
flann_params = dict(algorithm=6,  # FLANN_INDEX_LSH
                    table_number=6,
                    key_size=12,
                    multi_probe_level=1)

flann = cv2.FlannBasedMatcher(flann_params, {})

source = 0
results = []
img = None

matches = flann.knnMatch(des1, des2, k=2)

# store all the good matches as per Lowe's ratio test
good_matches = [m[0] for m in matches if len(m) == 2 and
                m[0].distance < m[1].distance * 0.7]

results.append(len(good_matches))

avg = sum(results) / len(results)

print("Base: {} Target: {}".format(args.base, args.target))
if avg > args.min_matches:
    print("Match found! ({} > {})".format(avg, args.min_matches))
else:
    print("Something may show up on the screen!!")
    raise SystemExit(
        "ERROR: Not enough matches are found - {} < {}".format(
            avg,
            args.min_matches))

scan_all.py

#!/usr/bin/env python3
#
# Author: Taihsiang Ho <[email protected]>
#
import argparse
import subprocess
from pprint import pprint
from os import listdir
from os.path import isfile, join


parser = argparse.ArgumentParser(
    description='''
Wrapper of diff_image.py to scan all images in a folder
''',
    formatter_class=argparse.RawDescriptionHelpFormatter
)
parser.add_argument('base',
                    metavar='BASE',
                    help='Input file to use as base image')
parser.add_argument('target',
                    metavar='TARGET',
                    default=None,
                    help='Input folder to use as target images')
parser.add_argument('-f', '--flow',
                    type=int,
                    default=None,
                    help='Flow mode. This will override base image'
                    'and use the previous N image as the base.')
# benchmark: the same picture is 1652
# 435 for lab monitoring - a half of human showing up in front of the camera (< 1m)
parser.add_argument('--min_matches',
                    type=int,
                    default=1000,
                    help='Minimum threshold value to validate a \
                          positive match')
parser.add_argument('-x', '--xdg-open',
                    default=False,
                    action='store_true',
                    help='If opening the failed target images for you.'
                    'Use this option carefully.'
                    'You may open a lot of images.')
args = parser.parse_args()

# list all files in the folder
all_images_name = listdir(args.target)
# all_images_name_sorted
all_images_name.sort()
all_images = [ join(args.target, f) for f in all_images_name \
               if isfile(join(args.target, f))]

if args.flow:
    msg = 'Flow mode is on. Use previous {} th image as base image'
    print(msg.format(args.flow))

# diff all of them and show results
all_images_number = len(all_images)
progress_counter = 0

failed_images = []
for image in all_images:
    progress_counter += 1
    if progress_counter % 10 == 0:
        print("Progress images: {} out of {}".format(progress_counter, all_images_number))

    try:
        if args.flow:
            image_index = all_images.index(image)
            if image_index == 0:
                print("1st image. Skipped.")
                image_base = all_images[0]
            elif image_index < 5:
                print("The previous 5 images."
                      "Force to use the previous 1 image as base image.")
                image_base = all_images[image_index - 1]
            else:
                image_base = all_images[image_index - args.flow]
        else:
            image_base = str(args.base)
        command = ["./diff_image.py", image_base, image,
                   '--min_matches', str(args.min_matches)]
        subprocess.check_output(command)

    except subprocess.CalledProcessError:
        failed_images.append(image)
        if args.xdg_open:
            subprocess.call(['xdg-open', image])

if len(failed_images) > 0:
    print("\n=====================================\n")
    print("Total failed image number: {}):".format(len(failed_images)))
    print("The following images are suspicious:")
    pprint(failed_images)
else:
    print("Nothing suspicious is found.")