159 lines
4.7 KiB
Python
159 lines
4.7 KiB
Python
"""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
|