Add disk_to_pmtiles: Convert a directory of tiles to pmtiles

python/bin/pmtiles-convert

@@ -5,19 +5,28 @@ import argparse
 import os
 import shutil
 
-from pmtiles.convert import mbtiles_to_pmtiles, pmtiles_to_mbtiles, pmtiles_to_dir
+from pmtiles.convert import mbtiles_to_pmtiles, pmtiles_to_mbtiles, pmtiles_to_dir, disk_to_pmtiles
 
 parser = argparse.ArgumentParser(
     description="Convert between PMTiles and other archive formats."
 )
-parser.add_argument("input", help="Input .mbtiles or .pmtiles")
+parser.add_argument("input", help="Input .mbtiles, .pmtiles, or directory")
 parser.add_argument("output", help="Output .mbtiles, .pmtiles, or directory")
 parser.add_argument(
-    "--maxzoom", help="the maximum zoom level to include in the output."
+    "--maxzoom", help="The maximum zoom level to include in the output. Set to 'auto' when converting from directory to use the highest zoom."
 )
 parser.add_argument(
     "--overwrite", help="Overwrite the existing output.", action="store_true"
 )
+parser.add_argument(
+    "--scheme", help="Tiling scheme of the input directory ('ags', 'gwc', 'zyx', 'zxy' (default))."
+)
+parser.add_argument(
+    "--format", help="Image format of tiles in the input directory, if not provided in the metadata ('png', 'jpeg', 'pbf', 'webp', 'avif').", dest="tile_format"
+)
+parser.add_argument(
+    "--verbose", help="Print progress when converting a directory to .pmtiles.", action="store_true"
+)
 args = parser.parse_args()
 
 if os.path.exists(args.output) and not args.overwrite:
@@ -39,5 +48,8 @@ elif args.input.endswith(".pmtiles") and args.output.endswith(".mbtiles"):
 elif args.input.endswith(".pmtiles"):
     pmtiles_to_dir(args.input, args.output)
 
+elif args.output.endswith(".pmtiles"):
+    disk_to_pmtiles(args.input, args.output, args.maxzoom, scheme=args.scheme, tile_format=args.tile_format, verbose=args.verbose)
+
 else:
     print("Conversion not implemented")

python/pmtiles/convert.py

@@ -178,3 +178,167 @@ def pmtiles_to_dir(input, output):
             os.makedirs(directory, exist_ok=True)
             with open(path, "wb") as f:
                 f.write(tile_data)
