mirror of
https://github.com/protomaps/PMTiles.git
synced 2026-02-04 02:41:09 +00:00
refactor: type header (#597)
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
import gzip
|
import gzip
|
||||||
import io
|
import io
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
|
from typing import TypedDict
|
||||||
|
|
||||||
|
|
||||||
class Entry:
|
class Entry:
|
||||||
@@ -186,7 +187,35 @@ class MagicNumberNotFound(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def deserialize_header(buf):
|
class HeaderDict(TypedDict):
|
||||||
|
version: int
|
||||||
|
root_offset: int
|
||||||
|
root_length: int
|
||||||
|
metadata_offset: int
|
||||||
|
metadata_length: int
|
||||||
|
leaf_directory_offset: int
|
||||||
|
leaf_directory_length: int
|
||||||
|
tile_data_offset: int
|
||||||
|
tile_data_length: int
|
||||||
|
addressed_tiles_count: int
|
||||||
|
tile_entries_count: int
|
||||||
|
tile_contents_count: int
|
||||||
|
clustered: bool
|
||||||
|
internal_compression: Compression
|
||||||
|
tile_compression: Compression
|
||||||
|
tile_type: TileType
|
||||||
|
min_zoom: int
|
||||||
|
max_zoom: int
|
||||||
|
min_lon_e7: int
|
||||||
|
min_lat_e7: int
|
||||||
|
max_lon_e7: int
|
||||||
|
max_lat_e7: int
|
||||||
|
center_zoom: int
|
||||||
|
center_lon_e7: int
|
||||||
|
center_lat_e7: int
|
||||||
|
|
||||||
|
|
||||||
|
def deserialize_header(buf) -> HeaderDict:
|
||||||
if buf[0:7].decode() != "PMTiles":
|
if buf[0:7].decode() != "PMTiles":
|
||||||
raise MagicNumberNotFound()
|
raise MagicNumberNotFound()
|
||||||
|
|
||||||
@@ -228,7 +257,7 @@ def deserialize_header(buf):
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def serialize_header(h):
|
def serialize_header(h: HeaderDict):
|
||||||
b_io = io.BytesIO()
|
b_io = io.BytesIO()
|
||||||
|
|
||||||
def write_uint64(i):
|
def write_uint64(i):
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import unittest
|
import unittest
|
||||||
from pmtiles.tile import zxy_to_tileid, tileid_to_zxy, Entry
|
from pmtiles.tile import zxy_to_tileid, tileid_to_zxy, Entry
|
||||||
from pmtiles.tile import read_varint, write_varint
|
from pmtiles.tile import read_varint, write_varint
|
||||||
from pmtiles.tile import Entry, find_tile, Compression, TileType
|
from pmtiles.tile import Entry, find_tile, Compression, TileType, HeaderDict
|
||||||
from pmtiles.tile import serialize_directory, deserialize_directory
|
from pmtiles.tile import serialize_directory, deserialize_directory
|
||||||
from pmtiles.tile import serialize_header, deserialize_header, SpecVersionUnsupported, MagicNumberNotFound
|
from pmtiles.tile import serialize_header, deserialize_header, SpecVersionUnsupported, MagicNumberNotFound
|
||||||
import io
|
import io
|
||||||
@@ -194,3 +194,9 @@ class TestHeader(unittest.TestCase):
|
|||||||
|
|
||||||
with self.assertRaises(MagicNumberNotFound):
|
with self.assertRaises(MagicNumberNotFound):
|
||||||
result = deserialize_header(b'PM\x00\x02')
|
result = deserialize_header(b'PM\x00\x02')
|
||||||
|
|
||||||
|
def test_type(self):
|
||||||
|
result = deserialize_header(b"PMTiles\x03" + b"\x00" * 120)
|
||||||
|
self.assertIsInstance(result, dict)
|
||||||
|
expected_keys = set(HeaderDict.__annotations__.keys())
|
||||||
|
self.assertTrue(expected_keys.issubset(result.keys()))
|
||||||
|
|||||||
Reference in New Issue
Block a user