bin/pmtiles.retile.py

pmtiles.retile.py

#!/usr/bin/env python3
"""Retile a raster PMTiles archive and report the effect of deduplication.

The input tiles are expected to be twice the output tile size. Each source
tile is decoded and split into four child tiles, so a 512px z/x/y tile
becomes four 256px tiles at z+1.

Usage:
    pmtiles.retile.py INPUT.pmtiles OUTPUT.pmtiles
    pmtiles.retile.py INPUT.pmtiles OUTPUT.pmtiles --quality 90
"""

from __future__ import annotations

import argparse
from dataclasses import dataclass
from io import BytesIO
import gzip
import json
from pathlib import Path
import os
import struct
import sys
import tempfile
from typing import BinaryIO, Iterator


HEADER_SIZE = 127
MAX_ROOT_DIRECTORY_LENGTH = 16_384 - HEADER_SIZE
MAGIC = b"PMTiles"
SPEC_VERSION = 3

COMPRESSION_UNKNOWN = 0
COMPRESSION_NONE = 1
COMPRESSION_GZIP = 2

TILE_TYPE_NAMES = {
    2: ("PNG", "png"),
    3: ("JPEG", "jpeg"),
    4: ("WebP", "webp"),
}


@dataclass(frozen=True)
class Header:
    root_offset: int
    root_length: int
    metadata_offset: int
    metadata_length: int
    leaf_offset: int
    leaf_length: int
    tile_data_offset: int
    tile_data_length: int
    num_addressed_tiles: int
    num_tile_entries: int
    num_tile_contents: int
    clustered: bool
    internal_compression: int
    tile_compression: int
    tile_type: int
    min_zoom: int
    max_zoom: int
    min_lon: float
    min_lat: float
    max_lon: float
    max_lat: float
    center_zoom: int
    center_lon: float
    center_lat: float


@dataclass(frozen=True)
class Entry:
    tile_id: int
    offset: int
    length: int
    run_length: int


@dataclass(frozen=True)
class ArchiveStats:
    file_size: int
    tile_data_length: int
    referenced_tile_bytes: int
    addressed_tiles: int
    tile_entries: int
    tile_contents: int


def read_varint(data: bytes, position: int) -> tuple[int, int]:
    value = 0
    shift = 0
    while position < len(data):
        byte = data[position]
        position += 1
        value |= (byte & 0x7F) << shift
        if byte < 0x80:
            return value, position
        shift += 7
        if shift > 63:
            raise ValueError("PMTiles varint is too large")
    raise ValueError("truncated PMTiles varint")


def write_varint(value: int) -> bytes:
    if value < 0:
        raise ValueError(f"cannot encode negative varint: {value}")
    result = bytearray()
    while value >= 0x80:
        result.append((value & 0x7F) | 0x80)
        value >>= 7
    result.append(value)
    return bytes(result)


def compress_directory(data: bytes, compression: int) -> bytes:
    if compression in (COMPRESSION_UNKNOWN, COMPRESSION_NONE):
        return data
    if compression == COMPRESSION_GZIP:
        return gzip.compress(data, mtime=0)
    raise ValueError(
        f"unsupported PMTiles internal compression {compression}; "
        "only none and gzip are supported"
    )


def decompress_directory(data: bytes, compression: int) -> bytes:
    if compression in (COMPRESSION_UNKNOWN, COMPRESSION_NONE):
        return data
    if compression == COMPRESSION_GZIP:
        return gzip.decompress(data)
    raise ValueError(
        f"unsupported PMTiles internal compression {compression}; "
        "only none and gzip are supported"
    )


