diff --git a/backend/app/dinsar_engines/registry.py b/backend/app/dinsar_engines/registry.py index 50703a2..408f214 100644 --- a/backend/app/dinsar_engines/registry.py +++ b/backend/app/dinsar_engines/registry.py @@ -29,12 +29,11 @@ def list_engines() -> List[DinsarEngine]: def _bootstrap() -> None: """Imports and registers all built-in engines.""" - from .isce2_engine import Isce2Engine from .landsar_engine import LandsarEngine from .pyint_engine import PyintEngine from .sarscape_engine import SarscapeEngine - for engine in (SarscapeEngine(), Isce2Engine(), PyintEngine(), LandsarEngine()): + for engine in (SarscapeEngine(), PyintEngine(), LandsarEngine()): register(engine) diff --git a/backend/app/models/schemas.py b/backend/app/models/schemas.py index b7b8312..716d190 100644 --- a/backend/app/models/schemas.py +++ b/backend/app/models/schemas.py @@ -567,6 +567,7 @@ class DinsarTaskItem(BaseModel): scene_center_distance_meters: Optional[float] = None status: str remark: Optional[str] = None + engine_results: Optional[Dict[str, Any]] = None created_at: datetime updated_at: datetime diff --git a/backend/app/routers/dinsar_production.py b/backend/app/routers/dinsar_production.py index 0d99dfb..047a348 100644 --- a/backend/app/routers/dinsar_production.py +++ b/backend/app/routers/dinsar_production.py @@ -28,12 +28,6 @@ DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( minimum=1, maximum=20, ) -ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( - "ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS", - 1, - minimum=1, - maximum=10, -) PYINT_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( "PYINT_PRODUCTION_JOB_MAX_ATTEMPTS", 1, @@ -49,7 +43,7 @@ LANDSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( class RunJobRequest(BaseModel): - engine_code: str = Field(..., description="Engine code: sarscape / isce2 / pyint / landsar") + engine_code: str = Field(..., description="Engine code: sarscape / pyint / landsar") profile: str = Field(..., description="Engine profile, for example custom6 / lt1_stripmap / lt1_gamma_dinsar") root_dir: str = Field(..., description="Windows root directory") num_to_process: int = Field(default=0, ge=0, description="How many tasks to process; 0 means all") @@ -251,7 +245,6 @@ async def submit_run( from ..services.job_handlers import ( JOB_TYPE_IDL_RUN_DINSAR, - JOB_TYPE_ISCE2_RUN, JOB_TYPE_LANDSAR_RUN, JOB_TYPE_PYINT_RUN, ) @@ -263,18 +256,14 @@ 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", "landsar"}: + elif req.engine_code in {"pyint", "landsar"}: if hasattr(engine, "normalize_extra"): try: payload["extra"] = engine.normalize_extra(payload["extra"]) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc normalized_extra = dict(payload["extra"]) - if req.engine_code == "isce2": - job_type = JOB_TYPE_ISCE2_RUN - max_attempts = ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS - create_managed_run = True - elif req.engine_code == "pyint": + if req.engine_code == "pyint": job_type = JOB_TYPE_PYINT_RUN max_attempts = PYINT_PRODUCTION_JOB_MAX_ATTEMPTS create_managed_run = True diff --git a/backend/app/routers/dinsar_products.py b/backend/app/routers/dinsar_products.py index 1d0efbc..45d3938 100644 --- a/backend/app/routers/dinsar_products.py +++ b/backend/app/routers/dinsar_products.py @@ -20,6 +20,7 @@ from ..services.result_catalog_service import ( TASK_TYPE_REBUILD_DINSAR_CATALOG, result_catalog_service, ) +from ..services.dinsar_intermediate_cleanup_service import dinsar_intermediate_cleanup_service from ..services.task_service import task_service from .dependencies import ( _add_operation_audit_log, @@ -233,6 +234,42 @@ async def list_dinsar_products( ) +@router.get("/dinsar-products/pairs") +async def list_dinsar_product_pairs( + limit: int = 100, + offset: int = 0, + engine_code: Optional[str] = None, + status: Optional[str] = None, + query: Optional[str] = None, + include_legacy: bool = False, + current_user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + _ = current_user + return await result_catalog_service.list_product_pairs( + db, + limit=limit, + offset=offset, + engine_code=engine_code, + status=status, + query=query, + include_legacy=include_legacy, + ) + + +@router.get("/dinsar-products/pairs/{pair_key}/cleanup-intermediates/plan") +async def get_dinsar_pair_intermediate_cleanup_plan( + pair_key: str, + current_user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + _ = current_user + plan = await dinsar_intermediate_cleanup_service.build_pair_plan(db, pair_key=pair_key) + if int(plan.get("product_count") or 0) == 0: + raise HTTPException(status_code=404, detail="No D-InSAR products found for pair_key") + return plan + + @router.get("/dinsar-products/{product_db_id}") async def get_dinsar_product_detail( product_db_id: int, @@ -246,6 +283,19 @@ async def get_dinsar_product_detail( return detail +@router.get("/dinsar-products/{product_db_id}/cleanup-intermediates/plan") +async def get_dinsar_product_intermediate_cleanup_plan( + product_db_id: int, + current_user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + _ = current_user + plan = await dinsar_intermediate_cleanup_service.build_product_plan(db, product_db_id=product_db_id) + if plan is None: + raise HTTPException(status_code=404, detail="Result product not found") + return plan + + @router.get("/dinsar-products/{product_db_id}/preview") async def get_dinsar_product_preview( product_db_id: int, diff --git a/backend/app/routers/task_batches.py b/backend/app/routers/task_batches.py index 772d9a2..a96efc8 100644 --- a/backend/app/routers/task_batches.py +++ b/backend/app/routers/task_batches.py @@ -28,6 +28,7 @@ from ..models import ( TimeseriesStackPlanItemORM, TimeseriesStackPlanORM, ) +from ..services.dinsar_engine_matrix import build_engine_results_for_task_items from .dependencies import ( _add_operation_audit_log, _refresh_dinsar_batch_summary, @@ -361,7 +362,14 @@ async def list_dinsar_batch_items_endpoint( .offset(safe_offset) .limit(safe_limit) ) - return [DinsarTaskItem.model_validate(i) for i in result.scalars().all()] + items = result.scalars().all() + engine_results_by_item_id = await build_engine_results_for_task_items(db, items) + payload: List[DinsarTaskItem] = [] + for item in items: + row = DinsarTaskItem.model_validate(item) + row.engine_results = engine_results_by_item_id.get(int(item.id or 0), {}) + payload.append(row) + return payload @router.patch("/task-batches/dinsar/{batch_id}/complete-all", response_model=DinsarTaskBatch) diff --git a/backend/app/services/dinsar_engine_matrix.py b/backend/app/services/dinsar_engine_matrix.py new file mode 100644 index 0000000..a1e8396 --- /dev/null +++ b/backend/app/services/dinsar_engine_matrix.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, Iterable, List, Optional + +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import DinsarTaskItemORM, ResultProductORM +from ..utils import normalize_satellite_family + + +DINSAR_CATALOG_NAME = "dinsar" +CURRENT_DINSAR_ENGINE_ORDER = ("sarscape", "landsar", "pyint") +LEGACY_DINSAR_ENGINE_CODES = {"isce2"} + +DEFAULT_PROFILE_BY_ENGINE = { + "sarscape": "custom6", + "landsar": "lt1_dinsar", + "pyint": "lt1_gamma_dinsar", +} + +S1_PROFILE_BY_ENGINE = { + "pyint": "s1_gamma_dinsar", +} + + +def normalize_dinsar_engine_code(value: Any) -> str: + text = str(value or "").strip().lower() + if text == "gamma": + return "pyint" + return text + + +def is_current_dinsar_engine(value: Any) -> bool: + return normalize_dinsar_engine_code(value) in CURRENT_DINSAR_ENGINE_ORDER + + +def _flatten_text_tokens(value: Any, output: List[str]) -> None: + if value is None: + return + if isinstance(value, dict): + for key, item in value.items(): + _flatten_text_tokens(key, output) + _flatten_text_tokens(item, output) + return + if isinstance(value, (list, tuple, set)): + for item in value: + _flatten_text_tokens(item, output) + return + text = str(value).strip().lower() + if text: + output.append(text) + + +def infer_dinsar_data_family(*values: Any) -> str: + tokens: List[str] = [] + for value in values: + _flatten_text_tokens(value, tokens) + joined = " ".join(tokens) + normalized = normalize_satellite_family(joined) + if normalized == "s1" or "sentinel" in joined or "sentinel-1" in joined or joined.startswith("s1"): + return "s1" + if normalized == "lt1" or "lt1" in joined or "陆探" in joined: + return "lt1" + return normalized or "unknown" + + +def allowed_engines_for_data_family(data_family: str) -> set[str]: + normalized = str(data_family or "").strip().lower() + if normalized == "s1": + return {"pyint"} + return set(CURRENT_DINSAR_ENGINE_ORDER) + + +def default_profile_for_engine(engine_code: str, data_family: str = "unknown") -> str: + engine = normalize_dinsar_engine_code(engine_code) + family = str(data_family or "").strip().lower() + if family == "s1" and engine in S1_PROFILE_BY_ENGINE: + return S1_PROFILE_BY_ENGINE[engine] + return DEFAULT_PROFILE_BY_ENGINE.get(engine, engine) + + +def _timestamp(value: Optional[datetime]) -> float: + if value is None: + return 0.0 + try: + return value.timestamp() + except Exception: + return 0.0 + + +def _product_status(product: ResultProductORM) -> str: + status = str(product.status or "").strip().upper() + health = str(product.health_status or "").strip().upper() + if status in {"READY", "COMPLETED", "SUCCESS"} and health not in {"ERROR", "FAILED"}: + return "ready" + if status in {"FAILED", "ERROR"} or health in {"ERROR", "FAILED"}: + return "failed" + if status: + return status.lower() + return "ready" + + +def serialize_engine_result( + *, + engine_code: str, + data_family: str, + product: Optional[ResultProductORM], + allowed: bool, + legacy: bool = False, +) -> Dict[str, Any]: + engine = normalize_dinsar_engine_code(engine_code) + if legacy: + status = "legacy" + can_dispatch = False + skip_reason = "legacy_engine" + elif not allowed: + status = "blocked" + can_dispatch = False + skip_reason = "unsupported_data_family" + elif product is None: + status = "missing" + can_dispatch = True + skip_reason = None + else: + status = _product_status(product) + can_dispatch = status in {"missing", "failed"} + skip_reason = "result_exists" if status == "ready" else None + + return { + "engine_code": engine, + "allowed": bool(allowed), + "legacy": bool(legacy), + "status": status, + "profile_code": ( + product.profile_code + if product is not None and product.profile_code + else default_profile_for_engine(engine, data_family) + ), + "latest_product_id": product.id if product is not None else None, + "product_id": product.product_id if product is not None else None, + "run_key": product.run_key if product is not None else None, + "published_at": product.published_at if product is not None else None, + "health_status": product.health_status if product is not None else None, + "primary_asset_path": product.primary_asset_path if product is not None else None, + "preview_path": product.preview_path if product is not None else None, + "can_dispatch": can_dispatch, + "skip_reason": skip_reason, + } + + +def build_engine_results( + *, + products: Iterable[ResultProductORM], + data_family: str = "unknown", + include_legacy: bool = False, +) -> Dict[str, Dict[str, Any]]: + latest_by_engine: Dict[str, ResultProductORM] = {} + legacy_latest: Dict[str, ResultProductORM] = {} + for product in products: + engine = normalize_dinsar_engine_code(product.engine_code) + target = legacy_latest if engine in LEGACY_DINSAR_ENGINE_CODES else latest_by_engine + current = target.get(engine) + if current is None or (_timestamp(product.published_at), product.id or 0) > ( + _timestamp(current.published_at), + current.id or 0, + ): + target[engine] = product + + allowed = allowed_engines_for_data_family(data_family) + matrix: Dict[str, Dict[str, Any]] = {} + for engine in CURRENT_DINSAR_ENGINE_ORDER: + matrix[engine] = serialize_engine_result( + engine_code=engine, + data_family=data_family, + product=latest_by_engine.get(engine), + allowed=engine in allowed, + ) + + if include_legacy: + for engine, product in sorted(legacy_latest.items()): + matrix[engine] = serialize_engine_result( + engine_code=engine, + data_family=data_family, + product=product, + allowed=False, + legacy=True, + ) + return matrix + + +async def build_engine_results_for_task_items( + db: AsyncSession, + items: List[DinsarTaskItemORM], +) -> Dict[int, Dict[str, Dict[str, Any]]]: + if not items: + return {} + + pair_keys = sorted({str(item.pair_key or "").strip() for item in items if str(item.pair_key or "").strip()}) + aliases = sorted( + { + str(item.task_alias or item.task_name or "").strip() + for item in items + if str(item.task_alias or item.task_name or "").strip() + } + ) + conditions = [] + if pair_keys: + conditions.append(ResultProductORM.pair_key.in_(pair_keys)) + if aliases: + conditions.append(ResultProductORM.task_alias.in_(aliases)) + conditions.append(ResultProductORM.task_name.in_(aliases)) + if not conditions: + return { + int(item.id): build_engine_results( + products=[], + data_family=infer_dinsar_data_family(item.master_satellite, item.slave_satellite), + ) + for item in items + if item.id is not None + } + + result = await db.execute( + select(ResultProductORM) + .where(ResultProductORM.catalog_name == DINSAR_CATALOG_NAME) + .where(or_(*conditions)) + .order_by(ResultProductORM.published_at.desc().nullslast(), ResultProductORM.id.desc()) + ) + products = result.scalars().all() + + by_pair_key: Dict[str, List[ResultProductORM]] = {} + by_alias: Dict[str, List[ResultProductORM]] = {} + for product in products: + pair_key = str(product.pair_key or "").strip() + if pair_key: + by_pair_key.setdefault(pair_key, []).append(product) + for alias in {str(product.task_alias or "").strip(), str(product.task_name or "").strip()}: + if alias: + by_alias.setdefault(alias, []).append(product) + + output: Dict[int, Dict[str, Dict[str, Any]]] = {} + for item in items: + item_products: List[ResultProductORM] = [] + pair_key = str(item.pair_key or "").strip() + alias = str(item.task_alias or item.task_name or "").strip() + if pair_key: + item_products.extend(by_pair_key.get(pair_key, [])) + if not item_products and alias: + item_products.extend(by_alias.get(alias, [])) + data_family = infer_dinsar_data_family(item.master_satellite, item.slave_satellite) + output[int(item.id)] = build_engine_results(products=item_products, data_family=data_family) + return output diff --git a/backend/app/services/dinsar_intermediate_cleanup_service.py b/backend/app/services/dinsar_intermediate_cleanup_service.py new file mode 100644 index 0000000..dec8036 --- /dev/null +++ b/backend/app/services/dinsar_intermediate_cleanup_service.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import os +from typing import Any, Dict, List, Optional, Set + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import ResultAssetORM, ResultProductORM +from .dinsar_engine_matrix import is_current_dinsar_engine + + +DINSAR_CATALOG_NAME = "dinsar" +PRESERVED_DIR_NAMES = {"assets", "preview", "current"} +PRESERVED_FILE_NAMES = { + "manifest.json", + "execution_manifest.json", + ".dinsar_run.json", + ".dinsar_pair.json", + "task_manifest.json", +} + + +def _norm_path(value: Any) -> str: + text = str(value or "").strip() + if not text: + return "" + return os.path.normcase(os.path.normpath(os.path.abspath(text))) + + +def _is_within(child: str, parent: str) -> bool: + child_norm = _norm_path(child) + parent_norm = _norm_path(parent) + if not child_norm or not parent_norm: + return False + try: + return os.path.commonpath([child_norm, parent_norm]) == parent_norm + except ValueError: + return False + + +def _path_size(path: str) -> int: + if not os.path.exists(path): + return 0 + if os.path.isfile(path): + try: + return int(os.path.getsize(path) or 0) + except OSError: + return 0 + total = 0 + for root, _dirs, files in os.walk(path): + for name in files: + file_path = os.path.join(root, name) + try: + total += int(os.path.getsize(file_path) or 0) + except OSError: + continue + return total + + +def _required_assets_ok(assets: List[ResultAssetORM]) -> tuple[bool, List[Dict[str, Any]]]: + missing: List[Dict[str, Any]] = [] + for asset in assets: + if not asset.is_required: + continue + path = _norm_path(asset.absolute_path) + exists = bool(path and os.path.exists(path)) + if not exists: + missing.append( + { + "asset_id": asset.id, + "asset_role": asset.asset_role, + "asset_name": asset.asset_name, + "absolute_path": asset.absolute_path, + } + ) + return len(missing) == 0, missing + + +def _is_preserved_path(path: str, product: ResultProductORM, asset_paths: Set[str]) -> bool: + normalized = _norm_path(path) + if not normalized: + return True + if normalized in asset_paths: + return True + if any(_is_within(asset_path, normalized) for asset_path in asset_paths): + return True + for preserve in [ + product.publish_dir, + product.manifest_path, + product.preview_path, + product.primary_asset_path, + ]: + preserve_norm = _norm_path(preserve) + if preserve_norm and normalized == preserve_norm: + return True + if preserve_norm and _is_within(preserve_norm, normalized): + return True + name = os.path.basename(normalized) + if name in PRESERVED_FILE_NAMES or name in PRESERVED_DIR_NAMES: + return True + return False + + +class DinsarIntermediateCleanupService: + async def build_product_plan( + self, + db: AsyncSession, + *, + product_db_id: int, + ) -> Optional[Dict[str, Any]]: + product_result = await db.execute( + select(ResultProductORM).where( + ResultProductORM.id == product_db_id, + ResultProductORM.catalog_name == DINSAR_CATALOG_NAME, + ) + ) + product = product_result.scalar_one_or_none() + if product is None: + return None + + asset_result = await db.execute( + select(ResultAssetORM) + .where(ResultAssetORM.product_ref_id == product.id) + .order_by(ResultAssetORM.asset_role.asc(), ResultAssetORM.id.asc()) + ) + assets = asset_result.scalars().all() + asset_paths = {_norm_path(asset.absolute_path) for asset in assets if _norm_path(asset.absolute_path)} + required_ok, missing_required_assets = _required_assets_ok(assets) + manifest_exists = bool(product.manifest_path and os.path.isfile(_norm_path(product.manifest_path))) + current_engine = is_current_dinsar_engine(product.engine_code) + + blockers: List[str] = [] + if not current_engine: + blockers.append("legacy_or_unknown_engine") + if not manifest_exists: + blockers.append("manifest_missing") + if not required_ok: + blockers.append("required_assets_missing") + + candidates: List[Dict[str, Any]] = [] + seen: Set[str] = set() + + def add_candidate(path: Any, reason: str) -> None: + normalized = _norm_path(path) + if not normalized or normalized in seen: + return + seen.add(normalized) + if _is_preserved_path(normalized, product, asset_paths): + return + publish_dir = _norm_path(product.publish_dir) + native_output_dir = _norm_path(product.native_output_dir) + if publish_dir and normalized == publish_dir: + return + if publish_dir and not _is_within(normalized, publish_dir) and normalized != native_output_dir: + blockers.append(f"candidate_outside_publish_dir:{normalized}") + return + candidates.append( + { + "path": normalized, + "reason": reason, + "exists": os.path.exists(normalized), + "is_dir": os.path.isdir(normalized), + "size_bytes": _path_size(normalized), + } + ) + + native_output_dir = _norm_path(product.native_output_dir) + if native_output_dir: + add_candidate(native_output_dir, "native_output_dir") + + publish_dir = _norm_path(product.publish_dir) + if publish_dir: + add_candidate(os.path.join(publish_dir, "native"), "managed_native_dir") + for name in ("work", "tmp", "temp", "intermediate", "landsar_input", "landsar_output"): + add_candidate(os.path.join(publish_dir, name), f"managed_{name}_dir") + + deletable = len(blockers) == 0 + total_size = sum(int(item.get("size_bytes") or 0) for item in candidates if item.get("exists")) + return { + "schema": "insar.dinsar-intermediate-cleanup-plan/v1", + "dry_run": True, + "deletable": deletable, + "product": { + "id": product.id, + "product_id": product.product_id, + "pair_key": product.pair_key, + "run_key": product.run_key, + "engine_code": product.engine_code, + "profile_code": product.profile_code, + "publish_dir": product.publish_dir, + "manifest_path": product.manifest_path, + "native_output_dir": product.native_output_dir, + }, + "checks": { + "manifest_exists": manifest_exists, + "required_assets_ok": required_ok, + "current_engine": current_engine, + "missing_required_assets": missing_required_assets, + }, + "blockers": blockers, + "candidates": candidates, + "candidate_count": len(candidates), + "total_size_bytes": total_size, + "preserve": { + "directories": sorted(PRESERVED_DIR_NAMES), + "files": sorted(PRESERVED_FILE_NAMES), + "asset_count": len(asset_paths), + }, + } + + async def build_pair_plan( + self, + db: AsyncSession, + *, + pair_key: str, + ) -> Dict[str, Any]: + normalized_pair_key = str(pair_key or "").strip() + result = await db.execute( + select(ResultProductORM) + .where( + ResultProductORM.catalog_name == DINSAR_CATALOG_NAME, + ResultProductORM.pair_key == normalized_pair_key, + ) + .order_by(ResultProductORM.published_at.desc().nullslast(), ResultProductORM.id.desc()) + ) + products = result.scalars().all() + product_plans: List[Dict[str, Any]] = [] + for product in products: + plan = await self.build_product_plan(db, product_db_id=int(product.id)) + if plan is not None: + product_plans.append(plan) + return { + "schema": "insar.dinsar-intermediate-cleanup-pair-plan/v1", + "dry_run": True, + "pair_key": normalized_pair_key, + "product_count": len(product_plans), + "deletable_product_count": sum(1 for item in product_plans if item.get("deletable")), + "total_size_bytes": sum(int(item.get("total_size_bytes") or 0) for item in product_plans), + "products": product_plans, + } + + +dinsar_intermediate_cleanup_service = DinsarIntermediateCleanupService() diff --git a/backend/app/services/dinsar_production_service.py b/backend/app/services/dinsar_production_service.py index 4053f04..b86d082 100644 --- a/backend/app/services/dinsar_production_service.py +++ b/backend/app/services/dinsar_production_service.py @@ -33,7 +33,6 @@ from .workflow_service import workflow_service TASK_TYPE_DINSAR_PRODUCTION = "IDL_RUN_DINSAR" -TASK_TYPE_ISCE2_DINSAR_PRODUCTION = "ISCE2_RUN" TASK_TYPE_PYINT_DINSAR_PRODUCTION = "PYINT_RUN" TASK_TYPE_LANDSAR_DINSAR_PRODUCTION = "LANDSAR_RUN" RUN_STATUS_PENDING = "PENDING" @@ -82,8 +81,6 @@ def _task_type_for_engine(engine_code: str) -> str: normalized = str(engine_code or "").strip().lower() if normalized == "sarscape": return TASK_TYPE_DINSAR_PRODUCTION - if normalized == "isce2": - return TASK_TYPE_ISCE2_DINSAR_PRODUCTION if normalized in {"pyint", "gamma"}: return TASK_TYPE_PYINT_DINSAR_PRODUCTION if normalized == "landsar": @@ -95,8 +92,6 @@ def _workflow_name_for_engine(engine_code: str) -> str: normalized = str(engine_code or "").strip().lower() if normalized == "sarscape": return "dinsar_sarscape_production" - if normalized == "isce2": - return "dinsar_isce2_production" if normalized in {"pyint", "gamma"}: return "dinsar_pyint_gamma_production" if normalized == "landsar": @@ -108,8 +103,6 @@ def _workflow_step_name_for_engine(engine_code: str) -> str: normalized = str(engine_code or "").strip().lower() if normalized == "sarscape": return RUNS_STEP_NAME - if normalized == "isce2": - return "Execute ISCE2 D-InSAR items" if normalized in {"pyint", "gamma"}: return "Execute PyINT/Gamma D-InSAR items" if normalized == "landsar": @@ -546,8 +539,6 @@ def _safe_epoch(value: Optional[datetime]) -> Optional[int]: def _runtime_id_for_engine(engine_code: Optional[str]) -> Optional[str]: normalized = str(engine_code or "").strip().lower() - if normalized == "isce2": - return settings.ISCE2_RUNTIME_ID or None if normalized in {"pyint", "gamma"}: return settings.PYINT_RUNTIME_ID or None return None diff --git a/backend/app/services/result_catalog_service.py b/backend/app/services/result_catalog_service.py index 63fbb81..c78cb29 100644 --- a/backend/app/services/result_catalog_service.py +++ b/backend/app/services/result_catalog_service.py @@ -48,6 +48,13 @@ from .dinsar_result_layout_service import ( is_standard_envi_disp_file, is_standard_isce2_disp_file, ) +from .dinsar_engine_matrix import ( + CURRENT_DINSAR_ENGINE_ORDER, + build_engine_results, + infer_dinsar_data_family, + is_current_dinsar_engine, + normalize_dinsar_engine_code, +) from .product_package_schema import build_canonical_descriptor, normalize_package_manifest from .product_packaging import build_dinsar_package_manifest @@ -1459,6 +1466,130 @@ class ResultCatalogService: "has_more": offset + len(items) < total, } + async def list_product_pairs( + self, + db: AsyncSession, + *, + limit: int = 100, + offset: int = 0, + engine_code: Optional[str] = None, + status: Optional[str] = None, + query: Optional[str] = None, + include_legacy: bool = False, + ) -> Dict[str, Any]: + limit = max(1, min(int(limit or 100), 500)) + offset = max(0, int(offset or 0)) + normalized_engine = normalize_dinsar_engine_code(engine_code) if engine_code else None + + stmt = select(ResultProductORM).where(ResultProductORM.catalog_name == DINSAR_CATALOG_NAME) + if not include_legacy: + stmt = stmt.where(ResultProductORM.engine_code.in_(list(CURRENT_DINSAR_ENGINE_ORDER))) + if normalized_engine: + stmt = stmt.where(ResultProductORM.engine_code == normalized_engine) + if status: + stmt = stmt.where(ResultProductORM.status == status) + if query: + like_value = f"%{query.strip()}%" + stmt = stmt.where( + or_( + ResultProductORM.display_name.ilike(like_value), + ResultProductORM.product_id.ilike(like_value), + ResultProductORM.task_name.ilike(like_value), + ResultProductORM.task_alias.ilike(like_value), + ResultProductORM.pair_key.ilike(like_value), + ResultProductORM.run_key.ilike(like_value), + ) + ) + + result = await db.execute( + stmt.order_by( + ResultProductORM.published_at.desc().nullslast(), + ResultProductORM.id.desc(), + ) + ) + products = result.scalars().all() + + groups: Dict[str, Dict[str, Any]] = {} + for product in products: + pair_key = str(product.pair_key or "").strip() or f"product:{product.id}" + group = groups.setdefault( + pair_key, + { + "pair_key": product.pair_key, + "pair_uid": product.pair_uid, + "task_name": product.task_name, + "task_alias": product.task_alias, + "network_run_id": product.network_run_id, + "network_edge_id": product.network_edge_id, + "policy_version": product.policy_version, + "selection_strategy": product.selection_strategy, + "latest_published_at": product.published_at, + "data_family": infer_dinsar_data_family( + product.task_name, + product.task_alias, + product.profile_code, + product.summary_json if isinstance(product.summary_json, dict) else None, + ), + "_products": [], + }, + ) + group["_products"].append(product) + if product.published_at and ( + group.get("latest_published_at") is None + or product.published_at > group["latest_published_at"] + ): + group["latest_published_at"] = product.published_at + for key in ("pair_uid", "task_name", "task_alias", "network_run_id", "network_edge_id", "policy_version", "selection_strategy"): + if group.get(key) in (None, "") and getattr(product, key) not in (None, ""): + group[key] = getattr(product, key) + + items: List[Dict[str, Any]] = [] + for group in groups.values(): + products_for_group: List[ResultProductORM] = group.pop("_products") + engine_results = build_engine_results( + products=products_for_group, + data_family=group.get("data_family") or "unknown", + include_legacy=include_legacy, + ) + available_results = [ + result_payload + for result_payload in engine_results.values() + if result_payload.get("latest_product_id") is not None + and (include_legacy or is_current_dinsar_engine(result_payload.get("engine_code"))) + ] + ready_count = sum(1 for item in engine_results.values() if item.get("status") == "ready") + primary_result = next( + ( + item + for engine in CURRENT_DINSAR_ENGINE_ORDER + for item in [engine_results.get(engine)] + if item and item.get("latest_product_id") is not None + ), + None, + ) + group.update( + { + "engine_results": engine_results, + "available_engine_count": len(available_results), + "ready_engine_count": ready_count, + "primary_product_id": primary_result.get("latest_product_id") if primary_result else None, + "primary_engine_code": primary_result.get("engine_code") if primary_result else None, + "status": "ready" if ready_count else "missing", + } + ) + items.append(group) + + items.sort(key=lambda item: item.get("latest_published_at") or datetime.min, reverse=True) + total = len(items) + paged = items[offset : offset + limit] + return { + "items": paged, + "total": total, + "limit": limit, + "offset": offset, + "has_more": offset + len(paged) < total, + } + async def get_product_detail( self, db: AsyncSession, diff --git a/frontend/src/App.css b/frontend/src/App.css index 8a13ab3..4c53f86 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -428,6 +428,71 @@ button:focus-visible { font-size: 11px; color: var(--color-text-muted); } + +.batch-engine-results, +.dinsar-engine-result-row { + display: flex; + flex-wrap: wrap; + gap: 6px; + margin-top: 6px; +} + +.batch-engine-chip, +.dinsar-engine-result-chip { + display: inline-flex; + align-items: center; + justify-content: center; + min-height: 22px; + padding: 3px 8px; + border-radius: 999px; + border: 1px solid rgba(100, 116, 139, 0.24); + background: rgba(148, 163, 184, 0.1); + color: #475569; + font-size: 11px; + font-weight: 700; + line-height: 1.2; +} + +.batch-engine-chip.tone-ready, +.dinsar-engine-result-chip.tone-ready { + border-color: rgba(22, 163, 74, 0.26); + background: rgba(22, 163, 74, 0.1); + color: #166534; +} + +.batch-engine-chip.tone-error, +.dinsar-engine-result-chip.tone-error { + border-color: rgba(220, 38, 38, 0.26); + background: rgba(220, 38, 38, 0.1); + color: #991b1b; +} + +.batch-engine-chip.tone-warn, +.dinsar-engine-result-chip.tone-warn { + border-color: rgba(217, 119, 6, 0.28); + background: rgba(245, 158, 11, 0.12); + color: #92400e; +} + +.batch-engine-chip.tone-info, +.dinsar-engine-result-chip.tone-info { + border-color: rgba(37, 99, 235, 0.26); + background: rgba(37, 99, 235, 0.1); + color: #1d4ed8; +} + +.dinsar-catalog-cleanup-head { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 12px; +} + +.dinsar-catalog-cleanup-plan { + display: grid; + gap: 10px; + margin-top: 10px; +} .batch-item select, .batch-item input { font-size: 12px; diff --git a/frontend/src/DinsarProductionPanel.jsx b/frontend/src/DinsarProductionPanel.jsx index 9b1b148..7c1ea35 100644 --- a/frontend/src/DinsarProductionPanel.jsx +++ b/frontend/src/DinsarProductionPanel.jsx @@ -38,18 +38,16 @@ const ENGINE_STATUS_LABEL = { const ENGINE_LABEL = { sarscape: 'SARscape', - isce2: 'ISCE2', pyint: 'PyINT / Gamma', landsar: 'LANDSAR', }; const TASK_TYPE_LABEL = { - ISCE2_RUN: 'ISCE2生产', PYINT_RUN: 'PyINT/Gamma生产', LANDSAR_RUN: 'LandSAR生产', IDL_RUN_DINSAR: 'ENVI生产', }; -const DINSAR_PRODUCTION_TASK_TYPES = ['ISCE2_RUN', 'PYINT_RUN', 'LANDSAR_RUN', 'IDL_RUN_DINSAR']; +const DINSAR_PRODUCTION_TASK_TYPES = ['PYINT_RUN', 'LANDSAR_RUN', 'IDL_RUN_DINSAR']; const STATUS_LABEL = { PENDING: '等待中', @@ -112,7 +110,6 @@ function formatStatus(status) { } function taskTypeToEngine(taskType) { - if (taskType === 'ISCE2_RUN') return 'isce2'; if (taskType === 'PYINT_RUN') return 'pyint'; if (taskType === 'LANDSAR_RUN') return 'landsar'; if (taskType === 'IDL_RUN_DINSAR') return 'sarscape'; @@ -602,8 +599,6 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) ? 'PyINT/Gamma 会按目标网格尺寸自动换算多视；新增 Gamma 残余重去平在解缠后执行 rascc_mask/quad_fit/quad_sub，再导出 native 和标准 GeoTIFF。' : selectedEngine === 'landsar' ? 'LandSAR 当前使用已跑通的稳定参数。GACOS 大气相位改正需要外部大气延迟文件，未配置文件前不可启用；垂直向形变为可选输出，默认关闭。' - : selectedEngine === 'isce2' - ? '这些参数现在按执行、交付、增强分组展示。结果异常时，优先尝试关闭增强项，再回看基础几何和配对质量。' : '这些参数影响当前引擎的生产模板。建议先使用默认值，只有在结果边界、噪声或几何表现异常时再逐项调整。'; const pyintPreviewBlocksSubmit = selectedEngine === 'pyint' && pyintPreview && pyintPreview.allow_submit === false; const taskMonitor = useTaskMonitor({ @@ -617,9 +612,7 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) ? { task_id: latestRunWithTask.task_id, task_type: - latestRunWithTask.engine === 'isce2' - ? 'ISCE2_RUN' - : latestRunWithTask.engine === 'pyint' + latestRunWithTask.engine === 'pyint' ? 'PYINT_RUN' : latestRunWithTask.engine === 'landsar' ? 'LANDSAR_RUN' @@ -1305,11 +1298,6 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) disabled={readOnly} style={{ width: 140, padding: '5px 8px', borderRadius: 4, border: '1px solid #e2e8f0', fontSize: 13 }} /> - {selectedEngine === 'isce2' && currentDefaultTimeoutSec > 0 && ( -
- ISCE2 默认按单对任务使用 {currentDefaultTimeoutSec} 秒；批量目录会串行逐对套用该超时。 -
- )} {selectedEngine === 'pyint' && currentDefaultTimeoutSec > 0 && (
PyINT 默认按单对任务使用 {currentDefaultTimeoutSec} 秒；当前会逐对串行创建工作区并运行外部 PyINT / Gamma 流程，native 会在主流程和重去平后统一写入。 diff --git a/frontend/src/api/dinsarProducts.js b/frontend/src/api/dinsarProducts.js index d2285f2..f2b13c5 100644 --- a/frontend/src/api/dinsarProducts.js +++ b/frontend/src/api/dinsarProducts.js @@ -12,5 +12,14 @@ export const queueDinsarCatalogRebuild = payload => export const listDinsarProducts = (params = {}) => apiClient.get('/dinsar-products', { params }).then(r => r.data); +export const listDinsarProductPairs = (params = {}) => + apiClient.get('/dinsar-products/pairs', { params }).then(r => r.data); + export const getDinsarProductDetail = productId => apiClient.get(`/dinsar-products/${encodeURIComponent(productId)}`).then(r => r.data); + +export const getDinsarProductCleanupPlan = productId => + apiClient.get(`/dinsar-products/${encodeURIComponent(productId)}/cleanup-intermediates/plan`).then(r => r.data); + +export const getDinsarPairCleanupPlan = pairKey => + apiClient.get(`/dinsar-products/pairs/${encodeURIComponent(pairKey)}/cleanup-intermediates/plan`).then(r => r.data); diff --git a/frontend/src/components/DinsarCatalogPanel.jsx b/frontend/src/components/DinsarCatalogPanel.jsx index 5daecdb..3783a7d 100644 --- a/frontend/src/components/DinsarCatalogPanel.jsx +++ b/frontend/src/components/DinsarCatalogPanel.jsx @@ -4,7 +4,8 @@ import apiClient from '../api/client'; import { getDinsarCatalogStatus, getDinsarProductDetail, - listDinsarProducts, + getDinsarProductCleanupPlan, + listDinsarProductPairs, queueDinsarCatalogRebuild, queueDinsarProductPublish, } from '../api/dinsarProducts'; @@ -22,6 +23,11 @@ const STATUS_TONE_MAP = { WARN: 'warn', ERROR: 'error', REBUILDING: 'info', + ready: 'ready', + missing: 'neutral', + failed: 'error', + blocked: 'warn', + legacy: 'neutral', }; function formatDateTime(value) { @@ -33,6 +39,19 @@ function formatDateTime(value) { } } +function formatBytes(value) { + const size = Number(value || 0); + if (!Number.isFinite(size) || size <= 0) return '0 B'; + const units = ['B', 'KB', 'MB', 'GB', 'TB']; + let next = size; + let index = 0; + while (next >= 1024 && index < units.length - 1) { + next /= 1024; + index += 1; + } + return `${next.toFixed(index === 0 ? 0 : 1)} ${units[index]}`; +} + function parseDirectoryList(value) { return [...new Set( String(value || '') @@ -50,6 +69,15 @@ function StatusPill({ label, tone = 'neutral' }) { return {label}; } +function engineStatusTone(status) { + const normalized = String(status || '').toLowerCase(); + if (normalized === 'ready') return 'ready'; + if (normalized === 'failed') return 'error'; + if (normalized === 'blocked') return 'warn'; + if (normalized === 'running') return 'info'; + return 'neutral'; +} + function MetaField({ label, value, multiline = false }) { const displayValue = value === null || value === undefined || value === '' ? '-' : value; return ( @@ -68,8 +96,12 @@ export default function DinsarCatalogPanel({ }) { const [catalogStatus, setCatalogStatus] = useState(null); const [products, setProducts] = useState([]); + const [productPairs, setProductPairs] = useState([]); + const [selectedPairKey, setSelectedPairKey] = useState(''); const [selectedProductId, setSelectedProductId] = useState(null); const [selectedProduct, setSelectedProduct] = useState(null); + const [cleanupPlan, setCleanupPlan] = useState(null); + const [cleanupPlanLoading, setCleanupPlanLoading] = useState(false); const [loading, setLoading] = useState(false); const [detailLoading, setDetailLoading] = useState(false); const [actionLoading, setActionLoading] = useState(false); @@ -94,8 +126,8 @@ export default function DinsarCatalogPanel({ ); const engineOptions = useMemo( - () => buildDinsarEngineOptions(products, { includeKnown: true }), - [products] + () => buildDinsarEngineOptions([], { includeKnown: true }), + [] ); const selectedEngineMeta = useMemo( () => (engineFilter === DINSAR_ENGINE_ALL ? null : getDinsarEngineMeta(engineFilter)), @@ -114,7 +146,7 @@ export default function DinsarCatalogPanel({ try { const [statusData, productData] = await Promise.all([ getDinsarCatalogStatus(), - listDinsarProducts({ + listDinsarProductPairs({ limit: listLimit, offset: 0, engine_code: engineFilter === DINSAR_ENGINE_ALL ? undefined : engineFilter, @@ -122,18 +154,35 @@ export default function DinsarCatalogPanel({ }), ]); setCatalogStatus(statusData); - const nextItems = Array.isArray(productData?.items) ? productData.items : []; + const nextPairs = Array.isArray(productData?.items) ? productData.items : []; + setProductPairs(nextPairs); + const nextItems = nextPairs + .map((item) => ({ + id: item.primary_product_id, + engine_code: item.primary_engine_code, + pair_key: item.pair_key, + })) + .filter((item) => item.id); setProducts(nextItems); + setSelectedPairKey((current) => { + if (current && nextPairs.some((item) => (item.pair_key || `pair:${item.primary_product_id}`) === current)) { + return current; + } + const first = nextPairs[0]; + return first ? (first.pair_key || `pair:${first.primary_product_id}`) : ''; + }); setSelectedProductId((current) => { if (current && nextItems.some((item) => item.id === current)) { return current; } - return nextItems[0]?.id ?? null; + return nextPairs[0]?.primary_product_id ?? null; }); } catch (error) { setActionMessage(`结果目录状态加载失败：${error?.response?.data?.detail || error.message}`); setCatalogStatus(null); setProducts([]); + setProductPairs([]); + setSelectedPairKey(''); setSelectedProductId(null); } finally { setLoading(false); @@ -143,9 +192,11 @@ export default function DinsarCatalogPanel({ const loadProductDetail = useCallback(async (productId) => { if (!productId) { setSelectedProduct(null); + setCleanupPlan(null); return; } setSelectedProduct(null); + setCleanupPlan(null); setDetailLoading(true); try { const detail = await getDinsarProductDetail(productId); @@ -159,6 +210,21 @@ export default function DinsarCatalogPanel({ } }, []); + const loadCleanupPlan = useCallback(async () => { + if (!selectedProductId) return; + setCleanupPlanLoading(true); + try { + const plan = await getDinsarProductCleanupPlan(selectedProductId); + setCleanupPlan(plan); + } catch (error) { + setCleanupPlan({ + error: error?.response?.data?.detail || error.message || '中间文件清理计划加载失败', + }); + } finally { + setCleanupPlanLoading(false); + } + }, [selectedProductId]); + useEffect(() => { loadCatalog(); }, [loadCatalog]); @@ -292,7 +358,7 @@ export default function DinsarCatalogPanel({ 手动发布与目录重建

这里用于把既有结果目录重新发布为标准结果包，并按最新规则重建目录索引。 - 如果同一对影像存在 ENVI 与 ISCE2 两套结果，它们会依赖 `engine_code` 与 `run_key` 分别登记，不会互相覆盖。 + 同一对影像的 ENVI/SARscape、LandSAR、Gamma/PyINT 结果会按任务聚合展示，底层仍依赖 `engine_code` 与 `run_key` 分别登记。

@@ -335,7 +401,7 @@ export default function DinsarCatalogPanel({
结果包列表 - {loading ? '加载中...' : `当前展示 ${products.length} 条`} + {loading ? '加载中...' : `当前展示 ${productPairs.length} 个任务`}
{queryApplied && } @@ -375,36 +441,52 @@ export default function DinsarCatalogPanel({
- {products.length === 0 ? ( + {productPairs.length === 0 ? (
{loading ? '正在加载结果包...' : '当前筛选条件下没有结果包。'}
) : (
- {products.map((item) => { + {productPairs.map((item) => { const tone = STATUS_TONE_MAP[item.status] || 'neutral'; - const engineMeta = getDinsarEngineMeta(item.engine_code); const satelliteFamily = inferSatelliteFamilyFromResultLike(item); + const rowKey = item.pair_key || `pair:${item.primary_product_id}`; return (
+ +
+
中间文件清理计划
+
+
+ +
+ +
+ {!cleanupPlan ? ( +
尚未生成清理计划。
+ ) : cleanupPlan.error ? ( +
{cleanupPlan.error}
+ ) : ( +
+
+
+ + + +
+
+ + + +
+
+ {Array.isArray(cleanupPlan.blockers) && cleanupPlan.blockers.length > 0 && ( +
+ {cleanupPlan.blockers.map((blocker) => ( +
+ {blocker} +
+ ))} +
+ )} +
+ {(cleanupPlan.candidates || []).map((candidate) => ( +
+
+ {candidate.reason} + {candidate.exists ? formatBytes(candidate.size_bytes) : '不存在'} +
+
{candidate.path}
+
+ ))} +
+
+ )} +
)} diff --git a/frontend/src/panels/BatchPanel.jsx b/frontend/src/panels/BatchPanel.jsx index 22b69e5..4a4ba41 100644 --- a/frontend/src/panels/BatchPanel.jsx +++ b/frontend/src/panels/BatchPanel.jsx @@ -2,6 +2,16 @@ import { useEffect } from 'react'; import { useBatchStore, useUiStore, useAuthStore } from '../store'; import { useI18n } from '../i18n/I18nContext'; import useBatchOperations from '../hooks/useBatchOperations'; +import { getDinsarEngineMeta } from '../utils/dinsarEngines'; + +function engineResultTone(status) { + const normalized = String(status || '').toLowerCase(); + if (normalized === 'ready') return 'ready'; + if (normalized === 'failed') return 'error'; + if (normalized === 'blocked') return 'warn'; + if (normalized === 'running') return 'info'; + return 'neutral'; +} export default function BatchPanel() { const { language } = useI18n(); @@ -124,6 +134,24 @@ export default function BatchPanel() { M: {item.master_imaging_date || '-'} / S: {item.slave_imaging_date || '-'} )} + {batchTab === 'dinsar' && ( +
+ {['sarscape', 'landsar', 'pyint'].map((engineCode) => { + const engineMeta = getDinsarEngineMeta(engineCode); + const result = item.engine_results?.[engineCode] || {}; + const status = result.status || 'missing'; + return ( + + {engineMeta.shortLabel}: {status} + + ); + })} +
+ )}