"""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))