def read_header(file: BinaryIO) -> Header:
    raw = file.read(HEADER_SIZE)
    if len(raw) != HEADER_SIZE:
        raise ValueError("file is shorter than a PMTiles v3 header")
    if raw[:7] != MAGIC:
        raise ValueError("file does not start with the PMTiles magic number")
    if raw[7] != SPEC_VERSION:
        raise ValueError(f"only PMTiles v3 is supported, found v{raw[7]}")

    (
        root_offset,
        root_length,
        metadata_offset,
        metadata_length,
        leaf_offset,
        leaf_length,
        tile_data_offset,
        tile_data_length,
    ) = struct.unpack_from("<8Q", raw, 8)
    (
        num_addressed_tiles,
        num_tile_entries,
        num_tile_contents,
    ) = struct.unpack_from("<3Q", raw, 72)

    return Header(
        root_offset=root_offset,
        root_length=root_length,
        metadata_offset=metadata_offset,
        metadata_length=metadata_length,
        leaf_offset=leaf_offset,
        leaf_length=leaf_length,
        tile_data_offset=tile_data_offset,
        tile_data_length=tile_data_length,
        num_addressed_tiles=num_addressed_tiles,
        num_tile_entries=num_tile_entries,
        num_tile_contents=num_tile_contents,
        clustered=raw[96] == 1,
        internal_compression=raw[97],
        tile_compression=raw[98],
        tile_type=raw[99],
        min_zoom=raw[100],
        max_zoom=raw[101],
        min_lon=struct.unpack_from("<i", raw, 102)[0] / 10_000_000,
        min_lat=struct.unpack_from("<i", raw, 106)[0] / 10_000_000,
        max_lon=struct.unpack_from("<i", raw, 110)[0] / 10_000_000,
        max_lat=struct.unpack_from("<i", raw, 114)[0] / 10_000_000,
        center_zoom=raw[118],
        center_lon=struct.unpack_from("<i", raw, 119)[0] / 10_000_000,
        center_lat=struct.unpack_from("<i", raw, 123)[0] / 10_000_000,
    )


def read_at(file: BinaryIO, offset: int, length: int) -> bytes:
    file.seek(offset)
    data = file.read(length)
    if len(data) != length:
        raise ValueError(
            f"unexpected end of file at byte {offset}: "
            f"wanted {length} bytes, got {len(data)}"
        )
    return data


def decode_directory(data: bytes, compression: int) -> list[Entry]:
    data = decompress_directory(data, compression)
    position = 0
    count, position = read_varint(data, position)
    if count == 0:
        raise ValueError("PMTiles directories must not be empty")

    tile_ids: list[int] = []
    last_tile_id = 0
    for _ in range(count):
        delta, position = read_varint(data, position)
        last_tile_id += delta
        tile_ids.append(last_tile_id)

    run_lengths: list[int] = []
    for _ in range(count):
        run_length, position = read_varint(data, position)
        run_lengths.append(run_length)

    lengths: list[int] = []
    for _ in range(count):
        length, position = read_varint(data, position)
        if length == 0:
            raise ValueError("PMTiles directory entry has zero length")
        lengths.append(length)

    offsets: list[int] = []
    for index in range(count):
        encoded, position = read_varint(data, position)
        if encoded == 0 and index > 0:
            offset = offsets[index - 1] + lengths[index - 1]
        else:
            if encoded == 0:
                raise ValueError("first PMTiles directory offset cannot be zero")
            offset = encoded - 1
        offsets.append(offset)

    if position != len(data):
        raise ValueError(
            f"PMTiles directory has {len(data) - position} trailing bytes"
        )
    return [
        Entry(tile_id, offsets[i], lengths[i], run_lengths[i])
        for i, tile_id in enumerate(tile_ids)
    ]


