API Reference

Core optimization

optimize_image(source: Path | str, output: Path | str | None = None, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, strip_metadata: bool = True, output_format: OutputFormat = OutputFormat.AUTO, keep_aspect_ratio: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE, auto_orient: bool = True, progressive: bool = True, optimize: bool = True, overwrite: bool = False, lossless: bool = False, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, keep_exif_groups: set[EXIFGroup] | None = None) -> OptimizationResult

Optimize a single image.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str | None	Path for the optimized image. If None, overwrites source (if overwrite=True).	None
max_width	int | None	Maximum width in pixels. None means no resize.	None
max_height	int | None	Maximum height in pixels. None means no resize.	None
quality	int	JPEG/WEBP quality (1-100). Higher is better quality, larger file.	85
strip_metadata	bool	Remove EXIF and other metadata.	True
output_format	OutputFormat	Target format. AUTO infers from output path or original.	AUTO
keep_aspect_ratio	bool	Maintain aspect ratio when resizing (legacy when fit is None).	True
fit	FitMode | str | None	Resize fit mode: down, cover, contain, fill.	None
anchor	Anchor | str	Anchor point for cover/contain cropping and positioning.	CENTER
aspect_ratio	tuple[int, int] | str | None	Target aspect ratio as '16:9' or (16, 9).	None
background_color	tuple[int, int, int] | str	RGB tuple or hex color for contain padding.	WHITE
auto_orient	bool	Apply EXIF orientation before processing.	True
progressive	bool	Use progressive JPEG encoding.	True
optimize	bool	Enable Pillow optimization flags.	True
overwrite	bool	Allow overwriting the source file when output is None.	False
lossless	bool	Use lossless compression for PNG/WEBP. Ignored for JPEG.	False
backup_dir	Path | str | None	Directory to copy the original file into before processing.	None
min_size_bytes	int | None	Skip files already smaller than this threshold (bytes).	None

Returns:

Type	Description
OptimizationResult	OptimizationResult with details of the operation.

Source code in pixopt/optimizer.py

def optimize_image(
    source: Path | str,
    output: Path | str | None = None,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    strip_metadata: bool = True,
    output_format: OutputFormat = OutputFormat.AUTO,
    keep_aspect_ratio: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
    auto_orient: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    overwrite: bool = False,
    lossless: bool = False,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    keep_exif_groups: set[EXIFGroup] | None = None,
) -> OptimizationResult:
    """Optimize a single image.

    Args:
        source: Path to the source image.
        output: Path for the optimized image. If None, overwrites source (if overwrite=True).
        max_width: Maximum width in pixels. None means no resize.
        max_height: Maximum height in pixels. None means no resize.
        quality: JPEG/WEBP quality (1-100). Higher is better quality, larger file.
        strip_metadata: Remove EXIF and other metadata.
        output_format: Target format. AUTO infers from output path or original.
        keep_aspect_ratio: Maintain aspect ratio when resizing (legacy when fit is None).
        fit: Resize fit mode: down, cover, contain, fill.
        anchor: Anchor point for cover/contain cropping and positioning.
        aspect_ratio: Target aspect ratio as '16:9' or (16, 9).
        background_color: RGB tuple or hex color for contain padding.
        auto_orient: Apply EXIF orientation before processing.
        progressive: Use progressive JPEG encoding.
        optimize: Enable Pillow optimization flags.
        overwrite: Allow overwriting the source file when output is None.
        lossless: Use lossless compression for PNG/WEBP. Ignored for JPEG.
        backup_dir: Directory to copy the original file into before processing.
        min_size_bytes: Skip files already smaller than this threshold (bytes).

    Returns:
        OptimizationResult with details of the operation.

    """
    source_path = Path(source)

    if error := validate_no_parent_references(source_path, "source"):
        return _error_result(source_path, error)

    try:
        original_size = source_path.stat().st_size
    except FileNotFoundError:
        _logger.warning("File not found", extra={"operation": "optimize", "path": str(source_path)})
        return _error_result(source_path, f"File not found: {source_path}")
    except OSError as exc:
        return _error_result(source_path, f"Cannot access file: {exc}")

    if not isinstance(output_format, OutputFormat):
        _logger.warning(
            "Invalid output_format", extra={"operation": "optimize", "path": str(source_path)}
        )
        return _error_result(
            source_path,
            f"output_format must be an OutputFormat value, got {type(output_format).__name__}",
        )

    if validation_error := validate_optimize_params(
        quality=quality,
        max_width=max_width,
        max_height=max_height,
        min_size_bytes=min_size_bytes,
    ):
        return _error_result(source_path, validation_error)

    try:
        if isinstance(fit, str) and fit:
            fit = FitMode(fit)
        if isinstance(anchor, str):
            anchor = Anchor(anchor)
    except ValueError as exc:
        return _error_result(source_path, str(exc))

    if original_size > MAX_INPUT_BYTES:
        return _error_result(
            source_path,
            f"Input too large (max {MAX_INPUT_BYTES} bytes)",
            original_size=original_size,
        )

    if output is not None:
        out = Path(output)
        if error := validate_no_parent_references(out, "output"):
            return _error_result(source_path, error, original_size=original_size)

    if backup_dir is not None:
        backup = Path(backup_dir)
        if error := validate_no_parent_references(backup, "backup_dir"):
            return _error_result(source_path, error, original_size=original_size)
        try:
            backup.mkdir(parents=True, exist_ok=True)
            shutil.copy2(source_path, backup / source_path.name)
        except OSError as exc:
            return _error_result(
                source_path,
                f"Failed to create backup: {exc}",
                original_size=original_size,
            )

    if min_size_bytes is not None and original_size <= min_size_bytes:
        return OptimizationResult(
            source_path=source_path,
            output_path=source_path,
            original_size=original_size,
            optimized_size=original_size,
            savings_bytes=0,
            savings_percent=0.0,
            width=0,
            height=0,
            format="",
            metadata_removed=False,
            success=True,
            error=f"Skipped: file already below {min_size_bytes} bytes",
        )

    output_path: Path
    if output is None:
        if not overwrite:
            return _error_result(
                source_path,
                "Output path required unless overwrite=True",
                original_size=original_size,
            )
        output_path = source_path
    else:
        output_path = Path(output)

    # Handle SVG files with pure-Python optimizer
    if source_path.suffix.lower() == ".svg":
        return _optimize_svg(source_path, output_path, original_size)

    try:
        image: Image.Image
        with _open_image(source_path, label="source") as opened_img:
            image = opened_img
            output_path, pillow_fmt = resolve_and_adjust_path(
                image,
                output_path,
                output_format,
            )

            is_animated = (
                getattr(image, "is_animated", False)
                or getattr(
                    image,
                    "n_frames",
                    1,
                )
                > 1
            )

            if is_animated and pillow_fmt == "WEBP":
                return _optimize_animated_gif(
                    image,
                    source_path,
                    output_path,
                    original_size,
                    pillow_fmt,
                    max_width=max_width,
                    max_height=max_height,
                    keep_aspect_ratio=keep_aspect_ratio,
                    fit=fit,
                    anchor=anchor,
                    aspect_ratio=aspect_ratio,
                    background_color=background_color,
                    quality=quality,
                    strip_metadata=strip_metadata,
                    optimize=optimize,
                    lossless=lossless,
                )

            if auto_orient and not is_animated:
                image = apply_exif_orientation(image)

            img = convert_mode(image, pillow_fmt)
            img = resize_image(
                img,
                max_width=max_width,
                max_height=max_height,
                keep_aspect_ratio=keep_aspect_ratio,
                fit=fit,
                anchor=anchor,
                aspect_ratio=aspect_ratio,
                background_color=background_color,
            )
            new_width, new_height = img.size

            save_kwargs = build_save_kwargs(
                pillow_fmt,
                quality=quality,
                progressive=progressive,
                optimize=optimize,
                strip_metadata=strip_metadata,
                lossless=lossless,
            )
            img = strip_metadata_pillow(img, pillow_fmt)

            output_path.parent.mkdir(parents=True, exist_ok=True)
            try:
                img.save(output_path, format=pillow_fmt, **save_kwargs)
            finally:
                img.close()

        if strip_metadata and keep_exif_groups is not None:
            # Selective EXIF: keep only specified groups.
            # First save without EXIF, then re-apply filtered EXIF from source.
            import piexif

            try:
                src_exif = piexif.load(str(source_path))
                from pixopt.exif import filter_exif

                filtered = filter_exif(src_exif, keep_exif_groups)
                has_data = any(
                    filtered.get(ifd) for ifd in ("0th", "Exif", "GPS", "1st", "Interoperability")
                ) or filtered.get("thumbnail")
                if has_data:
                    exif_bytes = piexif.dump(filtered)
                    # Re-save with filtered EXIF.
                    with Image.open(output_path) as _re:
                        _re.save(output_path, format=pillow_fmt, exif=exif_bytes)
            except (OSError, ValueError, KeyError) as exc:
                _logger.warning("Failed to apply filtered EXIF", exc_info=exc)
        elif strip_metadata:
            strip_exif_post_process(output_path, pillow_fmt)

        optimized_size = output_path.stat().st_size
        savings = original_size - optimized_size

        return OptimizationResult(
            source_path=source_path,
            output_path=output_path,
            original_size=original_size,
            optimized_size=optimized_size,
            savings_bytes=savings,
            savings_percent=(savings / original_size * PERCENT) if original_size > 0 else 0.0,
            width=new_width,
            height=new_height,
            format=pillow_fmt,
            metadata_removed=strip_metadata,
            success=True,
        )

    except Image.DecompressionBombError as exc:
        return _error_result(
            source_path,
            f"Image too large or possible decompression bomb: {exc}",
            original_size=original_size,
            output=output_path,
        )
    except (OSError, ValueError) as exc:
        return _error_result(source_path, str(exc), original_size=original_size, output=output_path)

batch_optimize(sources: Iterable[Path | str], output_dir: Path | str, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, strip_metadata: bool = True, output_format: OutputFormat = OutputFormat.AUTO, keep_aspect_ratio: bool = True, progressive: bool = True, optimize: bool = True, overwrite: bool = False, lossless: bool = False, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE, auto_orient: bool = True, keep_exif_groups: set[EXIFGroup] | None = None, on_progress: ProgressCallback | None = None) -> BatchReport

Optimize multiple image files and return an aggregated report.

Parameters:

Name	Type	Description	Default
sources	Iterable[Path | str]	Iterable of source image paths.	required
output_dir	Path | str	Directory where all optimized images are written.	required

Returns:

Name	Type	Description
A	BatchReport	class:BatchReport with totals, savings, failures and elapsed time.

Source code in pixopt/optimizer.py

def batch_optimize(
    sources: Iterable[Path | str],
    output_dir: Path | str,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    strip_metadata: bool = True,
    output_format: OutputFormat = OutputFormat.AUTO,
    keep_aspect_ratio: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    overwrite: bool = False,
    lossless: bool = False,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
    auto_orient: bool = True,
    keep_exif_groups: set[EXIFGroup] | None = None,
    on_progress: ProgressCallback | None = None,
) -> BatchReport:
    """Optimize multiple image files and return an aggregated report.

    Args:
        sources: Iterable of source image paths.
        output_dir: Directory where all optimized images are written.

    Returns:
        A :class:`BatchReport` with totals, savings, failures and elapsed time.
    """
    import time

    from pixopt._units import PERCENT

    out_dir = Path(output_dir)
    if error := validate_no_parent_references(out_dir, "output_dir"):
        raise ValueError(error)
    out_dir.mkdir(parents=True, exist_ok=True)

    if error := validate_optimize_params(
        quality=quality,
        max_width=max_width,
        max_height=max_height,
        min_size_bytes=min_size_bytes,
    ):
        raise ValueError(error)

    source_list = list(sources)
    total = len(source_list)
    if total > MAX_DIRECTORY_SCAN:
        raise ValueError(f"Too many files in batch (max {MAX_DIRECTORY_SCAN}), got {total}")

    logger = get_logger("batch")
    logger.info("Batch optimization started", extra={"operation": "batch", "size_bytes": total})

    start = time.perf_counter()
    results: list[OptimizationResult] = []

    for idx, src in enumerate(source_list, 1):
        src_path = Path(src)
        out_path = out_dir / src_path.name
        result = optimize_image(
            src_path,
            out_path,
            max_width=max_width,
            max_height=max_height,
            quality=quality,
            strip_metadata=strip_metadata,
            output_format=output_format,
            keep_aspect_ratio=keep_aspect_ratio,
            progressive=progressive,
            optimize=optimize,
            overwrite=overwrite,
            lossless=lossless,
            backup_dir=backup_dir,
            min_size_bytes=min_size_bytes,
            fit=fit,
            anchor=anchor,
            aspect_ratio=aspect_ratio,
            background_color=background_color,
            auto_orient=auto_orient,
            keep_exif_groups=keep_exif_groups,
        )
        results.append(result)

        if on_progress is not None:
            on_progress(
                ProgressInfo(
                    current=idx,
                    total=total,
                    current_file=src_path,
                    success=result.success,
                    message=result.error or "",
                )
            )

    elapsed = time.perf_counter() - start

    total_files = len(results)
    succeeded = sum(1 for r in results if r.success)
    failed = total_files - succeeded
    total_original = sum(r.original_size for r in results)
    total_optimized = sum(r.optimized_size for r in results if r.success)
    total_savings = total_original - total_optimized
    savings_percent = (total_savings / total_original * PERCENT) if total_original > 0 else 0.0

    return BatchReport(
        results=results,
        total_files=total_files,
        succeeded=succeeded,
        failed=failed,
        total_original_size=total_original,
        total_optimized_size=total_optimized,
        total_savings_bytes=total_savings,
        total_savings_percent=savings_percent,
        elapsed_seconds=elapsed,
    )

optimize_directory(source_dir: Path | str, output_dir: Path | str | None = None, *, recursive: bool = False, extensions: Iterable[str] | None = None, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, on_progress: ProgressCallback | None = None, **kwargs: Any) -> list[OptimizationResult]

Optimize all images in a directory.

Parameters:

Name	Type	Description	Default
source_dir	Path | str	Directory containing images.	required
output_dir	Path | str | None	Destination directory. If None, overwrites in-place.	None
recursive	bool	Search subdirectories.	False
extensions	Iterable[str] | None	File extensions to process. Defaults to common image types.	None
backup_dir	Path | str | None	Directory to copy originals into before processing.	None
min_size_bytes	int | None	Skip files already smaller than this threshold (bytes).	None
on_progress	ProgressCallback | None	Optional callback invoked after each file is processed.	None
**kwargs	Any	Passed to optimize_image.	{}

Returns:

Type	Description
list[OptimizationResult]	List of OptimizationResult for each processed file.

Source code in pixopt/optimizer.py

def optimize_directory(
    source_dir: Path | str,
    output_dir: Path | str | None = None,
    *,
    recursive: bool = False,
    extensions: Iterable[str] | None = None,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    on_progress: ProgressCallback | None = None,
    **kwargs: Any,
) -> list[OptimizationResult]:
    """Optimize all images in a directory.

    Args:
        source_dir: Directory containing images.
        output_dir: Destination directory. If None, overwrites in-place.
        recursive: Search subdirectories.
        extensions: File extensions to process. Defaults to common image types.
        backup_dir: Directory to copy originals into before processing.
        min_size_bytes: Skip files already smaller than this threshold (bytes).
        on_progress: Optional callback invoked after each file is processed.
        **kwargs: Passed to optimize_image.

    Returns:
        List of OptimizationResult for each processed file.

    """
    src = Path(source_dir)
    if output_dir is not None:
        out_dir_raw = Path(output_dir)
        if error := validate_no_parent_references(out_dir_raw, "output_dir"):
            raise ValueError(error)
        out_dir = out_dir_raw.resolve()
    else:
        out_dir = None
    if backup_dir is not None and (
        error := validate_no_parent_references(Path(backup_dir), "backup_dir")
    ):
        raise ValueError(error)
    results: list[OptimizationResult] = []

    files = list(discover_images(src, recursive=recursive, extensions=extensions))
    total = len(files)
    if total > MAX_DIRECTORY_SCAN:
        raise ValueError(f"Too many files in directory (max {MAX_DIRECTORY_SCAN}), got {total}")

    for idx, file_path in enumerate(files, 1):
        if out_dir is not None:
            try:
                rel = file_path.relative_to(src)
            except ValueError as exc:
                raise ValueError(f"File path is not inside source directory: {file_path}") from exc
            out = out_dir / rel
            if not out.resolve().is_relative_to(out_dir):
                raise ValueError(f"Output path escapes target directory: {out}")
        else:
            out = None

        result = optimize_image(
            file_path,
            out,
            overwrite=(output_dir is None),
            backup_dir=backup_dir,
            min_size_bytes=min_size_bytes,
            **kwargs,
        )
        results.append(result)

        if on_progress is not None:
            on_progress(
                ProgressInfo(
                    current=idx,
                    total=total,
                    current_file=file_path,
                    success=result.success,
                    message=result.error or "",
                )
            )

    return results

validate_optimize_params(*, quality: int, max_width: int | None, max_height: int | None, min_size_bytes: int | None) -> str | None

Return an error message if any parameter is invalid, otherwise None.

Source code in pixopt/optimizer.py

def validate_optimize_params(
    *,
    quality: int,
    max_width: int | None,
    max_height: int | None,
    min_size_bytes: int | None,
) -> str | None:
    """Return an error message if any parameter is invalid, otherwise None."""
    if not MIN_QUALITY <= quality <= MAX_QUALITY:
        return f"quality must be between {MIN_QUALITY} and {MAX_QUALITY}, got {quality}"
    if max_width is not None and max_width <= 0:
        return f"max_width must be a positive integer, got {max_width}"
    if max_height is not None and max_height <= 0:
        return f"max_height must be a positive integer, got {max_height}"
    if min_size_bytes is not None and min_size_bytes < 0:
        return f"min_size_bytes must be non-negative, got {min_size_bytes}"
    return None

change_extension(source: Path | str, output: Path | str | None = None, *, output_format: OutputFormat = OutputFormat.AUTO, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, **kwargs: Any) -> OptimizationResult

Convert an image to a different file format / extension.

