3746140319

Build Type

push

github

Committed by GitHub

Commit Message

Pre-commit tools (#121)

Run Details

74 of 96 new or added lines in 6 files covered. (77.08%)

2 existing lines in 1 file now uncovered.

558 of 727 relevant lines covered (76.75%)

3.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.52

/mapbox_vector_tile/decoder.py

from mapbox_vector_tile.Mapbox import vector_tile_pb2 as vector_tile

cmd_bits = 3

CMD_MOVE_TO = 1
CMD_LINE_TO = 2
CMD_SEG_END = 7

UNKNOWN = 0
POINT = 1
LINESTRING = 2
POLYGON = 3


class TileData:
    def __init__(self):
        self.tile = vector_tile.tile()

    def getMessage(self, pbf_data, y_coord_down=False):
        self.tile.ParseFromString(pbf_data)

        tile = {}
        for layer in self.tile.layers:
            keys = layer.keys
            vals = layer.values

            features = []
            for feature in layer.features:
                tags = feature.tags
                props = {}
                assert len(tags) % 2 == 0, "Unexpected number of tags"
                for key_idx, val_idx in zip(tags[::2], tags[1::2]):
                    key = keys[key_idx]
                    val = vals[val_idx]
                    value = self.parse_value(val)
                    props[key] = value

                geometry = self.parse_geometry(feature.geometry, feature.type, layer.extent, y_coord_down)
                new_feature = {"geometry": geometry, "properties": props, "id": feature.id, "type": feature.type}
                features.append(new_feature)

            tile[layer.name] = {
                "extent": layer.extent,
                "version": layer.version,
                "features": features,
            }
        return tile

    def zero_pad(self, val):
        return "0" + val if val[0] == "b" else val

    def parse_value(self, val):
        for candidate in (
            "bool_value",
            "double_value",
            "float_value",
            "int_value",
            "sint_value",
            "string_value",
            "uint_value",
        ):
            if val.HasField(candidate):
                return getattr(val, candidate)
        raise ValueError(f"{val} is an unknown value")

    def zig_zag_decode(self, n):
        return (n >> 1) ^ (-(n & 1))

    def parse_geometry(self, geom, ftype, extent, y_coord_down):  # noqa:C901
        # [9 0 8192 26 0 10 2 0 0 2 15]
        i = 0
        coords = []
        dx = 0
        dy = 0
        parts = []  # for multi linestrings and polygons

        while i != len(geom):
            item = bin(geom[i])
            ilen = len(item)
            cmd = int(self.zero_pad(item[(ilen - cmd_bits) : ilen]), 2)
            cmd_len = int(self.zero_pad(item[: ilen - cmd_bits]), 2)

            i = i + 1

            def _ensure_polygon_closed(coords):
                if coords and coords[0] != coords[-1]:
                    coords.append(coords[0])

            if cmd == CMD_SEG_END:
                if ftype == POLYGON:
                    _ensure_polygon_closed(coords)
                parts.append(coords)
                coords = []

            elif cmd == CMD_MOVE_TO or cmd == CMD_LINE_TO:

                if coords and cmd == CMD_MOVE_TO:
                    if ftype in (LINESTRING, POLYGON):
                        # multi line string or polygon
                        # our encoder includes CMD_SEG_END to denote
                        # the end of a polygon ring, but this path
                        # would also handle the case where we receive
                        # a move without a previous close on polygons

                        # for polygons, we want to ensure that it is
                        # closed
                        if ftype == POLYGON:
                            _ensure_polygon_closed(coords)
                        parts.append(coords)
                        coords = []

                for point in range(0, cmd_len):
                    x = geom[i]
                    i = i + 1

                    y = geom[i]
                    i = i + 1

                    # zipzag decode
                    x = self.zig_zag_decode(x)
                    y = self.zig_zag_decode(y)

                    x = x + dx
                    y = y + dy

                    dx = x
                    dy = y

                    if not y_coord_down:
                        y = extent - y

                    coords.append([x, y])

        if ftype == POINT:
            if len(coords) == 1:
                return {"type": "Point", "coordinates": coords[0]}
            else:
                return {"type": "MultiPoint", "coordinates": coords}
        elif ftype == LINESTRING:
            if parts:
                if coords:
                    parts.append(coords)
                if len(parts) == 1:
                    return {"type": "LineString", "coordinates": parts[0]}
                else:
                    return {"type": "MultiLineString", "coordinates": parts}
            else:
                return {"type": "LineString", "coordinates": coords}
        elif ftype == POLYGON:
            if coords:
                parts.append(coords)

            def _area_sign(ring):
                a = sum(ring[i][0] * ring[i + 1][1] - ring[i + 1][0] * ring[i][1] for i in range(0, len(ring) - 1))
                return -1 if a < 0 else 1 if a > 0 else 0

            polygon = []
            polygons = []
            winding = 0

            for ring in parts:
                a = _area_sign(ring)
                if a == 0:
                    continue
                if winding == 0:
                    winding = a

                if winding == a:
                    if polygon:
                        polygons.append(polygon)
                    polygon = [ring]
                else:
                    polygon.append(ring)

            if polygon:
                polygons.append(polygon)

            if len(polygons) == 1:
                return {"type": "Polygon", "coordinates": polygons[0]}
            else:
                return {"type": "MultiPolygon", "coordinates": polygons}

        else:
            raise ValueError(f"Unknown geometry type: {ftype}")

1	from mapbox_vector_tile.Mapbox import vector_tile_pb2 as vector_tile	5✔
2
3	cmd_bits = 3	5✔
4
5	CMD_MOVE_TO = 1	5✔
6	CMD_LINE_TO = 2	5✔
7	CMD_SEG_END = 7	5✔
8
9	UNKNOWN = 0	5✔
10	POINT = 1	5✔
11	LINESTRING = 2	5✔
12	POLYGON = 3	5✔
13
14
15	class TileData:	5✔
16	def __init__(self):	5✔
17	self.tile = vector_tile.tile()	5✔
18
19	def getMessage(self, pbf_data, y_coord_down=False):	5✔
20	self.tile.ParseFromString(pbf_data)	5✔
21
22	tile = {}	5✔
23	for layer in self.tile.layers:	5✔
24	keys = layer.keys	5✔
25	vals = layer.values	5✔
26
27	features = []	5✔
28	for feature in layer.features:	5✔
29	tags = feature.tags	5✔
30	props = {}	5✔
31	assert len(tags) % 2 == 0, "Unexpected number of tags"	5✔
32	for key_idx, val_idx in zip(tags[::2], tags[1::2]):	5✔
33	key = keys[key_idx]	5✔
34	val = vals[val_idx]	5✔
35	value = self.parse_value(val)	5✔
36	props[key] = value	5✔
37
38	geometry = self.parse_geometry(feature.geometry, feature.type, layer.extent, y_coord_down)	5✔
39	new_feature = {"geometry": geometry, "properties": props, "id": feature.id, "type": feature.type}	5✔
40	features.append(new_feature)	5✔
41
42	tile[layer.name] = {	5✔
43	"extent": layer.extent,
44	"version": layer.version,
45	"features": features,
46	}
47	return tile	5✔
48
49	def zero_pad(self, val):	5✔
50	return "0" + val if val[0] == "b" else val	5✔
51
52	def parse_value(self, val):	5✔
53	for candidate in (	5✔
54	"bool_value",
55	"double_value",
56	"float_value",
57	"int_value",
58	"sint_value",
59	"string_value",
60	"uint_value",
61	):
62	if val.HasField(candidate):	5✔
63	return getattr(val, candidate)	5✔
NEW 64	raise ValueError(f"{val} is an unknown value")	×
65
66	def zig_zag_decode(self, n):	5✔
67	return (n >> 1) ^ (-(n & 1))	5✔
68
69	def parse_geometry(self, geom, ftype, extent, y_coord_down): # noqa:C901	5✔
70	# [9 0 8192 26 0 10 2 0 0 2 15]
71	i = 0	5✔
72	coords = []	5✔
73	dx = 0	5✔
74	dy = 0	5✔
75	parts = [] # for multi linestrings and polygons	5✔
76
77	while i != len(geom):	5✔
78	item = bin(geom[i])	5✔
79	ilen = len(item)	5✔
80	cmd = int(self.zero_pad(item[(ilen - cmd_bits) : ilen]), 2)	5✔
81	cmd_len = int(self.zero_pad(item[: ilen - cmd_bits]), 2)	5✔
82
83	i = i + 1	5✔
84
85	def _ensure_polygon_closed(coords):	5✔
86	if coords and coords[0] != coords[-1]:	5✔
87	coords.append(coords[0])	5✔
88
89	if cmd == CMD_SEG_END:	5✔
90	if ftype == POLYGON:	5✔
91	_ensure_polygon_closed(coords)	5✔
92	parts.append(coords)	5✔
93	coords = []	5✔
94
95	elif cmd == CMD_MOVE_TO or cmd == CMD_LINE_TO:	5✔
96
97	if coords and cmd == CMD_MOVE_TO:	5✔
98	if ftype in (LINESTRING, POLYGON):	5✔
99	# multi line string or polygon
100	# our encoder includes CMD_SEG_END to denote
101	# the end of a polygon ring, but this path
102	# would also handle the case where we receive
103	# a move without a previous close on polygons
104
105	# for polygons, we want to ensure that it is
106	# closed
107	if ftype == POLYGON:	5✔
108	_ensure_polygon_closed(coords)	5✔
109	parts.append(coords)	5✔
110	coords = []	5✔
111
112	for point in range(0, cmd_len):	5✔
113	x = geom[i]	5✔
114	i = i + 1	5✔
115
116	y = geom[i]	5✔
117	i = i + 1	5✔
118
119	# zipzag decode
120	x = self.zig_zag_decode(x)	5✔
121	y = self.zig_zag_decode(y)	5✔
122
123	x = x + dx	5✔
124	y = y + dy	5✔
125
126	dx = x	5✔
127	dy = y	5✔
128
129	if not y_coord_down:	5✔
130	y = extent - y	5✔
131
132	coords.append([x, y])	5✔
133
134	if ftype == POINT:	5✔
135	if len(coords) == 1:	5✔
136	return {"type": "Point", "coordinates": coords[0]}	5✔
137	else:
138	return {"type": "MultiPoint", "coordinates": coords}	5✔
139	elif ftype == LINESTRING:	5✔
140	if parts:	5✔
141	if coords:	5✔
142	parts.append(coords)	5✔
143	if len(parts) == 1:	5✔
NEW 144	return {"type": "LineString", "coordinates": parts[0]}	×
145	else:
146	return {"type": "MultiLineString", "coordinates": parts}	5✔
147	else:
148	return {"type": "LineString", "coordinates": coords}	5✔
149	elif ftype == POLYGON:	5✔
150	if coords:	5✔
151	parts.append(coords)	5✔
152
153	def _area_sign(ring):	5✔
154	a = sum(ring[i][0] * ring[i + 1][1] - ring[i + 1][0] * ring[i][1] for i in range(0, len(ring) - 1))	5✔
155	return -1 if a < 0 else 1 if a > 0 else 0	5✔
156
157	polygon = []	5✔
158	polygons = []	5✔
159	winding = 0	5✔
160
161	for ring in parts:	5✔
162	a = _area_sign(ring)	5✔
163	if a == 0:	5✔
164	continue	5✔
165	if winding == 0:	5✔
166	winding = a	5✔
167
168	if winding == a:	5✔
169	if polygon:	5✔
170	polygons.append(polygon)	5✔
171	polygon = [ring]	5✔
172	else:
173	polygon.append(ring)	5✔
174
175	if polygon:	5✔
176	polygons.append(polygon)	5✔
177
178	if len(polygons) == 1:	5✔
179	return {"type": "Polygon", "coordinates": polygons[0]}	5✔
180	else:
181	return {"type": "MultiPolygon", "coordinates": polygons}	5✔
182
183	else:
NEW 184	raise ValueError(f"Unknown geometry type: {ftype}")	×

tilezen / mapbox-vector-tile / 3746140319

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous