diff --git a/etl/pipelines/openstreetmap/__init__.py b/etl/pipelines/openstreetmap/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/etl/pipelines/openstreetmap/donation_center/__init__.py b/etl/pipelines/openstreetmap/donation_center/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/etl/pipelines/openstreetmap/donation_center/normalizer.py b/etl/pipelines/openstreetmap/donation_center/normalizer.py new file mode 100644 index 0000000..b63c00d --- /dev/null +++ b/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 diff --git a/etl/pipelines/openstreetmap/donation_center/querier.py b/etl/pipelines/openstreetmap/donation_center/querier.py new file mode 100644 index 0000000..ee2354b --- /dev/null +++ b/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) diff --git a/etl/pipelines/openstreetmap/donation_center/test_pipeline.py b/etl/pipelines/openstreetmap/donation_center/test_pipeline.py new file mode 100644 index 0000000..e99e7af --- /dev/null +++ b/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) diff --git a/etl/pipelines/openstreetmap/openstreetmap_common.py b/etl/pipelines/openstreetmap/openstreetmap_common.py new file mode 100644 index 0000000..9fc6147 --- /dev/null +++ b/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)