cli.py

"""AgroSense command-line interface.

Usage:
    python cli.py "your question here"                 # one-shot
    python cli.py --location Belagavi "your question"  # + live weather
    python cli.py --satellite --prices "..."           # + satellite + mandi prices
    python cli.py --lang hi "..."                       # answer in Hindi
    python cli.py                                      # interactive REPL
"""
from __future__ import annotations

import sys
from dataclasses import dataclass

from agrosense import RAGEngine


@dataclass
class Flags:
    location: str | None = None
    satellite: bool = False
    prices: bool = False
    advisories: bool = False
    environment: bool = False
    planets: bool = False
    news: bool = False
    stage: str | None = None
    lang: str = "en"


def _parse_flags(argv: list[str]) -> tuple[list[str], Flags]:
    """Extract optional --location/--satellite/--prices/--lang flags from argv."""
    flags = Flags()
    rest: list[str] = []
    i = 0
    while i < len(argv):
        tok = argv[i]
        if tok in ("--location", "-l") and i + 1 < len(argv):
            flags.location = argv[i + 1]; i += 2; continue
        if tok in ("--lang",) and i + 1 < len(argv):
            flags.lang = argv[i + 1]; i += 2; continue
        if tok in ("--stage",) and i + 1 < len(argv):
            flags.stage = argv[i + 1]; i += 2; continue
        if tok in ("--satellite", "-s"):
            flags.satellite = True; i += 1; continue
        if tok in ("--prices", "-p"):
            flags.prices = True; i += 1; continue
        if tok in ("--advisories", "-a"):
            flags.advisories = True; i += 1; continue
        if tok in ("--environment", "-e"):
            flags.environment = True; i += 1; continue
        if tok in ("--planets", "-P"):
            flags.planets = True; i += 1; continue
        if tok in ("--news", "-n"):
            flags.news = True; i += 1; continue
        rest.append(tok); i += 1
    return rest, flags