+
+
+def disk_to_pmtiles(directory_path, output, maxzoom, **kwargs):
+    """Convert a directory of tiles on disk to PMTiles.
+
+    Requires metadata.json in the root of the directory.
+
+    Tiling scheme of the directory is assumed to be zxy unless specified.
+
+    Args:
+        directory_path (str): Root directory of tiles.
+        output (str): Path of PMTiles to be written.
+        maxzoom (int, "auto"): Max zoom level to use. If "auto", uses highest zoom in directory.
+
+    Keyword args:
+        scheme (str): Tiling scheme of the directory ('ags', 'gwc', 'zyx', 'zxy' (default)).
+        tile_format (str): Image format of the tiles ('png', 'jpeg', 'pbf', 'webp', 'avif') if not given in the metadata.
+        verbose (bool): Set True to print progress.
+
+    Uses modified elements of 'disk_to_mbtiles' from mbutil
+
+    Copyright (c), Development Seed
+    All rights reserved.
+
+    Licensed under BSD 3-Clause
+    """
+    verbose = kwargs.get("verbose")
+    try:
+        metadata = json.load(open(os.path.join(directory_path, 'metadata.json'), 'r'))
+    except IOError:
+        raise Exception("metadata.json not found in directory")
+
+    tile_format = kwargs.get('tile_format', metadata.get("format"))
+    if not tile_format:
+        raise Exception("tile format not found in metadata.json nor specified as keyword argument")
+    metadata["format"] = tile_format  # Add 'format' to metadata
+
+    scheme = kwargs.get('scheme')
+
+    # Collect a set of all tile IDs
+    z_set = []  # List of all zoom levels for auto-detecting maxzoom.
+    tileid_path_set = []  # List of tile (id, filepath) pairs
+    zoom_dirs = get_dirs(directory_path)
+    zoom_dirs.sort(key=len)
+    try:
+        collect_max = int(maxzoom)
+    except ValueError:
+        collect_max = 99
+    collect_min = metadata.get("minzoom", 0)
+    count = 0
+    warned = False
+    for zoom_dir in zoom_dirs:
+        if scheme == 'ags':
+            z = int(zoom_dir.replace("L", ""))
+        elif scheme == 'gwc':
+            z=int(zoom_dir[-2:])
+        else:
+            z = int(zoom_dir)
+        if not collect_min <= z <= collect_max:
+            continue
+        z_set.append(z)
+        if z > 9 and not warned:
+            print(" Warning: Large tilesets (z > 9) require extreme processing times.")
+            warned = True
+        if verbose:
+            print(" Searching for tiles at z=%s ..." % (z), end="", flush=True)
+        count = 0
+        for row_dir in get_dirs(os.path.join(directory_path, zoom_dir)):
+            if scheme == 'ags':
+                y = flip_y(z, int(row_dir.replace("R", ""), 16))
+            elif scheme == 'gwc':
+                pass
+            elif scheme == 'zyx':
+                y = flip_y(int(z), int(row_dir))
+            else:
+                x = int(row_dir)
+            for current_file in os.listdir(os.path.join(directory_path, zoom_dir, row_dir)):
+                if current_file == ".DS_Store":
+                    pass
+                else:
+                    file_name, _ = current_file.split('.',1)
+                    if scheme == 'xyz':
+                        y = flip_y(int(z), int(file_name))
+                    elif scheme == 'ags':
+                        x = int(file_name.replace("C", ""), 16)
+                    elif scheme == 'gwc':
+                        x, y = file_name.split('_')
+                        x = int(x)
+                        y = int(y)
+                    elif scheme == 'zyx':
+                        x = int(file_name)
+                    else:
+                        y = int(file_name)
+
+                    flipped = (1 << z) - 1 - y
+                    tileid = zxy_to_tileid(z, x, flipped)
+                    filepath = os.path.join(directory_path, zoom_dir, row_dir, current_file)
+                    tileid_path_set.append((tileid, filepath))
+                    count = count + 1
+        if verbose:
+            print(" found %s" % (count))
+
+    n_tiles = len(tileid_path_set)
+    if verbose:
+        print(" Sorting list of %s tile IDs ..." % (n_tiles), end="")
+    tileid_path_set.sort(key=lambda x: x[0])  # Sort by tileid
+    if verbose:
+        print(" done.")
+
+    maxzoom = max(z_set) if maxzoom == "auto" else int(maxzoom)
+    metadata["maxzoom"] = maxzoom
+
+    if not metadata.get("minzoom"):
+        metadata["minzoom"] = min(z_set)
+
+    is_pbf = tile_format == "pbf"
+
+    with write(output) as writer:
+
+        # read tiles in ascending tile order
+        count = 0
+        if verbose:
+            count_step = (2**(maxzoom-3))**2 if maxzoom <= 9 else (2**(9-3))**2
+            print(" Begin writing %s to .pmtiles ..." % (n_tiles), flush=True)
+        for tileid, filepath in tileid_path_set:
+            f = open(filepath, 'rb')
+            data = f.read()
+            # force gzip compression only for vector
+            if is_pbf and data[0:2] != b"\x1f\x8b":
+                data = gzip.compress(data)
+            writer.write_tile(tileid, data)
+            count = count + 1
+            if verbose and (count % count_step) == 0:
+                print(" %s tiles inserted of %s" % (count, n_tiles), flush=True)
+
+        if verbose and (count % count_step) != 0:
+            print(" %s tiles inserted of %s" % (count, n_tiles))
+
+        pmtiles_header, pmtiles_metadata = mbtiles_to_header_json(metadata)
+        pmtiles_header["max_zoom"] = maxzoom
+        result = writer.finalize(pmtiles_header, pmtiles_metadata)
+
+
+def get_dirs(path):
+    """'get_dirs' from mbutil
+
+    Copyright (c), Development Seed
+    All rights reserved
+
+    Licensed under BSD 3-Clause
+    """
+    return [name for name in os.listdir(path)
+        if os.path.isdir(os.path.join(path, name))]
+
+
+def flip_y(zoom, y):
+    """'flip_y' from mbutil
+
+    Copyright (c), Development Seed
+    All rights reserved
+
+    Licensed under BSD 3-Clause
+    """
+    return (2**zoom-1) - y