def encode_directory(entries: list[Entry], compression: int) -> bytes:
    if not entries:
        raise ValueError("cannot encode an empty PMTiles directory")
    if any(entry.length <= 0 for entry in entries):
        raise ValueError("PMTiles directory entries must have positive lengths")

    raw = bytearray()
    raw.extend(write_varint(len(entries)))

    previous_tile_id = 0
    for entry in entries:
        if entry.tile_id < previous_tile_id:
            raise ValueError("PMTiles directory tile IDs must be sorted")
        raw.extend(write_varint(entry.tile_id - previous_tile_id))
        previous_tile_id = entry.tile_id

    for entry in entries:
        raw.extend(write_varint(entry.run_length))
    for entry in entries:
        raw.extend(write_varint(entry.length))

    for index, entry in enumerate(entries):
        if index > 0:
            previous = entries[index - 1]
            if entry.offset == previous.offset + previous.length:
                raw.extend(write_varint(0))
                continue
        raw.extend(write_varint(entry.offset + 1))

    return compress_directory(bytes(raw), compression)


def rotate(n: int, x: int, y: int, rx: int, ry: int) -> tuple[int, int]:
    if ry == 0:
        if rx != 0:
            return n - 1 - y, n - 1 - x
        return y, x
    return x, y


def zxy_to_tile_id(z: int, x: int, y: int) -> int:
    if z < 0 or x < 0 or y < 0 or x >= (1 << z) or y >= (1 << z):
        raise ValueError(f"invalid z/x/y tile coordinate: {z}/{x}/{y}")
    if z == 0:
        return 0

    tile_id = ((1 << z) * (1 << z) - 1) // 3
    tx, ty = x, y
    for bit in range(z - 1, -1, -1):
        scale = 1 << bit
        rx = tx & scale
        ry = ty & scale
        tile_id += ((3 * rx) ^ ry) * scale
        tx, ty = rotate(scale, tx, ty, rx, ry)
    return tile_id


