Integrate SBAS workflows and redesign task center

2026-06-13 13:10:12 +08:00
parent b931e8db53
commit 58a87706ef
275 changed files with 18750 additions and 42524 deletions
@@ -40,6 +40,12 @@ PYINT_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env(
    minimum=1,
    maximum=10,
 )
+LANDSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env(
+    "LANDSAR_PRODUCTION_JOB_MAX_ATTEMPTS",
+    1,
+    minimum=1,
+    maximum=10,
+)


 class RunJobRequest(BaseModel):
@@ -243,7 +249,12 @@ async def submit_run(
        "extra": dict(req.extra or {}),
    }

-    from ..services.job_handlers import JOB_TYPE_IDL_RUN_DINSAR, JOB_TYPE_ISCE2_RUN, JOB_TYPE_PYINT_RUN
+    from ..services.job_handlers import (
+        JOB_TYPE_IDL_RUN_DINSAR,
+        JOB_TYPE_ISCE2_RUN,
+        JOB_TYPE_LANDSAR_RUN,
+        JOB_TYPE_PYINT_RUN,
+    )
    create_managed_run = False
    normalized_extra = dict(payload["extra"])

@@ -252,7 +263,7 @@ async def submit_run(
        job_type = JOB_TYPE_IDL_RUN_DINSAR
        max_attempts = DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS
        create_managed_run = True
-    elif req.engine_code in {"isce2", "pyint"}:
+    elif req.engine_code in {"isce2", "pyint", "landsar"}:
        if hasattr(engine, "normalize_extra"):
            try:
                payload["extra"] = engine.normalize_extra(payload["extra"])
@@ -263,10 +274,14 @@ async def submit_run(
            job_type = JOB_TYPE_ISCE2_RUN
            max_attempts = ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS
            create_managed_run = True
-        else:
+        elif req.engine_code == "pyint":
            job_type = JOB_TYPE_PYINT_RUN
            max_attempts = PYINT_PRODUCTION_JOB_MAX_ATTEMPTS
            create_managed_run = True
+        else:
+            job_type = JOB_TYPE_LANDSAR_RUN
+            max_attempts = LANDSAR_PRODUCTION_JOB_MAX_ATTEMPTS
+            create_managed_run = True
        if validation_summary is not None:
            validated_task_count = validation_summary.get("task_count", 0)
            payload["extra"].update(
@@ -4,18 +4,204 @@ import asyncio
 import mimetypes
 import subprocess

-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import FileResponse
 from pydantic import BaseModel, Field, field_validator, model_validator
+from sqlalchemy import String, cast, or_, select

+from .. import database
+from ..config import settings
+from ..models import AuthUserORM, SystemJobORM, SystemTaskORM, TaskLogORM
 from ..services.job_queue_service import job_queue_service
+from ..services.landsar_sbas_service import landsar_sbas_service
 from ..services.sbas_insar_production_service import sbas_insar_production_service
 from ..services.task_service import task_service
+from .dependencies import _require_admin


 router = APIRouter(prefix="/sbas-insar-production", tags=["sbas-insar-production"])


+def _new_session():
+    if database.AsyncSessionLocal is None:
+        database.init_db()
+    if database.AsyncSessionLocal is None:
+        raise RuntimeError("Database session factory is not initialized.")
+    return database.AsyncSessionLocal()
+
+
+def _dt(value):
+    return value.isoformat() if value else None
+
+
+def _task_payload_matches(run_id: str):
+    return or_(
+        cast(SystemTaskORM.params, String).ilike(f"%{run_id}%"),
+        SystemTaskORM.task_name.ilike(f"%{run_id}%"),
+    )
+
+
+def _job_payload_matches(run_id: str, task_ids: list[str]):
+    conditions = [
+        cast(SystemJobORM.payload, String).ilike(f"%{run_id}%"),
+        SystemJobORM.workflow_run_id == run_id,
+    ]
+    if task_ids:
+        conditions.append(SystemJobORM.task_id.in_(task_ids))
+    return or_(*conditions)
+
+
+async def _load_run_background_activity(run_id: str) -> dict:
+    try:
+        async with _new_session() as db:
+            task_result = await db.execute(
+                select(SystemTaskORM)
+                .where(_task_payload_matches(run_id))
+                .order_by(SystemTaskORM.created_at.desc(), SystemTaskORM.id.desc())
+                .limit(10)
+            )
+            tasks = list(task_result.scalars().all())
+            task_ids = [
+                str(task.task_id or "").strip()
+                for task in tasks
+                if str(task.task_id or "").strip()
+            ]
+            job_result = await db.execute(
+                select(SystemJobORM)
+                .where(_job_payload_matches(run_id, task_ids))
+                .order_by(SystemJobORM.created_at.desc(), SystemJobORM.id.desc())
+                .limit(10)
+            )
+            jobs = list(job_result.scalars().all())
+            if not task_ids:
+                task_ids = sorted({
+                    str(job.task_id or "").strip()
+                    for job in jobs
+                    if str(job.task_id or "").strip()
+                })
+            logs = []
+            if task_ids:
+                log_result = await db.execute(
+                    select(TaskLogORM)
+                    .where(TaskLogORM.task_id.in_(task_ids))
+                    .order_by(TaskLogORM.timestamp.desc(), TaskLogORM.id.desc())
+                    .limit(20)
+                )
+                logs = list(log_result.scalars().all())
+    except Exception as exc:
+        return {
+            "schema": "insar.sbas-background-activity/v1",
+            "error": str(exc),
+            "tasks": [],
+            "jobs": [],
+            "task_logs": [],
+            "active": False,
+        }
+
+    active_task_statuses = {"PENDING", "RUNNING"}
+    active_job_statuses = {"READY", "PENDING", "RUNNING", "RETRY"}
+    return {
+        "schema": "insar.sbas-background-activity/v1",
+        "tasks": [
+            {
+                "task_id": task.task_id,
+                "task_type": task.task_type,
+                "task_name": task.task_name,
+                "status": task.status,
+                "progress": task.progress,
+                "message": task.message,
+                "created_at": _dt(task.created_at),
+                "updated_at": _dt(task.updated_at),
+                "started_at": _dt(task.started_at),
+                "ended_at": _dt(task.ended_at),
+            }
+            for task in tasks
+        ],
+        "jobs": [
+            {
+                "job_id": job.job_id,
+                "job_type": job.job_type,
+                "status": job.status,
+                "attempts": job.attempts,
+                "max_attempts": job.max_attempts,
+                "locked_by": job.locked_by,
+                "locked_at": _dt(job.locked_at),
+                "heartbeat_at": _dt(job.heartbeat_at),
+                "created_at": _dt(job.created_at),
+                "updated_at": _dt(job.updated_at),
+                "started_at": _dt(job.started_at),
+                "finished_at": _dt(job.finished_at),
+                "last_error": job.last_error,
+                "task_id": job.task_id,
+            }
+            for job in jobs
+        ],
+        "task_logs": [
+            {
+                "task_id": log.task_id,
+                "level": log.log_level,
+                "message": log.message,
+                "timestamp": _dt(log.timestamp),
+            }
+            for log in logs
+        ],
+        "active": any(str(task.status or "").upper() in active_task_statuses for task in tasks)
+        or any(str(job.status or "").upper() in active_job_statuses for job in jobs),
+    }
+
+
+def _load_wsl_process_summary(run_id: str) -> dict:
+    distro = str(settings.GAMMA_SBAS_WSL_DISTRO or settings.WSL_DISTRO or "").strip()
+    command = [
+        "wsl.exe",
+        *([] if not distro else ["-d", distro]),
+        "--",
+        "bash",
+        "-lc",
+        (
+            "ps -eo pid,etime,stat,args --no-headers | "
+            "grep -E 'gamma|SLC_interp|base_calc|mk_diff|mk_unw|ts_rate| mb |python|bash' | "
+            "grep -v grep | head -20"
+        ),
+    ]
+    try:
+        completed = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            timeout=5,
+            check=False,
+            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+        )
+    except FileNotFoundError:
+        return {"available": False, "error": "wsl.exe not found", "processes": []}
+    except Exception as exc:
+        return {"available": False, "error": str(exc), "processes": []}
+
+    processes = []
+    for line in (completed.stdout or "").splitlines():
+        text = line.strip()
+        if not text:
+            continue
+        parts = text.split(None, 3)
+        processes.append(
+            {
+                "pid": parts[0] if len(parts) > 0 else "",
+                "etime": parts[1] if len(parts) > 1 else "",
+                "stat": parts[2] if len(parts) > 2 else "",
+                "command": parts[3] if len(parts) > 3 else text,
+                "matches_run": run_id in text,
+            }
+        )
+    return {
+        "available": completed.returncode in {0, 1},
+        "returncode": completed.returncode,
+        "distro": distro or None,
+        "processes": processes,
+        "stderr_tail": (completed.stderr or "")[-1200:],
+    }
+
+
 class SbasAoiBbox(BaseModel):
    min_lon: float = Field(ge=-180, le=180)
    min_lat: float = Field(ge=-90, le=90)
@@ -30,6 +216,7 @@ class SbasAoiBbox(BaseModel):


 class SbasStackDiscoverRequest(BaseModel):
+    sensor_family: str = Field(default="LT1", pattern="^(LT1|S1)$")
    source_roots: list[str] | None = None
    orbit_roots: list[str] | None = None
    min_scenes: int = Field(default=3, ge=2, le=100)
@@ -43,7 +230,11 @@ class SbasStackDiscoverRequest(BaseModel):
    discovery_mode: str = Field(default="strict", pattern="^(strict|aoi)$")
    aoi_bbox: SbasAoiBbox | None = None
    min_aoi_coverage_ratio: float = Field(default=0.01, ge=0, le=1)
-    min_common_overlap_ratio: float = Field(default=0.0, ge=0, le=1)
+    min_common_overlap_ratio: float = Field(
+        default_factory=lambda: float(settings.GAMMA_SBAS_MIN_COMMON_OVERLAP_RATIO or 0.30),
+        ge=0,
+        le=1,
+    )

    @field_validator("source_roots", "orbit_roots", mode="before")
    @classmethod
@@ -57,6 +248,14 @@ class SbasStackDiscoverRequest(BaseModel):
        cleaned = [str(item or "").strip() for item in items if str(item or "").strip()]
        return cleaned or None

+    @field_validator("sensor_family", mode="before")
+    @classmethod
+    def _normalize_sensor_family(cls, value):
+        text = str(value or "LT1").strip().upper()
+        if text in {"SENTINEL1", "SENTINEL-1"}:
+            return "S1"
+        return text or "LT1"
+
    @field_validator("platform", "relative_orbit", "orbit_direction", "admin_region", mode="before")
    @classmethod
    def _normalize_optional_text(cls, value):
@@ -165,9 +364,122 @@ class SbasWorkflowJobRequest(SbasWorkflowPrepareRequest):
    timeout_seconds: int = Field(default=172800, ge=60, le=604800)


+class LandsarSbasAutoWorkflowRequest(SbasStackDiscoverRequest):
+    sensor_family: str = Field(default="LT1", pattern="^LT1$")
+    require_orbits: bool = False
+    run_label: str | None = Field(default=None, max_length=160)
+    dem_path: str | None = Field(default=None, max_length=1024)
+    timeout_seconds: int | None = Field(default=None, ge=60, le=604800)
+    import_timeout_seconds: int | None = Field(default=None, ge=60, le=604800)
+    workflow_timeout_seconds: int | None = Field(default=None, ge=60, le=604800)
+    params: dict[str, object] = Field(default_factory=dict)
+
+
@router.get("/capabilities")
 async def get_sbas_insar_capabilities():
-    return sbas_insar_production_service.get_capabilities()
+    capabilities = sbas_insar_production_service.get_capabilities()
+    capabilities["processors"] = [
+        {
+            "processor_code": capabilities.get("processor_code"),
+            "profile_code": "lt1_gamma_sbas",
+            "engine_code": capabilities.get("engine_code"),
+            "label": "Gamma / IPTA SBAS",
+            "enabled": bool((capabilities.get("runtime") or {}).get("enabled", True)),
+        },
+        {
+            **landsar_sbas_service.get_capabilities(),
+            "label": "LandSAR SBAS",
+        },
+    ]
+    return capabilities
+
+
+@router.get("/landsar/capabilities")
+async def get_landsar_sbas_capabilities():
+    return landsar_sbas_service.get_capabilities()
+
+
+@router.post("/landsar/workflows/auto", status_code=202)
+async def submit_landsar_sbas_auto_workflow(request: LandsarSbasAutoWorkflowRequest):
+    try:
+        from ..services.job_handlers import JOB_TYPE_SBAS_LANDSAR_WORKFLOW
+
+        selection_request = {
+            "run_label": request.run_label,
+            "source_roots": request.source_roots,
+            "orbit_roots": request.orbit_roots,
+            "min_scenes": request.min_scenes,
+            "discovery_mode": request.discovery_mode,
+            "admin_region": request.admin_region,
+            "aoi_bbox": request.aoi_bbox.model_dump() if request.aoi_bbox else None,
+            "min_aoi_coverage_ratio": request.min_aoi_coverage_ratio,
+            "min_common_overlap_ratio": request.min_common_overlap_ratio,
+            "limit": request.limit,
+            "dem_path": request.dem_path,
+            "timeout_seconds": request.timeout_seconds,
+            "import_timeout_seconds": request.import_timeout_seconds,
+            "params": dict(request.params or {}),
+        }
+        payload = {
+            "auto_select": True,
+            "selection_request": selection_request,
+            "timeout_seconds": request.workflow_timeout_seconds or request.timeout_seconds,
+        }
+        task_id = await task_service.create_task(
+            task_type=JOB_TYPE_SBAS_LANDSAR_WORKFLOW,
+            task_name="LandSAR SBAS Auto Workflow",
+            params=payload,
+        )
+        job_id = await job_queue_service.create_job(
+            job_type=JOB_TYPE_SBAS_LANDSAR_WORKFLOW,
+            payload=payload,
+            task_id=task_id,
+            max_attempts=1,
+        )
+        return {
+            "message": "LandSAR SBAS auto workflow job queued. Stack selection and input import will run in the background.",
+            "run_id": None,
+            "selection_pending": True,
+            "task_id": task_id,
+            "job_id": job_id,
+            "job_type": JOB_TYPE_SBAS_LANDSAR_WORKFLOW,
+            "status": "QUEUED",
+        }
+    except ValueError as exc:
+        message = str(exc)
+        status_code = 409 if "already running" in message.lower() or "conflict" in message.lower() or "任务冲突" in message else 400
+        raise HTTPException(status_code=status_code, detail=message) from exc
+
+
+@router.get("/landsar/runs")
+async def list_landsar_sbas_runs():
+    return await asyncio.to_thread(landsar_sbas_service.list_runs)
+
+
+@router.get("/landsar/runs/{run_id}")
+async def get_landsar_sbas_run(run_id: str):
+    try:
+        return await asyncio.to_thread(landsar_sbas_service.get_run_detail, run_id)
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/landsar/runs/{run_id}/artifacts/{relative_path:path}")
+async def get_landsar_sbas_run_artifact(run_id: str, relative_path: str):
+    try:
+        artifact_path = await asyncio.to_thread(landsar_sbas_service.resolve_artifact, run_id, relative_path)
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    media_type = mimetypes.guess_type(str(artifact_path))[0] or "application/octet-stream"
+    return FileResponse(
+        artifact_path,
+        media_type=media_type,
+        filename=artifact_path.name,
+    )


@router.post("/stacks/discover")
@@ -175,6 +487,7 @@ async def discover_sbas_insar_stacks(request: SbasStackDiscoverRequest):
    try:
        return await asyncio.to_thread(
            sbas_insar_production_service.discover_stacks,
+            sensor_family=request.sensor_family,
            source_roots=request.source_roots,
            orbit_roots=request.orbit_roots,
            min_scenes=request.min_scenes,
@@ -200,6 +513,7 @@ async def audit_sbas_insar_stack(stack_id: str, request: SbasStackDiscoverReques
        return await asyncio.to_thread(
            sbas_insar_production_service.audit_stack,
            stack_id,
+            sensor_family=request.sensor_family,
            source_roots=request.source_roots,
            orbit_roots=request.orbit_roots,
            min_scenes=request.min_scenes,
@@ -222,6 +536,7 @@ async def submit_sbas_insar_run(stack_id: str, request: SbasRunSubmitRequest):
        return await asyncio.to_thread(
            sbas_insar_production_service.create_run,
            stack_id,
+            sensor_family=request.sensor_family,
            run_label=request.run_label,
            source_roots=request.source_roots,
            orbit_roots=request.orbit_roots,
@@ -253,13 +568,57 @@ async def list_sbas_insar_runs():
@router.get("/runs/{run_id}")
 async def get_sbas_insar_run(run_id: str):
    try:
-        return await asyncio.to_thread(sbas_insar_production_service.get_run_detail, run_id)
+        detail = await asyncio.to_thread(sbas_insar_production_service.get_run_detail, run_id)
+        background_activity, wsl_processes = await asyncio.gather(
+            _load_run_background_activity(run_id),
+            asyncio.to_thread(_load_wsl_process_summary, run_id),
+        )
+        runtime_status = dict(detail.get("runtime_status") or {})
+        runtime_status["background_activity"] = background_activity
+        runtime_status["wsl_processes"] = wsl_processes
+        runtime_status["active"] = bool(
+            runtime_status.get("active")
+            or background_activity.get("active")
+            or any(item.get("matches_run") for item in wsl_processes.get("processes") or [])
+        )
+        detail["runtime_status"] = runtime_status
+        return detail
    except FileNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


+@router.delete("/runs/{run_id}")
+async def delete_sbas_insar_run(
+    run_id: str,
+    current_user: AuthUserORM = Depends(_require_admin),
+):
+    _ = current_user
+    try:
+        async with _new_session() as db:
+            result = await sbas_insar_production_service.delete_run_record(run_id, db=db)
+            try:
+                from ..services.sbas_insar_catalog_service import sbas_insar_catalog_service
+
+                catalog_result = await sbas_insar_catalog_service.rebuild_catalog(db, full_rebuild=True)
+            except Exception as catalog_exc:
+                catalog_result = {
+                    "status": "WARN",
+                    "message": f"SBAS catalog rebuild failed after run deletion: {catalog_exc}",
+                }
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValueError as exc:
+        message = str(exc)
+        status_code = 409 if "active task/job" in message or "running" in message.lower() else 400
+        raise HTTPException(status_code=status_code, detail=message) from exc
+    return {
+        **result,
+        "catalog": catalog_result,
+    }
+
+
@router.post("/runs/{run_id}/workflow", status_code=202)
 async def prepare_sbas_insar_workflow(run_id: str, request: SbasWorkflowPrepareRequest):
    try:
@@ -5,10 +5,11 @@ import os
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

 from ..database import get_db
-from ..models import AuthUserORM
+from ..models import AuthUserORM, SystemTaskORM
 from ..services.job_queue_service import job_queue_service
 from ..services.sbas_insar_catalog_service import (
    JOB_TYPE_REBUILD_SBAS_INSAR_CATALOG,
@@ -26,6 +27,24 @@ class SbasInsarCatalogRebuildRequest(BaseModel):
    full_rebuild: bool = True


+class SbasInsarPointTimeseriesRequest(BaseModel):
+    lon: float
+    lat: float
+
+
+async def _get_active_sbas_catalog_rebuild_task(db: AsyncSession) -> SystemTaskORM | None:
+    result = await db.execute(
+        select(SystemTaskORM)
+        .where(
+            SystemTaskORM.task_type == TASK_TYPE_REBUILD_SBAS_INSAR_CATALOG,
+            SystemTaskORM.status.in_(["PENDING", "RUNNING"]),
+        )
+        .order_by(SystemTaskORM.created_at.desc(), SystemTaskORM.id.desc())
+        .limit(1)
+    )
+    return result.scalars().first()
+
+
@router.get("/sbas-insar-products/catalog-status")
 async def get_sbas_insar_catalog_status(
    current_user: AuthUserORM = Depends(_get_current_user),
@@ -43,12 +62,40 @@ async def queue_sbas_insar_catalog_rebuild(
    admin_user: AuthUserORM = Depends(_require_admin),
 ):
    _ = admin_user
-    task_id = await task_service.create_task(
-        TASK_TYPE_REBUILD_SBAS_INSAR_CATALOG,
-        "SBAS-InSAR result catalog rebuild",
-        params={"full_rebuild": request.full_rebuild},
-        db=db,
-    )
+    existing_task = await _get_active_sbas_catalog_rebuild_task(db)
+    if existing_task is not None:
+        await _add_operation_audit_log(
+            db,
+            request=http_request,
+            action="sbas_insar_catalog_rebuild_already_running",
+            resource="sbas-insar-products/rebuild",
+            detail={"task_id": existing_task.task_id, "full_rebuild": request.full_rebuild},
+        )
+        await db.commit()
+        return {
+            "message": "SBAS-InSAR result catalog rebuild is already queued or running.",
+            "task_id": existing_task.task_id,
+            "already_running": True,
+        }
+
+    try:
+        task_id = await task_service.create_task(
+            TASK_TYPE_REBUILD_SBAS_INSAR_CATALOG,
+            "SBAS-InSAR result catalog rebuild",
+            params={"full_rebuild": request.full_rebuild},
+            db=db,
+        )
+    except ValueError as exc:
+        await db.rollback()
+        existing_task = await _get_active_sbas_catalog_rebuild_task(db)
+        if existing_task is not None:
+            return {
+                "message": "SBAS-InSAR result catalog rebuild is already queued or running.",
+                "task_id": existing_task.task_id,
+                "already_running": True,
+            }
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
    await job_queue_service.create_job(
        JOB_TYPE_REBUILD_SBAS_INSAR_CATALOG,
        payload={"full_rebuild": request.full_rebuild},
@@ -103,6 +150,32 @@ async def get_sbas_insar_product_detail(
    return detail


+@router.post("/sbas-insar-products/{product_db_id}/point-timeseries")
+async def query_sbas_insar_point_timeseries(
+    product_db_id: int,
+    request: SbasInsarPointTimeseriesRequest,
+    current_user: AuthUserORM = Depends(_get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    _ = current_user
+    try:
+        detail = await sbas_insar_catalog_service.query_point_timeseries(
+            db,
+            product_db_id=product_db_id,
+            lon=request.lon,
+            lat=request.lat,
+        )
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+    if detail is None:
+        raise HTTPException(status_code=404, detail="SBAS-InSAR product not found")
+    return detail
+
+
@router.get("/sbas-insar-products/{product_db_id}/preview")
 async def get_sbas_insar_product_preview(
    product_db_id: int,