mirrors/PMTiles
1
0
Fork 0
mirror of https://github.com/protomaps/PMTiles.git synced 2026-02-04 10:51:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

reformat with black

Browse Source
This commit is contained in:
fscottfoti
2022-04-09 12:32:11 -07:00
parent 02c4dcc920
commit ed40933563
6 changed files with 252 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
python/pmtiles/__init__.py
View File

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

72
python/pmtiles/convert.py
View File

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

67
python/pmtiles/reader.py
View File

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

163
python/pmtiles/writer.py
View File

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

4
python/setup.py
View File

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

164
python/test/test_writer.py
View File

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