Spaces:
Running
Running
Place search (Open-Meteo + Gemini fallback), remembered farm location, small UI fixes
00b75ab verified | """Place-name search: Open-Meteo geocoding first, Gemini with Google Search as a fallback.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import os | |
| import re | |
| from typing import Any | |
| import requests | |
| LOGGER = logging.getLogger(__name__) | |
| OPEN_METEO_GEOCODING_URL = "https://geocoding-api.open-meteo.com/v1/search" | |
| GEOCODING_TIMEOUT_SECONDS = 10 | |
| GEMINI_GEOCODING_TIMEOUT_MS = 20_000 | |
| MAX_RESULTS = 5 | |
| # Generous bounding box around Pakistan; anything outside is rejected. | |
| PAKISTAN_BOUNDS = {"lat": (23.0, 37.5), "lon": (60.5, 77.5)} | |
| def _in_pakistan(latitude: Any, longitude: Any) -> bool: | |
| try: | |
| lat, lon = float(latitude), float(longitude) | |
| except (TypeError, ValueError): | |
| return False | |
| return ( | |
| PAKISTAN_BOUNDS["lat"][0] <= lat <= PAKISTAN_BOUNDS["lat"][1] | |
| and PAKISTAN_BOUNDS["lon"][0] <= lon <= PAKISTAN_BOUNDS["lon"][1] | |
| ) | |
| def search_open_meteo(query: str) -> list[dict[str, Any]]: | |
| response = requests.get( | |
| OPEN_METEO_GEOCODING_URL, | |
| params={"name": query, "count": MAX_RESULTS, "language": "en", "format": "json", "countryCode": "PK"}, | |
| timeout=GEOCODING_TIMEOUT_SECONDS, | |
| ) | |
| response.raise_for_status() | |
| results = [] | |
| for item in response.json().get("results") or []: | |
| if not _in_pakistan(item.get("latitude"), item.get("longitude")): | |
| continue | |
| parts = [item.get("name"), item.get("admin2"), item.get("admin1")] | |
| name = ", ".join(dict.fromkeys(part for part in parts if part)) | |
| results.append({ | |
| "name": name, | |
| "latitude": round(float(item["latitude"]), 4), | |
| "longitude": round(float(item["longitude"]), 4), | |
| "source": "open-meteo", | |
| }) | |
| return results | |
| def search_gemini(query: str) -> list[dict[str, Any]]: | |
| api_key = os.getenv("GEMINI_API_KEY") | |
| if not api_key: | |
| return [] | |
| from google import genai | |
| from google.genai import types | |
| client = genai.Client(api_key=api_key, http_options=types.HttpOptions(timeout=GEMINI_GEOCODING_TIMEOUT_MS)) | |
| prompt = ( | |
| "Find the geographic coordinates of this place in Pakistan, using Google Search to verify: " | |
| f"{json.dumps(query)}.\n" | |
| 'Reply with only a JSON object: {"name": "<place, district, province>", "latitude": <number>, "longitude": <number>}. ' | |
| 'If you cannot identify the place confidently, reply {"name": null}.' | |
| ) | |
| # Search grounding cannot be combined with a JSON response schema, so parse the text. | |
| response = client.models.generate_content( | |
| model=os.getenv("GEMINI_MODEL", "gemini-2.5-flash"), | |
| contents=prompt, | |
| config=types.GenerateContentConfig( | |
| tools=[types.Tool(google_search=types.GoogleSearch())], | |
| temperature=0, | |
| ), | |
| ) | |
| match = re.search(r"\{.*?\}", response.text or "", re.DOTALL) | |
| if not match: | |
| return [] | |
| data = json.loads(match.group(0)) | |
| if not data.get("name") or not _in_pakistan(data.get("latitude"), data.get("longitude")): | |
| return [] | |
| return [{ | |
| "name": str(data["name"])[:200], | |
| "latitude": round(float(data["latitude"]), 4), | |
| "longitude": round(float(data["longitude"]), 4), | |
| "source": "gemini", | |
| }] | |
| def search_places(query: str) -> list[dict[str, Any]]: | |
| """Return matching places in Pakistan; an empty list when nothing is found.""" | |
| try: | |
| results = search_open_meteo(query) | |
| if results: | |
| return results | |
| except (requests.RequestException, ValueError, KeyError): | |
| LOGGER.exception("Open-Meteo geocoding failed for %r", query) | |
| try: | |
| return search_gemini(query) | |
| except Exception: | |
| LOGGER.exception("Gemini geocoding failed for %r", query) | |
| return [] | |