Merge pull request #39 from mapcraftlabs/use-black

Use black

python/bin/pmtiles-convert

@@ -1,18 +1,26 @@
 #!/usr/bin/env python
 
-#pmtiles to files
+# pmtiles to files
 import argparse
 import os
 import shutil
 
 from pmtiles.convert import mbtiles_to_pmtiles, pmtiles_to_mbtiles, pmtiles_to_dir
 
-parser = argparse.ArgumentParser(description='Convert between PMTiles and other archive formats.')
-parser.add_argument('input',help='Input .mbtiles or .pmtiles')
-parser.add_argument('output',help='Output .mbtiles, .pmtiles, or directory')
-parser.add_argument('--maxzoom', help='the maximum zoom level to include in the output.')
-parser.add_argument('--gzip', help='The output should be gzip-compressed.',action='store_true')
-parser.add_argument('--overwrite', help='Overwrite the existing output.',action='store_true')
+parser = argparse.ArgumentParser(
+    description="Convert between PMTiles and other archive formats."
+)
+parser.add_argument("input", help="Input .mbtiles or .pmtiles")
+parser.add_argument("output", help="Output .mbtiles, .pmtiles, or directory")
+parser.add_argument(
+    "--maxzoom", help="the maximum zoom level to include in the output."
+)
+parser.add_argument(
+    "--gzip", help="The output should be gzip-compressed.", action="store_true"
+)
+parser.add_argument(
+    "--overwrite", help="Overwrite the existing output.", action="store_true"
+)
 args = parser.parse_args()
 
 if os.path.exists(args.output) and not args.overwrite:
@@ -26,10 +34,10 @@ if args.overwrite:
 
 print("compression:", "gzip" if args.gzip else "disabled")
 
-if args.input.endswith('.mbtiles') and args.output.endswith('.pmtiles'):
+if args.input.endswith(".mbtiles") and args.output.endswith(".pmtiles"):
     mbtiles_to_pmtiles(args.input, args.output, args.maxzoom, args.gzip)
 
-elif args.input.endswith('.pmtiles') and args.output.endswith('.mbtiles'):
+elif args.input.endswith(".pmtiles") and args.output.endswith(".mbtiles"):
     pmtiles_to_mbtiles(args.input, args.output, args.gzip)
 
 elif args.input.endswith(".pmtiles"):

python/bin/pmtiles-serve

@@ -13,51 +13,58 @@ from pmtiles.reader import read
 class ThreadingSimpleServer(ThreadingMixIn, http.server.HTTPServer):
     pass
 
-parser = argparse.ArgumentParser(description='Convert between PMTiles and other archive formats.')
-parser.add_argument('pmtiles_file',help='PMTiles archive to serve')
-parser.add_argument('port',help='Port to bind to')
-parser.add_argument('--bind', help='Address to bind server to: default localhost')
-parser.add_argument('--cors-allow-all', help='Return Access-Control-Allow-Origin:* header',action='store_true')
+
+parser = argparse.ArgumentParser(
+    description="Convert between PMTiles and other archive formats."
+)
+parser.add_argument("pmtiles_file", help="PMTiles archive to serve")
+parser.add_argument("port", help="Port to bind to")
+parser.add_argument("--bind", help="Address to bind server to: default localhost")
+parser.add_argument(
+    "--cors-allow-all",
+    help="Return Access-Control-Allow-Origin:* header",
+    action="store_true",
+)
 args = parser.parse_args()
 
 with read(args.pmtiles_file) as reader:
-    fmt = reader.metadata['format']
+    fmt = reader.metadata["format"]
 
     class Handler(http.server.SimpleHTTPRequestHandler):
         def do_GET(self):
             if self.path == "/metadata":
                 self.send_response(200)
                 if args.cors_allow_all:
-                    self.send_header('Access-Control-Allow-Origin','*')
+                    self.send_header("Access-Control-Allow-Origin", "*")
                 self.end_headers()
-                self.wfile.write(json.dumps(reader.metadata).encode('utf-8'))
+                self.wfile.write(json.dumps(reader.metadata).encode("utf-8"))
                 return
             match = re.match("/(\d+)/(\d+)/(\d+)." + fmt, self.path)
             if not match:
                 self.send_response(400)
                 self.end_headers()
-                self.wfile.write("bad request".encode('utf-8'))
+                self.wfile.write("bad request".encode("utf-8"))
                 return
             z = int(match.group(1))
             x = int(match.group(2))
             y = int(match.group(3))
-            data = reader.get(z,x,y)
+            data = reader.get(z, x, y)
             if not data:
                 self.send_response(404)
                 self.end_headers()
