"""Sub-agente: Amenities y walkability. Fuente: Overpass API (OpenStreetMap) — gratuita, sin key requerida. """ from __future__ import annotations import math import time import requests from data_fetchers.base import USER_AGENT OVERPASS_URL = "https://overpass-api.de/api/interpreter" CATEGORIES = { "supermarket": ["supermarket", "grocery"], "hospital": ["hospital", "clinic", "doctors", "pharmacy"], "restaurant": ["restaurant", "fast_food", "cafe"], "park": ["park"], "gym": ["fitness_centre", "sports_centre"], "school": ["school", "kindergarten"], "bank": ["bank", "atm"], "gas_station": ["fuel"], } def run(lat: float, lon: float, address: str) -> dict: result = { "categories": {}, "nearest": {}, "walk_score_estimate": None, "total_amenities": 0, "sources": ["OpenStreetMap/Overpass"], "errors": [], } try: amenities = _overpass_amenities(lat, lon) result["categories"] = amenities["by_category"] result["nearest"] = amenities["nearest"] result["total_amenities"] = amenities["total"] # Walk score estimado (basado en densidad de amenities en 1 milla) result["walk_score_estimate"] = _estimate_walk_score(amenities) except Exception as e: result["errors"].append(f"Overpass amenities: {e}") return result def _overpass_amenities(lat: float, lon: float, radius_m: int = 3200) -> dict: """Consulta Overpass API para amenities en radio de ~2 millas.""" amenity_values = "|".join( v for values in CATEGORIES.values() for v in values ) query = f""" [out:json][timeout:30]; ( node["amenity"~"{amenity_values}"](around:{radius_m},{lat},{lon}); ); out body; """ time.sleep(1) r = requests.post(OVERPASS_URL, data={"data": query}, headers={"User-Agent": USER_AGENT}, timeout=35) r.raise_for_status() elements = r.json().get("elements", []) by_category: dict = {cat: [] for cat in CATEGORIES} nearest: dict = {} for el in elements: tags = el.get("tags", {}) amenity = tags.get("amenity", "") name = tags.get("name", amenity) el_lat = el.get("lat", lat) el_lon = el.get("lon", lon) dist = _haversine(lat, lon, el_lat, el_lon) for cat, values in CATEGORIES.items(): if amenity in values: by_category[cat].append({"name": name, "dist_miles": round(dist, 2)}) if cat not in nearest or dist < nearest[cat]["dist_miles"]: nearest[cat] = {"name": name, "dist_miles": round(dist, 2)} break # Ordenar por distancia for cat in by_category: by_category[cat].sort(key=lambda x: x["dist_miles"]) by_category[cat] = by_category[cat][:5] total = sum(len(v) for v in by_category.values()) return {"by_category": by_category, "nearest": nearest, "total": total} def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Distancia en millas entre dos coordenadas.""" R = 3958.8 # radio Tierra en millas dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2 return R * 2 * math.asin(math.sqrt(a)) def _estimate_walk_score(amenities: dict) -> int: """Estima walk score 0-100 basado en densidad y diversidad de amenities.""" cats = amenities["by_category"] nearest = amenities["nearest"] score = 0 # Puntos por cercanía de supermercado (más importante) sup = nearest.get("supermarket", {}).get("dist_miles", 99) if sup <= 0.25: score += 25 elif sup <= 0.5: score += 18 elif sup <= 1.0: score += 10 elif sup <= 2.0: score += 5 # Restaurantes/cafes cercanos rest_count = len([x for x in cats.get("restaurant", []) if x["dist_miles"] <= 1.0]) score += min(20, rest_count * 3) # Diversidad de categorías con algo en <= 1 milla cats_nearby = sum( 1 for cat, items in cats.items() if any(x["dist_miles"] <= 1.0 for x in items) ) score += cats_nearby * 5 # Hospitales hosp = nearest.get("hospital", {}).get("dist_miles", 99) if hosp <= 2.0: score += 10 return min(100, max(0, score)) def score(data: dict) -> int: """Score 0-100 de amenities.""" ws = data.get("walk_score_estimate") if ws is not None: return ws total = data.get("total_amenities", 0) if total >= 50: return 85 elif total >= 30: return 70 elif total >= 15: return 55 elif total >= 5: return 40 return 25