"""Services for geocoding and address routing."""
import logging
import string
from abc import ABC, abstractmethod
from dataclasses import dataclass
from decimal import Decimal
from hashlib import sha256
from math import sin, cos, atan2, sqrt, radians
from typing import List, Tuple, Callable
from urllib.parse import urljoin
import requests
from django.core import mail
from django.core.cache import cache
from django.core.exceptions import ImproperlyConfigured
from django.db import connection
from ferrosoft.config import require_settings
from ferrosoft.decimal import zero
from ferrosoft.importutils import import_object
_config_key_map_config = "FERROBASE_MAP_CONFIG"
[docs]
class AddressServiceException(RuntimeError):
pass
[docs]
class GeocodeException(AddressServiceException):
"""Exception raised in geocoding methods."""
pass
[docs]
class CoordinatesNotFound(GeocodeException):
"""Exception raised if no coordinates could be determined for an address."""
def __init__(self, address: "Address"):
self.address = address
def __str__(self):
return "Coordinates not found for %s" % self.address.address_line
[docs]
class RoutingException(AddressServiceException):
pass
[docs]
@dataclass
class Address:
first_name: str = ""
last_name: str = ""
salutation: str = ""
company: str = ""
additional_lines: str = ""
street: str = ""
house_no: str = ""
city: str = ""
postal_code: str = ""
country: str = ""
@property
def address_line(self) -> str:
# noinspection PyTypeChecker
return self.join_fields(
", ", ["street", "house_no", "postal_code", "city", "country"]
)
[docs]
def join_fields(self, join: str | None, fields: List[str]) -> str | list:
values = list(
filter(
lambda it: it is not None and it != "",
[getattr(self, field, None) for field in fields],
)
)
if isinstance(join, str):
return join.join(values)
return values
[docs]
def geocode(self, service=None) -> "Location":
if service is None:
service = AddressRouterService.get_instance()
return service.geocode(self)
[docs]
@dataclass
class Coordinates:
lat: Decimal
lon: Decimal
def __init__(self, lat, lon):
self.lat = Decimal(lat)
self.lon = Decimal(lon)
[docs]
@dataclass
class Location:
coords: Coordinates
name: str
address: Address
[docs]
@dataclass
class RouteLeg:
start: Coordinates
destination: Coordinates
[docs]
@dataclass
class Route:
legs: List[RouteLeg]
[docs]
@dataclass
class Distance:
value: Decimal
unit: str
[docs]
class Geocoder(ABC):
"""Geocoder determines Location from address query."""
[docs]
@abstractmethod
def geocode(self, address: Address) -> Location:
raise NotImplementedError
[docs]
class FuzzyGeocoder(Geocoder):
SQL_QUERY = """
SELECT address_label, postal_code, city, longitude, latitude, country, street, house_no, 1 AS weight
FROM ferromaps_geocodedaddress
WHERE
(CASE
WHEN %(house_no)s = '' THEN
street LIKE %(street)s AND city LIKE %(city)s
ELSE
house_no = %(house_no)s AND street LIKE %(street)s AND city LIKE %(city)s
END)
UNION
SELECT address_label, postal_code, city, longitude, latitude, country, street, house_no, 2 AS weight
FROM ferromaps_geocodedaddress
WHERE
daitch_mokotoff(address_label) = daitch_mokotoff(%(address_line)s)
ORDER BY weight limit 1;
"""
[docs]
def geocode(self, address: Address) -> Location:
with connection.cursor() as cursor:
cursor.execute(
self.SQL_QUERY,
{
"house_no": address.house_no,
# Strip dot, as it confuses the search.
"street": "%s%%" % address.street.rstrip("."),
"city": "%s%%" % address.city,
"address_line": address.address_line,
},
)
results = cursor.fetchall()
column_names = [col[0] for col in cursor.description]
result = dict(zip(column_names, results[0])) if results else None
if not result:
raise CoordinatesNotFound(address)
postal_code = result.get("postal_code")
if address.postal_code != "" and address.postal_code != postal_code:
# daitch_mokotoff can return nonsensible results, because it just looks
# for similarity. Ensure postal code matches, so that it doesn't return
# addresses far off the expected result.
raise CoordinatesNotFound(address)
return Location(
coords=Coordinates(lat=result.get("latitude"), lon=result.get("longitude")),
name=result.get("address_label"),
address=Address(
first_name="",
last_name="",
salutation="",
company="",
additional_lines="",
street=result.get("street"),
house_no=result.get("house_no"),
city=result.get("city"),
postal_code=result.get("postal_code"),
country=result.get("country"),
),
)
[docs]
class DistanceRouter(ABC):
"""Distance router determines distance of a route."""
[docs]
@abstractmethod
def distance(self, route: Route) -> Distance | None:
raise NotImplementedError
[docs]
class GreatCircleDistanceRouter(DistanceRouter):
EARTH_RADIUS_KM = 6378.388
[docs]
def distance(self, route: Route) -> Distance | None:
distance_sum = zero
for leg in route.legs:
lat1 = radians(float(leg.start.lat))
lat2 = radians(float(leg.destination.lat))
lon1 = radians(float(leg.start.lon))
lon2 = radians(float(leg.destination.lon))
d_lat = lat2 - lat1
d_lon = lon2 - lon1
a = pow(sin(d_lat / 2.0), 2) + (
pow(sin(d_lon / 2.0), 2) * cos(lat1) * cos(lat2)
)
distance_sum += Decimal(
self.EARTH_RADIUS_KM * 2.0 * atan2(sqrt(a), sqrt(1.0 - a))
)
return Distance(distance_sum, "km")
[docs]
class AddressRouterService(Geocoder, DistanceRouter, ABC):
"""AddressRouter service bundles Geocoder and DistanceRouter providing a unified API."""
instance = None
"""The instance to return in get_instance. If None, a "sensible" default implementation is returned."""
[docs]
@dataclass
class Config:
enable_cache: bool
router_class: str
[docs]
@classmethod
def from_django_settings(cls):
from django.conf import settings
config = getattr(settings, _config_key_map_config)
return cls(
enable_cache=config.get("enable_cache", False),
router_class=getattr(settings, "FERROBASE_ADDRESS_ROUTER_CLASS", None),
)
[docs]
@classmethod
def default(cls):
return cls.get_instance()
[docs]
@classmethod
def get_instance(cls):
"""Return instance or create it using a sensible default suitable for production."""
if cls.instance is None:
config = cls.Config.from_django_settings()
if config.router_class is None:
raise ImproperlyConfigured("FERROBASE_ADDRESS_ROUTER_CLASS must be set")
router_class = import_object(config.router_class)
cls.instance = router_class.get_instance()
if config.enable_cache:
cls.instance = CachingAddressRouter(cls.instance)
return cls.instance
[docs]
class CachingAddressRouter(AddressRouterService):
CACHE_PREFIX = "ferromaps"
UPSTREAM_NONE = "__NONE__"
def __init__(self, upstream: AddressRouterService):
self._upstream = upstream
@classmethod
def _get_cached(
cls,
cache_key: str,
return_type,
upstream_func: Callable,
allow_none: bool | None = None,
):
result = cache.get(cache_key)
if isinstance(result, return_type):
return result
# If upstream returned None, return it only if None is allowed.
# Otherwise, try upstream again.
if result == cls.UPSTREAM_NONE and allow_none:
return None
result = upstream_func()
if result is None:
# Set a special None value in cache, to differentiate it
# between unset cache key and None returned from upstream.
cache.set(cache_key, cls.UPSTREAM_NONE)
else:
cache.set(cache_key, result)
return result
[docs]
def geocode(self, address: Address) -> Location:
key = "%s_geocode_%s" % (
self.CACHE_PREFIX,
sha256(address.address_line.encode()).hexdigest(),
)
return self._get_cached(key, Location, lambda: self._upstream.geocode(address))
[docs]
def distance(self, route: Route) -> Distance | None:
waypoints = ";".join(
[
"%s,%s;%s,%s"
% (
leg.start.lat,
leg.start.lon,
leg.destination.lat,
leg.destination.lon,
)
for leg in route.legs
]
)
key = "%s_distance_%s" % (
self.CACHE_PREFIX,
sha256(waypoints.encode()).hexdigest(),
)
return self._get_cached(
key,
Distance,
lambda: self._upstream.distance(route),
allow_none=True,
)
[docs]
class SwitchingAddressRouter(AddressRouterService):
def __init__(self, geocoder: Geocoder, distance_router: DistanceRouter):
self._geocoder = geocoder
self._distance_router = distance_router
[docs]
def geocode(self, address: Address) -> Location:
return self._geocoder.geocode(address)
[docs]
def distance(self, route: Route) -> Distance | None:
return self._distance_router.distance(route)
[docs]
class AddressRouterFake(AddressRouterService):
def __init__(self, distances: List[Tuple[int, str]]):
self._distances = iter(distances)
[docs]
def geocode(self, address: Address) -> Location:
return Location(
coords=Coordinates(lat=Decimal(42), lon=Decimal(-100)),
name=address.address_line,
address=Address(
first_name="",
last_name="",
salutation="",
company="",
additional_lines="",
street="Test Street",
house_no="12a",
city="Test City",
postal_code="10010",
country="XX",
),
)
[docs]
def distance(self, route: Route) -> Distance | None:
value, unit = next(self._distances, (None, None))
return None if value is None else Distance(value, unit)
[docs]
class AddressRouterGeoapify(AddressRouterService):
[docs]
@dataclass
class Config:
base_url: str
api_url: str
api_key: str
[docs]
@classmethod
def from_django_settings(cls):
from django.conf import settings
config = getattr(settings, _config_key_map_config)
return cls(
base_url=config.get("base_url", ""),
api_url=config.get("api_url", ""),
api_key=config.get("api_key", ""),
)
def __init__(self, config: Config):
# Django provides no clear flag to check if code runs in a test environment.
# One has to resort to this hack (see also Django's setup_test_environment).
if hasattr(mail, "outbox"):
raise RuntimeError("Geoapify router should not be used in test")
self.config = config
[docs]
@classmethod
def default(cls):
return cls.get_instance()
[docs]
@classmethod
def get_instance(cls):
return cls(cls.Config.from_django_settings())
[docs]
def geocode(self, address: Address) -> Location:
address_line = address.address_line
if address_line == "":
# Garbage addresses are completely empty and should not hit Geoapify.
raise CoordinatesNotFound(address)
location = self._geocode_text(address_line)
if location is not None:
return location
# Falling back to the city is better than returning no result.
address_line = address.join_fields(", ", ["postal_code", "city", "country"])
if address_line == "":
raise CoordinatesNotFound(address)
location = self._geocode_text(address_line)
if location is None:
raise CoordinatesNotFound(address)
return location
def _geocode_text(self, address_line: str) -> Location | None:
response = requests.get(
urljoin(self.config.api_url, "/v1/geocode/search"),
params={
"text": address_line,
"format": "json",
"apiKey": self.config.api_key,
},
)
if response.status_code > 200:
logging.error("Geoapify error response: %s", response.text)
response.raise_for_status()
api_result = response.json()
if not isinstance(api_result, dict):
raise GeocodeException("Invalid API response")
geocode_results = api_result.get("results", None)
if not isinstance(geocode_results, list) or len(geocode_results) == 0:
return None
first_result = geocode_results[0]
return Location(
Coordinates(
lat=first_result.get("lat", 0),
lon=first_result.get("lon", 0),
),
first_result.get("formatted", address_line),
Address(
first_name="",
last_name="",
salutation="",
company="",
additional_lines="",
street=first_result.get("street", ""),
house_no=first_result.get("housenumber", ""),
city=first_result.get("city", ""),
postal_code=first_result.get("postcode"),
country=first_result.get("country_code", "").upper(),
),
)
[docs]
def distance(self, route: Route) -> Distance | None:
response = requests.get(
urljoin(self.config.api_url, "/v1/routing"),
params={
"apiKey": self.config.api_key,
"waypoints": self._geoapify_waypoints(route),
"mode": "heavy_truck",
"units": "metric",
"format": "geojson",
},
)
if response.status_code > 200:
logging.error("Geoapify error response: %s", response.text)
response.raise_for_status()
geo_result = response.json()
if not isinstance(geo_result, dict):
logging.warning("Geoapify response: %s", response.text)
raise RoutingException("Invalid Geoapify response")
status_code = geo_result.get("statusCode", 0)
if status_code >= 400:
raise RoutingException(
geo_result.get("message", "Unexpected Geoapify error")
)
features = geo_result.get("features", [])
if isinstance(features, list) and len(features) > 0:
first_feature = features[0]
properties = first_feature.get("properties", {})
unit = properties.get("distance_units", "")
if unit != "meters":
raise RoutingException("Non metric response: {}".format(unit))
return Distance(
value=Decimal(properties.get("distance", 0)) / Decimal(1000),
unit="km",
)
else:
logging.warning("Geoapify response: %s", response.text)
raise RoutingException("Geoapify returned no results")
@staticmethod
def _geoapify_waypoints(route: Route) -> str:
waypoints = []
for i, leg in enumerate(route.legs):
start_waypoint = "%s,%s" % (leg.start.lat, leg.start.lon)
dest_waypoint = "%s,%s" % (leg.destination.lat, leg.destination.lon)
if len(waypoints) > 0:
# check if last waypoint is the same as start_waypoint, in which case
# do not add start_waypoint, avoiding duplicate subsequent waypoints.
# This reduces API costs.
last_waypoint = waypoints[-1]
if last_waypoint != start_waypoint:
waypoints.append(start_waypoint)
else:
waypoints.append(start_waypoint)
waypoints.append(dest_waypoint)
return "|".join(waypoints)