-                self.wfile.write("tile not found".encode('utf-8'))
+                self.wfile.write("tile not found".encode("utf-8"))
                 return
             self.send_response(200)
             if args.cors_allow_all:
-                self.send_header('Access-Control-Allow-Origin','*')
-            if fmt == 'pbf':
-                self.send_header('Content-Type','application/x-protobuf')
+                self.send_header("Access-Control-Allow-Origin", "*")
+            if fmt == "pbf":
+                self.send_header("Content-Type", "application/x-protobuf")
             else:
-                self.send_header('Content-Type','image/' + fmt)
+                self.send_header("Content-Type", "image/" + fmt)
             self.end_headers()
             self.wfile.write(data)
 
-    bind = args.bind or 'localhost'
-    print(f'serving {bind}:{args.port}/{{z}}/{{x}}/{{y}}.{fmt}, for development only')
-    httpd = ThreadingSimpleServer((args.bind or '', int(args.port)), Handler)
+    bind = args.bind or "localhost"
+    print(f"serving {bind}:{args.port}/{{z}}/{{x}}/{{y}}.{fmt}, for development only")
+    httpd = ThreadingSimpleServer((args.bind or "", int(args.port)), Handler)
     httpd.serve_forever()

python/bin/pmtiles-show

@@ -11,21 +11,21 @@ if len(sys.argv) <= 1:
 
 with read(sys.argv[1]) as reader:
     if len(sys.argv) == 2:
-        print('spec version: ',reader.version)
-        print('metadata:')
+        print("spec version: ", reader.version)
+        print("metadata:")
         for k, v in reader.metadata.items():
-            print(k,'=',v)
-        print('root entries:', reader.root_entries)
-        print('leaf directories:', len(set(reader.leaves.values())))
+            print(k, "=", v)
+        print("root entries:", reader.root_entries)
+        print("leaf directories:", len(set(reader.leaves.values())))
     elif len(sys.argv) == 3:
-        for k,v in reader.root_dir.items():
+        for k, v in reader.root_dir.items():
             print(f"{k[0]} {k[1]} {k[2]} {v[0]} {v[1]}")
         for val in set(reader.leaves.values()):
