Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def from_geojson_feature(cls, task_id, task_feature):
"""
Constructs and validates a task from a GeoJson feature object
:param task_id: Unique ID for the task
:param task_feature: A geoJSON feature object
:raises InvalidGeoJson, InvalidData
"""
if type(task_feature) is not geojson.Feature:
raise InvalidGeoJson('Task: Invalid GeoJson should be a feature')
task_geometry = task_feature.geometry
if type(task_geometry) is not geojson.MultiPolygon:
raise InvalidGeoJson('Task: Geometry must be a MultiPolygon')
is_valid_geojson = geojson.is_valid(task_geometry)
if is_valid_geojson['valid'] == 'no':
raise InvalidGeoJson(f"Task: Invalid MultiPolygon - {is_valid_geojson['message']}")
task = cls()
try:
task.x = task_feature.properties['x']
task.y = task_feature.properties['y']
task.zoom = task_feature.properties['zoom']
task.splittable = task_feature.properties['splittable']
except KeyError as e:
raise InvalidData(f'Task: Expected property not found: {str(e)}')
if 'extra_properties' in task_feature.properties:
task.extra_properties = json.dumps(
task_feature.properties['extra_properties'])
def _attach_tasks_to_project(draft_project: Project, tasks_geojson):
"""
Validates then iterates over the array of tasks and attach them to the draft project
:param draft_project: Draft project in scope
:param tasks_geojson: GeoJSON feature collection of mapping tasks
:raises InvalidGeoJson, InvalidData
"""
tasks = geojson.loads(json.dumps(tasks_geojson))
if type(tasks) is not geojson.FeatureCollection:
raise InvalidGeoJson('Tasks: Invalid GeoJson must be FeatureCollection')
is_valid_geojson = geojson.is_valid(tasks)
if is_valid_geojson['valid'] == 'no':
raise InvalidGeoJson(f"Tasks: Invalid FeatureCollection - {is_valid_geojson['message']}")
task_count = 1
for feature in tasks['features']:
try:
task = Task.from_geojson_feature(task_count, feature)
except (InvalidData, InvalidGeoJson) as e:
raise e
draft_project.tasks.append(task)
task_count += 1
task_count -= 1 # Remove last increment before falling out loop
draft_project.total_tasks = task_count
def _index_spatial_property(self, node, key, value, label):
sidx_key = u"{}_{}".format(label, key)
if sidx_key in self.sidx:
geo_value = geojson.loads(value)
is_valid_geojson = geojson.is_valid(geo_value)['valid'] != 'no'
if is_valid_geojson:
# Add node to index
index = self.sidx[sidx_key]["index"]
index_key = self.sidx[sidx_key]["key"]
wkt_value = shape(geo_value).wkt
node[index_key] = wkt_value
index.add(SPATIAL_INDEX_KEY, SPATIAL_INDEX_VALUE, node)
# Add node to layer
self.spatial.addNodeToLayer(layer=index.name, node=node.url)
:param dissolve: flag for wther to to dissolve internal boundaries.
:return: geojson.MultiPolygon
"""
parsed_geojson = GridService._to_shapely_geometries(
json.dumps(feature_collection)
)
multi_polygon = GridService._convert_to_multipolygon(parsed_geojson)
if dissolve:
multi_polygon = GridService._dissolve(multi_polygon)
aoi_multi_polygon_geojson = geojson.loads(json.dumps(mapping(multi_polygon)))
# validate the geometry
if type(aoi_multi_polygon_geojson) is not geojson.MultiPolygon:
raise InvalidGeoJson("Area Of Interest: geometry must be a MultiPolygon")
is_valid_geojson = geojson.is_valid(aoi_multi_polygon_geojson)
if is_valid_geojson["valid"] == "no":
raise InvalidGeoJson(
f"Area of Interest: Invalid MultiPolygon - {is_valid_geojson['message']}"
)
return aoi_multi_polygon_geojson
Merge all geometries to a single multipolygon
:param feature_collection: geojson feature collection str containing features
:param dissolve: flag for wther to to dissolve internal boundaries.
:return: geojson.MultiPolygon
"""
parsed_geojson = GridService._to_shapely_geometries(json.dumps(feature_collection))
multi_polygon = GridService._convert_to_multipolygon(parsed_geojson)
if dissolve:
multi_polygon = GridService._dissolve(multi_polygon)
aoi_multi_polygon_geojson = geojson.loads(json.dumps(mapping(multi_polygon)))
# validate the geometry
if type(aoi_multi_polygon_geojson) is not geojson.MultiPolygon:
raise InvalidGeoJson('Area Of Interest: geometry must be a MultiPolygon')
is_valid_geojson = geojson.is_valid(aoi_multi_polygon_geojson)
if is_valid_geojson['valid'] == 'no':
raise InvalidGeoJson(f"Area of Interest: Invalid MultiPolygon - {is_valid_geojson['message']}")
return aoi_multi_polygon_geojson
@app.route('/save', methods=['POST'])
def geojson_save():
try:
g = request.form['geojson']
f = geojson.loads(g)
except Exception, e:
err = "failed to load geojson, because %s" % e
return jsonify(ok=0, error=err)
# Does the input pass the smell check?
validation = geojson.is_valid(f)
if (validation['valid'] == 'no'):
error = "GeoJSON doesn't smell right: %s" % validation['message']
logging.error(error)
return jsonify(ok=0, error=error)
try:
ff = mapzen.whosonfirst.export.flatfile(root, debug=False)
path = ff.export_feature(f)
except Exception, e:
error = "failed to export geojson, because %s" % e
logging.error(error)
return jsonify(ok=0, error=error)
# Repeat back the file we just wrote