auto(0129): agents_dashboard — secret_store_cpp_infra + CMakeLists register #4

Open
dataforge wants to merge 615 commits from auto/0129 into master
19 changed files with 1568 additions and 0 deletions
Showing only changes of commit cbc4c5eafa - Show all commits
View File
@@ -0,0 +1,83 @@
---
name: build_guide_prompt
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def build_guide_prompt(location: dict, pois: list[dict], interests: list[dict], user_query: str = \"\", lang: str = \"es\") -> list[dict]"
description: "Construye un prompt contextual para el LLM combinando ubicación actual, POIs cercanos rankeados e intereses del usuario. Retorna mensajes en formato OpenAI/Ollama."
tags: [prompt, llm, guide, location, pois, interests, openai, ollama, context]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: location
desc: "Dict con info de ubicación de nominatim_reverse_geocode. Campos esperados: display_name, street, neighbourhood, city, country. Los campos vacíos se omiten del prompt."
- name: pois
desc: "Lista de POIs rankeados por relevancia (ya filtrados por match_pois_to_interests). Cada POI tiene: name, category, score, matched_interests, lat, lon, tags. Se usan como máximo los primeros 5."
- name: interests
desc: "Lista de intereses del usuario. Cada interés tiene: name (str), keywords (list[str]), weight (float). Se usan los names para el system prompt."
- name: user_query
desc: "Pregunta opcional del usuario. Si está vacía o no se proporciona, se usa '¿Qué me puedes contar de esta zona?' como default."
- name: lang
desc: "Código de idioma para la respuesta del LLM (ej: 'es', 'en', 'fr'). Por defecto 'es'."
output: "Lista de dos mensajes en formato OpenAI/Ollama: [{'role': 'system', 'content': '...'}, {'role': 'user', 'content': '...'}]. El system message define el rol de guía local con los intereses y el idioma. El user message incluye la ubicación, los POIs relevantes y la query."
tested: true
tests:
- "con ubicacion completa pois y query"
- "sin pois solo ubicacion"
- "sin query usa default"
- "max 5 pois"
- "lang aparece en system"
- "location sin neighbourhood"
test_file_path: "python/functions/core/build_guide_prompt_test.py"
file_path: "python/functions/core/build_guide_prompt.py"
---
## Ejemplo
```python
messages = build_guide_prompt(
location={
"display_name": "Puerta del Sol, Madrid",
"city": "Madrid",
"neighbourhood": "Sol",
"street": "Puerta del Sol",
"country": "España",
},
pois=[
{
"name": "Museo del Prado",
"category": "museum",
"score": 1.0,
"matched_interests": ["arte"],
"lat": 40.41,
"lon": -3.70,
"tags": {},
}
],
interests=[{"name": "arte", "keywords": ["museo"], "weight": 1.0}],
user_query="¿Qué hay interesante por aquí?",
)
# [
# {"role": "system", "content": "Eres un guía local experto y amigable. El usuario está caminando por Madrid y le interesan: arte..."},
# {"role": "user", "content": "📍 Estoy en: Puerta del Sol, Madrid\nBarrio: Sol, Madrid\n\nLugares cercanos que me pueden interesar:\n- Museo del Prado (museum) — relevante para: arte\n\n¿Qué hay interesante por aquí?"}
# ]
```
## Notas
Función pura — no hace I/O, no muta los inputs, determinista.
El system message establece el rol del LLM como guía local con conocimiento de los intereses del usuario y el idioma de respuesta. Incluye instrucción de ser conciso (3-4 frases) y conversacional.
El user message combina:
- Ubicación completa del usuario (solo campos no vacíos)
- Lista de hasta 5 POIs con nombre, categoría y los intereses que satisfacen
- La query del usuario o un fallback genérico
Los campos de `location` vacíos o ausentes se omiten automáticamente del prompt para no generar ruido. Compatible con el formato de salida de `nominatim_reverse_geocode_py_infra` y `match_pois_to_interests`.
@@ -0,0 +1,91 @@
"""Construye un prompt contextual para el LLM combinando ubicación, POIs e intereses del usuario."""
def build_guide_prompt(
location: dict,
pois: list[dict],
interests: list[dict],
user_query: str = "",
lang: str = "es",
) -> list[dict]:
"""Construye un prompt contextual para el LLM como guía local.
Combina la ubicación actual del usuario, POIs cercanos rankeados por relevancia
e intereses del usuario para generar mensajes en formato OpenAI/Ollama.
Args:
location: Dict con info de ubicación de nominatim_reverse_geocode.
Campos esperados: display_name, street, neighbourhood, city, country, etc.
pois: Lista de POIs rankeados con campos: name, category, score,
matched_interests, lat, lon, tags. Ya filtrados y ordenados por score.
interests: Lista de intereses del usuario. Cada uno con: name, keywords, weight.
user_query: Pregunta opcional del usuario. Si está vacía, se usa un default.
lang: Idioma para la respuesta del LLM (ej: "es", "en", "fr").
Returns:
Lista de mensajes en formato OpenAI/Ollama:
[{"role": "system", "content": "..."}, {"role": "user", "content": "..."}]
"""
# Extraer nombre de intereses para el system prompt
interest_names = [i.get("name", "") for i in interests if i.get("name")]
interests_str = ", ".join(interest_names) if interest_names else "lugares de interés general"
# Extraer ciudad para el system prompt
city = location.get("city", "") or location.get("town", "") or location.get("village", "") or "la zona"
# Construir system message
system_content = (
f"Eres un guía local experto y amigable. El usuario está caminando por {city} "
f"y le interesan: {interests_str}.\n"
"Tu objetivo es informar sobre lo que hay cerca, dar contexto histórico/cultural "
"relevante a sus intereses, y sugerir qué visitar.\n"
f"Responde en {lang}. Sé conciso (máximo 3-4 frases) y conversacional, "
"como si caminaras junto al usuario.\n"
"Si no hay POIs relevantes, comenta sobre el barrio o zona."
)
# Construir user message
user_parts = []
# Ubicación
display_name = location.get("display_name", "")
if display_name:
user_parts.append(f"📍 Estoy en: {display_name}")
neighbourhood = location.get("neighbourhood", "") or location.get("suburb", "")
city_label = location.get("city", "") or location.get("town", "") or location.get("village", "")
location_line_parts = [p for p in [neighbourhood, city_label] if p]
if location_line_parts:
user_parts.append(f"Barrio: {', '.join(location_line_parts)}")
# POIs (máximo 5)
top_pois = pois[:5]
if top_pois:
user_parts.append("")
user_parts.append("Lugares cercanos que me pueden interesar:")
for poi in top_pois:
name = poi.get("name", "Sin nombre")
category = poi.get("category", "")
matched = poi.get("matched_interests", [])
matched_str = ", ".join(matched) if matched else ""
poi_line = f"- {name}"
if category:
poi_line += f" ({category})"
if matched_str:
poi_line += f" — relevante para: {matched_str}"
user_parts.append(poi_line)
# Query del usuario
user_parts.append("")
if user_query and user_query.strip():
user_parts.append(user_query.strip())
else:
user_parts.append("¿Qué me puedes contar de esta zona?")
user_content = "\n".join(user_parts)
return [
{"role": "system", "content": system_content},
{"role": "user", "content": user_content},
]
@@ -0,0 +1,157 @@
"""Tests para build_guide_prompt."""
from build_guide_prompt import build_guide_prompt
def test_con_ubicacion_completa_pois_y_query():
location = {
"display_name": "Puerta del Sol, Madrid",
"street": "Puerta del Sol",
"neighbourhood": "Sol",
"city": "Madrid",
"country": "España",
}
pois = [
{
"name": "Museo del Prado",
"category": "museum",
"score": 1.0,
"matched_interests": ["arte"],
"lat": 40.41,
"lon": -3.70,
"tags": {},
},
{
"name": "Jardines del Retiro",
"category": "park",
"score": 0.8,
"matched_interests": ["naturaleza"],
"lat": 40.41,
"lon": -3.68,
"tags": {},
},
]
interests = [
{"name": "arte", "keywords": ["museo", "pintura"], "weight": 1.0},
{"name": "naturaleza", "keywords": ["parque", "jardín"], "weight": 0.8},
]
user_query = "¿Qué hay interesante por aquí?"
messages = build_guide_prompt(location, pois, interests, user_query)
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[1]["role"] == "user"
system = messages[0]["content"]
assert "Madrid" in system
assert "arte" in system
assert "naturaleza" in system
user = messages[1]["content"]
assert "Puerta del Sol, Madrid" in user
assert "Sol" in user
assert "Museo del Prado" in user
assert "museum" in user
assert "arte" in user
assert "Jardines del Retiro" in user
assert "¿Qué hay interesante por aquí?" in user
def test_sin_pois_solo_ubicacion():
location = {
"display_name": "Calle Mayor, Salamanca",
"street": "Calle Mayor",
"neighbourhood": "Centro Histórico",
"city": "Salamanca",
"country": "España",
}
pois = []
interests = [{"name": "historia", "keywords": ["catedral"], "weight": 1.0}]
user_query = "¿Algo interesante cerca?"
messages = build_guide_prompt(location, pois, interests, user_query)
assert len(messages) == 2
user = messages[1]["content"]
assert "Salamanca" in user
assert "Lugares cercanos" not in user
assert "¿Algo interesante cerca?" in user
def test_sin_query_usa_default():
location = {
"display_name": "Plaza Mayor, Sevilla",
"city": "Sevilla",
"neighbourhood": "Casco Antiguo",
"country": "España",
}
pois = [
{
"name": "Catedral de Sevilla",
"category": "cathedral",
"score": 0.95,
"matched_interests": ["historia", "arquitectura"],
"lat": 37.38,
"lon": -5.99,
"tags": {},
}
]
interests = [
{"name": "historia", "keywords": ["catedral"], "weight": 1.0},
{"name": "arquitectura", "keywords": ["edificio"], "weight": 0.9},
]
messages = build_guide_prompt(location, pois, interests)
assert len(messages) == 2
user = messages[1]["content"]
assert "¿Qué me puedes contar de esta zona?" in user
assert "Catedral de Sevilla" in user
assert "historia" in user
assert "arquitectura" in user
def test_max_5_pois():
location = {"display_name": "Barcelona", "city": "Barcelona"}
pois = [
{"name": f"Lugar {i}", "category": "poi", "score": 1.0 - i * 0.1, "matched_interests": ["cultura"], "lat": 0, "lon": 0, "tags": {}}
for i in range(10)
]
interests = [{"name": "cultura", "keywords": [], "weight": 1.0}]
messages = build_guide_prompt(location, pois, interests)
user = messages[1]["content"]
# Solo los primeros 5 POIs deben aparecer
for i in range(5):
assert f"Lugar {i}" in user
for i in range(5, 10):
assert f"Lugar {i}" not in user
def test_lang_aparece_en_system():
location = {"display_name": "Paris", "city": "Paris"}
pois = []
interests = [{"name": "art", "keywords": [], "weight": 1.0}]
messages = build_guide_prompt(location, pois, interests, lang="fr")
system = messages[0]["content"]
assert "fr" in system
def test_location_sin_neighbourhood():
location = {
"display_name": "Gran Via, Madrid",
"city": "Madrid",
}
pois = []
interests = []
messages = build_guide_prompt(location, pois, interests)
user = messages[1]["content"]
assert "Gran Via, Madrid" in user
# No debe romper si no hay neighbourhood
assert len(messages) == 2
@@ -0,0 +1,90 @@
---
name: generate_pwa_manifest
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def generate_pwa_manifest(name: str, short_name: str, description: str = \"\", start_url: str = \"/\", display: str = \"standalone\", orientation: str = \"portrait\", theme_color: str = \"#000000\", background_color: str = \"#000000\", icons: list[dict] | None = None, categories: list[str] | None = None, lang: str = \"es\") -> dict"
description: "Genera el contenido de un manifest.json para Progressive Web Apps segun el estandar W3C Web App Manifest. Retorna un dict listo para json.dumps(). Si icons es None genera placeholders estandar para 192x192 y 512x512."
tags: [pwa, manifest, web, json, frontend]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: name
desc: "Nombre completo de la aplicacion que aparece en la pantalla de instalacion."
- name: short_name
desc: "Nombre corto para launchers y pantallas de inicio (maximo 12 caracteres recomendado)."
- name: description
desc: "Descripcion de la aplicacion. Cadena vacia por defecto."
- name: start_url
desc: "URL de inicio al lanzar la PWA desde la pantalla de inicio. Por defecto '/'."
- name: display
desc: "Modo de visualizacion: standalone, fullscreen, minimal-ui o browser. Por defecto standalone."
- name: orientation
desc: "Orientacion preferida de pantalla: portrait, landscape o any. Por defecto portrait."
- name: theme_color
desc: "Color del tema en formato hex (#RRGGBB) para la barra de estado y UI del SO. Por defecto #000000."
- name: background_color
desc: "Color de fondo en formato hex (#RRGGBB) mostrado durante la carga. Por defecto #000000."
- name: icons
desc: "Lista de dicts con src, sizes, type y purpose. Si es None genera placeholders para 192x192 y 512x512."
- name: categories
desc: "Lista de categorias W3C (navigation, shopping, travel, etc). Si es None usa lista vacia."
- name: lang
desc: "Codigo de idioma BCP 47 del contenido principal de la app. Por defecto 'es'."
output: "Dict con la estructura completa W3C Web App Manifest: name, short_name, description, start_url, scope, display, orientation, theme_color, background_color, icons, categories y lang."
tested: true
tests:
- "valores por defecto generan manifest valido con icons placeholder"
- "icons custom se respetan"
- "todos los campos se incluyen correctamente"
- "categories none produce lista vacia"
- "json dumps compatible"
test_file_path: "python/functions/core/generate_pwa_manifest_test.py"
file_path: "python/functions/core/generate_pwa_manifest.py"
---
## Ejemplo
```python
manifest = generate_pwa_manifest(
name="Voice Guide",
short_name="VoiceGuide",
description="Guia por voz con POIs",
theme_color="#12b886",
background_color="#1a1b1e",
)
# {
# "name": "Voice Guide",
# "short_name": "VoiceGuide",
# "start_url": "/",
# "scope": "/",
# "display": "standalone",
# "orientation": "portrait",
# "theme_color": "#12b886",
# "background_color": "#1a1b1e",
# "icons": [
# {"src": "/icons/icon-192x192.png", "sizes": "192x192", "type": "image/png", "purpose": "any maskable"},
# {"src": "/icons/icon-512x512.png", "sizes": "512x512", "type": "image/png", "purpose": "any maskable"},
# ],
# "categories": [],
# "lang": "es",
# }
import json
with open("public/manifest.json", "w") as f:
json.dump(manifest, f, indent=2, ensure_ascii=False)
```
## Notas
Funcion pura — sin I/O, sin efectos secundarios. El caller es responsable de escribir el archivo con `json.dumps()` o `json.dump()`.
Los valores de `display` validos segun W3C son: `fullscreen`, `standalone`, `minimal-ui`, `browser`. La funcion no valida el valor — acepta cualquier string para permitir valores futuros del estandar.
El campo `scope` siempre se fija a `"/"`. Si se necesita un scope distinto, el caller puede modificar el dict resultante antes de serializar.
@@ -0,0 +1,67 @@
"""Genera el contenido de un manifest.json para Progressive Web Apps (W3C Web App Manifest)."""
def generate_pwa_manifest(
name: str,
short_name: str,
description: str = "",
start_url: str = "/",
display: str = "standalone",
orientation: str = "portrait",
theme_color: str = "#000000",
background_color: str = "#000000",
icons: list[dict] | None = None,
categories: list[str] | None = None,
lang: str = "es",
) -> dict:
"""Genera el contenido de un manifest.json para Progressive Web Apps.
Args:
name: Nombre completo de la aplicacion.
short_name: Nombre corto para launchers y pantallas de inicio.
description: Descripcion de la aplicacion.
start_url: URL de inicio de la PWA.
display: Modo de visualizacion (standalone, fullscreen, minimal-ui, browser).
orientation: Orientacion preferida (portrait, landscape, any).
theme_color: Color del tema en hex.
background_color: Color de fondo en hex.
icons: Lista de iconos con src, sizes, type y purpose. Si es None, usa placeholders estandar.
categories: Lista de categorias de la app. Si es None, usa lista vacia.
lang: Idioma principal de la aplicacion (BCP 47).
Returns:
Dict listo para json.dumps() con la estructura estandar W3C Web App Manifest.
"""
if icons is None:
icons = [
{
"src": "/icons/icon-192x192.png",
"sizes": "192x192",
"type": "image/png",
"purpose": "any maskable",
},
{
"src": "/icons/icon-512x512.png",
"sizes": "512x512",
"type": "image/png",
"purpose": "any maskable",
},
]
if categories is None:
categories = []
return {
"name": name,
"short_name": short_name,
"description": description,
"start_url": start_url,
"scope": "/",
"display": display,
"orientation": orientation,
"theme_color": theme_color,
"background_color": background_color,
"icons": icons,
"categories": categories,
"lang": lang,
}
@@ -0,0 +1,94 @@
"""Tests para generate_pwa_manifest."""
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from generate_pwa_manifest import generate_pwa_manifest
def test_valores_por_defecto_generan_manifest_valido_con_icons_placeholder():
result = generate_pwa_manifest(name="Mi App", short_name="App")
assert result["name"] == "Mi App"
assert result["short_name"] == "App"
assert result["start_url"] == "/"
assert result["scope"] == "/"
assert result["display"] == "standalone"
assert result["orientation"] == "portrait"
assert result["theme_color"] == "#000000"
assert result["background_color"] == "#000000"
assert result["lang"] == "es"
assert result["description"] == ""
assert result["categories"] == []
icons = result["icons"]
assert len(icons) == 2
assert icons[0]["src"] == "/icons/icon-192x192.png"
assert icons[0]["sizes"] == "192x192"
assert icons[0]["type"] == "image/png"
assert icons[0]["purpose"] == "any maskable"
assert icons[1]["src"] == "/icons/icon-512x512.png"
assert icons[1]["sizes"] == "512x512"
def test_icons_custom_se_respetan():
custom_icons = [
{"src": "/my-icon.png", "sizes": "256x256", "type": "image/png", "purpose": "any"},
]
result = generate_pwa_manifest(name="App", short_name="A", icons=custom_icons)
assert result["icons"] == custom_icons
assert len(result["icons"]) == 1
assert result["icons"][0]["src"] == "/my-icon.png"
assert result["icons"][0]["sizes"] == "256x256"
def test_todos_los_campos_se_incluyen_correctamente():
result = generate_pwa_manifest(
name="Voice Guide",
short_name="VoiceGuide",
description="Guia por voz con POIs",
start_url="/app",
display="fullscreen",
orientation="landscape",
theme_color="#12b886",
background_color="#1a1b1e",
categories=["navigation", "travel"],
lang="en",
)
assert result["name"] == "Voice Guide"
assert result["short_name"] == "VoiceGuide"
assert result["description"] == "Guia por voz con POIs"
assert result["start_url"] == "/app"
assert result["scope"] == "/"
assert result["display"] == "fullscreen"
assert result["orientation"] == "landscape"
assert result["theme_color"] == "#12b886"
assert result["background_color"] == "#1a1b1e"
assert result["categories"] == ["navigation", "travel"]
assert result["lang"] == "en"
# icons son placeholders por defecto
assert len(result["icons"]) == 2
def test_categories_none_produce_lista_vacia():
result = generate_pwa_manifest(name="App", short_name="A", categories=None)
assert result["categories"] == []
def test_json_dumps_compatible():
import json
result = generate_pwa_manifest(
name="Voice Guide",
short_name="VG",
theme_color="#12b886",
background_color="#1a1b1e",
)
serialized = json.dumps(result)
parsed = json.loads(serialized)
assert parsed["name"] == "Voice Guide"
assert parsed["theme_color"] == "#12b886"
@@ -0,0 +1,62 @@
---
name: generate_service_worker
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def generate_service_worker(cache_name: str = 'app-cache-v1', precache_urls: list[str] | None = None, network_first_patterns: list[str] | None = None) -> str"
description: "Genera el codigo JavaScript de un Service Worker basico para PWAs con estrategia cache-first para assets estaticos y network-first para API calls."
tags: [pwa, service-worker, javascript, cache, offline, codegen]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: cache_name
desc: "Nombre del cache para esta version del SW. Se usa como clave para identificar el cache activo y limpiar los anteriores en activate."
- name: precache_urls
desc: "Lista de URLs a pre-cachear durante el evento install. Por defecto: ['/', '/index.html']."
- name: network_first_patterns
desc: "Lista de patrones de URL (substrings) que usan estrategia network-first. Por defecto: ['/api/']. Cualquier URL que contenga alguno de estos patrones va a red primero."
output: "String con codigo JavaScript valido del Service Worker listo para escribir como sw.js. Incluye handlers de install (precacheo), activate (limpieza de caches antiguos) y fetch (cache-first / network-first segun patron)."
tested: true
tests:
- "SW por defecto contiene install/activate/fetch handlers"
- "Custom precache_urls aparecen en el codigo"
- "Custom network_first_patterns aparecen en el fetch handler"
- "El cache_name personalizado aparece en el codigo generado"
- "Los valores por defecto se usan cuando los parametros son None"
- "La funcion retorna un string no vacio"
- "El evento activate limpia caches antiguos"
- "La estrategia cache-first guarda respuestas de red en cache"
test_file_path: "python/functions/core/generate_service_worker_test.py"
file_path: "python/functions/core/generate_service_worker.py"
---
## Ejemplo
```python
sw_code = generate_service_worker(
cache_name="voice-guide-v1",
precache_urls=["/", "/index.html", "/manifest.json"],
network_first_patterns=["/api/"],
)
# Escribir al filesystem (operacion impura, fuera de esta funcion):
# with open("public/sw.js", "w") as f:
# f.write(sw_code)
```
## Notas
Funcion pura: no hace I/O, no tiene efectos secundarios, retorna siempre el mismo string dado los mismos inputs.
El codigo JS generado usa `self.skipWaiting()` en install y `self.clients.claim()` en activate para activacion inmediata del SW sin necesidad de refrescar la pagina.
La estrategia de fetch:
- **Network-first** (para `/api/` por defecto): intenta la red, guarda la respuesta exitosa en cache, y usa el cache como fallback si la red falla. Garantiza datos frescos cuando hay conexion.
- **Cache-first** (para todo lo demas): sirve desde cache si existe, si no va a red y guarda en cache. Optimo para assets estaticos (JS, CSS, imagenes) que cambian con el nombre de version.
Para invalidar el cache al desplegar una nueva version, basta con cambiar `cache_name` (ej: `app-cache-v2`). El evento activate limpia automaticamente cualquier cache con nombre distinto al actual.
@@ -0,0 +1,128 @@
"""Genera el codigo JavaScript de un Service Worker basico para PWAs."""
from __future__ import annotations
def generate_service_worker(
cache_name: str = "app-cache-v1",
precache_urls: list[str] | None = None,
network_first_patterns: list[str] | None = None,
) -> str:
"""Genera el codigo JavaScript de un Service Worker con estrategia cache-first
para assets estaticos y network-first para API calls.
Args:
cache_name: Nombre del cache a usar para esta version del SW.
precache_urls: URLs a pre-cachear en el evento install.
network_first_patterns: Patrones de URL que usan estrategia network-first.
Returns:
String con codigo JavaScript valido del Service Worker.
"""
if precache_urls is None:
precache_urls = ["/", "/index.html"]
if network_first_patterns is None:
network_first_patterns = ["/api/"]
precache_urls_js = _list_to_js_array(precache_urls, indent=2)
network_patterns_js = _list_to_js_array(network_first_patterns, indent=2)
network_check_js = _build_network_check(network_first_patterns)
return f"""\
// Service Worker — generado automaticamente
// Estrategia: cache-first para assets, network-first para API
const CACHE_NAME = '{cache_name}';
const PRECACHE_URLS = {precache_urls_js};
const NETWORK_FIRST_PATTERNS = {network_patterns_js};
// ── install ──────────────────────────────────────────────────────────────────
// Pre-cachea los assets estaticos esenciales al instalar el SW.
self.addEventListener('install', (event) => {{
event.waitUntil(
caches.open(CACHE_NAME).then((cache) => {{
return cache.addAll(PRECACHE_URLS);
}})
);
self.skipWaiting();
}});
// ── activate ─────────────────────────────────────────────────────────────────
// Limpia caches de versiones anteriores al activar la nueva version del SW.
self.addEventListener('activate', (event) => {{
event.waitUntil(
caches.keys().then((cacheNames) => {{
return Promise.all(
cacheNames
.filter((name) => name !== CACHE_NAME)
.map((name) => caches.delete(name))
);
}})
);
self.clients.claim();
}});
// ── fetch ─────────────────────────────────────────────────────────────────────
// Intercepta peticiones:
// - URLs que coinciden con NETWORK_FIRST_PATTERNS → network-first con fallback a cache
// - El resto → cache-first con fallback a network (y guarda en cache la respuesta)
self.addEventListener('fetch', (event) => {{
const url = event.request.url;
const isNetworkFirst = {network_check_js};
if (isNetworkFirst) {{
// Network-first: intenta red, si falla usa cache
event.respondWith(
fetch(event.request)
.then((response) => {{
if (response && response.status === 200) {{
const responseClone = response.clone();
caches.open(CACHE_NAME).then((cache) => {{
cache.put(event.request, responseClone);
}});
}}
return response;
}})
.catch(() => caches.match(event.request))
);
}} else {{
// Cache-first: sirve desde cache, si no existe va a red y guarda en cache
event.respondWith(
caches.match(event.request).then((cached) => {{
if (cached) {{
return cached;
}}
return fetch(event.request).then((response) => {{
if (response && response.status === 200) {{
const responseClone = response.clone();
caches.open(CACHE_NAME).then((cache) => {{
cache.put(event.request, responseClone);
}});
}}
return response;
}});
}})
);
}}
}});
"""
def _list_to_js_array(items: list[str], indent: int = 0) -> str:
"""Convierte una lista Python a un array JS multilínea."""
if not items:
return "[]"
pad = " " * indent
inner = f",\n{pad} ".join(f"'{item}'" for item in items)
return f"[\n{pad} {inner},\n{pad}]"
def _build_network_check(patterns: list[str]) -> str:
"""Genera la expresion JS que comprueba si una URL matchea algun patron."""
if not patterns:
return "false"
checks = " || ".join(f"url.includes('{p}')" for p in patterns)
return checks
@@ -0,0 +1,62 @@
"""Tests para generate_service_worker."""
from generate_service_worker import generate_service_worker
def test_default_sw_contains_install_activate_fetch_handlers():
"""SW por defecto contiene install/activate/fetch handlers."""
code = generate_service_worker()
assert "addEventListener('install'" in code
assert "addEventListener('activate'" in code
assert "addEventListener('fetch'" in code
def test_custom_precache_urls_appear_in_code():
"""Custom precache_urls aparecen en el codigo."""
urls = ["/", "/index.html", "/manifest.json", "/offline.html"]
code = generate_service_worker(precache_urls=urls)
for url in urls:
assert f"'{url}'" in code
def test_custom_network_first_patterns_appear_in_fetch_handler():
"""Custom network_first_patterns aparecen en el fetch handler."""
patterns = ["/api/", "/graphql"]
code = generate_service_worker(network_first_patterns=patterns)
for pattern in patterns:
assert f"url.includes('{pattern}')" in code
def test_cache_name_appears_in_code():
"""El cache_name personalizado aparece en el codigo generado."""
code = generate_service_worker(cache_name="voice-guide-v1")
assert "voice-guide-v1" in code
def test_default_values_are_used_when_none():
"""Los valores por defecto se usan cuando los parametros son None."""
code = generate_service_worker()
assert "'/'" in code
assert "'/index.html'" in code
assert "url.includes('/api/')" in code
def test_returns_valid_string():
"""La funcion retorna un string no vacio."""
code = generate_service_worker()
assert isinstance(code, str)
assert len(code) > 0
def test_activate_cleans_old_caches():
"""El evento activate limpia caches antiguos."""
code = generate_service_worker(cache_name="my-cache-v2")
assert "caches.keys()" in code
assert "caches.delete(name)" in code
assert "name !== CACHE_NAME" in code
def test_cache_first_fallback_saves_to_cache():
"""La estrategia cache-first guarda respuestas de red en cache."""
code = generate_service_worker()
assert "cache.put(event.request" in code
@@ -0,0 +1,58 @@
---
name: match_pois_to_interests
kind: function
lang: py
domain: core
version: "1.0.0"
purity: pure
signature: "def match_pois_to_interests(pois: list[dict], interests: list[dict], max_results: int = 10) -> list[dict]"
description: "Filtra y rankea una lista de POIs segun un perfil de intereses del usuario. Calcula un score por categoria (0.5), nombre (0.3) y tags (0.2), multiplicado por el weight del interes. Retorna los top max_results POIs con score > 0, enriquecidos con score y matched_interests."
tags: [poi, ranking, filtering, interests, geospatial, recommendation, pure]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
params:
- name: pois
desc: "Lista de POIs, cada uno con campos id, lat, lon, name, category y tags (dict). Formato compatible con la salida de overpass_nearby_pois."
- name: interests
desc: "Lista de intereses del usuario, cada uno con name (str), keywords (list[str]) y weight (float entre 0 y 1). Los keywords se comparan contra categoria, nombre y tags del POI."
- name: max_results
desc: "Numero maximo de POIs a retornar. Por defecto 10."
output: "Lista de POIs filtrados y ordenados por score descendente. Cada POI incluye todos sus campos originales mas score (float) y matched_interests (list[str] con los nombres de intereses que matchearon). Lista vacia si ningun POI supera score 0."
tested: true
tests:
- "pois que matchean parcialmente por categoria y nombre"
- "pois sin match retorna lista vacia"
- "multiples intereses con diferentes weights acumulan score"
test_file_path: "python/functions/core/match_pois_to_interests_test.py"
file_path: "python/functions/core/match_pois_to_interests.py"
---
## Ejemplo
```python
pois = [
{"id": 1, "lat": 40.41, "lon": -3.70, "name": "Museo del Prado", "category": "museum", "tags": {"tourism": "museum"}},
{"id": 2, "lat": 40.42, "lon": -3.71, "name": "Bar El Tigre", "category": "bar", "tags": {"amenity": "bar"}},
]
interests = [{"name": "arte", "keywords": ["museo", "museum", "arte"], "weight": 1.0}]
result = match_pois_to_interests(pois, interests)
# [{"id": 1, ..., "score": 1.0, "matched_interests": ["arte"]}]
```
## Notas
Funcion pura: no hace I/O, no muta los inputs, es determinista.
**Calculo de score por interes:**
- Match de categoria: 0.5 si `poi["category"]` esta en los keywords del interes
- Match de nombre: 0.3 si algun keyword aparece como substring en `poi["name"]` (case-insensitive)
- Match de tags: 0.2 si algun keyword aparece en los valores de `poi["tags"]` (case-insensitive)
- El subtotal del interes se multiplica por `weight`
Si un POI matchea multiples intereses, los scores se suman. Un POI puede obtener score maximo de `sum(weight_i * 1.0)` sobre todos los intereses que matcheen.
El campo `tags` puede ser dict con valores string o numericos — todos se convierten a string para la comparacion.
@@ -0,0 +1,62 @@
"""Filtra y rankea POIs segun un perfil de intereses del usuario."""
def match_pois_to_interests(
pois: list[dict],
interests: list[dict],
max_results: int = 10,
) -> list[dict]:
"""Filtra y rankea una lista de POIs segun un perfil de intereses.
Calcula un score para cada POI basado en matches de categoria, nombre
y tags contra los keywords de cada interes. Solo retorna POIs con score > 0,
ordenados por score descendente, hasta max_results.
Args:
pois: Lista de POIs con campos id, lat, lon, name, category, tags.
interests: Lista de intereses con campos name, keywords y weight.
max_results: Numero maximo de resultados a retornar.
Returns:
Lista de POIs enriquecidos con score y matched_interests, ordenados
por score descendente. Lista vacia si ningun POI matchea.
"""
scored: list[dict] = []
for poi in pois:
total_score = 0.0
matched: list[str] = []
category = (poi.get("category") or "").lower()
name = (poi.get("name") or "").lower()
tags = poi.get("tags") or {}
tag_values = " ".join(str(v) for v in tags.values()).lower()
for interest in interests:
keywords: list[str] = [kw.lower() for kw in (interest.get("keywords") or [])]
weight: float = float(interest.get("weight", 1.0))
interest_name: str = interest.get("name", "")
interest_score = 0.0
# Match de categoria: 0.5 puntos si la categoria aparece en keywords
if any(kw == category for kw in keywords):
interest_score += 0.5
# Match de nombre: 0.3 puntos si algun keyword es substring del nombre
if any(kw in name for kw in keywords):
interest_score += 0.3
# Match de tags: 0.2 puntos si algun keyword aparece en valores de tags
if any(kw in tag_values for kw in keywords):
interest_score += 0.2
if interest_score > 0:
total_score += interest_score * weight
matched.append(interest_name)
if total_score > 0:
scored.append({**poi, "score": round(total_score, 6), "matched_interests": matched})
scored.sort(key=lambda p: p["score"], reverse=True)
return scored[:max_results]
@@ -0,0 +1,131 @@
"""Tests para match_pois_to_interests."""
from match_pois_to_interests import match_pois_to_interests
POIS_SAMPLE = [
{
"id": 1,
"lat": 40.41,
"lon": -3.70,
"name": "Museo del Prado",
"category": "museum",
"tags": {"tourism": "museum", "name": "Museo del Prado"},
},
{
"id": 2,
"lat": 40.42,
"lon": -3.71,
"name": "Bar El Tigre",
"category": "bar",
"tags": {"amenity": "bar", "name": "Bar El Tigre"},
},
{
"id": 3,
"lat": 40.43,
"lon": -3.72,
"name": "Galería de Arte Moderno",
"category": "gallery",
"tags": {"tourism": "gallery", "art": "modern"},
},
]
def test_pois_que_matchean_parcialmente_por_categoria_y_nombre():
"""pois que matchean parcialmente por categoria y nombre"""
interests = [{"name": "arte", "keywords": ["museo", "museum", "arte", "galería"], "weight": 1.0}]
result = match_pois_to_interests(POIS_SAMPLE, interests)
ids = [p["id"] for p in result]
# museo matchea por category "museum" (0.5) + nombre "Museo" (0.3) + tag "museum" (0.2) = 1.0
assert 1 in ids
# galería: sin match de categoria exacta (gallery != keywords), pero nombre contiene "Arte" -> check keywords
# keywords: "museo", "museum", "arte", "galería" — "arte" en "galería de arte moderno" -> 0.3
assert 3 in ids
# bar no matchea ninguno
assert 2 not in ids
# el museo debe tener score mas alto que la galeria
museo = next(p for p in result if p["id"] == 1)
galeria = next(p for p in result if p["id"] == 3)
assert museo["score"] > galeria["score"]
assert "arte" in museo["matched_interests"]
assert "arte" in galeria["matched_interests"]
def test_pois_sin_match_retorna_lista_vacia():
"""pois sin match retorna lista vacia"""
pois = [
{"id": 10, "lat": 40.0, "lon": -3.0, "name": "Parking Norte", "category": "parking", "tags": {"amenity": "parking"}},
{"id": 11, "lat": 40.1, "lon": -3.1, "name": "Gasolinera BP", "category": "fuel", "tags": {"amenity": "fuel"}},
]
interests = [{"name": "arte", "keywords": ["museo", "museum", "galería", "pintura"], "weight": 1.0}]
result = match_pois_to_interests(pois, interests)
assert result == []
def test_multiples_intereses_con_diferentes_weights_acumulan_score():
"""multiples intereses con diferentes weights acumulan score"""
pois = [
{
"id": 20,
"lat": 40.5,
"lon": -3.5,
"name": "Café Museo",
"category": "cafe",
"tags": {"amenity": "cafe", "tourism": "museum"},
}
]
interests = [
{"name": "arte", "keywords": ["museum", "museo", "arte"], "weight": 1.0},
{"name": "gastronomia", "keywords": ["cafe", "restaurant", "bar"], "weight": 0.8},
]
result = match_pois_to_interests(pois, interests)
assert len(result) == 1
poi = result[0]
# arte: nombre "Café Museo" contiene "museo" (0.3) + tag "museum" (0.2) = 0.5 * 1.0 = 0.5
# gastronomia: category "cafe" in keywords (0.5) + nombre "Café" contiene "cafe" (0.3) = 0.8 * 0.8 = 0.64
# total ~ 1.14
assert poi["score"] > 1.0
assert "arte" in poi["matched_interests"]
assert "gastronomia" in poi["matched_interests"]
assert len(poi["matched_interests"]) == 2
def test_max_results_limita_resultados():
pois = [
{"id": i, "lat": 40.0, "lon": -3.0, "name": f"Museo {i}", "category": "museum", "tags": {}}
for i in range(20)
]
interests = [{"name": "arte", "keywords": ["museum", "museo"], "weight": 1.0}]
result = match_pois_to_interests(pois, interests, max_results=5)
assert len(result) == 5
def test_inputs_no_mutados():
pois = [
{"id": 1, "lat": 40.0, "lon": -3.0, "name": "Museo A", "category": "museum", "tags": {}}
]
interests = [{"name": "arte", "keywords": ["museum"], "weight": 1.0}]
original_poi_keys = set(pois[0].keys())
match_pois_to_interests(pois, interests)
# el poi original no debe tener score ni matched_interests
assert set(pois[0].keys()) == original_poi_keys
if __name__ == "__main__":
test_pois_que_matchean_parcialmente_por_categoria_y_nombre()
print("PASS: pois que matchean parcialmente por categoria y nombre")
test_pois_sin_match_retorna_lista_vacia()
print("PASS: pois sin match retorna lista vacia")
test_multiples_intereses_con_diferentes_weights_acumulan_score()
print("PASS: multiples intereses con diferentes weights acumulan score")
test_max_results_limita_resultados()
print("PASS: max_results limita resultados")
test_inputs_no_mutados()
print("PASS: inputs no mutados")
print("---")
print("All tests passed.")
@@ -0,0 +1,65 @@
---
name: nominatim_reverse_geocode
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def nominatim_reverse_geocode(lat: float, lon: float, lang: str = 'es') -> dict"
description: "Reverse geocoding usando Nominatim (OpenStreetMap). Convierte coordenadas lat/lon a una dirección estructurada con calle, ciudad, país y otros campos normalizados."
tags: [geocoding, nominatim, openstreetmap, osm, reverse-geocoding, geo, location, address]
params:
- name: lat
desc: "Latitud en grados decimales (rango -90 a 90)."
- name: lon
desc: "Longitud en grados decimales (rango -180 a 180)."
- name: lang
desc: "Código de idioma ISO 639-1 para la respuesta (ej: 'es', 'en', 'fr'). Por defecto 'es'."
output: "Diccionario con campos normalizados: display_name (dirección completa legible), street (calle/road), house_number, neighbourhood, city (city→town→village), state, country, postcode, lat, lon, osm_type (node/way/relation), osm_id. Campos ausentes retornan string vacío."
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: ["urllib.request", "urllib.parse", "json"]
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/infra/nominatim_reverse_geocode.py"
---
## Ejemplo
```python
from infra.nominatim_reverse_geocode import nominatim_reverse_geocode
addr = nominatim_reverse_geocode(40.4168, -3.7038)
# {
# "display_name": "Puerta del Sol, Sol, Madrid, ...",
# "street": "Puerta del Sol",
# "house_number": "",
# "neighbourhood": "Sol",
# "city": "Madrid",
# "state": "Comunidad de Madrid",
# "country": "España",
# "postcode": "28013",
# "lat": 40.4168,
# "lon": -3.7038,
# "osm_type": "way",
# "osm_id": 12345678
# }
# Con idioma inglés
addr_en = nominatim_reverse_geocode(48.8566, 2.3522, lang="en")
# {"city": "Paris", "country": "France", ...}
```
## Notas
- Usa `urllib.request` de stdlib — sin dependencias externas.
- El header `User-Agent: fn_registry/1.0` es obligatorio según la política de uso de Nominatim.
- Timeout de 5 segundos. Para coordenadas en zonas remotas puede necesitarse más.
- Campo `city` se resuelve con fallback: `city → town → village → ""`.
- `lat`/`lon` del resultado son los del objeto OSM encontrado, que puede diferir ligeramente de los de entrada.
- Nominatim tiene límite de 1 request/segundo por IP. Para uso intensivo, considerar un servidor propio.
- Lanza `RuntimeError` en errores HTTP o de red.
@@ -0,0 +1,74 @@
"""Reverse geocoding usando Nominatim (OpenStreetMap)."""
import json
import urllib.request
import urllib.parse
def nominatim_reverse_geocode(lat: float, lon: float, lang: str = "es") -> dict:
"""Convierte coordenadas lat/lon a una dirección estructurada usando Nominatim.
Args:
lat: Latitud en grados decimales.
lon: Longitud en grados decimales.
lang: Código de idioma para la respuesta (ISO 639-1). Por defecto "es".
Returns:
Diccionario con campos normalizados de la dirección.
Raises:
RuntimeError: Si la petición HTTP falla o el servidor retorna error.
"""
params = urllib.parse.urlencode({
"format": "jsonv2",
"lat": str(lat),
"lon": str(lon),
"accept-language": lang,
"zoom": "18",
})
url = f"https://nominatim.openstreetmap.org/reverse?{params}"
req = urllib.request.Request(
url,
headers={"User-Agent": "fn_registry/1.0"},
)
try:
with urllib.request.urlopen(req, timeout=5) as resp:
if resp.status != 200:
raise RuntimeError(
f"nominatim_reverse_geocode: HTTP {resp.status} para lat={lat}, lon={lon}"
)
data = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
raise RuntimeError(
f"nominatim_reverse_geocode: HTTP {e.code} para lat={lat}, lon={lon}"
) from e
except urllib.error.URLError as e:
raise RuntimeError(
f"nominatim_reverse_geocode: error de red — {e.reason}"
) from e
address = data.get("address", {})
city = (
address.get("city")
or address.get("town")
or address.get("village")
or ""
)
return {
"display_name": data.get("display_name", ""),
"street": address.get("road", ""),
"house_number": address.get("house_number", ""),
"neighbourhood": address.get("neighbourhood", ""),
"city": city,
"state": address.get("state", ""),
"country": address.get("country", ""),
"postcode": address.get("postcode", ""),
"lat": float(data.get("lat", lat)),
"lon": float(data.get("lon", lon)),
"osm_type": data.get("osm_type", ""),
"osm_id": int(data.get("osm_id", 0)),
}
+56
View File
@@ -0,0 +1,56 @@
---
name: ollama_chat
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def ollama_chat(messages: list[dict], model: str = 'llama3.1:8b', base_url: str = 'http://localhost:11434', temperature: float = 0.7, max_tokens: int = 1024) -> dict"
description: "Envía una solicitud de chat completion a Ollama (API local compatible con OpenAI). Retorna el contenido generado junto a métricas de duración y tokens."
tags: [ollama, llm, chat, inference, local-ai, http]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [json, urllib.request, urllib.error]
params:
- name: messages
desc: "Lista de mensajes de la conversación. Cada mensaje es un dict con 'role' (system|user|assistant) y 'content' (texto del mensaje)."
- name: model
desc: "Nombre del modelo Ollama a usar (ej: llama3.1:8b, mistral:7b, codellama:13b)."
- name: base_url
desc: "URL base de la instancia Ollama local. Por defecto http://localhost:11434."
- name: temperature
desc: "Temperatura de generación entre 0.0 (determinista) y 1.0 (creativo). Por defecto 0.7."
- name: max_tokens
desc: "Número máximo de tokens a generar en la respuesta. Por defecto 1024."
output: "Dict con 'content' (texto generado), 'model' (modelo usado), 'total_duration_ms' (duración total en milisegundos) y 'eval_count' (tokens generados)."
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/infra/ollama_chat.py"
---
## Ejemplo
```python
resp = ollama_chat([
{"role": "system", "content": "Eres un guía turístico experto."},
{"role": "user", "content": "¿Qué puedo ver cerca del Museo del Prado?"}
], model="llama3.1:8b")
# {"content": "Cerca del Museo del Prado encontrarás...", "model": "llama3.1:8b", "total_duration_ms": 3200, "eval_count": 150}
```
## Notas
Función impura: hace una solicitud HTTP POST a la API de Ollama. Solo usa stdlib Python (urllib.request, json) — sin dependencias externas.
La generación de LLMs puede ser lenta; el timeout está fijo en 60 segundos. Para modelos grandes o respuestas largas, ajustar `max_tokens` o considerar streaming.
Manejo de errores:
- Connection refused → `RuntimeError("Ollama no está corriendo en {base_url}")`
- HTTP error → `RuntimeError("Ollama retornó HTTP {code}: {reason}")`
- Otros errores de red → `RuntimeError("Error de conexión con Ollama: {reason}")`
El campo `total_duration` que retorna Ollama está en nanosegundos; se convierte a milisegundos para mayor legibilidad. El campo `eval_count` puede no estar presente en todas las versiones de Ollama (default 0).
+70
View File
@@ -0,0 +1,70 @@
"""Envía una solicitud de chat completion a Ollama (API local compatible con OpenAI)."""
import json
import urllib.request
import urllib.error
from typing import Any
def ollama_chat(
messages: list[dict],
model: str = "llama3.1:8b",
base_url: str = "http://localhost:11434",
temperature: float = 0.7,
max_tokens: int = 1024,
) -> dict:
"""Envía una solicitud de chat completion a Ollama.
Args:
messages: Lista de mensajes con formato {"role": "system"|"user"|"assistant", "content": str}.
model: Nombre del modelo Ollama a usar.
base_url: URL base de la instancia Ollama.
temperature: Temperatura de generación (0.0 - 1.0).
max_tokens: Número máximo de tokens a generar.
Returns:
Dict con content, model, total_duration_ms y eval_count.
Raises:
RuntimeError: Si Ollama no está corriendo o retorna error HTTP.
"""
url = f"{base_url}/api/chat"
payload = {
"model": model,
"messages": messages,
"stream": False,
"options": {
"temperature": temperature,
"num_predict": max_tokens,
},
}
body = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
url,
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=60) as resp:
raw = resp.read()
except urllib.error.URLError as exc:
reason = str(exc.reason) if hasattr(exc, "reason") else str(exc)
if "connection refused" in reason.lower() or "111" in reason:
raise RuntimeError(f"Ollama no está corriendo en {base_url}") from exc
raise RuntimeError(f"Error de conexión con Ollama: {reason}") from exc
except urllib.error.HTTPError as exc:
raise RuntimeError(f"Ollama retornó HTTP {exc.code}: {exc.reason}") from exc
response: dict[str, Any] = json.loads(raw)
content: str = response["message"]["content"]
total_duration_ns: int = response.get("total_duration", 0)
total_duration_ms: int = int(total_duration_ns / 1_000_000)
eval_count: int = response.get("eval_count", 0)
return {
"content": content,
"model": response.get("model", model),
"total_duration_ms": total_duration_ms,
"eval_count": eval_count,
}
@@ -0,0 +1,80 @@
---
name: overpass_nearby_pois
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "overpass_nearby_pois(lat: float, lon: float, radius_m: int = 500, categories: list[str] | None = None) -> list[dict]"
description: "Consulta la Overpass API (OpenStreetMap) para obtener POIs cercanos a una coordenada. Soporta 16 categorias mapeadas a tags OSM (amenity, tourism, historic, leisure, shop). Sin dependencias externas, solo stdlib."
tags: [osm, openstreetmap, overpass, poi, geolocation, nearby, geocoding, infra, http, stdlib]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: ["json", "urllib.error", "urllib.parse", "urllib.request"]
params:
- name: lat
desc: "latitud del punto central de busqueda (grados decimales, -90 a 90)"
- name: lon
desc: "longitud del punto central de busqueda (grados decimales, -180 a 180)"
- name: radius_m
desc: "radio de busqueda en metros alrededor del punto central (defecto 500)"
- name: categories
desc: "lista de categorias a buscar entre: restaurant, cafe, bar, museum, monument, park, library, theatre, cinema, gallery, historic, tourism, shop, hotel, pharmacy, hospital. Si es None, usa todas las categorias comunes excepto hotel, pharmacy, hospital"
output: "lista de dicts con campos id (int), lat (float), lon (float), name (str), category (str) y tags (dict con todos los tags OSM del nodo)"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/infra/overpass_nearby_pois.py"
---
## Ejemplo
```python
from infra.overpass_nearby_pois import overpass_nearby_pois
# POIs en el centro de Madrid (radio 300m)
pois = overpass_nearby_pois(40.4168, -3.7038, radius_m=300, categories=["museum", "monument"])
# [{"id": 123, "lat": 40.417, "lon": -3.704, "name": "Museo del Prado", "category": "museum", "tags": {...}}, ...]
# Todos los tipos comunes en radio 500m (defecto)
all_pois = overpass_nearby_pois(48.8566, 2.3522)
# Bares y restaurantes cerca de la Sagrada Familia
food = overpass_nearby_pois(41.4036, 2.1744, radius_m=200, categories=["restaurant", "cafe", "bar"])
```
## Mapeo de categorias a tags OSM
| Categoria | Tag OSM |
|-------------|----------------------|
| restaurant | amenity=restaurant |
| cafe | amenity=cafe |
| bar | amenity=bar |
| museum | tourism=museum |
| monument | historic=monument |
| park | leisure=park |
| library | amenity=library |
| theatre | amenity=theatre |
| cinema | amenity=cinema |
| gallery | tourism=gallery |
| historic | historic=* |
| tourism | tourism=* |
| shop | shop=* |
| hotel | tourism=hotel |
| pharmacy | amenity=pharmacy |
| hospital | amenity=hospital |
## Notas
Solo usa stdlib (urllib.request, urllib.parse, json). Sin dependencias externas.
La query Overpass QL usa union de nodos `(node[...]; node[...];)` con `out body` para obtener coordenadas y tags completos.
El endpoint publico `https://overpass-api.de/api/interpreter` tiene rate limits informales — para uso intensivo considerar un servidor Overpass propio o la instancia de kumi.systems.
El campo `category` refleja la primera coincidencia en el diccionario interno `_CATEGORY_TAGS`, que sigue el orden de insercion de Python 3.7+. Si un nodo tiene multiples tags relevantes (ej: un bar que tambien es restaurante), la categoria asignada sera la primera que matchee.
Timeout hardcodeado a 10 segundos tanto en la directiva Overpass QL `[timeout:10]` como en `urlopen(timeout=10)`.
@@ -0,0 +1,138 @@
"""Consulta la Overpass API para obtener POIs cercanos a una coordenada."""
import json
import urllib.error
import urllib.parse
import urllib.request
# Mapeo de categoría legible → (key OSM, value OSM)
# None como value significa wildcard (key=*)
_CATEGORY_TAGS: dict[str, tuple[str, str | None]] = {
"restaurant": ("amenity", "restaurant"),
"cafe": ("amenity", "cafe"),
"bar": ("amenity", "bar"),
"museum": ("tourism", "museum"),
"monument": ("historic", "monument"),
"park": ("leisure", "park"),
"library": ("amenity", "library"),
"theatre": ("amenity", "theatre"),
"cinema": ("amenity", "cinema"),
"gallery": ("tourism", "gallery"),
"historic": ("historic", None),
"tourism": ("tourism", None),
"shop": ("shop", None),
"hotel": ("tourism", "hotel"),
"pharmacy": ("amenity", "pharmacy"),
"hospital": ("amenity", "hospital"),
}
_DEFAULT_CATEGORIES = [
"restaurant", "cafe", "bar", "museum", "monument", "park",
"library", "theatre", "cinema", "gallery", "historic", "tourism", "shop",
]
_OVERPASS_ENDPOINT = "https://overpass-api.de/api/interpreter"
def _build_query(lat: float, lon: float, radius_m: int, categories: list[str]) -> str:
"""Construye una query Overpass QL con union de nodos."""
lines = ["[out:json][timeout:10];", "("]
for cat in categories:
if cat not in _CATEGORY_TAGS:
continue
key, value = _CATEGORY_TAGS[cat]
if value is None:
lines.append(f' node["{key}"](around:{radius_m},{lat},{lon});')
else:
lines.append(f' node["{key}"="{value}"](around:{radius_m},{lat},{lon});')
lines.append(");")
lines.append("out body;")
return "\n".join(lines)
def _extract_category(tags: dict) -> str:
"""Determina la categoría legible a partir de los tags OSM del nodo."""
for cat, (key, value) in _CATEGORY_TAGS.items():
if key not in tags:
continue
if value is None or tags[key] == value:
return cat
return "unknown"
def overpass_nearby_pois(
lat: float,
lon: float,
radius_m: int = 500,
categories: list[str] | None = None,
) -> list[dict]:
"""Obtiene POIs cercanos a una coordenada usando la Overpass API.
Args:
lat: latitud del punto central.
lon: longitud del punto central.
radius_m: radio de busqueda en metros (defecto 500).
categories: lista de categorias a buscar. Si es None, usa todas las
categorias comunes definidas internamente.
Returns:
lista de dicts con id, lat, lon, name, category y tags de cada POI.
Raises:
RuntimeError: si la request falla por error HTTP, timeout u otro error de red.
"""
active_categories = categories if categories is not None else _DEFAULT_CATEGORIES
query = _build_query(lat, lon, radius_m, active_categories)
encoded_query = urllib.parse.quote(query)
form_body = f"data={encoded_query}".encode("utf-8")
req = urllib.request.Request(
_OVERPASS_ENDPOINT,
data=form_body,
method="POST",
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
try:
with urllib.request.urlopen(req, timeout=10) as response:
raw = response.read()
except urllib.error.HTTPError as e:
body_preview = e.read(200).decode("utf-8", errors="replace")
raise RuntimeError(
f"Overpass API HTTP error {e.code} for ({lat},{lon}) r={radius_m}m: {body_preview}"
) from e
except urllib.error.URLError as e:
raise RuntimeError(
f"Overpass API request failed for ({lat},{lon}) r={radius_m}m: {e.reason}"
) from e
try:
result = json.loads(raw)
except json.JSONDecodeError as e:
raise RuntimeError(
f"Overpass API returned invalid JSON for ({lat},{lon}): {e}"
) from e
pois: list[dict] = []
for element in result.get("elements", []):
if element.get("type") != "node":
continue
tags = element.get("tags", {})
category = _extract_category(tags)
# Nombre: tag "name", o valor del tag de categoría, o "Unknown"
if "name" in tags:
name = tags["name"]
else:
key = _CATEGORY_TAGS.get(category, (None, None))[0]
name = tags.get(key, "Unknown") if key else "Unknown"
pois.append({
"id": element["id"],
"lat": element["lat"],
"lon": element["lon"],
"name": name,
"category": category,
"tags": tags,
})
return pois