Upload 9 files

planmate/attractions.py

@@ -0,0 +1,79 @@
+# planmate/attractions.py
+
+import requests
+from typing import List, Dict
+from .config import OPENTRIPMAP_BASE, get_opentripmap_key
+
+def _get(path: str, params: Dict):
+    url = f"{OPENTRIPMAP_BASE}{path}"
+    params = dict(params or {})
+    params["apikey"] = get_opentripmap_key()
+    r = requests.get(url, params=params, timeout=30)
+    try:
+        r.raise_for_status()
+    except requests.HTTPError as e:
+        # make API errors easier to debug in Streamlit
+        snippet = (r.text or "")[:300]
+        raise RuntimeError(f"OpenTripMap error {r.status_code}: {snippet}") from e
+    return r.json()
+
+def get_poi_radius(lat: float, lon: float, radius=7000, kinds="interesting_places", limit=50) -> List[Dict]:
+    return _get(
+        "/places/radius",
+        {
+            "lat": lat,
+            "lon": lon,
+            "radius": radius,
+            "kinds": kinds,
+            "format": "json",
+            "limit": limit,
+            "rate": 2,
+        },
+    )
+
+def get_details(xid: str) -> Dict:
+    return _get(f"/places/xid/{xid}", {})
+
+def enrich_pois(pois: List[Dict], fetch_details=False) -> List[Dict]:
+    out = []
+    for p in pois:
+        item = {
+            "name": p.get("name") or "(Unnamed)",
+            "dist": p.get("dist"),
+            "rate": p.get("rate"),
+            "kinds": p.get("kinds", ""),
+            "xid": p.get("xid"),
+            "point": p.get("point", {}),
+        }
+        if fetch_details and item["xid"]:
+            try:
+                det = get_details(item["xid"])
+                item["wikipedia"] = det.get("wikipedia")
+                item["url"] = det.get("url")
+                item["address"] = det.get("address", {})
+                item["otm"] = det.get("otm")
+            except Exception:
+                pass
+        out.append(item)
+    return out
+
+def get_attractions_and_stays(lat: float, lon: float, radius=7000, limit=40):
+    # Attractions remain unchanged
+    attractions_kinds = "interesting_places,cultural,historic,museums,architecture"
+
+    # IMPORTANT: Use taxonomy accepted by OpenTripMap.
+    # "accomodations" (sic) is the umbrella category; for hotels use "other_hotels", not "hotels".
+    # See OTM catalog: Accomodations (accomodations), Hotels (other_hotels), Hostels (hostels).
+    # https://dev.opentripmap.org/catalog
+    stays_kinds = (
+        "accomodations,other_hotels,hostels,apartments,guest_houses,"
+        "resorts,motels,villas_and_chalet,alpine_hut"
+    )
+
+    attractions = get_poi_radius(lat, lon, radius, attractions_kinds, limit)
+    stays = get_poi_radius(lat, lon, radius, stays_kinds, limit=30)
+
+    return {
+        "attractions": enrich_pois(attractions, fetch_details=False),
+        "stays": enrich_pois(stays, fetch_details=False),
+    }

planmate/config.py

@@ -0,0 +1,42 @@
+import os
+from dotenv import load_dotenv
+
+# Load .env if present (local dev convenience)
+load_dotenv()
+
+APP_TITLE = "PlanMate"
+APP_TAGLINE = "AI Power smart trip planner"
+
+THEME = {
+    "bg": "#fcfcfc",
+    "text": "#383838",
+    "label": "#153d15",
+    "border": "#153d15",
+}
+
+CURRENCY = "PKR"
+UNITS = "metric"
+LANGUAGE = "en"
+
+# API Base URLs
+AMADEUS_BASE = "https://test.api.amadeus.com"
+OPENWEATHER_BASE = "https://api.openweathermap.org"
+OPENTRIPMAP_BASE = "https://api.opentripmap.com/0.1/en"
+
+def get_env(key: str) -> str:
+    val = os.getenv(key)
+    if not val:
+        raise RuntimeError(f"Missing required environment variable: {key}")
+    return val
+
+def get_gemini_key():
+    return get_env("GEMINI_API_KEY")
+
+def get_amadeus_credentials():
+    return get_env("AMADEUS_CLIENT_ID"), get_env("AMADEUS_CLIENT_SECRET")
+
+def get_openweather_key():
+    return get_env("OPENWEATHER_API_KEY")
+
+def get_opentripmap_key():
+    return get_env("OPENTRIPMAP_API_KEY")

