Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_latest_predictions_in_bbox(model_id: int, version_id: int, bbox: list):
"""
Fetch latest predictions for the specified model intersecting
the given bbox
:param model_id, version_id, bbox
:return list of predictions
"""
query = db.session.query(Prediction.id, Prediction.created, Prediction.dockerhub_hash, ST_AsGeoJSON(ST_Envelope(Prediction.bbox)).label('bbox'),
Prediction.model_id, Prediction.tile_zoom, Prediction.version_id).filter(Prediction.model_id == model_id).filter(
Prediction.version_id == version_id).filter(ST_Intersects(
Prediction.bbox, ST_MakeEnvelope(bbox[0], bbox[1], bbox[2], bbox[3], 4326))).order_by(
Prediction.created.desc()).limit(1)
return query.all()
if tablename in seen_classes:
cls = current_app.class_references[tablename]
else:
db.metadata.reflect(bind=db.engine)
seen_classes.add(tablename)
cls = type(str(tablename), (GeoPoly, db.Model,), {'__tablename__':tablename,
'__table_args__' : {'extend_existing': True}})
current_app.class_references[tablename] = cls
if field == config.geom_column:
vector = cls.query.with_entities(geofuncs.ST_AsGeoJSON(getattr(cls, field))).all()
response['data'] = [v[0] for v in vector]
elif field == 'geojson':
#TODO: How can this be cleaner? Do I need 2 queries go get geojson?
#rows = cls.query.all()
geoms = cls.query.with_entities(geofuncs.ST_AsGeoJSON(getattr(cls, config.geom_column))).all()
features = []
for i, row in enumerate(geoms):
#attributes = row.as_dict()
#attributes.pop('wkb_geometry', None)
#for k, v in attributes.iteritems():
#if isinstance(v, decimal.Decimal):
#attributes[k] = float(v)
current_feature = {'type':'Feature',
'geometry':ast.literal_eval(geoms[i][0])}
#'properties':attributes}
features.append(current_feature)
geojson = {"type": "FeatureCollection","features": features}
#geojson = {"type":"FeatureCollection", "features": geoms}
response['data']['geojson'] = geojson
elif field == 'topojson':
#TODO: Add topojson support if the DB is postgresql
def get(prediction_id: int):
"""
Get prediction with the given ID
:param prediction_id
:return prediction if found otherwise None
"""
query = db.session.query(Prediction.id, Prediction.created, Prediction.dockerhub_hash,
ST_AsGeoJSON(ST_Envelope(Prediction.bbox)).label('bbox'), Prediction.model_id, Prediction.tile_zoom,
Prediction.version_id).filter(Prediction.id == prediction_id)
return query.one()
def get_predictions_by_model(model_id: int):
"""
Gets predictions for a specified ML Model
:param model_id: ml model ID in scope
:return predictions if found otherwise None
"""
query = db.session.query(Prediction.id, Prediction.created, Prediction.dockerhub_hash,
ST_AsGeoJSON(ST_Envelope(Prediction.bbox)).label('bbox'), Prediction.model_id, Prediction.tile_zoom,
Prediction.version_id).filter(Prediction.model_id == model_id)
return query.all()
def get_all_predictions_in_bbox(model_id: int, bbox: list):
"""
Fetch all predictions for the specified model intersecting the given
bbox
:param model_id, bbox
:return list of predictions
"""
query = db.session.query(Prediction.id, Prediction.created, Prediction.dockerhub_hash, ST_AsGeoJSON(ST_Envelope(Prediction.bbox)).label('bbox'),
Prediction.model_id, Prediction.tile_zoom, Prediction.version_id).filter(
Prediction.model_id == model_id).filter(
ST_Intersects(Prediction.bbox, ST_MakeEnvelope(bbox[0], bbox[1], bbox[2], bbox[3], 4326)))
return query.all()