Base Open Street Map Pipeline support & implementation for Donation Centers

etl/pipelines/openstreetmap/donation_center/normalizer.py

@@ -0,0 +1,30 @@
+from base.normalizer import BaseNormalizer
+from dtos import (
+    Activity,
+    NormalizedLocation,
+    RawLocation,
+)
+from pipelines.openstreetmap.donation_center.querier import DATA_SOURCE
+from pipelines.openstreetmap.openstreetmap_common import (
+    build_normalized_location
+)
+
+
+class DonationCenterNormalizer(BaseNormalizer):
+
+    def normalize(self, raw_locations: list[RawLocation]) -> list[NormalizedLocation]:
+        normalized_locations = []
+        for raw in raw_locations:
+            location = build_normalized_location(raw, DATA_SOURCE, self._get_activities)
+            if location is not None:
+                normalized_locations.append(location)
+        return normalized_locations
+
+    @staticmethod
+    def _get_activities(tags: dict) -> list[Activity]:
+        # Charity shops both accept donations and resell them.
+        activities = [Activity.DONATION_DROP]
+        if tags.get("shop") == "charity" or tags.get("second_hand") in ("yes", "only"):
+            activities.append(Activity.RESALE_BUY)
+
+        return activities

etl/pipelines/openstreetmap/donation_center/querier.py

@@ -0,0 +1,25 @@
+from base.querier import BaseQuerier
+from dtos import RawLocation
+from pipelines.openstreetmap.openstreetmap_common import (
+    GREATER_BOSTON_BBOX,
+    fetch_overpass,
+)
+
+
+DATA_SOURCE = "openstreetmap_donation"
+
+OVERPASS_QUERY = f"""
+[out:json][timeout:25];
+(
+  nwr["shop"="charity"]({GREATER_BOSTON_BBOX});
+  nwr["amenity"="recycling"]["recycling_type"="centre"]({GREATER_BOSTON_BBOX});
+  nwr["amenity"="social_facility"]["social_facility"="food_bank"]({GREATER_BOSTON_BBOX});
+);
+out center;
+"""
+
+
+class DonationCenterQuerier(BaseQuerier):
+
+    def fetch(self) -> list[RawLocation]:
+        return fetch_overpass(OVERPASS_QUERY, DATA_SOURCE)

etl/pipelines/openstreetmap/donation_center/test_pipeline.py

@@ -0,0 +1,9 @@
+from pipelines.openstreetmap.donation_center.querier import DonationCenterQuerier
+from pipelines.openstreetmap.donation_center.normalizer import DonationCenterNormalizer
+
+
+querier = DonationCenterQuerier()
+normalizer = DonationCenterNormalizer()
+
+locations = querier.fetch()
+normalized_locations = normalizer.normalize(locations)

etl/pipelines/openstreetmap/openstreetmap_common.py