This is a thin wrapper around optimize_image focused on format conversion. All other optimization parameters are forwarded.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str | None	Destination path. If None, overwrites source (requires overwrite=True).	None
output_format	OutputFormat	Target format. Defaults to inferring from output path.	AUTO
backup_dir	Path | str | None	Directory to copy originals into before processing.	None
min_size_bytes	int | None	Skip files already smaller than this threshold (bytes).	None
**kwargs	Any	Passed to optimize_image.	{}

Returns:

Type	Description
OptimizationResult	OptimizationResult with details of the conversion.

Source code in pixopt/optimizer.py

def change_extension(
    source: Path | str,
    output: Path | str | None = None,
    *,
    output_format: OutputFormat = OutputFormat.AUTO,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    **kwargs: Any,
) -> OptimizationResult:
    """Convert an image to a different file format / extension.

    This is a thin wrapper around optimize_image focused on format conversion.
    All other optimization parameters are forwarded.

    Args:
        source: Path to the source image.
        output: Destination path. If None, overwrites source (requires overwrite=True).
        output_format: Target format. Defaults to inferring from output path.
        backup_dir: Directory to copy originals into before processing.
        min_size_bytes: Skip files already smaller than this threshold (bytes).
        **kwargs: Passed to optimize_image.

    Returns:
        OptimizationResult with details of the conversion.

    """
    return optimize_image(
        source,
        output,
        output_format=output_format,
        backup_dir=backup_dir,
        min_size_bytes=min_size_bytes,
        **kwargs,
    )

convert_to_favicon(source: Path | str, output: Path | str | None = None, *, sizes: list[int] | None = None, background: tuple[int, int, int] = WHITE, keep_transparency: bool = True, auto_orient: bool = True) -> OptimizationResult

Convert an image to a multi-resolution ICO favicon.

Generates a .ico file containing multiple square resolutions suitable for browser tabs, bookmarks and high-DPI displays.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str | None	Output .ico path. If None, uses source name with .ico extension.	None
sizes	list[int] | None	List of square sizes to include. Default: [16, 32, 48, 64, 128, 256].	None
background	tuple[int, int, int]	RGB fill for transparent images when keep_transparency=False.	WHITE
keep_transparency	bool	Preserve alpha channel if present.	True
auto_orient	bool	Apply EXIF orientation before processing.	True

Returns:

Type	Description
OptimizationResult	OptimizationResult with details of the operation.

Source code in pixopt/optimizer.py

def convert_to_favicon(
    source: Path | str,
    output: Path | str | None = None,
    *,
    sizes: list[int] | None = None,
    background: tuple[int, int, int] = WHITE,
    keep_transparency: bool = True,
    auto_orient: bool = True,
) -> OptimizationResult:
    """Convert an image to a multi-resolution ICO favicon.

    Generates a .ico file containing multiple square resolutions suitable
    for browser tabs, bookmarks and high-DPI displays.

    Args:
        source: Path to the source image.
        output: Output .ico path. If None, uses source name with .ico extension.
        sizes: List of square sizes to include. Default: [16, 32, 48, 64, 128, 256].
        background: RGB fill for transparent images when keep_transparency=False.
        keep_transparency: Preserve alpha channel if present.
        auto_orient: Apply EXIF orientation before processing.

    Returns:
        OptimizationResult with details of the operation.

    """
    source_path = Path(source)
    try:
        original_size = source_path.stat().st_size
    except FileNotFoundError:
        return _error_result(source_path, f"File not found: {source_path}")
    except OSError as exc:
        return _error_result(source_path, f"Cannot access file: {exc}")

    if output is None:
        output_path = source_path.with_suffix(".ico")
    else:
        output_path = Path(output)
        if error := validate_no_parent_references(output_path, "output"):
            return _error_result(source_path, error, original_size=original_size)
        if not output_path.suffix:
            output_path = output_path.with_suffix(".ico")

    chosen_sizes = sizes if sizes is not None else DEFAULT_FAVICON_SIZES.copy()

    if not chosen_sizes:
        return _error_result(source_path, "sizes cannot be empty", original_size=original_size)

    if len(chosen_sizes) > MAX_FAVICON_SIZES:
        return _error_result(
            source_path,
            f"Too many favicon sizes (max {MAX_FAVICON_SIZES}), got {len(chosen_sizes)}",
            original_size=original_size,
        )

    for size in chosen_sizes:
        if size <= 0:
            return _error_result(
                source_path,
                f"favicon sizes must be positive integers, got {size!r}",
                original_size=original_size,
            )

    icons: list[Image.Image] = []
    image: Image.Image | None = None
    current_icon: Image.Image | None = None
    current_bg: Image.Image | None = None
    try:
        with _open_image(source_path, label="source") as opened_img:
            image = opened_img
            if auto_orient:
                image = apply_exif_orientation(image)
            # Work in RGBA so we can consistently composite on a background
            # when keep_transparency is False, regardless of the source mode.
            if image.mode != "RGBA":
                image = image.convert("RGBA")

            for size in chosen_sizes:
                current_icon = image.resize((size, size), Resampling.LANCZOS)
                if not keep_transparency:
                    current_bg = Image.new("RGB", (size, size), background)
                    channels = current_icon.split()
                    if len(channels) < 4:
                        raise ValueError(
                            f"Icon resized to {size} has unexpected channel count: {len(channels)}"
                        )
                    current_bg.paste(current_icon, mask=channels[3])
                    for ch in channels:
                        ch.close()
                    icons.append(current_bg)
                    current_bg = None
                    current_icon.close()
                    current_icon = None
                else:
                    icons.append(current_icon)
                    current_icon = None

            output_path.parent.mkdir(parents=True, exist_ok=True)
            icons[0].save(
                output_path,
                format="ICO",
                append_images=icons[1:],
            )
    except (OSError, ValueError) as exc:
        _logger.error(
            "Favicon conversion failed", extra={"operation": "favicon", "path": str(source_path)}
        )
        return _error_result(
            source_path,
            str(exc),
            original_size=original_size,
            output=output_path,
        )
    finally:
        if current_icon is not None:
            current_icon.close()
        if current_bg is not None:
            current_bg.close()
        for icon in icons:
            icon.close()
        if image is not None:
            image.close()

    optimized_size = output_path.stat().st_size
    savings = original_size - optimized_size
    max_size = max(chosen_sizes)

    return OptimizationResult(
        source_path=source_path,
        output_path=output_path,
        original_size=original_size,
        optimized_size=optimized_size,
        savings_bytes=savings,
        savings_percent=(savings / original_size * PERCENT) if original_size > 0 else 0.0,
        width=max_size,
        height=max_size,
        format="ICO",
        metadata_removed=True,
        success=True,
    )

Models

OptimizationResult(source_path: Path, output_path: Path, original_size: int, optimized_size: int, savings_bytes: int, savings_percent: float, width: int, height: int, format: str, metadata_removed: bool, success: bool, error: str | None = None) dataclass

Result of an image optimization operation.

Attributes

source_path: Path instance-attribute

output_path: Path instance-attribute

original_size: int instance-attribute

optimized_size: int instance-attribute

savings_bytes: int instance-attribute

savings_percent: float instance-attribute

width: int instance-attribute

height: int instance-attribute

format: str instance-attribute

metadata_removed: bool instance-attribute

success: bool instance-attribute

error: str | None = None class-attribute instance-attribute

human_original_size: str property

Return the original size as a human-readable string.

human_optimized_size: str property

Return the optimized size as a human-readable string.

human_savings: str property

Return the saved bytes as a human-readable string.

OutputFormat

Bases: str, Enum

Supported output formats.

Attributes

AUTO = 'auto' class-attribute instance-attribute

JPEG = 'jpeg' class-attribute instance-attribute

PNG = 'png' class-attribute instance-attribute

WEBP = 'webp' class-attribute instance-attribute

AVIF = 'avif' class-attribute instance-attribute

ORIGINAL = 'original' class-attribute instance-attribute

Placeholders

generate_placeholder(image_path: Path, *, placeholder_type: PlaceholderType | str = PlaceholderType.LQIP, lqip_size: int = 32, lqip_quality: int = 20) -> str

Generate a placeholder string for an image.

Parameters:

Name	Type	Description	Default
image_path	Path	Path to the source image.	required
placeholder_type	PlaceholderType | str	One of 'color', 'lqip', 'blurhash'.	LQIP
lqip_size	int	Max thumbnail dimension for LQIP.	32
lqip_quality	int	JPEG quality for LQIP.	20

Returns:

Type	Description
str	A CSS color string, base64 data URI, or blurhash string.

Source code in pixopt/placeholder.py

def generate_placeholder(
    image_path: Path,
    *,
    placeholder_type: PlaceholderType | str = PlaceholderType.LQIP,
    lqip_size: int = 32,
    lqip_quality: int = 20,
) -> str:
    """Generate a placeholder string for an image.

    Args:
        image_path: Path to the source image.
        placeholder_type: One of 'color', 'lqip', 'blurhash'.
        lqip_size: Max thumbnail dimension for LQIP.
        lqip_quality: JPEG quality for LQIP.

    Returns:
        A CSS color string, base64 data URI, or blurhash string.

    """
    ptype = (
        placeholder_type.value.lower()
        if isinstance(placeholder_type, PlaceholderType)
        else str(placeholder_type).lower()
    )
    if ptype not in ("color", "lqip", "blurhash"):
        raise ValueError(
            f"placeholder_type must be 'color', 'lqip' or 'blurhash', got {placeholder_type!r}",
        )

    with _open_image(image_path, label="source") as img:
        img.load()
        if img.width > MAX_IMAGE_DIMENSION or img.height > MAX_IMAGE_DIMENSION:
            raise ValueError(
                f"Image dimensions too large: {img.width}x{img.height} (max {MAX_IMAGE_DIMENSION})"
            )
        if ptype == "color":
            return extract_dominant_color(img)
        if ptype == "lqip":
            return generate_lqip_datauri(img, size=lqip_size, quality=lqip_quality)
        if ptype == "blurhash":
            return generate_blurhash(img)
    return ""

extract_dominant_color(img: Image.Image) -> str

Return the dominant color of an image as a hex CSS string.

Uses a downsample + average approach for accuracy.

Source code in pixopt/placeholder.py

def extract_dominant_color(img: Image.Image) -> str:
    """Return the dominant color of an image as a hex CSS string.

    Uses a downsample + average approach for accuracy.
    """
    rgb = img.convert("RGB")
    # Average all pixels by resizing to 1x1 with high-quality filter
    pixel = rgb.resize((1, 1), Image.Resampling.LANCZOS).getpixel((0, 0))
    if not isinstance(pixel, tuple):
        return "#000000"
    return f"#{pixel[0]:02x}{pixel[1]:02x}{pixel[2]:02x}"

generate_lqip_datauri(img: Image.Image, *, size: int = 32, quality: int = 20) -> str

Generate a tiny blurred placeholder image as a base64 data URI.

Parameters:

Name	Type	Description	Default
img	Image	Source PIL Image.	required
size	int	Maximum dimension of the thumbnail (maintains aspect ratio).	32
quality	int	JPEG quality for the tiny image (low = smaller).	20

Returns:

Type	Description
str	A base64 data URI string like 'data:image/jpeg;base64,/9j/4AAQ...'.

Source code in pixopt/placeholder.py

def generate_lqip_datauri(img: Image.Image, *, size: int = 32, quality: int = 20) -> str:
    """Generate a tiny blurred placeholder image as a base64 data URI.

    Args:
        img: Source PIL Image.
        size: Maximum dimension of the thumbnail (maintains aspect ratio).
        quality: JPEG quality for the tiny image (low = smaller).

    Returns:
        A base64 data URI string like 'data:image/jpeg;base64,/9j/4AAQ...'.

    """
    if size <= 0 or size > MAX_IMAGE_DIMENSION:
        raise ValueError(f"size must be between 1 and {MAX_IMAGE_DIMENSION}, got {size}")
    if not MIN_QUALITY <= quality <= MAX_QUALITY:
        raise ValueError(
            f"quality must be between {MIN_QUALITY} and {MAX_QUALITY}, got {quality}",
        )

    thumb = img.copy()
    thumb = thumb.convert("RGB")
    thumb.thumbnail((size, size), Image.Resampling.LANCZOS)
    thumb = thumb.filter(ImageFilter.GaussianBlur(radius=2))

    with io.BytesIO() as buf:
        thumb.save(buf, format="JPEG", quality=quality, optimize=True)
        b64 = base64.b64encode(buf.getvalue()).decode("ascii")
    thumb.close()
    return f"data:image/jpeg;base64,{b64}"

generate_blurhash(img: Image.Image, *, components_x: int = 4, components_y: int = 3) -> str

Generate a simplified blurhash-like string from an image.

This is a pure-Python approximation that encodes average colors of a grid into a compact base-83 string. It is NOT the official BlurHash algorithm, but produces visually similar short placeholders.

Parameters:

Name	Type	Description	Default
img	Image	Source PIL Image.	required
components_x	int	Number of horizontal grid cells.	4
components_y	int	Number of vertical grid cells.	3

Returns:

Type	Description
str	A short blurhash-like string.

Source code in pixopt/placeholder.py

