Examples¶
The scripts in examples/
run from the repository root on libphash's test images; each one's docstring shows the
command and what it prints, and the test suite runs them all and compares. Four are
twins of libphash's C examples,
and the tests require both to print the same.
Find duplicates¶
"""Find the pictures that appear more than once among many files.
Hashes every file with pHash on all CPUs, prints each hash as it would be stored, then
every pair within a threshold, closest first. The threshold is the most bits two hashes
of one picture may differ in; measure it on your own images (``threshold.py``). The twin
of libphash's ``examples/find_duplicates.c``: the same files give the same output. Run::
python examples/find_duplicates.py 7 native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg native/libphash/tests/data/photo.png
prints::
# algorithm=phash version=2000002 native-jpeg=1
7eb66192bc394247 native/libphash/tests/data/photo.jpeg
7eb66192bc394247 native/libphash/tests/data/photo_copy.jpeg
0100000101000139 native/libphash/tests/data/photo.png
0 bits native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg
pairs within 7 bits: 1
"""
from __future__ import annotations
import sys
from itertools import combinations
import libphash as ph
def main(argv: list[str]) -> int:
if len(argv) < 3:
print(f"usage: {argv[0]} <max-bits> <image>...", file=sys.stderr)
return 1
try:
max_bits = int(argv[1])
except ValueError:
max_bits = -1
if not 0 <= max_bits <= 64:
print("max-bits must be a number from 0 to 64", file=sys.stderr)
return 1
# Every file hashed with pHash on all CPUs; a file that cannot be read fails alone.
results = ph.hash_files(argv[2:], ph.Algorithm.PHASH)
# A stored hash carries what decides its value: the algorithm, the library version,
# and whether JPEG went through libjpeg-turbo or stb_image. The settings are the
# defaults here; a program that changes them stores them too.
major, minor, patch = (int(n) for n in ph.libphash_version().split("."))
version = major * 1_000_000 + minor * 1000 + patch
jpeg = int(ph.can_decode("jpeg"))
print(f"# algorithm={ph.Algorithm.PHASH} version={version} native-jpeg={jpeg}")
hashed: list[tuple[str, int]] = []
for result in results:
if result.hashes is None:
print(result.error, file=sys.stderr)
else:
value = result.hashes[ph.Algorithm.PHASH]
print(f"{ph.hash_to_hex(value)} {result.source}")
hashed.append((result.source, value))
# Every pair compared once: n * (n - 1) / 2 Hamming distances.
pairs = [
(ph.hamming_distance(ha, hb), a, b)
for (a, ha), (b, hb) in combinations(hashed, 2)
if ph.hamming_distance(ha, hb) <= max_bits
]
for distance, a, b in sorted(pairs, key=lambda p: p[0]):
print(f"{distance:2d} bits {a} {b}")
print(f"pairs within {max_bits} bits: {len(pairs)}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Hash distance between two images¶
"""Hash two images with the four 64-bit algorithms; print how many bits each pair differs in.
The twin of libphash's ``examples/hash_distance.c``: the same images give the same output.
Run from the repository root::
python examples/hash_distance.py native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_rotated_90.jpeg
prints::
aHash 000c6461c1c3ffff 1f07634303333f0f 26 bits differ
dHash b159cdc98b1f1f1e 6c1e879786e6fcdc 35 bits differ
pHash 7eb66192bc394247 a7a313dc73c92c25 34 bits differ
wHash ffffc38386263100 f2fcccc0c2c6e0f8 26 bits differ
"""
from __future__ import annotations
import sys
from typing import cast
import libphash as ph
from libphash import Algorithm
ALL_FOUR = (Algorithm.AHASH, Algorithm.DHASH, Algorithm.PHASH, Algorithm.WHASH)
NAMES = {Algorithm.AHASH: "aHash", Algorithm.DHASH: "dHash",
Algorithm.PHASH: "pHash", Algorithm.WHASH: "wHash"} # fmt: skip
def hash_file(path: str) -> dict[Algorithm, int]:
"""Return aHash, dHash, pHash and wHash of one file, from one pass over the image."""
with ph.ImageContext() as ctx:
ctx.load_file(path)
hashes = ctx.hashes(*ALL_FOUR)
# The 64-bit algorithms give an int, the others a Digest.
return {algo: cast("int", hashes[algo]) for algo in ALL_FOUR}
def main(argv: list[str]) -> int:
if len(argv) != 3:
print(f"usage: {argv[0]} <image-a> <image-b>", file=sys.stderr)
return 1
try:
a, b = hash_file(argv[1]), hash_file(argv[2])
except ph.PhashError as exc:
print(exc, file=sys.stderr)
return 1
for algo in ALL_FOUR:
distance = ph.hamming_distance(a[algo], b[algo])
print(
f"{NAMES[algo]} {a[algo]:016x} {b[algo]:016x} {distance:2d} bits differ"
)
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
One picture from three sources¶
"""One picture loaded three ways, and the same pHash from each.
As pixels already in memory (numpy, with a row stride), as an encoded file in memory,
and as a file on disk. The twin of libphash's ``examples/load_sources.c``: the same
output. Needs numpy. Run::
python examples/load_sources.py frame.ppm
(``frame.ppm`` is the file it writes and then loads) prints::
pixels pHash a036b6b65676c8c9
memory pHash a036b6b65676c8c9
file pHash a036b6b65676c8c9
one picture, one hash
"""
from __future__ import annotations
import sys
from pathlib import Path
import numpy as np
import numpy.typing as npt
import libphash as ph
# Each row is padded to 512 bytes: the stride is the distance from one row to the next,
# which need not be the width times the channels.
WIDTH, HEIGHT, STRIDE = 150, 100, 512
def make_frame() -> npt.NDArray[np.uint8]:
"""Return a frame some other code produced: RGB, a gradient with a bright disc."""
buffer = bytearray(HEIGHT * STRIDE)
# Rows STRIDE bytes apart, pixels 3, channels 1: the padding stays between rows.
frame: npt.NDArray[np.uint8] = np.ndarray(
(HEIGHT, WIDTH, 3), np.uint8, buffer, strides=(STRIDE, 3, 1)
)
y, x = np.mgrid[:HEIGHT, :WIDTH]
frame[..., 0] = x
frame[..., 1] = y * 2
frame[..., 2] = 160
frame[(x - 100) ** 2 + (y - 40) ** 2 < 30 * 30] = (250, 220, 90)
return frame
def print_phash(ctx: ph.ImageContext, source: str) -> int:
"""Hash what ``ctx`` holds and print it with the name of the source."""
value = ctx.phash()
print(f"{source:<16} pHash {ph.hash_to_hex(value)}")
return value
def main(argv: list[str]) -> int:
if len(argv) != 2:
print(f"usage: {argv[0]} <file to write>", file=sys.stderr)
return 1
frame = make_frame()
# The same frame encoded as a binary PPM, whole, in one buffer.
encoded = f"P6\n{WIDTH} {HEIGHT}\n255\n".encode() + frame.tobytes()
with ph.ImageContext() as ctx:
try:
# Pixels: an array of shape (height, width, channels); its row step goes
# to C as the stride, without a copy.
ctx.load_pixels(frame)
from_pixels = print_phash(ctx, "pixels")
# Memory: an encoded file -- from a network, a database, an archive.
ctx.load_bytes(encoded)
from_memory = print_phash(ctx, "memory")
# File: the same bytes on disk.
Path(argv[1]).write_bytes(encoded)
ctx.load_file(argv[1])
from_file = print_phash(ctx, "file")
except ph.PhashError as exc:
print(exc, file=sys.stderr)
return 1
if not from_pixels == from_memory == from_file:
print("the three sources disagree")
return 1
print("one picture, one hash")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Every kind of source¶
"""Hash one image from every kind of source: a path, bytes, a file object, numpy, Pillow.
A decoded image (numpy, Pillow, OpenCV) is hashed from its pixels, so it gives the hash
of the file only when its decoder produced the same pixels: true of PNG, not in general
of JPEG, which every decoder rounds its own way. OpenCV's arrays are BGR, and say so
with ``channel_order="bgr"``. Needs numpy and Pillow; OpenCV is used if installed. Run::
python examples/sources.py native/libphash/tests/data/photo.png
prints::
path 0100000101000139
bytes 0100000101000139
file object 0100000101000139
numpy 0100000101000139
Pillow 0100000101000139
"""
from __future__ import annotations
import io
import sys
from pathlib import Path
import numpy as np
from PIL import Image
import libphash as ph
def main(argv: list[str]) -> int:
if len(argv) != 2:
print(f"usage: {argv[0]} <png image>", file=sys.stderr)
return 1
path = Path(argv[1])
data = path.read_bytes()
with Image.open(path) as opened:
image = opened.convert("RGB")
sources: dict[str, ph.Source] = {
"path": path,
"bytes": data,
"file object": io.BytesIO(data),
"numpy": np.asarray(image), # shape (height, width, 3), RGB
"Pillow": image,
}
hashes = {name: ph.hash(source) for name, source in sources.items()}
for name, value in hashes.items():
print(f"{name:<12} {ph.hash_to_hex(value)}")
try:
import cv2 # noqa: PLC0415 -- optional
except ImportError:
return 0
bgr = cv2.imread(str(path), cv2.IMREAD_COLOR)
if bgr is None:
print(f"OpenCV cannot read {path}", file=sys.stderr)
return 1
with ph.ImageContext() as ctx:
ctx.load_pixels(bgr, channel_order="bgr")
if ctx.phash() != hashes["path"]:
print("OpenCV's pixels hash differently")
return 1
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Configuration¶
"""One Config for every context, a refused value, and what each setting is read by.
The twin of libphash's ``examples/configure.c``. In Python the settings are one
immutable :class:`libphash.Config`, checked when it is created: a value out of range
raises there, naming the field, and no context ever holds it. A context keeps the
Config it was created with, so a different setting is a different context. Run::
python examples/configure.py native/libphash/tests/data/photo.jpeg
prints::
BMH digest at block_size 8: 8 bytes
phash_reduction_size=9: refused
pHash, 8x8 block: 7eb66192bc394247
pHash, 6x6 block: 0000000852fb9487
ColorHash in color: a digest
ColorHash in grayscale: RequiresColorError
"""
from __future__ import annotations
import sys
import libphash as ph
# Every hash to be compared with another is computed with this one Config, so that all
# of them are computed alike.
CONFIG = ph.Config(
max_pixels=50_000_000, alpha_mode="white", gray_weights=(1, 1, 1), bmh_block_size=8
)
def main(argv: list[str]) -> int:
if len(argv) != 2:
print(f"usage: {argv[0]} <image>", file=sys.stderr)
return 1
path = argv[1]
# A digest's size follows from the settings; no image is needed to know it.
size, _ = ph.digest_info("bmh", CONFIG)
print(f"BMH digest at block_size 8: {size} bytes")
# A 9x9 block does not fit 64 bits: refused when the Config is made.
try:
CONFIG.replace(phash_reduction_size=9)
except ph.InvalidArgumentError:
print("phash_reduction_size=9: refused")
try:
before = ph.hash(path, config=CONFIG)
after = ph.hash(path, config=CONFIG.replace(phash_reduction_size=6))
except ph.PhashError as exc:
print(exc, file=sys.stderr)
return 1
print(f"pHash, 8x8 block: {before:016x}\npHash, 6x6 block: {after:016x}")
# A load setting decides how the image is decoded: in grayscale there is no color
# left for a color hash.
for name, cfg in (("color", CONFIG), ("grayscale", CONFIG.replace(grayscale=True))):
try:
ph.hash(path, "color_hash", config=cfg)
outcome = "a digest"
except ph.RequiresColorError as exc:
outcome = type(exc).__name__
print(f"ColorHash in {name + ':':<11} {outcome}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Storing in SQLite¶
"""Store hashes in SQLite, and find an image's near duplicates there.
A 64-bit hash fits SQLite's ``INTEGER``, which is signed: a hash at or above 2**63 is
stored as the negative number with the same 64 bits, and read back the same way. A
digest is stored as its text form, which keeps its kind, so a digest read back still
refuses a comparison that does not fit it. Store the library version and the
settings beside the hashes: hashes are comparable only when both match. Run::
python examples/store_sqlite.py native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg native/libphash/tests/data/photo_rotated_90.jpeg native/libphash/tests/data/photo.png
prints::
stored 4 images with libphash 2.0.2
within 10 bits of native/libphash/tests/data/photo.jpeg:
0 native/libphash/tests/data/photo.jpeg
0 native/libphash/tests/data/photo_copy.jpeg
mHash similarity of the first two, read back: 1.000
"""
from __future__ import annotations
import sqlite3
import sys
import libphash as ph
def to_int64(value: int) -> int:
"""Return the signed 64-bit integer with the bits of an unsigned 64-bit hash."""
return value - (1 << 64) if value >= 1 << 63 else value
def from_int64(value: int) -> int:
"""Return the unsigned 64-bit hash stored as a signed integer."""
return value + (1 << 64) if value < 0 else value
def hamming(a: int, b: int) -> int:
"""Hamming distance of two hashes in their stored form, for SQL."""
return ph.hamming_distance(from_int64(a), from_int64(b))
def main(argv: list[str]) -> int:
if len(argv) < 2:
print(f"usage: {argv[0]} <image>...", file=sys.stderr)
return 1
db = sqlite3.connect(":memory:")
db.execute(
"CREATE TABLE images (path TEXT PRIMARY KEY, phash INTEGER NOT NULL,"
" mhash TEXT NOT NULL, libphash TEXT NOT NULL)"
)
db.create_function("hamming", 2, hamming, deterministic=True)
with ph.ImageContext() as ctx:
for path in argv[1:]:
ctx.load_file(path)
db.execute(
"INSERT INTO images VALUES (?, ?, ?, ?)",
(path, to_int64(ctx.phash()), ctx.mhash().hex(), ph.libphash_version()),
)
print(f"stored {len(argv) - 1} images with libphash {ph.libphash_version()}")
query = to_int64(ph.hash(argv[1]))
print(f"within 10 bits of {argv[1]}:")
rows = db.execute(
"SELECT hamming(phash, ?) AS d, path FROM images WHERE d <= 10 ORDER BY d, path",
(query,),
)
for distance, path in rows:
print(f" {distance:2d} {path}")
a, b = (
ph.Digest.from_hex(text)
for (text,) in db.execute("SELECT mhash FROM images ORDER BY rowid LIMIT 2")
)
print(f"mHash similarity of the first two, read back: {a.similarity(b):.3f}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Nearest hashes with numpy¶
"""Find the nearest hashes to a query among many, with numpy.
The 64-bit hashes of a collection go into one ``uint64`` array; the distance to a query
is the population count of the XOR, computed for all of them at once
(``numpy.bitwise_count``, numpy 2.0 or later). Needs numpy. Run::
python examples/nearest_numpy.py native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg native/libphash/tests/data/photo_rotated_90.jpeg native/libphash/tests/data/photo.png native/libphash/tests/data/photo_complex.png
prints::
the 3 nearest to native/libphash/tests/data/photo.jpeg:
0 native/libphash/tests/data/photo.jpeg
0 native/libphash/tests/data/photo_copy.jpeg
31 native/libphash/tests/data/photo_complex.png
"""
from __future__ import annotations
import sys
import numpy as np
import libphash as ph
def main(argv: list[str]) -> int:
if len(argv) < 2:
print(f"usage: {argv[0]} <query image> <image>...", file=sys.stderr)
return 1
paths = argv[1:]
results = ph.hash_files(paths, ph.Algorithm.PHASH)
known = [(r.source, r.hashes[ph.Algorithm.PHASH]) for r in results if r.hashes]
hashes = np.array([value for _, value in known], dtype=np.uint64)
query = np.uint64(ph.hash(paths[0]))
distances = np.bitwise_count(hashes ^ query)
nearest = distances.argsort(kind="stable")[:3]
print(f"the {len(nearest)} nearest to {paths[0]}:")
for i in nearest:
print(f" {distances[i]:2d} {known[i][0]}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Groups with a BK-tree¶
"""Group near-duplicate images with a BK-tree instead of comparing every pair.
A BK-tree indexes the hashes by their Hamming distance to one another, so a query
"everything within N bits" visits a part of the collection instead of all of it; the
groups are the images linked by a chain of such matches. Comparing every pair (as
``find_duplicates.py`` does) is simpler and fine for a few thousand images. Run::
python examples/bk_tree.py 6 native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg native/libphash/tests/data/jpeg/s420.jpg native/libphash/tests/data/jpeg/s444.jpg native/libphash/tests/data/photo.png
prints::
group of 2:
native/libphash/tests/data/photo.jpeg
native/libphash/tests/data/photo_copy.jpeg
group of 2:
native/libphash/tests/data/jpeg/s420.jpg
native/libphash/tests/data/jpeg/s444.jpg
"""
from __future__ import annotations
import sys
from dataclasses import dataclass, field
import libphash as ph
@dataclass
class Node:
"""A hash, the items that have it, and children keyed by their distance to it."""
value: int
items: list[str]
children: dict[int, Node] = field(default_factory=dict)
class BKTree:
"""A BK-tree of 64-bit hashes under the Hamming distance."""
def __init__(self) -> None:
self.root: Node | None = None
def add(self, value: int, item: str) -> None:
"""Index ``item`` under its hash ``value``."""
if self.root is None:
self.root = Node(value, [item])
return
node = self.root
while True:
d = ph.hamming_distance(value, node.value)
if d == 0:
node.items.append(item)
return
if d not in node.children:
node.children[d] = Node(value, [item])
return
node = node.children[d]
def search(self, value: int, max_bits: int) -> list[str]:
"""Return the items whose hash is within ``max_bits`` of ``value``."""
found: list[str] = []
pending = [self.root] if self.root else []
while pending:
node = pending.pop()
d = ph.hamming_distance(value, node.value)
if d <= max_bits:
found += node.items
# Only children at a distance within max_bits of d can hold a match.
for key, child in node.children.items():
if d - max_bits <= key <= d + max_bits:
pending.append(child)
return found
def main(argv: list[str]) -> int:
if len(argv) < 3:
print(f"usage: {argv[0]} <max-bits> <image>...", file=sys.stderr)
return 1
max_bits = int(argv[1])
hashes = {
r.source: r.hashes[ph.Algorithm.PHASH]
for r in ph.hash_files(argv[2:], ph.Algorithm.PHASH)
if r.hashes
}
tree = BKTree()
for path, value in hashes.items():
tree.add(value, path)
# Each image not yet grouped starts a group, which grows by its matches' matches.
grouped: set[str] = set()
for path in hashes:
if path in grouped:
continue
group, pending = {path}, [path]
while pending:
for match in tree.search(hashes[pending.pop()], max_bits):
if match not in group:
group.add(match)
pending.append(match)
grouped |= group
if len(group) > 1:
print(f"group of {len(group)}:")
for member in sorted(group, key=list(hashes).index):
print(f" {member}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Threads, processes and asyncio¶
"""Hash many files concurrently: a batch, threads from asyncio, and processes.
* ``hash_files`` runs libphash's own pool of threads; from asyncio it goes to a thread
(``asyncio.to_thread``) so the event loop stays free.
* Threads of your own each need their own ``ImageContext`` (one per thread, kept in a
``threading.local``): one shared context serializes the work on its lock.
* Processes (``ProcessPoolExecutor``) suit a corpus too large for one machine's memory
budget per process, or work that does more in Python per image than hashing; give each
a slice of the files. Results and errors pickle.
All three give the same hashes. Run::
python examples/concurrency.py native/libphash/tests/data/photo.jpeg native/libphash/tests/data/photo_copy.jpeg native/libphash/tests/data/photo.png native/libphash/tests/data/photo_complex.webp
prints::
batch from asyncio: 4 files
threads: 4 files
processes: 4 files
the same hashes three ways
"""
from __future__ import annotations
import asyncio
import sys
import threading
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import libphash as ph
_local = threading.local()
def hash_in_thread(path: str) -> int:
"""Hash one file with this thread's own context."""
ctx = getattr(_local, "ctx", None)
if ctx is None:
ctx = _local.ctx = ph.ImageContext()
ctx.load_file(path)
return ctx.phash()
def hash_slice(paths: list[str]) -> dict[str, int]:
"""Hash a slice of the files in a worker process, on a batch of its own."""
results = ph.hash_files(paths, ph.Algorithm.PHASH, threads=1)
return {r.source: r.hashes[ph.Algorithm.PHASH] for r in results if r.hashes}
async def with_asyncio(paths: list[str]) -> dict[str, int]:
results = await asyncio.to_thread(ph.hash_files, paths, ph.Algorithm.PHASH)
return {r.source: r.hashes[ph.Algorithm.PHASH] for r in results if r.hashes}
def main(argv: list[str]) -> int:
paths = argv[1:]
if not paths:
print(f"usage: {argv[0]} <image>...", file=sys.stderr)
return 1
batch = asyncio.run(with_asyncio(paths))
print(f"batch from asyncio: {len(batch)} files")
with ThreadPoolExecutor(max_workers=4) as pool:
threads = dict(zip(paths, pool.map(hash_in_thread, paths), strict=True))
print(f"threads: {len(threads)} files")
slices = [paths[i::2] for i in range(2)]
processes: dict[str, int] = {}
with ProcessPoolExecutor(max_workers=2) as pool:
for part in pool.map(hash_slice, slices):
processes.update(part)
print(f"processes: {len(processes)} files")
if not batch == threads == processes:
print("the hashes differ")
return 1
print("the same hashes three ways")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Choosing a threshold¶
"""Choose a threshold for your own images from pairs you have labeled.
A threshold trades missed copies against false matches, and where it lies depends on
the images and on the edits a copy goes through: measure it. List pairs of images as
"same" picture or "different" in a CSV file (``first,second,label``, paths relative to
the file), and this prints, for each threshold, how many copies it misses and how many
different pairs it takes for copies; the largest threshold without a false match keeps
every margin there is. libphash's guide explains the method and its pitfalls:
https://gudoshnikovn.github.io/libphash/theory/comparing/. Run::
python examples/threshold.py examples/pairs.csv
prints::
bits missed copies false matches
0 2 of 6 0 of 6
2 0 of 6 0 of 6
4 0 of 6 0 of 6
6 0 of 6 0 of 6
8 0 of 6 1 of 6
10 0 of 6 1 of 6
12 0 of 6 2 of 6
largest threshold without a false match: 6 bits
"""
from __future__ import annotations
import csv
import sys
from pathlib import Path
import libphash as ph
def main(argv: list[str]) -> int:
if len(argv) != 2:
print(f"usage: {argv[0]} <pairs.csv>", file=sys.stderr)
return 1
table = Path(argv[1])
same: list[int] = []
different: list[int] = []
with table.open(newline="") as f:
for row in csv.DictReader(f):
a = ph.hash(table.parent / row["first"])
b = ph.hash(table.parent / row["second"])
(same if row["label"] == "same" else different).append(
ph.hamming_distance(a, b)
)
print("bits missed copies false matches")
for bits in range(0, 13, 2):
missed = sum(d > bits for d in same)
false = sum(d <= bits for d in different)
missed_text = f"{missed} of {len(same)}"
false_text = f"{false} of {len(different)}"
print(f"{bits:4d} {missed_text:>13} {false_text:>13}")
closest_different = min(different, default=65)
print(f"largest threshold without a false match: {closest_different - 1} bits")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))