Initial project implementation

This commit is contained in:
2026-04-28 10:44:45 +08:00
commit 4e81c36630
41 changed files with 7378 additions and 0 deletions
@@ -0,0 +1,3 @@
from .cli import main
__all__ = ["main"]
@@ -0,0 +1,5 @@
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1 @@
"""FastAPI application package."""
@@ -0,0 +1,57 @@
from __future__ import annotations
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from sentinel_orbit_downloader.api.routes_downloads import router as downloads_router
from sentinel_orbit_downloader.api.routes_health import router as health_router
from sentinel_orbit_downloader.api.routes_jobs import router as jobs_router
from sentinel_orbit_downloader.api.routes_projects import router as projects_router
from sentinel_orbit_downloader.api.routes_regions import router as regions_router
from sentinel_orbit_downloader.api.routes_search import router as search_router
from sentinel_orbit_downloader.api.routes_settings import router as settings_router
from sentinel_orbit_downloader.core.config import configured_api_host, configured_api_port
def create_app() -> FastAPI:
app = FastAPI(
title="Sentinel Orbit Downloader API",
version="0.1.0",
description="Local API for Sentinel-1 search, orbit matching, and task management.",
)
web_root = Path(__file__).resolve().parents[1] / "web"
app.mount("/static", StaticFiles(directory=web_root), name="static")
@app.get("/", include_in_schema=False)
def index() -> FileResponse:
return FileResponse(web_root / "index.html")
@app.get("/app", include_in_schema=False)
def app_index() -> FileResponse:
return FileResponse(web_root / "index.html")
app.include_router(health_router, prefix="/api")
app.include_router(regions_router, prefix="/api")
app.include_router(projects_router, prefix="/api")
app.include_router(jobs_router, prefix="/api")
app.include_router(search_router, prefix="/api")
app.include_router(downloads_router, prefix="/api")
app.include_router(settings_router, prefix="/api")
return app
app = create_app()
def main() -> None:
uvicorn.run(
"sentinel_orbit_downloader.api.app:app",
host=configured_api_host(),
port=configured_api_port(),
reload=False,
)
@@ -0,0 +1,9 @@
from __future__ import annotations
from pathlib import Path
from sentinel_orbit_downloader.core.config import configured_db_path
def get_db_path() -> Path:
return configured_db_path()
@@ -0,0 +1,67 @@
from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import DownloadJobCreateRequest, DownloadJobCreateResponse, JobSummary
from sentinel_orbit_downloader.services.asf_client import credentials_available
from sentinel_orbit_downloader.services.download_manager import DownloadJobRequest, execute_download_job, queue_download_job
from sentinel_orbit_downloader.services.project_store import get_job
from sentinel_orbit_downloader.services.settings_store import get_saved_earthdata_credentials
router = APIRouter(prefix="/download-jobs", tags=["download-jobs"])
@router.post("", response_model=DownloadJobCreateResponse, status_code=status.HTTP_202_ACCEPTED)
def create_download_job_endpoint(
request: DownloadJobCreateRequest,
background_tasks: BackgroundTasks,
db_path=Depends(get_db_path),
) -> DownloadJobCreateResponse:
job = get_job(db_path, request.job_id)
if job is None:
raise HTTPException(status_code=404, detail=f"Job not found: {request.job_id}")
saved_username, saved_password = get_saved_earthdata_credentials(db_path)
resolved_username = request.username or saved_username
resolved_password = request.password or saved_password
if not request.skip_data and not credentials_available(resolved_username, resolved_password):
raise HTTPException(
status_code=400,
detail=(
"下载 Sentinel-1 原始产品需要 Earthdata 用户名和密码。"
"请在页面填写 Earthdata 用户名/密码,或先保存到本地配置,或先设置 EARTHDATA_USER 和 EARTHDATA_PASS。"
"如果这次只想补轨道,请把“跳过产品”设为“是”。"
),
)
manager_request = DownloadJobRequest(
job_id=request.job_id,
data_dir=Path(request.data_dir) if request.data_dir else None,
orbit_dir=Path(request.orbit_dir) if request.orbit_dir else None,
manifest_path=Path(request.manifest_path) if request.manifest_path else None,
skip_data=request.skip_data,
skip_orbits=request.skip_orbits,
processes=request.processes,
ignore_orbit_errors=request.ignore_orbit_errors,
username=resolved_username,
password=resolved_password,
)
try:
queue_download_job(db_path, manager_request)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
background_tasks.add_task(execute_download_job, db_path, manager_request)
queued_job = get_job(db_path, request.job_id)
if queued_job is None:
raise HTTPException(status_code=500, detail=f"Job disappeared after queueing: {request.job_id}")
return DownloadJobCreateResponse(
message="Download job queued",
job=JobSummary.model_validate(queued_job),
)
@@ -0,0 +1,14 @@
from __future__ import annotations
from fastapi import APIRouter, Depends
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import HealthResponse
router = APIRouter(tags=["health"])
@router.get("/health", response_model=HealthResponse)
def health(db_path=Depends(get_db_path)) -> HealthResponse:
return HealthResponse(status="ok", db_path=str(db_path))
@@ -0,0 +1,90 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import (
JobDeleteResponse,
JobDetailResponse,
JobEventResponse,
JobSceneResponse,
JobSummary,
)
from sentinel_orbit_downloader.services.project_store import (
clear_jobs,
delete_job,
get_job,
list_job_events,
list_job_scenes,
list_jobs,
)
router = APIRouter(prefix="/jobs", tags=["jobs"])
@router.get("", response_model=list[JobSummary])
def jobs(
project_id: int | None = Query(default=None),
limit: int = Query(default=20, ge=1, le=200),
db_path=Depends(get_db_path),
) -> list[JobSummary]:
return [JobSummary.model_validate(job) for job in list_jobs(db_path, project_id=project_id, limit=limit)]
@router.delete("", response_model=JobDeleteResponse)
def clear_jobs_endpoint(
project_id: int | None = Query(default=None),
force: bool = Query(default=False),
db_path=Depends(get_db_path),
) -> JobDeleteResponse:
deleted_job_ids = clear_jobs(db_path, project_id=project_id, include_running=force)
return JobDeleteResponse(
message="任务已清空" if deleted_job_ids else "没有可清空的任务",
deleted_count=len(deleted_job_ids),
deleted_job_ids=deleted_job_ids,
)
@router.get("/{job_id}", response_model=JobDetailResponse)
def job_detail(
job_id: int,
include_events: bool = Query(default=False),
db_path=Depends(get_db_path),
) -> JobDetailResponse:
job = get_job(db_path, job_id)
if job is None:
raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")
scenes = [JobSceneResponse(**scene) for scene in list_job_scenes(db_path, job_id)]
events = []
if include_events:
events = [JobEventResponse(**event) for event in list_job_events(db_path, job_id)]
return JobDetailResponse(
job=JobSummary.model_validate(job),
scenes=scenes,
events=events,
)
@router.delete("/{job_id}", response_model=JobDeleteResponse, status_code=status.HTTP_200_OK)
def delete_job_endpoint(
job_id: int,
force: bool = Query(default=False),
db_path=Depends(get_db_path),
) -> JobDeleteResponse:
job = get_job(db_path, job_id)
if job is None:
raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")
if job.status == "running" and not force:
raise HTTPException(
status_code=409,
detail="任务仍在运行。请先停止服务后再删,或使用强制删除。",
)
deleted = delete_job(db_path, job_id)
if not deleted:
raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")
return JobDeleteResponse(
message="任务已删除",
deleted_count=1,
deleted_job_ids=[job_id],
)
@@ -0,0 +1,45 @@
from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import ProjectCreateRequest, ProjectResponse
from sentinel_orbit_downloader.services.project_store import create_project, get_project, list_projects
router = APIRouter(prefix="/projects", tags=["projects"])
@router.get("", response_model=list[ProjectResponse])
def projects(db_path=Depends(get_db_path)) -> list[ProjectResponse]:
return [ProjectResponse.model_validate(project) for project in list_projects(db_path)]
@router.post("", response_model=ProjectResponse)
def create_project_endpoint(
request: ProjectCreateRequest,
db_path=Depends(get_db_path),
) -> ProjectResponse:
try:
project = create_project(
db_path,
name=request.name,
data_root=Path(request.data_root),
description=request.description,
slug=request.slug,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except PermissionError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return ProjectResponse.model_validate(project)
@router.get("/{project_id}", response_model=ProjectResponse)
def project_detail(project_id: int, db_path=Depends(get_db_path)) -> ProjectResponse:
project = get_project(db_path, project_id)
if project is None:
raise HTTPException(status_code=404, detail=f"Project not found: {project_id}")
return ProjectResponse.model_validate(project)
@@ -0,0 +1,60 @@
from __future__ import annotations
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Query
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import (
RegionDetail,
RegionImportRequest,
RegionImportResponse,
RegionSummary,
)
from sentinel_orbit_downloader.api.serializers import region_detail, region_summary
from sentinel_orbit_downloader.services.region_catalog import (
get_region,
import_regions,
search_regions,
)
router = APIRouter(prefix="/regions", tags=["regions"])
@router.post("/import", response_model=RegionImportResponse)
def import_region_catalog(
request: RegionImportRequest,
db_path=Depends(get_db_path),
) -> RegionImportResponse:
source_path = Path(request.source_path)
if not source_path.exists():
raise HTTPException(status_code=404, detail=f"GeoJSON file not found: {source_path}")
stats = import_regions(db_path, source_path, simplify_tolerance=request.simplify_tolerance)
return RegionImportResponse(
source_path=str(stats.source_path),
simplify_tolerance=stats.simplify_tolerance,
total_features=stats.total_features,
imported=stats.imported,
)
@router.get("/search", response_model=list[RegionSummary])
def search_region_catalog(
q: str = Query(..., min_length=1),
level: str | None = Query(default=None),
limit: int = Query(default=20, ge=1, le=200),
db_path=Depends(get_db_path),
) -> list[RegionSummary]:
return [
region_summary(region)
for region in search_regions(db_path, q, limit=limit, level=level)
]
@router.get("/{adcode}", response_model=RegionDetail)
def get_region_detail(adcode: str, db_path=Depends(get_db_path)) -> RegionDetail:
region = get_region(db_path, adcode)
if region is None:
raise HTTPException(status_code=404, detail=f"Region not found: {adcode}")
return region_detail(region)
@@ -0,0 +1,121 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import JobSceneResponse, JobSummary, SearchJobCreateRequest, SearchJobCreateResponse
from sentinel_orbit_downloader.services.aoi import bbox_to_wkt
from sentinel_orbit_downloader.services.asf_client import SearchParams, query_scenes
from sentinel_orbit_downloader.services.project_store import (
create_search_job,
get_job,
get_project,
list_job_scenes,
record_job_event,
save_query_results,
update_search_job,
)
from sentinel_orbit_downloader.services.region_catalog import get_region, region_wkt
router = APIRouter(prefix="/search-jobs", tags=["search-jobs"])
@router.post("", response_model=SearchJobCreateResponse)
def create_search_job_endpoint(
request: SearchJobCreateRequest,
db_path=Depends(get_db_path),
) -> SearchJobCreateResponse:
if request.project_id is not None and get_project(db_path, request.project_id) is None:
raise HTTPException(status_code=404, detail=f"Project not found: {request.project_id}")
try:
aoi_meta = resolve_search_aoi(request, db_path)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
job = create_search_job(
db_path,
project_id=request.project_id,
source_type=aoi_meta["source_type"],
source_value=aoi_meta["source_value"],
region_adcode=aoi_meta["region_adcode"],
region_name=aoi_meta["region_name"],
start_time=request.start,
end_time=request.end,
processing_level=request.processing_level,
beam_mode=request.beam_mode,
flight_direction=request.flight_direction.upper() if request.flight_direction else None,
polarization=request.polarization,
relative_orbit=request.relative_orbit,
max_results=request.max_results,
manifest_path=request.manifest_path,
)
record_job_event(db_path, job.id, "info", "API search job created")
try:
params = SearchParams(
aoi_wkt=aoi_meta["aoi_wkt"],
start=request.start,
end=request.end,
processing_level=request.processing_level,
beam_mode=request.beam_mode,
flight_direction=request.flight_direction,
polarization=request.polarization,
relative_orbit=request.relative_orbit,
max_results=request.max_results,
)
results = query_scenes(params)
save_query_results(db_path, job.id, results)
update_search_job(
db_path,
job.id,
status="succeeded",
result_count=len(results),
manifest_path=request.manifest_path,
)
record_job_event(db_path, job.id, "info", f"API query returned {len(results)} scene(s)")
except Exception as exc:
update_search_job(db_path, job.id, status="failed", error_message=str(exc))
record_job_event(db_path, job.id, "error", f"API search failed: {exc}")
raise HTTPException(status_code=500, detail=str(exc)) from exc
final_job = get_job(db_path, job.id)
if final_job is None:
raise HTTPException(status_code=500, detail=f"Job disappeared after creation: {job.id}")
scenes = [JobSceneResponse(**scene) for scene in list_job_scenes(db_path, job.id)]
return SearchJobCreateResponse(
job=JobSummary.model_validate(final_job),
scenes=scenes,
)
def resolve_search_aoi(request: SearchJobCreateRequest, db_path) -> dict[str, str | None]:
if request.bbox:
return {
"aoi_wkt": bbox_to_wkt(request.bbox),
"source_type": "bbox",
"source_value": request.bbox,
"region_adcode": None,
"region_name": None,
}
if request.wkt:
return {
"aoi_wkt": request.wkt.strip(),
"source_type": "wkt",
"source_value": request.wkt.strip(),
"region_adcode": None,
"region_name": None,
}
if request.region_adcode:
region = get_region(db_path, request.region_adcode)
if region is None:
raise ValueError(f"Region not found: {request.region_adcode}")
return {
"aoi_wkt": region_wkt(db_path, region.adcode, simplified=not request.raw_region),
"source_type": "region",
"source_value": region.label,
"region_adcode": region.adcode,
"region_name": region.label,
}
raise ValueError("one AOI input is required")
@@ -0,0 +1,52 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from sentinel_orbit_downloader.api.deps import get_db_path
from sentinel_orbit_downloader.api.schemas import (
EarthdataSettingsRequest,
EarthdataSettingsResponse,
MessageResponse,
)
from sentinel_orbit_downloader.services.settings_store import (
clear_earthdata_credentials,
get_earthdata_settings,
save_earthdata_credentials,
)
router = APIRouter(prefix="/settings", tags=["settings"])
@router.get("/earthdata", response_model=EarthdataSettingsResponse)
def earthdata_settings(db_path=Depends(get_db_path)) -> EarthdataSettingsResponse:
settings = get_earthdata_settings(db_path)
return EarthdataSettingsResponse(
username=settings.username,
has_password=settings.has_password,
)
@router.put("/earthdata", response_model=EarthdataSettingsResponse, status_code=status.HTTP_200_OK)
def save_earthdata_settings(
request: EarthdataSettingsRequest,
db_path=Depends(get_db_path),
) -> EarthdataSettingsResponse:
try:
settings = save_earthdata_credentials(
db_path,
username=request.username,
password=request.password,
)
except (ValueError, RuntimeError) as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return EarthdataSettingsResponse(
username=settings.username,
has_password=settings.has_password,
)
@router.delete("/earthdata", response_model=MessageResponse)
def delete_earthdata_settings(db_path=Depends(get_db_path)) -> MessageResponse:
clear_earthdata_credentials(db_path)
return MessageResponse(message="Earthdata 凭据已清除")
@@ -0,0 +1,195 @@
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, model_validator
class HealthResponse(BaseModel):
status: str
db_path: str
class RegionSummary(BaseModel):
adcode: str
name: str
level: str
parent_adcode: str | None = None
parent_name: str | None = None
province_adcode: str | None = None
province_name: str | None = None
label: str
bbox: list[float] | None = None
simplify_tolerance: float
class RegionDetail(RegionSummary):
center: list[float] | None = None
centroid: list[float] | None = None
geometry_json: dict[str, Any]
wkt: str
simplified_wkt: str
children_num: int | None = None
tree_id: str | None = None
class RegionImportRequest(BaseModel):
source_path: str = Field(default="\u5168\u56fd\u884c\u653f\u533a.geojson")
simplify_tolerance: float = Field(default=0.01, gt=0)
class RegionImportResponse(BaseModel):
source_path: str
simplify_tolerance: float
total_features: int
imported: int
class ProjectCreateRequest(BaseModel):
name: str
data_root: str
description: str | None = None
slug: str | None = None
class ProjectResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
slug: str
data_root: str
description: str | None = None
created_at: str
updated_at: str
class JobSummary(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
project_id: int | None = None
source_type: str
source_value: str | None = None
region_adcode: str | None = None
region_name: str | None = None
start_time: str
end_time: str
processing_level: str
beam_mode: str | None = None
flight_direction: str | None = None
polarization: str | None = None
relative_orbit: int | None = None
max_results: int
status: str
result_count: int | None = None
data_dir: str | None = None
orbit_dir: str | None = None
manifest_path: str | None = None
error_message: str | None = None
created_at: str
finished_at: str | None = None
class JobSceneResponse(BaseModel):
job_id: int
scene_id: str
selected: int
product_status: str
orbit_status: str
product_path: str | None = None
orbit_path: str | None = None
platform: str | None = None
start_time: str | None = None
stop_time: str | None = None
processing_level: str | None = None
flight_direction: str | None = None
relative_orbit: int | None = None
size_mb: float | None = None
class JobEventResponse(BaseModel):
id: int
level: str
message: str
created_at: str
class JobDetailResponse(BaseModel):
job: JobSummary
scenes: list[JobSceneResponse]
events: list[JobEventResponse] = Field(default_factory=list)
class SearchJobCreateRequest(BaseModel):
project_id: int | None = None
bbox: str | None = None
wkt: str | None = None
region_adcode: str | None = None
start: str
end: str
processing_level: str = "SLC"
beam_mode: str | None = "IW"
flight_direction: str | None = None
polarization: str | None = None
relative_orbit: int | None = None
max_results: int = Field(default=50, ge=1)
manifest_path: str | None = None
raw_region: bool = False
@model_validator(mode="after")
def validate_aoi(self) -> "SearchJobCreateRequest":
provided = [self.bbox, self.wkt, self.region_adcode]
count = sum(1 for value in provided if value)
if count != 1:
raise ValueError("provide exactly one of bbox, wkt, or region_adcode")
return self
class SearchJobCreateResponse(BaseModel):
job: JobSummary
scenes: list[JobSceneResponse]
class DownloadJobCreateRequest(BaseModel):
job_id: int = Field(ge=1)
data_dir: str | None = None
orbit_dir: str | None = None
manifest_path: str | None = None
skip_data: bool = False
skip_orbits: bool = False
processes: int = Field(default=2, ge=1)
ignore_orbit_errors: bool = False
username: str | None = None
password: str | None = None
@model_validator(mode="after")
def validate_download_flags(self) -> "DownloadJobCreateRequest":
if self.skip_data and self.skip_orbits:
raise ValueError("skip_data and skip_orbits cannot both be true")
return self
class DownloadJobCreateResponse(BaseModel):
message: str
job: JobSummary
class MessageResponse(BaseModel):
message: str
class EarthdataSettingsRequest(BaseModel):
username: str
password: str
class EarthdataSettingsResponse(BaseModel):
username: str | None = None
has_password: bool = False
class JobDeleteResponse(BaseModel):
message: str
deleted_count: int
deleted_job_ids: list[int] = Field(default_factory=list)
@@ -0,0 +1,45 @@
from __future__ import annotations
import json
from sentinel_orbit_downloader.api.schemas import RegionDetail, RegionSummary
def region_summary(region) -> RegionSummary:
bbox = list(region.bbox) if region.bbox is not None else None
return RegionSummary(
adcode=region.adcode,
name=region.name,
level=region.level,
parent_adcode=region.parent_adcode,
parent_name=region.parent_name,
province_adcode=region.province_adcode,
province_name=region.province_name,
label=region.label,
bbox=bbox,
simplify_tolerance=region.simplify_tolerance,
)
def region_detail(region) -> RegionDetail:
summary = region_summary(region)
center = (
[region.center_lon, region.center_lat]
if region.center_lon is not None and region.center_lat is not None
else None
)
centroid = (
[region.centroid_lon, region.centroid_lat]
if region.centroid_lon is not None and region.centroid_lat is not None
else None
)
return RegionDetail(
**summary.model_dump(),
center=center,
centroid=centroid,
geometry_json=json.loads(region.geometry_json),
wkt=region.wkt,
simplified_wkt=region.simplified_wkt,
children_num=region.children_num,
tree_id=region.tree_id,
)
@@ -0,0 +1,815 @@
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any, Sequence
from sentinel_orbit_downloader.core.config import DEFAULT_DB_PATH
from sentinel_orbit_downloader.services.aoi import aoi_file_to_wkt, bbox_to_wkt
from sentinel_orbit_downloader.services.asf_client import (
SearchParams,
build_session,
product_size_mb,
query_scenes,
scene_name_from_product,
write_manifest,
)
from sentinel_orbit_downloader.services.orbit_client import (
collect_orbit_scenes,
download_orbits,
scene_names_from_results,
)
from sentinel_orbit_downloader.services.project_store import (
create_project,
create_search_job,
get_job,
get_project,
list_job_events,
list_job_scenes,
list_jobs,
list_projects,
record_download_result,
record_job_event,
save_query_results,
update_search_job,
)
from sentinel_orbit_downloader.services.region_catalog import (
get_region,
import_regions,
region_wkt,
search_regions,
)
DEFAULT_REGION_FILE = Path("\u5168\u56fd\u884c\u653f\u533a.geojson")
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
return args.func(args)
except KeyboardInterrupt:
print("Interrupted.", file=sys.stderr)
return 130
except ValueError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 2
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="s1-dl",
description="Query and download Sentinel-1 scenes plus matching orbit files.",
)
subparsers = parser.add_subparsers(dest="command", required=True)
search_parser = subparsers.add_parser("search", help="query Sentinel-1 scenes")
add_database_arg(search_parser)
add_project_arg(search_parser)
add_query_args(search_parser)
add_manifest_arg(search_parser)
search_parser.set_defaults(func=cmd_search)
download_parser = subparsers.add_parser(
"download",
help="query scenes, download products, and fetch matching orbit files",
)
add_database_arg(download_parser)
add_project_arg(download_parser)
add_query_args(download_parser)
download_parser.add_argument(
"--out",
type=Path,
default=Path("downloads"),
help="output root directory; data and orbits subdirectories are created here",
)
download_parser.add_argument(
"--data-dir",
type=Path,
help="override product download directory; default is <out>/data",
)
download_parser.add_argument(
"--orbit-dir",
type=Path,
help="override orbit download directory; default is <out>/orbits",
)
download_parser.add_argument(
"--username",
help="NASA Earthdata username; defaults to EARTHDATA_USER or EARTHDATA_USERNAME",
)
download_parser.add_argument(
"--password",
help="NASA Earthdata password; defaults to EARTHDATA_PASS or EARTHDATA_PASSWORD",
)
download_parser.add_argument(
"--ask-password",
action="store_true",
help="prompt for the Earthdata password instead of reading an argument/env var",
)
download_parser.add_argument(
"--processes",
type=positive_int,
default=2,
help="parallel ASF download processes",
)
download_parser.add_argument(
"--skip-data",
action="store_true",
help="do not download product zip files; useful for orbit-only runs from query results",
)
download_parser.add_argument(
"--skip-orbits",
action="store_true",
help="do not download matching orbit files",
)
download_parser.add_argument(
"--ignore-orbit-errors",
action="store_true",
help="return success even if some orbit files cannot be found",
)
add_manifest_arg(download_parser)
download_parser.set_defaults(func=cmd_download)
orbit_parser = subparsers.add_parser(
"orbits",
help="download orbit files for existing scenes or local .zip/.SAFE products",
)
scene_group = orbit_parser.add_mutually_exclusive_group(required=True)
scene_group.add_argument(
"--scene",
action="append",
help="Sentinel-1 scene name; may be repeated",
)
scene_group.add_argument(
"--scene-file",
type=Path,
help="text file containing one scene name per line",
)
scene_group.add_argument(
"--data-dir",
type=Path,
help="directory containing Sentinel-1 .zip files or .SAFE folders",
)
orbit_parser.add_argument(
"--out",
type=Path,
default=Path("orbits"),
help="orbit output directory",
)
orbit_parser.add_argument(
"--ignore-errors",
action="store_true",
help="return success even if some orbit files cannot be found",
)
orbit_parser.set_defaults(func=cmd_orbits)
regions_parser = subparsers.add_parser(
"regions",
help="import and query administrative boundary regions",
)
add_database_arg(regions_parser)
region_subparsers = regions_parser.add_subparsers(dest="regions_command", required=True)
regions_import = region_subparsers.add_parser("import", help="import a GeoJSON region catalog")
add_database_arg(regions_import)
regions_import.add_argument(
"source",
nargs="?",
type=Path,
default=DEFAULT_REGION_FILE,
help=f"GeoJSON source file; default is .\\{DEFAULT_REGION_FILE.name}",
)
regions_import.add_argument(
"--simplify-tolerance",
type=float,
default=0.01,
help="Shapely simplify tolerance used to derive query WKT",
)
regions_import.set_defaults(func=cmd_regions_import)
regions_search_parser = region_subparsers.add_parser("search", help="search regions by name or adcode")
add_database_arg(regions_search_parser)
regions_search_parser.add_argument("query", help="query text, for example mudanjiang or 231000")
regions_search_parser.add_argument(
"--limit",
type=positive_int,
default=20,
help="maximum number of matches to return",
)
regions_search_parser.add_argument(
"--level",
choices=("country", "province", "city", "district"),
help="optional level filter",
)
regions_search_parser.set_defaults(func=cmd_regions_search)
regions_export_parser = region_subparsers.add_parser("export", help="export region geometry or metadata")
add_database_arg(regions_export_parser)
regions_export_parser.add_argument("adcode", help="region adcode")
regions_export_parser.add_argument(
"--format",
choices=("wkt", "geojson", "bbox", "json"),
default="wkt",
help="output format",
)
regions_export_parser.add_argument(
"--raw",
action="store_true",
help="use the original geometry instead of simplified WKT/json",
)
regions_export_parser.add_argument(
"--out",
type=Path,
help="optional file to write the exported value to",
)
regions_export_parser.set_defaults(func=cmd_regions_export)
projects_parser = subparsers.add_parser("projects", help="manage local projects")
add_database_arg(projects_parser)
project_subparsers = projects_parser.add_subparsers(dest="projects_command", required=True)
projects_create = project_subparsers.add_parser("create", help="create a project")
add_database_arg(projects_create)
projects_create.add_argument("--name", required=True, help="project name")
projects_create.add_argument("--data-root", type=Path, required=True, help="project data root")
projects_create.add_argument("--description", help="optional description")
projects_create.add_argument("--slug", help="optional directory-friendly slug")
projects_create.set_defaults(func=cmd_projects_create)
projects_list_parser = project_subparsers.add_parser("list", help="list projects")
add_database_arg(projects_list_parser)
projects_list_parser.set_defaults(func=cmd_projects_list)
projects_show_parser = project_subparsers.add_parser("show", help="show one project")
add_database_arg(projects_show_parser)
projects_show_parser.add_argument("project_id", type=positive_int, help="project id")
projects_show_parser.set_defaults(func=cmd_projects_show)
jobs_parser = subparsers.add_parser("jobs", help="inspect saved query and download jobs")
add_database_arg(jobs_parser)
jobs_subparsers = jobs_parser.add_subparsers(dest="jobs_command", required=True)
jobs_list_parser = jobs_subparsers.add_parser("list", help="list recent jobs")
add_database_arg(jobs_list_parser)
jobs_list_parser.add_argument("--project-id", type=positive_int, help="optional project filter")
jobs_list_parser.add_argument("--limit", type=positive_int, default=20, help="maximum jobs to list")
jobs_list_parser.set_defaults(func=cmd_jobs_list)
jobs_show_parser = jobs_subparsers.add_parser("show", help="show one job and its scenes")
add_database_arg(jobs_show_parser)
jobs_show_parser.add_argument("job_id", type=positive_int, help="job id")
jobs_show_parser.add_argument(
"--events",
action="store_true",
help="include stored job events",
)
jobs_show_parser.set_defaults(func=cmd_jobs_show)
return parser
def add_database_arg(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--db",
type=Path,
default=DEFAULT_DB_PATH,
help=f"SQLite database path; default is {DEFAULT_DB_PATH}",
)
def add_project_arg(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--project-id",
type=positive_int,
help="optional project id used to group saved jobs",
)
def add_query_args(parser: argparse.ArgumentParser) -> None:
aoi = parser.add_mutually_exclusive_group(required=True)
aoi.add_argument("--bbox", help="AOI bounding box as min_lon,min_lat,max_lon,max_lat")
aoi.add_argument("--wkt", help="AOI as WKT, for example POLYGON((...))")
aoi.add_argument(
"--aoi-file",
type=Path,
help="AOI file containing WKT, GeoJSON Feature, or GeoJSON geometry",
)
aoi.add_argument(
"--region-adcode",
help="AOI from imported region catalog; use `s1-dl regions search` first",
)
parser.add_argument(
"--raw-region",
action="store_true",
help="when using --region-adcode, use the original geometry instead of simplified WKT",
)
parser.add_argument("--start", required=True, help="start date/time, e.g. 2024-01-01")
parser.add_argument("--end", required=True, help="end date/time, e.g. 2024-02-01")
parser.add_argument(
"--processing-level",
default="SLC",
help="ASF processing level, e.g. SLC, GRD_HD, GRD_MD, OCN",
)
parser.add_argument("--beam-mode", default="IW", help="beam mode, e.g. IW, EW, WV")
parser.add_argument(
"--flight-direction",
choices=("ASCENDING", "DESCENDING", "ascending", "descending"),
help="optional orbit direction filter",
)
parser.add_argument("--polarization", help="optional polarization filter, e.g. VV+VH")
parser.add_argument(
"--relative-orbit",
type=positive_int,
help="optional Sentinel-1 relative orbit/track number",
)
parser.add_argument(
"--max-results",
type=positive_int,
default=50,
help="maximum scenes to return",
)
def add_manifest_arg(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--manifest", type=Path, help="write search results to .csv or .json")
def cmd_search(args: argparse.Namespace) -> int:
ensure_project_exists(args.db, args.project_id)
aoi_meta = resolve_aoi(args)
job = create_search_job(
args.db,
project_id=args.project_id,
source_type=aoi_meta["source_type"],
source_value=aoi_meta["source_value"],
region_adcode=aoi_meta["region_adcode"],
region_name=aoi_meta["region_name"],
start_time=args.start,
end_time=args.end,
processing_level=args.processing_level,
beam_mode=args.beam_mode,
flight_direction=args.flight_direction.upper() if args.flight_direction else None,
polarization=args.polarization,
relative_orbit=args.relative_orbit,
max_results=args.max_results,
manifest_path=str(args.manifest) if args.manifest else None,
)
record_job_event(args.db, job.id, "info", "Query created")
try:
results = perform_query(args, aoi_meta["aoi_wkt"])
save_query_results(args.db, job.id, results)
print_results(results)
if args.manifest:
write_manifest(results, args.manifest)
print(f"Wrote manifest: {args.manifest}")
update_search_job(
args.db,
job.id,
status="succeeded",
result_count=len(results),
manifest_path=str(args.manifest) if args.manifest else None,
)
record_job_event(args.db, job.id, "info", f"Query returned {len(results)} scene(s)")
except Exception as exc:
update_search_job(args.db, job.id, status="failed", error_message=str(exc))
record_job_event(args.db, job.id, "error", f"Query failed: {exc}")
raise
print(f"Saved job: {job.id}")
return 0
def cmd_download(args: argparse.Namespace) -> int:
ensure_project_exists(args.db, args.project_id)
aoi_meta = resolve_aoi(args)
out_root = args.out
data_dir = args.data_dir or out_root / "data"
orbit_dir = args.orbit_dir or out_root / "orbits"
out_root.mkdir(parents=True, exist_ok=True)
manifest = args.manifest or out_root / "manifest.csv"
job = create_search_job(
args.db,
project_id=args.project_id,
source_type=aoi_meta["source_type"],
source_value=aoi_meta["source_value"],
region_adcode=aoi_meta["region_adcode"],
region_name=aoi_meta["region_name"],
start_time=args.start,
end_time=args.end,
processing_level=args.processing_level,
beam_mode=args.beam_mode,
flight_direction=args.flight_direction.upper() if args.flight_direction else None,
polarization=args.polarization,
relative_orbit=args.relative_orbit,
max_results=args.max_results,
data_dir=str(data_dir),
orbit_dir=str(orbit_dir),
manifest_path=str(manifest),
)
record_job_event(args.db, job.id, "info", "Download job created")
try:
results = perform_query(args, aoi_meta["aoi_wkt"])
save_query_results(args.db, job.id, results)
record_job_event(args.db, job.id, "info", f"Query returned {len(results)} scene(s)")
print_results(results)
write_manifest(results, manifest)
print(f"Wrote manifest: {manifest}")
if not results:
update_search_job(
args.db,
job.id,
status="succeeded",
result_count=0,
data_dir=str(data_dir),
orbit_dir=str(orbit_dir),
manifest_path=str(manifest),
)
record_job_event(args.db, job.id, "info", "No scenes matched the query")
print(f"Saved job: {job.id}")
return 0
if not args.skip_data:
data_dir.mkdir(parents=True, exist_ok=True)
session = build_session(
username=args.username,
password=args.password,
ask_password=args.ask_password,
)
print(f"Downloading {len(results)} product(s) to {data_dir} ...")
results.download(path=str(data_dir), session=session, processes=args.processes)
print("Product download finished.")
record_job_event(args.db, job.id, "info", f"Product download finished into {data_dir}")
record_product_downloads(args.db, job.id, results, data_dir)
else:
record_job_event(args.db, job.id, "info", "Product download skipped by request")
record_product_downloads(args.db, job.id, results, None)
orbit_failures = 0
if not args.skip_orbits:
scenes = scene_names_from_results(results)
print(f"Downloading orbit files for {len(scenes)} scene(s) to {orbit_dir} ...")
orbit_results = download_orbits(scenes, orbit_dir)
orbit_failures = summarize_orbit_results(orbit_results)
record_orbit_downloads(args.db, job.id, orbit_results)
record_job_event(
args.db,
job.id,
"info",
f"Orbit download finished: {len(orbit_results) - orbit_failures} ok, {orbit_failures} failed",
)
else:
record_job_event(args.db, job.id, "info", "Orbit download skipped by request")
final_status = "partial_failed" if orbit_failures else "succeeded"
update_search_job(
args.db,
job.id,
status=final_status,
result_count=len(results),
data_dir=str(data_dir),
orbit_dir=str(orbit_dir),
manifest_path=str(manifest),
)
print(f"Saved job: {job.id}")
if orbit_failures and not args.ignore_orbit_errors:
return 1
return 0
except Exception as exc:
update_search_job(
args.db,
job.id,
status="failed",
result_count=None,
data_dir=str(data_dir),
orbit_dir=str(orbit_dir),
manifest_path=str(manifest),
error_message=str(exc),
)
record_job_event(args.db, job.id, "error", f"Download job failed: {exc}")
raise
def cmd_orbits(args: argparse.Namespace) -> int:
scenes = collect_orbit_scenes(
scenes=args.scene,
scene_file=args.scene_file,
data_dir=args.data_dir,
)
if not scenes:
print("No Sentinel-1 scenes found.", file=sys.stderr)
return 1
print(f"Downloading orbit files for {len(scenes)} scene(s) to {args.out} ...")
results = download_orbits(scenes, args.out)
failures = summarize_orbit_results(results)
if failures and not args.ignore_errors:
return 1
return 0
def cmd_regions_import(args: argparse.Namespace) -> int:
if not args.source.exists():
raise ValueError(f"GeoJSON file does not exist: {args.source}")
stats = import_regions(args.db, args.source, simplify_tolerance=args.simplify_tolerance)
print(
"Imported region catalog: "
f"{stats.imported}/{stats.total_features} features "
f"from {stats.source_path} with simplify_tolerance={stats.simplify_tolerance}"
)
return 0
def cmd_regions_search(args: argparse.Namespace) -> int:
matches = search_regions(args.db, args.query, limit=args.limit, level=args.level)
if not matches:
print("No regions matched.")
return 0
for region in matches:
bbox_text = ""
if region.bbox is not None:
bbox_text = ",".join(f"{value:.6f}" for value in region.bbox)
print(f"{region.adcode} | {region.level:<8} | {region.label} | {bbox_text}")
return 0
def cmd_regions_export(args: argparse.Namespace) -> int:
region = get_region(args.db, args.adcode)
if region is None:
raise ValueError(f"region not found: {args.adcode}")
if args.format == "wkt":
output = region.wkt if args.raw else region.simplified_wkt
elif args.format == "geojson":
output = region.geometry_json
elif args.format == "bbox":
if region.bbox is None:
raise ValueError(f"region has no bbox: {args.adcode}")
output = ",".join(str(value) for value in region.bbox)
else:
payload: dict[str, Any] = {
"adcode": region.adcode,
"name": region.name,
"level": region.level,
"parent_adcode": region.parent_adcode,
"parent_name": region.parent_name,
"province_adcode": region.province_adcode,
"province_name": region.province_name,
"bbox": region.bbox,
"simplify_tolerance": region.simplify_tolerance,
"label": region.label,
}
output = json.dumps(payload, ensure_ascii=False, indent=2)
if args.out:
args.out.parent.mkdir(parents=True, exist_ok=True)
args.out.write_text(output, encoding="utf-8")
print(f"Wrote region export: {args.out}")
else:
print(output)
return 0
def cmd_projects_create(args: argparse.Namespace) -> int:
project = create_project(
args.db,
name=args.name,
data_root=args.data_root,
description=args.description,
slug=args.slug,
)
print(f"Created project: {project.id} | {project.slug} | {project.data_root}")
return 0
def cmd_projects_list(args: argparse.Namespace) -> int:
projects = list_projects(args.db)
if not projects:
print("No projects found.")
return 0
for project in projects:
desc = f" | {project.description}" if project.description else ""
print(f"{project.id} | {project.slug} | {project.name} | {project.data_root}{desc}")
return 0
def cmd_projects_show(args: argparse.Namespace) -> int:
project = get_project(args.db, args.project_id)
if project is None:
raise ValueError(f"project not found: {args.project_id}")
print(json.dumps(project.__dict__, ensure_ascii=False, indent=2))
return 0
def cmd_jobs_list(args: argparse.Namespace) -> int:
jobs = list_jobs(args.db, project_id=args.project_id, limit=args.limit)
if not jobs:
print("No jobs found.")
return 0
for job in jobs:
project_text = job.project_id if job.project_id is not None else "-"
target = job.region_name or job.source_value or job.source_type
print(
f"{job.id} | project={project_text} | {job.status:<14} | "
f"{job.start_time}..{job.end_time} | {target} | count={job.result_count}"
)
return 0
def cmd_jobs_show(args: argparse.Namespace) -> int:
job = get_job(args.db, args.job_id)
if job is None:
raise ValueError(f"job not found: {args.job_id}")
print(json.dumps(job.__dict__, ensure_ascii=False, indent=2))
scenes = list_job_scenes(args.db, args.job_id)
if scenes:
print("\nScenes:")
for scene in scenes:
size_text = f"{scene['size_mb']:.1f} MB" if scene["size_mb"] is not None else ""
print(
f" {scene['scene_id']} | {scene['start_time']} | {scene['processing_level']} | "
f"product={scene['product_status']} | orbit={scene['orbit_status']} | {size_text}"
)
if args.events:
events = list_job_events(args.db, args.job_id)
if events:
print("\nEvents:")
for event in events:
print(f" [{event['created_at']}] {event['level']}: {event['message']}")
return 0
def perform_query(args: argparse.Namespace, aoi_wkt: str):
params = SearchParams(
aoi_wkt=aoi_wkt,
start=args.start,
end=args.end,
processing_level=args.processing_level,
beam_mode=args.beam_mode,
flight_direction=args.flight_direction,
polarization=args.polarization,
relative_orbit=args.relative_orbit,
max_results=args.max_results,
)
print("Querying ASF Search ...")
return query_scenes(params)
def resolve_aoi(args: argparse.Namespace) -> dict[str, Any]:
if args.bbox:
return {
"aoi_wkt": bbox_to_wkt(args.bbox),
"source_type": "bbox",
"source_value": args.bbox,
"region_adcode": None,
"region_name": None,
}
if args.wkt:
return {
"aoi_wkt": args.wkt.strip(),
"source_type": "wkt",
"source_value": args.wkt.strip(),
"region_adcode": None,
"region_name": None,
}
if args.aoi_file:
return {
"aoi_wkt": aoi_file_to_wkt(args.aoi_file),
"source_type": "aoi_file",
"source_value": str(args.aoi_file),
"region_adcode": None,
"region_name": None,
}
if args.region_adcode:
region = get_region(args.db, args.region_adcode)
if region is None:
raise ValueError(f"region not found: {args.region_adcode}")
return {
"aoi_wkt": region_wkt(args.db, args.region_adcode, simplified=not args.raw_region),
"source_type": "region",
"source_value": region.label,
"region_adcode": region.adcode,
"region_name": region.label,
}
raise ValueError("one AOI input is required")
def print_results(results) -> None:
count = len(results)
print(f"Found {count} scene(s).")
if not count:
return
for index, product in enumerate(results, start=1):
props = product.properties
scene = scene_name_from_product(product) or props.get("fileID") or "<unknown>"
start = props.get("startTime") or ""
level = props.get("processingLevel") or ""
direction = props.get("flightDirection") or ""
rel_orbit = props.get("pathNumber") or ""
size = product_size_mb(props)
size_text = f"{size:.1f} MB" if size is not None else ""
print(f"{index:>3}. {scene} | {start} | {level} | {direction} | rel={rel_orbit} | {size_text}")
def record_product_downloads(db_path: Path, job_id: int, results, data_dir: Path | None) -> None:
for product in results:
scene_id = scene_name_from_product(product)
if scene_id is None:
continue
props = product.properties
if data_dir is None:
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="product",
status="skipped",
path=None,
url=string_value(props.get("url")),
bytes_total=float(props["bytes"]) if props.get("bytes") is not None else None,
bytes_done=0,
)
continue
expected = expected_product_path(product, data_dir)
status = "succeeded" if expected.exists() else "failed"
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="product",
status=status,
path=str(expected) if expected.exists() else None,
url=string_value(props.get("url")),
bytes_total=float(props["bytes"]) if props.get("bytes") is not None else None,
bytes_done=float(props["bytes"]) if expected.exists() and props.get("bytes") is not None else None,
error_message=None if expected.exists() else "Downloaded file not found at expected path",
)
def record_orbit_downloads(db_path: Path, job_id: int, orbit_results) -> None:
for result in orbit_results:
record_download_result(
db_path,
job_id=job_id,
scene_id=result.scene,
kind="orbit",
status="succeeded" if result.error is None else "failed",
path=str(result.path) if result.path is not None else None,
error_message=result.error,
)
def expected_product_path(product, data_dir: Path) -> Path:
props = product.properties
file_name = props.get("fileName")
if file_name:
return data_dir / str(file_name)
url = props.get("url")
if url:
return data_dir / Path(str(url)).name
scene = scene_name_from_product(product) or "unknown_scene"
return data_dir / f"{scene}.zip"
def summarize_orbit_results(results) -> int:
failures = [result for result in results if result.error]
for result in results:
if result.error:
print(f"Orbit failed: {result.scene}: {result.error}", file=sys.stderr)
else:
print(f"Orbit: {result.scene} -> {result.path.name}")
print(f"Orbit download finished: {len(results) - len(failures)} ok, {len(failures)} failed.")
return len(failures)
def ensure_project_exists(db_path: Path, project_id: int | None) -> None:
if project_id is None:
return
if get_project(db_path, project_id) is None:
raise ValueError(f"project not found: {project_id}")
def string_value(value: Any) -> str | None:
if value is None:
return None
return str(value)
def positive_int(value: str) -> int:
number = int(value)
if number < 1:
raise argparse.ArgumentTypeError("must be >= 1")
return number
@@ -0,0 +1 @@
"""Core configuration and constants."""
@@ -0,0 +1,27 @@
from __future__ import annotations
import os
from pathlib import Path
DEFAULT_DB_PATH = Path(".s1dl/app.db")
DEFAULT_API_HOST = "127.0.0.1"
DEFAULT_API_PORT = 8010
def configured_db_path() -> Path:
raw = os.getenv("S1DL_DB_PATH")
if not raw:
return DEFAULT_DB_PATH
return Path(raw)
def configured_api_host() -> str:
return os.getenv("S1DL_API_HOST", DEFAULT_API_HOST)
def configured_api_port() -> int:
raw = os.getenv("S1DL_API_PORT")
if not raw:
return DEFAULT_API_PORT
return int(raw)
@@ -0,0 +1,40 @@
from __future__ import annotations
PROVINCES: dict[str, str] = {
"110000": "北京市",
"120000": "天津市",
"130000": "河北省",
"140000": "山西省",
"150000": "内蒙古自治区",
"210000": "辽宁省",
"220000": "吉林省",
"230000": "黑龙江省",
"310000": "上海市",
"320000": "江苏省",
"330000": "浙江省",
"340000": "安徽省",
"350000": "福建省",
"360000": "江西省",
"370000": "山东省",
"410000": "河南省",
"420000": "湖北省",
"430000": "湖南省",
"440000": "广东省",
"450000": "广西壮族自治区",
"460000": "海南省",
"500000": "重庆市",
"510000": "四川省",
"520000": "贵州省",
"530000": "云南省",
"540000": "西藏自治区",
"610000": "陕西省",
"620000": "甘肃省",
"630000": "青海省",
"640000": "宁夏回族自治区",
"650000": "新疆维吾尔自治区",
"710000": "台湾省",
"810000": "香港特别行政区",
"820000": "澳门特别行政区",
}
@@ -0,0 +1 @@
"""Application services."""
@@ -0,0 +1,75 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Iterable, Sequence
def bbox_to_wkt(value: str) -> str:
parts = [part.strip() for part in value.split(",")]
if len(parts) != 4:
raise ValueError("--bbox must be min_lon,min_lat,max_lon,max_lat")
try:
min_lon, min_lat, max_lon, max_lat = (float(part) for part in parts)
except ValueError as exc:
raise ValueError("--bbox values must be numbers") from exc
if min_lon >= max_lon or min_lat >= max_lat:
raise ValueError("--bbox min values must be smaller than max values")
return (
f"POLYGON(({min_lon} {min_lat},{max_lon} {min_lat},"
f"{max_lon} {max_lat},{min_lon} {max_lat},{min_lon} {min_lat}))"
)
def aoi_file_to_wkt(path: Path) -> str:
text = path.read_text(encoding="utf-8").strip()
if not text:
raise ValueError(f"AOI file is empty: {path}")
if text[0] != "{":
return text
data = json.loads(text)
geometry = geojson_geometry(data)
return geometry_to_wkt(geometry)
def geojson_geometry(data: dict[str, Any]) -> dict[str, Any]:
kind = data.get("type")
if kind == "FeatureCollection":
features = data.get("features") or []
if not features:
raise ValueError("GeoJSON FeatureCollection has no features")
return geojson_geometry(features[0])
if kind == "Feature":
geometry = data.get("geometry")
if not geometry:
raise ValueError("GeoJSON Feature has no geometry")
return geometry
if kind in {"Polygon", "MultiPolygon", "Point", "MultiPoint", "LineString"}:
return data
raise ValueError(f"unsupported GeoJSON type: {kind}")
def geometry_to_wkt(geometry: dict[str, Any]) -> str:
kind = geometry.get("type")
coordinates = geometry.get("coordinates")
if kind == "Point":
lon, lat = coordinates
return f"POINT({lon} {lat})"
if kind == "MultiPoint":
return "MULTIPOINT(" + ",".join(f"({lon} {lat})" for lon, lat in coordinates) + ")"
if kind == "LineString":
return "LINESTRING(" + coords_to_text(coordinates) + ")"
if kind == "Polygon":
return "POLYGON(" + ",".join(f"({coords_to_text(ring)})" for ring in coordinates) + ")"
if kind == "MultiPolygon":
polygons = []
for polygon in coordinates:
rings = ",".join(f"({coords_to_text(ring)})" for ring in polygon)
polygons.append(f"({rings})")
return "MULTIPOLYGON(" + ",".join(polygons) + ")"
raise ValueError(f"unsupported GeoJSON geometry type: {kind}")
def coords_to_text(coords: Iterable[Sequence[float]]) -> str:
return ",".join(f"{lon} {lat}" for lon, lat, *_ in coords)
@@ -0,0 +1,151 @@
from __future__ import annotations
import csv
import getpass
import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence
import asf_search as asf
DEFAULT_FIELDS = (
"scene",
"startTime",
"stopTime",
"processingLevel",
"beamModeType",
"flightDirection",
"polarization",
"pathNumber",
"platform",
"sizeMB",
)
@dataclass(frozen=True)
class SearchParams:
aoi_wkt: str
start: str
end: str
processing_level: str = "SLC"
beam_mode: str | None = "IW"
flight_direction: str | None = None
polarization: str | None = None
relative_orbit: int | None = None
max_results: int = 50
def query_scenes(params: SearchParams):
kwargs: dict[str, Any] = {
"platform": [asf.PLATFORM.SENTINEL1],
"processingLevel": params.processing_level,
"intersectsWith": params.aoi_wkt,
"start": params.start,
"end": params.end,
"maxResults": params.max_results,
}
if params.beam_mode:
kwargs["beamMode"] = params.beam_mode
if params.flight_direction:
kwargs["flightDirection"] = params.flight_direction.upper()
if params.polarization:
kwargs["polarization"] = params.polarization
if params.relative_orbit:
kwargs["relativeOrbit"] = params.relative_orbit
return asf.geo_search(**kwargs)
def resolve_credentials(
username: str | None = None,
password: str | None = None,
) -> tuple[str | None, str | None]:
resolved_username = username or os.getenv("EARTHDATA_USER") or os.getenv("EARTHDATA_USERNAME")
resolved_password = password or os.getenv("EARTHDATA_PASS") or os.getenv("EARTHDATA_PASSWORD")
return resolved_username, resolved_password
def credentials_available(
username: str | None = None,
password: str | None = None,
) -> bool:
resolved_username, resolved_password = resolve_credentials(username, password)
return bool(resolved_username and resolved_password)
def build_session(
username: str | None = None,
password: str | None = None,
ask_password: bool = False,
) -> asf.ASFSession:
username, password = resolve_credentials(username, password)
if ask_password:
password = getpass.getpass("Earthdata password: ")
if not username or not password:
raise ValueError(
"ASF product downloads require NASA Earthdata credentials. "
"Set EARTHDATA_USER/EARTHDATA_PASS or pass --username/--ask-password."
)
return asf.ASFSession().auth_with_creds(username, password)
def product_size_mb(props: dict[str, Any]) -> float | None:
if props.get("sizeMB") is not None:
return float(props["sizeMB"])
if props.get("bytes") is not None:
return float(props["bytes"]) / (1024 * 1024)
return None
def scene_name_from_product(product) -> str | None:
from .orbit_client import normalize_scene_name
props = getattr(product, "properties", {})
candidates: Sequence[Any] = (
props.get("sceneName"),
props.get("fileID"),
props.get("fileName"),
props.get("url"),
)
for candidate in candidates:
if not candidate:
continue
scene = normalize_scene_name(str(candidate))
if scene:
return scene
return None
def manifest_row(product) -> dict[str, Any]:
props = product.properties
return {
"scene": scene_name_from_product(product),
"startTime": props.get("startTime"),
"stopTime": props.get("stopTime"),
"processingLevel": props.get("processingLevel"),
"beamModeType": props.get("beamModeType"),
"flightDirection": props.get("flightDirection"),
"polarization": props.get("polarization"),
"pathNumber": props.get("pathNumber"),
"platform": props.get("platform"),
"sizeMB": product_size_mb(props),
}
def write_manifest(results, path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
suffix = path.suffix.lower()
rows = [manifest_row(product) for product in results]
if suffix == ".json":
path.write_text(json.dumps(rows, indent=2, ensure_ascii=False, default=str), encoding="utf-8")
return
if suffix not in {"", ".csv"}:
raise ValueError("manifest path must end with .csv or .json")
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=DEFAULT_FIELDS)
writer.writeheader()
writer.writerows(rows)
@@ -0,0 +1,370 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence
import asf_search as asf
from sentinel_orbit_downloader.services.asf_client import build_session, credentials_available, scene_name_from_product
from sentinel_orbit_downloader.services.orbit_client import download_orbits
from sentinel_orbit_downloader.services.project_store import (
get_job,
get_project,
list_job_scenes,
record_download_result,
record_job_event,
update_search_job,
)
@dataclass(frozen=True)
class DownloadJobRequest:
job_id: int
data_dir: Path | None = None
orbit_dir: Path | None = None
manifest_path: Path | None = None
skip_data: bool = False
skip_orbits: bool = False
processes: int = 2
ignore_orbit_errors: bool = False
username: str | None = None
password: str | None = None
@dataclass(frozen=True)
class DownloadJobPaths:
root: Path
data_dir: Path
orbit_dir: Path
manifest_path: Path
def resolve_download_paths(db_path: Path, request: DownloadJobRequest) -> DownloadJobPaths:
job = get_job(db_path, request.job_id)
if job is None:
raise ValueError(f"Job not found: {request.job_id}")
root = None
if request.data_dir is not None:
root = request.data_dir.parent if request.orbit_dir is None else request.data_dir.parent
elif request.orbit_dir is not None:
root = request.orbit_dir.parent
elif job.data_dir:
root = Path(job.data_dir).parent
elif job.project_id is not None:
project = get_project(db_path, job.project_id)
if project is not None:
root = Path(project.data_root) / "jobs" / f"job-{job.id:06d}"
if root is None:
root = Path(".s1dl") / "downloads" / f"job-{job.id:06d}"
data_dir = request.data_dir or (Path(job.data_dir) if job.data_dir else root / "data")
orbit_dir = request.orbit_dir or (Path(job.orbit_dir) if job.orbit_dir else root / "orbits")
manifest_path = request.manifest_path or (
Path(job.manifest_path) if job.manifest_path else root / "manifest.csv"
)
return DownloadJobPaths(root=root, data_dir=data_dir, orbit_dir=orbit_dir, manifest_path=manifest_path)
def queue_download_job(db_path: Path, request: DownloadJobRequest) -> DownloadJobPaths:
job = get_job(db_path, request.job_id)
if job is None:
raise ValueError(f"Job not found: {request.job_id}")
scenes = list_job_scenes(db_path, request.job_id)
if not scenes:
raise ValueError(f"Job has no scenes to download: {request.job_id}")
paths = resolve_download_paths(db_path, request)
paths.root.mkdir(parents=True, exist_ok=True)
update_search_job(
db_path,
request.job_id,
status="running",
result_count=job.result_count,
data_dir=str(paths.data_dir),
orbit_dir=str(paths.orbit_dir),
manifest_path=str(paths.manifest_path),
error_message=None,
)
record_job_event(
db_path,
request.job_id,
"info",
f"Download queued for {len(scenes)} scene(s) into {paths.root}",
)
return paths
def execute_download_job(db_path: Path, request: DownloadJobRequest) -> None:
job = get_job(db_path, request.job_id)
if job is None:
return
paths = resolve_download_paths(db_path, request)
paths.root.mkdir(parents=True, exist_ok=True)
try:
scenes = list_job_scenes(db_path, request.job_id)
scene_ids = [scene["scene_id"] for scene in scenes]
if not scene_ids:
update_search_job(
db_path,
request.job_id,
status="failed",
result_count=0,
data_dir=str(paths.data_dir),
orbit_dir=str(paths.orbit_dir),
manifest_path=str(paths.manifest_path),
error_message="No scenes available for download",
)
record_job_event(db_path, request.job_id, "error", "Download job failed: no scenes available")
return
if not request.skip_data and not credentials_available(request.username, request.password):
message = (
"Download failed: Sentinel-1 product downloads require NASA Earthdata credentials. "
"Fill in Earthdata username/password, or set EARTHDATA_USER and EARTHDATA_PASS first."
)
for scene_id in scene_ids:
record_download_result(
db_path,
job_id=request.job_id,
scene_id=scene_id,
kind="product",
status="failed",
path=None,
error_message=message,
)
record_download_result(
db_path,
job_id=request.job_id,
scene_id=scene_id,
kind="orbit",
status="skipped",
path=None,
)
update_search_job(
db_path,
request.job_id,
status="failed",
result_count=job.result_count,
data_dir=str(paths.data_dir),
orbit_dir=str(paths.orbit_dir),
manifest_path=str(paths.manifest_path),
error_message=message,
)
record_job_event(db_path, request.job_id, "error", message)
return
product_error: str | None = None
if request.skip_data:
record_job_event(db_path, request.job_id, "info", "Product download skipped by request")
record_product_skips(db_path, request.job_id, scene_ids)
else:
results = asf.granule_search(scene_ids)
products = {scene_name_from_product(product): product for product in results}
ordered_products = [products[scene_id] for scene_id in scene_ids if scene_id in products]
missing_scene_ids = [scene_id for scene_id in scene_ids if scene_id not in products]
if missing_scene_ids:
record_job_event(
db_path,
request.job_id,
"warning",
f"{len(missing_scene_ids)} scene(s) could not be reloaded from ASF search",
)
for scene_id in missing_scene_ids:
record_download_result(
db_path,
job_id=request.job_id,
scene_id=scene_id,
kind="product",
status="failed",
path=None,
error_message="Scene was not returned by ASF granule_search",
)
if request.skip_orbits:
record_download_result(
db_path,
job_id=request.job_id,
scene_id=scene_id,
kind="orbit",
status="skipped",
path=None,
)
paths.data_dir.mkdir(parents=True, exist_ok=True)
try:
session = build_session(username=request.username, password=request.password)
results.download(path=str(paths.data_dir), session=session, processes=request.processes)
record_job_event(
db_path,
request.job_id,
"info",
f"Product download finished into {paths.data_dir}",
)
except Exception as exc:
product_error = str(exc)
record_job_event(db_path, request.job_id, "error", f"Product download error: {exc}")
record_product_downloads(db_path, request.job_id, ordered_products, paths.data_dir, product_error)
orbit_failures = 0
if request.skip_orbits:
record_job_event(db_path, request.job_id, "info", "Orbit download skipped by request")
record_orbit_skips(db_path, request.job_id, scene_ids)
else:
paths.orbit_dir.mkdir(parents=True, exist_ok=True)
orbit_results = download_orbits(scene_ids, paths.orbit_dir)
orbit_failures = sum(1 for result in orbit_results if result.error)
for result in orbit_results:
record_download_result(
db_path,
job_id=request.job_id,
scene_id=result.scene,
kind="orbit",
status="succeeded" if result.error is None else "failed",
path=str(result.path) if result.path is not None else None,
error_message=result.error,
)
record_job_event(
db_path,
request.job_id,
"info",
f"Orbit download finished: {len(orbit_results) - orbit_failures} ok, {orbit_failures} failed",
)
product_failures = count_download_failures(db_path, request.job_id, "product")
effective_orbit_failures = 0 if request.ignore_orbit_errors else orbit_failures
final_status = "succeeded"
if product_failures or effective_orbit_failures:
final_status = "partial_failed"
if product_failures and (request.skip_orbits or effective_orbit_failures == len(scene_ids)):
final_status = "failed"
update_search_job(
db_path,
request.job_id,
status=final_status,
result_count=job.result_count,
data_dir=str(paths.data_dir),
orbit_dir=str(paths.orbit_dir),
manifest_path=str(paths.manifest_path),
error_message=product_error,
)
except Exception as exc:
update_search_job(
db_path,
request.job_id,
status="failed",
result_count=job.result_count,
data_dir=str(paths.data_dir),
orbit_dir=str(paths.orbit_dir),
manifest_path=str(paths.manifest_path),
error_message=str(exc),
)
record_job_event(db_path, request.job_id, "error", f"Download execution failed: {exc}")
def record_product_downloads(
db_path: Path,
job_id: int,
products: Sequence,
data_dir: Path | None,
fallback_error: str | None = None,
) -> None:
for product in products:
scene_id = scene_name_from_product(product)
if scene_id is None:
continue
props = product.properties
if data_dir is None:
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="product",
status="skipped",
path=None,
url=string_value(props.get("url")),
bytes_total=float(props["bytes"]) if props.get("bytes") is not None else None,
bytes_done=0,
)
continue
expected = expected_product_path(product, data_dir)
exists = expected.exists()
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="product",
status="succeeded" if exists else "failed",
path=str(expected) if exists else None,
url=string_value(props.get("url")),
bytes_total=float(props["bytes"]) if props.get("bytes") is not None else None,
bytes_done=float(props["bytes"]) if exists and props.get("bytes") is not None else None,
error_message=None if exists else (fallback_error or "Downloaded file not found at expected path"),
)
def record_product_skips(db_path: Path, job_id: int, scene_ids: Sequence[str]) -> None:
for scene_id in scene_ids:
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="product",
status="skipped",
path=None,
)
def record_orbit_skips(db_path: Path, job_id: int, scene_ids: Sequence[str]) -> None:
for scene_id in scene_ids:
record_download_result(
db_path,
job_id=job_id,
scene_id=scene_id,
kind="orbit",
status="skipped",
path=None,
)
def expected_product_path(product, data_dir: Path) -> Path:
props = product.properties
file_name = props.get("fileName")
if file_name:
return data_dir / str(file_name)
url = props.get("url")
if url:
return data_dir / Path(str(url)).name
scene = scene_name_from_product(product) or "unknown_scene"
return data_dir / f"{scene}.zip"
def string_value(value) -> str | None:
if value is None:
return None
return str(value)
def count_download_failures(db_path: Path, job_id: int, kind: str) -> int:
from sentinel_orbit_downloader.storage.db import connect
if kind == "product":
column = "product_status"
elif kind == "orbit":
column = "orbit_status"
else:
raise ValueError(f"unsupported download kind: {kind}")
with connect(db_path) as connection:
row = connection.execute(
f"""
SELECT COUNT(*) AS failure_count
FROM job_scenes
WHERE job_id = ? AND {column} = 'failed'
""",
(job_id,),
).fetchone()
return int(row["failure_count"])
@@ -0,0 +1,91 @@
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Sequence
import s1_orbits
from s1_orbits.exceptions import InvalidSceneError, OrbitNotFoundError
SCENE_RE = re.compile(r"^(S1[A-D]_.+?)(?:\.SAFE|\.zip)?$", re.IGNORECASE)
@dataclass(frozen=True)
class OrbitResult:
scene: str
path: Path | None
error: str | None = None
def normalize_scene_name(value: str) -> str | None:
name = Path(value).name.strip()
for suffix in (".SAFE", ".safe", ".ZIP", ".zip"):
if name.endswith(suffix):
name = name[: -len(suffix)]
break
match = SCENE_RE.match(name)
if not match:
return None
return match.group(1)
def unique(values: Iterable[str]) -> list[str]:
seen = set()
output = []
for value in values:
if value in seen:
continue
seen.add(value)
output.append(value)
return output
def collect_orbit_scenes(
scenes: Sequence[str] | None = None,
scene_file: Path | None = None,
data_dir: Path | None = None,
) -> list[str]:
if scenes:
return unique(filter(None, (normalize_scene_name(scene) for scene in scenes)))
if scene_file:
scene_names = []
for line in scene_file.read_text(encoding="utf-8").splitlines():
clean = line.strip()
if not clean or clean.startswith("#"):
continue
scene_names.append(normalize_scene_name(clean))
return unique(filter(None, scene_names))
if not data_dir:
return []
found = []
for path in data_dir.iterdir():
if not path.name.upper().startswith(("S1A_", "S1B_", "S1C_", "S1D_")):
continue
scene = normalize_scene_name(path.name)
if scene:
found.append(scene)
return unique(found)
def scene_names_from_results(results) -> list[str]:
from .asf_client import scene_name_from_product
scenes = [scene_name_from_product(product) for product in results]
return unique(scene for scene in scenes if scene)
def download_orbits(scenes: Sequence[str], orbit_dir: Path) -> list[OrbitResult]:
orbit_dir.mkdir(parents=True, exist_ok=True)
results: list[OrbitResult] = []
for scene in scenes:
try:
path = s1_orbits.fetch_for_scene(scene, dir=orbit_dir)
results.append(OrbitResult(scene=scene, path=path))
except (InvalidSceneError, OrbitNotFoundError) as exc:
message = str(exc) or exc.__class__.__name__
results.append(OrbitResult(scene=scene, path=None, error=message))
return results
@@ -0,0 +1,545 @@
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from sentinel_orbit_downloader.storage.db import connect
NON_SLUG_RE = re.compile(r"[^\w\u4e00-\u9fff-]+", re.UNICODE)
@dataclass(frozen=True)
class ProjectRecord:
id: int
name: str
slug: str
data_root: str
description: str | None
created_at: str
updated_at: str
def create_project(
db_path: Path,
name: str,
data_root: Path,
description: str | None = None,
slug: str | None = None,
) -> ProjectRecord:
final_slug = slug or slugify(name)
if not final_slug:
raise ValueError("project slug is empty; choose a clearer project name")
data_root.mkdir(parents=True, exist_ok=True)
with connect(db_path) as connection:
exists = connection.execute("SELECT id FROM projects WHERE slug = ?", (final_slug,)).fetchone()
if exists is not None:
raise ValueError(f"project slug already exists: {final_slug}")
cursor = connection.execute(
"""
INSERT INTO projects(name, slug, data_root, description)
VALUES (?, ?, ?, ?)
""",
(name, final_slug, str(data_root), description),
)
project_id = int(cursor.lastrowid)
row = connection.execute("SELECT * FROM projects WHERE id = ?", (project_id,)).fetchone()
return row_to_project(row)
def list_projects(db_path: Path) -> list[ProjectRecord]:
with connect(db_path) as connection:
rows = connection.execute(
"SELECT * FROM projects ORDER BY created_at DESC, id DESC"
).fetchall()
return [row_to_project(row) for row in rows]
def get_project(db_path: Path, project_id: int) -> ProjectRecord | None:
with connect(db_path) as connection:
row = connection.execute("SELECT * FROM projects WHERE id = ?", (project_id,)).fetchone()
if row is None:
return None
return row_to_project(row)
def slugify(value: str) -> str:
text = value.strip().lower().replace(" ", "-")
text = NON_SLUG_RE.sub("-", text)
text = re.sub(r"-{2,}", "-", text)
text = text.strip("-")
return text
def row_to_project(row) -> ProjectRecord:
return ProjectRecord(
id=int(row["id"]),
name=str(row["name"]),
slug=str(row["slug"]),
data_root=str(row["data_root"]),
description=row["description"],
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
@dataclass(frozen=True)
class SearchJobRecord:
id: int
project_id: int | None
aoi_id: int | None
source_type: str
source_value: str | None
region_adcode: str | None
region_name: str | None
start_time: str
end_time: str
processing_level: str
beam_mode: str | None
flight_direction: str | None
polarization: str | None
relative_orbit: int | None
max_results: int
status: str
result_count: int | None
data_dir: str | None
orbit_dir: str | None
manifest_path: str | None
error_message: str | None
created_at: str
finished_at: str | None
def create_search_job(
db_path: Path,
*,
project_id: int | None,
source_type: str,
source_value: str | None,
region_adcode: str | None,
region_name: str | None,
start_time: str,
end_time: str,
processing_level: str,
beam_mode: str | None,
flight_direction: str | None,
polarization: str | None,
relative_orbit: int | None,
max_results: int,
status: str = "running",
data_dir: str | None = None,
orbit_dir: str | None = None,
manifest_path: str | None = None,
) -> SearchJobRecord:
with connect(db_path) as connection:
cursor = connection.execute(
"""
INSERT INTO search_jobs(
project_id, aoi_id, source_type, source_value, region_adcode, region_name,
start_time, end_time, processing_level, beam_mode, flight_direction,
polarization, relative_orbit, max_results, status, data_dir, orbit_dir, manifest_path
)
VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
project_id,
source_type,
source_value,
region_adcode,
region_name,
start_time,
end_time,
processing_level,
beam_mode,
flight_direction,
polarization,
relative_orbit,
max_results,
status,
data_dir,
orbit_dir,
manifest_path,
),
)
job_id = int(cursor.lastrowid)
row = connection.execute("SELECT * FROM search_jobs WHERE id = ?", (job_id,)).fetchone()
return row_to_search_job(row)
def update_search_job(
db_path: Path,
job_id: int,
*,
status: str,
result_count: int | None = None,
data_dir: str | None = None,
orbit_dir: str | None = None,
manifest_path: str | None = None,
error_message: str | None = None,
) -> None:
final_statuses = {"succeeded", "failed", "partial_failed", "canceled"}
with connect(db_path) as connection:
if status in final_statuses:
connection.execute(
"""
UPDATE search_jobs
SET
status = ?,
result_count = COALESCE(?, result_count),
data_dir = COALESCE(?, data_dir),
orbit_dir = COALESCE(?, orbit_dir),
manifest_path = COALESCE(?, manifest_path),
error_message = ?,
finished_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(
status,
result_count,
data_dir,
orbit_dir,
manifest_path,
error_message,
job_id,
),
)
else:
connection.execute(
"""
UPDATE search_jobs
SET
status = ?,
result_count = COALESCE(?, result_count),
data_dir = COALESCE(?, data_dir),
orbit_dir = COALESCE(?, orbit_dir),
manifest_path = COALESCE(?, manifest_path),
error_message = ?,
finished_at = NULL
WHERE id = ?
""",
(
status,
result_count,
data_dir,
orbit_dir,
manifest_path,
error_message,
job_id,
),
)
def record_job_event(db_path: Path, job_id: int, level: str, message: str) -> None:
with connect(db_path) as connection:
if not job_exists(connection, job_id):
return
connection.execute(
"INSERT INTO job_events(job_id, level, message) VALUES (?, ?, ?)",
(job_id, level, message),
)
def save_query_results(db_path: Path, job_id: int, results) -> None:
with connect(db_path) as connection:
if not job_exists(connection, job_id):
return
for product in results:
scene_id = scene_id_from_product(product)
if scene_id is None:
continue
props = product.properties
geometry_json = json.dumps(product.geometry, ensure_ascii=False, default=str)
properties_json = json.dumps(props, ensure_ascii=False, default=str)
connection.execute(
"""
INSERT INTO scenes(
scene_id, platform, start_time, stop_time, processing_level, beam_mode,
flight_direction, polarization, relative_orbit, size_bytes, size_mb,
download_url, geometry_json, properties_json
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(scene_id) DO UPDATE SET
platform = excluded.platform,
start_time = excluded.start_time,
stop_time = excluded.stop_time,
processing_level = excluded.processing_level,
beam_mode = excluded.beam_mode,
flight_direction = excluded.flight_direction,
polarization = excluded.polarization,
relative_orbit = excluded.relative_orbit,
size_bytes = excluded.size_bytes,
size_mb = excluded.size_mb,
download_url = excluded.download_url,
geometry_json = excluded.geometry_json,
properties_json = excluded.properties_json
""",
(
scene_id,
platform_from_scene(scene_id),
string_value(props.get("startTime")),
string_value(props.get("stopTime")),
string_value(props.get("processingLevel")),
string_value(props.get("beamModeType")),
string_value(props.get("flightDirection")),
string_value(props.get("polarization")),
int(props["pathNumber"]) if props.get("pathNumber") is not None else None,
float(props["bytes"]) if props.get("bytes") is not None else None,
product_size_mb(props),
string_value(props.get("url")),
geometry_json,
properties_json,
),
)
connection.execute(
"""
INSERT INTO job_scenes(job_id, scene_id, selected, product_status, orbit_status)
VALUES (?, ?, 1, 'pending', 'pending')
ON CONFLICT(job_id, scene_id) DO NOTHING
""",
(job_id, scene_id),
)
def record_download_result(
db_path: Path,
*,
job_id: int,
scene_id: str,
kind: str,
status: str,
path: str | None,
url: str | None = None,
bytes_total: float | None = None,
bytes_done: float | None = None,
error_message: str | None = None,
) -> None:
with connect(db_path) as connection:
if not job_exists(connection, job_id):
return
connection.execute(
"""
INSERT INTO downloads(
job_id, scene_id, kind, status, url, path, bytes_total, bytes_done, error_message
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
job_id,
scene_id,
kind,
status,
url,
path,
bytes_total,
bytes_done,
error_message,
),
)
if kind == "product":
connection.execute(
"""
UPDATE job_scenes
SET product_status = ?, product_path = COALESCE(?, product_path)
WHERE job_id = ? AND scene_id = ?
""",
(status, path, job_id, scene_id),
)
elif kind == "orbit":
connection.execute(
"""
UPDATE job_scenes
SET orbit_status = ?, orbit_path = COALESCE(?, orbit_path)
WHERE job_id = ? AND scene_id = ?
""",
(status, path, job_id, scene_id),
)
def delete_job(db_path: Path, job_id: int) -> bool:
with connect(db_path) as connection:
if not job_exists(connection, job_id):
return False
delete_job_rows(connection, job_id)
prune_orphan_scenes(connection)
return True
def clear_jobs(
db_path: Path,
project_id: int | None = None,
include_running: bool = False,
) -> list[int]:
sql = "SELECT id FROM search_jobs"
conditions: list[str] = []
params: list[Any] = []
if project_id is not None:
conditions.append("project_id = ?")
params.append(project_id)
if not include_running:
conditions.append("status != 'running'")
if conditions:
sql += " WHERE " + " AND ".join(conditions)
sql += " ORDER BY id"
with connect(db_path) as connection:
rows = connection.execute(sql, params).fetchall()
job_ids = [int(row["id"]) for row in rows]
for job_id in job_ids:
delete_job_rows(connection, job_id)
prune_orphan_scenes(connection)
return job_ids
def list_jobs(db_path: Path, project_id: int | None = None, limit: int = 20) -> list[SearchJobRecord]:
sql = """
SELECT *
FROM search_jobs
"""
params: list[Any] = []
if project_id is not None:
sql += " WHERE project_id = ?"
params.append(project_id)
sql += " ORDER BY id DESC LIMIT ?"
params.append(limit)
with connect(db_path) as connection:
rows = connection.execute(sql, params).fetchall()
return [row_to_search_job(row) for row in rows]
def get_job(db_path: Path, job_id: int) -> SearchJobRecord | None:
with connect(db_path) as connection:
row = connection.execute("SELECT * FROM search_jobs WHERE id = ?", (job_id,)).fetchone()
if row is None:
return None
return row_to_search_job(row)
def list_job_scenes(db_path: Path, job_id: int) -> list[dict[str, Any]]:
with connect(db_path) as connection:
rows = connection.execute(
"""
SELECT
js.job_id,
js.scene_id,
js.selected,
js.product_status,
js.orbit_status,
js.product_path,
js.orbit_path,
s.platform,
s.start_time,
s.stop_time,
s.processing_level,
s.flight_direction,
s.relative_orbit,
s.size_mb
FROM job_scenes AS js
JOIN scenes AS s ON s.scene_id = js.scene_id
WHERE js.job_id = ?
ORDER BY s.start_time, js.scene_id
""",
(job_id,),
).fetchall()
return [dict(row) for row in rows]
def list_job_events(db_path: Path, job_id: int) -> list[dict[str, Any]]:
with connect(db_path) as connection:
rows = connection.execute(
"SELECT id, level, message, created_at FROM job_events WHERE job_id = ? ORDER BY id",
(job_id,),
).fetchall()
return [dict(row) for row in rows]
def row_to_search_job(row) -> SearchJobRecord:
return SearchJobRecord(
id=int(row["id"]),
project_id=int(row["project_id"]) if row["project_id"] is not None else None,
aoi_id=int(row["aoi_id"]) if row["aoi_id"] is not None else None,
source_type=str(row["source_type"]),
source_value=row["source_value"],
region_adcode=row["region_adcode"],
region_name=row["region_name"],
start_time=str(row["start_time"]),
end_time=str(row["end_time"]),
processing_level=str(row["processing_level"]),
beam_mode=row["beam_mode"],
flight_direction=row["flight_direction"],
polarization=row["polarization"],
relative_orbit=int(row["relative_orbit"]) if row["relative_orbit"] is not None else None,
max_results=int(row["max_results"]),
status=str(row["status"]),
result_count=int(row["result_count"]) if row["result_count"] is not None else None,
data_dir=row["data_dir"],
orbit_dir=row["orbit_dir"],
manifest_path=row["manifest_path"],
error_message=row["error_message"],
created_at=str(row["created_at"]),
finished_at=row["finished_at"],
)
def scene_id_from_product(product) -> str | None:
from sentinel_orbit_downloader.services.asf_client import scene_name_from_product
return scene_name_from_product(product)
def platform_from_scene(scene_id: str) -> str | None:
parts = scene_id.split("_", 1)
if not parts:
return None
return parts[0]
def product_size_mb(props: dict[str, Any]) -> float | None:
if props.get("sizeMB") is not None:
return float(props["sizeMB"])
if props.get("bytes") is not None:
return float(props["bytes"]) / (1024 * 1024)
return None
def string_value(value: Any) -> str | None:
if value is None:
return None
return str(value)
def job_exists(connection, job_id: int) -> bool:
row = connection.execute(
"SELECT 1 FROM search_jobs WHERE id = ?",
(job_id,),
).fetchone()
return row is not None
def delete_job_rows(connection, job_id: int) -> None:
connection.execute("DELETE FROM job_events WHERE job_id = ?", (job_id,))
connection.execute("DELETE FROM downloads WHERE job_id = ?", (job_id,))
connection.execute("DELETE FROM job_scenes WHERE job_id = ?", (job_id,))
connection.execute("DELETE FROM search_jobs WHERE id = ?", (job_id,))
def prune_orphan_scenes(connection) -> None:
connection.execute(
"""
DELETE FROM scenes
WHERE NOT EXISTS (
SELECT 1
FROM job_scenes
WHERE job_scenes.scene_id = scenes.scene_id
)
"""
)
@@ -0,0 +1,285 @@
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from shapely.geometry import shape
from sentinel_orbit_downloader.core.provinces import PROVINCES
from sentinel_orbit_downloader.storage.db import connect, set_setting
@dataclass(frozen=True)
class RegionImportStats:
source_path: Path
simplify_tolerance: float
total_features: int
imported: int
@dataclass(frozen=True)
class RegionRecord:
adcode: str
name: str
level: str
parent_adcode: str | None
parent_name: str | None
province_adcode: str | None
province_name: str | None
center_lon: float | None
center_lat: float | None
centroid_lon: float | None
centroid_lat: float | None
min_lon: float | None
min_lat: float | None
max_lon: float | None
max_lat: float | None
children_num: int | None
tree_id: str | None
source_path: str | None
geometry_json: str
wkt: str
simplified_wkt: str
simplify_tolerance: float
geometry_hash: str
@property
def bbox(self) -> tuple[float, float, float, float] | None:
values = (self.min_lon, self.min_lat, self.max_lon, self.max_lat)
if any(value is None for value in values):
return None
return values # type: ignore[return-value]
@property
def label(self) -> str:
parts = []
if self.province_name and self.province_name != self.name:
parts.append(self.province_name)
if self.parent_name and self.parent_name not in parts and self.parent_name != self.name:
parts.append(self.parent_name)
parts.append(self.name)
return " / ".join(parts)
def import_regions(
db_path: Path,
source_path: Path,
simplify_tolerance: float = 0.01,
) -> RegionImportStats:
data = json.loads(source_path.read_text(encoding="utf-8"))
features = data.get("features") or []
rows = [feature_to_row(feature, source_path, simplify_tolerance) for feature in features]
with connect(db_path) as connection:
connection.execute("DELETE FROM regions")
connection.executemany(
"""
INSERT INTO regions (
adcode, name, level, parent_adcode, province_adcode, province_name,
center_lon, center_lat, centroid_lon, centroid_lat,
min_lon, min_lat, max_lon, max_lat, children_num, tree_id,
source_path, geometry_json, wkt, simplified_wkt, simplify_tolerance, geometry_hash
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
set_setting(connection, "regions.source_path", str(source_path.resolve()))
set_setting(connection, "regions.simplify_tolerance", str(simplify_tolerance))
set_setting(connection, "regions.feature_count", str(len(features)))
return RegionImportStats(
source_path=source_path,
simplify_tolerance=simplify_tolerance,
total_features=len(features),
imported=len(rows),
)
def search_regions(
db_path: Path,
query: str,
limit: int = 20,
level: str | None = None,
) -> list[RegionRecord]:
sql = """
SELECT
r.*,
parent.name AS parent_name
FROM regions AS r
LEFT JOIN regions AS parent ON parent.adcode = r.parent_adcode
WHERE (
r.name LIKE :like_query
OR r.adcode LIKE :like_query
OR IFNULL(r.province_name, '') LIKE :like_query
)
"""
params: dict[str, Any] = {
"like_query": f"%{query}%",
"exact_query": query,
"prefix_query": f"{query}%",
"limit": limit,
}
if level:
sql += " AND r.level = :level"
params["level"] = level
sql += """
ORDER BY
CASE
WHEN r.name = :exact_query THEN 0
WHEN r.name LIKE :prefix_query THEN 1
ELSE 2
END,
LENGTH(r.name),
r.level,
r.adcode
LIMIT :limit
"""
with connect(db_path) as connection:
rows = connection.execute(sql, params).fetchall()
return [row_to_region(row) for row in rows]
def get_region(db_path: Path, adcode: str) -> RegionRecord | None:
with connect(db_path) as connection:
row = connection.execute(
"""
SELECT
r.*,
parent.name AS parent_name
FROM regions AS r
LEFT JOIN regions AS parent ON parent.adcode = r.parent_adcode
WHERE r.adcode = ?
""",
(str(adcode),),
).fetchone()
if row is None:
return None
return row_to_region(row)
def region_wkt(db_path: Path, adcode: str, simplified: bool = True) -> str:
region = get_region(db_path, adcode)
if region is None:
raise ValueError(f"region not found: {adcode}")
return region.simplified_wkt if simplified else region.wkt
def feature_to_row(feature: dict[str, Any], source_path: Path, simplify_tolerance: float) -> tuple[Any, ...]:
properties = feature.get("properties") or {}
geometry = feature.get("geometry")
if geometry is None:
raise ValueError("feature geometry is missing")
geometry_text = json.dumps(geometry, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
geometry_hash = hashlib.sha256(geometry_text.encode("utf-8")).hexdigest()
geom = shape(geometry)
simplified = geom.simplify(simplify_tolerance, preserve_topology=True)
if simplified.is_empty:
simplified = geom
adcode = str(properties.get("adcode"))
name = str(properties.get("name"))
level = str(properties.get("level") or "unknown")
parent_adcode = normalize_adcode((properties.get("parent") or {}).get("adcode"))
province_adcode = derive_province_adcode(adcode, level, parent_adcode, properties.get("acroutes"))
province_name = PROVINCES.get(province_adcode) if province_adcode else None
center = normalize_point(properties.get("center"))
centroid = normalize_point(properties.get("centroid"))
min_lon, min_lat, max_lon, max_lat = geom.bounds
return (
adcode,
name,
level,
parent_adcode,
province_adcode,
province_name,
center[0] if center else None,
center[1] if center else None,
centroid[0] if centroid else None,
centroid[1] if centroid else None,
min_lon,
min_lat,
max_lon,
max_lat,
int(properties["childrenNum"]) if properties.get("childrenNum") is not None else None,
properties.get("treeID"),
str(source_path),
geometry_text,
geom.wkt,
simplified.wkt,
simplify_tolerance,
geometry_hash,
)
def derive_province_adcode(
adcode: str,
level: str,
parent_adcode: str | None,
acroutes: Any,
) -> str | None:
if level == "province" and adcode in PROVINCES:
return adcode
routes = [normalize_adcode(value) for value in acroutes or []]
for route in routes:
if route in PROVINCES:
return route
if parent_adcode in PROVINCES:
return parent_adcode
if adcode.isdigit() and len(adcode) == 6:
candidate = f"{adcode[:2]}0000"
if candidate in PROVINCES:
return candidate
return None
def normalize_adcode(value: Any) -> str | None:
if value is None:
return None
text = str(value)
return text if text else None
def normalize_point(value: Any) -> tuple[float, float] | None:
if not isinstance(value, list) or len(value) < 2:
return None
return (float(value[0]), float(value[1]))
def row_to_region(row) -> RegionRecord:
return RegionRecord(
adcode=str(row["adcode"]),
name=str(row["name"]),
level=str(row["level"]),
parent_adcode=row["parent_adcode"],
parent_name=row["parent_name"],
province_adcode=row["province_adcode"],
province_name=row["province_name"],
center_lon=row["center_lon"],
center_lat=row["center_lat"],
centroid_lon=row["centroid_lon"],
centroid_lat=row["centroid_lat"],
min_lon=row["min_lon"],
min_lat=row["min_lat"],
max_lon=row["max_lon"],
max_lat=row["max_lat"],
children_num=row["children_num"],
tree_id=row["tree_id"],
source_path=row["source_path"],
geometry_json=row["geometry_json"],
wkt=row["wkt"],
simplified_wkt=row["simplified_wkt"],
simplify_tolerance=float(row["simplify_tolerance"]),
geometry_hash=str(row["geometry_hash"]),
)
@@ -0,0 +1,145 @@
from __future__ import annotations
import base64
import ctypes
from dataclasses import dataclass
from pathlib import Path
from sentinel_orbit_downloader.storage.db import connect, get_setting, set_setting
EARTHDATA_USERNAME_KEY = "earthdata.username"
EARTHDATA_PASSWORD_KEY = "earthdata.password_dpapi"
@dataclass(frozen=True)
class EarthdataSettings:
username: str | None
has_password: bool
def get_earthdata_settings(db_path: Path) -> EarthdataSettings:
with connect(db_path) as connection:
username = get_setting(connection, EARTHDATA_USERNAME_KEY)
encrypted_password = get_setting(connection, EARTHDATA_PASSWORD_KEY)
return EarthdataSettings(
username=username,
has_password=bool(encrypted_password),
)
def get_saved_earthdata_credentials(db_path: Path) -> tuple[str | None, str | None]:
with connect(db_path) as connection:
username = get_setting(connection, EARTHDATA_USERNAME_KEY)
encrypted_password = get_setting(connection, EARTHDATA_PASSWORD_KEY)
if not encrypted_password:
return username, None
return username, dpapi_decrypt(encrypted_password)
def save_earthdata_credentials(db_path: Path, username: str, password: str) -> EarthdataSettings:
final_username = username.strip()
if not final_username:
raise ValueError("Earthdata username cannot be empty")
if not password:
raise ValueError("Earthdata password cannot be empty")
encrypted_password = dpapi_encrypt(password)
with connect(db_path) as connection:
set_setting(connection, EARTHDATA_USERNAME_KEY, final_username)
set_setting(connection, EARTHDATA_PASSWORD_KEY, encrypted_password)
return EarthdataSettings(username=final_username, has_password=True)
def clear_earthdata_credentials(db_path: Path) -> None:
with connect(db_path) as connection:
connection.execute(
"DELETE FROM settings WHERE key IN (?, ?)",
(EARTHDATA_USERNAME_KEY, EARTHDATA_PASSWORD_KEY),
)
class DATA_BLOB(ctypes.Structure):
_fields_ = [
("cbData", ctypes.c_uint32),
("pbData", ctypes.POINTER(ctypes.c_ubyte)),
]
try:
kernel32 = ctypes.windll.kernel32
crypt32 = ctypes.windll.crypt32
except AttributeError:
kernel32 = None
crypt32 = None
CRYPTPROTECT_UI_FORBIDDEN = 0x01
def dpapi_encrypt(value: str) -> str:
ensure_windows_dpapi()
raw = value.encode("utf-8")
in_blob = bytes_to_blob(raw)
out_blob = DATA_BLOB()
if not crypt32.CryptProtectData(
ctypes.byref(in_blob),
"sentinel-orbit-downloader",
None,
None,
None,
CRYPTPROTECT_UI_FORBIDDEN,
ctypes.byref(out_blob),
):
raise ctypes.WinError()
try:
encrypted = blob_to_bytes(out_blob)
finally:
free_blob(out_blob)
return base64.b64encode(encrypted).decode("ascii")
def dpapi_decrypt(encoded_value: str) -> str:
ensure_windows_dpapi()
encrypted = base64.b64decode(encoded_value.encode("ascii"))
in_blob = bytes_to_blob(encrypted)
out_blob = DATA_BLOB()
if not crypt32.CryptUnprotectData(
ctypes.byref(in_blob),
None,
None,
None,
None,
CRYPTPROTECT_UI_FORBIDDEN,
ctypes.byref(out_blob),
):
raise ctypes.WinError()
try:
decrypted = blob_to_bytes(out_blob)
finally:
free_blob(out_blob)
return decrypted.decode("utf-8")
def ensure_windows_dpapi() -> None:
if crypt32 is None or kernel32 is None:
raise RuntimeError("Earthdata local credential storage currently requires Windows")
def bytes_to_blob(value: bytes) -> DATA_BLOB:
if not value:
buffer = (ctypes.c_ubyte * 1)()
return DATA_BLOB(0, ctypes.cast(buffer, ctypes.POINTER(ctypes.c_ubyte)))
buffer = (ctypes.c_ubyte * len(value)).from_buffer_copy(value)
return DATA_BLOB(len(value), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_ubyte)))
def blob_to_bytes(blob: DATA_BLOB) -> bytes:
if not blob.cbData or not blob.pbData:
return b""
return bytes(ctypes.string_at(blob.pbData, blob.cbData))
def free_blob(blob: DATA_BLOB) -> None:
if blob.pbData:
kernel32.LocalFree(blob.pbData)
@@ -0,0 +1 @@
"""Storage layer."""
@@ -0,0 +1,193 @@
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA = """
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS regions (
adcode TEXT PRIMARY KEY,
name TEXT NOT NULL,
level TEXT NOT NULL,
parent_adcode TEXT,
province_adcode TEXT,
province_name TEXT,
center_lon REAL,
center_lat REAL,
centroid_lon REAL,
centroid_lat REAL,
min_lon REAL,
min_lat REAL,
max_lon REAL,
max_lat REAL,
children_num INTEGER,
tree_id TEXT,
source_path TEXT,
geometry_json TEXT NOT NULL,
wkt TEXT NOT NULL,
simplified_wkt TEXT NOT NULL,
simplify_tolerance REAL NOT NULL,
geometry_hash TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_regions_name ON regions(name);
CREATE INDEX IF NOT EXISTS idx_regions_level ON regions(level);
CREATE INDEX IF NOT EXISTS idx_regions_parent ON regions(parent_adcode);
CREATE INDEX IF NOT EXISTS idx_regions_province ON regions(province_adcode);
CREATE TABLE IF NOT EXISTS projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
slug TEXT NOT NULL UNIQUE,
data_root TEXT NOT NULL,
description TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS aois (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
name TEXT NOT NULL,
source_type TEXT NOT NULL,
region_adcode TEXT,
wkt TEXT NOT NULL,
bbox_json TEXT,
geometry_json TEXT,
simplification_tolerance REAL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(project_id) REFERENCES projects(id)
);
CREATE TABLE IF NOT EXISTS search_jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
aoi_id INTEGER,
source_type TEXT NOT NULL,
source_value TEXT,
region_adcode TEXT,
region_name TEXT,
start_time TEXT NOT NULL,
end_time TEXT NOT NULL,
processing_level TEXT NOT NULL,
beam_mode TEXT,
flight_direction TEXT,
polarization TEXT,
relative_orbit INTEGER,
max_results INTEGER NOT NULL,
status TEXT NOT NULL,
result_count INTEGER,
data_dir TEXT,
orbit_dir TEXT,
manifest_path TEXT,
error_message TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
finished_at TEXT,
FOREIGN KEY(project_id) REFERENCES projects(id),
FOREIGN KEY(aoi_id) REFERENCES aois(id)
);
CREATE INDEX IF NOT EXISTS idx_search_jobs_project ON search_jobs(project_id);
CREATE INDEX IF NOT EXISTS idx_search_jobs_created ON search_jobs(created_at);
CREATE INDEX IF NOT EXISTS idx_search_jobs_region ON search_jobs(region_adcode);
CREATE TABLE IF NOT EXISTS scenes (
scene_id TEXT PRIMARY KEY,
platform TEXT,
start_time TEXT,
stop_time TEXT,
processing_level TEXT,
beam_mode TEXT,
flight_direction TEXT,
polarization TEXT,
relative_orbit INTEGER,
size_bytes REAL,
size_mb REAL,
download_url TEXT,
geometry_json TEXT,
properties_json TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS job_scenes (
job_id INTEGER NOT NULL,
scene_id TEXT NOT NULL,
selected INTEGER NOT NULL DEFAULT 1,
product_status TEXT NOT NULL DEFAULT 'pending',
orbit_status TEXT NOT NULL DEFAULT 'pending',
product_path TEXT,
orbit_path TEXT,
PRIMARY KEY(job_id, scene_id),
FOREIGN KEY(job_id) REFERENCES search_jobs(id),
FOREIGN KEY(scene_id) REFERENCES scenes(scene_id)
);
CREATE INDEX IF NOT EXISTS idx_job_scenes_scene ON job_scenes(scene_id);
CREATE TABLE IF NOT EXISTS downloads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_id INTEGER NOT NULL,
scene_id TEXT NOT NULL,
kind TEXT NOT NULL,
status TEXT NOT NULL,
url TEXT,
path TEXT,
bytes_total REAL,
bytes_done REAL,
attempt_count INTEGER NOT NULL DEFAULT 1,
error_message TEXT,
started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
finished_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(job_id) REFERENCES search_jobs(id),
FOREIGN KEY(scene_id) REFERENCES scenes(scene_id)
);
CREATE INDEX IF NOT EXISTS idx_downloads_job ON downloads(job_id);
CREATE INDEX IF NOT EXISTS idx_downloads_scene ON downloads(scene_id);
CREATE TABLE IF NOT EXISTS job_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_id INTEGER NOT NULL,
level TEXT NOT NULL,
message TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(job_id) REFERENCES search_jobs(id)
);
CREATE INDEX IF NOT EXISTS idx_job_events_job ON job_events(job_id);
"""
def connect(db_path: Path) -> sqlite3.Connection:
db_path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(str(db_path))
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
return connection
def set_setting(connection: sqlite3.Connection, key: str, value: str) -> None:
connection.execute(
"""
INSERT INTO settings(key, value, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(key) DO UPDATE SET
value = excluded.value,
updated_at = CURRENT_TIMESTAMP
""",
(key, value),
)
def get_setting(connection: sqlite3.Connection, key: str) -> str | None:
row = connection.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
if row is None:
return None
return str(row["value"])
@@ -0,0 +1,573 @@
:root {
color-scheme: light;
--bg: #f3f5f9;
--panel: #ffffff;
--panel-soft: #f8fafc;
--line: #d8dee8;
--text: #15253b;
--muted: #5e6c80;
--accent: #2563eb;
--accent-soft: #e7efff;
--success: #157347;
--success-soft: #dbf5e7;
--warning: #9a6700;
--warning-soft: #fff0cf;
--danger: #b42318;
--danger-soft: #ffe3df;
--shadow: 0 1px 2px rgba(16, 24, 40, 0.05);
}
* {
box-sizing: border-box;
}
html,
body {
margin: 0;
min-height: 100%;
font-family: "Segoe UI", system-ui, sans-serif;
background: var(--bg);
color: var(--text);
}
button,
input,
select,
textarea {
font: inherit;
}
button,
input,
select,
textarea,
.item,
.chip,
.status-pill {
transition: border-color 120ms ease, background 120ms ease, color 120ms ease;
}
.shell {
min-height: 100vh;
display: grid;
grid-template-columns: 320px minmax(0, 1fr);
}
.sidebar {
padding: 20px;
border-right: 1px solid var(--line);
background: var(--panel);
display: flex;
flex-direction: column;
gap: 16px;
}
.brand h1,
.topbar h2,
.panel h2 {
margin: 0;
}
.brand p,
.topbar p {
margin: 4px 0 0;
color: var(--muted);
font-size: 13px;
}
.content {
padding: 20px;
display: flex;
flex-direction: column;
gap: 16px;
min-width: 0;
}
.topbar {
display: flex;
justify-content: space-between;
align-items: flex-start;
gap: 16px;
}
.message {
min-height: 38px;
display: inline-flex;
align-items: center;
padding: 8px 12px;
border: 1px solid var(--line);
border-radius: 6px;
background: var(--panel);
box-shadow: var(--shadow);
font-size: 13px;
line-height: 1.4;
}
.message.info {
color: var(--text);
}
.message.success {
border-color: #b8e3c9;
background: var(--success-soft);
color: var(--success);
}
.message.warning {
border-color: #f0d48a;
background: var(--warning-soft);
color: var(--warning);
}
.message.error {
border-color: #f3b8b1;
background: var(--danger-soft);
color: var(--danger);
}
.panel {
background: var(--panel);
border: 1px solid var(--line);
border-radius: 8px;
box-shadow: var(--shadow);
padding: 16px;
min-width: 0;
}
.section-head {
display: flex;
align-items: center;
justify-content: space-between;
gap: 12px;
}
.section-head h2 {
font-size: 15px;
}
.stack {
display: flex;
flex-direction: column;
gap: 12px;
}
.layout.two-col {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 16px;
}
.grid {
display: grid;
gap: 12px;
}
.grid.two {
grid-template-columns: repeat(2, minmax(0, 1fr));
}
.grid.three {
grid-template-columns: repeat(3, minmax(0, 1fr));
}
.field {
display: flex;
flex-direction: column;
gap: 6px;
min-width: 0;
}
.field label,
.field span {
font-size: 12px;
color: var(--muted);
}
.field input,
.field select,
.field textarea {
width: 100%;
min-height: 38px;
padding: 8px 10px;
border: 1px solid var(--line);
border-radius: 6px;
background: #fff;
color: var(--text);
}
.field textarea {
min-height: 84px;
resize: vertical;
}
.field input:focus,
.field select:focus,
.field textarea:focus {
outline: none;
border-color: #94b5ff;
box-shadow: 0 0 0 3px rgba(37, 99, 235, 0.12);
}
.inline-field input {
max-width: 160px;
}
.toolbar {
display: flex;
flex-wrap: wrap;
gap: 8px;
align-items: center;
}
button {
min-height: 38px;
padding: 8px 12px;
border-radius: 6px;
border: 1px solid var(--line);
background: #fff;
color: var(--text);
cursor: pointer;
}
button:hover {
border-color: #bcc8d9;
}
button.primary {
background: var(--accent);
color: #fff;
border-color: var(--accent);
}
button.primary:hover {
background: #1d4ed8;
border-color: #1d4ed8;
}
button.ghost {
background: var(--panel-soft);
}
button.danger {
border-color: #f0c3bc;
background: #fff6f4;
color: var(--danger);
}
button.danger:hover {
border-color: #e5a59c;
background: #ffeceb;
}
button:disabled {
opacity: 0.55;
cursor: not-allowed;
}
.status-block {
display: flex;
align-items: center;
gap: 10px;
}
.status-dot {
width: 10px;
height: 10px;
border-radius: 999px;
background: #98a2b3;
flex: none;
}
.status-dot.ok {
background: var(--success);
}
.status-dot.error {
background: var(--danger);
}
.status-title {
font-size: 13px;
font-weight: 600;
}
.status-subtle {
margin-top: 2px;
color: var(--muted);
font-size: 12px;
}
.mono {
font-family: ui-monospace, "SFMono-Regular", Consolas, monospace;
}
.list {
display: flex;
flex-direction: column;
gap: 8px;
min-height: 0;
}
.list.tall {
max-height: 320px;
overflow: auto;
}
.item {
width: 100%;
border: 1px solid var(--line);
border-radius: 6px;
background: #fff;
text-align: left;
padding: 10px 12px;
display: flex;
flex-direction: column;
gap: 4px;
}
.item:hover {
border-color: #c1d2fa;
}
.item.active {
border-color: var(--accent);
background: #f5f8ff;
}
.item-title {
font-size: 13px;
font-weight: 600;
line-height: 1.45;
}
.item-subtle {
font-size: 12px;
color: var(--muted);
line-height: 1.45;
}
.detail-card,
.events {
border: 1px solid var(--line);
border-radius: 6px;
background: var(--panel-soft);
padding: 12px;
}
.detail-card {
min-height: 150px;
}
.detail-grid {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 10px;
}
.detail-row {
display: flex;
flex-direction: column;
gap: 3px;
}
.detail-label {
font-size: 12px;
color: var(--muted);
}
.detail-value {
font-size: 13px;
line-height: 1.4;
}
.chip-row {
display: flex;
flex-wrap: wrap;
gap: 8px;
}
.chip {
display: inline-flex;
align-items: center;
min-height: 32px;
padding: 6px 10px;
border-radius: 999px;
background: var(--panel-soft);
border: 1px solid var(--line);
font-size: 12px;
line-height: 1.4;
}
.status-pill {
display: inline-flex;
align-items: center;
min-height: 24px;
padding: 2px 8px;
border-radius: 999px;
font-size: 12px;
line-height: 1.4;
background: var(--panel-soft);
color: var(--muted);
}
.status-pill.pending,
.status-pill.queued,
.status-pill.skipped {
background: #edf1f6;
color: var(--muted);
}
.status-pill.running {
background: var(--accent-soft);
color: var(--accent);
}
.status-pill.succeeded {
background: var(--success-soft);
color: var(--success);
}
.status-pill.partial_failed {
background: var(--warning-soft);
color: var(--warning);
}
.status-pill.failed,
.status-pill.error {
background: var(--danger-soft);
color: var(--danger);
}
.toggle-row {
display: flex;
flex-wrap: wrap;
gap: 12px;
}
.toggle {
display: inline-flex;
align-items: center;
gap: 8px;
min-height: 38px;
padding: 0;
color: var(--text);
}
.toggle input {
width: 16px;
height: 16px;
}
.note {
font-size: 12px;
line-height: 1.5;
color: var(--muted);
background: var(--panel-soft);
border: 1px solid var(--line);
border-radius: 6px;
padding: 10px 12px;
}
.detail-layout {
display: grid;
grid-template-columns: minmax(0, 1.25fr) minmax(260px, 0.75fr);
gap: 16px;
}
.table-wrap {
overflow: auto;
border: 1px solid var(--line);
border-radius: 6px;
background: #fff;
}
table {
width: 100%;
border-collapse: collapse;
table-layout: fixed;
font-size: 12px;
}
thead {
background: #fafbfd;
}
th,
td {
border-top: 1px solid var(--line);
text-align: left;
vertical-align: top;
padding: 9px 8px;
word-break: break-word;
}
thead th {
border-top: 0;
color: var(--muted);
font-weight: 600;
}
.events {
min-height: 280px;
max-height: 420px;
overflow: auto;
}
.event {
padding: 8px 0;
border-top: 1px solid #e8edf4;
}
.event:first-child {
border-top: 0;
padding-top: 0;
}
.event-title {
font-size: 12px;
font-weight: 600;
}
.event-meta {
margin-top: 2px;
font-size: 12px;
color: var(--muted);
}
.event-message {
margin-top: 4px;
font-size: 12px;
line-height: 1.5;
}
.empty {
color: var(--muted);
font-size: 12px;
padding: 8px 0;
}
@media (max-width: 1200px) {
.shell {
grid-template-columns: 1fr;
}
.sidebar {
border-right: 0;
border-bottom: 1px solid var(--line);
}
}
@media (max-width: 900px) {
.layout.two-col,
.grid.two,
.grid.three,
.detail-layout,
.detail-grid {
grid-template-columns: 1fr;
}
.topbar {
flex-direction: column;
align-items: stretch;
}
}
@@ -0,0 +1,864 @@
const DEFAULT_CATALOG_PATH = "\u5168\u56fd\u884c\u653f\u533a.geojson";
const FINAL_STATUSES = new Set(["succeeded", "failed", "partial_failed", "canceled"]);
const api = {
async request(path, options = {}) {
const init = { ...options };
init.headers = {
...(options.body ? { "Content-Type": "application/json" } : {}),
...(options.headers || {}),
};
const response = await fetch(path, init);
if (!response.ok) {
let detail = response.statusText;
try {
const payload = await response.json();
detail = payload.detail || JSON.stringify(payload);
} catch (_) {}
throw new Error(detail);
}
if (response.status === 204) {
return null;
}
return response.json();
},
health() {
return this.request("/api/health");
},
earthdataSettings() {
return this.request("/api/settings/earthdata");
},
saveEarthdataSettings(payload) {
return this.request("/api/settings/earthdata", {
method: "PUT",
body: JSON.stringify(payload),
});
},
clearEarthdataSettings() {
return this.request("/api/settings/earthdata", {
method: "DELETE",
});
},
importRegions(payload) {
return this.request("/api/regions/import", {
method: "POST",
body: JSON.stringify(payload),
});
},
searchRegions(query) {
return this.request(`/api/regions/search?q=${encodeURIComponent(query)}`);
},
regionDetail(adcode) {
return this.request(`/api/regions/${encodeURIComponent(adcode)}`);
},
projects() {
return this.request("/api/projects");
},
createProject(payload) {
return this.request("/api/projects", {
method: "POST",
body: JSON.stringify(payload),
});
},
jobs(projectId) {
const suffix = projectId ? `?project_id=${projectId}&limit=50` : "?limit=50";
return this.request(`/api/jobs${suffix}`);
},
job(jobId) {
return this.request(`/api/jobs/${jobId}?include_events=true`);
},
createSearchJob(payload) {
return this.request("/api/search-jobs", {
method: "POST",
body: JSON.stringify(payload),
});
},
createDownloadJob(payload) {
return this.request("/api/download-jobs", {
method: "POST",
body: JSON.stringify(payload),
});
},
deleteJob(jobId, force = false) {
return this.request(`/api/jobs/${jobId}?force=${force ? "true" : "false"}`, {
method: "DELETE",
});
},
clearJobs(projectId, force = false) {
const projectPart = projectId ? `project_id=${projectId}&` : "";
return this.request(`/api/jobs?${projectPart}force=${force ? "true" : "false"}`, {
method: "DELETE",
});
},
};
const state = {
projects: [],
selectedProjectId: null,
regions: [],
selectedRegion: null,
selectedRegionDetail: null,
earthdataSettings: null,
jobs: [],
selectedJobId: null,
selectedJobDetail: null,
pollTimer: null,
};
const el = {
healthDot: document.getElementById("healthDot"),
healthLabel: document.getElementById("healthLabel"),
healthDetail: document.getElementById("healthDetail"),
catalogPath: document.getElementById("catalogPath"),
simplifyTolerance: document.getElementById("simplifyTolerance"),
projectName: document.getElementById("projectName"),
projectPath: document.getElementById("projectPath"),
projectDesc: document.getElementById("projectDesc"),
projectList: document.getElementById("projectList"),
messageBar: document.getElementById("messageBar"),
regionKeyword: document.getElementById("regionKeyword"),
regionList: document.getElementById("regionList"),
regionDetailCard: document.getElementById("regionDetailCard"),
startDate: document.getElementById("startDate"),
endDate: document.getElementById("endDate"),
maxResults: document.getElementById("maxResults"),
processingLevel: document.getElementById("processingLevel"),
beamMode: document.getElementById("beamMode"),
flightDirection: document.getElementById("flightDirection"),
polarization: document.getElementById("polarization"),
relativeOrbit: document.getElementById("relativeOrbit"),
manifestPath: document.getElementById("manifestPath"),
rawRegion: document.getElementById("rawRegion"),
selectionSummary: document.getElementById("selectionSummary"),
dataDir: document.getElementById("dataDir"),
orbitDir: document.getElementById("orbitDir"),
skipData: document.getElementById("skipData"),
skipOrbits: document.getElementById("skipOrbits"),
downloadProcesses: document.getElementById("downloadProcesses"),
ignoreOrbitErrors: document.getElementById("ignoreOrbitErrors"),
earthdataUser: document.getElementById("earthdataUser"),
earthdataPass: document.getElementById("earthdataPass"),
earthdataSummary: document.getElementById("earthdataSummary"),
jobFilterSummary: document.getElementById("jobFilterSummary"),
jobList: document.getElementById("jobList"),
deleteJobBtn: document.getElementById("deleteJobBtn"),
clearJobsBtn: document.getElementById("clearJobsBtn"),
jobMeta: document.getElementById("jobMeta"),
sceneTableBody: document.getElementById("sceneTableBody"),
eventList: document.getElementById("eventList"),
};
function setMessage(text, tone = "info") {
el.messageBar.textContent = text;
el.messageBar.className = `message ${tone}`;
}
function setHealth(ok, label, detail) {
el.healthDot.className = `status-dot ${ok ? "ok" : "error"}`;
el.healthLabel.textContent = label;
el.healthDetail.textContent = detail;
}
function makeChip(text) {
const span = document.createElement("span");
span.className = "chip";
span.textContent = text;
return span;
}
function makeStatusPill(value) {
const span = document.createElement("span");
span.className = `status-pill ${value || ""}`.trim();
span.textContent = value || "-";
return span;
}
function renderEmpty(container, text) {
container.innerHTML = `<div class="empty">${escapeHtml(text)}</div>`;
}
function formatMb(value) {
if (value == null || Number.isNaN(Number(value))) {
return "";
}
return `${Number(value).toFixed(1)} MB`;
}
function formatRelativeOrbit(value) {
return value == null ? "" : String(value);
}
function formatBBox(bbox) {
if (!Array.isArray(bbox) || bbox.length !== 4) {
return "-";
}
return bbox.map((value) => Number(value).toFixed(4)).join(", ");
}
function formatPoint(point) {
if (!Array.isArray(point) || point.length < 2) {
return "-";
}
return point.slice(0, 2).map((value) => Number(value).toFixed(4)).join(", ");
}
function emptyToNull(value) {
const text = (value || "").trim();
return text ? text : null;
}
function numberOrNull(value) {
const text = (value || "").trim();
if (!text) {
return null;
}
const number = Number(text);
return Number.isFinite(number) ? number : null;
}
function escapeHtml(text) {
const div = document.createElement("div");
div.textContent = text || "";
return div.innerHTML;
}
function statusCounts(scenes) {
const product = {};
const orbit = {};
for (const scene of scenes) {
product[scene.product_status] = (product[scene.product_status] || 0) + 1;
orbit[scene.orbit_status] = (orbit[scene.orbit_status] || 0) + 1;
}
return { product, orbit };
}
function currentProject() {
return state.projects.find((project) => project.id === state.selectedProjectId) || null;
}
function autoFillProjectPaths(project) {
if (!project || !project.data_root) {
return;
}
el.projectPath.value = project.data_root;
el.dataDir.value = `${project.data_root}\\data`;
el.orbitDir.value = `${project.data_root}\\orbits`;
if (!el.manifestPath.value) {
el.manifestPath.value = `${project.data_root}\\manifest.csv`;
}
}
function renderProjects() {
el.projectList.innerHTML = "";
if (!state.projects.length) {
renderEmpty(el.projectList, "暂无项目");
return;
}
for (const project of state.projects) {
const button = document.createElement("button");
button.type = "button";
button.className = `item ${state.selectedProjectId === project.id ? "active" : ""}`;
button.innerHTML = `
<div class="item-title">${escapeHtml(project.name)}</div>
<div class="item-subtle mono">${escapeHtml(project.data_root)}</div>
<div class="item-subtle">${escapeHtml(project.slug)}</div>
`;
button.addEventListener("click", async () => {
state.selectedProjectId = project.id;
autoFillProjectPaths(project);
renderProjects();
updateSelectionSummary();
await loadJobs();
await loadSelectedJobDetail();
});
el.projectList.appendChild(button);
}
}
function renderRegions() {
el.regionList.innerHTML = "";
if (!state.regions.length) {
renderEmpty(el.regionList, "暂无行政区结果");
return;
}
for (const region of state.regions) {
const button = document.createElement("button");
button.type = "button";
button.className = `item ${state.selectedRegion && state.selectedRegion.adcode === region.adcode ? "active" : ""}`;
button.innerHTML = `
<div class="item-title">${escapeHtml(region.label)}</div>
<div class="item-subtle mono">${escapeHtml(region.adcode)}</div>
<div class="item-subtle">${escapeHtml(region.level)} | ${escapeHtml(formatBBox(region.bbox))}</div>
`;
button.addEventListener("click", async () => {
state.selectedRegion = region;
renderRegions();
updateSelectionSummary();
await loadRegionDetail(region.adcode);
});
el.regionList.appendChild(button);
}
}
function renderRegionDetail() {
const region = state.selectedRegionDetail;
if (!region) {
el.regionDetailCard.innerHTML = '<div class="empty">未选择行政区</div>';
return;
}
el.regionDetailCard.innerHTML = `
<div class="stack">
<div class="chip-row">
<span class="status-pill succeeded">${escapeHtml(region.level)}</span>
<span class="chip mono">${escapeHtml(region.adcode)}</span>
</div>
<div class="detail-grid">
<div class="detail-row">
<div class="detail-label">行政区</div>
<div class="detail-value">${escapeHtml(region.label)}</div>
</div>
<div class="detail-row">
<div class="detail-label">简化容差</div>
<div class="detail-value">${escapeHtml(String(region.simplify_tolerance))}</div>
</div>
<div class="detail-row">
<div class="detail-label">范围框</div>
<div class="detail-value mono">${escapeHtml(formatBBox(region.bbox))}</div>
</div>
<div class="detail-row">
<div class="detail-label">中心点</div>
<div class="detail-value mono">${escapeHtml(formatPoint(region.center))}</div>
</div>
</div>
</div>
`;
}
function renderEarthdataSummary() {
el.earthdataSummary.innerHTML = "";
const settings = state.earthdataSettings;
if (!settings || (!settings.username && !settings.has_password)) {
el.earthdataSummary.appendChild(makeChip("Earthdata 本地凭据:未保存"));
return;
}
if (settings.username) {
el.earthdataSummary.appendChild(makeChip(`Earthdata 用户:${settings.username}`));
}
el.earthdataSummary.appendChild(
makeChip(settings.has_password ? "Earthdata 密码:已保存" : "Earthdata 密码:未保存"),
);
}
function renderJobFilterSummary() {
el.jobFilterSummary.innerHTML = "";
const project = currentProject();
if (project) {
el.jobFilterSummary.appendChild(makeChip(`项目 ${project.name}`));
} else {
el.jobFilterSummary.appendChild(makeChip("全部项目"));
}
el.jobFilterSummary.appendChild(makeChip(`任务数 ${state.jobs.length}`));
}
function renderJobs() {
el.jobList.innerHTML = "";
if (!state.jobs.length) {
renderEmpty(el.jobList, "暂无任务");
renderJobFilterSummary();
return;
}
for (const job of state.jobs) {
const button = document.createElement("button");
button.type = "button";
button.className = `item ${state.selectedJobId === job.id ? "active" : ""}`;
button.appendChild(makeStatusPill(job.status));
const title = document.createElement("div");
title.className = "item-title";
title.textContent = `#${job.id} ${job.region_name || job.source_value || job.source_type}`;
button.appendChild(title);
const subtles = [
`${job.start_time} - ${job.end_time}`,
`${job.processing_level}${job.beam_mode ? ` | ${job.beam_mode}` : ""}`,
`结果 ${job.result_count ?? "-"}`,
];
for (const text of subtles) {
const div = document.createElement("div");
div.className = "item-subtle";
div.textContent = text;
button.appendChild(div);
}
button.addEventListener("click", async () => {
state.selectedJobId = job.id;
renderJobs();
updateSelectionSummary();
await loadSelectedJobDetail();
});
el.jobList.appendChild(button);
}
renderJobFilterSummary();
}
function renderJobDetail() {
const payload = state.selectedJobDetail;
el.jobMeta.innerHTML = "";
el.sceneTableBody.innerHTML = "";
el.eventList.innerHTML = "";
if (!payload) {
renderEmpty(el.jobMeta, "未选择任务");
el.sceneTableBody.innerHTML = '<tr><td colspan="7" class="empty">暂无场景</td></tr>';
renderEmpty(el.eventList, "暂无事件");
return;
}
const { job, scenes, events } = payload;
const counts = statusCounts(scenes);
const chips = [
`Job #${job.id}`,
`状态 ${job.status}`,
`时间 ${job.start_time} - ${job.end_time}`,
`产品 ${job.processing_level}`,
`结果 ${job.result_count ?? scenes.length}`,
];
if (job.region_name) {
chips.push(`区域 ${job.region_name}`);
}
if (job.data_dir) {
chips.push(`数据目录 ${job.data_dir}`);
}
if (job.orbit_dir) {
chips.push(`轨道目录 ${job.orbit_dir}`);
}
if (job.error_message) {
chips.push(`错误 ${job.error_message}`);
}
for (const text of chips) {
el.jobMeta.appendChild(makeChip(text));
}
for (const [status, count] of Object.entries(counts.product)) {
el.jobMeta.appendChild(makeChip(`产品 ${status} ${count}`));
}
for (const [status, count] of Object.entries(counts.orbit)) {
el.jobMeta.appendChild(makeChip(`轨道 ${status} ${count}`));
}
if (!scenes.length) {
el.sceneTableBody.innerHTML = '<tr><td colspan="7" class="empty">暂无场景</td></tr>';
} else {
for (const scene of scenes) {
const row = document.createElement("tr");
const sceneCell = document.createElement("td");
sceneCell.className = "mono";
sceneCell.textContent = scene.scene_id;
row.appendChild(sceneCell);
const startCell = document.createElement("td");
startCell.textContent = scene.start_time || "";
row.appendChild(startCell);
const levelCell = document.createElement("td");
levelCell.textContent = scene.processing_level || "";
row.appendChild(levelCell);
const orbitCell = document.createElement("td");
orbitCell.textContent = formatRelativeOrbit(scene.relative_orbit);
row.appendChild(orbitCell);
const productCell = document.createElement("td");
productCell.appendChild(makeStatusPill(scene.product_status));
row.appendChild(productCell);
const eofCell = document.createElement("td");
eofCell.appendChild(makeStatusPill(scene.orbit_status));
row.appendChild(eofCell);
const sizeCell = document.createElement("td");
sizeCell.textContent = formatMb(scene.size_mb);
row.appendChild(sizeCell);
el.sceneTableBody.appendChild(row);
}
}
if (!events.length) {
renderEmpty(el.eventList, "暂无事件");
} else {
for (const event of events) {
const div = document.createElement("div");
div.className = "event";
div.innerHTML = `
<div class="event-title">${escapeHtml(event.level)}</div>
<div class="event-meta">${escapeHtml(event.created_at)}</div>
<div class="event-message">${escapeHtml(event.message)}</div>
`;
el.eventList.appendChild(div);
}
}
syncPolling();
}
function updateSelectionSummary() {
el.selectionSummary.innerHTML = "";
const project = currentProject();
if (project) {
el.selectionSummary.appendChild(makeChip(`项目 ${project.name}`));
}
if (state.selectedRegion) {
el.selectionSummary.appendChild(makeChip(`行政区 ${state.selectedRegion.label}`));
}
el.selectionSummary.appendChild(makeChip(`时间 ${el.startDate.value} - ${el.endDate.value}`));
el.selectionSummary.appendChild(makeChip(`产品 ${el.processingLevel.value}`));
if (el.beamMode.value) {
el.selectionSummary.appendChild(makeChip(`波束 ${el.beamMode.value}`));
}
if (state.selectedJobId) {
el.selectionSummary.appendChild(makeChip(`当前任务 #${state.selectedJobId}`));
}
}
async function loadHealth() {
try {
const health = await api.health();
setHealth(true, "服务已连接", health.db_path);
} catch (error) {
setHealth(false, "服务不可用", error.message || String(error));
throw error;
}
}
async function loadEarthdataSettings() {
state.earthdataSettings = await api.earthdataSettings();
if (state.earthdataSettings && state.earthdataSettings.username && !el.earthdataUser.value) {
el.earthdataUser.value = state.earthdataSettings.username;
}
renderEarthdataSummary();
}
async function loadProjects() {
state.projects = await api.projects();
if (!state.selectedProjectId && state.projects.length) {
state.selectedProjectId = state.projects[0].id;
autoFillProjectPaths(state.projects[0]);
} else {
const project = currentProject();
if (project) {
autoFillProjectPaths(project);
}
}
renderProjects();
updateSelectionSummary();
}
async function loadRegions() {
const keyword = el.regionKeyword.value.trim();
if (!keyword) {
state.regions = [];
state.selectedRegion = null;
state.selectedRegionDetail = null;
renderRegions();
renderRegionDetail();
updateSelectionSummary();
return;
}
state.regions = await api.searchRegions(keyword);
state.selectedRegion = state.regions[0] || null;
renderRegions();
if (state.selectedRegion) {
await loadRegionDetail(state.selectedRegion.adcode);
} else {
state.selectedRegionDetail = null;
renderRegionDetail();
}
updateSelectionSummary();
}
async function loadRegionDetail(adcode) {
state.selectedRegionDetail = await api.regionDetail(adcode);
renderRegionDetail();
}
async function loadJobs() {
state.jobs = await api.jobs(state.selectedProjectId);
if (state.selectedJobId && !state.jobs.some((job) => job.id === state.selectedJobId)) {
state.selectedJobId = null;
state.selectedJobDetail = null;
}
renderJobs();
updateSelectionSummary();
}
async function loadSelectedJobDetail() {
if (!state.selectedJobId) {
state.selectedJobDetail = null;
renderJobDetail();
return;
}
state.selectedJobDetail = await api.job(state.selectedJobId);
const updatedJob = state.selectedJobDetail.job;
state.jobs = state.jobs.map((job) => (job.id === updatedJob.id ? updatedJob : job));
renderJobs();
renderJobDetail();
updateSelectionSummary();
}
async function importCatalog() {
const payload = {
source_path: el.catalogPath.value.trim() || DEFAULT_CATALOG_PATH,
simplify_tolerance: Number(el.simplifyTolerance.value || 0.01),
};
const result = await api.importRegions(payload);
setMessage(`行政区已导入:${result.imported}/${result.total_features}`, "success");
await loadRegions();
}
async function saveEarthdataSettings() {
const username = el.earthdataUser.value.trim();
const password = el.earthdataPass.value;
if (!username || !password) {
throw new Error("保存 Earthdata 凭据时必须填写用户名和密码");
}
const response = await api.saveEarthdataSettings({ username, password });
state.earthdataSettings = response;
el.earthdataPass.value = "";
renderEarthdataSummary();
setMessage("Earthdata 凭据已保存到本地配置", "success");
}
async function clearEarthdataSettings() {
if (!window.confirm("确定清除本地保存的 Earthdata 用户名和密码吗?")) {
return;
}
await api.clearEarthdataSettings();
state.earthdataSettings = { username: null, has_password: false };
el.earthdataUser.value = "";
el.earthdataPass.value = "";
renderEarthdataSummary();
setMessage("Earthdata 本地凭据已清除", "success");
}
async function createProject() {
const payload = {
name: el.projectName.value.trim(),
data_root: el.projectPath.value.trim(),
description: emptyToNull(el.projectDesc.value),
};
if (!payload.name || !payload.data_root) {
throw new Error("项目名称和数据根目录不能为空");
}
const project = await api.createProject(payload);
state.selectedProjectId = project.id;
setMessage(`项目已创建:${project.name}`, "success");
await loadProjects();
await loadJobs();
}
async function createSearchJob() {
if (!state.selectedProjectId) {
throw new Error("请先选择项目");
}
if (!state.selectedRegion) {
throw new Error("请先选择行政区");
}
const payload = {
project_id: state.selectedProjectId,
region_adcode: state.selectedRegion.adcode,
start: el.startDate.value,
end: el.endDate.value,
processing_level: el.processingLevel.value,
beam_mode: emptyToNull(el.beamMode.value),
flight_direction: emptyToNull(el.flightDirection.value),
polarization: emptyToNull(el.polarization.value),
relative_orbit: numberOrNull(el.relativeOrbit.value),
max_results: Number(el.maxResults.value || 20),
manifest_path: emptyToNull(el.manifestPath.value),
raw_region: el.rawRegion.checked,
};
const response = await api.createSearchJob(payload);
state.selectedJobId = response.job.id;
state.selectedJobDetail = {
job: response.job,
scenes: response.scenes,
events: [],
};
setMessage(`检索完成:job #${response.job.id},场景 ${response.scenes.length}`, "success");
await loadJobs();
await loadSelectedJobDetail();
}
async function queueDownloadJob() {
if (!state.selectedJobId) {
throw new Error("请先选择任务");
}
const payload = {
job_id: state.selectedJobId,
data_dir: emptyToNull(el.dataDir.value),
orbit_dir: emptyToNull(el.orbitDir.value),
skip_data: el.skipData.value === "true",
skip_orbits: el.skipOrbits.value === "true",
processes: Number(el.downloadProcesses.value || 2),
ignore_orbit_errors: el.ignoreOrbitErrors.checked,
username: emptyToNull(el.earthdataUser.value),
password: emptyToNull(el.earthdataPass.value),
};
const response = await api.createDownloadJob(payload);
setMessage(`下载任务已排队:job #${response.job.id}`, "success");
await loadSelectedJobDetail();
}
async function deleteSelectedJob() {
if (!state.selectedJobId) {
throw new Error("请先选择任务");
}
const selectedJob =
(state.selectedJobDetail && state.selectedJobDetail.job) ||
state.jobs.find((job) => job.id === state.selectedJobId) ||
null;
const isRunning = Boolean(selectedJob && selectedJob.status === "running");
const message = isRunning
? "当前任务仍显示为运行中。强制删除只会清理记录,不能保证后台下载立刻停止。确定继续删除吗?"
: `确定删除任务 #${state.selectedJobId} 吗?`;
if (!window.confirm(message)) {
return;
}
const deletedJobId = state.selectedJobId;
const response = await api.deleteJob(deletedJobId, isRunning);
state.selectedJobId = null;
state.selectedJobDetail = null;
state.jobs = state.jobs.filter((job) => job.id !== deletedJobId);
renderJobs();
renderJobDetail();
updateSelectionSummary();
setMessage(`已删除 ${response.deleted_count} 个任务`, "success");
}
async function clearCurrentJobs() {
if (!state.selectedProjectId) {
throw new Error("请先选择项目");
}
const project = currentProject();
const projectName = project ? project.name : `#${state.selectedProjectId}`;
if (!window.confirm(`确定清空项目“${projectName}”下的非运行任务吗?`)) {
return;
}
const response = await api.clearJobs(state.selectedProjectId, false);
const deletedIds = new Set(response.deleted_job_ids || []);
if (state.selectedJobId && deletedIds.has(state.selectedJobId)) {
state.selectedJobId = null;
state.selectedJobDetail = null;
}
state.jobs = state.jobs.filter((job) => !deletedIds.has(job.id));
renderJobs();
renderJobDetail();
updateSelectionSummary();
setMessage(`已清空 ${response.deleted_count} 个任务`, "success");
}
function stopPolling() {
if (state.pollTimer) {
window.clearInterval(state.pollTimer);
state.pollTimer = null;
}
}
function syncPolling() {
const activeJob = state.selectedJobDetail && state.selectedJobDetail.job;
if (!activeJob || FINAL_STATUSES.has(activeJob.status)) {
stopPolling();
return;
}
if (state.pollTimer) {
return;
}
state.pollTimer = window.setInterval(async () => {
if (!state.selectedJobId) {
stopPolling();
return;
}
try {
await loadSelectedJobDetail();
} catch (_) {}
}, 5000);
}
function runAction(fn, pendingText) {
return async () => {
try {
if (pendingText) {
setMessage(pendingText, "info");
}
await fn();
} catch (error) {
setMessage(error.message || String(error), "error");
}
};
}
function bindEvents() {
document.getElementById("saveEarthdataBtn").addEventListener("click", runAction(saveEarthdataSettings, "正在保存 Earthdata 凭据"));
document.getElementById("clearEarthdataBtn").addEventListener("click", runAction(clearEarthdataSettings, "正在清除 Earthdata 凭据"));
document.getElementById("importCatalogBtn").addEventListener("click", runAction(importCatalog, "正在导入行政区"));
document.getElementById("createProjectBtn").addEventListener("click", runAction(createProject, "正在创建项目"));
document.getElementById("refreshProjectsBtn").addEventListener("click", runAction(loadProjects, "正在刷新项目"));
document.getElementById("searchRegionsBtn").addEventListener("click", runAction(loadRegions, "正在查询行政区"));
document.getElementById("clearRegionsBtn").addEventListener("click", () => {
state.regions = [];
state.selectedRegion = null;
state.selectedRegionDetail = null;
el.regionKeyword.value = "";
renderRegions();
renderRegionDetail();
updateSelectionSummary();
setMessage("行政区查询已清空", "info");
});
document.getElementById("runSearchBtn").addEventListener("click", runAction(createSearchJob, "正在检索 Sentinel-1"));
document.getElementById("queueDownloadBtn").addEventListener("click", runAction(queueDownloadJob, "正在排队下载任务"));
document.getElementById("deleteJobBtn").addEventListener("click", runAction(deleteSelectedJob, "正在删除任务"));
document.getElementById("clearJobsBtn").addEventListener("click", runAction(clearCurrentJobs, "正在清空任务"));
document.getElementById("refreshJobsBtn").addEventListener("click", runAction(async () => {
await loadJobs();
await loadSelectedJobDetail();
}, "正在刷新任务"));
document.getElementById("refreshDetailBtn").addEventListener("click", runAction(loadSelectedJobDetail, "正在刷新任务详情"));
}
async function bootstrap() {
bindEvents();
if (!el.catalogPath.value) {
el.catalogPath.value = DEFAULT_CATALOG_PATH;
}
el.healthDetail.textContent = window.location.origin;
setMessage("正在初始化工作台", "info");
await loadHealth();
await loadEarthdataSettings();
await loadProjects();
await loadRegions();
await loadJobs();
await loadSelectedJobDetail();
setMessage("工作台已就绪", "success");
}
bootstrap().catch((error) => {
setMessage(error.message || String(error), "error");
});
@@ -0,0 +1,271 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Sentinel Orbit Downloader</title>
<link rel="stylesheet" href="/static/app.css">
</head>
<body>
<div class="shell">
<aside class="sidebar">
<div class="brand">
<h1>Sentinel Orbit Downloader</h1>
<p>本地查询、下载与精密轨道管理</p>
</div>
<section class="panel stack">
<div class="section-head">
<h2>环境配置</h2>
</div>
<div class="status-block">
<span class="status-dot" id="healthDot"></span>
<div>
<div class="status-title" id="healthLabel">正在连接服务</div>
<div class="status-subtle mono" id="healthDetail">http://127.0.0.1:8011</div>
</div>
</div>
<div class="field">
<label for="catalogPath">行政区 GeoJSON</label>
<input id="catalogPath" value="">
</div>
<div class="field inline-field">
<label for="simplifyTolerance">简化容差</label>
<input id="simplifyTolerance" type="number" min="0.0001" step="0.001" value="0.01">
</div>
<div class="toolbar">
<button class="primary" id="importCatalogBtn" type="button">导入行政区</button>
</div>
</section>
<section class="panel stack">
<div class="section-head">
<h2>项目</h2>
<button class="ghost" id="refreshProjectsBtn" type="button">刷新</button>
</div>
<div class="field">
<label for="projectName">项目名称</label>
<input id="projectName" value="牡丹江 Sentinel-1 数据">
</div>
<div class="field">
<label for="projectPath">数据根目录</label>
<input id="projectPath" value=".local-data\\mudanjiang">
</div>
<div class="field">
<label for="projectDesc">说明</label>
<textarea id="projectDesc">牡丹江 Sentinel-1 场景、精密轨道与任务记录。</textarea>
</div>
<div class="toolbar">
<button class="primary" id="createProjectBtn" type="button">新建项目</button>
</div>
<div id="projectList" class="list"></div>
</section>
</aside>
<main class="content">
<header class="topbar">
<div>
<h2>任务工作台</h2>
<p>按行政区和时间检索 Sentinel-1,并统一管理数据与精密轨道下载。</p>
</div>
<div class="message" id="messageBar">等待操作</div>
</header>
<section class="layout two-col">
<section class="panel stack">
<div class="section-head">
<h2>行政区选择</h2>
<div class="toolbar">
<button class="ghost" id="searchRegionsBtn" type="button">查询</button>
<button class="ghost" id="clearRegionsBtn" type="button">清空</button>
</div>
</div>
<div class="field">
<label for="regionKeyword">名称或 adcode</label>
<input id="regionKeyword" value="牡丹江">
</div>
<div id="regionList" class="list tall"></div>
<div class="detail-card" id="regionDetailCard">
<div class="empty">未选择行政区</div>
</div>
</section>
<section class="panel stack">
<div class="section-head">
<h2>检索参数</h2>
<button class="primary" id="runSearchBtn" type="button">发起检索</button>
</div>
<div class="grid three">
<div class="field">
<label for="startDate">开始日期</label>
<input id="startDate" type="date" value="2024-01-01">
</div>
<div class="field">
<label for="endDate">结束日期</label>
<input id="endDate" type="date" value="2024-03-01">
</div>
<div class="field">
<label for="maxResults">最大结果数</label>
<input id="maxResults" type="number" min="1" value="20">
</div>
</div>
<div class="grid three">
<div class="field">
<label for="processingLevel">产品级别</label>
<select id="processingLevel">
<option value="SLC" selected>SLC</option>
<option value="GRD_HD">GRD_HD</option>
<option value="GRD_MD">GRD_MD</option>
<option value="GRD_MS">GRD_MS</option>
<option value="OCN">OCN</option>
</select>
</div>
<div class="field">
<label for="beamMode">波束模式</label>
<select id="beamMode">
<option value="IW" selected>IW</option>
<option value="EW">EW</option>
<option value="WV">WV</option>
</select>
</div>
<div class="field">
<label for="flightDirection">轨道方向</label>
<select id="flightDirection">
<option value="">不限</option>
<option value="ASCENDING">ASCENDING</option>
<option value="DESCENDING">DESCENDING</option>
</select>
</div>
</div>
<div class="grid three">
<div class="field">
<label for="polarization">极化</label>
<input id="polarization" placeholder="VV+VH">
</div>
<div class="field">
<label for="relativeOrbit">相对轨道号</label>
<input id="relativeOrbit" type="number" min="1" placeholder="可留空">
</div>
<div class="field">
<label for="manifestPath">清单 CSV</label>
<input id="manifestPath" placeholder=".local-data\\mudanjiang\\manifest.csv">
</div>
</div>
<div class="toggle-row">
<label class="toggle">
<input id="rawRegion" type="checkbox">
<span>使用原始行政区边界</span>
</label>
</div>
<div class="chip-row" id="selectionSummary"></div>
</section>
</section>
<section class="layout two-col">
<section class="panel stack">
<div class="section-head">
<h2>下载参数</h2>
<button class="primary" id="queueDownloadBtn" type="button">启动下载</button>
</div>
<div class="grid two">
<div class="field">
<label for="dataDir">数据目录</label>
<input id="dataDir" value=".local-data\\mudanjiang\\data">
</div>
<div class="field">
<label for="orbitDir">轨道目录</label>
<input id="orbitDir" value=".local-data\\mudanjiang\\orbits">
</div>
</div>
<div class="grid three">
<label class="field">
<span>跳过产品</span>
<select id="skipData">
<option value="false" selected>否</option>
<option value="true">是</option>
</select>
</label>
<label class="field">
<span>跳过轨道</span>
<select id="skipOrbits">
<option value="false" selected>否</option>
<option value="true">是</option>
</select>
</label>
<div class="field">
<label for="downloadProcesses">并发数</label>
<input id="downloadProcesses" type="number" min="1" value="2">
</div>
</div>
<div class="toggle-row">
<label class="toggle">
<input id="ignoreOrbitErrors" type="checkbox">
<span>忽略轨道失败并允许任务成功结束</span>
</label>
</div>
<div class="grid two">
<div class="field">
<label for="earthdataUser">Earthdata 用户名</label>
<input id="earthdataUser" autocomplete="username" placeholder="可留空">
</div>
<div class="field">
<label for="earthdataPass">Earthdata 密码</label>
<input id="earthdataPass" type="password" autocomplete="current-password" placeholder="可留空">
</div>
</div>
<div class="toolbar">
<button class="ghost" id="saveEarthdataBtn" type="button">保存凭据</button>
<button class="ghost danger" id="clearEarthdataBtn" type="button">清除凭据</button>
</div>
<div class="chip-row" id="earthdataSummary"></div>
<div class="note">
下载原始 Sentinel-1 产品必须提供 Earthdata 用户名和密码。你可以手填后直接下载,也可以先保存到本地配置。当前阶段只下载原始产品包和匹配的 EOF 轨道文件,暂不包含 GeoTIFF 处理。
</div>
</section>
<section class="panel stack">
<div class="section-head">
<h2>任务</h2>
<div class="toolbar">
<button class="ghost" id="refreshJobsBtn" type="button">刷新任务</button>
<button class="ghost" id="refreshDetailBtn" type="button">刷新详情</button>
<button class="ghost danger" id="deleteJobBtn" type="button">删除选中</button>
<button class="ghost danger" id="clearJobsBtn" type="button">清空当前项目</button>
</div>
</div>
<div class="chip-row" id="jobFilterSummary"></div>
<div id="jobList" class="list tall"></div>
</section>
</section>
<section class="panel stack">
<div class="section-head">
<h2>任务详情</h2>
</div>
<div id="jobMeta" class="chip-row"></div>
<div class="detail-layout">
<div class="table-wrap">
<table>
<thead>
<tr>
<th style="width: 32%;">Scene</th>
<th style="width: 16%;">开始时间</th>
<th style="width: 10%;">级别</th>
<th style="width: 10%;">相对轨道</th>
<th style="width: 12%;">产品</th>
<th style="width: 12%;">轨道</th>
<th style="width: 8%;">大小</th>
</tr>
</thead>
<tbody id="sceneTableBody"></tbody>
</table>
</div>
<div class="events" id="eventList"></div>
</div>
</section>
</main>
</div>
<script src="/static/app.js" defer></script>
</body>
</html>