"""Voice Guide — Backend FastAPI. Compone funciones del registry para guiar al usuario por voz según su ubicación e intereses. """ import sys import os import json from pathlib import Path from typing import Optional from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel # Registry functions path _registry_root = Path(__file__).resolve().parents[3] sys.path.insert(0, str(_registry_root / "python" / "functions")) from infra.overpass_nearby_pois import overpass_nearby_pois from infra.nominatim_reverse_geocode import nominatim_reverse_geocode from infra.ollama_chat import ollama_chat from core.match_pois_to_interests import match_pois_to_interests from core.build_guide_prompt import build_guide_prompt app = FastAPI(title="Voice Guide API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # --- In-memory store (per session, no DB needed) --- _interests: list[dict] = [] _INTERESTS_FILE = Path(__file__).parent / "interests.json" def _load_interests() -> list[dict]: global _interests if _INTERESTS_FILE.exists(): _interests = json.loads(_INTERESTS_FILE.read_text()) return _interests def _save_interests() -> None: _INTERESTS_FILE.write_text(json.dumps(_interests, ensure_ascii=False, indent=2)) _load_interests() # --- Models --- class Interest(BaseModel): name: str keywords: list[str] weight: float = 1.0 class GuideRequest(BaseModel): lat: float lon: float query: str = "" radius_m: int = 500 categories: Optional[list[str]] = None model: str = "llama3.1:8b" ollama_url: str = "http://localhost:11434" # --- Endpoints --- @app.get("/api/interests") def list_interests(): return _interests @app.post("/api/interests") def add_interest(interest: Interest): _interests.append(interest.model_dump()) _save_interests() return {"ok": True, "interests": _interests} @app.delete("/api/interests/{index}") def delete_interest(index: int): if 0 <= index < len(_interests): removed = _interests.pop(index) _save_interests() return {"ok": True, "removed": removed} return {"ok": False, "error": "index out of range"} @app.post("/api/guide") def guide(req: GuideRequest): """Pipeline principal: ubicación + POIs + intereses + LLM → guía.""" # 1. Reverse geocode try: location = nominatim_reverse_geocode(req.lat, req.lon) except RuntimeError: location = {"display_name": "", "street": "", "house_number": "", "neighbourhood": "", "city": "", "state": "", "country": "", "postcode": "", "lat": req.lat, "lon": req.lon, "osm_type": "", "osm_id": 0} # 2. POIs cercanos try: pois = overpass_nearby_pois( req.lat, req.lon, radius_m=req.radius_m, categories=req.categories, ) except RuntimeError: pois = [] # 3. Filtrar por intereses if _interests: matched = match_pois_to_interests(pois, _interests, max_results=10) else: # Sin intereses, devolver todos con score 0 matched = [ {**p, "score": 0, "matched_interests": []} for p in pois[:10] ] # 4. Construir prompt messages = build_guide_prompt( location=location, pois=matched, interests=_interests, user_query=req.query, ) # 5. LLM try: llm_resp = ollama_chat( messages=messages, model=req.model, base_url=req.ollama_url, ) guide_text = llm_resp["content"] except RuntimeError as e: guide_text = _fallback_guide(location, matched, req.query) return { "guide": guide_text, "location": location, "pois": matched, "interests": _interests, } def _fallback_guide(location: dict, pois: list[dict], query: str) -> str: """Respuesta sin LLM cuando Ollama no está disponible.""" city = location.get("city", "tu zona") street = location.get("street", "") where = f"{street}, {city}" if street else city if not pois: return f"Estás en {where}. No encontré lugares destacados cerca con tus intereses." names = ", ".join(p["name"] for p in pois[:3]) return f"Estás en {where}. Cerca tienes: {names}." @app.get("/api/health") def health(): return {"status": "ok"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8787)