diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 229bcd3..49c0f10 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -30,6 +30,9 @@ from .orm import ( SystemWorkerHeartbeatORM, DinsarTaskBatchORM, DinsarTaskItemORM, + DinsarProductionRunORM, + DinsarProductionRunItemORM, + DinsarProductionExecutionORM, PsTaskBatchORM, PsTaskItemORM, PsTimeseriesRunORM, @@ -85,7 +88,9 @@ __all__ = [ "ManagedRootORM", "ScanCursorORM", "PathInventoryORM", "WorkflowDefORM", "WorkflowRunORM", "WorkflowStepORM", "WorkflowArtifactORM", "SystemWorkerHeartbeatORM", - "DinsarTaskBatchORM", "DinsarTaskItemORM", "PsTaskBatchORM", "PsTaskItemORM", + "DinsarTaskBatchORM", "DinsarTaskItemORM", + "DinsarProductionRunORM", "DinsarProductionRunItemORM", "DinsarProductionExecutionORM", + "PsTaskBatchORM", "PsTaskItemORM", "PsTimeseriesRunORM", "AuthUserORM", "AuthSessionORM", "AuthAuditLogORM", "AuthRateLimitORM", "SARSceneGeoORM", "FloodDetectionORM", diff --git a/backend/app/models/orm.py b/backend/app/models/orm.py index 261befa..6dadddd 100644 --- a/backend/app/models/orm.py +++ b/backend/app/models/orm.py @@ -781,6 +781,109 @@ class DinsarTaskItemORM(Base): batch = relationship("DinsarTaskBatchORM", back_populates="items") +class DinsarProductionRunORM(Base): + __tablename__ = "dinsar_production_runs" + + id = Column(Integer, primary_key=True, autoincrement=True) + run_id = Column(String(64), unique=True, index=True, nullable=False) + task_id = Column(String, index=True, nullable=True) + workflow_run_id = Column(String, index=True, nullable=True) + + engine_code = Column(String(32), index=True, nullable=False, default="sarscape") + profile_code = Column(String(64), index=True, nullable=False, default="custom6") + mode = Column(String(32), index=True, nullable=False, default="custom") + source_root = Column(String, nullable=False) + status = Column(String(32), index=True, nullable=False, default="PENDING") + cancel_requested = Column(Boolean, nullable=False, default=False) + + total_items = Column(Integer, nullable=False, default=0) + completed_items = Column(Integer, nullable=False, default=0) + failed_items = Column(Integer, nullable=False, default=0) + skipped_items = Column(Integer, nullable=False, default=0) + + latest_message = Column(Text, nullable=True) + params_json = Column(JSON, nullable=True) + summary_json = Column(JSON, nullable=True) + created_by = Column(String(128), nullable=True) + + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + started_at = Column(DateTime, nullable=True) + ended_at = Column(DateTime, nullable=True) + + items = relationship("DinsarProductionRunItemORM", back_populates="run", cascade="all, delete-orphan") + + +class DinsarProductionRunItemORM(Base): + __tablename__ = "dinsar_production_run_items" + + id = Column(Integer, primary_key=True, autoincrement=True) + run_id = Column(String(64), ForeignKey("dinsar_production_runs.run_id"), index=True, nullable=False) + + order_index = Column(Integer, nullable=False, default=0) + task_name = Column(String(255), nullable=True) + task_alias = Column(String(255), index=True, nullable=True) + pair_key = Column(String(128), index=True, nullable=True) + pair_uid = Column(String(64), index=True, nullable=True) + network_run_id = Column(String(64), index=True, nullable=True) + network_edge_id = Column(Integer, nullable=True) + policy_version = Column(String(32), index=True, nullable=True) + selection_strategy = Column(String(32), index=True, nullable=True) + + source_task_dir = Column(String, nullable=False) + results_root_dir = Column(String, nullable=False) + status = Column(String(32), index=True, nullable=False, default="PENDING") + current_step = Column(String(64), nullable=True) + attempt_count = Column(Integer, nullable=False, default=0) + latest_run_key = Column(String(128), index=True, nullable=True) + latest_output_dir = Column(String, nullable=True) + latest_manifest_path = Column(String, nullable=True) + latest_log_path = Column(String, nullable=True) + last_error = Column(Text, nullable=True) + metrics_json = Column(JSON, nullable=True) + + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + started_at = Column(DateTime, nullable=True) + ended_at = Column(DateTime, nullable=True) + + run = relationship("DinsarProductionRunORM", back_populates="items") + executions = relationship("DinsarProductionExecutionORM", back_populates="item", cascade="all, delete-orphan") + + __table_args__ = ( + Index("idx_dinsar_run_items_run_order", "run_id", "order_index"), + ) + + +class DinsarProductionExecutionORM(Base): + __tablename__ = "dinsar_production_executions" + + id = Column(Integer, primary_key=True, autoincrement=True) + execution_id = Column(String(128), unique=True, index=True, nullable=False) + run_id = Column(String(64), ForeignKey("dinsar_production_runs.run_id"), index=True, nullable=False) + item_id = Column(Integer, ForeignKey("dinsar_production_run_items.id"), index=True, nullable=False) + + run_key = Column(String(128), index=True, nullable=False) + status = Column(String(32), index=True, nullable=False, default="PENDING") + output_dir = Column(String, nullable=False) + manifest_path = Column(String, nullable=True) + log_path = Column(String, nullable=True) + subprocess_pid = Column(Integer, nullable=True) + error_message = Column(Text, nullable=True) + metrics_json = Column(JSON, nullable=True) + + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + started_at = Column(DateTime, nullable=True) + ended_at = Column(DateTime, nullable=True) + + item = relationship("DinsarProductionRunItemORM", back_populates="executions") + + __table_args__ = ( + Index("idx_dinsar_exec_run_item", "run_id", "item_id"), + ) + + class PsTaskBatchORM(Base): __tablename__ = "ps_task_batches" diff --git a/backend/app/routers/dinsar_production.py b/backend/app/routers/dinsar_production.py index f0d1e9c..e8fc778 100644 --- a/backend/app/routers/dinsar_production.py +++ b/backend/app/routers/dinsar_production.py @@ -9,19 +9,14 @@ from pydantic import BaseModel, Field from .dependencies import _get_current_user, _require_admin from ..config import read_int_env, settings +from ..database import AsyncSessionLocal from ..models import AuthUserORM -from ..services import envi_service as _envi_svc +from ..services.dinsar_production_service import dinsar_production_service from ..services.job_queue_service import job_queue_service from ..services.task_service import task_service router = APIRouter(prefix="/dinsar-production", tags=["dinsar-production"]) -DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( - "DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS", - 1, - minimum=1, - maximum=20, -) ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( "ISCE2_PRODUCTION_JOB_MAX_ATTEMPTS", 1, @@ -181,8 +176,34 @@ async def submit_run( detail=f"Engine '{req.engine_code}' does not support queued production.", ) - task_name = f"D-InSAR production: {req.engine_code}/{req.profile}" try: + if req.engine_code == "sarscape": + if AsyncSessionLocal is None: + raise RuntimeError("Database session factory is not initialized.") + async with AsyncSessionLocal() as db: + result = await dinsar_production_service.create_run( + engine_code=req.engine_code, + profile_code=req.profile, + root_dir=req.root_dir, + num_to_process=req.num_to_process, + timeout_seconds=req.timeout_seconds, + extra=req.extra, + created_by=getattr(current_user, "username", None), + db=db, + ) + return { + "task_id": result["task_id"], + "job_id": result.get("workflow_run_id"), + "run_id": result["run_id"], + "workflow_run_id": result.get("workflow_run_id"), + "job_type": job_type, + "engine_code": req.engine_code, + "profile": req.profile, + "selected_task_count": result.get("selected_task_count", 0), + "message": "Task queued.", + } + + task_name = f"D-InSAR production: {req.engine_code}/{req.profile}" task_id = await task_service.create_task( task_type=job_type, task_name=task_name, @@ -212,5 +233,8 @@ async def submit_run( @router.get("/runs") async def list_runs(limit: int = 20): - runs = await asyncio.to_thread(_envi_svc.list_recent_runs, limit) - return {"runs": runs, "total": len(runs)} + if AsyncSessionLocal is None: + raise HTTPException(status_code=500, detail="Database session factory is not initialized.") + async with AsyncSessionLocal() as db: + result = await dinsar_production_service.list_runs(db, limit=limit) + return result diff --git a/backend/app/routers/tasks.py b/backend/app/routers/tasks.py index b44d32b..6ae1196 100644 --- a/backend/app/routers/tasks.py +++ b/backend/app/routers/tasks.py @@ -10,6 +10,7 @@ from pydantic import BaseModel from ..auth_utils import verify_password from ..models import AuthUserORM, TaskInfo +from ..services.dinsar_production_service import dinsar_production_service from ..services.task_service import ( TASK_ACTIVE_DEFAULT_LIMIT, TASK_ACTIVE_MAX_LIMIT, @@ -125,5 +126,18 @@ async def force_cancel_task( raise HTTPException(status_code=404, detail="任务未找到") if task.status not in ("PENDING", "RUNNING"): raise HTTPException(status_code=400, detail="任务已结束，无需取消") + killed_pid = None + if AsyncSessionLocal is not None: + async with AsyncSessionLocal() as db: + run = await dinsar_production_service.request_cancel(task_id, db=db) + if run is not None: + killed_pid = await dinsar_production_service.kill_active_execution_by_task_id( + task_id, + db=db, + ) await task_service.update_task(task_id, status="CANCELLED", message="管理员强制取消") - return {"message": "任务已强制取消", "task_id": task_id} + return { + "message": "任务已强制取消", + "task_id": task_id, + "killed_pid": killed_pid, + } diff --git a/backend/app/services/dinsar_production_service.py b/backend/app/services/dinsar_production_service.py new file mode 100644 index 0000000..158f6ab --- /dev/null +++ b/backend/app/services/dinsar_production_service.py @@ -0,0 +1,717 @@ +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +import uuid +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import database +from ..models import ( + DinsarProductionExecutionORM, + DinsarProductionRunItemORM, + DinsarProductionRunORM, +) +from .envi_service import RUNTIME_DIR, _collect_task_folders, _resolve_dinsar_pair_identity, _to_local_path +from .task_service import task_service +from .workflow_service import workflow_service + + +TASK_TYPE_DINSAR_PRODUCTION = "IDL_RUN_DINSAR" +RUN_STATUS_PENDING = "PENDING" +RUN_STATUS_RUNNING = "RUNNING" +RUN_STATUS_COMPLETED = "COMPLETED" +RUN_STATUS_FAILED = "FAILED" +RUN_STATUS_CANCELLED = "CANCELLED" +RUN_ITEM_STATUS_PENDING = "PENDING" +RUN_ITEM_STATUS_RUNNING = "RUNNING" +RUN_ITEM_STATUS_COMPLETED = "COMPLETED" +RUN_ITEM_STATUS_FAILED = "FAILED" +RUN_ITEM_STATUS_SKIPPED = "SKIPPED" +RUN_ITEM_STATUS_CANCELLED = "CANCELLED" +EXECUTION_STATUS_PENDING = "PENDING" +EXECUTION_STATUS_RUNNING = "RUNNING" +EXECUTION_STATUS_COMPLETED = "COMPLETED" +EXECUTION_STATUS_FAILED = "FAILED" +EXECUTION_STATUS_CANCELLED = "CANCELLED" + +CURRENT_POINTER_FILENAME = "current.json" +EXECUTION_MANIFEST_FILENAME = "execution_manifest.json" +RUNS_STEP_ID = "execute_items" +RUNS_STEP_NAME = "Execute ENVI D-InSAR items" +TERMINAL_RUN_STATUSES = { + RUN_STATUS_COMPLETED, + RUN_STATUS_FAILED, + RUN_STATUS_CANCELLED, +} +TERMINAL_ITEM_STATUSES = { + RUN_ITEM_STATUS_COMPLETED, + RUN_ITEM_STATUS_FAILED, + RUN_ITEM_STATUS_SKIPPED, + RUN_ITEM_STATUS_CANCELLED, +} + + +def _new_session() -> AsyncSession: + 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 _utcnow() -> datetime: + return datetime.utcnow() + + +def _utc_text(value: Optional[datetime] = None) -> str: + stamp = value or _utcnow() + return stamp.isoformat(timespec="seconds") + "Z" + + +def _normalize_dir(path: str, label: str) -> str: + normalized = os.path.normpath(os.path.abspath(_to_local_path(path))) + if not os.path.isdir(normalized): + raise ValueError(f"{label} does not exist: {path}") + return normalized + + +def _ensure_dir(path: str) -> str: + normalized = os.path.normpath(os.path.abspath(path)) + os.makedirs(normalized, exist_ok=True) + return normalized + + +def _write_json(path: str, payload: Dict[str, Any]) -> str: + target = os.path.normpath(os.path.abspath(path)) + os.makedirs(os.path.dirname(target), exist_ok=True) + with open(target, "w", encoding="utf-8") as fp: + json.dump(payload, fp, ensure_ascii=False, indent=2) + return target + + +def _looks_like_task_dir(path: str) -> bool: + return os.path.isdir(os.path.join(path, "master")) and os.path.isdir(os.path.join(path, "slave")) + + +def _discover_run_items(root_dir: str, num_to_process: int) -> List[Dict[str, Any]]: + task_folders = [root_dir] if _looks_like_task_dir(root_dir) else _collect_task_folders(root_dir) + if num_to_process > 0: + task_folders = task_folders[:num_to_process] + + items: List[Dict[str, Any]] = [] + for order_index, folder in enumerate(task_folders, start=1): + task_name = os.path.basename(folder) + task_alias, pair_key, pair_meta = _resolve_dinsar_pair_identity(folder, task_name) + items.append( + { + "order_index": order_index, + "task_name": task_name, + "task_alias": task_alias, + "pair_key": pair_key, + "pair_uid": pair_meta.get("pair_uid") or pair_meta.get("scene_pair_uid"), + "network_run_id": pair_meta.get("network_run_id"), + "network_edge_id": pair_meta.get("network_edge_id"), + "policy_version": pair_meta.get("policy_version"), + "selection_strategy": pair_meta.get("selection_strategy"), + "source_task_dir": folder, + "results_root_dir": os.path.join(folder, "dinsar_results"), + } + ) + return items + + +def _run_log_path(run_id: str) -> str: + return os.path.join(RUNTIME_DIR, f"{run_id}.log") + + +def _append_run_log_sync(run_id: str, message: str) -> str: + _ensure_dir(RUNTIME_DIR) + log_path = _run_log_path(run_id) + line = str(message or "").rstrip() + if not line: + return log_path + with open(log_path, "a", encoding="utf-8") as fp: + fp.write(line + "\n") + return log_path + + +def _kill_process_tree_sync(pid: int) -> None: + try: + import psutil + + parent = psutil.Process(pid) + children = parent.children(recursive=True) + for child in children: + try: + child.kill() + except psutil.NoSuchProcess: + pass + try: + parent.kill() + except psutil.NoSuchProcess: + pass + psutil.wait_procs(children + [parent], timeout=10) + return + except ImportError: + pass + except Exception: + pass + + try: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(pid)], + capture_output=True, + timeout=15, + check=False, + ) + except Exception: + pass + + +def _execution_dir(item: DinsarProductionRunItemORM, run_key: str) -> str: + return os.path.join(item.results_root_dir, "runs", run_key) + + +def _current_pointer_path(item: DinsarProductionRunItemORM) -> str: + return os.path.join(item.results_root_dir, CURRENT_POINTER_FILENAME) + + +def _execution_manifest_path(execution_dir: str) -> str: + return os.path.join(execution_dir, EXECUTION_MANIFEST_FILENAME) + + +def _safe_epoch(value: Optional[datetime]) -> Optional[int]: + if value is None: + return None + return int(value.timestamp()) + + +def _public_run_status(value: str) -> str: + normalized = str(value or "").strip().upper() + if normalized == RUN_STATUS_COMPLETED: + return "success" + if normalized == RUN_STATUS_FAILED: + return "failed" + if normalized == RUN_STATUS_CANCELLED: + return "cancelled" + if normalized == RUN_STATUS_RUNNING: + return "running" + return "pending" + + +class DinsarProductionService: + async def create_run( + self, + *, + engine_code: str, + profile_code: str, + root_dir: str, + num_to_process: int, + timeout_seconds: Optional[int], + extra: Optional[Dict[str, Any]], + created_by: Optional[str], + db: AsyncSession, + ) -> Dict[str, Any]: + normalized_engine = str(engine_code or "").strip().lower() + normalized_profile = str(profile_code or "").strip() + if normalized_engine != "sarscape": + raise ValueError(f"Unsupported engine for D-InSAR production run: {engine_code}") + + normalized_root = _normalize_dir(root_dir, "root_dir") + item_payloads = await asyncio.to_thread( + _discover_run_items, + normalized_root, + max(0, int(num_to_process or 0)), + ) + if not item_payloads: + raise ValueError(f"No Task_* directories found under: {normalized_root}") + + run_id = str(uuid.uuid4()) + mode = "custom" if normalized_profile == "custom6" else "metatask" + task_name = f"D-InSAR production: {normalized_engine}/{normalized_profile}" + task_params = { + "engine_code": normalized_engine, + "profile": normalized_profile, + "root_dir": normalized_root, + "num_to_process": int(num_to_process or 0), + "timeout_seconds": timeout_seconds, + "extra": dict(extra or {}), + "mode": mode, + "production_run_id": run_id, + } + + task_id: Optional[str] = None + try: + task_id = await task_service.create_task( + task_type=TASK_TYPE_DINSAR_PRODUCTION, + task_name=task_name, + params=task_params, + db=db, + ) + + run = DinsarProductionRunORM( + run_id=run_id, + task_id=task_id, + engine_code=normalized_engine, + profile_code=normalized_profile, + mode=mode, + source_root=normalized_root, + status=RUN_STATUS_PENDING, + cancel_requested=False, + total_items=len(item_payloads), + completed_items=0, + failed_items=0, + skipped_items=0, + latest_message="Queued", + params_json=task_params, + summary_json={ + "phase": "queued", + "selected_task_count": len(item_payloads), + }, + created_by=created_by, + ) + db.add(run) + await db.flush() + + for item_payload in item_payloads: + db.add( + DinsarProductionRunItemORM( + run_id=run_id, + order_index=item_payload["order_index"], + task_name=item_payload["task_name"], + task_alias=item_payload["task_alias"], + pair_key=item_payload["pair_key"], + pair_uid=item_payload["pair_uid"], + network_run_id=item_payload["network_run_id"], + network_edge_id=item_payload["network_edge_id"], + policy_version=item_payload["policy_version"], + selection_strategy=item_payload["selection_strategy"], + source_task_dir=item_payload["source_task_dir"], + results_root_dir=item_payload["results_root_dir"], + status=RUN_ITEM_STATUS_PENDING, + ) + ) + + await db.flush() + + workflow_run_id = await workflow_service.create_run( + workflow_name="dinsar_sarscape_production", + steps=[ + { + "step_id": RUNS_STEP_ID, + "step_name": RUNS_STEP_NAME, + "job_type": TASK_TYPE_DINSAR_PRODUCTION, + "payload": {"production_run_id": run_id}, + "task_id": task_id, + "max_attempts": 1, + } + ], + params={ + "production_run_id": run_id, + "engine_code": normalized_engine, + "profile_code": normalized_profile, + "root_dir": normalized_root, + }, + tags={ + "engine_code": normalized_engine, + "profile_code": normalized_profile, + "source_root": normalized_root, + }, + created_by=created_by, + db=db, + ) + run.workflow_run_id = workflow_run_id + await db.commit() + await db.refresh(run) + except Exception as exc: + await db.rollback() + if task_id: + try: + await task_service.update_task( + task_id, + status="FAILED", + message=f"Failed to create D-InSAR production run: {exc}", + ) + except Exception: + pass + raise + + await asyncio.to_thread( + _append_run_log_sync, + run_id, + f"[queued] run_id={run_id} profile={normalized_profile} root={normalized_root} items={len(item_payloads)}", + ) + return { + "run_id": run_id, + "task_id": task_id, + "workflow_run_id": run.workflow_run_id, + "status": run.status, + "selected_task_count": len(item_payloads), + } + + async def list_runs( + self, + db: AsyncSession, + *, + limit: int = 20, + offset: int = 0, + ) -> Dict[str, Any]: + safe_limit = max(1, min(200, int(limit or 20))) + safe_offset = max(0, int(offset or 0)) + total_result = await db.execute(select(func.count(DinsarProductionRunORM.id))) + total = int(total_result.scalar_one() or 0) + stmt = ( + select(DinsarProductionRunORM) + .order_by(DinsarProductionRunORM.created_at.desc()) + .offset(safe_offset) + .limit(safe_limit) + ) + result = await db.execute(stmt) + runs = result.scalars().all() + return { + "runs": [ + { + "run_id": run.run_id, + "engine": run.engine_code, + "profile_code": run.profile_code, + "status": _public_run_status(run.status), + "raw_status": run.status, + "started_at": _safe_epoch(run.started_at or run.created_at), + "ended_at": _safe_epoch(run.ended_at), + "task_id": run.task_id, + "workflow_run_id": run.workflow_run_id, + "root_dir": run.source_root, + "message": run.latest_message, + "total_items": run.total_items, + "completed_items": run.completed_items, + "failed_items": run.failed_items, + "skipped_items": run.skipped_items, + } + for run in runs + ], + "total": total, + } + + async def get_run(self, run_id: str, db: AsyncSession) -> Optional[DinsarProductionRunORM]: + result = await db.execute( + select(DinsarProductionRunORM).where(DinsarProductionRunORM.run_id == str(run_id or "").strip()) + ) + return result.scalar_one_or_none() + + async def get_run_by_task_id(self, task_id: str, db: AsyncSession) -> Optional[DinsarProductionRunORM]: + result = await db.execute( + select(DinsarProductionRunORM).where(DinsarProductionRunORM.task_id == str(task_id or "").strip()) + ) + return result.scalar_one_or_none() + + async def list_run_items(self, run_id: str, db: AsyncSession) -> List[DinsarProductionRunItemORM]: + result = await db.execute( + select(DinsarProductionRunItemORM) + .where(DinsarProductionRunItemORM.run_id == run_id) + .order_by(DinsarProductionRunItemORM.order_index.asc(), DinsarProductionRunItemORM.id.asc()) + ) + return result.scalars().all() + + async def request_cancel(self, task_id: str, *, db: AsyncSession) -> Optional[DinsarProductionRunORM]: + run = await self.get_run_by_task_id(task_id, db) + if run is None or run.status in TERMINAL_RUN_STATUSES: + return run + run.cancel_requested = True + run.latest_message = "Cancellation requested" + await db.commit() + await asyncio.to_thread(_append_run_log_sync, run.run_id, "[cancel] cancellation requested") + return run + + async def refresh_run_counters( + self, + run: DinsarProductionRunORM, + *, + db: AsyncSession, + latest_message: Optional[str] = None, + ) -> DinsarProductionRunORM: + rows = await db.execute( + select(DinsarProductionRunItemORM.status, func.count(DinsarProductionRunItemORM.id)) + .where(DinsarProductionRunItemORM.run_id == run.run_id) + .group_by(DinsarProductionRunItemORM.status) + ) + counts = {str(status or "").upper(): int(count or 0) for status, count in rows.fetchall()} + run.completed_items = counts.get(RUN_ITEM_STATUS_COMPLETED, 0) + run.failed_items = counts.get(RUN_ITEM_STATUS_FAILED, 0) + run.skipped_items = counts.get(RUN_ITEM_STATUS_SKIPPED, 0) + if latest_message is not None: + run.latest_message = latest_message + return run + + async def mark_run_started( + self, + run: DinsarProductionRunORM, + *, + db: AsyncSession, + message: str, + ) -> None: + if run.started_at is None: + run.started_at = _utcnow() + run.status = RUN_STATUS_RUNNING + run.latest_message = message + await self.refresh_run_counters(run, db=db) + await db.commit() + + async def begin_item_execution( + self, + *, + run: DinsarProductionRunORM, + item: DinsarProductionRunItemORM, + run_key: str, + db: AsyncSession, + ) -> DinsarProductionExecutionORM: + output_dir = _execution_dir(item, run_key) + _ensure_dir(output_dir) + execution = DinsarProductionExecutionORM( + execution_id=run_key, + run_id=run.run_id, + item_id=item.id, + run_key=run_key, + status=EXECUTION_STATUS_RUNNING, + output_dir=output_dir, + log_path=_run_log_path(run.run_id), + started_at=_utcnow(), + ) + db.add(execution) + item.status = RUN_ITEM_STATUS_RUNNING + item.current_step = "queued" + item.attempt_count = int(item.attempt_count or 0) + 1 + item.last_error = None + item.latest_run_key = run_key + item.latest_output_dir = output_dir + item.latest_log_path = execution.log_path + item.started_at = item.started_at or _utcnow() + run.status = RUN_STATUS_RUNNING + run.latest_message = f"Running {item.task_alias or item.task_name}" + await db.commit() + await db.refresh(execution) + return execution + + async def set_execution_pid( + self, + execution_id: str, + pid: int, + *, + db: AsyncSession, + ) -> None: + result = await db.execute( + select(DinsarProductionExecutionORM).where(DinsarProductionExecutionORM.execution_id == execution_id) + ) + execution = result.scalar_one_or_none() + if execution is None: + return + execution.subprocess_pid = int(pid or 0) or None + await db.commit() + + async def update_item_step( + self, + item_id: int, + step_name: str, + *, + db: AsyncSession, + ) -> None: + result = await db.execute(select(DinsarProductionRunItemORM).where(DinsarProductionRunItemORM.id == item_id)) + item = result.scalar_one_or_none() + if item is None: + return + item.current_step = str(step_name or "").strip() or None + await db.commit() + + async def mark_item_completed( + self, + *, + run: DinsarProductionRunORM, + item: DinsarProductionRunItemORM, + execution: DinsarProductionExecutionORM, + manifest_path: str, + metrics: Optional[Dict[str, Any]], + db: AsyncSession, + ) -> None: + now = _utcnow() + execution.status = EXECUTION_STATUS_COMPLETED + execution.manifest_path = manifest_path + execution.metrics_json = metrics or {} + execution.ended_at = now + item.status = RUN_ITEM_STATUS_COMPLETED + item.current_step = "completed" + item.latest_manifest_path = manifest_path + item.metrics_json = metrics or {} + item.ended_at = now + item.last_error = None + await self.refresh_run_counters( + run, + db=db, + latest_message=f"Completed {item.task_alias or item.task_name}", + ) + await db.commit() + + async def mark_item_failed( + self, + *, + run: DinsarProductionRunORM, + item: DinsarProductionRunItemORM, + execution: DinsarProductionExecutionORM, + error_message: str, + db: AsyncSession, + ) -> None: + now = _utcnow() + execution.status = EXECUTION_STATUS_FAILED + execution.error_message = error_message + execution.ended_at = now + item.status = RUN_ITEM_STATUS_FAILED + item.current_step = "failed" + item.last_error = error_message + item.ended_at = now + await self.refresh_run_counters( + run, + db=db, + latest_message=f"Failed {item.task_alias or item.task_name}: {error_message}", + ) + await db.commit() + + async def mark_item_cancelled( + self, + *, + run: DinsarProductionRunORM, + item: DinsarProductionRunItemORM, + execution: DinsarProductionExecutionORM, + error_message: str, + db: AsyncSession, + ) -> None: + now = _utcnow() + execution.status = EXECUTION_STATUS_CANCELLED + execution.error_message = error_message + execution.ended_at = now + item.status = RUN_ITEM_STATUS_CANCELLED + item.current_step = "cancelled" + item.last_error = error_message + item.ended_at = now + run.status = RUN_STATUS_CANCELLED + run.cancel_requested = True + await self.refresh_run_counters(run, db=db, latest_message=error_message) + await db.commit() + + async def finalize_run( + self, + run: DinsarProductionRunORM, + *, + db: AsyncSession, + status: str, + summary_payload: Dict[str, Any], + latest_message: str, + ) -> None: + run.status = status + run.summary_json = summary_payload + run.latest_message = latest_message + run.ended_at = _utcnow() + await self.refresh_run_counters(run, db=db, latest_message=latest_message) + await db.commit() + + def append_run_log(self, run_id: str, message: str) -> str: + return _append_run_log_sync(run_id, message) + + def build_execution_manifest( + self, + *, + run: DinsarProductionRunORM, + item: DinsarProductionRunItemORM, + execution: DinsarProductionExecutionORM, + primary_file: str, + source_files: List[str], + metrics: Optional[Dict[str, Any]], + ) -> str: + manifest_payload = { + "format_version": 1, + "run_id": run.run_id, + "run_key": execution.run_key, + "task_id": run.task_id, + "engine_code": run.engine_code, + "profile_code": run.profile_code, + "mode": run.mode, + "task_name": item.task_name, + "task_alias": item.task_alias, + "pair_key": item.pair_key, + "pair_uid": item.pair_uid, + "network_run_id": item.network_run_id, + "network_edge_id": item.network_edge_id, + "policy_version": item.policy_version, + "selection_strategy": item.selection_strategy, + "source_root": run.source_root, + "source_task_dir": item.source_task_dir, + "output_dir": execution.output_dir, + "primary_file": primary_file, + "source_files": source_files, + "status": EXECUTION_STATUS_COMPLETED, + "metrics": metrics or {}, + "created_at": _utc_text(execution.started_at or _utcnow()), + "finished_at": _utc_text(), + } + manifest_path = _execution_manifest_path(execution.output_dir) + return _write_json(manifest_path, manifest_payload) + + def write_current_pointer( + self, + *, + item: DinsarProductionRunItemORM, + execution: DinsarProductionExecutionORM, + manifest_path: str, + primary_file: str, + source_files: List[str], + ) -> str: + pointer_payload = { + "format_version": 1, + "run_key": execution.run_key, + "execution_id": execution.execution_id, + "status": EXECUTION_STATUS_COMPLETED, + "output_dir": execution.output_dir, + "manifest_path": manifest_path, + "primary_file": primary_file, + "source_files": source_files, + "updated_at": _utc_text(), + } + return _write_json(_current_pointer_path(item), pointer_payload) + + async def get_active_execution_by_task_id( + self, + task_id: str, + *, + db: AsyncSession, + ) -> Optional[DinsarProductionExecutionORM]: + run = await self.get_run_by_task_id(task_id, db) + if run is None: + return None + result = await db.execute( + select(DinsarProductionExecutionORM) + .where( + DinsarProductionExecutionORM.run_id == run.run_id, + DinsarProductionExecutionORM.status == EXECUTION_STATUS_RUNNING, + ) + .order_by(DinsarProductionExecutionORM.started_at.desc(), DinsarProductionExecutionORM.id.desc()) + ) + return result.scalars().first() + + async def kill_active_execution_by_task_id( + self, + task_id: str, + *, + db: AsyncSession, + ) -> Optional[int]: + execution = await self.get_active_execution_by_task_id(task_id, db=db) + if execution is None or not execution.subprocess_pid: + return None + pid = int(execution.subprocess_pid) + await asyncio.to_thread(_kill_process_tree_sync, pid) + return pid + + +dinsar_production_service = DinsarProductionService() diff --git a/backend/app/services/engine_lock_service.py b/backend/app/services/engine_lock_service.py new file mode 100644 index 0000000..e893e9a --- /dev/null +++ b/backend/app/services/engine_lock_service.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import asyncio +import hashlib +import time +from contextlib import asynccontextmanager +from typing import AsyncIterator, Optional + +from sqlalchemy import text + +from .. import database + + +def _resource_lock_key(resource_name: str) -> int: + normalized = str(resource_name or "").strip().lower().encode("utf-8") + digest = hashlib.sha256(normalized).digest() + return int.from_bytes(digest[:8], byteorder="big", signed=True) + + +class EngineLockService: + async def _ensure_engine(self): + if database.engine is None: + database.init_db() + if database.engine is None: + raise RuntimeError("Database engine is not initialized.") + return database.engine + + @asynccontextmanager + async def acquire( + self, + resource_name: str, + *, + poll_interval: float = 2.0, + timeout_seconds: Optional[float] = None, + cancel_event: Optional[asyncio.Event] = None, + ) -> AsyncIterator[None]: + engine = await self._ensure_engine() + lock_key = _resource_lock_key(resource_name) + started = time.monotonic() + conn = await engine.connect() + + try: + acquired = False + while not acquired: + if cancel_event is not None and cancel_event.is_set(): + raise RuntimeError(f"Lock acquisition cancelled: {resource_name}") + + result = await conn.execute( + text("SELECT pg_try_advisory_lock(:lock_key)"), + {"lock_key": lock_key}, + ) + acquired = bool(result.scalar()) + if acquired: + break + + if timeout_seconds is not None and (time.monotonic() - started) >= float(timeout_seconds): + raise TimeoutError(f"Timeout waiting for engine resource lock: {resource_name}") + + await asyncio.sleep(max(0.5, float(poll_interval))) + + try: + yield + finally: + await conn.execute( + text("SELECT pg_advisory_unlock(:lock_key)"), + {"lock_key": lock_key}, + ) + finally: + await conn.close() + + +engine_lock_service = EngineLockService() diff --git a/backend/app/services/envi_extract.py b/backend/app/services/envi_extract.py index 4d5791d..c174487 100644 --- a/backend/app/services/envi_extract.py +++ b/backend/app/services/envi_extract.py @@ -1,6 +1,7 @@ """D-InSAR result extraction and task overview helpers.""" from __future__ import annotations +import json import os import re import shutil @@ -90,6 +91,29 @@ def _find_envi_task_result(task_dir: str) -> Optional[Dict[str, Any]]: if not os.path.isdir(dinsar_results_dir): return None + current_pointer_path = os.path.join(dinsar_results_dir, "current.json") + if os.path.isfile(current_pointer_path): + try: + with open(current_pointer_path, "r", encoding="utf-8") as fp: + pointer = json.load(fp) + primary_file = str(pointer.get("primary_file") or "").strip() + source_files = [ + os.path.normpath(os.path.abspath(path)) + for path in (pointer.get("source_files") or []) + if str(path or "").strip() + ] + if primary_file and os.path.isfile(primary_file): + if not source_files: + source_files = [os.path.normpath(os.path.abspath(primary_file))] + return { + "engine": "envi", + "task_name": os.path.basename(os.path.normpath(task_dir)), + "source_dir": os.path.dirname(primary_file), + "source_files": source_files, + } + except Exception: + pass + candidates: List[Tuple[str, str]] = [] try: for entry in os.scandir(dinsar_results_dir): diff --git a/backend/app/services/envi_runner_cli.py b/backend/app/services/envi_runner_cli.py index 163c914..2ef6a9c 100644 --- a/backend/app/services/envi_runner_cli.py +++ b/backend/app/services/envi_runner_cli.py @@ -28,10 +28,15 @@ def _parse_args() -> argparse.Namespace: required=True, choices=["import", "dinsar", "dinsar_custom"], ) - parser.add_argument("--root-dir", required=True) + parser.add_argument("--root-dir", required=False) + parser.add_argument("--task-dir", required=False) + parser.add_argument("--output-dir", required=False) + parser.add_argument("--source-root", required=False) parser.add_argument("--num-to-process", type=int, default=0) parser.add_argument("--timeout-seconds", type=int, default=None) parser.add_argument("--job-id", type=str, default=None) + parser.add_argument("--run-key", type=str, default=None) + parser.add_argument("--profile-code", type=str, default=None) return parser.parse_args() @@ -39,15 +44,31 @@ def main() -> int: ensure_project_env_loaded() args = _parse_args() try: - from .envi_service import run_workflow + from .envi_service import run_single_task_workflow, run_workflow - record = run_workflow( - workflow=args.workflow, - root_dir=args.root_dir, - num_to_process=args.num_to_process, - timeout=args.timeout_seconds or 14400, - job_id=args.job_id, - ) + if args.task_dir: + if not args.output_dir: + raise ValueError("--output-dir is required when --task-dir is used.") + record = run_single_task_workflow( + workflow=args.workflow, + task_dir=args.task_dir, + output_dir=args.output_dir, + source_root=args.source_root, + timeout=args.timeout_seconds or 14400, + job_id=args.job_id, + run_key=args.run_key, + profile_code=args.profile_code, + ) + else: + if not args.root_dir: + raise ValueError("--root-dir is required when --task-dir is not used.") + record = run_workflow( + workflow=args.workflow, + root_dir=args.root_dir, + num_to_process=args.num_to_process, + timeout=args.timeout_seconds or 14400, + job_id=args.job_id, + ) print(json.dumps(record, ensure_ascii=False)) return 0 except Exception as exc: diff --git a/backend/app/services/envi_service.py b/backend/app/services/envi_service.py index d238e31..ccf7fd9 100644 --- a/backend/app/services/envi_service.py +++ b/backend/app/services/envi_service.py @@ -14,7 +14,7 @@ import defusedxml.ElementTree as ET from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError from datetime import datetime from glob import glob -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional from ..config import get_env_text, settings from ..process_utils import is_any_process_running @@ -1592,6 +1592,189 @@ def run_dinsar_custom_workflow( } +# --------------------------------------------------------------------------- +# Single-task D-InSAR workflow entrypoints +# --------------------------------------------------------------------------- + +def run_single_task_workflow( + workflow: str, + task_dir: str, + output_dir: str, + *, + source_root: Optional[str] = None, + timeout: int = DEFAULT_TIMEOUT, + timeout_seconds: Optional[int] = None, + job_id: Optional[str] = None, + run_key: Optional[str] = None, + profile_code: Optional[str] = None, + started_at: Optional[datetime] = None, +) -> Dict[str, Any]: + """Run one Task_* directory into an isolated output directory.""" + normalized_workflow = str(workflow or "").strip().lower() + if normalized_workflow not in {"dinsar", "dinsar_custom"}: + raise ValueError(f"Unsupported single-task workflow: {workflow}") + + timeout = min(timeout_seconds or timeout or DEFAULT_TIMEOUT, MAX_TIMEOUT) + task_dir = _to_local_path(task_dir) + output_dir = _to_local_path(output_dir) + source_root = _to_local_path(source_root or os.path.dirname(task_dir)) + run_started_at = started_at or datetime.utcnow() + run_started_at_text = run_started_at.isoformat(timespec="seconds") + "Z" + resolved_profile_code = profile_code or ("custom6" if normalized_workflow == "dinsar_custom" else "metatask") + resolved_run_key = run_key or build_run_key("sarscape", resolved_profile_code, started_at=run_started_at) + task_name = os.path.basename(os.path.normpath(task_dir)) + task_alias, pair_key, pair_meta = _resolve_dinsar_pair_identity(task_dir, task_name) + + if not task_dir or not os.path.isdir(task_dir): + raise ValueError(f"D-InSAR task directory does not exist: {task_dir}") + if not output_dir: + raise ValueError("output_dir must not be empty.") + if not DEM_BASE_FILE: + raise ValueError("DEM path not configured. Set IDL_DINSAR_DEM_BASE_FILE in .env") + + master_dir = os.path.join(task_dir, "master") + slave_dir = os.path.join(task_dir, "slave") + if not os.path.isdir(master_dir) or not os.path.isdir(slave_dir): + raise RuntimeError(f"{task_name}: master/slave dir missing") + + os.makedirs(output_dir, exist_ok=True) + log_lines: List[str] = [ + f"[envi] single task workflow={normalized_workflow}", + f"[envi] source_root={source_root}", + f"[envi] task_dir={task_dir}", + f"[envi] output_dir={output_dir}", + f"[envi] dem={DEM_BASE_FILE}", + ] + auto_imported = 0 + + for side, side_dir in [("master", master_dir), ("slave", slave_dir)]: + if _has_sml(side_dir): + continue + meta_files = _find_meta_files(side_dir) + if not meta_files: + log_lines.append(f"[warn] {task_name}/{side}: no .sml and no .meta.xml") + continue + log_lines.append(f"[auto-import] {task_name}/{side}: importing {len(meta_files)} file(s)") + for meta_file in meta_files: + start = time.time() + try: + execute_envi_task( + "SARsImportLuTan1", + { + "INPUT_FILE_LIST": [meta_file], + "ROOT_URI_FOR_OUTPUT": side_dir, + }, + ) + elapsed = round(time.time() - start, 1) + auto_imported += 1 + log_lines.append(f"[auto-import ok] {task_name}/{side}: {os.path.basename(meta_file)} ({elapsed}s)") + except Exception as exc: + elapsed = round(time.time() - start, 1) + log_lines.append(f"[auto-import err] {task_name}/{side}: {os.path.basename(meta_file)} ({elapsed}s): {exc}") + + master_base = _first_sml_base(master_dir) + slave_base = _first_sml_base(slave_dir) + if not master_base or not slave_base: + raise RuntimeError( + f"{task_name}: missing .sml after import " + f"(master={'yes' if master_base else 'no'} slave={'yes' if slave_base else 'no'})" + ) + + start = time.time() + if normalized_workflow == "dinsar": + _write_progress(job_id, 1, 1, "Metatask D-InSAR", output_dir, 1, 1, task_name) + execute_envi_task( + "SARsMetataskInSARDisplacementGeneration", + { + "REFERENCE_SARSCAPEDATA": _build_sarscapedata(master_base), + "SECONDARY_SARSCAPEDATA": _build_sarscapedata(slave_base), + "DEM_SARSCAPEDATA": _build_sarscapedata(DEM_BASE_FILE), + "OUTPUT_FOLDER": _normalize_path(output_dir), + }, + ) + _write_progress(job_id, 1, 1, "Completed", output_dir, 1, 1, task_name) + else: + success = _run_dinsar_custom_single( + master_base, + slave_base, + DEM_BASE_FILE, + os.path.join(output_dir, "workflow"), + log_lines, + job_id=job_id, + pair_index=1, + total_pairs=1, + pair_name=task_name, + ) + if not success: + detail = "\n".join(log_lines[-20:]) + raise RuntimeError(f"{task_name}: custom D-InSAR workflow failed.\n{detail}") + + elapsed = round(time.time() - start, 1) + log_lines.append(f"[ok] {normalized_workflow} {task_name} ({elapsed}s)") + _write_envi_run_sidecar( + output_dir, + engine_code="sarscape", + profile_code=resolved_profile_code, + root_dir=source_root, + task_dir=task_dir, + task_name=task_name, + task_alias=task_alias, + pair_key=pair_key, + run_key=resolved_run_key, + pair_meta=pair_meta, + started_at=run_started_at_text, + params=( + { + "timeout_seconds": timeout, + "workflow": "custom6", + "dem_base_file": DEM_BASE_FILE, + "target_resolution_m": CUSTOM_TARGET_RESOLUTION_M, + "filter_method": CUSTOM_FILTER_METHOD, + "unwrap_coh_threshold": CUSTOM_UNWRAP_COH_THRESHOLD, + "gcp_coh_threshold": CUSTOM_GCP_COH_THRESHOLD, + "gcp_number": CUSTOM_GCP_NUMBER, + "geocoding_coh_threshold": CUSTOM_GEOCODING_COH_THRESHOLD, + "geocoding_pixel_size_m": CUSTOM_GEOCODING_PIXEL_SIZE_M, + } + if normalized_workflow == "dinsar_custom" + else { + "timeout_seconds": timeout, + "workflow": "metatask", + "dem_base_file": DEM_BASE_FILE, + } + ), + metrics={ + "elapsed_seconds": elapsed, + }, + ) + return { + "summary": { + "task_folders": 1, + "processed": 1, + "failed": 0, + "skipped": 0, + "auto_imported": auto_imported, + }, + "log_lines": log_lines, + "task_results": [ + { + "task_name": task_name, + "task_alias": task_alias, + "pair_key": pair_key, + "run_key": resolved_run_key, + "task_dir": task_dir, + "output_dir": output_dir, + "success": True, + "status": "ok", + "elapsed_seconds": elapsed, + } + ], + "output_dirs": [output_dir], + "run_key": resolved_run_key, + "profile_code": resolved_profile_code, + } + + # --------------------------------------------------------------------------- # Inspect (pre-check) functions # --------------------------------------------------------------------------- diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index ffc6ae3..b2e024d 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -5,10 +5,12 @@ import io import json import logging import os +import re import subprocess import sys import tempfile import time +import uuid from typing import Callable, Awaitable, Optional, Dict, Any, List logger = logging.getLogger(__name__) @@ -21,8 +23,11 @@ from ..models import SystemJobORM, DinsarResultORM, HazardPointORM, DinsarTaskIt from ..scheduler import scan_data_job from .data_service import data_service from .dinsar_compat_service import dinsar_compat_service +from .dinsar_naming import build_run_key +from .dinsar_production_service import dinsar_production_service from .dinsar_read_service import dinsar_read_service from .dinsar_scan_service import dinsar_scan_service +from .engine_lock_service import engine_lock_service from .psinsar_catalog_service import psinsar_catalog_service from .result_catalog_service import result_catalog_service from .task_service import task_service @@ -924,6 +929,258 @@ def _kill_process_tree(pid: int) -> None: print(f"[WARN] _kill_process_tree: pid={pid} — {exc}") +_ENVI_RESULT_NAME_RE = re.compile(r"^.+_rsp_disp$", re.IGNORECASE) + + +def _format_envi_keepalive(progress: Optional[Dict[str, Any]]) -> tuple[str, Optional[int]]: + if not progress: + return "ENVI processing...", None + + step = progress.get("step", 0) + total = progress.get("total_steps", 6) + step_msg = progress.get("message", "") + pair_index = progress.get("pair_index", 0) + total_pairs = progress.get("total_pairs", 0) + pair_name = progress.get("pair_name", "") + + step_part = f"Step {step}/{total}: {step_msg}" if step_msg else f"Step {step}/{total}" + if total_pairs > 0 and pair_index > 0: + pair_part = f"Pair {pair_index}/{total_pairs}" + if pair_name: + pair_part += f" ({pair_name})" + message = f"{pair_part} | {step_part}" + else: + message = step_part + + progress_value: Optional[int] = None + if isinstance(step, (int, float)) and isinstance(total, (int, float)) and total > 0: + if total_pairs > 0 and pair_index > 0: + pair_frac = (pair_index - 1 + step / total) / total_pairs + progress_value = min(90, 10 + int(80 * pair_frac)) + else: + progress_value = min(90, 10 + int(80 * step / total)) + return message, progress_value + + +def _clear_envi_progress_file(job_id: Optional[str]) -> None: + if not job_id: + return + try: + progress_path = _get_envi_progress_file(job_id) + if os.path.isfile(progress_path): + os.remove(progress_path) + except OSError: + pass + + +def _find_latest_envi_result(output_dir: str) -> Dict[str, Any]: + matches: List[tuple[float, str]] = [] + try: + for entry in os.scandir(output_dir): + if not entry.is_file(): + continue + if entry.name.lower().endswith((".hdr", ".sml")): + continue + if not _ENVI_RESULT_NAME_RE.match(entry.name): + continue + try: + stat = entry.stat() + matches.append((max(stat.st_mtime, stat.st_ctime), entry.path)) + except OSError: + matches.append((0.0, entry.path)) + except OSError as exc: + raise RuntimeError(f"Failed to scan ENVI output directory: {output_dir}: {exc}") from exc + + if not matches: + raise RuntimeError(f"No ENVI displacement result found under: {output_dir}") + + matches.sort(key=lambda item: (item[0], item[1]), reverse=True) + primary_file = matches[0][1] + source_files = [primary_file] + for ext in (".hdr", ".sml"): + sidecar = primary_file + ext + if os.path.isfile(sidecar): + source_files.append(sidecar) + return { + "primary_file": primary_file, + "source_files": source_files, + } + + +async def _run_envi_runner_command( + job: SystemJobORM, + runner_cmd: List[str], + *, + absolute_timeout_seconds: int, + keepalive_formatter: Optional[Callable[[Optional[Dict[str, Any]]], tuple[str, Optional[int]]]] = None, + register_pid: Optional[Callable[[int], Awaitable[None]]] = None, +) -> Dict[str, Any]: + if not job.task_id: + raise ValueError(f"{job.job_type} requires task_id for progress tracking.") + + formatter = keepalive_formatter or _format_envi_keepalive + progress_file = _get_envi_progress_file(job.job_id) + pid_ready = asyncio.Event() + proc_state: Dict[str, Any] = {} + loop = asyncio.get_running_loop() + + async def _task_keepalive(): + while True: + await asyncio.sleep(30) + try: + progress = _read_progress_file(progress_file) + message, progress_value = formatter(progress) + await task_service.update_task(job.task_id, message=message, progress=progress_value) + except Exception as keepalive_exc: + print(f"[keepalive] WARNING: failed to update task {job.task_id}: {keepalive_exc}") + + def _run_with_monitoring() -> Dict[str, Any]: + stdout_fd = None + stderr_fd = None + stdout_path = None + stderr_path = None + proc = None + try: + stdout_fd, stdout_path = tempfile.mkstemp(suffix="_stdout.txt") + stderr_fd, stderr_path = tempfile.mkstemp(suffix="_stderr.txt") + + proc = subprocess.Popen( + runner_cmd, + stdout=stdout_fd, + stderr=stderr_fd, + cwd=type(settings).PROJECT_ROOT, + ) + proc_state["pid"] = proc.pid + loop.call_soon_threadsafe(pid_ready.set) + + os.close(stdout_fd) + os.close(stderr_fd) + + absolute_start = time.time() + last_activity = time.time() + last_step_msg = "" + + while proc.poll() is None: + time.sleep(_ENVI_MONITOR_INTERVAL) + + now = time.time() + progress = _read_progress_file(progress_file) + if progress: + ts = progress.get("timestamp", 0) + if ts > last_activity: + last_activity = ts + msg = progress.get("message", "") + if msg and msg != last_step_msg: + last_step_msg = msg + + output_dir = "" + if progress and progress.get("output_dir"): + output_dir = progress["output_dir"] + + if output_dir: + dir_mtime = _scan_latest_mtime(output_dir) + if dir_mtime and dir_mtime > last_activity: + last_activity = dir_mtime + + if (now - last_activity) > _ENVI_FILE_STALE_SECONDS: + _kill_process_tree(proc.pid) + try: + proc.wait(timeout=15) + except Exception as exc: + print(f"[WARN] stale kill: proc.wait timeout — {exc}") + raise RuntimeError( + f"ENVI subprocess stale: no file activity for " + f"{int(now - last_activity)}s " + f"(threshold={_ENVI_FILE_STALE_SECONDS}s). " + f"Last step: {last_step_msg}" + ) + + if (now - absolute_start) > absolute_timeout_seconds: + _kill_process_tree(proc.pid) + try: + proc.wait(timeout=15) + except Exception: + pass + raise RuntimeError( + f"ENVI subprocess exceeded absolute timeout: " + f"{int(now - absolute_start)}s > {absolute_timeout_seconds}s. " + f"Last step: {last_step_msg}" + ) + + output_dir = "" + progress = _read_progress_file(progress_file) + if progress and progress.get("output_dir"): + output_dir = progress["output_dir"] + + if output_dir: + stable_count = 0 + prev_sizes: Dict[str, int] = {} + wait_start = time.time() + while stable_count < _ENVI_STABILITY_ROUNDS: + if (time.time() - wait_start) > _ENVI_STABILITY_MAX_WAIT: + break + time.sleep(_ENVI_STABILITY_CHECK_INTERVAL) + cur_sizes = _collect_file_sizes(output_dir) + if cur_sizes == prev_sizes: + stable_count += 1 + else: + stable_count = 0 + prev_sizes = cur_sizes + + with open(stdout_path, "r", encoding="utf-8", errors="replace") as fp: + stdout = fp.read() + with open(stderr_path, "r", encoding="utf-8", errors="replace") as fp: + stderr = fp.read() + finally: + for fd in (stdout_fd, stderr_fd): + if isinstance(fd, int): + try: + os.close(fd) + except OSError: + pass + for path in (stdout_path, stderr_path): + if path: + try: + os.unlink(path) + except OSError: + pass + + if proc is None: + raise RuntimeError("ENVI runner subprocess failed to start.") + if proc.returncode != 0: + raise RuntimeError( + "ENVI runner subprocess failed. " + f"rc={proc.returncode} " + f"stderr={(stderr or '').strip()[:1200]!r}" + ) + output = (stdout or "").strip() + if not output: + raise RuntimeError("ENVI runner subprocess returned empty output.") + try: + return json.loads(output.splitlines()[-1]) + except Exception as exc: + raise RuntimeError(f"ENVI runner returned non-JSON output: {output[:1200]!r}") from exc + + keepalive_task = asyncio.create_task(_task_keepalive()) + runner_task = asyncio.create_task(asyncio.to_thread(_run_with_monitoring)) + try: + if register_pid is not None: + try: + await asyncio.wait_for(pid_ready.wait(), timeout=30) + pid_value = int(proc_state.get("pid") or 0) + if pid_value > 0: + await register_pid(pid_value) + except asyncio.TimeoutError: + pass + return await runner_task + finally: + keepalive_task.cancel() + try: + await keepalive_task + except asyncio.CancelledError: + pass + + async def _run_envi_workflow_job( job: SystemJobORM, workflow: str, @@ -1215,23 +1472,359 @@ async def _run_envi_workflow_job( ) +async def _run_dinsar_production_controller(job: SystemJobORM) -> None: + if not job.task_id: + raise ValueError("IDL_RUN_DINSAR production controller requires task_id.") + + payload = job.payload or {} + production_run_id = str(payload.get("production_run_id") or "").strip() + if not production_run_id: + raise ValueError("IDL_RUN_DINSAR production controller requires production_run_id.") + + async with AsyncSessionLocal() as db: + run = await dinsar_production_service.get_run(production_run_id, db) + if run is None: + raise ValueError(f"D-InSAR production run not found: {production_run_id}") + + items = await dinsar_production_service.list_run_items(run.run_id, db) + if not items: + raise ValueError(f"D-InSAR production run has no items: {production_run_id}") + + workflow = "dinsar_custom" if str(run.mode or "").strip().lower() == "custom" else "dinsar" + params = run.params_json or {} + timeout_seconds_raw = params.get("timeout_seconds") + timeout_seconds = int(timeout_seconds_raw) if timeout_seconds_raw not in (None, "") else None + absolute_timeout_seconds = max(_ENVI_PER_TASK_TIMEOUT, int(timeout_seconds or 0)) if timeout_seconds else _ENVI_PER_TASK_TIMEOUT + total_items = len(items) + run_log = dinsar_production_service.append_run_log + + async def _refresh_cancel_state() -> bool: + await db.refresh(run) + current_task = await task_service.get_task(job.task_id) + task_cancelled = bool(current_task and current_task.status == "CANCELLED") + if task_cancelled and not run.cancel_requested: + run.cancel_requested = True + await db.commit() + return bool(run.cancel_requested or task_cancelled) + + await task_service.start_task( + job.task_id, + message=f"Starting ENVI D-InSAR production run {run.run_id} ({total_items} items)...", + ) + await task_service.add_log( + job.task_id, + "INFO", + ( + f"D-InSAR production controller started. run_id={run.run_id} " + f"profile={run.profile_code} mode={run.mode} items={total_items}" + ), + ) + await dinsar_production_service.mark_run_started( + run, + db=db, + message=f"Preparing {total_items} D-InSAR item(s)", + ) + run_log(run.run_id, f"[start] workflow={workflow} items={total_items} source_root={run.source_root}") + + successful_output_dirs: List[str] = [] + async with engine_lock_service.acquire("envi_taskengine"): + await task_service.update_task( + job.task_id, + progress=5, + message=f"ENVI engine acquired. Preparing {total_items} item(s)...", + ) + for item_index, item in enumerate(items, start=1): + if await _refresh_cancel_state(): + await task_service.add_log( + job.task_id, + "WARNING", + f"Cancellation detected before item {item.task_alias or item.task_name}.", + ) + break + + await db.refresh(item) + if str(item.status or "").upper() in {"COMPLETED", "FAILED", "SKIPPED", "CANCELLED"}: + if item.status == "COMPLETED" and item.latest_output_dir and os.path.isdir(item.latest_output_dir): + successful_output_dirs.append(item.latest_output_dir) + continue + + run_key = ( + f"{build_run_key('sarscape', run.profile_code, started_at=datetime.utcnow())}" + f"_{item.id}_{uuid.uuid4().hex[:6]}" + ) + execution = await dinsar_production_service.begin_item_execution( + run=run, + item=item, + run_key=run_key, + db=db, + ) + + runner_cmd = [ + sys.executable, + "-m", + "backend.app.services.envi_runner_cli", + "--workflow", + workflow, + "--task-dir", + str(item.source_task_dir), + "--output-dir", + str(execution.output_dir), + "--source-root", + str(run.source_root), + "--job-id", + str(job.job_id), + "--run-key", + str(run_key), + "--profile-code", + str(run.profile_code), + ] + if timeout_seconds is not None: + runner_cmd.extend(["--timeout-seconds", str(timeout_seconds)]) + + def _keepalive_formatter(progress: Optional[Dict[str, Any]]) -> tuple[str, Optional[int]]: + message, progress_value = _format_envi_keepalive(progress) + prefix = f"{item_index}/{total_items} {item.task_alias or item.task_name}: " + if progress_value is None: + return prefix + message, None + local_fraction = max(0.0, min(1.0, progress_value / 100.0)) + overall = ((item_index - 1) + local_fraction) / max(1, total_items) + return prefix + message, min(98, max(1, int(overall * 100))) + + async def _register_pid(pid: int) -> None: + async with AsyncSessionLocal() as pid_db: + await dinsar_production_service.set_execution_pid( + execution.execution_id, + pid, + db=pid_db, + ) + + await task_service.add_log( + job.task_id, + "INFO", + ( + f"[{item_index}/{total_items}] Launching {item.task_alias or item.task_name} " + f"-> {execution.output_dir}" + ), + ) + run_log( + run.run_id, + ( + f"[item-start] {item_index}/{total_items} " + f"{item.task_alias or item.task_name} run_key={run_key} output={execution.output_dir}" + ), + ) + + try: + run_meta = await _run_envi_runner_command( + job, + runner_cmd, + absolute_timeout_seconds=absolute_timeout_seconds, + keepalive_formatter=_keepalive_formatter, + register_pid=_register_pid, + ) + result_files = await asyncio.to_thread(_find_latest_envi_result, execution.output_dir) + metrics = { + "duration_seconds": run_meta.get("duration_seconds"), + "summary": run_meta.get("summary") or {}, + } + if run_meta.get("task_results"): + metrics["task_result"] = run_meta["task_results"][0] + + manifest_path = await asyncio.to_thread( + dinsar_production_service.build_execution_manifest, + run=run, + item=item, + execution=execution, + primary_file=result_files["primary_file"], + source_files=result_files["source_files"], + metrics=metrics, + ) + await asyncio.to_thread( + dinsar_production_service.write_current_pointer, + item=item, + execution=execution, + manifest_path=manifest_path, + primary_file=result_files["primary_file"], + source_files=result_files["source_files"], + ) + await dinsar_production_service.mark_item_completed( + run=run, + item=item, + execution=execution, + manifest_path=manifest_path, + metrics=metrics, + db=db, + ) + successful_output_dirs.append(execution.output_dir) + await task_service.add_log( + job.task_id, + "INFO", + f"[{item_index}/{total_items}] Completed {item.task_alias or item.task_name}", + ) + run_log( + run.run_id, + f"[item-ok] {item_index}/{total_items} {item.task_alias or item.task_name}", + ) + except Exception as exc: + cancelled = await _refresh_cancel_state() + if cancelled: + await dinsar_production_service.mark_item_cancelled( + run=run, + item=item, + execution=execution, + error_message=f"Cancelled while processing {item.task_alias or item.task_name}", + db=db, + ) + await task_service.add_log( + job.task_id, + "WARNING", + f"[{item_index}/{total_items}] Cancelled {item.task_alias or item.task_name}: {exc}", + ) + run_log( + run.run_id, + f"[item-cancelled] {item_index}/{total_items} {item.task_alias or item.task_name}: {exc}", + ) + break + + error_message = str(exc) + await dinsar_production_service.mark_item_failed( + run=run, + item=item, + execution=execution, + error_message=error_message, + db=db, + ) + await task_service.add_log( + job.task_id, + "WARNING", + f"[{item_index}/{total_items}] Failed {item.task_alias or item.task_name}: {error_message}", + ) + run_log( + run.run_id, + f"[item-failed] {item_index}/{total_items} {item.task_alias or item.task_name}: {error_message}", + ) + finally: + _clear_envi_progress_file(job.job_id) + + publish_result = None + publish_error = None + publish_dirs = _dedupe_existing_dirs(successful_output_dirs) + if publish_dirs: + await task_service.update_task( + job.task_id, + progress=99, + message=f"Publishing {len(publish_dirs)} successful result package(s)...", + ) + try: + publish_result = await result_catalog_service.publish_from_sources(db, publish_dirs) + await task_service.add_log( + job.task_id, + "INFO", + f"Published {publish_result.get('processed', 0)} result package(s).", + ) + run_log( + run.run_id, + f"[publish] processed={publish_result.get('processed', 0)} failed={publish_result.get('failed', 0)}", + ) + except Exception as exc: + publish_error = str(exc) + await task_service.add_log( + job.task_id, + "WARNING", + f"Result catalog publish failed: {publish_error}", + ) + run_log(run.run_id, f"[publish-failed] {publish_error}") + + cancelled = await _refresh_cancel_state() + await db.refresh(run) + latest_message = "" + final_status = "COMPLETED" + if publish_error: + final_status = "FAILED" + latest_message = f"Result catalog publish failed: {publish_error}" + elif cancelled: + final_status = "CANCELLED" + latest_message = ( + f"D-InSAR production cancelled. completed={run.completed_items} " + f"failed={run.failed_items} total={run.total_items}" + ) + elif int(run.failed_items or 0) > 0: + final_status = "FAILED" + latest_message = ( + f"D-InSAR production finished with failures. completed={run.completed_items} " + f"failed={run.failed_items} total={run.total_items}" + ) + else: + latest_message = ( + f"D-InSAR production completed. completed={run.completed_items} " + f"failed={run.failed_items} total={run.total_items}" + ) + + summary_payload = { + "workflow": workflow, + "engine_code": run.engine_code, + "profile_code": run.profile_code, + "mode": run.mode, + "total_items": run.total_items, + "completed_items": run.completed_items, + "failed_items": run.failed_items, + "skipped_items": run.skipped_items, + "publish": publish_result, + "publish_error": publish_error, + "published_output_dirs": publish_dirs, + } + await dinsar_production_service.finalize_run( + run, + db=db, + status=final_status, + summary_payload=summary_payload, + latest_message=latest_message, + ) + run_log(run.run_id, f"[finish] status={final_status} message={latest_message}") + + if final_status == "COMPLETED": + await task_service.update_task( + job.task_id, + status="COMPLETED", + progress=100, + message=latest_message, + ) + return + + task_status = "CANCELLED" if final_status == "CANCELLED" else "FAILED" + await task_service.update_task( + job.task_id, + status=task_status, + progress=100, + message=latest_message, + ) + raise RuntimeError(latest_message) + + async def _handle_idl_run_import(job: SystemJobORM) -> None: - await _run_envi_workflow_job( - job, - workflow="import", - start_message="Starting ENVI Import workflow...", - ) + async with engine_lock_service.acquire("envi_taskengine"): + await _run_envi_workflow_job( + job, + workflow="import", + start_message="Starting ENVI Import workflow...", + ) async def _handle_idl_run_dinsar(job: SystemJobORM) -> None: payload = job.payload or {} + production_run_id = str(payload.get("production_run_id") or "").strip() + if production_run_id: + await _run_dinsar_production_controller(job) + return + mode = payload.get("mode", "metatask") workflow = "dinsar_custom" if mode == "custom" else "dinsar" - await _run_envi_workflow_job( - job, - workflow=workflow, - start_message=f"Starting ENVI D-InSAR workflow (mode={mode})...", - ) + async with engine_lock_service.acquire("envi_taskengine"): + await _run_envi_workflow_job( + job, + workflow=workflow, + start_message=f"Starting ENVI D-InSAR workflow (mode={mode})...", + ) async def _handle_isce2_run(job: SystemJobORM) -> None: diff --git a/backend/app/services/job_worker.py b/backend/app/services/job_worker.py index 64dd91d..8a41f3a 100644 --- a/backend/app/services/job_worker.py +++ b/backend/app/services/job_worker.py @@ -96,7 +96,11 @@ async def _run_job(job: SystemJobORM) -> None: if status == JOB_STATUS_RETRY: await task_service.update_task(job.task_id, message=f"任务失败，稍后重试: {err}") else: - await task_service.update_task(job.task_id, status="FAILED", message=err) + current_task = await task_service.get_task(job.task_id) + final_status = "FAILED" + if current_task and current_task.status == "CANCELLED": + final_status = "CANCELLED" + await task_service.update_task(job.task_id, status=final_status, message=err) if job.workflow_run_id and job.workflow_step_id and status != JOB_STATUS_RETRY: await workflow_service.mark_step_failed( job.workflow_run_id, diff --git a/backend/app/services/workflow_service.py b/backend/app/services/workflow_service.py index 9347533..14f8e8c 100644 --- a/backend/app/services/workflow_service.py +++ b/backend/app/services/workflow_service.py @@ -57,6 +57,7 @@ class WorkflowService: "payload": step.get("payload") or {}, "task_id": step.get("task_id"), "optional": bool(step.get("optional", False)), + "max_attempts": step.get("max_attempts"), } db.add( WorkflowStepORM( @@ -101,12 +102,14 @@ class WorkflowService: continue payload = params.get("payload") or {} task_id = params.get("task_id") + max_attempts = params.get("max_attempts") await job_queue_service.create_job( job_type, payload=payload, workflow_run_id=run_id, workflow_step_id=step.step_id, task_id=task_id, + max_attempts=max_attempts if max_attempts is not None else 3, db=db, ) step.status = "RUNNING"