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

96.86

/mapbox_vector_tile/encoder.py

import decimal
from numbers import Number

from shapely.geometry.base import BaseGeometry
from shapely.geometry.multipolygon import MultiPolygon
from shapely.geometry.polygon import orient, Polygon
from shapely.ops import transform
from shapely.wkb import loads as load_wkb
from shapely.wkt import loads as load_wkt

from mapbox_vector_tile.Mapbox import vector_tile_pb2 as vector_tile
from mapbox_vector_tile.geom_encoder import GeometryEncoder
from mapbox_vector_tile.polygon import make_it_valid
from mapbox_vector_tile.simple_shape import SimpleShape


def apply_map(fn, x):
    return list(map(fn, x))


def on_invalid_geometry_raise(shape):
    raise ValueError(f"Invalid geometry: {shape.wkt}")


def on_invalid_geometry_ignore(shape):
    return None


def on_invalid_geometry_make_valid(shape):
    return make_it_valid(shape)


class VectorTile:
    def __init__(
        self, extents, on_invalid_geometry=None, max_geometry_validate_tries=5, round_fn=None, check_winding_order=True
    ):
        self.tile = vector_tile.tile()
        self.extents = extents
        self.on_invalid_geometry = on_invalid_geometry
        self.check_winding_order = check_winding_order
        self.max_geometry_validate_tries = max_geometry_validate_tries
        if round_fn:
            self._round = round_fn
        else:
            self._round = self._round_quantize

    def _round_quantize(self, val):

        # round() has different behavior in python 2/3
        # For consistency between 2 and 3 we use quantize, however
        # it is slower than the built in round function.
        d = decimal.Decimal(val)
        rounded = d.quantize(1, rounding=decimal.ROUND_HALF_EVEN)
        return float(rounded)

    def addFeatures(self, features, layer_name="", quantize_bounds=None, y_coord_down=False):

        self.layer = self.tile.layers.add()
        self.layer.name = layer_name
        self.layer.version = 1
        self.layer.extent = self.extents

        self.key_idx = 0
        self.val_idx = 0
        self.seen_keys_idx = {}
        self.seen_values_idx = {}
        self.seen_values_bool_idx = {}

        for feature in features:

            # skip missing or empty geometries
            geometry_spec = feature.get("geometry")
            if geometry_spec is None:
                continue
            shape = self._load_geometry(geometry_spec)

            if shape is None:
                raise NotImplementedError("Can't do geometries that are not wkt, wkb, or shapely geometries")

            if shape.is_empty:
                continue

            if quantize_bounds:
                shape = self.quantize(shape, quantize_bounds)
            if self.check_winding_order:
                shape = self.enforce_winding_order(shape, y_coord_down)

            if shape is not None and not shape.is_empty:
                self.addFeature(feature, shape, y_coord_down)

    def enforce_winding_order(self, shape, y_coord_down, n_try=1):
        if shape.type == "MultiPolygon":
            # If we are a multipolygon, we need to ensure that the
            # winding orders of the consituent polygons are
            # correct. In particular, the winding order of the
            # interior rings need to be the opposite of the
            # exterior ones, and all interior rings need to follow
            # the exterior one. This is how the end of one polygon
            # and the beginning of another are signaled.
            shape = self.enforce_multipolygon_winding_order(shape, y_coord_down, n_try)

        elif shape.type == "Polygon":
            # Ensure that polygons are also oriented with the
            # appropriate winding order. Their exterior rings must
            # have a clockwise order, which is translated into a
            # clockwise order in MVT's tile-local coordinates with
            # the Y axis in "screen" (i.e: +ve down) configuration.
            # Note that while the Y axis flips, we also invert the
            # Y coordinate to get the tile-local value, which means
            # the clockwise orientation is unchanged.
            shape = self.enforce_polygon_winding_order(shape, y_coord_down, n_try)

        # other shapes just get passed through
        return shape

    def quantize(self, shape, bounds):
        minx, miny, maxx, maxy = bounds

        def fn(x, y, z=None):
            xfac = self.extents / (maxx - minx)
            yfac = self.extents / (maxy - miny)
            x = xfac * (x - minx)
            y = yfac * (y - miny)
            return self._round(x), self._round(y)

        return transform(fn, shape)

    def handle_shape_validity(self, shape, y_coord_down, n_try):
        if shape.is_valid:
            return shape

        if n_try >= self.max_geometry_validate_tries:
            # ensure that we don't recurse indefinitely with an
            # invalid geometry handler that doesn't validate
            # geometries
            return None

        if self.on_invalid_geometry:
            shape = self.on_invalid_geometry(shape)
            if shape is not None and not shape.is_empty:
                # this means that we have a handler that might have
                # altered the geometry. We'll run through the process
                # again, but keep track of which attempt we are on to
                # terminate the recursion.
                shape = self.enforce_winding_order(shape, y_coord_down, n_try + 1)

        return shape

    def enforce_multipolygon_winding_order(self, shape, y_coord_down, n_try):
        assert shape.type == "MultiPolygon"

        parts = []
        for part in shape.geoms:
            part = self.enforce_polygon_winding_order(part, y_coord_down, n_try)
            if part is not None and not part.is_empty:
                if part.geom_type == "MultiPolygon":
                    parts.extend(part.geoms)
                else:
                    parts.append(part)

        if not parts:
            return None

        if len(parts) == 1:
            oriented_shape = parts[0]
        else:
            oriented_shape = MultiPolygon(parts)

        oriented_shape = self.handle_shape_validity(oriented_shape, y_coord_down, n_try)
        return oriented_shape

    def enforce_polygon_winding_order(self, shape, y_coord_down, n_try):
        assert shape.type == "Polygon"

        def fn(point):
            x, y = point
            return self._round(x), self._round(y)

        exterior = apply_map(fn, shape.exterior.coords)
        rings = None

        if len(shape.interiors) > 0:
            rings = [apply_map(fn, ring.coords) for ring in shape.interiors]

        sign = 1.0 if y_coord_down else -1.0
        oriented_shape = orient(Polygon(exterior, rings), sign=sign)
        oriented_shape = self.handle_shape_validity(oriented_shape, y_coord_down, n_try)
        return oriented_shape

    def _load_geometry(self, geometry_spec):
        if isinstance(geometry_spec, BaseGeometry):
            return geometry_spec

        if isinstance(geometry_spec, dict):
            return SimpleShape(geometry_spec["coordinates"], geometry_spec["type"])

        try:
            return load_wkb(geometry_spec)
        except Exception:
            try:
                return load_wkt(geometry_spec)
            except Exception:
                return None

    def addFeature(self, feature, shape, y_coord_down):
        geom_encoder = GeometryEncoder(y_coord_down, self.extents, self._round)
        geometry = geom_encoder.encode(shape)

        feature_type = self._get_feature_type(shape)
        if len(geometry) == 0:
            # Don't add geometry if it's too small
            return
        f = self.layer.features.add()

        fid = feature.get("id")
        if fid is not None:
            if isinstance(fid, Number) and fid >= 0:
                f.id = fid

        # properties
        properties = feature.get("properties")
        if properties is not None:
            self._handle_attr(self.layer, f, properties)

        f.type = feature_type
        f.geometry.extend(geometry)

    def _get_feature_type(self, shape):
        if shape.type == "Point" or shape.type == "MultiPoint":
            return self.tile.Point
        elif shape.type == "LineString" or shape.type == "MultiLineString":
            return self.tile.LineString
        elif shape.type == "Polygon" or shape.type == "MultiPolygon":
            return self.tile.Polygon
        elif shape.type == "GeometryCollection":
            raise ValueError("Encoding geometry collections not supported")
        else:
            raise ValueError(f"Cannot encode unknown geometry type: {shape.type}")

    def _chunker(self, seq, size):

        return [seq[pos : pos + size] for pos in range(0, len(seq), size)]

    def _can_handle_key(self, k):
        return isinstance(k, str)

    def _can_handle_val(self, v):
        if isinstance(v, str):
            return True
        elif isinstance(v, bool):
            return True
        elif isinstance(v, int):
            return True
        elif isinstance(v, float):
            return True

        return False

    def _can_handle_attr(self, k, v):
        return self._can_handle_key(k) and self._can_handle_val(v)

    def _handle_attr(self, layer, feature, props):
        for k, v in props.items():
            if self._can_handle_attr(k, v):
                if k not in self.seen_keys_idx:
                    layer.keys.append(k)
                    self.seen_keys_idx[k] = self.key_idx
                    self.key_idx += 1

                feature.tags.append(self.seen_keys_idx[k])

                if isinstance(v, bool):
                    values_idx = self.seen_values_bool_idx
                else:
                    values_idx = self.seen_values_idx

                if v not in values_idx:
                    values_idx[v] = self.val_idx
                    self.val_idx += 1

                    val = layer.values.add()
                    if isinstance(v, bool):
                        val.bool_value = v
                    elif isinstance(v, str):
                        val.string_value = v
                    elif isinstance(v, int):
                        val.int_value = v
                    elif isinstance(v, float):
                        val.double_value = v

                feature.tags.append(values_idx[v])

