Add types and docstrings.

This commit is contained in:
Edward Betts 2023-10-10 10:11:23 +01:00
parent 5c01d9aebf
commit e975e86af5
6 changed files with 181 additions and 87 deletions

View file

@ -1,3 +1,6 @@
"""Reverse geocode for Wikidata and Wikimedia Commons."""
headers = {"User-Agent": "UK gecode/0.1 (edward@4angle.com)"} headers = {"User-Agent": "UK gecode/0.1 (edward@4angle.com)"}
samples = [ samples = [

View file

@ -1,25 +1,31 @@
import flask
import sqlalchemy
from sqlalchemy import create_engine, func from sqlalchemy import create_engine, func
from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.orm import scoped_session, sessionmaker
session = scoped_session(sessionmaker()) session = scoped_session(sessionmaker())
def init_db(db_url): def init_db(db_url: str, echo: bool = False) -> None:
session.configure(bind=get_engine(db_url)) """Initialise databsae."""
session.configure(bind=get_engine(db_url, echo=echo))
def get_engine(db_url, echo=False): def get_engine(db_url: str, echo: bool = False) -> sqlalchemy.engine.base.Engine:
"""Create an engine object."""
return create_engine(db_url, pool_recycle=3600, echo=echo) return create_engine(db_url, pool_recycle=3600, echo=echo)
def init_app(app, echo=False): def init_app(app: flask.app.Flask, echo: bool = False) -> None:
"""Initialise database connection within flask app."""
db_url = app.config["DB_URL"] db_url = app.config["DB_URL"]
session.configure(bind=get_engine(db_url, echo=echo)) session.configure(bind=get_engine(db_url, echo=echo))
@app.teardown_appcontext @app.teardown_appcontext
def shutdown_session(exception=None): def shutdown_session(exception: Exception | None = None) -> None:
session.remove() session.remove()
def now_utc(): def now_utc():
"""Time in UTC via SQL."""
return func.timezone("utc", func.now()) return func.timezone("utc", func.now())

View file

@ -1,12 +1,15 @@
from sqlalchemy.ext.declarative import declarative_base """Database model."""
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, Float, Numeric, String
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import column_property
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import func, cast
import sqlalchemy
from geoalchemy2 import Geometry from geoalchemy2 import Geometry
from sqlalchemy import cast, func
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import column_property
from sqlalchemy.schema import Column
from sqlalchemy.types import Float, Integer, Numeric, String
from .database import session from .database import session
Base = declarative_base() Base = declarative_base()
@ -14,6 +17,8 @@ Base.query = session.query_property()
class Polygon(Base): class Polygon(Base):
"""Polygon."""
__tablename__ = "planet_osm_polygon" __tablename__ = "planet_osm_polygon"
osm_id = Column(Integer, primary_key=True, autoincrement=False) osm_id = Column(Integer, primary_key=True, autoincrement=False)
@ -25,25 +30,33 @@ class Polygon(Base):
area = column_property(func.ST_Area(way, False)) area = column_property(func.ST_Area(way, False))
@property @property
def osm_url(self): def osm_url(self) -> str:
"""OSM URL for polygon."""
osm_type = "way" if self.osm_id > 0 else "relation" osm_type = "way" if self.osm_id > 0 else "relation"
return f"https://www.openstreetmap.org/{osm_type}/{abs(self.osm_id)}" return f"https://www.openstreetmap.org/{osm_type}/{abs(self.osm_id)}"
@hybrid_property @hybrid_property
def area_in_sq_km(self): def area_in_sq_km(self) -> float:
return self.area / (1000 * 1000) """Area in square kilometers."""
return float(self.area) / (1000 * 1000)
@classmethod @classmethod
def coords_within(cls, lat, lon): def coords_within(
cls, lat: str | float, lon: str | float
) -> sqlalchemy.orm.query.Query["Polygon"]:
"""Polygons that contain given coordinates."""
point = func.ST_SetSRID(func.ST_MakePoint(lon, lat), 4326) point = func.ST_SetSRID(func.ST_MakePoint(lon, lat), 4326)
return cls.query.filter( q: sqlalchemy.orm.query.Query["Polygon"] = cls.query.filter(
cls.admin_level.isnot(None), cls.admin_level.isnot(None),
cls.admin_level.regexp_match("^\d+$"), cls.admin_level.regexp_match("^\d+$"),
func.ST_Within(point, cls.way), func.ST_Within(point, cls.way),
).order_by(cls.area, cast(cls.admin_level, Integer).desc()) ).order_by(cls.area, cast(cls.admin_level, Integer).desc())
return q
class Scotland(Base): class Scotland(Base):
"""Civil parishes in Scotland."""
__tablename__ = "scotland" __tablename__ = "scotland"
gid = Column(Integer, primary_key=True) gid = Column(Integer, primary_key=True)

View file

@ -1,8 +1,11 @@
from flask import current_app """Reverse geocode civil parishes in Scotland."""
import psycopg2 import psycopg2
from flask import current_app
def get_scotland_code(lat, lon): def get_scotland_code(lat: str | float, lon: str | float) -> str | None:
"""Find civil parish in Scotland for given lat/lon."""
conn = psycopg2.connect(**current_app.config["DB_PARAMS"]) conn = psycopg2.connect(**current_app.config["DB_PARAMS"])
cur = conn.cursor() cur = conn.cursor()

View file

@ -1,41 +1,54 @@
from flask import render_template """Wikidata API functions."""
import requests
import simplejson import typing
from . import headers
import urllib.parse import urllib.parse
import requests
import simplejson.errors
from flask import render_template
from . import headers
wikidata_query_api_url = "https://query.wikidata.org/bigdata/namespace/wdq/sparql" wikidata_query_api_url = "https://query.wikidata.org/bigdata/namespace/wdq/sparql"
wd_entity = "http://www.wikidata.org/entity/Q" wd_entity = "http://www.wikidata.org/entity/Q"
commons_cat_start = "https://commons.wikimedia.org/wiki/Category:" commons_cat_start = "https://commons.wikimedia.org/wiki/Category:"
class QueryError(Exception): class QueryError(Exception):
def __init__(self, query, r): """Query error."""
def __init__(self, query: str, r: requests.Response):
"""Init."""
self.query = query self.query = query
self.r = r self.r = r
def api_call(params): def api_call(params: dict[str, str | int]) -> dict[str, typing.Any]:
return requests.get( """Wikidata API call."""
"https://www.wikidata.org/w/api.php", api_params: dict[str, str | int] = {"format": "json", "formatversion": 2, **params}
params={"format": "json", "formatversion": 2, **params}, r = requests.get(
headers=headers, "https://www.wikidata.org/w/api.php", params=api_params, headers=headers
).json() )
return typing.cast(dict[str, typing.Any], r.json())
def get_entity(qid): def get_entity(qid: str) -> dict[str, typing.Any] | None:
"""Get Wikidata entity."""
json_data = api_call({"action": "wbgetentities", "ids": qid}) json_data = api_call({"action": "wbgetentities", "ids": qid})
try: try:
entity = list(json_data["entities"].values())[0] entity: dict[str, typing.Any] = list(json_data["entities"].values())[0]
except KeyError: except KeyError:
return return None
if "missing" not in entity: return entity if "missing" not in entity else None
return entity
def qid_to_commons_category(qid): def qid_to_commons_category(qid: str) -> str | None:
"""Commons category for a given Wikidata item."""
entity = get_entity(qid) entity = get_entity(qid)
if not entity:
return None
commons_cat: str | None
try: try:
commons_cat = entity["claims"]["P373"][0]["mainsnak"]["datavalue"]["value"] commons_cat = entity["claims"]["P373"][0]["mainsnak"]["datavalue"]["value"]
except Exception: except Exception:
@ -44,29 +57,36 @@ def qid_to_commons_category(qid):
return commons_cat return commons_cat
def wdqs(query): Row = dict[str, dict[str, typing.Any]]
def wdqs(query: str) -> list[Row]:
"""Pass query to the Wikidata Query Service."""
r = requests.post( r = requests.post(
wikidata_query_api_url, data={"query": query, "format": "json"}, headers=headers wikidata_query_api_url, data={"query": query, "format": "json"}, headers=headers
) )
try: try:
return r.json()["results"]["bindings"] return typing.cast(list[Row], r.json()["results"]["bindings"])
except simplejson.errors.JSONDecodeError: except simplejson.errors.JSONDecodeError:
raise QueryError(query, r) raise QueryError(query, r)
def wd_to_qid(wd): def wd_to_qid(wd: dict[str, str]) -> str:
"""Convert Wikidata URL from WDQS to QID."""
# expecting {"type": "url", "value": "https://www.wikidata.org/wiki/Q30"} # expecting {"type": "url", "value": "https://www.wikidata.org/wiki/Q30"}
if wd["type"] == "uri": assert wd["type"] == "uri"
return wd_uri_to_qid(wd["value"]) return wd_uri_to_qid(wd["value"])
def wd_uri_to_qid(value): def wd_uri_to_qid(value: str) -> str:
"""Convert URL like https://www.wikidata.org/wiki/Q30 to QID."""
assert value.startswith(wd_entity) assert value.startswith(wd_entity)
return value[len(wd_entity) - 1 :] return value[len(wd_entity) - 1 :]
def geosearch_query(lat, lon): def geosearch_query(lat: str | float, lon: str | float) -> list[Row]:
"""Geosearch via WDQS."""
if isinstance(lat, float): if isinstance(lat, float):
lat = f"{lat:f}" lat = f"{lat:f}"
if isinstance(lon, float): if isinstance(lon, float):
@ -76,7 +96,8 @@ def geosearch_query(lat, lon):
return wdqs(query) return wdqs(query)
def geosearch(lat, lon): def geosearch(lat: str | float, lon: str | float) -> Row | None:
"""Geosearch."""
default_max_dist = 1 default_max_dist = 1
rows = geosearch_query(lat, lon) rows = geosearch_query(lat, lon)
max_dist = { max_dist = {
@ -105,30 +126,37 @@ def geosearch(lat, lon):
break break
return row return row
return None
def lookup_scottish_parish_in_wikidata(code): def lookup_scottish_parish_in_wikidata(code: str) -> list[Row]:
query = render_template("sparql/scottish_parish.sparql", code=code) """Lookup scottish parish in Wikidata."""
return wdqs(query) return wdqs(render_template("sparql/scottish_parish.sparql", code=code))
def lookup_gss_in_wikidata(gss): def lookup_gss_in_wikidata(gss: str) -> list[Row]:
query = render_template("sparql/lookup_gss.sparql", gss=gss) """Lookup GSS in Wikidata."""
return wdqs(query) return wdqs(render_template("sparql/lookup_gss.sparql", gss=gss))
def lookup_wikidata_by_name(name, lat, lon): def lookup_wikidata_by_name(name: str, lat: float | str, lon: float | str) -> list[Row]:
"""Lookup place in Wikidata by name."""
query = render_template( query = render_template(
"sparql/lookup_by_name.sparql", name=repr(name), lat=str(lat), lon=str(lon) "sparql/lookup_by_name.sparql", name=repr(name), lat=str(lat), lon=str(lon)
) )
return wdqs(query) return wdqs(query)
def unescape_title(t): def unescape_title(t: str) -> str:
"""Unescape article title."""
return urllib.parse.unquote(t.replace("_", " ")) return urllib.parse.unquote(t.replace("_", " "))
def commons_from_rows(rows): Hit = dict[str, str | None]
def commons_from_rows(rows: list[Row]) -> Hit | None:
"""Commons from rows."""
for row in rows: for row in rows:
if "commonsCat" in row: if "commonsCat" in row:
qid = wd_to_qid(row["item"]) qid = wd_to_qid(row["item"])
@ -138,22 +166,28 @@ def commons_from_rows(rows):
qid = wd_to_qid(row["item"]) qid = wd_to_qid(row["item"])
cat = unescape_title(site_link[len(commons_cat_start) :]) cat = unescape_title(site_link[len(commons_cat_start) :])
return {"wikidata": qid, "commons_cat": cat} return {"wikidata": qid, "commons_cat": cat}
return None
def get_commons_cat_from_gss(gss): def get_commons_cat_from_gss(gss: str) -> Hit | None:
"""Get commons from GSS via Wikidata."""
return commons_from_rows(lookup_gss_in_wikidata(gss)) return commons_from_rows(lookup_gss_in_wikidata(gss))
def build_dict(hit, lat, lon): WikidataDict = dict[str, None | bool | str | int | dict[str, typing.Any]]
def build_dict(hit: Hit | None, lat: str | float, lon: str | float) -> WikidataDict:
"""Build dict."""
coords = {"lat": lat, "lon": lon} coords = {"lat": lat, "lon": lon}
if hit is None: if hit is None:
return dict(commons_cat=None, missing=True, coords=coords) return {"commons_cat": None, "missing": True, "coords": coords}
commons_cat = hit["commons_cat"] commons_cat = hit["commons_cat"]
ret = dict( ret: WikidataDict = {
coords=coords, "coords": coords,
admin_level=hit.get("admin_level"), "admin_level": hit.get("admin_level"),
wikidata=hit["wikidata"], "wikidata": hit["wikidata"],
) }
if not commons_cat: if not commons_cat:
return ret return ret

View file

@ -1,9 +1,14 @@
#!/usr/bin/python3 #!/usr/bin/python3
from flask import Flask, render_template, request, jsonify, redirect, url_for
from geocode import wikidata, scotland, database, model
import geocode
import random import random
import typing
import sqlalchemy
from flask import Flask, jsonify, redirect, render_template, request, url_for
from werkzeug.wrappers import Response
import geocode
from geocode import database, model, scotland, wikidata
city_of_london_qid = "Q23311" city_of_london_qid = "Q23311"
app = Flask(__name__) app = Flask(__name__)
@ -11,8 +16,8 @@ app.config.from_object("config.default")
database.init_app(app) database.init_app(app)
def get_random_lat_lon(): def get_random_lat_lon() -> tuple[float, float]:
"""Select random lat/lon within the UK""" """Select random lat/lon within the UK."""
south, east = 50.8520, 0.3536 south, east = 50.8520, 0.3536
north, west = 53.7984, -2.7296 north, west = 53.7984, -2.7296
@ -23,7 +28,12 @@ def get_random_lat_lon():
return lat, lon return lat, lon
def do_lookup(elements, lat, lon): Elements = sqlalchemy.orm.query.Query[model.Polygon]
def do_lookup(
elements: Elements, lat: str | float, lon: str | float
) -> wikidata.WikidataDict:
try: try:
hit = osm_lookup(elements, lat, lon) hit = osm_lookup(elements, lat, lon)
except wikidata.QueryError as e: except wikidata.QueryError as e:
@ -36,9 +46,10 @@ def do_lookup(elements, lat, lon):
return wikidata.build_dict(hit, lat, lon) return wikidata.build_dict(hit, lat, lon)
def lat_lon_to_wikidata(lat, lon): def lat_lon_to_wikidata(lat: str | float, lon: str | float) -> dict[str, typing.Any]:
scotland_code = scotland.get_scotland_code(lat, lon) scotland_code = scotland.get_scotland_code(lat, lon)
elements: typing.Any
if scotland_code: if scotland_code:
rows = wikidata.lookup_scottish_parish_in_wikidata(scotland_code) rows = wikidata.lookup_scottish_parish_in_wikidata(scotland_code)
hit = wikidata.commons_from_rows(rows) hit = wikidata.commons_from_rows(rows)
@ -55,6 +66,7 @@ def lat_lon_to_wikidata(lat, lon):
return {"elements": elements, "result": result} return {"elements": elements, "result": result}
admin_level = result.get("admin_level") admin_level = result.get("admin_level")
assert isinstance(admin_level, int)
if not admin_level or admin_level >= 7: if not admin_level or admin_level >= 7:
return {"elements": elements, "result": result} return {"elements": elements, "result": result}
@ -68,11 +80,17 @@ def lat_lon_to_wikidata(lat, lon):
return {"elements": elements, "result": result} return {"elements": elements, "result": result}
def osm_lookup(elements, lat, lon): def osm_lookup(
elements: Elements, lat: str | float, lon: str | float
) -> wikidata.Hit | None:
"""OSM lookup."""
ret: wikidata.Hit | None
for e in elements: for e in elements:
tags = e.tags assert isinstance(e, model.Polygon)
assert e.tags
tags: typing.Mapping[str, typing.Any] = e.tags
admin_level_tag = tags.get("admin_level") admin_level_tag = tags.get("admin_level")
admin_level = ( admin_level: int | None = (
int(admin_level_tag) int(admin_level_tag)
if admin_level_tag and admin_level_tag.isdigit() if admin_level_tag and admin_level_tag.isdigit()
else None else None
@ -108,24 +126,36 @@ def osm_lookup(elements, lat, lon):
ret["admin_level"] = admin_level ret["admin_level"] = admin_level
return ret return ret
has_wikidata_tag = [e.tags for e in elements if e.tags.get("wikidata")] has_wikidata_tag = [
e.tags for e in elements if e.tags.get("wikidata") # type: ignore
]
if len(has_wikidata_tag) != 1: if len(has_wikidata_tag) != 1:
return return None
assert has_wikidata_tag[0]
qid = has_wikidata_tag[0]["wikidata"] qid = has_wikidata_tag[0]["wikidata"]
return { return typing.cast(
wikidata.Hit,
{
"wikidata": qid, "wikidata": qid,
"commons_cat": wikidata.qid_to_commons_category(qid), "commons_cat": wikidata.qid_to_commons_category(qid),
"admin_level": admin_level, "admin_level": admin_level,
} },
)
def redirect_to_detail(q: str) -> Response:
"""Redirect to detail page."""
lat, lon = [v.strip() for v in q.split(",", 1)]
return redirect(url_for("detail_page", lat=lat, lon=lon))
@app.route("/") @app.route("/")
def index(): def index() -> str | Response:
"""Index page."""
q = request.args.get("q") q = request.args.get("q")
if q and q.strip(): if q and q.strip():
lat, lon = [v.strip() for v in q.split(",", 1)] return redirect_to_detail(q)
return redirect(url_for("detail_page", lat=lat, lon=lon))
lat, lon = request.args.get("lat"), request.args.get("lon") lat, lon = request.args.get("lat"), request.args.get("lon")
@ -137,7 +167,8 @@ def index():
@app.route("/random") @app.route("/random")
def random_location(): def random_location() -> str:
"""Return detail page for random lat/lon."""
lat, lon = get_random_lat_lon() lat, lon = get_random_lat_lon()
elements = model.Polygon.coords_within(lat, lon) elements = model.Polygon.coords_within(lat, lon)
@ -149,12 +180,14 @@ def random_location():
@app.route("/wikidata_tag") @app.route("/wikidata_tag")
def wikidata_tag(): def wikidata_tag() -> str:
lat = float(request.args.get("lat")) """Lookup Wikidata tag for lat/lon."""
lon = float(request.args.get("lon")) lat_str, lon_str = request.args["lat"], request.args["lon"]
lat, lon = float(lat_str), float(lon_str)
scotland_code = scotland.get_scotland_code(lat, lon) scotland_code = scotland.get_scotland_code(lat, lon)
elements: typing.Any
if scotland_code: if scotland_code:
rows = wikidata.lookup_scottish_parish_in_wikidata(scotland_code) rows = wikidata.lookup_scottish_parish_in_wikidata(scotland_code)
hit = wikidata.commons_from_rows(rows) hit = wikidata.commons_from_rows(rows)
@ -170,9 +203,11 @@ def wikidata_tag():
@app.route("/detail") @app.route("/detail")
def detail_page(): def detail_page() -> Response | str:
"""Detail page."""
try: try:
lat, lon = [float(request.args.get(param)) for param in ("lat", "lon")] lat_str, lon_str = request.args["lat"], request.args["lon"]
lat, lon = float(lat_str), float(lon_str)
except TypeError: except TypeError:
return redirect(url_for("index")) return redirect(url_for("index"))
try: try: