diff --git a/imageset_thumbnails.py b/imageset_thumbnails.py new file mode 100755 index 0000000..c8ce2c7 --- /dev/null +++ b/imageset_thumbnails.py @@ -0,0 +1,413 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["Pillow"] +# /// + +import argparse +import atexit +import logging +import os +import re +import shutil +import subprocess +import sys +from concurrent.futures import FIRST_COMPLETED, Future, ProcessPoolExecutor, wait +from contextlib import ExitStack +from dataclasses import dataclass +from functools import partial +from io import BytesIO +from multiprocessing import get_context +from pathlib import Path +from time import perf_counter +from typing import BinaryIO, Callable +from zipfile import BadZipFile, ZipFile, ZipInfo, is_zipfile + +from PIL import Image, ImageColor, ImageOps + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") +worker_archive: ZipFile | None = None + + +@dataclass +class Source: + name: str + path: Path | None = None + member: ZipInfo | None = None + ratio: float = 1.0 + + +@dataclass +class Row: + sources: list[Source] + height: int + widths: list[int] + + +@dataclass +class Cell: + source: Source + position: tuple[int, int] + size: tuple[int, int] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Create a separate justified image contact sheet for each directory or ZIP.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("inputs", nargs="+", type=Path, help="Image directories or ZIP archives; missing paths are skipped") + parser.add_argument("--output", type=Path, help="JPEG output for a single existing input; defaults to _thumbs.jpg beside each input") + parser.add_argument("--width", type=int, default=2500, help="Sheet width in pixels") + parser.add_argument("--target-height", type=int, default=300, help="Preferred row height in pixels; rows may be taller or shorter") + parser.add_argument("--gap", type=int, default=8, help="Gap between images and at sheet edges in pixels") + parser.add_argument("--bg-color", default="white", help="Background color as a color name or hex value") + parser.add_argument("--workers", type=int, default=os.cpu_count() or 1, help="Number of resize worker processes") + arguments = parser.parse_args() + if arguments.target_height <= 0 or arguments.width <= 0 or arguments.gap < 0: + parser.error("width and target height must be positive; gap must be nonnegative") + if arguments.width <= 2 * arguments.gap: + parser.error("width must exceed twice the gap") + if arguments.workers <= 0: + parser.error("workers must be positive") + try: + ImageColor.getcolor(color=arguments.bg_color, mode="RGB") + except ValueError as error: + parser.error(f"Invalid background color: {error}") + return arguments + + +def read_image(source: Source, archive: ZipFile | None) -> Image.Image: + with ExitStack() as stack: + if source.member is not None: + if archive is None: + raise ValueError("ZIP source requires an open archive") + stream = stack.enter_context(archive.open(name=source.member)) + original = stack.enter_context(Image.open(fp=stream)) + else: + original = stack.enter_context(Image.open(fp=source.path)) + oriented = ImageOps.exif_transpose(image=original) + result = oriented.convert(mode="RGBA") + if oriented is not original: + oriented.close() + return result + + +def read_dims_jpeg(stream: BinaryIO) -> tuple[int, int]: + EXIF_ORIENTATION_TAG = 274 + ORIENTATIONS_SWAPPING_WIDTH_HEIGHT = {5, 6, 7, 8} + with Image.open(fp=stream) as image: + width, height = image.size + if image.getexif().get(EXIF_ORIENTATION_TAG) in ORIENTATIONS_SWAPPING_WIDTH_HEIGHT: + return height, width + return width, height + + +def read_dims(image_path: Path | BinaryIO) -> tuple[int, int]: + EXIF_ORIENTATION_TAG = 274 + ORIENTATIONS_SWAPPING_WIDTH_HEIGHT = {5, 6, 7, 8} + with ExitStack() as stack: + stream = stack.enter_context(image_path.open(mode="rb")) if isinstance(image_path, Path) else image_path + signature = stream.read(12) + stream.seek(0) + if signature.startswith(b"\xff\xd8"): + return read_dims_jpeg(stream=stream) + if signature.startswith(b"\x89PNG\r\n\x1a\n"): + raise ValueError("PNG images are not supported") + if signature[:4] == b"RIFF" and signature[8:12] == b"WEBP": + raise ValueError("WebP images are not supported") + with Image.open(fp=stream) as image: + if image.format == "TIFF": + raise ValueError("TIFF images are not supported") + width, height = image.size + if image.getexif().get(EXIF_ORIENTATION_TAG) in ORIENTATIONS_SWAPPING_WIDTH_HEIGHT: + return height, width + return width, height + + +def natural_sort_key(name: str) -> tuple[str | int, ...]: + return tuple(int(part) if index % 2 else part.casefold() for index, part in enumerate(re.split(pattern=r"([0-9]+)", string=name))) + + +def discover_sources(input_path: Path, output_path: Path, archive: ZipFile | None) -> list[Source]: + Image.init() + extensions = {extension for extension, image_format in Image.registered_extensions().items() if image_format not in {"PNG", "WEBP", "TIFF"}} + if archive is not None: + candidates = [ + Source(name=member.filename, member=member) + for member in archive.infolist() + if not member.is_dir() and Path(member.filename).suffix.lower() in extensions + ] + else: + candidates = [ + Source(name=path.relative_to(input_path).as_posix(), path=path) + for path in input_path.rglob("*") + if path.is_file() and path.suffix.lower() in extensions and path.resolve() != output_path.resolve() + ] + sources = [] + logging.info(f"Reading metadata for {len(candidates)} images") + for index, source in enumerate(sorted(candidates, key=lambda source: (natural_sort_key(name=source.name), source.name)), start=1): + try: + if source.member is not None and archive is not None: + with archive.open(name=source.member) as stream: + width, height = read_dims(image_path=stream) + elif source.path is not None: + width, height = read_dims(image_path=source.path) + else: + raise ValueError("Image source has no file or ZIP member") + source.ratio = width / height + sources.append(source) + except (OSError, ValueError, SyntaxError, RuntimeError, BadZipFile, Image.DecompressionBombError) as error: + logging.warning(f"Skipping {source.name}: {error}") + if index % 100 == 0 or index == len(candidates): + logging.info(f"Metadata {index}/{len(candidates)} ({index / len(candidates):.0%}); {len(sources)} usable images") + if not sources: + raise ValueError("No usable images found") + return sources + + +def fitted_height(sources: list[Source], width: int, gap: int) -> float: + available = width - gap * (len(sources) + 1) + return available / sum(source.ratio for source in sources) + + +def make_row(sources: list[Source], width: int, gap: int, target_height: int, justify: bool) -> Row: + available = width - gap * (len(sources) + 1) + ideal_height = fitted_height(sources=sources, width=width, gap=gap) + height = max(1, round(ideal_height if justify else min(target_height, ideal_height))) + ratio_sum = sum(source.ratio for source in sources) + total_width = available if justify else min(available, max(len(sources), round(ratio_sum * height))) + widths = [] + cumulative = 0.0 + previous = 0 + for index, source in enumerate(sources): + cumulative += source.ratio + boundary = min( + total_width - (len(sources) - index - 1), + max(previous + 1, round(total_width * cumulative / ratio_sum)), + ) + widths.append(boundary - previous) + previous = boundary + return Row(sources=sources, height=height, widths=widths) + + +def layout_rows(sources: list[Source], width: int, gap: int, target_height: int) -> list[Row]: + rows = [] + start = 0 + while start < len(sources): + end = start + 1 + while end < len(sources) and fitted_height(sources=sources[start:end], width=width, gap=gap) > target_height: + candidate = sources[start : end + 1] + if width - gap * (len(candidate) + 1) < len(candidate): + break + candidate_height = fitted_height(sources=candidate, width=width, gap=gap) + current_height = fitted_height(sources=sources[start:end], width=width, gap=gap) + if candidate_height <= target_height and abs(current_height - target_height) <= abs(candidate_height - target_height): + break + end += 1 + group = sources[start:end] + justify = end < len(sources) or fitted_height(sources=group, width=width, gap=gap) <= target_height + rows.append(make_row(sources=group, width=width, gap=gap, target_height=target_height, justify=justify)) + start = end + return rows + + +def initialize_worker(archive_path: Path | None) -> None: + global worker_archive + if archive_path is not None: + worker_archive = ZipFile(file=archive_path) + atexit.register(worker_archive.close) + + +def resize_with_pillow(cell: Cell) -> bytes: + with read_image(source=cell.source, archive=worker_archive) as image: + with image.resize(size=cell.size, resample=Image.Resampling.LANCZOS) as thumbnail: + return thumbnail.tobytes() + + +def resize_with_vips(cell: Cell, vips_binary: str) -> bytes: + input_bytes = None + if cell.source.member is not None: + if worker_archive is None: + raise ValueError("ZIP source requires an open archive") + input_bytes = worker_archive.read(name=cell.source.member) + operation, input_name = "thumbnail_source", "[descriptor=0]" + elif cell.source.path is not None: + operation, input_name = "thumbnail", str(cell.source.path) + else: + raise ValueError("Image source has no file or ZIP member") + try: + result = subprocess.run( + args=[ + vips_binary, + "--vips-concurrency=1", + operation, + input_name, + ".png", + str(cell.size[0]), + "--height", + str(cell.size[1]), + "--size", + "both", + ], + input=input_bytes, + check=True, + capture_output=True, + ) + except (OSError, subprocess.CalledProcessError) as error: + detail = error.stderr.decode(errors="replace").strip() if isinstance(error, subprocess.CalledProcessError) and error.stderr else str(error) + logging.warning(f"vips failed for {cell.source.name}; using Pillow: {detail}") + return resize_with_pillow(cell=cell) + with Image.open(fp=BytesIO(result.stdout)) as image: + with image.convert(mode="RGBA") as thumbnail: + if thumbnail.size == cell.size: + return thumbnail.tobytes() + with thumbnail.resize(size=cell.size, resample=Image.Resampling.LANCZOS) as exact_thumbnail: + return exact_thumbnail.tobytes() + + +def resize_thumbnail(cell: Cell, processor: Callable[..., bytes]) -> tuple[bytes, float]: + started = perf_counter() + logging.info(f"Resizing {cell.source.name} to {cell.size[0]}x{cell.size[1]}") + pixels = processor(cell=cell) + return pixels, perf_counter() - started + + +def render_rows( + sheet: Image.Image, + rows: list[Row], + input_path: Path, + workers: int, + gap: int, + processor: Callable[..., bytes], +) -> int: + cells = [] + top = gap + for row in rows: + left = gap + for source, cell_width in zip(row.sources, row.widths): + cells.append(Cell(source=source, position=(left, top), size=(cell_width, row.height))) + left += cell_width + gap + top += row.height + gap + remaining = iter(cells) + pending: dict[Future[tuple[bytes, float]], Cell] = {} + processed_count = rendered_count = 0 + archive_path = None if input_path.is_dir() else input_path + with ProcessPoolExecutor( + max_workers=workers, + mp_context=get_context(method="spawn"), + initializer=initialize_worker, + initargs=(archive_path,), + ) as pool: + while True: + while len(pending) < workers * 2: + cell = next(remaining, None) + if cell is None: + break + future = pool.submit(resize_thumbnail, cell=cell, processor=processor) + pending[future] = cell + if not pending: + break + completed, _ = wait(fs=pending, return_when=FIRST_COMPLETED) + for future in completed: + cell = pending.pop(future) + processed_count += 1 + try: + pixels, duration = future.result() + with Image.frombytes(mode="RGBA", size=cell.size, data=pixels) as thumbnail: + sheet.paste(im=thumbnail, box=cell.position, mask=thumbnail) + rendered_count += 1 + logging.info(f"Placed {cell.source.name} (resized in {duration:.1f}s)") + except (OSError, ValueError, SyntaxError, RuntimeError, BadZipFile, Image.DecompressionBombError) as error: + logging.warning(f"Skipping {cell.source.name} during rendering: {error}") + logging.info(f"Progress {processed_count}/{len(cells)} ({processed_count / len(cells):.0%})") + return rendered_count + + +def create_sheet( + input_path: Path, + output_path: Path, + width: int, + target_height: int, + gap: int, + bg_color: str = "white", + workers: int | None = None, + processor: Callable[..., bytes] = resize_with_pillow, +) -> None: + if width <= 0 or target_height <= 0 or gap < 0 or width <= 2 * gap: + raise ValueError("Invalid sheet dimensions or gap") + if output_path.suffix.lower() not in {".jpg", ".jpeg"}: + raise ValueError("Output must have a .jpg or .jpeg extension") + if not input_path.is_dir() and not (input_path.is_file() and is_zipfile(filename=input_path)): + raise ValueError(f"Expected a directory or ZIP archive: {input_path}") + if input_path.resolve() == output_path.resolve(): + raise ValueError("Output must differ from input") + if workers is not None and workers <= 0: + raise ValueError("Workers must be positive") + background = ImageColor.getcolor(color=bg_color, mode="RGB") + with ExitStack() as stack: + archive = None if input_path.is_dir() else stack.enter_context(ZipFile(file=input_path)) + sources = discover_sources(input_path=input_path, output_path=output_path, archive=archive) + rows = layout_rows(sources=sources, width=width, gap=gap, target_height=target_height) + height = sum(row.height for row in rows) + gap * (len(rows) + 1) + worker_count = min(workers if workers is not None else os.cpu_count() or 1, len(sources)) + logging.info(f"Rendering {len(sources)} images in {len(rows)} rows on a {width}x{height} sheet using {worker_count} workers") + sheet = stack.enter_context(Image.new(mode="RGB", size=(width, height), color=background)) + started = perf_counter() + rendered_count = render_rows( + sheet=sheet, + rows=rows, + input_path=input_path, + workers=worker_count, + gap=gap, + processor=processor, + ) + if not rendered_count: + raise ValueError("No usable images could be decoded") + logging.info(f"Rendering finished in {perf_counter() - started:.1f}s; saving {output_path}") + output_path.parent.mkdir(parents=True, exist_ok=True) + sheet.save(fp=output_path, quality=95) + logging.info(f"Wrote {output_path} ({rendered_count} images, {width}x{height})") + + +def main() -> None: + arguments = parse_args() + input_paths = [] + for path in arguments.inputs: + if path.exists(): + input_paths.append(path.resolve()) + else: + logging.warning(f"Skipping non-existing input: {path}") + if not input_paths: + logging.error("No existing input paths") + sys.exit(1) + if arguments.output is not None and len(input_paths) > 1: + logging.error("--output requires a single existing input") + sys.exit(2) + vips_binary = shutil.which(cmd="vips") + processor = partial(resize_with_vips, vips_binary=vips_binary) if vips_binary else resize_with_pillow + logging.info(f"Resize backend: {vips_binary or 'Pillow'}") + failed_count = 0 + for index, input_path in enumerate(input_paths, start=1): + logging.info(f"Processing input {index}/{len(input_paths)}: {input_path}") + input_name = input_path.name if input_path.is_dir() else input_path.stem + output_path = arguments.output or input_path.with_name(f"{input_name}_thumbs.jpg") + try: + create_sheet( + input_path=input_path, + output_path=output_path, + width=arguments.width, + target_height=arguments.target_height, + gap=arguments.gap, + bg_color=arguments.bg_color, + workers=arguments.workers, + processor=processor, + ) + except (OSError, ValueError, RuntimeError, BadZipFile) as error: + failed_count += 1 + logging.error(f"Could not create contact sheet for {input_path}: {error}") + if failed_count: + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/imageset_thumbnails_test.py b/imageset_thumbnails_test.py new file mode 100755 index 0000000..0509fe6 --- /dev/null +++ b/imageset_thumbnails_test.py @@ -0,0 +1,476 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["pytest", "Pillow"] +# /// + +from functools import partial +from io import BytesIO +from pathlib import Path +import shutil +import subprocess +import sys +from zipfile import ZIP_DEFLATED, ZipFile + +from PIL import Image +import pytest + +import imageset_thumbnails as script + + +@pytest.fixture +def rendered_sheets(monkeypatch: pytest.MonkeyPatch) -> list[Image.Image]: + sheets = [] + original_save = Image.Image.save + + def capture_sheet(image: Image.Image, **keywords: object) -> None: + if Path(keywords["fp"]).suffix == ".jpg": + sheets.append(image.copy()) + original_save(image, **keywords) + + monkeypatch.setattr(Image.Image, "save", capture_sheet) + return sheets + + +def test_directory_and_zip_natural_order(tmp_path: Path) -> None: + directory = tmp_path / "images" + directory.mkdir() + for name, color in [("prefix_10.bmp", "blue"), ("prefix_2.bmp", "green"), ("prefix_1.bmp", "red"), ("nested_10/prefix_1.bmp", "yellow"), ("nested_2/prefix_1.bmp", "black")]: + path = directory / name + path.parent.mkdir(parents=True, exist_ok=True) + with Image.new(mode="RGB", size=(20, 20), color=color) as image: + image.save(fp=path) + (directory / "broken.bmp").write_bytes(data=b"invalid") + (directory / "notes.txt").write_text(data="not an image") + archive_path = tmp_path / "images.zip" + with ZipFile(file=archive_path, mode="w") as archive: + for path in sorted(directory.rglob("*"), reverse=True): + if path.is_file(): + archive.write(filename=path, arcname=path.relative_to(directory).as_posix()) + output = tmp_path / "directory.jpg" + sources = script.discover_sources(input_path=directory, output_path=output, archive=None) + assert [source.name for source in sources] == ["nested_2/prefix_1.bmp", "nested_10/prefix_1.bmp", "prefix_1.bmp", "prefix_2.bmp", "prefix_10.bmp"] + script.create_sheet(input_path=directory, output_path=output, width=68, target_height=20, gap=2) + zip_output = tmp_path / "zip.jpg" + script.create_sheet(input_path=archive_path, output_path=zip_output, width=68, target_height=20, gap=2) + with Image.open(fp=output) as first, Image.open(fp=zip_output) as second: + assert first.size == second.size + assert first.tobytes() == second.tobytes() + assert first.getpixel(xy=(10, 10)) == pytest.approx((0, 0, 0), abs=5) + assert first.getpixel(xy=(32, 10)) == pytest.approx((255, 255, 0), abs=5) + assert first.getpixel(xy=(54, 10)) == pytest.approx((255, 0, 0), abs=5) + + +@pytest.mark.parametrize("ratios,width,target,gap", [ + ([1.0] * 7, 100, 30, 2), + ([0.5, 2.0, 1.0, 0.75, 1.5], 100, 25, 3), + ([100.0], 100, 30, 2), + ([1.0] * 5, 90, 30, 0), + ([0.001, 100.0, 0.001], 100, 30, 1), + ([0.5] * 4, 3, 20, 1), +]) +def test_layout_geometry(ratios: list[float], width: int, target: int, gap: int) -> None: + sources = [script.Source(name=str(index), ratio=ratio) for index, ratio in enumerate(ratios)] + rows = script.layout_rows(sources=sources, width=width, target_height=target, gap=gap) + assert [source for row in rows for source in row.sources] == sources + for index, row in enumerate(rows): + assert row.height >= 1 + assert all(cell_width >= 1 for cell_width in row.widths) + occupied = sum(row.widths) + gap * (len(row.sources) + 1) + assert occupied <= width + if index < len(rows) - 1: + assert occupied == width + + +def test_render_gaps_equal_heights_and_last_row(tmp_path: Path, rendered_sheets: list[Image.Image]) -> None: + directory = tmp_path / "images" + directory.mkdir() + for index in range(4): + with Image.new(mode="RGB", size=(40, 40), color="red") as image: + image.save(fp=directory / f"image_{index}.bmp") + output = tmp_path / "sheet.jpg" + script.create_sheet(input_path=directory, output_path=output, width=100, target_height=30, gap=2) + with rendered_sheets[0] as sheet: + assert sheet.size == (100, 67) + for left in (2, 35, 67): + assert sheet.getpixel(xy=(left, 2)) == (255, 0, 0) + assert sheet.getpixel(xy=(left, 32)) == (255, 0, 0) + for position in [(0, 0), (33, 10), (65, 10), (99, 10), (10, 33), (32, 40)]: + assert sheet.getpixel(xy=position) == (255, 255, 255) + assert sheet.getpixel(xy=(2, 35)) == (255, 0, 0) + + +def test_orientation_and_transparency(tmp_path: Path) -> None: + path = tmp_path / "oriented.jpg" + with Image.new(mode="RGB", size=(40, 20), color="red") as image: + exif = Image.Exif() + exif[274] = 6 + image.save(fp=path, exif=exif) + with script.read_image(source=script.Source(name=path.name, path=path), archive=None) as image: + assert image.size == (20, 40) + path = tmp_path / "transparent.gif" + with Image.new(mode="RGBA", size=(10, 10), color=(255, 0, 0, 0)) as image: + image.save(fp=path) + with script.read_image(source=script.Source(name=path.name, path=path), archive=None) as image: + assert image.mode == "RGBA" + assert image.getpixel(xy=(0, 0)) == (255, 0, 0, 0) + + +@pytest.mark.parametrize("bg_color,expected", [("white", (255, 255, 255)), ("black", (0, 0, 0)), ("#123456", (18, 52, 86))]) +@pytest.mark.parametrize("use_vips", [False, True]) +def test_sheet_background( + tmp_path: Path, rendered_sheets: list[Image.Image], bg_color: str, + expected: tuple[int, int, int], use_vips: bool, +) -> None: + vips_binary = shutil.which(cmd="vips") if use_vips else None + if use_vips and vips_binary is None: + pytest.skip("vips is not installed") + processor = partial(script.resize_with_vips, vips_binary=vips_binary) if vips_binary else script.resize_with_pillow + directory = tmp_path / "images" + directory.mkdir() + with Image.new(mode="RGBA", size=(20, 20), color=(255, 0, 0, 0)) as image: + image.save(fp=directory / "transparent.gif") + with Image.new(mode="RGB", size=(20, 20), color="red") as image: + image.save(fp=directory / "red.bmp") + output = tmp_path / "sheet.jpg" + script.create_sheet(input_path=directory, output_path=output, width=100, target_height=20, gap=2, bg_color=bg_color, processor=processor) + with rendered_sheets[0] as sheet: + assert sheet.getpixel(xy=(0, 0)) == expected + assert sheet.getpixel(xy=(22, 5)) == expected + assert sheet.getpixel(xy=(25, 5)) == expected + assert sheet.getpixel(xy=(99, 5)) == expected + assert sheet.getpixel(xy=(3, 5)) == (255, 0, 0) + + +@pytest.mark.parametrize("extension", ["jpg"]) +@pytest.mark.parametrize("orientation", [1, 6, 8]) +def test_metadata_dimensions_without_pixel_decoding( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + extension: str, orientation: int, +) -> None: + directory = tmp_path / "images" + directory.mkdir() + path = directory / f"image.{extension}" + with Image.new(mode="RGB", size=(40, 20), color="red") as image: + exif = Image.Exif() + exif[274] = orientation + image.save(fp=path, exif=exif) + archive_path = tmp_path / "images.zip" + with ZipFile(file=archive_path, mode="w") as archive: + archive.write(filename=path, arcname=path.name) + + def reject_decoding(image: Image.Image, *arguments: object, **keywords: object) -> None: + pytest.fail("Metadata discovery must not decode image pixels") + + Image.init() + monkeypatch.setattr(Image.Image, "load", reject_decoding) + expected = (40, 20) if orientation == 1 else (20, 40) + assert script.read_dims(image_path=path) == expected + sources = script.discover_sources(input_path=directory, output_path=tmp_path / "sheet.jpg", archive=None) + assert sources[0].ratio == expected[0] / expected[1] + with ZipFile(file=archive_path) as archive: + with archive.open(name=path.name) as stream: + assert script.read_dims(image_path=stream) == expected + sources = script.discover_sources(input_path=archive_path, output_path=tmp_path / "sheet.jpg", archive=archive) + assert sources[0].ratio == expected[0] / expected[1] + + +def test_rendering_skips_pixel_corruption(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + directory = tmp_path / "images" + directory.mkdir() + valid = directory / "1.bmp" + broken = directory / "2.jpg" + with Image.new(mode="RGB", size=(20, 20), color="red") as image: + image.save(fp=valid) + image.save(fp=broken) + broken.write_bytes(data=broken.read_bytes()[:-10]) + assert script.read_dims(image_path=broken) == (20, 20) + output = tmp_path / "sheet.jpg" + script.create_sheet(input_path=directory, output_path=output, width=100, target_height=20, gap=0) + assert "Skipping 2.jpg during rendering" in caplog.text + with Image.open(fp=output) as sheet: + assert sheet.getpixel(xy=(10, 10)) == pytest.approx((255, 0, 0), abs=5) + valid.unlink() + with pytest.raises(ValueError, match="No usable images could be decoded"): + script.create_sheet(input_path=directory, output_path=tmp_path / "failed.jpg", width=100, target_height=20, gap=0) + assert not (tmp_path / "failed.jpg").exists() + + +@pytest.mark.parametrize("extension,expected_format", [("jpg", "JPEG"), ("jpeg", "JPEG")]) +def test_output_format_and_exclusion(tmp_path: Path, extension: str, expected_format: str) -> None: + with Image.new(mode="RGB", size=(20, 20), color="blue") as image: + image.save(fp=tmp_path / "input.bmp") + output = tmp_path / f"sheet.{extension}" + script.create_sheet(input_path=tmp_path, output_path=output, width=100, target_height=20, gap=0) + sources = script.discover_sources(input_path=tmp_path, output_path=output, archive=None) + assert [source.name for source in sources] == ["input.bmp"] + with Image.open(fp=output) as sheet: + assert sheet.format == expected_format + assert sheet.size == (100, 20) + + +def test_first_animation_frame(tmp_path: Path) -> None: + path = tmp_path / "animated.gif" + with Image.new(mode="RGB", size=(10, 10), color="red") as first, Image.new(mode="RGB", size=(10, 10), color="blue") as second: + first.save(fp=path, save_all=True, append_images=[second]) + with script.read_image(source=script.Source(name=path.name, path=path), archive=None) as image: + assert image.getpixel(xy=(0, 0)) == (255, 0, 0, 255) + + +@pytest.mark.parametrize("input_kind", ["directory", "zip"]) +def test_cli_process_pool_matches_single_worker(tmp_path: Path, input_kind: str) -> None: + directory = tmp_path / "images" + directory.mkdir() + for index in [10, 2, 1, 3, 4, 5, 6]: + with Image.new(mode="RGBA", size=(20 + index * 5, 40), color=(index * 20, 100, 50, 128)) as image: + exif = Image.Exif() + exif[274] = 6 if index == 2 else 1 + image.convert(mode="RGB").save(fp=directory / f"image_{index}.jpg", exif=exif) + input_path = directory + if input_kind == "zip": + input_path = tmp_path / "images.zip" + with ZipFile(file=input_path, mode="w", compression=ZIP_DEFLATED) as archive: + for path in directory.iterdir(): + archive.write(filename=path, arcname=f"nested/{path.name}") + outputs = [] + for workers in [1, 2]: + output = tmp_path / f"workers_{workers}.jpg" + result = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(input_path), + "--output", str(output), "--workers", str(workers), "--width", "180", + "--target-height", "40", "--gap", "3", "--bg-color", "#123456"], + check=True, capture_output=True, text=True, timeout=30, + ) + assert "Progress 7/7 (100%)" in result.stderr + with Image.open(fp=output) as image: + outputs.append((image.size, image.tobytes())) + assert outputs[0] == outputs[1] + + +@pytest.mark.parametrize("options", [ + ["--width", "0"], ["--target-height", "0"], ["--gap", "-1"], + ["--width", "16", "--gap", "8"], + ["--bg-color", "invalid-color"], + ["--workers", "0"], ["--workers", "-1"], +]) +def test_invalid_arguments(monkeypatch: pytest.MonkeyPatch, options: list[str]) -> None: + monkeypatch.setattr(sys, "argv", ["imageset_thumbnails.py", "images", *options]) + with pytest.raises(SystemExit) as error: + script.parse_args() + assert error.value.code == 2 + + +def test_cli_names(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sys, "argv", ["imageset_thumbnails.py", "images", "--target-height", "180", "--gap", "4", "--bg-color", "#123456", "--workers", "3"]) + arguments = script.parse_args() + assert arguments.target_height == 180 + assert arguments.gap == 4 + assert arguments.bg_color == "#123456" + assert arguments.workers == 3 + assert arguments.inputs == [Path("images")] + + +@pytest.mark.parametrize("bad_input_kind,expected_status", [("missing", 0), ("invalid", 1)]) +def test_cli_multiple_inputs(tmp_path: Path, bad_input_kind: str, expected_status: int) -> None: + directory = tmp_path / "photos" + directory.mkdir() + with Image.new(mode="RGB", size=(20, 20), color="red") as image: + image.save(fp=directory / "image.bmp") + archive_path = tmp_path / "archive.zip" + with ZipFile(file=archive_path, mode="w", compression=ZIP_DEFLATED) as archive: + archive.write(filename=directory / "image.bmp", arcname="image.bmp") + bad_input = tmp_path / "bad_input" + if bad_input_kind == "invalid": + bad_input.write_text(data="not a ZIP") + result = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(bad_input), + str(directory), str(archive_path), "--workers", "2"], + check=False, capture_output=True, text=True, timeout=30, + ) + assert result.returncode == expected_status + if bad_input_kind == "missing": + assert "WARNING: Skipping non-existing input" in result.stderr + else: + assert "Could not create contact sheet for" in result.stderr + for output in [tmp_path / "photos_thumbs.jpg", tmp_path / "archive_thumbs.jpg"]: + with Image.open(fp=output) as sheet: + assert sheet.size == (2500, 316) + + +def test_cli_all_inputs_missing(tmp_path: Path) -> None: + result = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), + str(tmp_path / "missing.zip"), str(tmp_path / "missing_directory")], + check=False, capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 1 + assert result.stderr.count("Skipping non-existing input") == 2 + assert "No existing input paths" in result.stderr + assert list(tmp_path.iterdir()) == [] + + +def test_cli_output_requires_single_existing_input(tmp_path: Path) -> None: + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + output = tmp_path / "sheet.jpg" + result = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(first), + str(second), "--output", str(output)], + check=False, capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 2 + assert "--output requires a single existing input" in result.stderr + assert not output.exists() + with Image.new(mode="RGB", size=(20, 20), color="blue") as image: + image.save(fp=first / "image.bmp") + result = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(first), + str(tmp_path / "missing"), "--output", str(output)], + check=True, capture_output=True, text=True, timeout=30, + ) + assert "Skipping non-existing input" in result.stderr + assert output.exists() + + +@pytest.mark.parametrize("use_vips", [False, True]) +def test_main_selects_processor_once( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, use_vips: bool, +) -> None: + vips_binary = shutil.which(cmd="vips") if use_vips else None + if use_vips and vips_binary is None: + pytest.skip("vips is not installed") + directories = [tmp_path / "first", tmp_path / "second"] + for directory in directories: + directory.mkdir() + with Image.new(mode="RGB", size=(40, 20), color="red") as image: + image.save(fp=directory / "image_1.bmp") + image.save(fp=directory / "image_2.bmp") + lookups = [] + + def find_binary(cmd: str) -> str | None: + lookups.append(cmd) + return vips_binary + + monkeypatch.setattr(script.shutil, "which", find_binary) + monkeypatch.setattr(sys, "argv", ["imageset_thumbnails.py", *map(str, directories), + "--width", "100", "--target-height", "20", "--workers", "2"]) + script.main() + assert lookups == ["vips"] + for name in ["first", "second"]: + with Image.open(fp=tmp_path / f"{name}_thumbs.jpg") as sheet: + assert sheet.width == 100 + + +@pytest.mark.parametrize("input_kind", ["directory", "zip"]) +def test_vips_resizes_with_legacy_options_before_pillow_reads_pixels( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, input_kind: str, +) -> None: + vips_binary = shutil.which(cmd="vips") + if vips_binary is None: + pytest.skip("vips is not installed") + path = tmp_path / "image.jpg" + with Image.new(mode="RGB", size=(400, 200), color="red") as image: + image.paste(im="blue", box=(200, 0, 400, 200)) + exif = Image.Exif() + exif[274] = 6 + image.save(fp=path, exif=exif) + cell = script.Cell(source=script.Source(name=path.name, path=path), position=(0, 0), size=(10, 20)) + archive_path = tmp_path / "images.zip" + with ZipFile(file=archive_path, mode="w") as archive: + archive.write(filename=path, arcname=path.name) + original_open = Image.open + original_run = subprocess.run + reads = [] + + def run_legacy_vips(args: list[str], **keywords: object) -> subprocess.CompletedProcess[str]: + assert ".png" in args + if input_kind == "zip": + assert "thumbnail_source" in args + assert keywords["input"] == path.read_bytes() + else: + assert "thumbnail" in args + assert keywords["input"] is None + if "--fail-on" in args: + raise subprocess.CalledProcessError(returncode=1, cmd=args, stderr=b"Unknown option --fail-on") + return original_run(args=args, **keywords) + + def read_thumbnail(fp: BytesIO) -> Image.Image: + assert isinstance(fp, BytesIO) + image = original_open(fp=fp) + reads.append(image.size) + return image + + monkeypatch.setattr(script.Image, "open", read_thumbnail) + monkeypatch.setattr(script.subprocess, "run", run_legacy_vips) + with ZipFile(file=archive_path) as archive: + if input_kind == "zip": + cell.source = script.Source(name=path.name, member=archive.getinfo(name=path.name)) + monkeypatch.setattr(script, "worker_archive", archive) + pixels = script.resize_with_vips(cell=cell, vips_binary=vips_binary) + assert reads == [cell.size] + with Image.frombytes(mode="RGBA", size=cell.size, data=pixels) as thumbnail: + assert thumbnail.getpixel(xy=(5, 2)) == pytest.approx((255, 0, 0, 255), abs=5) + assert thumbnail.getpixel(xy=(5, 17)) == pytest.approx((0, 0, 255, 255), abs=5) + + +def test_vips_failure_falls_back_to_pillow( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, +) -> None: + path = tmp_path / "image.gif" + with Image.new(mode="RGBA", size=(40, 20), color=(255, 0, 0, 128)) as image: + image.save(fp=path) + cell = script.Cell(source=script.Source(name=path.name, path=path), position=(0, 0), size=(10, 5)) + expected = script.resize_with_pillow(cell=cell) + + def fail_command(args: list[str], **keywords: object) -> subprocess.CompletedProcess[str]: + raise subprocess.CalledProcessError(returncode=1, cmd=args, stderr=b"unsupported format") + + monkeypatch.setattr(script.subprocess, "run", fail_command) + assert script.resize_with_vips(cell=cell, vips_binary="vips") == expected + assert "using Pillow: unsupported format" in caplog.text + + +@pytest.mark.parametrize("input_kind", ["directory", "zip"]) +@pytest.mark.parametrize("image_format,format_name", [("PNG", "PNG"), ("WEBP", "WebP"), ("TIFF", "TIFF")]) +def test_unsupported_inputs_are_skipped(tmp_path: Path, input_kind: str, image_format: str, format_name: str) -> None: + directory = tmp_path / "images" + directory.mkdir() + with Image.new(mode="RGB", size=(20, 20), color="red") as image: + image.save(fp=directory / "supported.jpg") + image.save(fp=directory / f"ignored.{image_format}", format=image_format) + image.save(fp=directory / "disguised.jpg", format=image_format) + output = tmp_path / "sheet.jpg" + with pytest.raises(ValueError, match=f"{format_name} images are not supported"): + script.read_dims(image_path=directory / f"ignored.{image_format}") + if input_kind == "zip": + archive_path = tmp_path / "images.zip" + with ZipFile(file=archive_path, mode="w") as archive: + for path in directory.iterdir(): + archive.write(filename=path, arcname=path.name) + with ZipFile(file=archive_path) as archive: + sources = script.discover_sources(input_path=archive_path, output_path=output, archive=archive) + else: + sources = script.discover_sources(input_path=directory, output_path=output, archive=None) + assert [source.name for source in sources] == ["supported.jpg"] + + +@pytest.mark.parametrize("extension", ["png", "webp", "tif", "tiff"]) +def test_unsupported_output_is_rejected(tmp_path: Path, extension: str) -> None: + output = tmp_path / f"sheet.{extension}" + with pytest.raises(ValueError, match="Output must have a .jpg or .jpeg extension"): + script.create_sheet(input_path=tmp_path, output_path=output, width=100, target_height=20, gap=0) + assert not output.exists() + + +def test_empty_and_invalid_input(tmp_path: Path) -> None: + output = tmp_path / "sheet.jpg" + with pytest.raises(ValueError, match="No usable images"): + script.create_sheet(input_path=tmp_path, output_path=output, width=100, target_height=20, gap=0) + with pytest.raises(ValueError, match="Expected a directory or ZIP"): + script.create_sheet(input_path=tmp_path / "missing", output_path=output, width=100, target_height=20, gap=0) + assert not output.exists() + + +if __name__ == "__main__": + sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))