Merge a directory of PMTiles files into a single file

merge-pmtiles.py

#!/usr/bin/env python3

import argparse
import os
from pmtiles.reader import MmapSource, Reader, all_tiles
from pmtiles.writer import Writer
from pmtiles.tile import Compression
from pmtiles.tile import zxy_to_tileid
from tqdm import tqdm
def merge_pmtiles(input_dir: str, output_file: str) -> None:
    if not os.path.exists(input_dir):
        print(f"Error: Input directory '{input_dir}' does not exist.")
        exit(1)

    input_files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f)) and f.endswith('.pmtiles')]

    if not input_files:
        print(f"No PMTiles files found in '{input_dir}'.")
        exit(1)

    output_pmtiles = open(output_file, "wb")
    writer = Writer(output_pmtiles)

    tile_type = None
    bounds : list[tuple[int, int, int, int]] = []
    for input_file in tqdm(input_files, "Merging PMTiles", unit=" files"):
        input_path = os.path.join(input_dir, input_file)
        with open(input_path, 'rb') as f:
            source = MmapSource(f)
            reader = Reader(source)
            metadata = reader.header()
            if tile_type is None:
                tile_type = metadata['tile_type']
            bounds.append((
                metadata['min_lon_e7'],
                metadata['min_lat_e7'],
                metadata['max_lon_e7'],
                metadata['max_lat_e7']
            ))
            for zxy, tile_data in all_tiles(source):
                z, x, y = zxy
                writer.write_tile(zxy_to_tileid(z, x, y), tile_data)

    merged_bounds = (
        min([b[0] for b in bounds]),
        min([b[1] for b in bounds]),
        max([b[2] for b in bounds]),
        max([b[3] for b in bounds]),
    )
    writer.finalize(
        header={
            "tile_compression": Compression.NONE,
            "tile_type": tile_type,
            "min_lon_e7": merged_bounds[0],
            "min_lat_e7": merged_bounds[1],
            "max_lon_e7": merged_bounds[2],
            "max_lat_e7": merged_bounds[3],
        },
        metadata={},
    )

    output_pmtiles.close()

def main():
    parser = argparse.ArgumentParser(description="Merge PMTiles files into a single output file.")
    parser.add_argument("input_dir", type=str, help="Path to the input directory containing PMTiles files.")
    parser.add_argument("output_file", type=str, help="Path to the output file.")
    args = parser.parse_args()
    merge_pmtiles(args.input_dir, args.output_file)

if __name__ == "__main__":
    main()