planmate/flights.py

@@ -0,0 +1,100 @@
+from typing import List, Dict, Tuple
+from .clients.amadeus_client import AmadeusClient
+from .utils import format_price
+
+
+def resolve_city_to_iata(city_name: str) -> Tuple[str, str]:
+    """Resolve a human city name to an IATA city code using Amadeus.
+    Returns (iata_code, display_label). Raises RuntimeError if not found.
+    """
+    client = AmadeusClient()
+    res = client.locations(city_name, subtypes="CITY")
+    data = res.get("data", [])
+    if not data:
+        # try airports as fallback and take the cityName + iataCode of airport
+        res = client.locations(city_name, subtypes="AIRPORT")
+        data = res.get("data", [])
+    if not data:
+        raise RuntimeError(f"Could not resolve city to IATA: {city_name}")
+
+    # Prefer exact name match then highest relevance (first)
+    city_lower = city_name.strip().lower()
+    chosen = None
+    for d in data:
+        name = (d.get("name") or "").lower()
+        address_city = (d.get("address", {}).get("cityName") or "").lower()
+        if name == city_lower or address_city == city_lower:
+            chosen = d
+            break
+    if not chosen:
+        chosen = data[0]
+
+    code = chosen.get("iataCode")
+    label = f"{chosen.get('name')} ({code})"
+    if not code:
+        raise RuntimeError(f"No IATA code found for: {city_name}")
+    return code, label
+
+
+def search_airports(keyword: str) -> List[Dict]:
+    client = AmadeusClient()
+    res = client.locations(keyword)
+    items = []
+    for d in res.get("data", []):
+        items.append(
+            {
+                "name": d.get("name"),
+                "iataCode": d.get("iataCode"),
+                "subType": d.get("subType"),
+                "address": d.get("address", {}).get("cityName")
+                or d.get("address", {}).get("countryName"),
+            }
+        )
+    return items
+
+
+def search_flights(
+    origin_code: str,
+    dest_code: str,
+    depart: str,
+    ret: str,
+    adults=1,
+    currency="PKR",
+    non_stop: bool = False,
+) -> Dict:
+    client = AmadeusClient()
+    res = client.flight_offers(
+        origin_code, dest_code, depart, ret, adults, currency, non_stop
+    )
+    dicts = res.get("dictionaries", {})
+    carriers = dicts.get("carriers", {})
+    result = []
+    for offer in res.get("data", []):
+        price = offer.get("price", {}).get("total")
+        itineraries = offer.get("itineraries", [])
+        legs = []
+        for it in itineraries:
+            segs = []
+            for s in it.get("segments", []):
+                carrier = s.get("carrierCode")
+                segs.append(
+                    {
+                        "from": s.get("departure", {}).get("iataCode"),
+                        "to": s.get("arrival", {}).get("iataCode"),
+                        "dep": s.get("departure", {}).get("at"),
+                        "arr": s.get("arrival", {}).get("at"),
+                        "carrier": carriers.get(carrier, carrier),
+                        "number": s.get("number"),
+                        "duration": it.get("duration", ""),
+                    }
+                )
+            legs.append(segs)
+        result.append(
+            {
+                "price": price,
+                "price_label": format_price(price, currency),
+                "legs": legs,
+                "oneWay": (ret is None or ret == ""),
+            }
+        )
+    return {"flights": result, "carriers": carriers}

planmate/itinerary.py

@@ -0,0 +1,15 @@
+from typing import List, Dict
+from .llm import generate_itinerary
+
+
+def build_itinerary(
+    city: str,
+    start_date: str,
+    days: int,
+    selected_attractions: List[Dict],
+    selected_stay: Dict | None,
+    weather_summary: str,
+) -> str:
+    return generate_itinerary(
+        city, start_date, days, selected_attractions, selected_stay, weather_summary
+    )

planmate/llm.py

@@ -0,0 +1,136 @@
+import json
+from typing import List, Dict, Optional, Tuple
+import google.generativeai as genai
+from .config import get_gemini_key
+from .utils import is_iata_code
+
+# Initialize once
+_gen_inited = False
+
+def _ensure_init():
+    global _gen_inited
+    if not _gen_inited:
+        genai.configure(api_key=get_gemini_key())
+        _gen_inited = True
+
+
+def _model():
+    _ensure_init()
+    return genai.GenerativeModel(
+        model_name="gemini-2.0-flash-lite",
+        generation_config={
+            "temperature": 0.2,
+            "top_p": 0.9,
+            "top_k": 40,
+            "max_output_tokens": 600,
+        },
+    )
+
+
+def resolve_city_to_iata_ai(city: str, country_hint: Optional[str] = None) -> Tuple[str, str, str]:
+    """Use the LLM to resolve a human-readable city to an IATA code.
+    Returns (code, canonical_name, kind) where kind is 'CITY' or 'AIRPORT'.
+    Raises RuntimeError on failure.
+    """
+    m = _model()
+    prompt = f"""
+You convert a city name into a 3-letter IATA code for flight search.
+Rules:
+- If the metro area has multiple major airports with an aggregate IATA city code, return the CITY code (e.g., New York→NYC, London→LON, Paris→PAR, Tokyo→TYO).
+- If the city typically uses a single primary commercial airport, return that AIRPORT code (e.g., Dubai→DXB, Doha→DOH, Lahore→LHE, Karachi→KHI, Istanbul→IST).
+- Return empty code "" if no suitable airport exists.
+- The code MUST be exactly 3 uppercase letters A–Z.
+- Respond with ONLY a compact JSON object, no additional text, using this schema:
+  {{"code":"XXX","name":"Canonical City or Airport Name","kind":"CITY|AIRPORT","alternates":["AAA","BBB"]}}
+
+Input:
+- city: {city}
+- country_hint: {country_hint or 'unknown'}
+"""
+    resp = m.generate_content(prompt)
+    text = (resp.text or "").strip()
+
+    # Extract JSON block
+    mjson = None
+    if "{" in text:
+        try:
+            start = text.find("{")
+            end = text.rfind("}") + 1
+            mjson = json.loads(text[start:end])
+        except Exception:
+            pass
+    if not mjson or not isinstance(mjson, dict):
+        raise RuntimeError("AI could not produce a valid JSON mapping for the city.")
+
+    code = str(mjson.get("code", "")).strip().upper()
+    name = str(mjson.get("name", city)).strip()
+    kind = str(mjson.get("kind", "")).strip().upper()
+
+    if not is_iata_code(code):
+        raise RuntimeError(f"AI returned an invalid IATA code: {code}")
+    if kind not in ("CITY", "AIRPORT"):
+        kind = "AIRPORT"
+    return code, name, kind
+
+
+def rank_accommodations(accommodations: List[Dict], prefs: str = "") -> List[Dict]:
+    """Add an 'llm_score' to each accommodation and sort by it."""
+    if not accommodations:
+        return []
+    m = _model()
+    lines = "\n".join(
+        [
+            f"- {a.get('name','(no name)')} | rate:{a.get('rate')} | dist:{int(a.get('dist',0))}m | kinds:{a.get('kinds','')}"
+            for a in accommodations[:50]
+        ]
+    )
+    prompt = f"""Rank the following accommodations for a tourist trip. Prefer central, well-rated options.
+User preferences: {prefs or 'not specified'}.
+Return a JSON array of objects with 'name' and an integer 'score' 1-100.
+
+Items:
+{lines}
+"""
+    resp = m.generate_content(prompt)
+    try:
+        text = resp.text or ""
+        if "```json" in text:
+            data = json.loads(text.split("```json")[-1].split("```")[0])
+        else:
+            data = json.loads(text)
+    except Exception:
+        data = []
+    score_map = {d.get("name", ""): int(d.get("score", 50)) for d in data if isinstance(d, dict)}
+    for a in accommodations:
+        a["llm_score"] = score_map.get(a.get("name", ""), 50)
+    return sorted(accommodations, key=lambda x: x.get("llm_score", 50), reverse=True)
+
+
+def generate_itinerary(
+    city: str,
+    start_date: str,
+    days: int,
+    selected_attractions: List[Dict],
+    selected_stay: Dict | None,
+    weather_summary: str,
+) -> str:
+    m = _model()
+    attractions_text = "\n".join([f"- {a.get('name')} ({a.get('kinds','')})" for a in selected_attractions])
+    hotel_text = selected_stay.get("name") if selected_stay else "TBD"
+    prompt = f"""
+Create a practical, day-by-day itinerary for a trip.
+
+City: {city}
+Start Date: {start_date}
+Days: {days}
+Hotel: {hotel_text}
+Weather (summary): {weather_summary}
+
+Attractions to consider:
+{attractions_text}
+
+Constraints: group nearby sights, account for weather, add meal/time suggestions, include commute notes.
+Return Markdown with sections Day 1, Day 2, ..., and a brief daily plan (morning/afternoon/evening).
+"""
+    resp = m.generate_content(prompt)
+    return resp.text