def main(argv: list[str]) -> int:
    args, flags = _parse_flags(argv[1:])

    from agrosense.calendars import datetime_header
    h = datetime_header()
    _moon = f"  |  🌙 {h['lunar_day']}" if h.get('lunar_day') else ""
    print(f"📅 {h['gregorian']}  |  🪔 {h['indian_national']}{_moon}  |  🕐 {h['ist_time']} IST")
    if h.get("panchang"):
        p = h["panchang"]
        print(f"   Panchang: {p['vaara']} | {p['tithi']} | Nakshatra {p['nakshatra']} "
              f"| Yoga {p['yoga']} | Karana {p['karana']}")

    print("Loading AgroSense knowledge base...")
    engine = RAGEngine()
    print(f"Ready. {engine.num_documents} KB documents indexed.")
    if flags.news:
        items = engine.get_news(limit=6)
        if items:
            print("\n📰 Latest from Google News:")
            for it in items:
                print(f"  - {it.title}")
    if flags.location:
        extras = "live weather" + (" + satellite" if flags.satellite else "")
        print(f"Location set to '{flags.location}' ({extras} will be attached).")
    if flags.lang != "en":
        print(f"Answer language: {flags.lang}")
    print()

    if args:
        _answer(engine, " ".join(args), flags)
        return 0

    print("Type a question (or 'quit' to exit).")
    while True:
        try:
            q = input("\n> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if q.lower() in {"quit", "exit", "q"}:
            break
        if q:
            _answer(engine, q, flags)
    return 0


def _answer(engine: RAGEngine, query: str, flags: Flags) -> None:
    ans = engine.answer(query, location=flags.location,
                        include_satellite=flags.satellite,
                        include_prices=flags.prices, language=flags.lang,
                        include_advisories=flags.advisories, stage=flags.stage)
    print("\n" + ans.text)
    if ans.advisories:
        print("\n" + _advisories_text(ans.advisories))
    elif flags.advisories:
        print("\n(Decision advisories unavailable — offline or location not found.)")
    if flags.environment:
        profile = engine.get_environment(location=flags.location)
        if profile:
            print("\n" + _environment_text(profile.to_dict()))
        else:
            print("\n(Environment data unavailable — offline or location not found.)")
    if flags.planets:
        pl = engine.get_planetary(location=flags.location)
        if pl:
            print("\n" + pl.to_context())
        else:
            print("\n(Planetary data unavailable — location not found.)")
    if flags.lang != "en" and not ans.translation_backend:
        print("\n(Translation backend unavailable — showing English. "
              "Install argostranslate or deep-translator.)")
    if ans.satellite:
        print("\n" + _satellite_text(ans.satellite))
    elif flags.satellite:
        print("\n(Satellite monitoring unavailable — offline or location not found.)")
    if ans.prices:
        print("\n" + _prices_text(ans.prices))
    elif flags.prices:
        print("\n(Market prices unavailable — set AGROSENSE_DATAGOV_API_KEY.)")
    print(f"\n[{ans.latency_ms:.0f} ms · embeddings={ans.embedding_backend} "
          f"· vector={ans.vector_backend} · gen={ans.generation_backend}]")


def _satellite_text(sat: dict) -> str:
    lines = [f"**Satellite monitoring - {sat['location_name']}** (source: {sat['source']})",
             f"- True-color image ({sat['truecolor_date']}): {sat['imagery'].get('true_color')}",
             f"- NDVI image ({sat['ndvi_date']}): {sat['imagery'].get('ndvi')}"]
    ac = sat.get("agroclimate")
    if ac:
        lines.append(
            f"- Agroclimate {ac['start']} to {ac['end']}: avg solar {ac['avg_solar_mj']} "
            f"MJ/m2/day, temp {ac['avg_tmin_c']}-{ac['avg_tmax_c']} C, "
            f"total rain {ac['total_precip_mm']} mm"
        )
        for n in ac.get("notes", []):
            lines.append(f"  - {n}")
    nd = sat.get("numeric_ndvi")
    if nd and nd.get("latest") is not None:
        lines.append(
            f"- Field NDVI ({nd['source']}): latest {nd['latest']} on "
            f"{nd.get('latest_date')}, mean {nd['mean']}, trend {nd.get('trend')} "
            f"-> {nd.get('status')}"
        )
        for n in nd.get("notes", []):
            lines.append(f"  - {n}")
    if sat["imagery"].get("worldview"):
        lines.append(f"- Interactive: {sat['imagery']['worldview']}")
    return "\n".join(lines)


def _environment_text(p: dict) -> str:
    sun, wind, aq = p.get("sunlight", {}), p.get("wind", {}), p.get("air_quality", {})
    gw, pollen = p.get("groundwater", {}), p.get("pollen", {})
    lines = [f"**Location & environment - {p['location_name']}**",
             f"- Latitude/Longitude: {p['latitude']}, {p['longitude']}",
             f"- Altitude (sea level): {p.get('elevation_m')} m",
             f"- Local population: {p.get('population')}",
             f"- Humidity: {p.get('humidity_pct')} %",
             f"- Sunlight: sunshine {sun.get('sunshine_hours')} h/day, "
             f"UV max {sun.get('uv_index_max')}, solar now {sun.get('shortwave_wm2')} W/m2",
             f"- Wind: {wind.get('speed_kmh')} km/h from {wind.get('direction_deg')}deg "
             f"({wind.get('direction_compass')})",
             f"- Air quality: US AQI {aq.get('us_aqi')} ({aq.get('category')}), "
             f"PM2.5 {aq.get('pm2_5')}, PM10 {aq.get('pm10')} ug/m3",
             "- Pollen: " + (", ".join(f"{k}={v}" for k, v in pollen.get("values", {}).items())
                             if pollen.get("available") else pollen.get("note", "unavailable")),
             (f"- Ground water table: {gw['level_m']} m below ground"
              if gw.get("level_m") is not None
              else f"- Ground water (soil-moisture proxy 3-9cm): "
                   f"{gw.get('soil_moisture_m3m3')} m3/m3"),
             f"  note: {gw.get('note')}"]
    return "\n".join(lines)


def _advisories_text(adv: dict) -> str:
    lines = [f"**Decision advisories - {adv['location_name']}**"
             + (f" (crop: {adv['crop']})" if adv.get("crop") else ""),
             f"({adv['source']})"]
    items = adv.get("advisories", [])
    if not items:
        lines.append("- No urgent signals; conditions look unremarkable.")
    for a in items:
        lines.append(f"- [{a['urgency'].upper()}] {a['title']}: {a['action']} "
                     f"(why: {a['rationale']})")
    return "\n".join(lines)


def _prices_text(prices: dict) -> str:
    lines = [f"**Market prices - {prices.get('commodity')}** (source: {prices['source']}, "
             "Rs/quintal)"]
    s = prices.get("summary") or {}
    if s:
        lines.append(f"- Modal across {s['count']} market(s): {s['modal_min']}-"
                     f"{s['modal_max']} (avg {s['modal_avg']})")
    for r in prices.get("records", [])[:5]:
        lines.append(f"  - {r['market']} ({r['state']}): modal {r['modal_price']} "
                     f"[{r['min_price']}-{r['max_price']}] on {r['arrival_date']}")
    for n in prices.get("notes", []):
        lines.append(f"- {n}")
    return "\n".join(lines)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))