GeoSight MCP Server

""" GeoSight CLI - Command Line Interface for management and testing. """ import asyncio import click import json from rich.console import Console from rich.table import Table from rich.panel import Panel from rich.progress import Progress, SpinnerColumn, TextColumn console = Console() @click.group() @click.version_option(version="1.0.0") def cli(): """🛰️ GeoSight MCP Server CLI""" pass @cli.command() @click.option("--mode", type=click.Choice(["stdio", "http"]), default="stdio") @click.option("--host", default="0.0.0.0") @click.option("--port", default=8000) def serve(mode, host, port): """Start the MCP server.""" console.print(Panel.fit( "[bold green]🛰️ Starting GeoSight MCP Server[/bold green]", subtitle=f"Mode: {mode}" )) import os os.environ["SERVER_MODE"] = mode os.environ["SERVER_HOST"] = host os.environ["SERVER_PORT"] = str(port) from geosight.server import main main() @cli.command() @click.argument("location") @click.option("--start-date", "-s", required=True, help="Start date (YYYY-MM-DD)") @click.option("--end-date", "-e", required=True, help="End date (YYYY-MM-DD)") @click.option("--radius", "-r", default=10.0, help="Radius in km") def ndvi(location, start_date, end_date, radius): """Calculate NDVI for a location.""" async def run(): from geosight.tools.indices import calculate_ndvi with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: progress.add_task("Calculating NDVI...", total=None) result = await calculate_ndvi( start_date=start_date, end_date=end_date, location_name=location, radius_km=radius, ) console.print(Panel(result["summary"], title="NDVI Results")) if "statistics" in result: table = Table(title="Statistics") table.add_column("Metric", style="cyan") table.add_column("Value", style="green") for key, value in result["statistics"].items(): if isinstance(value, float): table.add_row(key, f"{value:.4f}") else: table.add_row(key, str(value)) console.print(table) asyncio.run(run()) @cli.command() @click.argument("location") @click.option("--date", "-d", required=True, help="Date (YYYY-MM-DD)") @click.option("--radius", "-r", default=10.0, help="Radius in km") @click.option("--model", "-m", default="eurosat", help="Model to use") def landcover(location, date, radius, model): """Classify land cover for a location.""" async def run(): from geosight.tools.classification import detect_land_cover with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: progress.add_task("Classifying land cover...", total=None) result = await detect_land_cover( date=date, location_name=location, radius_km=radius, model=model, ) console.print(Panel(result["summary"], title="Land Cover Results")) asyncio.run(run()) @cli.command() @click.argument("location") @click.option("--before", "-b", required=True, help="Before date (YYYY-MM-DD)") @click.option("--after", "-a", required=True, help="After date (YYYY-MM-DD)") @click.option("--radius", "-r", default=10.0, help="Radius in km") def changes(location, before, after, radius): """Detect changes between two dates.""" async def run(): from geosight.tools.change_detection import detect_changes with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: progress.add_task("Detecting changes...", total=None) result = await detect_changes( date_before=before, date_after=after, location_name=location, radius_km=radius, ) console.print(Panel(result["summary"], title="Change Detection Results")) asyncio.run(run()) @cli.command() @click.argument("location") def geocode(location): """Geocode a location name.""" async def run(): from geosight.tools.geocoding import get_location_info result = await get_location_info(location_name=location) console.print(Panel(result["summary"], title="Location Info")) if "coordinates" in result: table = Table(title="Coordinates") table.add_column("Property", style="cyan") table.add_column("Value", style="green") table.add_row("Latitude", f"{result['coordinates']['latitude']:.6f}") table.add_row("Longitude", f"{result['coordinates']['longitude']:.6f}") console.print(table) asyncio.run(run()) @cli.command() @click.argument("location") @click.option("--start-date", "-s", required=True) @click.option("--end-date", "-e", required=True) @click.option("--max-cloud", "-c", default=20.0, help="Max cloud cover %") def search(location, start_date, end_date, max_cloud): """Search for available imagery.""" async def run(): from geosight.tools.imagery import search_imagery with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: progress.add_task("Searching imagery...", total=None) result = await search_imagery( start_date=start_date, end_date=end_date, location_name=location, max_cloud_cover=max_cloud, ) console.print(Panel(result["summary"], title="Search Results")) asyncio.run(run()) @cli.command() def tools(): """List all available MCP tools.""" tools_list = [ ("search_imagery", "Search for available satellite imagery"), ("calculate_ndvi", "Calculate vegetation index"), ("calculate_ndwi", "Calculate water index"), ("calculate_ndbi", "Calculate built-up index"), ("detect_land_cover", "Classify land cover types"), ("detect_changes", "Detect changes between dates"), ("detect_objects", "Detect objects (ships, buildings, etc.)"), ("generate_report", "Generate analysis report"), ("get_location_info", "Geocode location names"), ] table = Table(title="🛰️ GeoSight MCP Tools") table.add_column("Tool", style="cyan", no_wrap=True) table.add_column("Description", style="green") for tool, desc in tools_list: table.add_row(tool, desc) console.print(table) @cli.command() def check(): """Check system configuration and dependencies.""" console.print(Panel.fit("[bold]System Check[/bold]")) checks = [] # Check Python version import sys py_version = f"{sys.version_info.major}.{sys.version_info.minor}" checks.append(("Python Version", py_version, sys.version_info >= (3, 11))) # Check dependencies try: import numpy checks.append(("NumPy", numpy.__version__, True)) except ImportError: checks.append(("NumPy", "Not installed", False)) try: import torch checks.append(("PyTorch", torch.__version__, True)) except ImportError: checks.append(("PyTorch", "Not installed", False)) try: import rasterio checks.append(("Rasterio", rasterio.__version__, True)) except ImportError: checks.append(("Rasterio", "Not installed", False)) try: import sentinelhub checks.append(("SentinelHub", sentinelhub.__version__, True)) except ImportError: checks.append(("SentinelHub", "Not installed", False)) # Check environment variables import os sh_configured = bool(os.environ.get("SENTINEL_HUB_CLIENT_ID")) checks.append(("Sentinel Hub Config", "Configured" if sh_configured else "Not configured", sh_configured)) # Display results table = Table(title="Dependency Check") table.add_column("Component", style="cyan") table.add_column("Status", style="green") table.add_column("OK", justify="center") for name, status, ok in checks: icon = "✅" if ok else "❌" table.add_row(name, status, icon) console.print(table) all_ok = all(ok for _, _, ok in checks) if all_ok: console.print("\n[bold green]✅ All checks passed![/bold green]") else: console.print("\n[bold yellow]⚠️ Some checks failed. See above for details.[/bold yellow]") @cli.command() @click.option("--output", "-o", default="geosight_demo.html", help="Output file") def demo(output): """Generate a demo HTML report.""" async def run(): from geosight.tools.reports import generate_report with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: progress.add_task("Generating demo report...", total=None) result = await generate_report( start_date="2024-01-01", end_date="2024-06-01", location_name="Mumbai, India", radius_km=15, analyses=["ndvi", "land_cover"], output_format="html", report_title="GeoSight Demo Report - Mumbai", ) # Save report with open(output, "w") as f: f.write(result.get("report_content", "")) console.print(f"\n[bold green]✅ Demo report saved to: {output}[/bold green]") console.print(result["summary"]) asyncio.run(run()) def main(): """Entry point for CLI.""" cli() if __name__ == "__main__": main()