Commit 3a64a9f3 authored by Bhoopalam's avatar Bhoopalam
Browse files

Initial backend

parent f2aa86d9
from fastapi import APIRouter, HTTPException
from app.schemas import SearchBody, SearchResponse
from app.connectors.copernicus_stac import search_cdse, normalize_cdse_item
from app.connectors.usgs_m2m import search_usgs, normalize_usgs_item
from app.util import bbox_to_list
router = APIRouter(prefix="/search", tags=["search"])
@router.post("", response_model=SearchResponse)
async def search(body: SearchBody):
bbox_list = None
if body.bbox:
bbox_list = [body.bbox.minx, body.bbox.miny, body.bbox.maxx, body.bbox.maxy]
if body.provider == "cdse":
raw = await search_cdse(
bbox=bbox_list,
geometry=body.geometry,
datetime=body.datetime,
max_cloud=body.max_cloud,
collections=body.collections,
limit=body.limit,
sort=body.sort,
)
items = [normalize_cdse_item(f) for f in raw.get("features", [])]
return SearchResponse(items=items, next=raw.get("links", [{}])[-1].get("href"))
if body.provider == "usgs":
scenes = await search_usgs(
bbox=bbox_list,
geometry=body.geometry,
datetime=body.datetime,
datasets=body.collections,
max_results=body.limit,
)
items = [normalize_usgs_item(s) for s in scenes]
return SearchResponse(items=items)
# Optional AWS COG source kept for future expansion
raise HTTPException(400, detail="Unsupported provider for search")
\ No newline at end of file
from fastapi import APIRouter
from titiler.core.factory import TilerFactory
router = APIRouter()
# Mount TiTiler COG endpoints at /tiles/cog
cog = TilerFactory(router_prefix="/tiles/cog")
router.include_router(cog.router)
# Notes:
# - Use /tiles/cog/tilejson.json?url=<COG_URL>&assets=B04,B03,B02 for RGB
# - For S2 COGs on AWS, pass the HTTPS URL (RequesterPays OFF for Element84 COGs)
# - Frontend can consume the returned tilejson.tiles[0] template directly
\ No newline at end of file
from fastapi import APIRouter, HTTPException
from typing import List
from app.schemas import SearchBody, TimeSeriesResponse, TimeSlice
from app.routers.search import search as do_search
router = APIRouter(prefix="/timeseries", tags=["timeseries"])
@router.post("", response_model=TimeSeriesResponse)
async def timeseries(body: SearchBody):
# Reuse /search, then map to time slices; front-end can plug this into a slider
res = await do_search(body)
slices: List[TimeSlice] = []
for it in sorted(res.items, key=lambda x: x.datetime):
# If you mount /tiles (TiTiler), front-end can request dynamic RGB tiles per item/asset
slices.append(TimeSlice(
iso_date=it.datetime,
label=it.datetime.split("T")[0] if it.datetime else "",
item_id=it.id,
provider=it.provider,
# tile_template can be filled after you implement your own tiling logic
tile_template=None,
tilejson=None,
))
return TimeSeriesResponse(slices=slices)
\ No newline at end of file
from pydantic import BaseModel, Field
from typing import Literal, Optional, List, Dict, Any
Provider = Literal["cdse", "usgs", "aws_s2_cog"]
class BBox(BaseModel):
# WebMercator / WGS84 lon/lat
minx: float
miny: float
maxx: float
maxy: float
class SearchBody(BaseModel):
provider: Provider
bbox: Optional[BBox] = None
geometry: Optional[Dict[str, Any]] = None # GeoJSON
datetime: str = Field(..., description="STAC datetime interval e.g. 2020-01-01/2020-12-31")
max_cloud: Optional[float] = 30
collections: Optional[List[str]] = None # override defaults per provider
limit: int = 50
sort: Literal["asc", "desc"] = "desc"
class ItemAsset(BaseModel):
key: str
href: str
roles: Optional[List[str]] = None
type: Optional[str] = None
class ImageryItem(BaseModel):
id: str
provider: Provider
collection: str
datetime: str
geometry: Dict[str, Any]
bbox: List[float]
assets: Dict[str, ItemAsset]
preview: Optional[str] = None # thumbnail URL if available
linkouts: Optional[Dict[str, str]] = None # Copernicus Browser / EarthExplorer links
class SearchResponse(BaseModel):
items: List[ImageryItem]
next: Optional[str] = None
class TimeSlice(BaseModel):
iso_date: str
label: str
item_id: str
provider: Provider
tilejson: Optional[Dict[str, Any]] = None
tile_template: Optional[str] = None
class TimeSeriesResponse(BaseModel):
slices: List[TimeSlice]
\ No newline at end of file
from pydantic_settings import BaseSettings
from typing import List
class Settings(BaseSettings):
cdse_stac_endpoint: str = "https://stac.dataspace.copernicus.eu/v1"
cdse_oauth_token_url: str | None = None
cdse_client_id: str | None = None
cdse_client_secret: str | None = None
usgs_m2m_base: str = "https://m2m.cr.usgs.gov"
usgs_m2m_username: str | None = None
usgs_m2m_app_token: str | None = None
sentinel_hub_instance_id: str | None = None
tile_cache_dir: str = "./cache"
allowed_origins: List[str] = ["*"]
class Config:
env_file = ".env"
settings = Settings()
\ No newline at end of file
from typing import Dict, Any, List, Optional
def bbox_to_list(bbox) -> List[float]:
return [bbox.minx, bbox.miny, bbox.maxx, bbox.maxy]
def stac_datetime(start: str, end: str) -> str:
return f"{start}/{end}"
def iso_to_label(iso: str) -> str:
# 2020-05-17T10:11:12Z -> 2020‑05‑17
return iso.split("T")[0]
def build_copernicus_browser_link(lat: float, lon: float, date_iso: str, cloud: int = 30) -> str:
base = "https://browser.dataspace.copernicus.eu/"
return (f"{base}?zoom=13&lat={lat:.6f}&lng={lon:.6f}"
f"&themeId=DEFAULT-THEME&datasetId=S2_L2A_CDAS&cloudCoverage={cloud}"
f"&dateMode=SINGLE&toTime={date_iso}")
def build_earth_explorer_link() -> str:
# EE does not accept bbox in URL; provide homepage link
return "https://earthexplorer.usgs.gov/"
\ No newline at end of file
# === Copernicus Data Space Ecosystem ===
CDSE_STAC_ENDPOINT=https://stac.dataspace.copernicus.eu/v1
# OAuth client (needed for S3/download/process later; STAC search can be public)
CDSE_OAUTH_TOKEN_URL=https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token
CDSE_CLIENT_ID=
CDSE_CLIENT_SECRET=
# === USGS M2M ===
# As of 2025, use Application Token with the login-token endpoint
USGS_M2M_USERNAME=
USGS_M2M_APP_TOKEN=
USGS_M2M_BASE=https://m2m.cr.usgs.gov
# === Optional: Sentinel Hub OGC (WMS/WMTS) passthrough ===
SENTINEL_HUB_INSTANCE_ID=
# === Tile cache & general ===
TILE_CACHE_DIR=./cache
ALLOWED_ORIGINS=*
LOG_LEVEL=info
\ No newline at end of file
fastapi==0.115.0
uvicorn[standard]==0.30.6
httpx==0.27.2
pydantic==2.9.2
pydantic-settings==2.5.2
python-dotenv==1.0.1
# Optional (for /tiles with COGs)
titiler.core==0.16.2
rio-tiler==6.7.0
\ No newline at end of file
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment