Spaces:
Sleeping
Sleeping
| """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 | |
| 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)) | |