@@ -0,0 +1,162 @@
+import json
+from datetime import datetime, timezone
+from typing import Callable
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+from dtos import (
+    Activity,
+    Address,
+    Availability,
+    Contact,
+    ItemCategory,
+    NormalizedLocation,
+    RawLocation,
+    Service,
+)
+
+OVERPASS_URL = "https://overpass-api.de/api/interpreter"
+GREATER_BOSTON_BBOX = "42.2,-71.2,42.5,-70.9"
+
+# OSM shop= values mapped to the ItemCategory of the goods sold/serviced.
+SHOP_TO_CATEGORY: dict[str, ItemCategory] = {
+    "tailor": ItemCategory.CLOTHING,
+    "clothes": ItemCategory.CLOTHING,
+    "shoes": ItemCategory.SHOES,
+    "computer": ItemCategory.ELECTRONICS,
+    "electronics": ItemCategory.ELECTRONICS,
+    "mobile_phone": ItemCategory.ELECTRONICS,
+    "furniture": ItemCategory.FURNITURE,
+    "books": ItemCategory.BOOKS,
+    "tool_hire": ItemCategory.TOOLS,
+    "hardware": ItemCategory.TOOLS,
+}
+
+def fetch_overpass(query: str, data_source: str) -> list[RawLocation]:
+    """POST an Overpass QL query and wrap each element as a RawLocation."""
+    body = urlencode({"data": query}).encode("utf-8")
+    request = Request(
+        OVERPASS_URL,
+        data=body,
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "User-Agent": "boston-circular-economy-etl/0.1",
+        },
+    )
+    with urlopen(request, timeout=60) as response:
+        data = json.load(response)
+
+    fetched_at = datetime.now(timezone.utc)
+    raw_locations = []
+    for element in data.get("elements", []):
+        raw_locations.append(RawLocation(
+            data_source=data_source,
+            data_source_id=f"osm-{element['type']}-{element['id']}",
+            fetched_at=fetched_at,
+            payload=element,
+        ))
+    return raw_locations
+
+
+def build_normalized_location(
+    raw: RawLocation,
+    data_source: str,
+    get_activities: Callable[[dict], list[Activity]],
+) -> NormalizedLocation | None:
+    """Validate an Overpass element and assemble a NormalizedLocation.
+
+    Returns None if the element lacks a name, coordinates, or services.
+    """
+    element = raw.payload
+    tags = element.get("tags", {})
+
+    name = tags.get("name")
+    lat, lon = get_coordinates(element)
+    categories = infer_item_categories(tags)
+    if not name or lat is None or lon is None or not categories:
+        return None
+
+    activities = get_activities(tags)
+    services = [
+        Service(activity=activity, item_category=category)
+        for activity in activities
+        for category in categories
+    ]
+
+    return NormalizedLocation(
+        data_source_id=raw.data_source_id,
+        data_source=data_source,
+        name=name,
+        lat=lat,
+        lon=lon,
+        address=build_address(tags),
+        contact=build_contact(tags),
+        services=services,
+        availability=build_availability(tags),
+        last_verified=tags.get("check_date"),
+    )
+
+
+def get_coordinates(element: dict) -> tuple[float | None, float | None]:
+    target = element if element["type"] == "node" else element.get("center", {})
+    return target.get("lat"), target.get("lon")
+
+
+def build_address(tags: dict) -> Address:
+    houseNumber = tags.get("addr:housenumber")
+    street_name = tags.get("addr:street")
+    if houseNumber and street_name:
+        street = f"{houseNumber} {street_name}"
+    else:
+        street = street_name or houseNumber
+    return Address(
+        street=street,
+        city=tags.get("addr:city"),
+        state=tags.get("addr:state"),
+        postcode=tags.get("addr:postcode"),
+    )
+
+
+def build_contact(tags: dict) -> Contact:
+    return Contact(
+        phone=tags.get("phone") or tags.get("contact:phone"),
+    #investigate
+        email=tags.get("email") or tags.get("contact:email"),
+        website=tags.get("website") or tags.get("contact:website"),
+    )
+
+
+def build_availability(tags: dict) -> Availability:
+    return Availability(opening_hours=tags.get("opening_hours"))
+
+# not great
+def infer_item_categories(tags: dict) -> list[ItemCategory]:
+    """Infer item categories from shop type and free-text fields."""
+    categories: set[ItemCategory] = set()
+
+    #investigate
+    shop = tags.get("shop")
+    category = SHOP_TO_CATEGORY.get(shop)
+    if category:
+        categories.add(category)
+
+    text = " ".join([
+        tags.get("name", ""),
+        #investigate
+        tags.get("name:en", ""),
+        tags.get("description", ""),
+    ]).lower()
+
+    keyword_map = {
+        ItemCategory.CLOTHING: ["clothing", "clothes", "apparel", "tailor"],
+        ItemCategory.ELECTRONICS: ["electronics", "computer", "phone", "printer"],
+        ItemCategory.FURNITURE: ["furniture"],
+        ItemCategory.SHOES: ["shoe", "cobbler"],
+        ItemCategory.BOOKS: ["book"],
+        ItemCategory.TOOLS: ["tool"],
+    }
+    for category, keywords in keyword_map.items():
+        if any(kw in text for kw in keywords):
+            categories.add(category)
+
+    return sorted(categories, key=lambda c: c.value)