Skip to content

Instantly share code, notes, and snippets.

@smcveigh-phunware
Last active January 20, 2022 02:06
Show Gist options
  • Save smcveigh-phunware/ad055e47abfa35394c9bd4e7aa17af61 to your computer and use it in GitHub Desktop.
Save smcveigh-phunware/ad055e47abfa35394c9bd4e7aa17af61 to your computer and use it in GitHub Desktop.
Test Shapely with MongoDB GeoJSON examples
#!/usr/bin/env python
# -*- coding: utf-8 -*
import pytest
import typing as typ
import pymongo as pm
import mongoengine as me
import shapely.geometry as geo
GEOMETRIES = [
pytest.param(
{
'type': "Point",
'coordinates': [40.0, 5.0]
},
id="a GeoJSON Point"
),
pytest.param(
{
'type': "LineString",
'coordinates': [[40.0, 5.0], [41.0, 6.0]]
},
id="a GeoJSON LineString"
),
pytest.param(
{
'type': "Polygon",
'coordinates': [
[[0.0, 0.0], [3.0, 6.0], [6.0, 1.0], [0.0, 0.0]]
]
},
id="a GeoJSON Polygon with a single ring"
),
pytest.param(
{
'type': "Polygon",
'coordinates': [
[[0.0, 0.0], [3.0, 6.0], [6.0, 1.0], [0.0, 0.0]],
[[2.0, 2.0], [3.0, 3.0], [4.0, 2.0], [2.0, 2.0]]
]
},
id="a GeoJSON Polygon with multiple rings"
),
pytest.param(
{
'type': "MultiPoint",
'coordinates': [
[-73.9580, 40.8003],
[-73.9498, 40.7968],
[-73.9737, 40.7648],
[-73.9814, 40.7681]
]
},
id="a GeoJSON MultiPoint"
),
pytest.param(
{
'type': "MultiLineString",
'coordinates': [
[[-73.96943, 40.78519], [-73.96082, 40.78095]],
[[-73.96415, 40.79229], [-73.95544, 40.78854]],
[[-73.97162, 40.78205], [-73.96374, 40.77715]],
[[-73.97880, 40.77247], [-73.97036, 40.76811]]
]
},
id="a GeoJSON MultiLineString"
),
pytest.param(
{
'type': "MultiPolygon",
'coordinates': [
[[[-73.958, 40.8003], [-73.9498, 40.7968], [-73.9737, 40.7648], [-73.9814, 40.7681], [-73.958, 40.8003]]],
[[[-73.958, 40.8003], [-73.9498, 40.7968], [-73.9737, 40.7648], [-73.958, 40.8003]]]
]
},
id="a GeoJSON MultiPolygon",
marks=pytest.mark.xfail(
raises=AssertionError,
reason="MultiPolygon has polygon intersection"
)
),
pytest.param(
{
'type': "GeometryCollection",
'geometries': [
{
'type': "MultiPoint",
'coordinates': [
[-73.9580, 40.8003],
[-73.9498, 40.7968],
[-73.9737, 40.7648],
[-73.9814, 40.7681]
]
},
{
'type': "MultiLineString",
'coordinates': [
[[-73.96943, 40.78519], [-73.96082, 40.78095]],
[[-73.96415, 40.79229], [-73.95544, 40.78854]],
[[-73.97162, 40.78205], [-73.96374, 40.77715]],
[[-73.97880, 40.77247], [-73.97036, 40.76811]]
]
}
]
},
id="a GeoJSON GeometryCollection",
marks=pytest.mark.xfail(
raises=AttributeError,
reason="MongoEngine has no a GeometryCollectionField"
)
)
]
GeoJSON = typ.Dict[str, typ.Any]
@pytest.fixture
def query() -> GeoJSON:
return {
'type': "Polygon",
'coordinates': [
[[-179, -89], [180, -89], [180, 90], [-179, 90], [-179, -89]]
]
}
@pytest.mark.parametrize('value', GEOMETRIES)
def test_shape_from_geojson(value: GeoJSON, query: GeoJSON):
"""
Using sample GeoJSON-like dictionaries from MongoDB GeoJSON examples_,
test, using Shapley_, by:
- interpret GeoJSON data as Shapely geometry
- assert geometry conforms to OpenGIS standards_
- convert Shapely geometry back to GeoJSON data
- assert original and converted GeoJSON consistent
- retrieve the MongoEngine field type matching the geometry
- connect to MongoDB and register 'tmpdb' with MongoEngine
- drop document collection if it exists
- define a document with matching GeoJSON field type
- store the mapped GeoJSON data in the field values
- validate the document
- save the document to MongoDB
- assert collection exists after save
- assert saved field value matches GeoJSON data
- query collection with huge Polygon
- assert document found from query
- drop document collection
- assert document collection does not exist
.. _examples: https://docs.mongodb.com/manual/reference/geojson/
.. _Shapley: http://toblerity.org/shapely/
.. _standards: http://www.opengeospatial.org/standards/sfa
:param value: GeoJSON feature to store
:param query: GeoJSON Polygon to query with
"""
shape = geo.shape(value)
assert shape.is_valid, "shape failed OpenGIS validation"
mapped = geo.mapping(shape)
# NOTE 'mapping` uses tuples, BSON/JSON uses lists
# assert mapped == value, "mapped value does not match original"
field = getattr(me, "{:s}Field".format(value['type']))
cc = me.connect('tmpdb') # type: pm.MongoClient
db = cc.get_database('tmpdb')
if 'doc' in db.list_collection_names():
db.drop_collection('doc')
assert 'doc' not in db.list_collection_names(), "collection should not exist after drop"
class Doc(me.Document):
value = field()
query = me.PolygonField()
doc = Doc()
doc.value = mapped
doc.query = query
doc.validate()
doc.save()
assert 'doc' in db.list_collection_names(), "expected collection to exist after save"
assert Doc.objects[0].value == value, "value from mongo does not match original"
qry = Doc.objects(value__geo_within=doc.query) # type: me.QuerySet
assert qry.first() is not None, "expected successful query"
Doc.drop_collection()
assert 'doc' not in db.list_collection_names(), "failed to drop collection"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment