Python OpenCV ImageHash

find_matches.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) Odd Stråbø <[email protected]>
#

import os

from opencv_test import Session, Image, Hash, HASHERS
from gmpy2 import hamdist

session = Session()

for htype in [2]: #HASHERS.keys():
    hashes = session.query(Hash).filter_by(type=htype).all()
    for hash in hashes:
        hash.num = int.from_bytes(hash.hash, 'big')
    
    hlen = len(hashes[0].hash)

    for h1 in hashes:
        for h2 in hashes:
            if h1 is h2: continue

            d = hamdist(h1.num, h2.num) / (hlen-1)
            confedence = (1-d) * 100
            if 0 < d and d <= 0.05:
                print("{} {}/{}: `{}` = `{}` {:.3f}%".format(htype, d, len(h1.hash*8), h1.image.path, h2.image.path, confedence))
                #os.system('"{}"'.format(h1.image.path))
                #os.system('"{}"'.format(h2.image.path))
                #input()

opencv_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) Odd Stråbø <[email protected]>
#

import os

import sqlalchemy
from cv2 import cv2
#import magic

SOURCES = [
    "U:/Dropbox",
    "U:/Backup/Phone/DCIM",
    "U:/Backup/i7w7/G/Pictures",
    "N:/Pictures",
]

IMAGE_EXTENSIONS = [
    ".png",
    ".jpg",
    ".jpeg",
    ".jpe",
    ".jp2",
    ".bmp",
]

AVERAGE_HASH = 0
BLOCK_MEAN_HASH = 1
MARR_HILDRETH_HASH = 2
P_HASH = 3
RADIAL_VARIANCE_HASH = 4

HASHERS = {
    AVERAGE_HASH: cv2.img_hash.averageHash,
    BLOCK_MEAN_HASH: cv2.img_hash.blockMeanHash,
    MARR_HILDRETH_HASH: cv2.img_hash.marrHildrethHash,
    P_HASH: cv2.img_hash.pHash,
    RADIAL_VARIANCE_HASH: cv2.img_hash.radialVarianceHash,
}

engine = sqlalchemy.create_engine("sqlite:///db.sqlite3")

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

Base = declarative_base()
Session = sessionmaker(bind=engine)

from sqlalchemy import Column, Integer, String, LargeBinary

class Image(Base):
    __tablename__ = 'images'

    id = Column(Integer, primary_key=True)

    path = Column(String)
    mimetype = Column(String)

    def __repr__(self):
       return "<Image(path='{path}')>".format(path=self.path)

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Hash(Base):
    __tablename__ = 'hashes'
    id = Column(Integer, primary_key=True)
    type = Column(Integer)
    hash = Column(LargeBinary)
    image_id = Column(Integer, ForeignKey('images.id'))

    image = relationship("Image", back_populates="hashes")

    def __repr__(self):
        return "<Hash(type='{}', hash='{}')>".format(type, hash)

Image.hashes = relationship("Hash", order_by=Hash.type, back_populates="image")

Base.metadata.create_all(engine)

#session = Session()
if __name__ == "__main__":
    for topdir in SOURCES:
        for root, dirs, files in os.walk(topdir, topdown=False):
            for name in files:
                path = os.path.normpath(os.path.join(root, name))
                ext = os.path.splitext(path)[-1]

                if ext not in IMAGE_EXTENSIONS:
                    continue

                print(path)
                session = Session()

                try:
                    dbim = session.query(Image).filter_by(path=path).first()

                    if not dbim:
                        h = hash(path)
                        dbim = Image(path=path)
                        session.add(dbim)
                    
                    # -----------------------------------------------------------
                    
                    im = None
                    
                    for h, f in HASHERS.items():
                        dbh = session.query(Hash).filter_by(image_id=dbim.id, type=h).first()
                        if not dbh:
                            if im is None:
                                im = cv2.imread(path)
                            dbh = Hash(image=dbim, type=h, hash=f(im).tobytes())
                            print((h, repr(dbh.hash)))
                            session.add(dbh)
                    
                    # -----------------------------------------------------------

                    session.commit()
                except cv2.error as e:
                    print(e)
                    pass
                
                session.close()
            for name in dirs:
                path = os.path.normpath(os.path.join(root, name))