1	import decimal	5✔
2	from numbers import Number	5✔
3
4	from shapely.geometry.base import BaseGeometry	5✔
5	from shapely.geometry.multipolygon import MultiPolygon	5✔
6	from shapely.geometry.polygon import orient, Polygon	5✔
7	from shapely.ops import transform	5✔
8	from shapely.wkb import loads as load_wkb	5✔
9	from shapely.wkt import loads as load_wkt	5✔
10
11	from mapbox_vector_tile.Mapbox import vector_tile_pb2 as vector_tile	5✔
12	from mapbox_vector_tile.geom_encoder import GeometryEncoder	5✔
13	from mapbox_vector_tile.polygon import make_it_valid	5✔
14	from mapbox_vector_tile.simple_shape import SimpleShape	5✔
15
16
17	def apply_map(fn, x):	5✔
18	return list(map(fn, x))	5✔
19
20
21	def on_invalid_geometry_raise(shape):	5✔
22	raise ValueError(f"Invalid geometry: {shape.wkt}")	5✔
23
24
25	def on_invalid_geometry_ignore(shape):	5✔
26	return None	5✔
27
28
29	def on_invalid_geometry_make_valid(shape):	5✔
30	return make_it_valid(shape)	5✔
31
32
33	class VectorTile:	5✔
34	def __init__(	5✔
35	self, extents, on_invalid_geometry=None, max_geometry_validate_tries=5, round_fn=None, check_winding_order=True
36	):
37	self.tile = vector_tile.tile()	5✔
38	self.extents = extents	5✔
39	self.on_invalid_geometry = on_invalid_geometry	5✔
40	self.check_winding_order = check_winding_order	5✔
41	self.max_geometry_validate_tries = max_geometry_validate_tries	5✔
42	if round_fn:	5✔
43	self._round = round_fn	5✔
44	else:
45	self._round = self._round_quantize	5✔
46
47	def _round_quantize(self, val):	5✔
48
49	# round() has different behavior in python 2/3
50	# For consistency between 2 and 3 we use quantize, however
51	# it is slower than the built in round function.
52	d = decimal.Decimal(val)	5✔
53	rounded = d.quantize(1, rounding=decimal.ROUND_HALF_EVEN)	5✔
54	return float(rounded)	5✔
55
56	def addFeatures(self, features, layer_name="", quantize_bounds=None, y_coord_down=False):	5✔
57
58	self.layer = self.tile.layers.add()	5✔
59	self.layer.name = layer_name	5✔
60	self.layer.version = 1	5✔
61	self.layer.extent = self.extents	5✔
62
63	self.key_idx = 0	5✔
64	self.val_idx = 0	5✔
65	self.seen_keys_idx = {}	5✔
66	self.seen_values_idx = {}	5✔
67	self.seen_values_bool_idx = {}	5✔
68
69	for feature in features:	5✔
70
71	# skip missing or empty geometries
72	geometry_spec = feature.get("geometry")	5✔
73	if geometry_spec is None:	5✔
74	continue	×
75	shape = self._load_geometry(geometry_spec)	5✔
76
77	if shape is None:	5✔
78	raise NotImplementedError("Can't do geometries that are not wkt, wkb, or shapely geometries")	5✔
79
80	if shape.is_empty:	5✔
81	continue	×
82
83	if quantize_bounds:	5✔
84	shape = self.quantize(shape, quantize_bounds)	5✔
85	if self.check_winding_order:	5✔
86	shape = self.enforce_winding_order(shape, y_coord_down)	5✔
87
88	if shape is not None and not shape.is_empty:	5✔
89	self.addFeature(feature, shape, y_coord_down)	5✔
90
91	def enforce_winding_order(self, shape, y_coord_down, n_try=1):	5✔
92	if shape.type == "MultiPolygon":	5✔
93	# If we are a multipolygon, we need to ensure that the
94	# winding orders of the consituent polygons are
95	# correct. In particular, the winding order of the
96	# interior rings need to be the opposite of the
97	# exterior ones, and all interior rings need to follow
98	# the exterior one. This is how the end of one polygon
99	# and the beginning of another are signaled.
100	shape = self.enforce_multipolygon_winding_order(shape, y_coord_down, n_try)	5✔
101
102	elif shape.type == "Polygon":	5✔
103	# Ensure that polygons are also oriented with the
104	# appropriate winding order. Their exterior rings must
105	# have a clockwise order, which is translated into a
106	# clockwise order in MVT's tile-local coordinates with
107	# the Y axis in "screen" (i.e: +ve down) configuration.
108	# Note that while the Y axis flips, we also invert the
109	# Y coordinate to get the tile-local value, which means
110	# the clockwise orientation is unchanged.
111	shape = self.enforce_polygon_winding_order(shape, y_coord_down, n_try)	5✔
112
113	# other shapes just get passed through
114	return shape	5✔
115
116	def quantize(self, shape, bounds):	5✔
117	minx, miny, maxx, maxy = bounds	5✔
118
119	def fn(x, y, z=None):	5✔
120	xfac = self.extents / (maxx - minx)	5✔
121	yfac = self.extents / (maxy - miny)	5✔
122	x = xfac * (x - minx)	5✔
123	y = yfac * (y - miny)	5✔
124	return self._round(x), self._round(y)	5✔
125
126	return transform(fn, shape)	5✔
127
128	def handle_shape_validity(self, shape, y_coord_down, n_try):	5✔
129	if shape.is_valid:	5✔
130	return shape	5✔
131
132	if n_try >= self.max_geometry_validate_tries:	5✔
133	# ensure that we don't recurse indefinitely with an
134	# invalid geometry handler that doesn't validate
135	# geometries
136	return None	×
137
138	if self.on_invalid_geometry:	5✔
139	shape = self.on_invalid_geometry(shape)	5✔
140	if shape is not None and not shape.is_empty:	5✔
141	# this means that we have a handler that might have
142	# altered the geometry. We'll run through the process
143	# again, but keep track of which attempt we are on to
144	# terminate the recursion.
145	shape = self.enforce_winding_order(shape, y_coord_down, n_try + 1)	5✔
146
147	return shape	5✔
148
149	def enforce_multipolygon_winding_order(self, shape, y_coord_down, n_try):	5✔
150	assert shape.type == "MultiPolygon"	5✔
151
152	parts = []	5✔
153	for part in shape.geoms:	5✔
154	part = self.enforce_polygon_winding_order(part, y_coord_down, n_try)	5✔
155	if part is not None and not part.is_empty:	5✔
156	if part.geom_type == "MultiPolygon":	5✔
157	parts.extend(part.geoms)	5✔
158	else:
159	parts.append(part)	5✔
160
161	if not parts:	5✔
162	return None	×
163
164	if len(parts) == 1:	5✔
165	oriented_shape = parts[0]	5✔
166	else:
167	oriented_shape = MultiPolygon(parts)	5✔
168
169	oriented_shape = self.handle_shape_validity(oriented_shape, y_coord_down, n_try)	5✔
170	return oriented_shape	5✔
171
172	def enforce_polygon_winding_order(self, shape, y_coord_down, n_try):	5✔
173	assert shape.type == "Polygon"	5✔
174
175	def fn(point):	5✔
176	x, y = point	5✔
177	return self._round(x), self._round(y)	5✔
178
179	exterior = apply_map(fn, shape.exterior.coords)	5✔
180	rings = None	5✔
181
182	if len(shape.interiors) > 0:	5✔
183	rings = [apply_map(fn, ring.coords) for ring in shape.interiors]	5✔
184
185	sign = 1.0 if y_coord_down else -1.0	5✔
186	oriented_shape = orient(Polygon(exterior, rings), sign=sign)	5✔
187	oriented_shape = self.handle_shape_validity(oriented_shape, y_coord_down, n_try)	5✔
188	return oriented_shape	5✔
189
190	def _load_geometry(self, geometry_spec):	5✔
191	if isinstance(geometry_spec, BaseGeometry):	5✔
192	return geometry_spec	5✔
193
194	if isinstance(geometry_spec, dict):	5✔
195	return SimpleShape(geometry_spec["coordinates"], geometry_spec["type"])	5✔
196
197	try:	5✔
198	return load_wkb(geometry_spec)	5✔
199	except Exception:	5✔
200	try:	5✔
201	return load_wkt(geometry_spec)	5✔
202	except Exception:	5✔
203	return None	5✔
204
205	def addFeature(self, feature, shape, y_coord_down):	5✔
206	geom_encoder = GeometryEncoder(y_coord_down, self.extents, self._round)	5✔
207	geometry = geom_encoder.encode(shape)	5✔
208
209	feature_type = self._get_feature_type(shape)	5✔
210	if len(geometry) == 0:	5✔
211	# Don't add geometry if it's too small
212	return	5✔
213	f = self.layer.features.add()	5✔
214
215	fid = feature.get("id")	5✔
216	if fid is not None:	5✔
217	if isinstance(fid, Number) and fid >= 0:	5✔
218	f.id = fid	5✔
219
220	# properties
221	properties = feature.get("properties")	5✔
222	if properties is not None:	5✔
223	self._handle_attr(self.layer, f, properties)	5✔
224
225	f.type = feature_type	5✔
226	f.geometry.extend(geometry)	5✔
227
228	def _get_feature_type(self, shape):	5✔
229	if shape.type == "Point" or shape.type == "MultiPoint":	5✔
230	return self.tile.Point	5✔
231	elif shape.type == "LineString" or shape.type == "MultiLineString":	5✔
232	return self.tile.LineString	5✔
233	elif shape.type == "Polygon" or shape.type == "MultiPolygon":	5✔
234	return self.tile.Polygon	5✔
235	elif shape.type == "GeometryCollection":	5✔
236	raise ValueError("Encoding geometry collections not supported")	5✔
237	else:
NEW 238	raise ValueError(f"Cannot encode unknown geometry type: {shape.type}")	×
239
240	def _chunker(self, seq, size):	5✔
241
NEW 242	return [seq[pos : pos + size] for pos in range(0, len(seq), size)]	×
243
244	def _can_handle_key(self, k):	5✔
245	return isinstance(k, str)	5✔
246
247	def _can_handle_val(self, v):	5✔
248	if isinstance(v, str):	5✔
249	return True	5✔
250	elif isinstance(v, bool):	5✔
251	return True	5✔
252	elif isinstance(v, int):	5✔
253	return True	5✔
254	elif isinstance(v, float):	5✔
255	return True	5✔
256
257	return False	5✔
258
259	def _can_handle_attr(self, k, v):	5✔
260	return self._can_handle_key(k) and self._can_handle_val(v)	5✔
261
262	def _handle_attr(self, layer, feature, props):	5✔
263	for k, v in props.items():	5✔
264	if self._can_handle_attr(k, v):	5✔
265	if k not in self.seen_keys_idx:	5✔
266	layer.keys.append(k)	5✔
267	self.seen_keys_idx[k] = self.key_idx	5✔
268	self.key_idx += 1	5✔
269
270	feature.tags.append(self.seen_keys_idx[k])	5✔
271
272	if isinstance(v, bool):	5✔
273	values_idx = self.seen_values_bool_idx	5✔
274	else:
275	values_idx = self.seen_values_idx	5✔
276
277	if v not in values_idx:	5✔
278	values_idx[v] = self.val_idx	5✔
279	self.val_idx += 1	5✔
280
281	val = layer.values.add()	5✔
282	if isinstance(v, bool):	5✔
283	val.bool_value = v	5✔
284	elif isinstance(v, str):	5✔
285	val.string_value = v	5✔
286	elif isinstance(v, int):	5✔
287	val.int_value = v	5✔
288	elif isinstance(v, float):	5✔
289	val.double_value = v	5✔
290
291	feature.tags.append(values_idx[v])	5✔

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