[EIE] Add STAC Statistics Calculation Tool

[paridhi-parajuli]

Summary

Add MCP-compatible tool for returning statictics for the given list of items over a spatial bbox.

What it does

• Returns statistics for each item
• Uses the /cog/statistics endpoint of the raster API
• Registers with the MCP tool registry via @mcp_tool decorator

Files changed

• akd_ext/tools/stac_item_stats.py — New tool implementation
• akd_ext/tools/init.py — Added exports
• pyproject.toml — Added httpx as dependency

Testing

import asyncio
from akd_ext.tools.eie.stac_stats import StacItemInfo
from akd_ext.tools.eie.stac_stats import (
    STACItemStatsTool,
    STACItemStatsInputSchema,
    PlaceResult,
    StacSearchResult,
)

async def test_stats_tool_realistic():
    tool = STACItemStatsTool()

    # Use the STAC search output you provided
    stac_result = StacSearchResult(
        item_ids=[
            "OMI_trno2_0.10x0.10_202112_Col3_V4.nc",
            "OMI_trno2_0.10x0.10_202111_Col3_V4.nc",
            "OMI_trno2_0.10x0.10_202110_Col3_V4.nc",
            "OMI_trno2_0.10x0.10_202109_Col3_V4.nc",
            "OMI_trno2_0.10x0.10_202108_Col3_V4.nc",
        ],
        items=[
            StacItemInfo(
                id="OMI_trno2_0.10x0.10_202112_Col3_V4.nc",
                collection="no2-monthly",
                datetime="2021-12-01T00:00:00Z",
                asset_url="s3://veda-data-store-staging/no2-monthly/OMI_trno2_0.10x0.10_202112_Col3_V4.nc.tif"
            ),
            StacItemInfo(
                id="OMI_trno2_0.10x0.10_202111_Col3_V4.nc",
                collection="no2-monthly",
                datetime="2021-11-01T00:00:00Z",
                asset_url="s3://veda-data-store-staging/no2-monthly/OMI_trno2_0.10x0.10_202111_Col3_V4.nc.tif"
            ),
            StacItemInfo(
                id="OMI_trno2_0.10x0.10_202110_Col3_V4.nc",
                collection="no2-monthly",
                datetime="2021-10-01T00:00:00Z",
                asset_url="s3://veda-data-store-staging/no2-monthly/OMI_trno2_0.10x0.10_202110_Col3_V4.nc.tif"
            ),
            StacItemInfo(
                id="OMI_trno2_0.10x0.10_202109_Col3_V4.nc",
                collection="no2-monthly",
                datetime="2021-09-01T00:00:00Z",
                asset_url="s3://veda-data-store-staging/no2-monthly/OMI_trno2_0.10x0.10_202109_Col3_V4.nc.tif"
            ),
            StacItemInfo(
                id="OMI_trno2_0.10x0.10_202108_Col3_V4.nc",
                collection="no2-monthly",
                datetime="2021-08-01T00:00:00Z",
                asset_url="s3://veda-data-store-staging/no2-monthly/OMI_trno2_0.10x0.10_202108_Col3_V4.nc.tif"
            ),
        ],
        count=5
    )

    # Mocked place result (simple polygon covering the USA bbox)
    place_result = PlaceResult(
        place="USA",
        bbox=[-125.0, 24.5, -66.9, 49.5],
        geometry={
            "type": "Polygon",
            "coordinates": [[
                [-125.0, 24.5],
                [-66.9, 24.5],
                [-66.9, 49.5],
                [-125.0, 49.5],
                [-125.0, 24.5]
            ]]
        }
    )

    params = STACItemStatsInputSchema(
        stac_result=stac_result,
        place_result=place_result
    )

    result = await tool.arun(params)

    print("\n=== STAC Item Stats Result ===")
    print("Error:", result.error)
    print("Number of items:", len(result.items))
    for item in result.items:
        print("-" * 40)
        print("ID:", item.id)
        print("Datetime:", item.datetime)
        print("Statistics:", item.statistics)
        print("Error:", item.error)

    # Basic sanity checks
    assert isinstance(result.items, list)
    assert result.error is None or isinstance(result.error, str)
    assert len(result.items) == stac_result.count


async def main():
    await test_stats_tool_realistic()


if __name__ == "__main__":
    asyncio.run(main())