planmate/utils.py

@@ -0,0 +1,40 @@
+from datetime import date, datetime, timedelta
+from typing import List
+import pytz
+import re
+
+def clamp_days(n: int, min_v=1, max_v=30):
+    return max(min_v, min(max_v, n))
+
+
+def to_iso(d: date) -> str:
+    return d.strftime("%Y-%m-%d")
+
+
+def parse_date_str(s: str) -> date:
+    return datetime.strptime(s, "%Y-%m-%d").date()
+
+
+def compute_return_date(start: date, days: int) -> date:
+    days = clamp_days(days)
+    return start + timedelta(days=days)
+
+
+def format_price(amount: str, currency: str = "PKR") -> str:
+    try:
+        f = float(amount)
+        return f"{currency} {f:,.0f}"
+    except Exception:
+        return f"{currency} {amount}"
+
+
+def day_list(start: date, days: int) -> List[date]:
+    days = clamp_days(days)
+    return [start + timedelta(days=i) for i in range(days)]
+
+
+def as_local(dt: datetime, tz="Asia/Karachi") -> datetime:
+    return dt.astimezone(pytz.timezone(tz))
+
+def is_iata_code(s: str) -> bool:
+    return bool(re.fullmatch(r"[A-Z]{3}", s or ""))

planmate/validation.py

@@ -0,0 +1,37 @@
+from typing import Dict
+from .clients.amadeus_client import AmadeusClient
+from .weather import geocode_city
+from .config import MAX_AIRPORT_KM
+
+
+def nearest_airports_for_city(city: str, max_km: int = MAX_AIRPORT_KM) -> Dict:
+    """
+    Returns nearest airports to the given city (within max_km).
+    Output: {"location": {...}, "airports": [airport_dicts_with__distance_km], "raw": raw_list}
+    Raises on geocoding or network errors.
+    """
+    loc = geocode_city(city)  # {name, lat, lon, country}
+    client = AmadeusClient()
+    res = client.airports_nearby(lat=loc["lat"], lon=loc["lon"])
+
+    filtered = []
+    for a in res.get("data", []):
+        dist = a.get("distance", {})
+        val = dist.get("value")
+        unit = (dist.get("unit") or "").upper()
+        if val is None:
+            continue
+        km = float(val) if unit == "KM" else float(val) * 1.60934
+        a["_distance_km"] = km
+        if km <= max_km:
+            filtered.append(a)
+
+    return {"location": loc, "airports": filtered, "raw": res.get("data", [])}
+
+
+def assert_city_has_airport(city: str, max_km: int = MAX_AIRPORT_KM) -> Dict:
+    """Validates a city has at least one airport within max_km. Raises RuntimeError otherwise."""
+    data = nearest_airports_for_city(city, max_km)
+    if len(data["airports"]) == 0:
+        raise RuntimeError(f"No airport found within {max_km} km of '{city}'. Please try a nearby larger city.")
+    return data