-            leaf_dir, _ = reader.load_directory(val[0],val[1]//17)
-            for k,v in leaf_dir.items():
+            leaf_dir, _ = reader.load_directory(val[0], val[1] // 17)
+            for k, v in leaf_dir.items():
                 print(f"{k[0]} {k[1]} {k[2]} {v[0]} {v[1]}")
     else:
         z = int(sys.argv[2])
         x = int(sys.argv[3])
         y = int(sys.argv[4])
-        print(reader.get(z,x,y))
+        print(reader.get(z, x, y))

python/pmtiles/__init__.py

@@ -1,3 +1,3 @@
 from collections import namedtuple
 
-Entry = namedtuple('Entry',['z','x','y','offset','length','is_dir'])
+Entry = namedtuple("Entry", ["z", "x", "y", "offset", "length", "is_dir"])

python/pmtiles/convert.py

@@ -1,5 +1,4 @@
-
-#pmtiles to files
+# pmtiles to files
 import gzip
 import json
 import os
@@ -9,42 +8,47 @@ from pmtiles.writer import write
 
 # if the tile is GZIP-encoded, it won't work with range queries
 # until transfer-encoding: gzip is well supported.
-def force_compress(data,compress):
-    if compress and data[0:2] !=  b'\x1f\x8b':
+def force_compress(data, compress):
+    if compress and data[0:2] != b"\x1f\x8b":
         return gzip.compress(data)
-    if not compress and data[0:2] == b'\x1f\x8b':
+    if not compress and data[0:2] == b"\x1f\x8b":
         return gzip.decompress(data)
     return data
 
-def set_metadata_compression(metadata,gzip):
+
+def set_metadata_compression(metadata, gzip):
     if gzip:
-        metadata['compression'] = 'gzip'
+        metadata["compression"] = "gzip"
     else:
         try:
-            del metadata['compression']
+            del metadata["compression"]
         except:
             pass
     return metadata
 
+
 def mbtiles_to_pmtiles(input, output, maxzoom, gzip):
     conn = sqlite3.connect(input)
     cursor = conn.cursor()
 
     with write(output) as writer:
-        for row in cursor.execute('SELECT zoom_level,tile_column,tile_row,tile_data FROM tiles WHERE zoom_level <= ? ORDER BY zoom_level,tile_column,tile_row ASC',(maxzoom or 99,)):
+        for row in cursor.execute(
+            "SELECT zoom_level,tile_column,tile_row,tile_data FROM tiles WHERE zoom_level <= ? ORDER BY zoom_level,tile_column,tile_row ASC",
+            (maxzoom or 99,),
+        ):
             flipped = (1 << row[0]) - 1 - row[2]
-            writer.write_tile(row[0],row[1],flipped,force_compress(row[3],gzip))
+            writer.write_tile(row[0], row[1], flipped, force_compress(row[3], gzip))
 
         metadata = {}
-        for row in cursor.execute('SELECT name,value FROM metadata'):
+        for row in cursor.execute("SELECT name,value FROM metadata"):
             metadata[row[0]] = row[1]
         if maxzoom:
-            metadata['maxzoom'] = str(maxzoom)
-        metadata = set_metadata_compression(metadata,gzip)
+            metadata["maxzoom"] = str(maxzoom)
+        metadata = set_metadata_compression(metadata, gzip)
         result = writer.finalize(metadata)
-        print("Num tiles:",result['num_tiles'])
-        print("Num unique tiles:",result['num_unique_tiles'])
-        print("Num leaves:",result['num_leaves'])
+        print("Num tiles:", result["num_tiles"])
+        print("Num unique tiles:", result["num_unique_tiles"])
+        print("Num leaves:", result["num_leaves"])
 
     conn.close()
 
@@ -52,34 +56,42 @@ def mbtiles_to_pmtiles(input, output, maxzoom, gzip):
 def pmtiles_to_mbtiles(input, output, gzip):
     conn = sqlite3.connect(output)
     cursor = conn.cursor()
-    cursor.execute('CREATE TABLE metadata (name text, value text);')
-    cursor.execute('CREATE TABLE tiles (zoom_level integer, tile_column integer, tile_row integer, tile_data blob);')
+    cursor.execute("CREATE TABLE metadata (name text, value text);")
+    cursor.execute(
+        "CREATE TABLE tiles (zoom_level integer, tile_column integer, tile_row integer, tile_data blob);"
+    )
 
     with read(input) as reader:
         metadata = reader.metadata
-        metadata = set_metadata_compression(metadata,gzip)
-        for k,v in metadata.items():
-            cursor.execute('INSERT INTO metadata VALUES(?,?)',(k,v))
+        metadata = set_metadata_compression(metadata, gzip)
+        for k, v in metadata.items():
+            cursor.execute("INSERT INTO metadata VALUES(?,?)", (k, v))
         for tile, data in reader.tiles():
             flipped = (1 << tile[0]) - 1 - tile[2]
-            cursor.execute('INSERT INTO tiles VALUES(?,?,?,?)',(tile[0],tile[1],flipped,force_compress(data,gzip)))
+            cursor.execute(
+                "INSERT INTO tiles VALUES(?,?,?,?)",
+                (tile[0], tile[1], flipped, force_compress(data, gzip)),
+            )
 
-    cursor.execute('CREATE UNIQUE INDEX tile_index on tiles (zoom_level, tile_column, tile_row);')
+    cursor.execute(
+        "CREATE UNIQUE INDEX tile_index on tiles (zoom_level, tile_column, tile_row);"
+    )
     conn.commit()
     conn.close()
 
+
 def pmtiles_to_dir(input, output, gzip):
     os.makedirs(output)
 
     with read(input) as reader:
         metadata = reader.metadata
-        metadata = set_metadata_compression(metadata,gzip)
-        with open(os.path.join(output,'metadata.json'),'w') as f:
+        metadata = set_metadata_compression(metadata, gzip)
+        with open(os.path.join(output, "metadata.json"), "w") as f:
             f.write(json.dumps(metadata))
 
         for tile, data in reader.tiles():
-            directory = os.path.join(output,str(tile[0]),str(tile[1]))
-            path = os.path.join(directory,str(tile[2]) + '.' + metadata['format'])
-            os.makedirs(directory,exist_ok=True)
-            with open(path,'wb') as f:
-                f.write(force_compress(data,gzip))
+            directory = os.path.join(output, str(tile[0]), str(tile[1]))
+            path = os.path.join(directory, str(tile[2]) + "." + metadata["format"])
+            os.makedirs(directory, exist_ok=True)
+            with open(path, "wb") as f:
+                f.write(force_compress(data, gzip))

python/pmtiles/reader.py

@@ -2,6 +2,7 @@ import json
 import mmap
 from contextlib import contextmanager
 
+
 @contextmanager
 def read(fname):
     r = Reader(fname)
@@ -10,73 +11,75 @@ def read(fname):
     finally:
         r.close()
 
+
 class Reader:
-    def __init__(self,fname):
+    def __init__(self, fname):
         self.f = open(fname, "r+b")
         self.mmap = mmap.mmap(self.f.fileno(), 0)
-        assert int.from_bytes(self.mmap[0:2],byteorder='little') == 0x4D50
-        first_entry_idx = 10+self.metadata_len
-        self.root_dir, self.leaves = self.load_directory(first_entry_idx,self.root_entries)
+        assert int.from_bytes(self.mmap[0:2], byteorder="little") == 0x4D50
+        first_entry_idx = 10 + self.metadata_len
+        self.root_dir, self.leaves = self.load_directory(
+            first_entry_idx, self.root_entries
+        )
 
-    def load_directory(self,offset,num_entries):
+    def load_directory(self, offset, num_entries):
         directory = {}
         leaves = {}
-        for i in range(offset,offset+num_entries*17,17):
-            z = int.from_bytes(self.mmap[i:i+1],byteorder='little')
-            x = int.from_bytes(self.mmap[i+1:i+4],byteorder='little')
-            y = int.from_bytes(self.mmap[i+4:i+7],byteorder='little')
-            tile_off = int.from_bytes(self.mmap[i+7:i+13],byteorder='little')
-            tile_len = int.from_bytes(self.mmap[i+13:i+17],byteorder='little')
-            if (z & 0b10000000):
-                leaves[(z & 0b01111111,x,y)] = (tile_off,tile_len)
+        for i in range(offset, offset + num_entries * 17, 17):
+            z = int.from_bytes(self.mmap[i : i + 1], byteorder="little")
+            x = int.from_bytes(self.mmap[i + 1 : i + 4], byteorder="little")
+            y = int.from_bytes(self.mmap[i + 4 : i + 7], byteorder="little")
+            tile_off = int.from_bytes(self.mmap[i + 7 : i + 13], byteorder="little")
+            tile_len = int.from_bytes(self.mmap[i + 13 : i + 17], byteorder="little")
+            if z & 0b10000000:
+                leaves[(z & 0b01111111, x, y)] = (tile_off, tile_len)
             else:
-                directory[(z,x,y)] = (tile_off,tile_len)
-        return (directory,leaves)
+                directory[(z, x, y)] = (tile_off, tile_len)
+        return (directory, leaves)
 
     def close(self):
         self.f.close()
 
     @property
     def metadata_len(self):
-        return int.from_bytes(self.mmap[4:8],byteorder='little')
+        return int.from_bytes(self.mmap[4:8], byteorder="little")
 
     @property
     def metadata(self):
-        s = self.mmap[10:10+self.metadata_len]
+        s = self.mmap[10 : 10 + self.metadata_len]
         return json.loads(s)
 
     @property
     def version(self):
-        return int.from_bytes(self.mmap[2:4],byteorder='little')
+        return int.from_bytes(self.mmap[2:4], byteorder="little")
 
     @property
     def root_entries(self):
-        return int.from_bytes(self.mmap[8:10],byteorder='little')
+        return int.from_bytes(self.mmap[8:10], byteorder="little")
 
     @property
     def leaf_level(self):
         return next(iter(self.leaves))[0]
 
-    def get(self,z,x,y):
-        val = self.root_dir.get((z,x,y))
+    def get(self, z, x, y):
+        val = self.root_dir.get((z, x, y))
         if val:
-            return self.mmap[val[0]:val[0]+val[1]]
+            return self.mmap[val[0] : val[0] + val[1]]
         else:
             if len(self.leaves) > 0:
                 level_diff = z - self.leaf_level
-                leaf = (self.leaf_level,x // (1 << level_diff),y // (1 << level_diff))
+                leaf = (self.leaf_level, x // (1 << level_diff), y // (1 << level_diff))
                 val = self.leaves.get(leaf)
                 if val:
-                    directory, _ = self.load_directory(val[0],val[1]//17)
-                    val = directory.get((z,x,y))
+                    directory, _ = self.load_directory(val[0], val[1] // 17)
+                    val = directory.get((z, x, y))
                     if val:
-                        return self.mmap[val[0]:val[0]+val[1]]
+                        return self.mmap[val[0] : val[0] + val[1]]
 
     def tiles(self):
-        for k,v in self.root_dir.items():
-            yield (k,self.mmap[v[0]:v[0]+v[1]])
+        for k, v in self.root_dir.items():
+            yield (k, self.mmap[v[0] : v[0] + v[1]])
         for val in self.leaves.values():
-            leaf_dir, _ = self.load_directory(val[0],val[1]//17)
-            for k,v in leaf_dir.items():
-                yield (k,self.mmap[v[0]:v[0]+v[1]])
-
+            leaf_dir, _ = self.load_directory(val[0], val[1] // 17)
+            for k, v in leaf_dir.items():
+                yield (k, self.mmap[v[0] : v[0] + v[1]])

python/pmtiles/writer.py

@@ -3,70 +3,80 @@ import json
 from contextlib import contextmanager
 from pmtiles import Entry
 
+
 def entrysort(t):
-    return (t.z,t.x,t.y)
+    return (t.z, t.x, t.y)
+
 
 # Find best base zoom to avoid extra indirection for as many tiles as we can
 # precondition: entries is sorted, only tile entries, len(entries) > max_dir_size
-def find_leaf_level(entries,max_dir_size):
-  return entries[max_dir_size].z - 1
+def find_leaf_level(entries, max_dir_size):
+    return entries[max_dir_size].z - 1
 
-def make_pyramid(tile_entries,start_leaf_offset,max_dir_size=21845):
-  sorted_entries = sorted(tile_entries,key=entrysort)
-  if len(sorted_entries) <= max_dir_size:
-    return (sorted_entries,[])
 
-  leaf_dirs = []
+def make_pyramid(tile_entries, start_leaf_offset, max_dir_size=21845):
+    sorted_entries = sorted(tile_entries, key=entrysort)
+    if len(sorted_entries) <= max_dir_size:
+        return (sorted_entries, [])
 
-  # determine root leaf level
-  leaf_level = find_leaf_level(sorted_entries,max_dir_size)
+    leaf_dirs = []
 
-  def by_parent(e):
-    level_diff = e.z - leaf_level
-    return (leaf_level,e.x//(1 << level_diff),e.y//(1 << level_diff))
+    # determine root leaf level
+    leaf_level = find_leaf_level(sorted_entries, max_dir_size)
 
-  root_entries = [e for e in sorted_entries if e.z < leaf_level]
-  # get all entries greater than or equal to the leaf level
-  entries_in_leaves = [e for e in sorted_entries if e.z >= leaf_level]
+    def by_parent(e):
+        level_diff = e.z - leaf_level
+        return (leaf_level, e.x // (1 << level_diff), e.y // (1 << level_diff))
 
-  # group the entries by their parent (stable)
-  entries_in_leaves.sort(key=by_parent)
+    root_entries = [e for e in sorted_entries if e.z < leaf_level]
+    # get all entries greater than or equal to the leaf level
+    entries_in_leaves = [e for e in sorted_entries if e.z >= leaf_level]
 
-  current_offset = start_leaf_offset
-  # pack entries into groups
-  packed_entries = []
-  packed_roots = []
+    # group the entries by their parent (stable)
+    entries_in_leaves.sort(key=by_parent)
 
-  for group in itertools.groupby(entries_in_leaves,key=by_parent):
-    subpyramid_entries = list(group[1])
+    current_offset = start_leaf_offset
+    # pack entries into groups
+    packed_entries = []
+    packed_roots = []
 
-    root = by_parent(subpyramid_entries[0])
-    if len(packed_entries) + len(subpyramid_entries) <= max_dir_size:
-      packed_entries.extend(subpyramid_entries)
-      packed_roots.append((root[0],root[1],root[2]))
-    else:
-      # flush the current packed entries
+    for group in itertools.groupby(entries_in_leaves, key=by_parent):
+        subpyramid_entries = list(group[1])
 
-      for p in packed_roots:
-        root_entries.append(Entry(p[0],p[1],p[2],current_offset,17 * len(packed_entries),True))
-      # re-sort the packed_entries by ZXY order
-      packed_entries.sort(key=entrysort)
-      leaf_dirs.append(packed_entries)
+        root = by_parent(subpyramid_entries[0])
+        if len(packed_entries) + len(subpyramid_entries) <= max_dir_size:
+            packed_entries.extend(subpyramid_entries)
+            packed_roots.append((root[0], root[1], root[2]))
+        else:
+            # flush the current packed entries
 
-      current_offset += 17 * len(packed_entries)
-      packed_entries = subpyramid_entries
-      packed_roots = [(root[0],root[1],root[2])]
+            for p in packed_roots:
+                root_entries.append(
+                    Entry(
+                        p[0], p[1], p[2], current_offset, 17 * len(packed_entries), True
+                    )
+                )
+            # re-sort the packed_entries by ZXY order
+            packed_entries.sort(key=entrysort)
+            leaf_dirs.append(packed_entries)
 
-  # finalize the last set
-  if len(packed_entries):
+            current_offset += 17 * len(packed_entries)
+            packed_entries = subpyramid_entries
+            packed_roots = [(root[0], root[1], root[2])]
 
-    for p in packed_roots:
-      root_entries.append(Entry(p[0],p[1],p[2],current_offset,17 * len(packed_entries),True))
-    # re-sort the packed_entries by ZXY order
-    packed_entries.sort(key=entrysort)
-    leaf_dirs.append(packed_entries)
+    # finalize the last set
+    if len(packed_entries):
+
+        for p in packed_roots:
+            root_entries.append(
+                Entry(p[0], p[1], p[2], current_offset, 17 * len(packed_entries), True)
+            )
+        # re-sort the packed_entries by ZXY order
+        packed_entries.sort(key=entrysort)
+        leaf_dirs.append(packed_entries)
+
+    return (root_entries, leaf_dirs)
 
-  return (root_entries,leaf_dirs)
 
 @contextmanager
 def write(fname):
@@ -76,60 +86,67 @@ def write(fname):
     finally:
         w.close()
 
+
 class Writer:
-    def __init__(self,fname):
-        self.f = open(fname,'wb')
+    def __init__(self, fname):
+        self.f = open(fname, "wb")
         self.offset = 512000
-        self.f.write(b'\0' * self.offset)
+        self.f.write(b"\0" * self.offset)
         self.tile_entries = []
         self.hash_to_offset = {}
 
-    def write_tile(self,z,x,y,data):
+    def write_tile(self, z, x, y, data):
         hsh = hash(data)
         if hsh in self.hash_to_offset:
-            self.tile_entries.append(Entry(z,x,y,self.hash_to_offset[hsh],len(data),False))
+            self.tile_entries.append(
+                Entry(z, x, y, self.hash_to_offset[hsh], len(data), False)
+            )
         else:
             self.f.write(data)
-            self.tile_entries.append(Entry(z,x,y,self.offset,len(data),False))
+            self.tile_entries.append(Entry(z, x, y, self.offset, len(data), False))
             self.hash_to_offset[hsh] = self.offset
             self.offset = self.offset + len(data)
 
-    def write_entry(self,entry):
+    def write_entry(self, entry):
         if entry.is_dir:
-          z_bytes = 0b10000000 | entry.z
+            z_bytes = 0b10000000 | entry.z
         else:
-          z_bytes = entry.z
-        self.f.write(z_bytes.to_bytes(1,byteorder='little'))
-        self.f.write(entry.x.to_bytes(3,byteorder='little'))
-        self.f.write(entry.y.to_bytes(3,byteorder='little'))
-        self.f.write(entry.offset.to_bytes(6,byteorder='little'))
-        self.f.write(entry.length.to_bytes(4,byteorder='little'))
+            z_bytes = entry.z
+        self.f.write(z_bytes.to_bytes(1, byteorder="little"))
+        self.f.write(entry.x.to_bytes(3, byteorder="little"))
+        self.f.write(entry.y.to_bytes(3, byteorder="little"))
+        self.f.write(entry.offset.to_bytes(6, byteorder="little"))
+        self.f.write(entry.length.to_bytes(4, byteorder="little"))
 
-    def write_header(self,metadata,root_entries_len):
-        self.f.write((0x4D50).to_bytes(2,byteorder='little'))
-        self.f.write((2).to_bytes(2,byteorder='little'))
+    def write_header(self, metadata, root_entries_len):
+        self.f.write((0x4D50).to_bytes(2, byteorder="little"))
+        self.f.write((2).to_bytes(2, byteorder="little"))
         metadata_serialized = json.dumps(metadata)
         # 512000 - (17 * 21845) - 2 (magic) - 2 (version) - 4 (jsonlen) - 2 (dictentries) = 140625
         assert len(metadata_serialized) < 140625
-        self.f.write(len(metadata_serialized).to_bytes(4,byteorder='little'))
-        self.f.write(root_entries_len.to_bytes(2,byteorder='little'))
-        self.f.write(metadata_serialized.encode('utf-8'))
+        self.f.write(len(metadata_serialized).to_bytes(4, byteorder="little"))
+        self.f.write(root_entries_len.to_bytes(2, byteorder="little"))
+        self.f.write(metadata_serialized.encode("utf-8"))
 
-    def finalize(self,metadata = {}):
-        root_dir, leaf_dirs = make_pyramid(self.tile_entries,self.offset)
+    def finalize(self, metadata={}):
+        root_dir, leaf_dirs = make_pyramid(self.tile_entries, self.offset)
 
         if len(leaf_dirs) > 0:
-          for leaf_dir in leaf_dirs:
-            for entry in leaf_dir:
-              self.write_entry(entry)
+            for leaf_dir in leaf_dirs:
+                for entry in leaf_dir:
+                    self.write_entry(entry)
 
         self.f.seek(0)
-        self.write_header(metadata,len(root_dir))
+        self.write_header(metadata, len(root_dir))
 
         for entry in root_dir:
             self.write_entry(entry)
 
-        return {'num_tiles':len(self.tile_entries),'num_unique_tiles':len(self.hash_to_offset),'num_leaves':len(leaf_dirs)}
+        return {
+            "num_tiles": len(self.tile_entries),
+            "num_unique_tiles": len(self.hash_to_offset),
+            "num_leaves": len(leaf_dirs),
+        }
 
     def close(self):
         self.f.close()

python/setup.py

@@ -19,6 +19,6 @@ setuptools.setup(
         "License :: OSI Approved :: BSD License",
         "Operating System :: OS Independent",
     ],
-    scripts=['bin/pmtiles-convert','bin/pmtiles-serve','bin/pmtiles-show'],
-    requires_python='>=3.0'
+    scripts=["bin/pmtiles-convert", "bin/pmtiles-serve", "bin/pmtiles-show"],
+    requires_python=">=3.0",
 )

python/test/test_writer.py

@@ -2,95 +2,95 @@ import unittest
 from pmtiles import Entry
 from pmtiles.writer import find_leaf_level, make_pyramid
 
+
 class TestTilePyramid(unittest.TestCase):
     def test_root_sorted(self):
-      entries = [
-        Entry(1,0,0,1,1,False),
-        Entry(1,0,1,2,1,False),
-        Entry(1,1,0,3,1,False),
-        Entry(1,1,1,4,1,False),
-        Entry(0,0,0,0,1,False)
-      ]
-      root_entries, leaf_dirs = make_pyramid(entries,0,6)
-      self.assertEqual(len(root_entries),5)
-      self.assertEqual(len(leaf_dirs),0)
-      self.assertEqual(root_entries[0].z,0)
-      self.assertEqual(root_entries[4].z,1)
+        entries = [
+            Entry(1, 0, 0, 1, 1, False),
+            Entry(1, 0, 1, 2, 1, False),
+            Entry(1, 1, 0, 3, 1, False),
+            Entry(1, 1, 1, 4, 1, False),
+            Entry(0, 0, 0, 0, 1, False),
+        ]
+        root_entries, leaf_dirs = make_pyramid(entries, 0, 6)
+        self.assertEqual(len(root_entries), 5)
+        self.assertEqual(len(leaf_dirs), 0)
+        self.assertEqual(root_entries[0].z, 0)
+        self.assertEqual(root_entries[4].z, 1)
 
     def test_leafdir(self):
-      entries = [
-        Entry(0,0,0,0,1,False),
-        Entry(1,0,0,1,1,False),
-        Entry(1,0,1,2,1,False),
-        Entry(1,1,0,3,1,False),
-        Entry(1,1,1,4,1,False),
-        Entry(2,0,0,5,1,False),
-        Entry(3,0,0,6,1,False),
-        Entry(2,0,1,7,1,False),
-        Entry(3,0,2,8,1,False)
-      ]
-      root_entries, leaf_dirs = make_pyramid(entries,0,7)
-      self.assertEqual(len(root_entries),7)
-      self.assertEqual(root_entries[5].y,0)
-      self.assertEqual(root_entries[6].y,1)
-      self.assertEqual(len(leaf_dirs),1)
-      self.assertEqual(len(leaf_dirs[0]),4)
-      self.assertEqual(leaf_dirs[0][0].z,2)
-      self.assertEqual(leaf_dirs[0][1].z,2)
-      self.assertEqual(leaf_dirs[0][2].z,3)
-      self.assertEqual(leaf_dirs[0][3].z,3)
-
+        entries = [
+            Entry(0, 0, 0, 0, 1, False),
+            Entry(1, 0, 0, 1, 1, False),
+            Entry(1, 0, 1, 2, 1, False),
+            Entry(1, 1, 0, 3, 1, False),
+            Entry(1, 1, 1, 4, 1, False),
+            Entry(2, 0, 0, 5, 1, False),
+            Entry(3, 0, 0, 6, 1, False),
+            Entry(2, 0, 1, 7, 1, False),
+            Entry(3, 0, 2, 8, 1, False),
+        ]
+        root_entries, leaf_dirs = make_pyramid(entries, 0, 7)
+        self.assertEqual(len(root_entries), 7)
+        self.assertEqual(root_entries[5].y, 0)
+        self.assertEqual(root_entries[6].y, 1)
+        self.assertEqual(len(leaf_dirs), 1)
+        self.assertEqual(len(leaf_dirs[0]), 4)
+        self.assertEqual(leaf_dirs[0][0].z, 2)
+        self.assertEqual(leaf_dirs[0][1].z, 2)
+        self.assertEqual(leaf_dirs[0][2].z, 3)
+        self.assertEqual(leaf_dirs[0][3].z, 3)
 
     def test_leafdir_overflow(self):
-      entries = [
-        Entry(0,0,0,0,1,False),
-        Entry(1,0,0,1,1,False),
-        Entry(1,0,1,2,1,False),
-        Entry(1,1,0,3,1,False),
-        Entry(1,1,1,4,1,False),
-        Entry(2,0,0,5,1,False),
-        Entry(3,0,0,6,1,False),
-        Entry(3,0,1,7,1,False),
-        Entry(3,1,0,8,1,False),
-        Entry(3,1,1,9,1,False),
-        Entry(2,0,1,10,1,False),
-        Entry(3,0,2,11,1,False),
-        Entry(3,0,3,12,1,False),
-        Entry(3,1,2,13,1,False),
-        Entry(3,1,3,14,1,False)
-      ]
-      root_entries, leaf_dirs = make_pyramid(entries,0,7)
-      self.assertEqual(len(root_entries),7)
-      self.assertEqual(root_entries[5].y,0)
-      self.assertEqual(root_entries[6].y,1)
+        entries = [
+            Entry(0, 0, 0, 0, 1, False),
+            Entry(1, 0, 0, 1, 1, False),
+            Entry(1, 0, 1, 2, 1, False),
+            Entry(1, 1, 0, 3, 1, False),
+            Entry(1, 1, 1, 4, 1, False),
+            Entry(2, 0, 0, 5, 1, False),
+            Entry(3, 0, 0, 6, 1, False),
+            Entry(3, 0, 1, 7, 1, False),
+            Entry(3, 1, 0, 8, 1, False),
+            Entry(3, 1, 1, 9, 1, False),
+            Entry(2, 0, 1, 10, 1, False),
+            Entry(3, 0, 2, 11, 1, False),
+            Entry(3, 0, 3, 12, 1, False),
+            Entry(3, 1, 2, 13, 1, False),
+            Entry(3, 1, 3, 14, 1, False),
+        ]
+        root_entries, leaf_dirs = make_pyramid(entries, 0, 7)
+        self.assertEqual(len(root_entries), 7)
+        self.assertEqual(root_entries[5].y, 0)
+        self.assertEqual(root_entries[6].y, 1)
 
     def test_sparse_pyramid(self):
-      entries = [
-        Entry(0,0,0,0,1,False),
-        Entry(1,0,0,1,1,False),
-        Entry(1,0,1,2,1,False),
-        Entry(1,1,0,3,1,False),
-        Entry(1,1,1,4,1,False),
-        Entry(2,0,0,5,1,False),
-        Entry(3,0,0,6,1,False),
-        # Entry(2,0,1,7,1,False), make this entry missing
-        Entry(3,0,2,8,1,False)
-      ]
-      root_entries, leaf_dirs = make_pyramid(entries,0,7)
-      self.assertEqual(len(root_entries),7)
-      self.assertEqual(root_entries[6].z,2)
-      self.assertEqual(root_entries[6].x,0)
-      self.assertEqual(root_entries[6].y,1)
+        entries = [
+            Entry(0, 0, 0, 0, 1, False),
+            Entry(1, 0, 0, 1, 1, False),
+            Entry(1, 0, 1, 2, 1, False),
+            Entry(1, 1, 0, 3, 1, False),
+            Entry(1, 1, 1, 4, 1, False),
+            Entry(2, 0, 0, 5, 1, False),
+            Entry(3, 0, 0, 6, 1, False),
+            # Entry(2,0,1,7,1,False), make this entry missing
+            Entry(3, 0, 2, 8, 1, False),
+        ]
+        root_entries, leaf_dirs = make_pyramid(entries, 0, 7)
+        self.assertEqual(len(root_entries), 7)
+        self.assertEqual(root_entries[6].z, 2)
+        self.assertEqual(root_entries[6].x, 0)
+        self.assertEqual(root_entries[6].y, 1)
 
     def test_full_z7_pyramid(self):
-      entries = []
-      # create artificial 8 levels  
-      for z in range(0,9):
-        for x in range(0,pow(2,z)):
-          for y in range(0,pow(2,z)):
-            entries.append(Entry(z,x,y,0,0,False))
-      self.assertEqual(find_leaf_level(entries,21845),7)
-      root_entries, leaf_dirs = make_pyramid(entries,0)
-      self.assertEqual(len(root_entries),21845)
-      self.assertEqual(len(leaf_dirs),4)
-      self.assertTrue(len(leaf_dirs[0]) <= 21845)
+        entries = []
+        # create artificial 8 levels
+        for z in range(0, 9):
+            for x in range(0, pow(2, z)):
+                for y in range(0, pow(2, z)):
+                    entries.append(Entry(z, x, y, 0, 0, False))
+        self.assertEqual(find_leaf_level(entries, 21845), 7)
+        root_entries, leaf_dirs = make_pyramid(entries, 0)
+        self.assertEqual(len(root_entries), 21845)
+        self.assertEqual(len(leaf_dirs), 4)
+        self.assertTrue(len(leaf_dirs[0]) <= 21845)