def tile_id_to_zxy(tile_id: int) -> tuple[int, int, int]:
    if tile_id < 0:
        raise ValueError(f"invalid negative PMTiles tile ID: {tile_id}")
    z = ((3 * tile_id + 1).bit_length() - 1) // 2
    if z == 0:
        return 0, 0, 0

    base = ((1 << z) * (1 << z) - 1) // 3
    t = tile_id - base
    x = y = 0
    scale = 1
    while scale < (1 << z):
        rx = scale & (t // 2)
        ry = scale & (t ^ rx)
        x, y = rotate(scale, x, y, rx, ry)
        t //= 2
        x += rx
        y += ry
        scale <<= 1
    return z, x, y


def child_coordinates(z: int, x: int, y: int) -> list[tuple[int, int, int]]:
    children = [
        (z + 1, x * 2 + dx, y * 2 + dy)
        for dy in range(2)
        for dx in range(2)
    ]
    return sorted(children, key=lambda coordinate: zxy_to_tile_id(*coordinate))


class PMTilesReader:
    def __init__(self, path: Path):
        self.path = path
        self.file = path.open("rb")
        self.header = read_header(self.file)
        self.metadata = self._read_metadata()

    def __enter__(self) -> PMTilesReader:
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.file.close()

    def _read_metadata(self) -> dict:
        raw = read_at(
            self.file,
            self.header.metadata_offset,
            self.header.metadata_length,
        )
        raw = decompress_directory(raw, self.header.internal_compression)
        metadata = json.loads(raw)
        if not isinstance(metadata, dict):
            raise ValueError("PMTiles metadata must be a JSON object")
        return metadata

    def _read_directory(self, offset: int, length: int) -> list[Entry]:
        return decode_directory(
            read_at(self.file, offset, length),
            self.header.internal_compression,
        )

    def tile_entries(self) -> Iterator[Entry]:
        root = self._read_directory(
            self.header.root_offset,
            self.header.root_length,
        )
        for root_entry in root:
            if root_entry.run_length == 0:
                if self.header.leaf_length == 0:
                    raise ValueError("root references a missing leaf directory")
                leaf = self._read_directory(
                    self.header.leaf_offset + root_entry.offset,
                    root_entry.length,
                )
                for leaf_entry in leaf:
                    if leaf_entry.run_length == 0:
                        raise ValueError("nested PMTiles leaf directories are unsupported")
                    yield leaf_entry
            else:
                yield root_entry

    def read_tile(self, entry: Entry) -> bytes:
        raw = read_at(
            self.file,
            self.header.tile_data_offset + entry.offset,
            entry.length,
        )
        if self.header.tile_compression in (
            COMPRESSION_UNKNOWN,
            COMPRESSION_NONE,
        ):
            return raw
        if self.header.tile_compression == COMPRESSION_GZIP:
            return gzip.decompress(raw)
        raise ValueError(
            f"unsupported PMTiles tile compression "
            f"{self.header.tile_compression}; raster retile supports none and gzip"
        )

    def source_stats(self, entries: list[Entry]) -> ArchiveStats:
        addressed_tiles = sum(entry.run_length for entry in entries)
        referenced_tile_bytes = sum(entry.length * entry.run_length for entry in entries)
        return ArchiveStats(
            file_size=self.path.stat().st_size,
            tile_data_length=self.header.tile_data_length,
            referenced_tile_bytes=referenced_tile_bytes,
            addressed_tiles=addressed_tiles,
            tile_entries=sum(1 for entry in entries if entry.run_length > 0),
            tile_contents=self.header.num_tile_contents,
        )


def encode_image(
    image,
    image_format: str,
    quality: int | None,
    lossless: bool,
) -> bytes:
    if image_format == "JPEG" and image.mode not in ("L", "RGB", "CMYK"):
        image = image.convert("RGB")

    options = {}
    if quality is not None and image_format in ("JPEG", "WEBP"):
        options["quality"] = quality
    if image_format == "WEBP" and lossless:
        options["lossless"] = True
    if image_format == "PNG":
        options["optimize"] = True

    output = BytesIO()
    image.save(output, format=image_format, **options)
    return output.getvalue()


def make_output_tiles(
    reader: PMTilesReader,
    entries: list[Entry],
    input_tile_size: int,
    output_tile_size: int,
    quality: int | None,
    lossless: bool,
) -> tuple[list[tuple[int, bytes]], int]:
    if input_tile_size != output_tile_size * 2:
        raise ValueError("the input tile size must be exactly twice the output tile size")
    if reader.header.tile_type not in TILE_TYPE_NAMES:
        raise ValueError(
            f"PMTiles tile type {reader.header.tile_type} is not a supported "
            "raster type (PNG, JPEG, or WebP)"
        )
    if reader.header.max_zoom >= 255:
        raise ValueError("cannot increase a source max zoom of 255")

    try:
        from PIL import Image
    except ImportError as error:
        raise SystemExit("this script requires Pillow: python3 -m pip install Pillow") from error

    image_format, _ = TILE_TYPE_NAMES[reader.header.tile_type]
    output_tiles: list[tuple[int, bytes]] = []
    canonical_tiles: dict[bytes, bytes] = {}
    seen_tile_ids: set[int] = set()

    for entry in entries:
        source_data = reader.read_tile(entry)
        for tile_id in range(entry.tile_id, entry.tile_id + entry.run_length):
            z, x, y = tile_id_to_zxy(tile_id)
            with Image.open(BytesIO(source_data)) as image:
                image.load()
                if image.size != (input_tile_size, input_tile_size):
                    raise ValueError(
                        f"source tile {z}/{x}/{y} is {image.size[0]}x{image.size[1]}, "
                        f"expected {input_tile_size}x{input_tile_size}"
                    )
                for child_z, child_x, child_y in child_coordinates(z, x, y):
                    child_id = zxy_to_tile_id(child_z, child_x, child_y)
                    if child_id in seen_tile_ids:
                        raise ValueError(f"duplicate output tile ID {child_id}")
                    seen_tile_ids.add(child_id)

                    dx = child_x - x * 2
                    dy = child_y - y * 2
                    box = (
                        dx * output_tile_size,
                        dy * output_tile_size,
                        (dx + 1) * output_tile_size,
                        (dy + 1) * output_tile_size,
                    )
                    encoded = encode_image(
                        image.crop(box),
                        image_format,
                        quality,
                        lossless,
                    )
                    canonical = canonical_tiles.setdefault(encoded, encoded)
                    output_tiles.append((child_id, canonical))

    output_tiles.sort(key=lambda tile: tile[0])
    return output_tiles, sum(len(data) for _, data in output_tiles)


def deduplicate_tiles(
    output_tiles: list[tuple[int, bytes]],
) -> tuple[list[Entry], bytes, int]:
    entries: list[Entry] = []
    offsets: dict[bytes, tuple[int, int]] = {}
    tile_data = bytearray()

    previous_tile_id = None
    for tile_id, data in output_tiles:
        if previous_tile_id == tile_id:
            raise ValueError(f"duplicate output tile ID {tile_id}")
        previous_tile_id = tile_id

        location = offsets.get(data)
        if location is None:
            location = (len(tile_data), len(data))
            offsets[data] = location
            tile_data.extend(data)
        offset, length = location

        if entries:
            previous = entries[-1]
            if (
                previous.tile_id + previous.run_length == tile_id
                and previous.offset == offset
                and previous.length == length
            ):
                entries[-1] = Entry(
                    previous.tile_id,
                    previous.offset,
                    previous.length,
                    previous.run_length + 1,
                )
                continue
        entries.append(Entry(tile_id, offset, length, 1))

    return entries, bytes(tile_data), len(offsets)


def build_directories(
    tile_entries: list[Entry],
    internal_compression: int,
) -> tuple[bytes, bytes, int]:
    direct_root = encode_directory(tile_entries, internal_compression)
    if len(direct_root) <= MAX_ROOT_DIRECTORY_LENGTH:
        return direct_root, b"", 0

    chunk_size = 4096
    while True:
        leaves: list[bytes] = []
        root_entries: list[Entry] = []
        leaf_offset = 0
        for start in range(0, len(tile_entries), chunk_size):
            chunk = tile_entries[start : start + chunk_size]
            leaf = encode_directory(chunk, internal_compression)
            leaves.append(leaf)
            root_entries.append(Entry(chunk[0].tile_id, leaf_offset, len(leaf), 0))
            leaf_offset += len(leaf)

        root = encode_directory(root_entries, internal_compression)
        if len(root) <= MAX_ROOT_DIRECTORY_LENGTH:
            return root, b"".join(leaves), leaf_offset
        if chunk_size >= len(tile_entries):
            raise ValueError(
                "unable to fit the PMTiles root directory within the 16 KiB limit"
            )
        chunk_size = min(chunk_size * 2, len(tile_entries))


def build_header(
    *,
    root_length: int,
    metadata_length: int,
    leaf_length: int,
    tile_data_length: int,
    num_addressed_tiles: int,
    num_tile_entries: int,
    num_tile_contents: int,
    internal_compression: int,
    tile_type: int,
    min_zoom: int,
    max_zoom: int,
    min_lon: float,
    min_lat: float,
    max_lon: float,
    max_lat: float,
    center_zoom: int,
    center_lon: float,
    center_lat: float,
) -> bytes:
    root_offset = HEADER_SIZE
    metadata_offset = root_offset + root_length
    leaf_offset = metadata_offset + metadata_length if leaf_length else 0
    tile_data_offset = (
        (leaf_offset + leaf_length) if leaf_length else metadata_offset + metadata_length
    )

    header = bytearray(HEADER_SIZE)
    header[:7] = MAGIC
    header[7] = SPEC_VERSION
    struct.pack_into(
        "<8Q",
        header,
        8,
        root_offset,
        root_length,
        metadata_offset,
        metadata_length,
        leaf_offset,
        leaf_length,
        tile_data_offset,
        tile_data_length,
    )
    struct.pack_into(
        "<3Q",
        header,
        72,
        num_addressed_tiles,
        num_tile_entries,
        num_tile_contents,
    )
    header[96] = 1
    header[97] = internal_compression
    header[98] = COMPRESSION_NONE
    header[99] = tile_type
    header[100] = min_zoom
    header[101] = max_zoom
    struct.pack_into(
        "<4i",
        header,
        102,
        round(min_lon * 10_000_000),
        round(min_lat * 10_000_000),
        round(max_lon * 10_000_000),
        round(max_lat * 10_000_000),
    )
    header[118] = center_zoom
    struct.pack_into(
        "<2i",
        header,
        119,
        round(center_lon * 10_000_000),
        round(center_lat * 10_000_000),
    )
    return bytes(header)


def write_archive(
    output_path: Path,
    reader: PMTilesReader,
    metadata: dict,
    tile_entries: list[Entry],
    tile_data: bytes,
    num_addressed_tiles: int,
    num_tile_contents: int,
) -> ArchiveStats:
    root, leaves, leaf_length = build_directories(
        tile_entries,
        reader.header.internal_compression,
    )
    metadata_bytes = json.dumps(
        metadata,
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    header = build_header(
        root_length=len(root),
        metadata_length=len(metadata_bytes),
        leaf_length=leaf_length,
        tile_data_length=len(tile_data),
        num_addressed_tiles=num_addressed_tiles,
        num_tile_entries=len(tile_entries),
        num_tile_contents=num_tile_contents,
        internal_compression=reader.header.internal_compression,
        tile_type=reader.header.tile_type,
        min_zoom=reader.header.min_zoom + 1,
        max_zoom=reader.header.max_zoom + 1,
        min_lon=reader.header.min_lon,
        min_lat=reader.header.min_lat,
        max_lon=reader.header.max_lon,
        max_lat=reader.header.max_lat,
        center_zoom=min(reader.header.center_zoom + 1, reader.header.max_zoom + 1),
        center_lon=reader.header.center_lon,
        center_lat=reader.header.center_lat,
    )

    output_path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary_name = tempfile.mkstemp(
        prefix=f".{output_path.name}.",
        suffix=".tmp",
        dir=output_path.parent,
    )
    try:
        with os.fdopen(fd, "wb") as output:
            output.write(header)
            output.write(root)
            output.write(metadata_bytes)
            output.write(leaves)
            output.write(tile_data)
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary_name, output_path)
    except BaseException:
        try:
            os.unlink(temporary_name)
        except FileNotFoundError:
            pass
        raise

    return ArchiveStats(
        file_size=output_path.stat().st_size,
        tile_data_length=len(tile_data),
        referenced_tile_bytes=0,
        addressed_tiles=num_addressed_tiles,
        tile_entries=len(tile_entries),
        tile_contents=num_tile_contents,
    )


def format_bytes(value: int) -> str:
    units = ("B", "KiB", "MiB", "GiB")
    amount = float(value)
    for unit in units:
        if amount < 1024 or unit == units[-1]:
            return f"{amount:,.2f} {unit}"
        amount /= 1024
    raise AssertionError("unreachable")


def percentage_change(value: int, baseline: int) -> str:
    if baseline == 0:
        return "n/a"
    return f"{(value - baseline) / baseline * 100:+.2f}%"


def print_report(
    source: ArchiveStats,
    output: ArchiveStats,
    raw_output_tile_bytes: int,
    input_path: Path,
    output_path: Path,
) -> None:
    print(f"source:  {input_path} ({format_bytes(source.file_size)})")
    print(f"retiled: {output_path} ({format_bytes(output.file_size)})")
    print()
    print("                    source       retiled")
    print(f"addressed tiles     {source.addressed_tiles:>8,} {output.addressed_tiles:>13,}")
    print(f"tile entries        {source.tile_entries:>8,} {output.tile_entries:>13,}")
    print(f"unique tile blobs   {source.tile_contents:>8,} {output.tile_contents:>13,}")
    print(
        "tile data (dedup)   "
        f"{format_bytes(source.tile_data_length):>12} "
        f"{format_bytes(output.tile_data_length):>13}"
    )
    print(
        "tile refs (raw)     "
        f"{format_bytes(source.referenced_tile_bytes):>12} "
        f"{format_bytes(raw_output_tile_bytes):>13}"
    )
    print()
    print(
        "PMTiles dedup saving "
        f"(retiled raw -> retiled dedup): "
        f"{format_bytes(raw_output_tile_bytes - output.tile_data_length)} "
        f"({1 - output.tile_data_length / raw_output_tile_bytes:.2%})"
        if raw_output_tile_bytes
        else "PMTiles dedup saving: n/a"
    )
    print(
        "archive size change  "
        f"{percentage_change(output.file_size, source.file_size)} "
        f"({format_bytes(output.file_size - source.file_size) if output.file_size >= source.file_size else '-' + format_bytes(source.file_size - output.file_size)})"
    )
    print(
        "tile-data change     "
        f"{percentage_change(output.tile_data_length, source.tile_data_length)} "
        f"({format_bytes(output.tile_data_length - source.tile_data_length) if output.tile_data_length >= source.tile_data_length else '-' + format_bytes(source.tile_data_length - output.tile_data_length)})"
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("input", type=Path, help="512px raster PMTiles v3 archive")
    parser.add_argument("output", type=Path, help="path for the 256px PMTiles archive")
    parser.add_argument(
        "--input-tile-size",
        type=int,
        default=512,
        help="expected source tile size (default: 512)",
    )
    parser.add_argument(
        "--output-tile-size",
        type=int,
        default=256,
        help="output tile size (default: 256)",
    )
    parser.add_argument(
        "--quality",
        type=int,
        help="JPEG/WebP quality, 0-100 (Pillow's default is used when omitted)",
    )
    parser.add_argument(
        "--lossless",
        action="store_true",
        help="write lossless WebP output (cannot be used with JPEG)",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="replace an existing output archive",
    )
    args = parser.parse_args()
    if args.input_tile_size <= 0 or args.output_tile_size <= 0:
        parser.error("tile sizes must be positive")
    if args.quality is not None and not 0 <= args.quality <= 100:
        parser.error("--quality must be between 0 and 100")
    if args.lossless and args.quality is not None:
        parser.error("--lossless and --quality are mutually exclusive")
    if args.output.exists() and not args.force:
        parser.error(f"output already exists: {args.output} (use --force to replace)")
    return args


def main() -> None:
    args = parse_args()
    if not args.input.is_file():
        raise SystemExit(f"input does not exist or is not a file: {args.input}")

    with PMTilesReader(args.input) as reader:
        entries = list(reader.tile_entries())
        if not entries:
            raise SystemExit("input PMTiles archive contains no tile entries")
        source_stats = reader.source_stats(entries)
        output_tiles, raw_output_tile_bytes = make_output_tiles(
            reader,
            entries,
            args.input_tile_size,
            args.output_tile_size,
            args.quality,
            args.lossless,
        )
        output_entries, tile_data, num_tile_contents = deduplicate_tiles(output_tiles)

        metadata = dict(reader.metadata)
        metadata["minzoom"] = reader.header.min_zoom + 1
        metadata["maxzoom"] = reader.header.max_zoom + 1
        metadata["tile_size"] = args.output_tile_size

        output_stats = write_archive(
            args.output,
            reader,
            metadata,
            output_entries,
            tile_data,
            len(output_tiles),
            num_tile_contents,
        )

    print_report(
        source_stats,
        output_stats,
        raw_output_tile_bytes,
        args.input,
        args.output,
    )


if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, json.JSONDecodeError) as error:
        raise SystemExit(f"error: {error}") from error