Spaces:
Sleeping
Sleeping
File size: 3,061 Bytes
24eacbb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
import os
import json
import ee
import geemap
import gradio as gr
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from geemap.ee_tile_layers import _get_tile_url_format, _validate_palette
from starlette.middleware.cors import CORSMiddleware
# Earth Engine auth
if "EARTHENGINE_TOKEN" not in os.environ:
raise RuntimeError("EARTHENGINE_TOKEN environment variable not found")
try:
geemap.ee_initialize()
except Exception as e:
raise RuntimeError(f"Earth Engine authentication failed: {e}")
# ---- Shared Tile Logic ----
def get_tile(asset_id, vis_params=None):
try:
if asset_id.startswith("ee."):
ee_object = eval(asset_id)
else:
data_dict = ee.data.getAsset(asset_id)
data_type = data_dict["type"]
if data_type == "IMAGE":
ee_object = ee.Image(asset_id)
elif data_type == "IMAGE_COLLECTION":
ee_object = ee.ImageCollection(asset_id)
elif data_type in ["TABLE", "TABLE_COLLECTION"]:
ee_object = ee.FeatureCollection(asset_id)
else:
raise ValueError(f"Unsupported data type: {data_type}")
if vis_params is None:
vis_params = {}
if isinstance(vis_params, str):
if len(vis_params) == 0:
vis_params = "{}"
if vis_params.startswith("{") and vis_params.endswith("}"):
vis_params = json.loads(vis_params)
else:
raise ValueError(f"Unsupported vis_params type: {type(vis_params)}")
elif isinstance(vis_params, dict):
pass
else:
raise ValueError(f"Unsupported vis_params type: {type(vis_params)}")
if "palette" in vis_params:
vis_params["palette"] = _validate_palette(vis_params["palette"])
url = _get_tile_url_format(ee_object, vis_params)
return url
except Exception as e:
return f"Error: {str(e)}"
# ---- FastAPI ----
app = FastAPI()
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
)
class TileRequest(BaseModel):
asset_id: str
vis_params: dict | None = None
@app.post("/tile")
def get_tile_api(req: TileRequest):
result = get_tile(req.asset_id, req.vis_params)
if isinstance(result, str) and result.startswith("Error"):
raise HTTPException(status_code=400, detail=result)
return {"tile_url": result}
# ---- Gradio UI ----
gradio_ui = gr.Interface(
fn=get_tile,
inputs=[
gr.Textbox(label="Earth Engine Asset ID", placeholder="e.g., USGS/SRTMGL1_003"),
gr.Textbox(
label="Visualization Parameters (JSON)",
placeholder='{"min":0,"max":5000,"palette":"terrain"}',
),
],
outputs="text",
title="Earth Engine Tile URL Generator",
description="Supports ee.Image, ee.ImageCollection, ee.FeatureCollection. Tile URL is suitable for basemap usage.",
)
app = gr.mount_gradio_app(app, gradio_ui, path="/")
|