Merge pull request #368 from City-of-Turku/feature/filter-administrat…

…ive-divisions-by-address

Feature/filter administrative divisions by address

requirements.in

@@ -43,4 +43,5 @@ pyshp
 polyline
 drf-spectacular
 xmltodict
-freezegun
+freezegun
+geopy

requirements.txt

@@ -110,6 +110,10 @@ flake8-polyfill==1.0.2
  # via pep8-naming
 freezegun==1.5.1
  # via -r requirements.in
+geographiclib==2.0
+ # via geopy
+geopy==2.4.1
+ # via -r requirements.in
 idna==3.7
  # via requests
 inflection==0.5.1

services/api.py

@@ -42,6 +42,7 @@
 )
 from services.models.unit import ORGANIZER_TYPES, PROVIDER_TYPES
 from services.utils import check_valid_concrete_field
+from services.utils.geocode_address import geocode_address
 
 if settings.REST_FRAMEWORK and settings.REST_FRAMEWORK["DEFAULT_RENDERER_CLASSES"]:
  DEFAULT_RENDERERS = [
@@ -1173,6 +1174,24 @@ def to_representation(self, obj):
 class AdministrativeDivisionViewSet(munigeo_api.AdministrativeDivisionViewSet):
  serializer_class = AdministrativeDivisionSerializer
 
+ def get_queryset(self):
+ queryset = super().get_queryset()
+ filters = self.request.query_params
+
+ if "address" in filters and "municipality" in filters:
+ street_address = filters["address"]
+ municipality = filters["municipality"]
+ country = settings.DEFAULT_COUNTRY
+ address = f"{street_address}, {municipality}, {country}"
+ location_coordinates = geocode_address(address)
+ if location_coordinates:
+ point = Point(
+ location_coordinates[1], location_coordinates[0], srid=4326
+ )
+ queryset = queryset.filter(geometry__boundary__contains=point)
+
+ return queryset.order_by("id")
+
 
 register_view(AdministrativeDivisionViewSet, "administrative_division")
 

services/tests/test_administrative_division_view_set_api.py

@@ -0,0 +1,106 @@
+import pytest
+from django.conf import settings
+from django.contrib.gis.geos import MultiPolygon, Polygon
+from django.urls import reverse
+from munigeo.models import (
+ AdministrativeDivision,
+ AdministrativeDivisionGeometry,
+ AdministrativeDivisionType,
+ Municipality,
+)
+from rest_framework.test import APIClient
+
+from services.api import make_muni_ocd_id
+from services.tests.utils import get
+
+
+def create_administrative_divisions():
+ municipality_ids = ["helsinki", "espoo", "vantaa"]
+ division_type = AdministrativeDivisionType.objects.create(type="muni")
+ for municipality_id in municipality_ids:
+ municipality = Municipality.objects.create(
+ id=municipality_id, name=municipality_id
+ )
+ AdministrativeDivision.objects.create(
+ type=division_type,
+ name=municipality_id,
+ ocd_id=make_muni_ocd_id(municipality_id),
+ municipality=municipality,
+ )
+
+
+def create_test_area():
+ """
+ Create a simple test area in Helsinki center.
+ """
+ polygon_coords = [
+ (24.928, 60.178), # top left
+ (24.948, 60.178), # top right
+ (24.948, 60.159), # bottom right
+ (24.928, 60.159), # bottom left
+ (24.928, 60.178), # Close the ring by repeating the first point
+ ]
+ polygon = Polygon(polygon_coords, srid=4326) # WGS84 srid
+ multi_polygon = MultiPolygon(polygon, srid=4326)
+ multi_polygon.transform(settings.DEFAULT_SRID)
+ return multi_polygon
+
+
+@pytest.fixture
+def api_client():
+ return APIClient()
+
+
+@pytest.mark.django_db
+def test_get_administrative_division_list(api_client):
+ create_administrative_divisions()
+ response = get(api_client, reverse("administrativedivision-list"))
+ assert response.status_code == 200
+ assert response.data["count"] == 3
+
+
+@pytest.mark.django_db
+def test_municipality_filter(api_client):
+ create_administrative_divisions()
+ response = get(
+ api_client,
+ reverse("administrativedivision-list"),
+ data={"municipality": "helsinki"},
+ )
+ assert response.status_code == 200
+ assert response.data["count"] == 1
+ assert response.data["results"][0]["municipality"] == "helsinki"
+
+
+@pytest.mark.django_db
+def test_address_filter(api_client):
+ create_administrative_divisions()
+ division = AdministrativeDivision.objects.get(name="helsinki")
+ AdministrativeDivisionGeometry.objects.create(
+ division=division, boundary=create_test_area()
+ )
+
+ response = get(
+ api_client,
+ reverse("administrativedivision-list"),
+ data={
+ "municipality": "helsinki",
+ "address": "Kaivokatu 1",
+ }, # An address in the test area
+ )
+
+ assert response.status_code == 200
+ assert response.data["count"] == 1
+ assert response.data["results"][0]["municipality"] == "helsinki"
+
+ response = get(
+ api_client,
+ reverse("administrativedivision-list"),
+ data={
+ "municipality": "helsinki",
+ "address": "Katajanokanranta 1",
+ }, # An address outside the test area
+ )
+
+ assert response.status_code == 200
+ assert response.data["count"] == 0

services/utils/geocode_address.py

@@ -0,0 +1,12 @@
+from geopy.geocoders import Nominatim
+
+
+def geocode_address(address):
+ """
+ Geocodes address and returns location coordinates.
+ """
+ geolocator = Nominatim(user_agent="smbackend")
+ location = geolocator.geocode(address)
+ if location:
+ return location.latitude, location.longitude
+ return None