planmate/weather.py

@@ -0,0 +1,90 @@
+import requests
+from datetime import datetime
+from typing import Dict
+from .config import OPENWEATHER_BASE, get_openweather_key, UNITS
+from .utils import day_list
+
+
+def geocode_city(city: str) -> Dict:
+    url = f"{OPENWEATHER_BASE}/geo/1.0/direct"
+    r = requests.get(
+        url,
+        params={"q": city, "limit": 1, "appid": get_openweather_key()},
+        timeout=20,
+    )
+    r.raise_for_status()
+    arr = r.json()
+    if not arr:
+        raise RuntimeError(f"City not found: {city}")
+    d = arr[0]
+    return {
+        "name": d.get("name"),
+        "lat": d.get("lat"),
+        "lon": d.get("lon"),
+        "country": d.get("country"),
+    }
+
+
+def forecast_5day(lat: float, lon: float, units=UNITS) -> Dict:
+    # 5 day / 3-hour forecast
+    url = f"{OPENWEATHER_BASE}/data/2.5/forecast"
+    r = requests.get(
+        url,
+        params={"lat": lat, "lon": lon, "appid": get_openweather_key(), "units": units},
+        timeout=25,
+    )
+    r.raise_for_status()
+    return r.json()
+
+
+def summarize_forecast_for_range(city: str, start_date, days: int) -> Dict:
+    loc = geocode_city(city)
+    fc = forecast_5day(loc["lat"], loc["lon"])
+    daily = {}
+    for item in fc.get("list", []):
+        dt_txt = item.get("dt_txt")  # 'YYYY-MM-DD HH:MM:SS'
+        dt = datetime.strptime(dt_txt, "%Y-%m-%d %H:%M:%S")
+        dkey = dt.date().isoformat()
+        temp = item.get("main", {}).get("temp")
+        weather = item.get("weather", [{}])[0].get("description", "")
+        wind = item.get("wind", {}).get("speed", 0)
+        if dkey not in daily:
+            daily[dkey] = {"temps": [], "desc": {}, "wind": []}
+        daily[dkey]["temps"].append(temp)
+        daily[dkey]["wind"].append(wind)
+        daily[dkey]["desc"][weather] = daily[dkey]["desc"].get(weather, 0) + 1
+
+    dates = [d.isoformat() for d in day_list(start_date, days)]
+    rows = []
+    for dkey in dates:
+        if dkey in daily:
+            temps = daily[dkey]["temps"]
+            avg = sum(temps) / len(temps) if temps else None
+            wavg = (
+                sum(daily[dkey]["wind"]) / len(daily[dkey]["wind"]) if daily[dkey]["wind"] else 0
+            )
+            desc = max(daily[dkey]["desc"], key=daily[dkey]["desc"].get)
+            rows.append(
+                {
+                    "date": dkey,
+                    "temp_avg": round(avg, 1) if avg is not None else None,
+                    "wind": round(wavg, 1),
+                    "desc": desc,
+                }
+            )
+        else:
+            rows.append(
+                {
+                    "date": dkey,
+                    "temp_avg": None,
+                    "wind": None,
+                    "desc": "No forecast (beyond 5 days)",
+                }
+            )
+    summary_text = "\n".join(
+        [
+            f"{r['date']}: {r['desc']}, avg {r['temp_avg']}°C, wind {r['wind']} m/s"
+            for r in rows
+        ]
+    )
+    return {"location": loc, "daily": rows, "summary_text": summary_text}