def generate_blurhash(img: Image.Image, *, components_x: int = 4, components_y: int = 3) -> str:
    """Generate a simplified blurhash-like string from an image.

    This is a pure-Python approximation that encodes average colors of a
    grid into a compact base-83 string. It is NOT the official BlurHash
    algorithm, but produces visually similar short placeholders.

    Args:
        img: Source PIL Image.
        components_x: Number of horizontal grid cells.
        components_y: Number of vertical grid cells.

    Returns:
        A short blurhash-like string.

    """
    if components_x <= 0 or components_y <= 0:
        raise ValueError(
            f"components_x and components_y must be positive, "
            f"got {components_x!r} and {components_y!r}",
        )
    if components_x > MAX_BLURHASH_COMPONENTS or components_y > MAX_BLURHASH_COMPONENTS:
        raise ValueError(
            f"components_x and components_y must not exceed {MAX_BLURHASH_COMPONENTS}, "
            f"got {components_x!r} and {components_y!r}"
        )

    rgb = img.convert("RGB")
    try:
        w, h = rgb.size
        cell_w = max(1, w // components_x)
        cell_h = max(1, h // components_y)

        # Size flag (components - 1) each fits in one char
        size_flag = (components_y - 1) * 9 + (components_x - 1)
        parts: list[str] = [_encode_base83(size_flag, 1)]

        for cy in range(components_y):
            for cx in range(components_x):
                x1 = cx * cell_w
                y1 = cy * cell_h
                x2 = min(w, x1 + cell_w)
                y2 = min(h, y1 + cell_h)
                region = rgb.crop((x1, y1, x2, y2))
                try:
                    data = region.tobytes()
                    n = len(data) // 3
                    if n == 0:
                        r = g = b = 0
                    else:
                        r_total = g_total = b_total = 0
                        for i in range(0, len(data), 3):
                            r_total += data[i]
                            g_total += data[i + 1]
                            b_total += data[i + 2]
                        r = r_total // n
                        g = g_total // n
                        b = b_total // n
                    # Pack RGB into a single base83 value
                    packed = (r << 16) | (g << 8) | b
                    parts.append(_encode_base83(packed, 4))
                finally:
                    region.close()

        return "".join(parts)
    finally:
        rgb.close()

Smart format detection

detect_optimal_format(image_path: Path | str, *, allow_lossy: bool = True, allow_lossless: bool = True, allow_animation: bool = True) -> OutputFormat

Analyze an image and return the most efficient output format.

Rules

• Transparent image → WEBP (or PNG if lossless only)
• Animated image → WEBP
• Photograph with many colors → WEBP (or JPEG if no WEBP)
• Graphic/UI with few colors → WEBP lossless or PNG

Source code in pixopt/smart_format.py

def detect_optimal_format(
    image_path: Path | str,
    *,
    allow_lossy: bool = True,
    allow_lossless: bool = True,
    allow_animation: bool = True,
) -> OutputFormat:
    """Analyze an image and return the most efficient output format.

    Rules:
        - Transparent image → WEBP (or PNG if lossless only)
        - Animated image → WEBP
        - Photograph with many colors → WEBP (or JPEG if no WEBP)
        - Graphic/UI with few colors → WEBP lossless or PNG
    """
    path = Path(image_path)

    try:
        with _open_image(path, label="source") as img:
            img.load()
            if img.width > MAX_IMAGE_DIMENSION or img.height > MAX_IMAGE_DIMENSION:
                return OutputFormat.WEBP

            is_animated = getattr(img, "is_animated", False) or getattr(img, "n_frames", 1) > 1
            if is_animated and allow_animation:
                return OutputFormat.WEBP

            transparent = has_transparency(img)
            photo = is_photo(img)

            if transparent:
                if allow_lossless:
                    return OutputFormat.WEBP
                return OutputFormat.PNG

            if photo and allow_lossy:
                return OutputFormat.WEBP

            if not photo and allow_lossless:
                return OutputFormat.WEBP

            if allow_lossy:
                return OutputFormat.JPEG

            return OutputFormat.PNG
    except (UnidentifiedImageError, Image.DecompressionBombError, ValueError):
        # Corrupt, too large or otherwise unreadable images fall back to a
        # widely supported default so the caller can report a controlled error
        # through optimize_image instead of crashing.
        return OutputFormat.WEBP

has_transparency(img: Image.Image) -> bool

Check if the image contains any transparent or semi-transparent pixels.

Source code in pixopt/smart_format.py

def has_transparency(img: Image.Image) -> bool:
    """Check if the image contains any transparent or semi-transparent pixels."""
    mode = img.mode
    if mode in ("RGBA", "LA"):
        alpha = img.split()[-1]
        data = alpha.tobytes()
        return any(b < MAX_CHANNEL_VALUE for b in data)
    if mode == "P":
        # Check if palette has transparency
        if "transparency" in img.info:
            return True
        # Convert to RGBA and check
        rgba = img.convert("RGBA")
        try:
            alpha = rgba.split()[-1]
            data = alpha.tobytes()
            return any(b < MAX_CHANNEL_VALUE for b in data)
        finally:
            rgba.close()
    return False

count_unique_colors(img: Image.Image, max_colors: int = MAX_UNIQUE_COLORS) -> int

Count unique colors in the image, capped at max_colors.

Uses a histogram approach with reduced precision for performance.

Source code in pixopt/smart_format.py

def count_unique_colors(img: Image.Image, max_colors: int = MAX_UNIQUE_COLORS) -> int:
    """Count unique colors in the image, capped at max_colors.

    Uses a histogram approach with reduced precision for performance.
    """
    rgb = img.convert("RGB")
    small: Image.Image | None = None
    try:
        small = rgb.resize((COLOR_SAMPLE_SIZE, COLOR_SAMPLE_SIZE), Image.Resampling.LANCZOS)
        data = small.tobytes()
        colors: set[tuple[int, int, int]] = set()
        for i in range(0, len(data), 3):
            colors.add((data[i], data[i + 1], data[i + 2]))
            if len(colors) >= max_colors:
                return max_colors
        return len(colors)
    finally:
        if small is not None:
            small.close()
        rgb.close()

is_photo(img: Image.Image) -> bool

Heuristic: returns True if the image looks like a photograph.

Photos tend to have many unique colors and smooth gradients. Graphics/UI tend to have fewer colors and sharp edges.

Source code in pixopt/smart_format.py

def is_photo(img: Image.Image) -> bool:
    """Heuristic: returns True if the image looks like a photograph.

    Photos tend to have many unique colors and smooth gradients.
    Graphics/UI tend to have fewer colors and sharp edges.
    """
    unique = count_unique_colors(img, max_colors=512)
    return unique >= SMART_FORMAT_PHOTO_THRESHOLD

Srcset generation

generate_srcset_images(source: Path | str, output_dir: Path | str, widths: list[int], *, quality: int = 85, output_format: str = 'WEBP', strip_metadata: bool = True, progressive: bool = True, optimize: bool = True, lossless: bool = False) -> list[SrcsetImage]

Generate resized variants of an image for responsive srcset.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output_dir	Path | str	Directory where variants will be saved.	required
widths	list[int]	List of target widths in pixels. Each variant will have this width, preserving aspect ratio.	required
quality	int	JPEG/WEBP quality (1-100).	85
output_format	str	Pillow format string for output (e.g. "WEBP", "JPEG").	'WEBP'
strip_metadata	bool	Remove EXIF and other metadata.	True
progressive	bool	Use progressive JPEG encoding.	True
optimize	bool	Enable Pillow optimizer.	True
lossless	bool	Use lossless compression for PNG/WEBP.	False

Returns:

Type	Description
list[SrcsetImage]	List of SrcsetImage entries, sorted by width ascending.

Source code in pixopt/srcset_generator.py

def generate_srcset_images(
    source: Path | str,
    output_dir: Path | str,
    widths: list[int],
    *,
    quality: int = 85,
    output_format: str = "WEBP",
    strip_metadata: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    lossless: bool = False,
) -> list[SrcsetImage]:
    """Generate resized variants of an image for responsive srcset.

    Args:
        source: Path to the source image.
        output_dir: Directory where variants will be saved.
        widths: List of target widths in pixels. Each variant will have
            this width, preserving aspect ratio.
        quality: JPEG/WEBP quality (1-100).
        output_format: Pillow format string for output (e.g. "WEBP", "JPEG").
        strip_metadata: Remove EXIF and other metadata.
        progressive: Use progressive JPEG encoding.
        optimize: Enable Pillow optimizer.
        lossless: Use lossless compression for PNG/WEBP.

    Returns:
        List of SrcsetImage entries, sorted by width ascending.

    """
    source_path = Path(source)
    out_dir = Path(output_dir)
    if error := validate_no_parent_references(out_dir, "output_dir"):
        raise ValueError(error)
    out_dir.mkdir(parents=True, exist_ok=True)

    if not widths:
        raise ValueError("widths must contain at least one positive value")

    if len(widths) > MAX_SRCSET_WIDTHS:
        raise ValueError(f"Too many srcset widths (max {MAX_SRCSET_WIDTHS})")

    for w in widths:
        if w < 1 or w > MAX_IMAGE_DIMENSION:
            raise ValueError(f"width must be between 1 and {MAX_IMAGE_DIMENSION}, got {w}")

    fmt = _resolve_output_format(output_format)
    results: list[SrcsetImage] = []

    pillow_fmt = FORMAT_MAP[fmt]
    ext = FORMAT_TO_EXT[pillow_fmt]

    with _open_image(source_path, label="source") as img:
        img.load()
        orig_width = img.width

        for target_width in sorted({w for w in widths if w > 0}):
            if target_width > orig_width:
                continue

            suffix = f"-{target_width}w{ext}"
            out_path = out_dir / (source_path.stem + suffix)

            result = optimize_image(
                source_path,
                out_path,
                max_width=target_width,
                quality=quality,
                strip_metadata=strip_metadata,
                output_format=fmt,
                progressive=progressive,
                optimize=optimize,
                lossless=lossless,
            )

            if result.success:
                results.append(
                    SrcsetImage(
                        width=result.width,
                        output_path=result.output_path,
                        size_bytes=result.optimized_size,
                    ),
                )

    return results

SrcsetImage(width: int, output_path: Path, size_bytes: int) dataclass

A single responsive image variant.

Attributes

width: int instance-attribute

output_path: Path instance-attribute

size_bytes: int instance-attribute

Adaptive quality

find_quality_for_target_size(img: Image.Image, pillow_fmt: str, target_size: int, *, max_width: int | None = None, max_height: int | None = None, keep_aspect_ratio: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE, strip_metadata: bool = True, progressive: bool = True, optimize: bool = True, lossless: bool = False, min_quality: int = 1, max_quality: int = 100, tolerance: float = 0.05, max_iterations: int = 8) -> int

Find the JPEG/WEBP quality that produces a file closest to target_size.

Uses binary search over quality (1-100) and measures the actual encoded file size in memory. Returns the quality value that yields a size closest to but not exceeding the target.

Parameters:

Name	Type	Description	Default
img	Image	Open PIL Image.	required
pillow_fmt	str	Target Pillow format (JPEG or WEBP).	required
target_size	int	Target file size in bytes.	required
max_width	int | None	Maximum width in pixels, or None.	None
max_height	int | None	Maximum height in pixels, or None.	None
keep_aspect_ratio	bool	Whether to keep the original aspect ratio.	True
fit	FitMode | str | None	Resize fit mode.	None
anchor	Anchor | str	Anchor point for cover/contain.	CENTER
aspect_ratio	tuple[int, int] | str | None	Target aspect ratio.	None
background_color	tuple[int, int, int] | str	Background color for contain padding.	WHITE
strip_metadata	bool	Whether to strip metadata before saving.	True
progressive	bool	Whether to use progressive encoding.	True
optimize	bool	Whether to optimize the output.	True
lossless	bool	Whether to use lossless compression.	False
min_quality	int	Lowest quality to try.	1
max_quality	int	Highest quality to try.	100
tolerance	float	Fractional tolerance around target_size (e.g. 0.05 = 5%).	0.05
max_iterations	int	Maximum binary-search iterations.	8

Returns:

Type	Description
int	Quality integer (1-100).

Source code in pixopt/adaptive_quality.py

def find_quality_for_target_size(
    img: Image.Image,
    pillow_fmt: str,
    target_size: int,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    keep_aspect_ratio: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
    strip_metadata: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    lossless: bool = False,
    min_quality: int = 1,
    max_quality: int = 100,
    tolerance: float = 0.05,
    max_iterations: int = 8,
) -> int:
    """Find the JPEG/WEBP quality that produces a file closest to target_size.

    Uses binary search over quality (1-100) and measures the actual encoded
    file size in memory. Returns the quality value that yields a size
    closest to but not exceeding the target.

    Args:
        img: Open PIL Image.
        pillow_fmt: Target Pillow format (JPEG or WEBP).
        target_size: Target file size in bytes.
        max_width: Maximum width in pixels, or None.
        max_height: Maximum height in pixels, or None.
        keep_aspect_ratio: Whether to keep the original aspect ratio.
        fit: Resize fit mode.
        anchor: Anchor point for cover/contain.
        aspect_ratio: Target aspect ratio.
        background_color: Background color for contain padding.
        strip_metadata: Whether to strip metadata before saving.
        progressive: Whether to use progressive encoding.
        optimize: Whether to optimize the output.
        lossless: Whether to use lossless compression.
        min_quality: Lowest quality to try.
        max_quality: Highest quality to try.
        tolerance: Fractional tolerance around target_size (e.g. 0.05 = 5%).
        max_iterations: Maximum binary-search iterations.

    Returns:
        Quality integer (1-100).

    """
    if pillow_fmt not in ("JPEG", "WEBP"):
        return 85

    if target_size <= 0:
        raise ValueError(f"target_size must be a positive integer, got {target_size}")
    if not MIN_QUALITY <= min_quality <= MAX_QUALITY:
        raise ValueError(
            f"min_quality must be between {MIN_QUALITY} and {MAX_QUALITY}, got {min_quality}",
        )
    if not MIN_QUALITY <= max_quality <= MAX_QUALITY:
        raise ValueError(
            f"max_quality must be between {MIN_QUALITY} and {MAX_QUALITY}, got {max_quality}",
        )
    if min_quality > max_quality:
        raise ValueError(
            f"min_quality ({min_quality}) cannot exceed max_quality ({max_quality})",
        )
    if tolerance < 0:
        raise ValueError(f"tolerance must be non-negative, got {tolerance}")
    if max_iterations <= 0:
        raise ValueError(f"max_iterations must be positive, got {max_iterations}")

    working = img.copy()
    working = convert_mode(working, pillow_fmt)
    working = resize_image(
        working,
        max_width=max_width,
        max_height=max_height,
        keep_aspect_ratio=keep_aspect_ratio,
        fit=fit,
        anchor=anchor,
        aspect_ratio=aspect_ratio,
        background_color=background_color,
    )

    try:
        low = min_quality
        high = max_quality
        best_quality = low
        best_diff = float("inf")

        for _ in range(max_iterations):
            if low > high:
                break
            mid = (low + high) // 2

            with BytesIO() as buf:
                kwargs = build_save_kwargs(
                    pillow_fmt,
                    quality=mid,
                    progressive=progressive,
                    optimize=optimize,
                    strip_metadata=strip_metadata,
                    lossless=lossless,
                )
                working.save(buf, format=pillow_fmt, **kwargs)
                size = buf.tell()

            diff = abs(size - target_size)
            if diff < best_diff:
                best_diff = diff
                best_quality = mid

            # Within tolerance window?
            if abs(size - target_size) <= target_size * tolerance:
                best_quality = mid
                break

            if size > target_size:
                high = mid - 1
            else:
                low = mid + 1

        return best_quality
    finally:
        working.close()

Visual comparison

generate_comparison_html(before_path: Path, after_path: Path, output_html: Path, title: str = 'Image Comparison') -> Path

Generate a self-contained HTML file with an interactive before/after slider.

Parameters:

Name	Type	Description	Default
before_path	Path	Path to the original image.	required
after_path	Path	Path to the optimized image.	required
output_html	Path	Path where the HTML file will be saved.	required
title	str	Page title displayed above the slider.	'Image Comparison'

Returns:

Type	Description
Path	Path to the generated HTML file.

Source code in pixopt/html_comparison.py

def generate_comparison_html(
    before_path: Path,
    after_path: Path,
    output_html: Path,
    title: str = "Image Comparison",
) -> Path:
    """Generate a self-contained HTML file with an interactive before/after slider.

    Args:
        before_path: Path to the original image.
        after_path: Path to the optimized image.
        output_html: Path where the HTML file will be saved.
        title: Page title displayed above the slider.

    Returns:
        Path to the generated HTML file.

    """
    output_html = Path(output_html)
    if error := validate_no_parent_references(output_html, "output_html"):
        raise ValueError(error)

    before_b64 = _img_to_base64(before_path)
    after_b64 = _img_to_base64(after_path)

    try:
        with _open_image(before_path, label="source") as img:
            width = img.width
    except (OSError, ValueError):
        # Pillow cannot read the file (e.g. an SVG). Use a sensible default
        # width; the responsive CSS still keeps the slider usable.
        width = 800

    orig_size = before_path.stat().st_size
    opt_size = after_path.stat().st_size
    savings = orig_size - opt_size
    pct = savings / orig_size * PERCENT if orig_size > 0 else 0

    def _human(size: int) -> str:
        if size < BYTES_PER_KB:
            return f"{size} B"
        if size < BYTES_PER_MB:
            return f"{size / BYTES_PER_KB:.1f} KB"
        return f"{size / BYTES_PER_MB:.2f} MB"

    meta = (
        f"Original: {_human(orig_size)}  |  "
        f"Optimized: {_human(opt_size)}  |  "
        f"Savings: {_human(savings)} ({pct:.1f}%)"
    )

    safe_title = html.escape(title)

    html_content = HTML_TEMPLATE.format(
        title=safe_title,
        meta=meta,
        before_b64=before_b64,
        after_b64=after_b64,
        width=width,
    )

    output_html.parent.mkdir(parents=True, exist_ok=True)
    output_html.write_text(html_content, encoding="utf-8")
    return output_html

Watermark

add_text_watermark(source: Path | str, output: Path | str, text: str, *, position: WatermarkPosition = WatermarkPosition.BOTTOM_RIGHT, opacity: float = 0.5, padding: int = 20, font_size: int = _DEFAULT_FONT_SIZE, font_path: Path | str | None = None, color: tuple[int, int, int] = (255, 255, 255)) -> WatermarkResult

Add a text watermark to an image.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str	Path for the output image.	required
text	str	Watermark text.	required
position	WatermarkPosition	Where to place the watermark.	BOTTOM_RIGHT
opacity	float	Opacity from 0.0 (transparent) to 1.0 (opaque).	0.5
padding	int	Pixel padding from the edge.	20
font_size	int	Font size in pixels.	_DEFAULT_FONT_SIZE
font_path	Path | str | None	Optional path to a TTF/OTF font file.	None
color	tuple[int, int, int]	Text color as (R, G, B).	(255, 255, 255)

Returns:

Name	Type	Description
A	WatermarkResult	class:WatermarkResult.

Source code in pixopt/watermark.py

def add_text_watermark(
    source: Path | str,
    output: Path | str,
    text: str,
    *,
    position: WatermarkPosition = WatermarkPosition.BOTTOM_RIGHT,
    opacity: float = 0.5,
    padding: int = 20,
    font_size: int = _DEFAULT_FONT_SIZE,
    font_path: Path | str | None = None,
    color: tuple[int, int, int] = (255, 255, 255),
) -> WatermarkResult:
    """Add a text watermark to an image.

    Args:
        source: Path to the source image.
        output: Path for the output image.
        text: Watermark text.
        position: Where to place the watermark.
        opacity: Opacity from 0.0 (transparent) to 1.0 (opaque).
        padding: Pixel padding from the edge.
        font_size: Font size in pixels.
        font_path: Optional path to a TTF/OTF font file.
        color: Text color as (R, G, B).

    Returns:
        A :class:`WatermarkResult`.
    """
    if padding < 0 or padding > MAX_IMAGE_DIMENSION:
        raise ValueError(f"padding must be between 0 and {MAX_IMAGE_DIMENSION}, got {padding}")
    if font_size < 1 or font_size > 1000:
        raise ValueError(f"font_size must be between 1 and 1000, got {font_size}")
    if not isinstance(text, str):
        raise ValueError(f"text must be a string, got {type(text).__name__}")
    if len(text) > _MAX_TEXT_LENGTH:
        raise ValueError(f"text exceeds maximum length of {_MAX_TEXT_LENGTH} characters")

    src_path = Path(source)
    out_path = Path(output)
    if error := validate_no_parent_references(out_path, "output"):
        raise ValueError(error)

    with _open_image(src_path, label="source") as _base:
        _base.load()
        base = _base.convert("RGBA")
        bw, bh = base.size

        if bw > MAX_IMAGE_DIMENSION or bh > MAX_IMAGE_DIMENSION:
            msg = (
                f"Image dimensions too large: {bw}x{bh} "
                f"(max {MAX_IMAGE_DIMENSION}x{MAX_IMAGE_DIMENSION})"
            )
            raise ValueError(msg)

        # Create transparent overlay for text.
        overlay = Image.new("RGBA", base.size, (0, 0, 0, 0))
        try:
            draw = ImageDraw.Draw(overlay)

            # Load font.
            font: ImageFont.ImageFont | ImageFont.FreeTypeFont
            if font_path:
                font_path_obj = Path(font_path)
                if error := validate_no_parent_references(font_path_obj, "font_path"):
                    raise ValueError(error)
                if font_path_obj.suffix.lower() not in {".ttf", ".otf"}:
                    raise ValueError("font_path must be a .ttf or .otf file")
                try:
                    stat = font_path_obj.stat()
                except FileNotFoundError as exc:
                    raise FileNotFoundError(f"Font file not found: {font_path_obj}") from exc
                if stat.st_size > 10 * 1024 * 1024:
                    raise ValueError("font file too large (max 10 MB)")
                font = ImageFont.truetype(str(font_path), font_size)
            else:
                try:
                    font = ImageFont.truetype("arial.ttf", font_size)
                except OSError:
                    font = ImageFont.load_default()

            # Measure text size.
            bbox = draw.textbbox((0, 0), text, font=font)
            tw = int(bbox[2] - bbox[0])
            th = int(bbox[3] - bbox[1])

            x, y = _resolve_position((bw, bh), (tw, th), position, padding)

            # Draw text with opacity.
            alpha = int(255 * max(0.0, min(1.0, opacity)))
            draw.text((x, y), text, fill=(color[0], color[1], color[2], alpha), font=font)

            # Composite.
            result = Image.alpha_composite(base, overlay)
            result = result.convert("RGB")

            out_path.parent.mkdir(parents=True, exist_ok=True)
            result.save(out_path)
            result.close()
        finally:
            overlay.close()
            base.close()

    return WatermarkResult(
        source_path=src_path,
        output_path=out_path,
        width=bw,
        height=bh,
        watermark_type="text",
    )

add_image_watermark(source: Path | str, output: Path | str, watermark: Path | str, *, position: WatermarkPosition = WatermarkPosition.BOTTOM_RIGHT, opacity: float = 0.5, padding: int = 20, scale: float | None = None) -> WatermarkResult

Add an image watermark (logo/overlay) onto a base image.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str	Path for the output image.	required
watermark	Path | str	Path to the watermark image (PNG with alpha recommended).	required
position	WatermarkPosition	Where to place the watermark.	BOTTOM_RIGHT
opacity	float	Opacity from 0.0 (transparent) to 1.0 (opaque).	0.5
padding	int	Pixel padding from the edge.	20
scale	float | None	Scale factor for the watermark relative to the base image width. If None, the watermark is used at its original size.	None

Returns:

Name	Type	Description
A	WatermarkResult	class:WatermarkResult.

Source code in pixopt/watermark.py

def add_image_watermark(
    source: Path | str,
    output: Path | str,
    watermark: Path | str,
    *,
    position: WatermarkPosition = WatermarkPosition.BOTTOM_RIGHT,
    opacity: float = 0.5,
    padding: int = 20,
    scale: float | None = None,
) -> WatermarkResult:
    """Add an image watermark (logo/overlay) onto a base image.

    Args:
        source: Path to the source image.
        output: Path for the output image.
        watermark: Path to the watermark image (PNG with alpha recommended).
        position: Where to place the watermark.
        opacity: Opacity from 0.0 (transparent) to 1.0 (opaque).
        padding: Pixel padding from the edge.
        scale: Scale factor for the watermark relative to the base image width.
            If None, the watermark is used at its original size.

    Returns:
        A :class:`WatermarkResult`.
    """
    if padding < 0 or padding > MAX_IMAGE_DIMENSION:
        raise ValueError(f"padding must be between 0 and {MAX_IMAGE_DIMENSION}, got {padding}")
    if scale is not None and (scale <= 0 or scale > 10):
        raise ValueError(f"scale must be between 0 and 10, got {scale}")

    src_path = Path(source)
    wm_path = Path(watermark)
    out_path = Path(output)
    if error := validate_no_parent_references(out_path, "output"):
        raise ValueError(error)
    if error := validate_no_parent_references(wm_path, "watermark"):
        raise ValueError(error)

    with _open_image(src_path, label="source") as _base:
        _base.load()
        base = _base.convert("RGBA")
        wm: Image.Image | None = None
        try:
            bw, bh = base.size

            if bw > MAX_IMAGE_DIMENSION or bh > MAX_IMAGE_DIMENSION:
                msg = (
                    f"Image dimensions too large: {bw}x{bh} "
                    f"(max {MAX_IMAGE_DIMENSION}x{MAX_IMAGE_DIMENSION})"
                )
                raise ValueError(msg)

            with _open_image(wm_path, label="watermark") as _wm:
                _wm.load()
                wm = _wm.convert("RGBA")

                if scale is not None:
                    if wm.width <= 0 or wm.height <= 0:
                        raise ValueError(
                            f"Watermark image has invalid dimensions: {wm.width}x{wm.height}"
                        )
                    new_w = int(bw * scale)
                    new_h = int(wm.height * (new_w / wm.width))
                    resized = wm.resize((new_w, new_h), Image.Resampling.LANCZOS)
                    wm.close()
                    wm = resized

                opacity = max(0.0, min(1.0, opacity))
                original_wm = wm
                wm = _apply_opacity(wm, opacity)
                original_wm.close()

                x, y = _resolve_position(base.size, wm.size, position, padding)

                # Composite watermark onto base.
                base.paste(wm, (x, y), wm)
                result = base.convert("RGB")

                out_path.parent.mkdir(parents=True, exist_ok=True)
                result.save(out_path)
                result.close()
        finally:
            if wm is not None:
                wm.close()
            base.close()

    return WatermarkResult(
        source_path=src_path,
        output_path=out_path,
        width=bw,
        height=bh,
        watermark_type="image",
    )

Sprite

create_sprite(images: Sequence[Path | str], output: Path | str, *, cell_width: int | None = None, cell_height: int | None = None, columns: int | None = None, layout: SpriteLayout | str = SpriteLayout.GRID, padding: int = 0, background: tuple[int, int, int] = WHITE, fmt: str = 'PNG') -> SpriteResult

Combine multiple images into a single sprite sheet.

Each image is resized to fit within cell_width × cell_height (keeping aspect ratio by default) and placed in a grid.

Parameters:

Name	Type	Description	Default
images	Sequence[Path | str]	List of image file paths.	required
output	Path | str	Output sprite sheet path.	required
cell_width	int | None	Width of each cell. If None, uses the max image width.	None
cell_height	int | None	Height of each cell. If None, uses the max image height.	None
columns	int | None	Number of columns (grid layout). If None, auto-calculated.	None
layout	SpriteLayout | str	"grid", "horizontal", or "vertical".	GRID
padding	int	Pixels between cells.	0
background	tuple[int, int, int]	Background color for empty areas.	WHITE
fmt	str	Output image format (PNG, JPEG, WEBP).	'PNG'

Returns:

Name	Type	Description
A	SpriteResult	class:SpriteResult with layout info and slot positions.

Raises:

Type	Description
ValueError	If the image list is empty.

Source code in pixopt/sprite.py

def create_sprite(
    images: Sequence[Path | str],
    output: Path | str,
    *,
    cell_width: int | None = None,
    cell_height: int | None = None,
    columns: int | None = None,
    layout: SpriteLayout | str = SpriteLayout.GRID,
    padding: int = 0,
    background: tuple[int, int, int] = WHITE,
    fmt: str = "PNG",
) -> SpriteResult:
    """Combine multiple images into a single sprite sheet.

    Each image is resized to fit within ``cell_width × cell_height``
    (keeping aspect ratio by default) and placed in a grid.

    Args:
        images: List of image file paths.
        output: Output sprite sheet path.
        cell_width: Width of each cell. If None, uses the max image width.
        cell_height: Height of each cell. If None, uses the max image height.
        columns: Number of columns (grid layout). If None, auto-calculated.
        layout: ``"grid"``, ``"horizontal"``, or ``"vertical"``.
        padding: Pixels between cells.
        background: Background color for empty areas.
        fmt: Output image format (PNG, JPEG, WEBP).

    Returns:
        A :class:`SpriteResult` with layout info and slot positions.

    Raises:
        ValueError: If the image list is empty.

    """
    _validate_sprite_params(cell_width, cell_height, columns, padding)

    output_path = Path(output)
    if error := validate_no_parent_references(output_path, "output"):
        raise ValueError(error)

    if not images:
        raise ValueError("images list cannot be empty")

    if len(images) > MAX_SPRITE_IMAGES:
        raise ValueError(f"Too many images for sprite (max {MAX_SPRITE_IMAGES}), got {len(images)}")

    for img_path in images:
        if error := validate_no_parent_references(Path(img_path), "source"):
            raise ValueError(error)

    if isinstance(layout, str):
        layout = SpriteLayout(layout)

    output_path.parent.mkdir(parents=True, exist_ok=True)

    _logger.debug("Creating sprite", extra={"operation": "sprite", "path": str(output_path)})

    pil_images: list[Image.Image] = []
    try:
        total_source_pixels = 0
        for img_path in images:
            image: Image.Image
            with _open_image(Path(img_path), label="source") as opened:
                image = opened
            total_source_pixels += image.width * image.height
            if total_source_pixels > MAX_SPRITE_TOTAL_PIXELS:
                raise ValueError(
                    f"Sprite source pixel budget exceeded: {total_source_pixels} "
                    f"(max {MAX_SPRITE_TOTAL_PIXELS})"
                )
            pil_images.append(image)

        # Determine cell dimensions
        if cell_width is None:
            cell_width = max(img.width for img in pil_images)
        if cell_height is None:
            cell_height = max(img.height for img in pil_images)

        total = len(pil_images)

        # Determine layout
        if layout == SpriteLayout.HORIZONTAL:
            cols = total
            rows = 1
        elif layout == SpriteLayout.VERTICAL:
            cols = 1
            rows = total
        else:
            # Grid
            import math

            cols = columns if columns is not None else max(1, int(math.ceil(total**0.5)))
            rows = (total + cols - 1) // cols

        sheet_width = cols * cell_width + (cols - 1) * padding
        sheet_height = rows * cell_height + (rows - 1) * padding

        if sheet_width > MAX_IMAGE_DIMENSION or sheet_height > MAX_IMAGE_DIMENSION:
            raise ValueError(
                f"Sprite sheet dimensions too large: {sheet_width}x{sheet_height} "
                f"(max {MAX_IMAGE_DIMENSION})"
            )

        sheet = Image.new("RGB", (sheet_width, sheet_height), background)

        slots: list[SpriteSlot] = []
        for idx, img in enumerate(pil_images):
            row = idx // cols
            col = idx % cols
            x = col * (cell_width + padding)
            y = row * (cell_height + padding)

            # Resize image to fit within cell, preserving aspect ratio
            resized = _fit_image(img, cell_width, cell_height)

            # Center within cell
            offset_x = x + (cell_width - resized.width) // 2
            offset_y = y + (cell_height - resized.height) // 2

            sheet.paste(resized, (offset_x, offset_y))

            slots.append(
                SpriteSlot(
                    index=idx,
                    source=Path(images[idx]),
                    x=offset_x,
                    y=offset_y,
                    width=resized.width,
                    height=resized.height,
                )
            )

        pillow_fmt = fmt.upper()
        save_kwargs: dict[str, Any] = {}
        if pillow_fmt == "JPEG":
            save_kwargs["quality"] = 95
        sheet.save(output_path, pillow_fmt, **save_kwargs)

        result = SpriteResult(
            output=output_path,
            layout=layout.value,
            columns=cols,
            rows=rows,
            cell_width=cell_width,
            cell_height=cell_height,
            total_images=total,
            slots=slots,
            width=sheet_width,
            height=sheet_height,
            success=True,
        )

        _logger.info(
            "Sprite created",
            extra={"operation": "sprite", "path": str(output_path), "size_bytes": total},
        )

        return result

    except (OSError, ValueError, Image.DecompressionBombError) as exc:
        _logger.error(
            "Sprite creation failed", extra={"operation": "sprite", "path": str(output_path)}
        )
        return SpriteResult(
            output=output_path,
            layout=layout.value if isinstance(layout, SpriteLayout) else str(layout),
            columns=0,
            rows=0,
            cell_width=cell_width or 0,
            cell_height=cell_height or 0,
            total_images=len(images),
            success=False,
            error=str(exc),
        )
    finally:
        for img in pil_images:
            img.close()

create_contact_sheet(images: Sequence[Path | str], output: Path | str, *, cell_width: int = 200, cell_height: int = 200, columns: int | None = None, padding: int = 10, background: tuple[int, int, int] = WHITE, label_height: int = 20, fmt: str = 'PNG') -> SpriteResult

Create a contact sheet with labels showing image filenames.

A contact sheet is a grid of thumbnails with text labels below each image, similar to a photo proof sheet.

Parameters:

Name	Type	Description	Default
images	Sequence[Path | str]	List of image file paths.	required
output	Path | str	Output contact sheet path.	required
cell_width	int	Width of each thumbnail cell.	200
cell_height	int	Height of each thumbnail cell.	200
columns	int | None	Number of columns. If None, auto-calculated.	None
padding	int	Pixels between cells.	10
background	tuple[int, int, int]	Background color.	WHITE
label_height	int	Height of the label area below each image.	20
fmt	str	Output image format.	'PNG'

Returns:

Name	Type	Description
A	SpriteResult	class:SpriteResult with layout info and slot positions.

Raises:

Type	Description
ValueError	If the image list is empty.

Source code in pixopt/sprite.py

def create_contact_sheet(
    images: Sequence[Path | str],
    output: Path | str,
    *,
    cell_width: int = 200,
    cell_height: int = 200,
    columns: int | None = None,
    padding: int = 10,
    background: tuple[int, int, int] = WHITE,
    label_height: int = 20,
    fmt: str = "PNG",
) -> SpriteResult:
    """Create a contact sheet with labels showing image filenames.

    A contact sheet is a grid of thumbnails with text labels below each
    image, similar to a photo proof sheet.

    Args:
        images: List of image file paths.
        output: Output contact sheet path.
        cell_width: Width of each thumbnail cell.
        cell_height: Height of each thumbnail cell.
        columns: Number of columns. If None, auto-calculated.
        padding: Pixels between cells.
        background: Background color.
        label_height: Height of the label area below each image.
        fmt: Output image format.

    Returns:
        A :class:`SpriteResult` with layout info and slot positions.

    Raises:
        ValueError: If the image list is empty.

    """
    _validate_sprite_params(cell_width, cell_height, columns, padding, label_height)

    output_path = Path(output)
    if error := validate_no_parent_references(output_path, "output"):
        raise ValueError(error)

    if not images:
        raise ValueError("images list cannot be empty")

    if len(images) > MAX_SPRITE_IMAGES:
        raise ValueError(
            f"Too many images for contact sheet (max {MAX_SPRITE_IMAGES}), got {len(images)}"
        )

    for img_path in images:
        if error := validate_no_parent_references(Path(img_path), "source"):
            raise ValueError(error)

    import math

    output_path.parent.mkdir(parents=True, exist_ok=True)

    _logger.debug(
        "Creating contact sheet", extra={"operation": "contact_sheet", "path": str(output_path)}
    )

    pil_images: list[Image.Image] = []
    try:
        total_source_pixels = 0
        for img_path in images:
            image: Image.Image
            with _open_image(Path(img_path), label="source") as opened:
                image = opened
            total_source_pixels += image.width * image.height
            if total_source_pixels > MAX_SPRITE_TOTAL_PIXELS:
                raise ValueError(
                    f"Contact sheet source pixel budget exceeded: {total_source_pixels} "
                    f"(max {MAX_SPRITE_TOTAL_PIXELS})"
                )
            pil_images.append(image)

        total = len(pil_images)
        cols = columns if columns is not None else max(1, int(math.ceil(total**0.5)))
        rows = (total + cols - 1) // cols

        full_cell_height = cell_height + label_height
        sheet_width = cols * cell_width + (cols + 1) * padding
        sheet_height = rows * full_cell_height + (rows + 1) * padding

        if sheet_width > MAX_IMAGE_DIMENSION or sheet_height > MAX_IMAGE_DIMENSION:
            raise ValueError(
                f"Contact sheet dimensions too large: {sheet_width}x{sheet_height} "
                f"(max {MAX_IMAGE_DIMENSION})"
            )

        sheet = Image.new("RGB", (sheet_width, sheet_height), background)
        draw = ImageDraw.Draw(sheet)

        slots: list[SpriteSlot] = []
        for idx, img in enumerate(pil_images):
            row = idx // cols
            col = idx % cols
            x = padding + col * (cell_width + padding)
            y = padding + row * (full_cell_height + padding)

            resized = _fit_image(img, cell_width, cell_height)
            offset_x = x + (cell_width - resized.width) // 2
            offset_y = y + (cell_height - resized.height) // 2

            sheet.paste(resized, (offset_x, offset_y))

            # Draw label
            label = Path(images[idx]).name
            label_y = y + cell_height + 2
            draw.text((x + 2, label_y), label, fill=_BLACK)

            # Draw border
            draw.rectangle(
                [x - 1, y - 1, x + cell_width, y + cell_height],
                outline=_GRAY,
            )

            slots.append(
                SpriteSlot(
                    index=idx,
                    source=Path(images[idx]),
                    x=offset_x,
                    y=offset_y,
                    width=resized.width,
                    height=resized.height,
                )
            )

        pillow_fmt = fmt.upper()
        save_kwargs: dict[str, Any] = {}
        if pillow_fmt == "JPEG":
            save_kwargs["quality"] = 95
        sheet.save(output_path, pillow_fmt, **save_kwargs)

        result = SpriteResult(
            output=output_path,
            layout="contact_sheet",
            columns=cols,
            rows=rows,
            cell_width=cell_width,
            cell_height=cell_height,
            total_images=total,
            slots=slots,
            width=sheet_width,
            height=sheet_height,
            success=True,
        )

        _logger.info(
            "Contact sheet created",
            extra={"operation": "contact_sheet", "path": str(output_path), "size_bytes": total},
        )

        return result

    except (OSError, ValueError, Image.DecompressionBombError) as exc:
        _logger.error(
            "Contact sheet failed", extra={"operation": "contact_sheet", "path": str(output_path)}
        )
        return SpriteResult(
            output=output_path,
            layout="contact_sheet",
            columns=0,
            rows=0,
            cell_width=cell_width,
            cell_height=cell_height,
            total_images=len(images),
            success=False,
            error=str(exc),
        )
    finally:
        for img in pil_images:
            img.close()

Bundle

generate_asset_bundle(source: Path | str, output_dir: Path | str, *, options: BundleOptions | None = None) -> AssetBundle

Generate a complete asset bundle from a single source image.

Produces hero image, thumbnail, og:image, favicon, srcset variants, LQIP data URI, blurhash, dominant color, and color palette.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output_dir	Path | str	Directory where all generated assets are written.	required
options	BundleOptions | None	Configuration for what to generate and sizing parameters.	None

Returns:

Name	Type	Description
An	AssetBundle	class:AssetBundle with all generated assets.

Raises:

Type	Description
FileNotFoundError	If the source image does not exist.

Source code in pixopt/bundle.py

def generate_asset_bundle(
    source: Path | str,
    output_dir: Path | str,
    *,
    options: BundleOptions | None = None,
) -> AssetBundle:
    """Generate a complete asset bundle from a single source image.

    Produces hero image, thumbnail, og:image, favicon, srcset variants,
    LQIP data URI, blurhash, dominant color, and color palette.

    Args:
        source: Path to the source image.
        output_dir: Directory where all generated assets are written.
        options: Configuration for what to generate and sizing parameters.

    Returns:
        An :class:`AssetBundle` with all generated assets.

    Raises:
        FileNotFoundError: If the source image does not exist.

    """
    source_path = Path(source)
    if not source_path.exists():
        raise ImageNotFoundError(source_path)

    opts = options or BundleOptions()
    out_dir = Path(output_dir)
    if error := validate_no_parent_references(out_dir, "output_dir"):
        raise ValueError(error)
    out_dir.mkdir(parents=True, exist_ok=True)

    logger = get_logger("bundle")
    logger.info(
        "Asset bundle generation started", extra={"operation": "bundle", "path": str(source_path)}
    )

    stem = source_path.stem
    bundle = AssetBundle(source_path=source_path, output_dir=out_dir)

    # Hero image
    if opts.generate_hero:
        hero_path = out_dir / f"{stem}-hero.webp"
        bundle.hero = optimize_image(
            source_path,
            hero_path,
            max_width=opts.hero_width,
            quality=opts.quality,
            output_format=OutputFormat.WEBP,
        )

    # Thumbnail
    if opts.generate_thumbnail:
        thumb_path = out_dir / f"{stem}-thumb.webp"
        bundle.thumbnail = optimize_image(
            source_path,
            thumb_path,
            max_width=opts.thumbnail_width,
            quality=opts.quality,
            output_format=OutputFormat.WEBP,
        )

    # og:image (1200x630, cover crop)
    if opts.generate_og:
        og_path = out_dir / f"{stem}-og.jpg"
        bundle.og_image = optimize_image(
            source_path,
            og_path,
            max_width=opts.og_width,
            max_height=opts.og_height,
            fit="cover",
            quality=opts.quality,
            output_format=OutputFormat.JPEG,
        )

    # Favicon
    if opts.generate_favicon:
        fav_path = out_dir / f"{stem}.ico"
        bundle.favicon = convert_to_favicon(
            source_path,
            fav_path,
            sizes=opts.favicon_sizes,
        )

    # Srcset variants
    if opts.generate_srcset:
        srcset_dir = out_dir / "srcset"
        bundle.srcset_images = generate_srcset_images(
            source_path,
            srcset_dir,
            opts.srcset_widths,
            quality=opts.quality,
            output_format="WEBP",
        )

    # LQIP, blurhash, dominant color — all from the loaded image
    if opts.generate_lqip or opts.generate_blurhash or opts.generate_dominant_color:
        with _open_image(source_path, label="source") as img:
            img.load()
            if opts.generate_lqip:
                bundle.lqip_data_uri = generate_lqip_datauri(
                    img,
                    size=opts.lqip_size,
                    quality=opts.lqip_quality,
                )
            if opts.generate_blurhash:
                bundle.blurhash = generate_blurhash(img)
            if opts.generate_dominant_color:
                bundle.dominant_color = extract_dominant_color(img)

    # Color palette
    if opts.generate_palette:
        bundle.palette = extract_palette(source_path, n=opts.palette_n)

    return bundle

In-memory I/O

optimize_bytes(data: bytes, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, output_format: OutputFormat | str = OutputFormat.WEBP, progressive: bool = True, optimize: bool = True, strip_metadata: bool = True, lossless: bool = False, auto_orient: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE) -> BytesResult

Optimize an image from raw bytes and return optimized bytes.

Parameters:

Name	Type	Description	Default
data	bytes	Raw image file bytes (JPEG, PNG, WEBP, etc.).	required
max_width	int | None	Maximum width in pixels.	None
max_height	int | None	Maximum height in pixels.	None
quality	int	JPEG/WEBP quality (1-100).	85
output_format	OutputFormat | str	Target format as :class:OutputFormat or string.	WEBP
progressive	bool	Use progressive JPEG encoding.	True
optimize	bool	Enable Pillow optimization flags.	True
strip_metadata	bool	Remove EXIF and other metadata.	True
lossless	bool	Use lossless compression for PNG/WEBP.	False
auto_orient	bool	Apply EXIF orientation before processing.	True
fit	FitMode | str | None	Resize fit mode: down, cover, contain, fill.	None
anchor	Anchor | str	Anchor point for cover/contain cropping.	CENTER
aspect_ratio	tuple[int, int] | str | None	Target aspect ratio as '16:9' or (16, 9).	None
background_color	tuple[int, int, int] | str	RGB tuple or hex color for contain padding.	WHITE

Returns:

Name	Type	Description
A	BytesResult	class:BytesResult with the optimized image bytes and metadata.

Source code in pixopt/io_bytes.py

def optimize_bytes(
    data: bytes,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    output_format: OutputFormat | str = OutputFormat.WEBP,
    progressive: bool = True,
    optimize: bool = True,
    strip_metadata: bool = True,
    lossless: bool = False,
    auto_orient: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
) -> BytesResult:
    """Optimize an image from raw bytes and return optimized bytes.

    Args:
        data: Raw image file bytes (JPEG, PNG, WEBP, etc.).
        max_width: Maximum width in pixels.
        max_height: Maximum height in pixels.
        quality: JPEG/WEBP quality (1-100).
        output_format: Target format as :class:`OutputFormat` or string.
        progressive: Use progressive JPEG encoding.
        optimize: Enable Pillow optimization flags.
        strip_metadata: Remove EXIF and other metadata.
        lossless: Use lossless compression for PNG/WEBP.
        auto_orient: Apply EXIF orientation before processing.
        fit: Resize fit mode: down, cover, contain, fill.
        anchor: Anchor point for cover/contain cropping.
        aspect_ratio: Target aspect ratio as '16:9' or (16, 9).
        background_color: RGB tuple or hex color for contain padding.

    Returns:
        A :class:`BytesResult` with the optimized image bytes and metadata.

    """
    if len(data) > MAX_INPUT_BYTES:
        return _bytes_error(
            f"Input too large (max {MAX_INPUT_BYTES} bytes)",
            original_size=len(data),
        )

    original_size = len(data)

    try:
        if isinstance(output_format, str):
            output_format = OutputFormat(output_format)
    except ValueError as exc:
        return _bytes_error(str(exc), original_size=original_size)

    try:
        if isinstance(fit, str) and fit:
            fit = FitMode(fit)
        if isinstance(anchor, str):
            anchor = Anchor(anchor)
    except ValueError as exc:
        return _bytes_error(str(exc), original_size=original_size)

    try:
        with Image.open(io.BytesIO(data)) as img:
            image: Image.Image = img
            image.load()
            if image.width > MAX_IMAGE_DIMENSION or image.height > MAX_IMAGE_DIMENSION:
                return _bytes_error(
                    f"Image dimensions too large: {image.width}x{image.height} "
                    f"(max {MAX_IMAGE_DIMENSION})",
                    original_size=original_size,
                )
            # Determine target format.
            source_fmt = image.format or "JPEG"
            if output_format in (OutputFormat.AUTO, OutputFormat.ORIGINAL):
                pillow_fmt = source_fmt
            else:
                pillow_fmt = FORMAT_MAP.get(output_format, "WEBP")

            is_animated = getattr(image, "is_animated", False) or getattr(image, "n_frames", 1) > 1

            if auto_orient and not is_animated:
                image = apply_exif_orientation(image)

            image = convert_mode(image, pillow_fmt)
            image = resize_image(
                image,
                max_width=max_width,
                max_height=max_height,
                fit=fit,
                anchor=anchor,
                aspect_ratio=aspect_ratio,
                background_color=background_color,
            )
            new_width, new_height = image.size

            try:
                optimized_data = image_to_bytes(
                    image,
                    fmt=pillow_fmt,
                    quality=quality,
                    progressive=progressive,
                    optimize=optimize,
                    strip_metadata=strip_metadata,
                    lossless=lossless,
                )
            finally:
                image.close()

            optimized_size = len(optimized_data)
            savings = original_size - optimized_size
            savings_pct = (savings / original_size * 100.0) if original_size > 0 else 0.0

            return BytesResult(
                data=optimized_data,
                format=pillow_fmt,
                width=new_width,
                height=new_height,
                original_size=original_size,
                optimized_size=optimized_size,
                savings_bytes=savings,
                savings_percent=savings_pct,
                success=True,
            )
    except (OSError, ValueError, Image.DecompressionBombError) as exc:
        _logger.error("Optimization failed", extra={"operation": "optimize_bytes"})
        return _bytes_error(str(exc), original_size=original_size)
    except (KeyboardInterrupt, SystemExit):
        raise

optimize_base64(b64_str: str, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, output_format: OutputFormat | str = OutputFormat.WEBP, progressive: bool = True, optimize: bool = True, strip_metadata: bool = True, lossless: bool = False, auto_orient: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE) -> Base64Result

Optimize an image from a base64 string and return base64 + metadata.

Parameters:

Name	Type	Description	Default
b64_str	str	Base64-encoded image data.	required

Returns:

Name	Type	Description
A	Base64Result	class:Base64Result with base64 data and optimization metadata.

Source code in pixopt/io_bytes.py

def optimize_base64(
    b64_str: str,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    output_format: OutputFormat | str = OutputFormat.WEBP,
    progressive: bool = True,
    optimize: bool = True,
    strip_metadata: bool = True,
    lossless: bool = False,
    auto_orient: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
) -> Base64Result:
    """Optimize an image from a base64 string and return base64 + metadata.

    Args:
        b64_str: Base64-encoded image data.

    Returns:
        A :class:`Base64Result` with base64 data and optimization metadata.

    """
    try:
        data = _b64.b64decode(b64_str)
    except (ValueError, TypeError, binascii.Error) as exc:
        return Base64Result(
            base64=None,
            format="",
            width=0,
            height=0,
            original_size=0,
            optimized_size=0,
            savings_bytes=0,
            savings_percent=0.0,
            success=False,
            error=f"Invalid base64 data: {exc}",
        )

    result = optimize_bytes(
        data,
        max_width=max_width,
        max_height=max_height,
        quality=quality,
        output_format=output_format,
        progressive=progressive,
        optimize=optimize,
        strip_metadata=strip_metadata,
        lossless=lossless,
        auto_orient=auto_orient,
        fit=fit,
        anchor=anchor,
        aspect_ratio=aspect_ratio,
        background_color=background_color,
    )

    return Base64Result(
        base64=_b64.b64encode(result.data).decode("ascii") if result.success else None,
        format=result.format,
        width=result.width,
        height=result.height,
        original_size=result.original_size,
        optimized_size=result.optimized_size,
        savings_bytes=result.savings_bytes,
        savings_percent=result.savings_percent,
        success=result.success,
        error=result.error,
    )

BytesResult(data: bytes, format: str, width: int, height: int, original_size: int, optimized_size: int, savings_bytes: int, savings_percent: float, success: bool, error: str | None = None) dataclass

Result of an in-memory image optimization.

Attributes

data: bytes instance-attribute

format: str instance-attribute

width: int instance-attribute

height: int instance-attribute

original_size: int instance-attribute

optimized_size: int instance-attribute

savings_bytes: int instance-attribute

savings_percent: float instance-attribute

success: bool instance-attribute

error: str | None = None class-attribute instance-attribute

human_original_size: str property

human_optimized_size: str property

Methods:

to_dict() -> dict[str, Any]

Source code in pixopt/io_bytes.py

def to_dict(self) -> dict[str, Any]:
    return {
        "format": self.format,
        "width": self.width,
        "height": self.height,
        "original_size": self.original_size,
        "optimized_size": self.optimized_size,
        "savings_bytes": self.savings_bytes,
        "savings_percent": self.savings_percent,
        "success": self.success,
        "error": self.error,
    }

Base64Result(base64: str | None, format: str, width: int, height: int, original_size: int, optimized_size: int, savings_bytes: int, savings_percent: float, success: bool, error: str | None = None) dataclass

Result of an in-memory base64 image optimization.

Attributes

base64: str | None instance-attribute

format: str instance-attribute

width: int instance-attribute

height: int instance-attribute

original_size: int instance-attribute

optimized_size: int instance-attribute

savings_bytes: int instance-attribute

savings_percent: float instance-attribute

success: bool instance-attribute

error: str | None = None class-attribute instance-attribute

human_original_size: str property

human_optimized_size: str property

Methods:

to_dict() -> dict[str, Any]

Source code in pixopt/io_bytes.py

def to_dict(self) -> dict[str, Any]:
    return {
        "base64": self.base64,
        "format": self.format,
        "width": self.width,
        "height": self.height,
        "original_size": self.original_size,
        "optimized_size": self.optimized_size,
        "savings_bytes": self.savings_bytes,
        "savings_percent": self.savings_percent,
        "success": self.success,
        "error": self.error,
    }

PDF

images_to_pdf(images: Sequence[Path | str], output: Path | str, *, title: str | None = None) -> PdfExportResult

Combine multiple images into a single PDF file.

Parameters:

Name	Type	Description	Default
images	Sequence[Path | str]	List of image file paths. Each image becomes one page.	required
output	Path | str	Output PDF file path.	required
title	str | None	Optional PDF document title metadata.	None

Returns:

Name	Type	Description
A	PdfExportResult	class:PdfExportResult with the output path and page count.

Raises:

Type	Description
ValueError	If the image list is empty.

Source code in pixopt/pdf_io.py

def images_to_pdf(
    images: Sequence[Path | str],
    output: Path | str,
    *,
    title: str | None = None,
) -> PdfExportResult:
    """Combine multiple images into a single PDF file.

    Args:
        images: List of image file paths. Each image becomes one page.
        output: Output PDF file path.
        title: Optional PDF document title metadata.

    Returns:
        A :class:`PdfExportResult` with the output path and page count.

    Raises:
        ValueError: If the image list is empty.

    """
    if not images:
        raise ValueError("images list cannot be empty")

    output_path = Path(output)
    if error := validate_no_parent_references(output_path, "output"):
        raise ValueError(error)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    _logger.debug(
        "Converting images to PDF", extra={"operation": "pdf_export", "path": str(output_path)}
    )

    if len(images) > MAX_PDF_IMAGES:
        raise ValueError(f"Too many images for PDF (max {MAX_PDF_IMAGES}), got {len(images)}")

    pil_images: list[Image.Image] = []
    try:
        total_pixels = 0
        for img_path in images:
            image: Image.Image
            with _open_image(Path(img_path), label="source") as img:
                image = img
            if image.mode != "RGB":
                image = image.convert("RGB")
            total_pixels += image.width * image.height
            if total_pixels > MAX_PDF_TOTAL_PIXELS:
                raise ValueError(
                    f"PDF total pixel budget exceeded: {total_pixels} (max {MAX_PDF_TOTAL_PIXELS})"
                )
            pil_images.append(image)

        save_kwargs: dict[str, Any] = {}
        if title:
            save_kwargs["title"] = title

        pil_images[0].save(
            output_path,
            format="PDF",
            save_all=True,
            append_images=pil_images[1:],
            **save_kwargs,
        )

        result = PdfExportResult(
            output=output_path,
            page_count=len(pil_images),
            success=True,
        )

        _logger.info(
            "PDF export complete",
            extra={
                "operation": "pdf_export",
                "path": str(output_path),
                "size_bytes": len(pil_images),
            },
        )

        return result

    except (OSError, ValueError) as exc:
        _logger.error(
            "PDF export failed", extra={"operation": "pdf_export", "path": str(output_path)}
        )
        return PdfExportResult(
            output=output_path,
            page_count=0,
            success=False,
            error=str(exc),
        )
    finally:
        for img in pil_images:
            img.close()

pdf_to_images(source: Path | str, output_dir: Path | str, *, dpi: int = 150, fmt: str = 'PNG', prefix: str | None = None) -> PdfImportResult

Convert each page of a PDF to an image file.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the PDF file.	required
output_dir	Path | str	Directory where page images will be saved.	required
dpi	int	Render resolution in DPI (default 150).	150
fmt	str	Output image format ("PNG", "JPEG", "WEBP").	'PNG'
prefix	str | None	Filename prefix. Defaults to the PDF stem.	None

Returns:

Name	Type	Description
A	PdfImportResult	class:PdfImportResult with per-page info.

Raises:

Type	Description
FileNotFoundError	If the PDF does not exist.

Source code in pixopt/pdf_io.py

def pdf_to_images(
    source: Path | str,
    output_dir: Path | str,
    *,
    dpi: int = 150,
    fmt: str = "PNG",
    prefix: str | None = None,
) -> PdfImportResult:
    """Convert each page of a PDF to an image file.

    Args:
        source: Path to the PDF file.
        output_dir: Directory where page images will be saved.
        dpi: Render resolution in DPI (default 150).
        fmt: Output image format (``"PNG"``, ``"JPEG"``, ``"WEBP"``).
        prefix: Filename prefix. Defaults to the PDF stem.

    Returns:
        A :class:`PdfImportResult` with per-page info.

    Raises:
        FileNotFoundError: If the PDF does not exist.

    """
    source_path = Path(source)
    if not source_path.exists():
        raise FileNotFoundError(f"PDF file not found: {source_path}")

    if fitz is None:
        return PdfImportResult(
            source=source_path,
            success=False,
            error="PDF support requires PyMuPDF. Install it with: pip install pixopt[pdf]",
        )

    if source_path.stat().st_size > MAX_PDF_SIZE_BYTES:
        return PdfImportResult(
            source=source_path,
            success=False,
            error=f"PDF file exceeds maximum size of {MAX_PDF_SIZE_BYTES} bytes",
        )

    if dpi < 1 or dpi > MAX_PDF_DPI:
        return PdfImportResult(
            source=source_path,
            success=False,
            error=f"dpi must be between 1 and {MAX_PDF_DPI}, got {dpi}",
        )

    out_dir = Path(output_dir)
    if error := validate_no_parent_references(out_dir, "output_dir"):
        raise ValueError(error)
    out_dir.mkdir(parents=True, exist_ok=True)

    name_prefix = prefix or source_path.stem
    pillow_fmt = fmt.upper()

    _logger.debug(
        "Converting PDF to images", extra={"operation": "pdf_import", "path": str(source_path)}
    )

    doc: Any | None = None
    try:
        doc = fitz.open(source_path)

        if len(doc) > MAX_PDF_PAGES:
            return PdfImportResult(
                source=source_path,
                success=False,
                error=f"PDF has too many pages (max {MAX_PDF_PAGES})",
            )
        pages: list[PdfPageInfo] = []

        zoom = dpi / 72.0
        matrix = fitz.Matrix(zoom, zoom)

        for page_num in range(len(doc)):
            page = doc.load_page(page_num)
            pix = page.get_pixmap(matrix=matrix)
            if pix.width > MAX_IMAGE_DIMENSION or pix.height > MAX_IMAGE_DIMENSION:
                return PdfImportResult(
                    source=source_path,
                    success=False,
                    error=f"PDF page dimensions too large: {pix.width}x{pix.height} "
                    f"(max {MAX_IMAGE_DIMENSION})",
                )
            img = Image.frombytes("RGB", (pix.width, pix.height), pix.samples)

            out_path = out_dir / f"{name_prefix}_{page_num + 1:04d}.{pillow_fmt.lower()}"
            img.save(out_path, pillow_fmt)
            img.close()

            pages.append(
                PdfPageInfo(
                    page_number=page_num + 1,
                    width=pix.width,
                    height=pix.height,
                    output_path=out_path,
                )
            )

        result = PdfImportResult(
            source=source_path,
            pages=pages,
            total_pages=len(pages),
            success=True,
        )

        _logger.info(
            "PDF import complete",
            extra={"operation": "pdf_import", "path": str(source_path), "size_bytes": len(pages)},
        )

        return result

    except (OSError, ValueError) as exc:
        _logger.error(
            "PDF import failed", extra={"operation": "pdf_import", "path": str(source_path)}
        )
        return PdfImportResult(
            source=source_path,
            total_pages=0,
            success=False,
            error=str(exc),
        )
    finally:
        if doc is not None:
            doc.close()

Perceptual hashing

compute_hash(source: Path | str, *, algorithm: str = 'phash', hash_size: int = _DEFAULT_HASH_SIZE) -> HashResult

Compute a perceptual hash of an image.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the image file.	required
algorithm	str	One of "phash", "dhash", "ahash".	'phash'
hash_size	int	Hash size in bits per side (default 8 → 64-bit hash).	_DEFAULT_HASH_SIZE

Returns:

Name	Type	Description
A	HashResult	class:HashResult with the computed hash.

Source code in pixopt/perceptual.py

def compute_hash(
    source: Path | str,
    *,
    algorithm: str = "phash",
    hash_size: int = _DEFAULT_HASH_SIZE,
) -> HashResult:
    """Compute a perceptual hash of an image.

    Args:
        source: Path to the image file.
        algorithm: One of ``"phash"``, ``"dhash"``, ``"ahash"``.
        hash_size: Hash size in bits per side (default 8 → 64-bit hash).

    Returns:
        A :class:`HashResult` with the computed hash.

    """
    algorithm = algorithm.lower()
    if algorithm == "phash":
        h = phash(source, hash_size=hash_size)
    elif algorithm == "dhash":
        h = dhash(source, hash_size=hash_size)
    elif algorithm == "ahash":
        h = ahash(source, hash_size=hash_size)
    else:
        raise ValueError(
            f"Unknown hash algorithm: {algorithm!r}. Use 'phash', 'dhash', or 'ahash'."
        )

    return HashResult(
        file_path=Path(source),
        algorithm=algorithm,
        hash_hex=h,
        hash_size=hash_size,
    )

phash(source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE, highfreq_factor: int = 4) -> str

Compute the perceptual hash (pHash) of an image.

Uses a DCT-based approach: resize to hash_size * highfreq_factor, apply a slight blur, compute the DCT, and take the low-frequency components compared to the median.

Returns:

Type	Description
str	A hex string representing the hash.

Source code in pixopt/perceptual.py

def phash(
    source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE, highfreq_factor: int = 4
) -> str:
    """Compute the perceptual hash (pHash) of an image.

    Uses a DCT-based approach: resize to ``hash_size * highfreq_factor``,
    apply a slight blur, compute the DCT, and take the low-frequency
    components compared to the median.

    Returns:
        A hex string representing the hash.
    """

    if highfreq_factor < 1 or highfreq_factor > 16:
        raise ValueError(f"highfreq_factor must be between 1 and 16, got {highfreq_factor}")

    img_size = hash_size * highfreq_factor
    img = _img_to_grayscale(source, img_size)
    try:
        img = img.filter(ImageFilter.MedianFilter(size=3))
        pixels = _pixel_data(img)

        # Compute 1D DCT for each row, then for each column.
        dct_matrix = _compute_dct_2d(pixels, img_size, img_size)

        # Take the top-left hash_size × hash_size low-frequency block.
        low_freq = []
        for row in range(hash_size):
            for col in range(hash_size):
                low_freq.append(dct_matrix[row * img_size + col])

        # Compare to median (excluding the DC component at [0,0]).
        tail = low_freq[1:]
        median = sorted(tail)[len(tail) // 2] if tail else low_freq[0] if low_freq else 0
        bits = [1 if v > median else 0 for v in low_freq]
        return _bits_to_hex(bits)
    finally:
        img.close()

ahash(source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE) -> str

Compute the average hash (aHash) of an image.

The image is resized to hash_size × hash_size grayscale, then each pixel is compared to the mean: above → 1, below → 0.

Returns:

Type	Description
str	A hex string representing the hash.

Source code in pixopt/perceptual.py

def ahash(source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE) -> str:
    """Compute the average hash (aHash) of an image.

    The image is resized to ``hash_size × hash_size`` grayscale, then each
    pixel is compared to the mean: above → 1, below → 0.

    Returns:
        A hex string representing the hash.
    """
    img = _img_to_grayscale(source, hash_size)
    try:
        pixels = _pixel_data(img)
        avg = sum(pixels) / len(pixels) if pixels else 0
        bits = [1 if p >= avg else 0 for p in pixels]
        return _bits_to_hex(bits)
    finally:
        img.close()

dhash(source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE) -> str

Compute the difference hash (dHash) of an image.

Compares each pixel to its right neighbor: left < right → 1.

Returns:

Type	Description
str	A hex string representing the hash.

Source code in pixopt/perceptual.py

def dhash(source: Path | str, *, hash_size: int = _DEFAULT_HASH_SIZE) -> str:
    """Compute the difference hash (dHash) of an image.

    Compares each pixel to its right neighbor: left < right → 1.

    Returns:
        A hex string representing the hash.
    """
    img = _img_to_grayscale(source, hash_size)
    try:
        pixels = _pixel_data(img)
        width = hash_size
        bits = []
        for row in range(hash_size):
            for col in range(hash_size - 1):
                left = pixels[row * width + col]
                right = pixels[row * width + col + 1]
                bits.append(1 if left < right else 0)
        return _bits_to_hex(bits)
    finally:
        img.close()

find_duplicates(hashes: list[HashResult], *, threshold: int = 5) -> list[DuplicateGroup]

Find duplicate or near-duplicate images from a list of hashes.

Parameters:

Name	Type	Description	Default
hashes	list[HashResult]	List of :class:HashResult objects.	required
threshold	int	Maximum Hamming distance to consider images as duplicates.	5

Returns:

Type	Description
list[DuplicateGroup]	A list of :class:DuplicateGroup objects, each containing files
list[DuplicateGroup]	that are within threshold of each other.

Source code in pixopt/perceptual.py

def find_duplicates(
    hashes: list[HashResult],
    *,
    threshold: int = 5,
) -> list[DuplicateGroup]:
    """Find duplicate or near-duplicate images from a list of hashes.

    Args:
        hashes: List of :class:`HashResult` objects.
        threshold: Maximum Hamming distance to consider images as duplicates.

    Returns:
        A list of :class:`DuplicateGroup` objects, each containing files
        that are within ``threshold`` of each other.

    """
    if not hashes:
        return []

    limit = MAX_DUPLICATE_SCAN if threshold == 0 else MAX_NEAR_DUPLICATE_SCAN
    if len(hashes) > limit:
        raise ValueError(f"Too many hashes to compare (max {limit}), got {len(hashes)}")

    groups: list[DuplicateGroup] = []

    if threshold == 0:
        # Fast path for exact duplicates: O(n log n) sort and group.
        sorted_hashes = sorted(hashes, key=lambda h: (h.algorithm, h.hash_hex))
        i = 0
        while i < len(sorted_hashes):
            j = i + 1
            while (
                j < len(sorted_hashes)
                and sorted_hashes[j].algorithm == sorted_hashes[i].algorithm
                and sorted_hashes[j].hash_hex == sorted_hashes[i].hash_hex
            ):
                j += 1
            if j - i > 1:
                groups.append(
                    DuplicateGroup(
                        hash_hex=sorted_hashes[i].hash_hex,
                        algorithm=sorted_hashes[i].algorithm,
                        files=[h.file_path for h in sorted_hashes[i:j]],
                    )
                )
            i = j
        return groups

    assigned: set[int] = set()

    for i, h in enumerate(hashes):
        if i in assigned:
            continue

        group = DuplicateGroup(hash_hex=h.hash_hex, algorithm=h.algorithm, files=[h.file_path])
        assigned.add(i)

        for j in range(i + 1, len(hashes)):
            if j in assigned:
                continue
            if hashes[j].algorithm != h.algorithm:
                continue
            dist = hamming_distance(h.hash_hex, hashes[j].hash_hex)
            if dist <= threshold:
                group.files.append(hashes[j].file_path)
                group.distances[str(hashes[j].file_path)] = dist
                assigned.add(j)

        if group.count > 1:
            groups.append(group)

    return groups

scan_duplicates(directory: Path | str, *, algorithm: str = 'phash', threshold: int = 5, recursive: bool = False, hash_size: int = _DEFAULT_HASH_SIZE) -> DuplicateReport

Scan a directory for duplicate or near-duplicate images.

Parameters:

Name	Type	Description	Default
directory	Path | str	Directory to scan.	required
algorithm	str	Hash algorithm: "phash", "dhash", or "ahash".	'phash'
threshold	int	Maximum Hamming distance for duplicates.	5
recursive	bool	Scan subdirectories recursively.	False
hash_size	int	Hash size per side.	_DEFAULT_HASH_SIZE

Returns:

Name	Type	Description
A	DuplicateReport	class:DuplicateReport with all duplicate groups found.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.

Source code in pixopt/perceptual.py

def scan_duplicates(
    directory: Path | str,
    *,
    algorithm: str = "phash",
    threshold: int = 5,
    recursive: bool = False,
    hash_size: int = _DEFAULT_HASH_SIZE,
) -> DuplicateReport:
    """Scan a directory for duplicate or near-duplicate images.

    Args:
        directory: Directory to scan.
        algorithm: Hash algorithm: ``"phash"``, ``"dhash"``, or ``"ahash"``.
        threshold: Maximum Hamming distance for duplicates.
        recursive: Scan subdirectories recursively.
        hash_size: Hash size per side.

    Returns:
        A :class:`DuplicateReport` with all duplicate groups found.

    Raises:
        FileNotFoundError: If the directory does not exist.

    """
    dir_path = Path(directory)
    if not dir_path.exists():
        raise FileNotFoundError(f"Directory not found: {dir_path}")

    _logger.debug(
        "Scanning for duplicates", extra={"operation": "duplicates", "path": str(dir_path)}
    )

    limit = MAX_DUPLICATE_SCAN if threshold == 0 else MAX_NEAR_DUPLICATE_SCAN
    hashes: list[HashResult] = []
    for file_path in discover_images(dir_path, recursive=recursive):
        if len(hashes) >= limit:
            _logger.warning(
                "Duplicate scan stopped after reaching the limit",
                extra={"operation": "duplicates", "limit": limit},
            )
            break
        try:
            h = compute_hash(file_path, algorithm=algorithm, hash_size=hash_size)
            hashes.append(h)
        except (OSError, ValueError, Image.DecompressionBombError):
            _logger.warning(
                "Failed to hash", extra={"operation": "duplicates", "path": str(file_path)}
            )

    groups = find_duplicates(hashes, threshold=threshold)

    report = DuplicateReport(
        directory=dir_path,
        algorithm=algorithm,
        threshold=threshold,
        total_files=len(hashes),
        duplicate_groups=groups,
        total_duplicates=sum(g.count for g in groups),
    )

    _logger.info(
        "Duplicate scan complete",
        extra={"operation": "duplicates", "path": str(dir_path), "size_bytes": len(groups)},
    )

    return report

hamming_distance(hash_a: str, hash_b: str) -> int

Compute the Hamming distance between two hex hash strings.

Returns:

Type	Description
int	The number of differing bits. 0 means identical.

Source code in pixopt/perceptual.py

def hamming_distance(hash_a: str, hash_b: str) -> int:
    """Compute the Hamming distance between two hex hash strings.

    Returns:
        The number of differing bits.  0 means identical.
    """
    if not hash_a or not hash_b:
        raise ValueError("hash strings cannot be empty")
    bits_a = _hex_to_bits(hash_a)
    bits_b = _hex_to_bits(hash_b)
    if len(bits_a) != len(bits_b):
        # Pad shorter with zeros
        max_len = max(len(bits_a), len(bits_b))
        bits_a = bits_a + [0] * (max_len - len(bits_a))
        bits_b = bits_b + [0] * (max_len - len(bits_b))
    return sum(a != b for a, b in zip(bits_a, bits_b, strict=True))

Async API

async_optimize_image(source: Path | str, output: Path | str | None = None, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, strip_metadata: bool = True, output_format: OutputFormat = OutputFormat.AUTO, keep_aspect_ratio: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE, auto_orient: bool = True, progressive: bool = True, optimize: bool = True, overwrite: bool = False, lossless: bool = False, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, keep_exif_groups: set[EXIFGroup] | None = None) -> OptimizationResult async

Async variant of :func:pixopt.optimize_image.

Runs the synchronous optimization in a background thread.

Source code in pixopt/async_api.py

async def async_optimize_image(
    source: Path | str,
    output: Path | str | None = None,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    strip_metadata: bool = True,
    output_format: OutputFormat = OutputFormat.AUTO,
    keep_aspect_ratio: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
    auto_orient: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    overwrite: bool = False,
    lossless: bool = False,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    keep_exif_groups: set[EXIFGroup] | None = None,
) -> OptimizationResult:
    """Async variant of :func:`pixopt.optimize_image`.

    Runs the synchronous optimization in a background thread.
    """
    if error := validate_optimize_params(
        quality=quality,
        max_width=max_width,
        max_height=max_height,
        min_size_bytes=min_size_bytes,
    ):
        raise ValueError(error)
    return await asyncio.to_thread(
        optimize_image,
        source,
        output,
        max_width=max_width,
        max_height=max_height,
        quality=quality,
        strip_metadata=strip_metadata,
        output_format=output_format,
        keep_aspect_ratio=keep_aspect_ratio,
        fit=fit,
        anchor=anchor,
        aspect_ratio=aspect_ratio,
        background_color=background_color,
        auto_orient=auto_orient,
        progressive=progressive,
        optimize=optimize,
        overwrite=overwrite,
        lossless=lossless,
        backup_dir=backup_dir,
        min_size_bytes=min_size_bytes,
        keep_exif_groups=keep_exif_groups,
    )

async_batch_optimize(sources: Sequence[Path | str], output_dir: Path | str, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, strip_metadata: bool = True, output_format: OutputFormat = OutputFormat.AUTO, keep_aspect_ratio: bool = True, progressive: bool = True, optimize: bool = True, overwrite: bool = False, lossless: bool = False, backup_dir: Path | str | None = None, min_size_bytes: int | None = None, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE, auto_orient: bool = True, keep_exif_groups: set[EXIFGroup] | None = None, max_concurrency: int = 4, on_progress: ProgressCallback | None = None) -> BatchReport async

Async variant of :func:pixopt.batch_optimize.

Processes images concurrently using max_concurrency parallel tasks, each running in a background thread.

Parameters:

Name	Type	Description	Default
sources	Sequence[Path | str]	List of source image paths.	required
output_dir	Path | str	Output directory for optimized images.	required
max_concurrency	int	Maximum number of images to process in parallel.	4
on_progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Name	Type	Description
A	BatchReport	class:BatchReport with aggregated results.

Source code in pixopt/async_api.py

async def async_batch_optimize(
    sources: Sequence[Path | str],
    output_dir: Path | str,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    strip_metadata: bool = True,
    output_format: OutputFormat = OutputFormat.AUTO,
    keep_aspect_ratio: bool = True,
    progressive: bool = True,
    optimize: bool = True,
    overwrite: bool = False,
    lossless: bool = False,
    backup_dir: Path | str | None = None,
    min_size_bytes: int | None = None,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
    auto_orient: bool = True,
    keep_exif_groups: set[EXIFGroup] | None = None,
    max_concurrency: int = 4,
    on_progress: ProgressCallback | None = None,
) -> BatchReport:
    """Async variant of :func:`pixopt.batch_optimize`.

    Processes images concurrently using ``max_concurrency`` parallel tasks,
    each running in a background thread.

    Args:
        sources: List of source image paths.
        output_dir: Output directory for optimized images.
        max_concurrency: Maximum number of images to process in parallel.
        on_progress: Optional progress callback.

    Returns:
        A :class:`BatchReport` with aggregated results.

    """
    import time

    out_dir = Path(output_dir)
    if error := validate_no_parent_references(out_dir, "output_dir"):
        raise ValueError(error)
    out_dir.mkdir(parents=True, exist_ok=True)

    if error := validate_optimize_params(
        quality=quality,
        max_width=max_width,
        max_height=max_height,
        min_size_bytes=min_size_bytes,
    ):
        raise ValueError(error)

    if max_concurrency <= 0:
        raise ValueError(f"max_concurrency must be positive, got {max_concurrency}")

    source_list = list(sources)
    total = len(source_list)

    start = time.perf_counter()
    results: list[OptimizationResult | None] = [None] * total

    semaphore = asyncio.Semaphore(max_concurrency)

    async def _process_one(idx: int, src: Path | str) -> None:
        src_path = Path(src)
        out_path = out_dir / src_path.name
        try:
            async with semaphore:
                result = await async_optimize_image(
                    src_path,
                    out_path,
                    max_width=max_width,
                    max_height=max_height,
                    quality=quality,
                    strip_metadata=strip_metadata,
                    output_format=output_format,
                    keep_aspect_ratio=keep_aspect_ratio,
                    progressive=progressive,
                    optimize=optimize,
                    fit=fit,
                    anchor=anchor,
                    aspect_ratio=aspect_ratio,
                    background_color=background_color,
                    auto_orient=auto_orient,
                    keep_exif_groups=keep_exif_groups,
                    overwrite=overwrite,
                    lossless=lossless,
                    backup_dir=backup_dir,
                    min_size_bytes=min_size_bytes,
                )
            results[idx] = result
        except (OSError, ValueError, Image.DecompressionBombError, RuntimeError) as exc:
            result = OptimizationResult(
                source_path=src_path,
                output_path=out_path,
                original_size=0,
                optimized_size=0,
                savings_bytes=0,
                savings_percent=0.0,
                width=0,
                height=0,
                format="",
                metadata_removed=False,
                success=False,
                error=f"{exc}",
            )
            results[idx] = result
        if on_progress is not None:
            await asyncio.get_running_loop().run_in_executor(
                None,
                on_progress,
                ProgressInfo(
                    current=idx + 1,
                    total=total,
                    current_file=src_path,
                    success=result.success,
                    message=result.error or "",
                ),
            )

    tasks = [asyncio.create_task(_process_one(i, src)) for i, src in enumerate(source_list)]
    gathered = await asyncio.gather(*tasks, return_exceptions=True)
    for i, value in enumerate(gathered):
        if isinstance(value, BaseException) and not isinstance(value, Exception):
            raise value
        if isinstance(value, Exception):
            if results[i] is not None:
                raise value
            src_path = Path(source_list[i])
            out_path = out_dir / src_path.name
            results[i] = OptimizationResult(
                source_path=src_path,
                output_path=out_path,
                original_size=0,
                optimized_size=0,
                savings_bytes=0,
                savings_percent=0.0,
                width=0,
                height=0,
                format="",
                metadata_removed=False,
                success=False,
                error=str(value),
            )

    elapsed = time.perf_counter() - start

    # Build the BatchReport from results
    valid_results = [r for r in results if r is not None]
    return _build_batch_report(valid_results, total, elapsed)

async_optimize_bytes(data: bytes, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, output_format: OutputFormat | str = OutputFormat.WEBP, progressive: bool = True, optimize: bool = True, strip_metadata: bool = True, lossless: bool = False, auto_orient: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE) -> BytesResult async

Async variant of :func:pixopt.optimize_bytes.

Runs in-memory optimization in a background thread.

Source code in pixopt/async_api.py

async def async_optimize_bytes(
    data: bytes,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    output_format: OutputFormat | str = OutputFormat.WEBP,
    progressive: bool = True,
    optimize: bool = True,
    strip_metadata: bool = True,
    lossless: bool = False,
    auto_orient: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
) -> BytesResult:
    """Async variant of :func:`pixopt.optimize_bytes`.

    Runs in-memory optimization in a background thread.
    """
    from pixopt.io_bytes import optimize_bytes

    return await asyncio.to_thread(
        optimize_bytes,
        data,
        max_width=max_width,
        max_height=max_height,
        quality=quality,
        output_format=output_format,
        progressive=progressive,
        optimize=optimize,
        strip_metadata=strip_metadata,
        lossless=lossless,
        auto_orient=auto_orient,
        fit=fit,
        anchor=anchor,
        aspect_ratio=aspect_ratio,
        background_color=background_color,
    )

async_optimize_base64(b64_str: str, *, max_width: int | None = None, max_height: int | None = None, quality: int = 85, output_format: OutputFormat | str = OutputFormat.WEBP, progressive: bool = True, optimize: bool = True, strip_metadata: bool = True, lossless: bool = False, auto_orient: bool = True, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str = WHITE) -> Base64Result async

Async variant of :func:pixopt.optimize_base64.

Runs in-memory base64 optimization in a background thread.

Source code in pixopt/async_api.py

async def async_optimize_base64(
    b64_str: str,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    quality: int = 85,
    output_format: OutputFormat | str = OutputFormat.WEBP,
    progressive: bool = True,
    optimize: bool = True,
    strip_metadata: bool = True,
    lossless: bool = False,
    auto_orient: bool = True,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str = WHITE,
) -> Base64Result:
    """Async variant of :func:`pixopt.optimize_base64`.

    Runs in-memory base64 optimization in a background thread.
    """
    from pixopt.io_bytes import optimize_base64

    return await asyncio.to_thread(
        optimize_base64,
        b64_str,
        max_width=max_width,
        max_height=max_height,
        quality=quality,
        output_format=output_format,
        progressive=progressive,
        optimize=optimize,
        strip_metadata=strip_metadata,
        lossless=lossless,
        auto_orient=auto_orient,
        fit=fit,
        anchor=anchor,
        aspect_ratio=aspect_ratio,
        background_color=background_color,
    )

async_inspect_image(source: Path | str) -> ImageInfo async

Async variant of :func:pixopt.inspect_image.

Runs image inspection in a background thread.

Source code in pixopt/async_api.py

async def async_inspect_image(source: Path | str) -> ImageInfo:
    """Async variant of :func:`pixopt.inspect_image`.

    Runs image inspection in a background thread.
    """
    from pixopt.inspect import inspect_image

    return await asyncio.to_thread(inspect_image, source)

async_scan_directory(directory: Path | str, *, recursive: bool = False, extensions: Iterable[str] | None = None) -> ScanReport async

Async variant of :func:pixopt.scan_directory.

Runs directory scan in a background thread.

Source code in pixopt/async_api.py

async def async_scan_directory(
    directory: Path | str,
    *,
    recursive: bool = False,
    extensions: Iterable[str] | None = None,
) -> ScanReport:
    """Async variant of :func:`pixopt.scan_directory`.

    Runs directory scan in a background thread.
    """
    return await asyncio.to_thread(
        scan_directory,
        directory,
        recursive=recursive,
        extensions=extensions,
    )

async_scan_duplicates(directory: Path | str, *, algorithm: str = 'phash', threshold: int = 5, recursive: bool = True, hash_size: int = 8) -> DuplicateReport async

Async variant of :func:pixopt.scan_duplicates.

Runs duplicate scan in a background thread.

Source code in pixopt/async_api.py

async def async_scan_duplicates(
    directory: Path | str,
    *,
    algorithm: str = "phash",
    threshold: int = 5,
    recursive: bool = True,
    hash_size: int = 8,
) -> DuplicateReport:
    """Async variant of :func:`pixopt.scan_duplicates`.

    Runs duplicate scan in a background thread.
    """
    return await asyncio.to_thread(
        scan_duplicates,
        directory,
        algorithm=algorithm,
        threshold=threshold,
        recursive=recursive,
        hash_size=hash_size,
    )

Pipeline

Pipeline()

Chainable image processing pipeline.

Build a sequence of operations and execute them in one call::

pipeline = (
    Pipeline()
    .open("photo.jpg")
    .resize(max_width=800)
    .watermark_text("© 2025", position="bottom-right")
    .optimize(quality=85, output_format="webp")
    .save("output/photo.webp")
)
result = pipeline.run()

Each method returns self for fluent chaining.

Source code in pixopt/pipeline.py

def __init__(self) -> None:
    self._ops: list[_Op] = []
    self._source: Path | None = None

Attributes

steps: list[str] property

Return a human-readable list of queued operation names.

Methods:

open(source: Path | str) -> Pipeline

Set the source image to process.

Source code in pixopt/pipeline.py

def open(self, source: Path | str) -> Pipeline:
    """Set the source image to process."""
    src = Path(source)
    if error := validate_no_parent_references(src, "source"):
        raise ValueError(error)
    self._source = src
    self._ops.append(_Op(_OpKind.OPEN, {"source": self._source}))
    return self

auto_orient() -> Pipeline

Apply EXIF orientation tag.

Source code in pixopt/pipeline.py

def auto_orient(self) -> Pipeline:
    """Apply EXIF orientation tag."""
    self._ops.append(_Op(_OpKind.AUTO_ORIENT))
    return self

resize(*, max_width: int | None = None, max_height: int | None = None, fit: FitMode | str | None = None, anchor: Anchor | str = Anchor.CENTER, aspect_ratio: tuple[int, int] | str | None = None, background_color: tuple[int, int, int] | str | None = None) -> Pipeline

Resize the image with optional fit mode.

Source code in pixopt/pipeline.py

def resize(
    self,
    *,
    max_width: int | None = None,
    max_height: int | None = None,
    fit: FitMode | str | None = None,
    anchor: Anchor | str = Anchor.CENTER,
    aspect_ratio: tuple[int, int] | str | None = None,
    background_color: tuple[int, int, int] | str | None = None,
) -> Pipeline:
    """Resize the image with optional fit mode."""
    self._ops.append(
        _Op(
            _OpKind.RESIZE,
            {
                "max_width": max_width,
                "max_height": max_height,
                "fit": fit,
                "anchor": anchor,
                "aspect_ratio": aspect_ratio,
                "background_color": background_color,
            },
        )
    )
    return self

convert(mode: str = 'RGB') -> Pipeline

Convert the image color mode (e.g. "RGB", "RGBA", "L").

Source code in pixopt/pipeline.py

def convert(self, mode: str = "RGB") -> Pipeline:
    """Convert the image color mode (e.g. ``"RGB"``, ``"RGBA"``, ``"L"``)."""
    self._ops.append(_Op(_OpKind.CONVERT, {"mode": mode}))
    return self

watermark_text(text: str, *, position: WatermarkPosition | str = WatermarkPosition.BOTTOM_RIGHT, opacity: float = 0.5, padding: int = 20, font_size: int = 48, font_path: Path | str | None = None, color: tuple[int, int, int] = (255, 255, 255)) -> Pipeline

Add a text watermark overlay.

Source code in pixopt/pipeline.py

def watermark_text(
    self,
    text: str,
    *,
    position: WatermarkPosition | str = WatermarkPosition.BOTTOM_RIGHT,
    opacity: float = 0.5,
    padding: int = 20,
    font_size: int = 48,
    font_path: Path | str | None = None,
    color: tuple[int, int, int] = (255, 255, 255),
) -> Pipeline:
    """Add a text watermark overlay."""
    if isinstance(position, str):
        position = WatermarkPosition(position)
    if font_path is not None and (
        error := validate_no_parent_references(Path(font_path), "font_path")
    ):
        raise ValueError(error)
    self._ops.append(
        _Op(
            _OpKind.WATERMARK_TEXT,
            {
                "text": text,
                "position": position,
                "opacity": opacity,
                "padding": padding,
                "font_size": font_size,
                "font_path": font_path,
                "color": color,
            },
        )
    )
    return self

watermark_image(watermark_path: Path | str, *, position: WatermarkPosition | str = WatermarkPosition.BOTTOM_RIGHT, opacity: float = 0.5, padding: int = 20, scale: float = 0.3) -> Pipeline

Add an image watermark overlay.

Source code in pixopt/pipeline.py

def watermark_image(
    self,
    watermark_path: Path | str,
    *,
    position: WatermarkPosition | str = WatermarkPosition.BOTTOM_RIGHT,
    opacity: float = 0.5,
    padding: int = 20,
    scale: float = 0.3,
) -> Pipeline:
    """Add an image watermark overlay."""
    if isinstance(position, str):
        position = WatermarkPosition(position)
    if error := validate_no_parent_references(Path(watermark_path), "watermark_path"):
        raise ValueError(error)
    self._ops.append(
        _Op(
            _OpKind.WATERMARK_IMAGE,
            {
                "watermark_path": Path(watermark_path),
                "position": position,
                "opacity": opacity,
                "padding": padding,
                "scale": scale,
            },
        )
    )
    return self

optimize(*, quality: int = 85, output_format: OutputFormat | str = OutputFormat.WEBP, progressive: bool = True, optimize: bool = True, lossless: bool = False, strip_metadata: bool = True) -> Pipeline

Configure optimization parameters for the final save.

Source code in pixopt/pipeline.py

def optimize(
    self,
    *,
    quality: int = 85,
    output_format: OutputFormat | str = OutputFormat.WEBP,
    progressive: bool = True,
    optimize: bool = True,
    lossless: bool = False,
    strip_metadata: bool = True,
) -> Pipeline:
    """Configure optimization parameters for the final save."""
    if isinstance(output_format, str):
        try:
            output_format = OutputFormat(output_format)
        except ValueError as exc:
            raise ValueError(f"Invalid output_format: {output_format}") from exc
    self._ops.append(
        _Op(
            _OpKind.OPTIMIZE,
            {
                "quality": quality,
                "output_format": output_format,
                "progressive": progressive,
                "optimize": optimize,
                "lossless": lossless,
                "strip_metadata": strip_metadata,
            },
        )
    )
    return self

save(output: Path | str) -> Pipeline

Set the output path for the final image.

Source code in pixopt/pipeline.py

def save(self, output: Path | str) -> Pipeline:
    """Set the output path for the final image."""
    out = Path(output)
    if error := validate_no_parent_references(out, "output"):
        raise ValueError(error)
    self._ops.append(_Op(_OpKind.OPEN, {"output": out}))  # placeholder
    self._output = out
    return self

run() -> PipelineResult

Execute all queued operations and return a :class:PipelineResult.

Raises:

Type	Description
ValueError	If no source image was set.
FileNotFoundError	If the source image does not exist.

Source code in pixopt/pipeline.py

def run(self) -> PipelineResult:
    """Execute all queued operations and return a :class:`PipelineResult`.

    Raises:
        ValueError: If no source image was set.
        FileNotFoundError: If the source image does not exist.

    """
    if self._source is None:
        raise ValueError("No source image set. Call .open(path) first.")

    logger = get_logger("pipeline")
    logger.debug("Pipeline started", extra={"operation": "pipeline", "path": str(self._source)})

    output = Path(getattr(self, "_output", None) or self._source.with_suffix(".webp"))
    if error := validate_no_parent_references(output, "output"):
        raise ValueError(error)

    steps: list[str] = []
    img: Image.Image | None = None
    opt_params: dict[str, Any] = {
        "quality": 85,
        "output_format": OutputFormat.WEBP,
        "progressive": True,
        "optimize": True,
        "lossless": False,
        "strip_metadata": True,
    }

    for op in self._ops:
        if op.kind == _OpKind.OPEN:
            if "source" in op.params:
                with _open_image(op.params["source"], label="source") as opened:
                    opened.load()
                    img = opened
                steps.append(f"open({op.params['source'].name})")
            elif "output" in op.params:
                pass  # output path already captured

        elif op.kind == _OpKind.AUTO_ORIENT and img is not None:
            img = apply_exif_orientation(img)
            steps.append("auto_orient")

        elif op.kind == _OpKind.RESIZE and img is not None:
            fit_val = op.params.get("fit")
            if isinstance(fit_val, str):
                try:
                    fit_val = FitMode(fit_val)
                except ValueError as exc:
                    raise ValueError(f"Invalid fit value: {fit_val}") from exc
            anchor_val = op.params.get("anchor", Anchor.CENTER)
            if isinstance(anchor_val, str):
                try:
                    anchor_val = Anchor(anchor_val)
                except ValueError as exc:
                    raise ValueError(f"Invalid anchor value: {anchor_val}") from exc
            img = resize_image(
                img,
                max_width=op.params.get("max_width"),
                max_height=op.params.get("max_height"),
                fit=fit_val,
                anchor=anchor_val,
                aspect_ratio=op.params.get("aspect_ratio"),
                background_color=op.params.get("background_color", WHITE),
            )
            steps.append(f"resize({img.width}x{img.height})")

        elif op.kind == _OpKind.CONVERT and img is not None:
            img = img.convert(op.params["mode"])
            steps.append(f"convert({op.params['mode']})")

        elif op.kind == _OpKind.WATERMARK_TEXT and img is not None:
            fd, tmp_name = tempfile.mkstemp(
                prefix="pixopt_wm_", suffix=".png", dir=tempfile.gettempdir()
            )
            os.close(fd)
            tmp_path = Path(tmp_name)
            try:
                img.save(tmp_path, "PNG")
                wm_result = add_text_watermark(
                    tmp_path,
                    tmp_path,
                    op.params["text"],
                    position=op.params["position"],
                    opacity=op.params["opacity"],
                    padding=op.params["padding"],
                    font_size=op.params["font_size"],
                    font_path=op.params["font_path"],
                    color=op.params["color"],
                )
                with _open_image(wm_result.output_path, label="watermark result") as opened:
                    opened.load()
                    img.close()
                    img = opened
            finally:
                tmp_path.unlink(missing_ok=True)
            steps.append(f"watermark_text({op.params['text']!r})")

        elif op.kind == _OpKind.WATERMARK_IMAGE and img is not None:
            fd, tmp_name = tempfile.mkstemp(
                prefix="pixopt_wm_", suffix=".png", dir=tempfile.gettempdir()
            )
            os.close(fd)
            tmp_path = Path(tmp_name)
            try:
                img.save(tmp_path, "PNG")
                wm_result = add_image_watermark(
                    tmp_path,
                    tmp_path,
                    op.params["watermark_path"],
                    position=op.params["position"],
                    opacity=op.params["opacity"],
                    padding=op.params["padding"],
                    scale=op.params["scale"],
                )
                with _open_image(wm_result.output_path, label="watermark result") as opened:
                    opened.load()
                    img = opened
            finally:
                tmp_path.unlink(missing_ok=True)
            steps.append("watermark_image")

        elif op.kind == _OpKind.OPTIMIZE:
            opt_params.update(op.params)
            steps.append(f"optimize(quality={op.params['quality']})")

    if img is None:
        raise ValueError("Pipeline produced no image. Did you call .open()?")

    output.parent.mkdir(parents=True, exist_ok=True)

    out_fmt: OutputFormat = opt_params["output_format"]
    pillow_fmt = FORMAT_MAP.get(out_fmt, "WEBP")
    ext = FORMAT_TO_EXT.get(pillow_fmt, ".webp")

    if output.suffix == "":
        output = output.with_suffix(ext)

    try:
        img = convert_mode(img, pillow_fmt)
        save_kwargs = build_save_kwargs(
            pillow_fmt,
            quality=opt_params["quality"],
            progressive=opt_params["progressive"],
            optimize=opt_params["optimize"],
            lossless=opt_params["lossless"],
            strip_metadata=opt_params["strip_metadata"],
        )

        if opt_params["strip_metadata"]:
            img = strip_metadata_pillow(img, pillow_fmt)

        img.save(output, format=pillow_fmt, **save_kwargs)

        if opt_params["strip_metadata"]:
            strip_exif_post_process(output, pillow_fmt)

        size_bytes = output.stat().st_size

        return PipelineResult(
            output_path=output,
            width=img.width,
            height=img.height,
            format=pillow_fmt,
            size_bytes=size_bytes,
            steps_executed=steps,
        )
    finally:
        img.close()

PipelineResult(output_path: Path, width: int, height: int, format: str, size_bytes: int, steps_executed: list[str] = list()) dataclass

Result of a pipeline execution.

Attributes

output_path: Path instance-attribute

width: int instance-attribute

height: int instance-attribute

format: str instance-attribute

size_bytes: int instance-attribute

steps_executed: list[str] = field(default_factory=list) class-attribute instance-attribute

Methods:

to_dict() -> dict[str, Any]

Source code in pixopt/pipeline.py

def to_dict(self) -> dict[str, Any]:
    return {
        "output_path": str(self.output_path),
        "width": self.width,
        "height": self.height,
        "format": self.format,
        "size_bytes": self.size_bytes,
        "steps_executed": list(self.steps_executed),
    }

Image inspection

inspect_image(source: Path | str) -> ImageInfo

Inspect an image file and return structured metadata.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the image file to inspect.	required

Returns:

Name	Type	Description
An	ImageInfo	class:ImageInfo dataclass with all detected metadata.

Raises:

Type	Description
FileNotFoundError	If source does not exist.
UnidentifiedImageError	If the file is not a valid image.

Source code in pixopt/inspect.py

def inspect_image(source: Path | str) -> ImageInfo:
    """Inspect an image file and return structured metadata.

    Args:
        source: Path to the image file to inspect.

    Returns:
        An :class:`ImageInfo` dataclass with all detected metadata.

    Raises:
        FileNotFoundError: If *source* does not exist.
        UnidentifiedImageError: If the file is not a valid image.

    """
    path = Path(source)
    if not path.exists():
        raise FileNotFoundError(f"Image file not found: {path}")

    try:
        with _open_image(path, label="source") as img:
            img.load()

            # Basic properties.
            width, height = img.size
            mode = img.mode
            fmt = img.format or "UNKNOWN"
            file_size = path.stat().st_size

            # DPI (may be absent).
            dpi = img.info.get("dpi")
            if dpi is not None and isinstance(dpi, tuple) and len(dpi) == 2:
                dpi_tuple: tuple[float, float] | None = (float(dpi[0]), float(dpi[1]))
            else:
                dpi_tuple = None

            # Alpha channel detection.
            has_alpha = mode in ("RGBA", "LA") or (mode == "P" and "transparency" in img.info)

            # Animation / frame count.
            n_frames = 1
            with contextlib.suppress(AttributeError, OSError, ValueError):
                n_frames = getattr(img, "n_frames", 1)
            # Fallback: some Pillow versions need an explicit seek.
            if n_frames <= 1:
                try:
                    img.seek(1)
                    n_frames = getattr(img, "n_frames", 1)
                    img.seek(0)
                except (EOFError, AttributeError):
                    n_frames = 1
            is_animated = n_frames > 1
            frame_count = n_frames

            # EXIF data.
            exif_dict: dict[str, object] = {}
            orientation: int | None = None
            try:
                exif = img.getexif()
            except (AttributeError, OSError, ValueError, KeyError):
                exif = None

            if exif:
                for tag_id, value in exif.items():
                    tag_name = Base(tag_id).name if tag_id in _EXIF_TAG_IDS else f"Tag_{tag_id}"
                    # Convert bytes to a readable representation.
                    if isinstance(value, bytes):
                        value = value.hex()
                    exif_dict[tag_name] = value

                orientation = exif.get(_ORIENTATION_TAG)

            # ICC profile presence.
            icc_profile = "icc_profile" in img.info

            return ImageInfo(
                file_path=path,
                file_size=file_size,
                width=width,
                height=height,
                mode=mode,
                format=fmt,
                dpi=dpi_tuple,
                has_alpha=has_alpha,
                is_animated=is_animated,
                frame_count=frame_count,
                exif=exif_dict,
                icc_profile=icc_profile,
                orientation=orientation,
            )

    except UnidentifiedImageError:
        raise
    except FileNotFoundError:
        raise
    except (OSError, ValueError, Image.DecompressionBombError) as exc:
        msg = f"Error reading image {path}: {exc}"
        raise UnidentifiedImageError(msg) from exc

scan_directory(directory: Path | str, *, recursive: bool = False, extensions: Iterable[str] | None = None) -> ScanReport

Scan a directory and return structured image info plus aggregate stats.

Parameters:

Name	Type	Description	Default
directory	Path | str	Directory to scan.	required
recursive	bool	If True, scan subdirectories recursively.	False
extensions	Iterable[str] | None	Optional list of file extensions to include (e.g. [".jpg", ".png"]). Defaults to the built-in set of supported image extensions.	None

Returns:

Name	Type	Description
A	ScanReport	class:ScanReport with per-file entries and aggregate statistics.

Raises:

Type	Description
FileNotFoundError	If the directory does not exist.

Source code in pixopt/inventory.py

def scan_directory(
    directory: Path | str,
    *,
    recursive: bool = False,
    extensions: Iterable[str] | None = None,
) -> ScanReport:
    """Scan a directory and return structured image info plus aggregate stats.

    Args:
        directory: Directory to scan.
        recursive: If True, scan subdirectories recursively.
        extensions: Optional list of file extensions to include (e.g. ``[".jpg", ".png"]``).
            Defaults to the built-in set of supported image extensions.

    Returns:
        A :class:`ScanReport` with per-file entries and aggregate statistics.

    Raises:
        FileNotFoundError: If the directory does not exist.

    """
    dir_path = Path(directory)
    if error := validate_no_parent_references(dir_path, "directory"):
        raise ValueError(error)
    try:
        dir_path.stat()
    except FileNotFoundError as exc:
        raise FileNotFoundError(f"Directory not found: {dir_path}") from exc

    _logger.debug("Scanning directory", extra={"operation": "scan", "path": str(dir_path)})

    exts = set(extensions) if extensions is not None else set(DEFAULT_EXTENSIONS)

    entries: list[ScanEntry] = []
    total_size = 0
    formats: dict[str, int] = {}
    largest_file: Path | None = None
    largest_size = 0
    smallest_file: Path | None = None
    smallest_size = MAX_INPUT_BYTES + 1
    error_count = 0
    valid_count = 0

    for file_path in discover_images(dir_path, recursive=recursive, extensions=exts):
        if valid_count >= MAX_SCAN_ENTRIES:
            _logger.warning(
                "Scan stopped after reaching the entry limit",
                extra={"operation": "scan", "limit": MAX_SCAN_ENTRIES},
            )
            break

        file_size = file_path.stat().st_size

        try:
            with _open_image(file_path, label="source") as img:
                img.load()
                fmt = img.format or "UNKNOWN"
                mode = img.mode
                width = img.width
                height = img.height
                has_alpha = "A" in mode or "transparency" in img.info
                is_animated = getattr(img, "is_animated", False)
                frame_count = getattr(img, "n_frames", 1)

            entry = ScanEntry(
                file_path=file_path,
                file_size=file_size,
                width=width,
                height=height,
                format=fmt,
                mode=mode,
                has_alpha=has_alpha,
                is_animated=is_animated,
                frame_count=frame_count,
            )
            valid_count += 1
            total_size += file_size
            formats[fmt] = formats.get(fmt, 0) + 1

            if largest_file is None or file_size > largest_size:
                largest_file = file_path
                largest_size = file_size
            if smallest_file is None or file_size < smallest_size:
                smallest_file = file_path
                smallest_size = file_size

        except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError) as exc:
            entry = ScanEntry(
                file_path=file_path,
                file_size=file_size,
                width=0,
                height=0,
                format="UNKNOWN",
                mode="",
                has_alpha=False,
                is_animated=False,
                frame_count=0,
                error=str(exc),
            )
            error_count += 1

        entries.append(entry)

    if smallest_file is None:
        smallest_size = 0

    report = ScanReport(
        directory=dir_path,
        total_files=len(entries),
        valid_images=valid_count,
        errors=error_count,
        total_size_bytes=total_size,
        entries=entries,
        formats=formats,
        largest_file=largest_file,
        largest_size=largest_size,
        smallest_file=smallest_file,
        smallest_size=smallest_size,
    )

    _logger.info(
        "Scan complete",
        extra={"operation": "scan", "path": str(dir_path), "size_bytes": valid_count},
    )

    return report

Next-generation formats

convert_to_nextgen(source: Path | str, output: Path | str, *, fmt: NextGenFormat | str = NextGenFormat.JXL, quality: int = 85, fallback: bool = True, fallback_format: str = 'WEBP') -> ConversionResult

Convert an image to a next-generation format.

If the target format is not supported and fallback is True, converts to fallback_format instead.

Parameters:

Name	Type	Description	Default
source	Path | str	Path to the source image.	required
output	Path | str	Output path. Extension will be adjusted if fallback occurs.	required
fmt	NextGenFormat | str	Target next-gen format ("jxl" or "webp2").	JXL
quality	int	Quality for lossy compression (1-100).	85
fallback	bool	If True, fall back to fallback_format when the target format is not available.	True
fallback_format	str	Format to use when falling back (default WEBP).	'WEBP'

Returns:

Name	Type	Description
A	ConversionResult	class:ConversionResult with conversion details.

Raises:

Type	Description
FileNotFoundError	If the source image does not exist.
ValueError	If the format is not supported and fallback is disabled.

Source code in pixopt/nextgen.py

def convert_to_nextgen(
    source: Path | str,
    output: Path | str,
    *,
    fmt: NextGenFormat | str = NextGenFormat.JXL,
    quality: int = 85,
    fallback: bool = True,
    fallback_format: str = "WEBP",
) -> ConversionResult:
    """Convert an image to a next-generation format.

    If the target format is not supported and ``fallback`` is True,
    converts to ``fallback_format`` instead.

    Args:
        source: Path to the source image.
        output: Output path. Extension will be adjusted if fallback occurs.
        fmt: Target next-gen format (``"jxl"`` or ``"webp2"``).
        quality: Quality for lossy compression (1-100).
        fallback: If True, fall back to ``fallback_format`` when the
            target format is not available.
        fallback_format: Format to use when falling back (default WEBP).

    Returns:
        A :class:`ConversionResult` with conversion details.

    Raises:
        FileNotFoundError: If the source image does not exist.
        ValueError: If the format is not supported and fallback is disabled.

    """
    if isinstance(fmt, NextGenFormat):
        fmt = fmt.value

    source_path = Path(source)
    if error := validate_no_parent_references(source_path, "source"):
        raise ValueError(error)
    try:
        original_size = source_path.stat().st_size
    except FileNotFoundError as exc:
        raise FileNotFoundError(f"Source file not found: {source_path}") from exc

    output_path = Path(output)
    if error := validate_no_parent_references(output_path, "output"):
        raise ValueError(error)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    supported = is_format_supported(fmt)

    if not supported:
        if not fallback:
            return ConversionResult(
                source=source_path,
                output=output_path,
                format=fmt,
                supported=False,
                success=False,
                original_size=original_size,
                error=f"Format '{fmt}' is not supported and fallback is disabled",
            )

        # Fall back to a supported format
        _logger.info(
            "Format %s not supported, falling back to %s",
            fmt,
            fallback_format,
            extra={"operation": "nextgen_convert", "format": fmt, "fallback": fallback_format},
        )

        pillow_fmt = fallback_format.upper()
        output_path = output_path.with_suffix(f".{fallback_format.lower()}")
    else:
        pillow_fmt = fmt.upper()

    try:
        image: Image.Image
        with _open_image(source_path, label="source") as img:
            img.load()
            image = img
            if image.mode != "RGB":
                image = image.convert("RGB")

            save_kwargs: dict[str, Any] = {}
            if pillow_fmt in ("JPEG", "WEBP", "JXL"):
                save_kwargs["quality"] = quality

            image.save(output_path, format=pillow_fmt, **save_kwargs)

        output_size = output_path.stat().st_size
        savings = original_size - output_size
        savings_pct = (savings / original_size * 100.0) if original_size > 0 else 0.0

        result = ConversionResult(
            source=source_path,
            output=output_path,
            format=fmt if supported else fallback_format,
            supported=supported,
            success=True,
            original_size=original_size,
            output_size=output_size,
            savings_bytes=savings,
            savings_percent=savings_pct,
            fallback=not supported,
            fallback_format=fallback_format if not supported else None,
        )

        _logger.info(
            "Conversion complete",
            extra={
                "operation": "nextgen_convert",
                "format": result.format,
                "fallback": result.fallback,
                "size_bytes": output_size,
            },
        )

        return result

    except (OSError, ValueError, Image.DecompressionBombError) as exc:
        _logger.error(
            "Conversion failed",
            extra={"operation": "nextgen_convert", "format": fmt},
        )
        return ConversionResult(
            source=source_path,
            output=output_path,
            format=fmt,
            supported=supported,
            success=False,
            original_size=original_size,
            error=str(exc),
        )

Quality metrics

compare_images(original: Path | str, compared: Path | str) -> QualityMetrics

Compare two images and return quality metrics.

Parameters:

Name	Type	Description	Default
original	Path | str	Path to the original image.	required
compared	Path | str	Path to the image to compare against.	required

Returns:

Name	Type	Description
A	QualityMetrics	class:QualityMetrics with SSIM, PSNR, and MSE.

Raises:

Type	Description
FileNotFoundError	If either path does not exist.
ValueError	If images have different dimensions.

Source code in pixopt/quality.py

def compare_images(
    original: Path | str,
    compared: Path | str,
) -> QualityMetrics:
    """Compare two images and return quality metrics.

    Args:
        original: Path to the original image.
        compared: Path to the image to compare against.

    Returns:
        A :class:`QualityMetrics` with SSIM, PSNR, and MSE.

    Raises:
        FileNotFoundError: If either path does not exist.
        ValueError: If images have different dimensions.

    """
    orig_path = Path(original)
    comp_path = Path(compared)

    if not orig_path.exists():
        raise FileNotFoundError(f"Original image not found: {orig_path}")
    if not comp_path.exists():
        raise FileNotFoundError(f"Compared image not found: {comp_path}")

    arr1 = _load_as_array(orig_path)
    arr2 = _load_as_array(comp_path)

    if arr1.shape != arr2.shape:
        msg = (
            f"Image dimensions don't match: "
            f"{arr1.shape[1]}x{arr1.shape[0]} vs {arr2.shape[1]}x{arr2.shape[0]}"
        )
        raise ValueError(msg)

    mse = compute_mse(arr1, arr2)
    psnr = compute_psnr(mse)
    ssim = compute_ssim(arr1, arr2)

    return QualityMetrics(
        ssim=round(ssim, 4),
        psnr=round(psnr, 2) if psnr is not None else None,
        mse=round(mse, 4),
        original_path=orig_path,
        compared_path=comp_path,
        width=arr1.shape[1],
        height=arr1.shape[0],
    )

compute_ssim(img1: npt.NDArray[Any], img2: npt.NDArray[Any], *, win_size: int = 7, data_range: float = 255.0) -> float

Compute the Structural Similarity Index (SSIM) between two images.

Uses a sliding window approach with a Gaussian-free uniform window. Works on grayscale (2D) or multi-channel (3D) arrays.

Parameters:

Name	Type	Description	Default
img1	NDArray[Any]	First image as a numpy array.	required
img2	NDArray[Any]	Second image as a numpy array (same shape as img1).	required
win_size	int	Side length of the sliding window (must be odd).	7
data_range	float	Maximum pixel value (255 for 8-bit images).	255.0

Returns:

Type	Description
float	SSIM value in [-1, 1], where 1 means identical structure.

Source code in pixopt/quality.py

def compute_ssim(
    img1: npt.NDArray[Any],
    img2: npt.NDArray[Any],
    *,
    win_size: int = 7,
    data_range: float = 255.0,
) -> float:
    """Compute the Structural Similarity Index (SSIM) between two images.

    Uses a sliding window approach with a Gaussian-free uniform window.
    Works on grayscale (2D) or multi-channel (3D) arrays.

    Args:
        img1: First image as a numpy array.
        img2: Second image as a numpy array (same shape as img1).
        win_size: Side length of the sliding window (must be odd).
        data_range: Maximum pixel value (255 for 8-bit images).

    Returns:
        SSIM value in [-1, 1], where 1 means identical structure.
    """
    if img1.shape != img2.shape:
        msg = f"Image shapes don't match: {img1.shape} vs {img2.shape}"
        raise ValueError(msg)

    if img1.shape[0] * img1.shape[1] > MAX_SSIM_PIXELS:
        raise ValueError(
            f"SSIM pixel budget exceeded: {img1.shape[0] * img1.shape[1]} (max {MAX_SSIM_PIXELS})"
        )

    if np.array_equal(img1, img2):
        return 1.0

    c1 = (0.01 * data_range) ** 2
    c2 = (0.03 * data_range) ** 2

    # Convert to grayscale if multi-channel by averaging.
    if img1.ndim == 3:
        img1 = img1.mean(axis=2)
        img2 = img2.mean(axis=2)

    # Pad images to handle borders.
    pad = win_size // 2
    img1_p = np.pad(img1, pad, mode="reflect")
    img2_p = np.pad(img2, pad, mode="reflect")

    h, w = img1.shape
    ssim_sum = 0.0
    count = 0

    for i in range(h):
        for j in range(w):
            w1 = img1_p[i : i + win_size, j : j + win_size]
            w2 = img2_p[i : i + win_size, j : j + win_size]

            mu1 = w1.mean()
            mu2 = w2.mean()
            sigma1 = w1.var()
            sigma2 = w2.var()
            sigma12 = ((w1 - mu1) * (w2 - mu2)).mean()

            ssim_val = ((2 * mu1 * mu2 + c1) * (2 * sigma12 + c2)) / (
                (mu1**2 + mu2**2 + c1) * (sigma1 + sigma2 + c2)
            )
            ssim_sum += ssim_val
            count += 1

    return float(ssim_sum / count) if count > 0 else 1.0

compute_psnr(mse: float, max_pixel: float = 255.0) -> float | None

Compute PSNR from MSE.

Returns None if MSE is 0 (images are identical, PSNR = infinity).

Source code in pixopt/quality.py

def compute_psnr(mse: float, max_pixel: float = 255.0) -> float | None:
    """Compute PSNR from MSE.

    Returns None if MSE is 0 (images are identical, PSNR = infinity).
    """
    if mse == 0:
        return None
    return 10.0 * math.log10((max_pixel**2) / mse)