Last active
January 20, 2021 09:46
-
-
Save thomaxxl/53256daf0c94b53832fd6cc1959a1451 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from flask import Flask | |
from flask_sqlalchemy import SQLAlchemy | |
from sqlalchemy import func | |
from geoalchemy2 import Geometry | |
from flask_sqlalchemy import SQLAlchemy | |
from safrs import SAFRSBase, SAFRSAPI | |
from flask.json import JSONEncoder | |
from flask import jsonify | |
import shapely_geojson | |
from safrs.json_encoder import SAFRSJSONEncoder | |
from geoalchemy2.functions import ST_AsGeoJSON | |
from geoalchemy2.elements import WKBElement, _SpatialElement | |
app = Flask(__name__) | |
db = SQLAlchemy() | |
class City(SAFRSBase, db.Model): | |
"""A city, including its geospatial data.""" | |
__tablename__ = "cities" | |
point_id = db.Column(db.Integer, primary_key=True, autoincrement=True) | |
location = db.Column(db.String(30)) | |
longitude = db.Column(db.Float) | |
latitude = db.Column(db.Float) | |
geo = db.Column(Geometry(geometry_type="POINT",srid=25833, dimension=2)) | |
def __init__(self, *args, **kwargs): | |
print(args, kwargs) | |
latitude = kwargs.get("latitude",0) | |
longitude = kwargs.get("longitude",0) | |
kwargs['geo'] = 'POINT({} {})'.format(longitude, latitude) | |
super().__init__(*args,**kwargs) | |
def __repr__(self): | |
return "<City {name} ({lat}, {lon})>".format( | |
name=self.location, lat=self.latitude, lon=self.longitude) | |
def __str__(self): | |
return self.__repr__() | |
def get_cities_within_radius(self, radius): | |
"""Return all cities within a given radius (in meters) of this city.""" | |
return City.query.filter(func.ST_Distance_Sphere(City.geo, self.geo) < radius).all() | |
@classmethod | |
def add_city(cls, location, longitude, latitude): | |
"""Put a new city in the database.""" | |
geo = 'POINT({} {})'.format(longitude, latitude) | |
city = City(location=location, | |
longitude=longitude, | |
latitude=latitude, | |
geo=geo) | |
db.session.add(city) | |
db.session.commit() | |
@classmethod | |
def update_geometries(cls): | |
"""Using each city's longitude and latitude, add geometry data to db.""" | |
cities = City.query.all() | |
for city in cities: | |
point = 'POINT({} {})'.format(city.longitude, city.latitude) | |
city.geo = point | |
db.session.commit() | |
def connect_to_db(app): | |
"""Connect the database to Flask app.""" | |
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgres:///testgis' | |
app.config['SQLALCHEMY_ECHO'] = False | |
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False | |
db.app = app | |
db.init_app(app) | |
def create_api(app, HOST="localhost", PORT=5000, API_PREFIX=""): | |
api = SAFRSAPI(app, host=HOST, port=PORT, prefix=API_PREFIX) | |
app.json_encoder = GeoJSONEncoder | |
api.expose_object(City) | |
print("Starting API: http://{}:{}/{}".format(HOST, PORT, API_PREFIX)) | |
class GeoJSONEncoder(SAFRSJSONEncoder): | |
def default(self, obj, **kwargs): | |
if isinstance(obj, _SpatialElement): | |
return str(obj) | |
return super().default(obj, **kwargs) | |
if __name__ == "__main__": | |
connect_to_db(app) | |
db.create_all() | |
host = '192.168.235.136' | |
with app.app_context(): | |
print(jsonify({})) | |
create_api(app, host) | |
for city in City.query.all(): | |
print(city, city.location) | |
sf = db.session.query(City).filter(City.location == 'San Francisco').one() | |
app.run(host=host) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
👍