python/setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="pmtiles",
-    version="3.3.0",
+    version="3.4.0",
     author="Brandon Liu",
     author_email="[email protected]",
     description="Library and utilities to write and read PMTiles archives - cloud-optimized archives of map tiles.",

python/test/test_convert.py

@@ -11,6 +11,7 @@
     pmtiles_to_dir,
     mbtiles_to_pmtiles,
     mbtiles_to_header_json,
+    disk_to_pmtiles
 )
 from pmtiles.tile import TileType, Compression
 
@@ -33,6 +34,10 @@ def tearDown(self):
             shutil.rmtree("test_dir")
         except:
             pass
+        try:
+            os.remove("test_tmp_from_dir.pmtiles")
+        except:
+            pass
 
     def test_roundtrip(self):
         with open("test_tmp.pmtiles", "wb") as f:
@@ -45,22 +50,41 @@ def test_roundtrip(self):
             writer.write_tile(5, b"5")
             writer.write_tile(6, b"6")
             writer.write_tile(7, b"7")
+
+            header = {
+                "tile_type": TileType.MVT,
+                "tile_compression": Compression.GZIP,
+                "min_zoom": 0,
+                "max_zoom": 2,
+                "min_lon_e7": 0,
+                "max_lon_e7": 0,
+                "min_lat_e7": 0,
+                "max_lat_e7": 0,
+                "center_zoom": 0,
+                "center_lon_e7": 0,
+                "center_lat_e7": 0,
+                }
+
+            metadata = {
+                "vector_layers": ['vector','layers'],
+                "tilestats":{'tile':'stats'},
+                }
+            metadata["minzoom"] = header["min_zoom"]
+            metadata["maxzoom"] = header["max_zoom"]
+            min_lon = header["min_lon_e7"] / 10000000
+            min_lat = header["min_lat_e7"] / 10000000
+            max_lon = header["max_lon_e7"] / 10000000
+            max_lat = header["max_lat_e7"] / 10000000
+            metadata["bounds"] = f"{min_lon},{min_lat},{max_lon},{max_lat}"
+            center_lon = header["center_lon_e7"] / 10000000
+            center_lat = header["center_lat_e7"] / 10000000
+            center_zoom = header["center_zoom"]
+            metadata["center"] = f"{center_lon},{center_lat},{center_zoom}"
+            metadata["format"] = "pbf"
+
             writer.finalize(
-                {
-                    "tile_type": TileType.MVT,
-                    "tile_compression": Compression.GZIP,
-                    "min_zoom": 0,
-                    "max_zoom": 2,
-                    "min_lon_e7": 0,
-                    "max_lon_e7": 0,
-                    "min_lat_e7": 0,
-                    "max_lat_e7": 0,
-                    "center_zoom": 0,
-                    "center_lon_e7": 0,
-                    "center_lat_e7": 0,
-                },
-                {"vector_layers": ['vector','layers'],
-                "tilestats":{'tile':'stats'}},
+                header,
+                metadata,
             )
 
         pmtiles_to_mbtiles("test_tmp.pmtiles", "test_tmp.mbtiles")
@@ -74,7 +98,9 @@ def test_roundtrip(self):
 
         mbtiles_to_pmtiles("test_tmp.mbtiles", "test_tmp_2.pmtiles", 3)
 
-        pmtiles_to_dir("test_tmp.pmtiles","test_dir")
+        pmtiles_to_dir("test_tmp.pmtiles", "test_dir")
+
+        disk_to_pmtiles("test_dir", "test_tmp_from_dir.pmtiles", maxzoom="auto", tile_format="pbz")
 
     def test_mbtiles_header(self):
         header, json_metadata = mbtiles_to_header_json(