161 lines
4.8 KiB
Python
161 lines
4.8 KiB
Python
"""Sub-agente: Estilo de vida náutico.
|
|
|
|
Fuentes: Overpass API (marinas, boat ramps, playas, acceso al agua).
|
|
"""
|
|
|
|
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"
|
|
|
|
|
|
def run(lat: float, lon: float, address: str) -> dict:
|
|
result = {
|
|
"marinas": [],
|
|
"boat_ramps": [],
|
|
"beaches": [],
|
|
"nearest_marina": None,
|
|
"nearest_beach": None,
|
|
"nearest_boat_ramp": None,
|
|
"ocean_access": False,
|
|
"waterway_nearby": False,
|
|
"sources": ["OpenStreetMap/Overpass"],
|
|
"errors": [],
|
|
}
|
|
|
|
try:
|
|
data = _overpass_nautical(lat, lon)
|
|
result.update(data)
|
|
except Exception as e:
|
|
result["errors"].append(f"Overpass lifestyle: {e}")
|
|
|
|
return result
|
|
|
|
|
|
def _overpass_nautical(lat: float, lon: float, radius_m: int = 16000) -> dict:
|
|
"""Consulta amenidades náuticas en radio de ~10 millas."""
|
|
query = f"""
|
|
[out:json][timeout:35];
|
|
(
|
|
node["leisure"="marina"](around:{radius_m},{lat},{lon});
|
|
way["leisure"="marina"](around:{radius_m},{lat},{lon});
|
|
node["leisure"="slipway"](around:{radius_m},{lat},{lon});
|
|
way["leisure"="slipway"](around:{radius_m},{lat},{lon});
|
|
node["natural"="beach"](around:{radius_m},{lat},{lon});
|
|
way["natural"="beach"](around:{radius_m},{lat},{lon});
|
|
node["waterway"="river"](around:3200,{lat},{lon});
|
|
node["natural"="water"](around:3200,{lat},{lon});
|
|
way["natural"="coastline"](around:8000,{lat},{lon});
|
|
);
|
|
out body center;
|
|
"""
|
|
time.sleep(1)
|
|
r = requests.post(OVERPASS_URL, data={"data": query},
|
|
headers={"User-Agent": USER_AGENT}, timeout=40)
|
|
r.raise_for_status()
|
|
elements = r.json().get("elements", [])
|
|
|
|
marinas, boat_ramps, beaches = [], [], []
|
|
waterway_nearby = False
|
|
ocean_access = False
|
|
|
|
for el in elements:
|
|
tags = el.get("tags", {})
|
|
name = tags.get("name", "Sin nombre")
|
|
|
|
# Obtener coords
|
|
if "center" in el:
|
|
el_lat = el["center"]["lat"]
|
|
el_lon = el["center"]["lon"]
|
|
else:
|
|
el_lat = el.get("lat", lat)
|
|
el_lon = el.get("lon", lon)
|
|
|
|
dist = _haversine(lat, lon, el_lat, el_lon)
|
|
|
|
leisure = tags.get("leisure", "")
|
|
natural = tags.get("natural", "")
|
|
waterway = tags.get("waterway", "")
|
|
|
|
if leisure == "marina":
|
|
entry = {
|
|
"name": name,
|
|
"dist_miles": round(dist, 2),
|
|
"fuel": tags.get("fuel", "unknown"),
|
|
"pump_out": tags.get("pump_out", "unknown"),
|
|
"depth": tags.get("maxdraught", tags.get("depth", "unknown")),
|
|
}
|
|
marinas.append(entry)
|
|
elif leisure == "slipway":
|
|
boat_ramps.append({"name": name, "dist_miles": round(dist, 2)})
|
|
elif natural == "beach":
|
|
beaches.append({"name": name, "dist_miles": round(dist, 2)})
|
|
elif natural == "coastline":
|
|
ocean_access = True
|
|
elif waterway in ("river", "canal") or natural in ("water", "bay"):
|
|
waterway_nearby = True
|
|
|
|
marinas.sort(key=lambda x: x["dist_miles"])
|
|
boat_ramps.sort(key=lambda x: x["dist_miles"])
|
|
beaches.sort(key=lambda x: x["dist_miles"])
|
|
|
|
return {
|
|
"marinas": marinas[:10],
|
|
"boat_ramps": boat_ramps[:10],
|
|
"beaches": beaches[:10],
|
|
"nearest_marina": marinas[0] if marinas else None,
|
|
"nearest_beach": beaches[0] if beaches else None,
|
|
"nearest_boat_ramp": boat_ramps[0] if boat_ramps else None,
|
|
"ocean_access": ocean_access,
|
|
"waterway_nearby": waterway_nearby,
|
|
}
|
|
|
|
|
|
def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
|
R = 3958.8
|
|
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 score(data: dict) -> int:
|
|
"""Score 0-100 para lifestyle náutico."""
|
|
s = 30 # base
|
|
|
|
nearest_marina = data.get("nearest_marina")
|
|
if nearest_marina:
|
|
d = nearest_marina["dist_miles"]
|
|
if d <= 1:
|
|
s += 30
|
|
elif d <= 3:
|
|
s += 20
|
|
elif d <= 5:
|
|
s += 12
|
|
elif d <= 10:
|
|
s += 6
|
|
|
|
nearest_beach = data.get("nearest_beach")
|
|
if nearest_beach:
|
|
d = nearest_beach["dist_miles"]
|
|
if d <= 1:
|
|
s += 20
|
|
elif d <= 3:
|
|
s += 12
|
|
elif d <= 5:
|
|
s += 6
|
|
|
|
if data.get("ocean_access"):
|
|
s += 10
|
|
if data.get("waterway_nearby"):
|
|
s += 5
|
|
|
|
boat_ramps = len(data.get("boat_ramps", []))
|
|
s += min(10, boat_ramps * 2)
|
|
|
|
return min(100, max(0, s))
|