"""ISCE2 D-InSAR engine backed by a WSL pipeline script.""" from __future__ import annotations import json import os import shlex from datetime import datetime from pathlib import Path from typing import Any, Dict, List from ..config import get_env_text, read_bool_env, settings from .base import DinsarEngine, EngineAvailability, EngineProfile, RunRequest, RunResult from ..services.dinsar_naming import ( PAIR_META_FILENAME, build_fallback_pair_key, build_run_key, find_json_sidecar, write_run_metadata, ) from ..services.dinsar_result_layout_service import ( normalize_isce2_run_layout, ) from ..services.isce2_result_validator import validate_isce2_result_files LT1_FIXED_WAVELENGTH = 0.23793052222222222 DEFAULT_TARGET_GRID_SIZE_M = 10 DEFAULT_BBOX_MARGIN = 0.05 DEFAULT_COH_THRESHOLD = 0.05 ORBIT_MARGIN_MIN_SEC = 60.0 ORBIT_MARGIN_MAX_SEC = 120.0 TARGET_GRID_SIZE_MIN_M = 5 TARGET_GRID_SIZE_MAX_M = 100 RERUN_MODE_UNFINISHED_ONLY = "unfinished_only" RESUME_STAGE_CHOICES = {"", "unwrap", "geocode", "export"} def _read_env(name: str, default: str = "") -> str: return get_env_text(name, default) or default def _read_bool_env(name: str, default: bool = False) -> bool: return read_bool_env(name, default) def _windows_path_to_wsl_mount(path: str) -> str: text = str(path or "").strip() if not text: return "" drive, tail = os.path.splitdrive(os.path.normpath(text)) if not drive: return text.replace("\\", "/") drive_letter = drive.rstrip(":").lower() normalized_tail = tail.replace("\\", "/") return f"/mnt/{drive_letter}/{normalized_tail}" def _join_argv_for_log(argv: List[str]) -> str: return " ".join(shlex.quote(str(item)) for item in argv if str(item)) def _normalize_rerun_mode(value: Any) -> str: normalized = str(value or "").strip().lower() return normalized if normalized == RERUN_MODE_UNFINISHED_ONLY else "rerun_all" def _normalize_optional_path(value: Any) -> str: text = str(value or "").strip() if not text: return "" return os.path.normpath(os.path.abspath(text)) def _prefer_prepared_dem_variant(path: str) -> str: normalized = _normalize_optional_path(path) if not normalized or normalized.lower().endswith(".wgs84"): return normalized prepared = normalized + ".wgs84" if os.path.isfile(prepared) and os.path.isfile(prepared + ".xml"): return prepared return normalized def _load_json_file(path: str) -> Dict[str, Any]: try: with open(path, "r", encoding="utf-8") as fp: payload = json.load(fp) return payload if isinstance(payload, dict) else {} except Exception: return {} def _find_isce2_export_outputs(export_dir: str) -> Dict[str, str]: normalized_export_dir = _normalize_optional_path(export_dir) if not normalized_export_dir or not os.path.isdir(normalized_export_dir): return {} outputs: Dict[str, str] = {} with os.scandir(normalized_export_dir) as entries: files = sorted( [entry for entry in entries if entry.is_file()], key=lambda entry: entry.name.lower(), ) for entry in files: lower_name = entry.name.lower() if lower_name.endswith(("_disp_full.tif", "_disp_full.tiff")): continue if "disp" not in outputs and ( lower_name in {"disp.tif", "disp.tiff"} or lower_name.endswith(("_disp.tif", "_disp.tiff")) ): outputs["disp"] = entry.path continue if "coh" not in outputs and ( lower_name in {"coh.tif", "coh.tiff"} or lower_name.endswith(("_coh.tif", "_coh.tiff")) ): outputs["coh"] = entry.path return outputs class Isce2Engine(DinsarEngine): @property def engine_code(self) -> str: return "isce2" @property def engine_label(self) -> str: return "ISCE2（WSL）" @property def default_timeout_seconds(self) -> int: return max(60, int(settings.ISCE2_PER_TASK_TIMEOUT_SECONDS or 43200)) # ------------------------------------------------------------------ # Config helpers # ------------------------------------------------------------------ @property def _enabled(self) -> bool: return _read_bool_env("ISCE2_ENABLED", False) @property def _distro(self) -> str: return _read_env("ISCE2_WSL_DISTRO", "Ubuntu-24.04") @property def _python(self) -> str: return _read_env( "ISCE2_PYTHON", "/home/administrator/miniconda3/envs/isce2/bin/python", ) @property def _runtime_id(self) -> str: return _read_env("ISCE2_RUNTIME_ID", settings.ISCE2_RUNTIME_ID or "isce2_runtime_v1") @property def _dem_path(self) -> str: explicit = _prefer_prepared_dem_variant(_read_env("ISCE2_DEM_PATH", "")) if explicit: return explicit base = _read_env("IDL_DINSAR_DEM_BASE_FILE", "") return _prefer_prepared_dem_variant(base) if base else "" @property def _orbit_pool_isce2(self) -> str: return _read_env("ORBIT_POOL_ISCE2", "") or _read_env("ISCE2_ORBIT_DIR", "") @property def _work_root(self) -> str: return _read_env("ISCE2_WORK_ROOT", "") @property def _output_root(self) -> str: return _read_env("ISCE2_OUTPUT_ROOT", "") @property def _smoke_test(self) -> bool: return _read_bool_env("ISCE2_SMOKE_TEST_ENABLED", False) @property def _stripmap_app(self) -> str: return _read_env( "ISCE2_STRIPMAP_APP", "/home/administrator/miniconda3/envs/isce2/lib/python3.11" "/site-packages/isce/applications/stripmapApp.py", ) @property def _pipeline_script(self) -> str: explicit = _read_env("ISCE2_PIPELINE_SCRIPT", "") if explicit: return explicit local_script = ( Path(__file__).resolve().parent.parent / "isce2_pipeline" / "run_lt1_dinsar_pipeline.py" ) return _windows_path_to_wsl_mount(str(local_script)) # ------------------------------------------------------------------ # Profiles # ------------------------------------------------------------------ def get_profiles(self) -> List[EngineProfile]: return [ EngineProfile( code="lt1_stripmap", label="LT-1 条带模式", description="通过 WSL 环境运行 LT-1 条带模式 D-InSAR 处理流程。", params_schema={ "force": { "label": "强制重建工作目录", "type": "boolean", "default": False, "description": "如果工作目录已经存在，先删除旧目录再重新处理。", "recommendation": "仅在确认旧结果可以丢弃时启用。", }, "target_grid_size_m": { "label": "目标网格尺寸（米）", "type": "number", "default": DEFAULT_TARGET_GRID_SIZE_M, "step": 1, "min": TARGET_GRID_SIZE_MIN_M, "max": TARGET_GRID_SIZE_MAX_M, "description": "控制多视尺度和地理编码输出粒度，系统会自动换算内部处理参数。", "recommendation": "建议优先使用 10；希望保留更多细节可尝试 5，若更看重稳定性和效率可提高到 15 或 20。", }, "bbox": { "label": "地理编码范围", "type": "string", "default": "", "placeholder": "南,北,西,东", "description": "手工指定地理编码范围，格式为南、北、西、东。", "recommendation": "通常留空即可，让系统自动估算；只有在你明确知道目标范围时再手工填写。", }, "coh_threshold": { "label": "相干性阈值", "type": "number", "default": DEFAULT_COH_THRESHOLD, "step": 0.01, "min": 0, "max": 1, "description": "导出位移结果时，低于该阈值的像元会被掩膜。", "recommendation": "默认 0.05 便于先看全量结果；正式成果更建议从 0.10 起用。", }, "bbox_margin": { "label": "范围外扩量（度）", "type": "number", "default": DEFAULT_BBOX_MARGIN, "step": 0.01, "min": 0, "description": "自动估算地理编码范围后，向四周额外扩展的角度。", "recommendation": "推荐 0.05；如果边缘仍有裁切，可逐步提高到 0.08 或 0.10。", }, "wavelength": { "label": "雷达波长（米）", "type": "number", "default": LT1_FIXED_WAVELENGTH, "step": 0.000001, "readonly": True, "include_in_payload": False, "description": "LT-1 固定参数，用于位移量换算。", "recommendation": "系统已锁定，不允许修改。", }, "orbit_margin_sec": { "label": "精轨裁剪时间余量（秒）", "type": "number", "default": ORBIT_MARGIN_MIN_SEC, "step": 1, "min": ORBIT_MARGIN_MIN_SEC, "max": ORBIT_MARGIN_MAX_SEC, "description": "生成精轨 XML 时，在场景开始和结束时刻前后额外保留的时间。", "recommendation": "建议优先使用 60；如果元数据时间标签偏紧或边界异常，可提高到 90 或 120。", }, }, ), ] def normalize_extra(self, extra: Dict[str, Any] | None) -> Dict[str, Any]: normalized: Dict[str, Any] = dict(extra or {}) normalized.pop("wavelength", None) if "force" in normalized: normalized["force"] = bool(normalized["force"]) if "bbox" in normalized and normalized["bbox"] is not None: normalized["bbox"] = str(normalized["bbox"]).strip() if "target_grid_size_m" in normalized and normalized["target_grid_size_m"] is not None: try: grid_size = float(normalized["target_grid_size_m"]) except (TypeError, ValueError) as exc: raise ValueError("目标网格尺寸必须为数字。") from exc if int(grid_size) != grid_size: raise ValueError("目标网格尺寸必须使用整数米。") grid_size = int(grid_size) if grid_size < TARGET_GRID_SIZE_MIN_M or grid_size > TARGET_GRID_SIZE_MAX_M: raise ValueError( f"目标网格尺寸必须在 {TARGET_GRID_SIZE_MIN_M} 到 {TARGET_GRID_SIZE_MAX_M} 米之间。" ) normalized["target_grid_size_m"] = grid_size if "coh_threshold" in normalized and normalized["coh_threshold"] is not None: try: coh_threshold = float(normalized["coh_threshold"]) except (TypeError, ValueError) as exc: raise ValueError("相干性阈值必须为数字。") from exc if coh_threshold < 0 or coh_threshold > 1: raise ValueError("相干性阈值必须在 0 到 1 之间。") normalized["coh_threshold"] = coh_threshold if "bbox_margin" in normalized and normalized["bbox_margin"] is not None: try: bbox_margin = float(normalized["bbox_margin"]) except (TypeError, ValueError) as exc: raise ValueError("范围外扩量必须为数字。") from exc if bbox_margin < 0: raise ValueError("范围外扩量不能小于 0。") normalized["bbox_margin"] = bbox_margin if "orbit_margin_sec" in normalized and normalized["orbit_margin_sec"] is not None: try: orbit_margin = float(normalized["orbit_margin_sec"]) except (TypeError, ValueError) as exc: raise ValueError("精轨裁剪时间余量必须为数字。") from exc if orbit_margin < ORBIT_MARGIN_MIN_SEC or orbit_margin > ORBIT_MARGIN_MAX_SEC: raise ValueError( f"精轨裁剪时间余量必须在 {int(ORBIT_MARGIN_MIN_SEC)} 到 {int(ORBIT_MARGIN_MAX_SEC)} 秒之间。" ) normalized["orbit_margin_sec"] = orbit_margin if "full_geocode" in normalized: normalized["full_geocode"] = bool(normalized["full_geocode"]) if "resume_from" in normalized and normalized["resume_from"] is not None: resume_from = str(normalized["resume_from"]).strip().lower() if resume_from not in RESUME_STAGE_CHOICES: raise ValueError("resume_from must be one of: unwrap, geocode, export") normalized["resume_from"] = resume_from if bool(normalized.get("force")) and str(normalized.get("resume_from") or "").strip(): raise ValueError("force cannot be used together with resume_from") return normalized # ------------------------------------------------------------------ # Availability # ------------------------------------------------------------------ def check_available(self) -> EngineAvailability: if not self._enabled: return EngineAvailability( engine_code=self.engine_code, status="unavailable", available=False, checks=[ { "name": "ISCE2_ENABLED", "ok": False, "detail": "ISCE2_ENABLED=false", } ], message="ISCE2 is disabled. Set ISCE2_ENABLED=true to enable it.", ) from ..services.wsl_service import check_wsl_environment report = check_wsl_environment( distro=self._distro, python_cmd=self._python, stripmap_app_path=self._stripmap_app, pipeline_script_path=self._pipeline_script, dem_path_win=self._dem_path, orbit_dir_win=self._orbit_pool_isce2, output_dir_win=self._output_root, smoke_test=self._smoke_test, ) checks_list = [ {"name": check.name, "ok": check.ok, "detail": check.detail, "skipped": check.skipped} for check in report.checks ] if report.overall_ok: status = "ok" available = True else: critical_failed = [check for check in report.checks if not check.ok and not check.skipped] status = "degraded" if critical_failed else "unavailable" available = False return EngineAvailability( engine_code=self.engine_code, status=status, available=available, checks=checks_list, message=report.message, ) # ------------------------------------------------------------------ # Task discovery # ------------------------------------------------------------------ def _has_completed_task_result(self, task_dir: str, profile_code: str) -> bool: task_name = os.path.basename(os.path.normpath(task_dir)) pair_meta = find_json_sidecar(task_dir, PAIR_META_FILENAME, max_levels=0) or {} task_alias = str(pair_meta.get("task_alias") or task_name).strip() or task_name pair_key = str(pair_meta.get("pair_key") or "").strip() or build_fallback_pair_key(task_alias, task_dir) pointer_path = os.path.join( settings.DINSAR_PRODUCT_DIR, pair_key, "current", f"{self.engine_code}__{str(profile_code or '').strip()}.json", ) if not os.path.isfile(pointer_path): return False payload = _load_json_file(pointer_path) if str(payload.get("status") or "").strip().upper() != "COMPLETED": return False primary_file = _normalize_optional_path(payload.get("primary_file")) source_files = payload.get("source_files") if isinstance(payload.get("source_files"), list) else [] validation = validate_isce2_result_files(primary_file, source_files) if not bool(validation.get("accepted")): return False manifest_path = _normalize_optional_path(payload.get("manifest_path")) if manifest_path and os.path.isfile(manifest_path): return True output_dir = _normalize_optional_path(payload.get("output_dir")) if output_dir and os.path.isfile(os.path.join(output_dir, "execution_manifest.json")): return True return False def validate_root_dir( self, root_dir: str, num_to_process: int = 0, rerun_mode: str = "rerun_all", ) -> Dict[str, Any]: normalized_root = os.path.normpath(os.path.abspath(str(root_dir or "").strip())) if not root_dir or not os.path.isdir(normalized_root): raise ValueError(f"ISCE2 root_dir does not exist or is not a directory: {root_dir}") if self._looks_like_task_dir(normalized_root): task_dirs = [normalized_root] invalid_candidates: List[Dict[str, Any]] = [] mode = "single_task_dir" else: task_dirs = [] invalid_candidates = [] for entry in self._iter_child_dirs(normalized_root): if not entry.name.lower().startswith("task_"): continue missing = self._missing_task_subdirs(entry.path) if missing: invalid_candidates.append( {"name": entry.name, "path": entry.path, "missing_subdirs": missing} ) continue task_dirs.append(os.path.normpath(entry.path)) mode = "task_root_dir" if not task_dirs: detail = "" if invalid_candidates: formatted = ", ".join( f"{item['name']} missing {','.join(item['missing_subdirs'])}" for item in invalid_candidates[:5] ) detail = f" Invalid candidates: {formatted}." raise ValueError( "ISCE2 root_dir must be either a single task directory containing " "'master' and 'slave', or a parent directory containing valid Task_* subdirectories." f"{detail}" ) discovered_task_count = len(task_dirs) skipped_completed_count = 0 selected_task_dirs = task_dirs if _normalize_rerun_mode(rerun_mode) == RERUN_MODE_UNFINISHED_ONLY: selected_task_dirs = [] for task_dir in task_dirs: if self._has_completed_task_result(task_dir, "lt1_stripmap"): skipped_completed_count += 1 continue selected_task_dirs.append(task_dir) selected_count = int(num_to_process or 0) if selected_count > 0: selected_task_dirs = selected_task_dirs[:selected_count] return { "root_dir": normalized_root, "mode": mode, "task_dirs": selected_task_dirs, "task_count": len(selected_task_dirs), "selected_task_count": len(selected_task_dirs), "discovered_task_count": discovered_task_count, "skipped_completed_count": skipped_completed_count, "invalid_candidates": invalid_candidates, } def _build_lt1_manifest_payload( self, *, request: RunRequest, task_dir: str, task_name: str, task_alias: str, pair_key: str, run_key: str, work_dir: str, output_dir: str, orbit_output_dir: str, wsl_task_dir: str, wsl_work_dir: str, wsl_output_dir: str, wsl_orbit_output_dir: str, wsl_orbit_root: str, wsl_dem: str, force: bool, target_grid_size_m: int, bbox: str, coh_threshold: Any, bbox_margin: Any, wavelength: Any, orbit_margin_sec: Any, full_geocode: bool, resume_from: str, pair_meta: Dict[str, Any], ) -> Dict[str, Any]: return { "job_id": request.job_id, "profile": request.profile, "task_name": task_name, "task_alias": task_alias, "pair_key": pair_key, "run_key": run_key, "paths": { "source_root_windows": os.path.normpath( str((request.extra or {}).get("__source_root_override") or request.root_dir) ), "task_dir_windows": os.path.normpath(task_dir), "task_dir_wsl": wsl_task_dir, "work_dir_windows": work_dir, "work_dir_wsl": wsl_work_dir, "output_dir_windows": output_dir, "output_dir_wsl": wsl_output_dir, "orbit_output_dir_windows": orbit_output_dir, "orbit_output_dir_wsl": wsl_orbit_output_dir, "orbit_root_windows": self._orbit_pool_isce2, "orbit_root_wsl": wsl_orbit_root, "dem_path_windows": self._dem_path, "dem_path_wsl": wsl_dem, }, "params": { "force": bool(force), "target_grid_size_m": int(target_grid_size_m), "bbox": str(bbox or "").strip(), "coh_threshold": coh_threshold, "bbox_margin": bbox_margin, "wavelength": wavelength, "orbit_margin_sec": orbit_margin_sec, "full_geocode": bool(full_geocode), "resume_from": str(resume_from or "").strip(), }, "pair_meta": dict(pair_meta or {}), } @staticmethod def _iter_child_dirs(root_dir: str): with os.scandir(root_dir) as entries: child_dirs = [entry for entry in entries if entry.is_dir()] child_dirs.sort(key=lambda entry: entry.name.lower()) return child_dirs @staticmethod def _missing_task_subdirs(task_dir: str) -> List[str]: missing: List[str] = [] for subdir in ("master", "slave"): if not os.path.isdir(os.path.join(task_dir, subdir)): missing.append(subdir) return missing def _looks_like_task_dir(self, task_dir: str) -> bool: return not self._missing_task_subdirs(task_dir) # ------------------------------------------------------------------ # Run # ------------------------------------------------------------------ def run(self, request: RunRequest) -> RunResult: if not self._enabled: return RunResult( success=False, engine_code=self.engine_code, profile=request.profile, job_id=request.job_id, error="ISCE2 is disabled.", ) if request.profile != "lt1_stripmap": return RunResult( success=False, engine_code=self.engine_code, profile=request.profile, job_id=request.job_id, error=f"Unknown profile: {request.profile}", ) return self._run_lt1_stripmap(request) def _run_lt1_stripmap(self, request: RunRequest) -> RunResult: from ..services.wsl_broker import wsl_broker from ..services.wsl_runtime_registry import get_wsl_runtime from ..services.wsl_service import windows_path_to_wsl extra = self.normalize_extra(request.extra) validation = self.validate_root_dir( request.root_dir, request.num_to_process, str((request.extra or {}).get("__rerun_mode") or "rerun_all"), ) task_dirs: List[str] = validation["task_dirs"] total_tasks = len(task_dirs) run_started_at = datetime.utcnow() run_started_at_text = run_started_at.isoformat(timespec="seconds") + "Z" run_key = str(extra.get("__managed_run_key") or "").strip() or build_run_key( self.engine_code, request.profile, started_at=run_started_at, ) progress_callback = request.progress_callback runtime = get_wsl_runtime(self._runtime_id) managed_run_dir_override = _normalize_optional_path(extra.get("__managed_run_dir")) managed_native_output_dir_override = _normalize_optional_path(extra.get("__managed_native_output_dir")) managed_work_dir_override = _normalize_optional_path(extra.get("__managed_work_dir")) managed_export_dir_override = _normalize_optional_path(extra.get("__managed_export_dir")) managed_orbit_output_dir_override = _normalize_optional_path(extra.get("__managed_orbit_output_dir")) source_root_override = _normalize_optional_path(extra.get("__source_root_override")) or os.path.normpath(request.root_dir) has_managed_override = any( [ managed_run_dir_override, managed_native_output_dir_override, managed_work_dir_override, managed_export_dir_override, managed_orbit_output_dir_override, ] ) if has_managed_override and total_tasks > 1: return RunResult( success=False, engine_code=self.engine_code, profile=request.profile, job_id=request.job_id, error="Managed ISCE2 directory overrides require a single task request.", ) def emit_progress(event_type: str, **payload: Any) -> None: if not callable(progress_callback): return try: progress_callback({"event": event_type, **payload}) except Exception: return timeout = max(60, int(request.timeout_seconds or self.default_timeout_seconds)) force = bool(extra.get("force")) target_grid_size_m = int(extra.get("target_grid_size_m", DEFAULT_TARGET_GRID_SIZE_M)) bbox = extra.get("bbox", "") coh_threshold = extra.get("coh_threshold") bbox_margin = extra.get("bbox_margin") wavelength = LT1_FIXED_WAVELENGTH orbit_margin_sec = extra.get("orbit_margin_sec") full_geocode = bool(extra.get("full_geocode")) resume_from = str(extra.get("resume_from") or "").strip().lower() wsl_isce2_pool = "" if self._orbit_pool_isce2: wsl_isce2_pool = windows_path_to_wsl(self._orbit_pool_isce2, distro=runtime.distro) wsl_dem = "" if self._dem_path: wsl_dem = windows_path_to_wsl(self._dem_path, distro=runtime.distro) task_results: List[Dict[str, Any]] = [] output_dirs: List[str] = [] pairs_processed = 0 pairs_failed = 0 for pair_index, task_dir in enumerate(task_dirs, start=1): task_name = os.path.basename(os.path.normpath(task_dir)) pair_meta = find_json_sidecar(task_dir, PAIR_META_FILENAME, max_levels=0) or {} task_alias = str(pair_meta.get("task_alias") or task_name).strip() or task_name pair_key = str(pair_meta.get("pair_key") or "").strip() or build_fallback_pair_key(task_alias, task_dir) output_root = self._output_root or os.path.join(task_dir, "isce2_output") run_dir = managed_run_dir_override or os.path.normpath( os.path.join(output_root, pair_key, "runs", run_key) ) native_output_dir = managed_native_output_dir_override or os.path.join(run_dir, "native") work_dir = managed_work_dir_override or os.path.join(native_output_dir, "workflow") export_dir = managed_export_dir_override or os.path.join(native_output_dir, "export") orbit_output_dir = managed_orbit_output_dir_override or os.path.join(work_dir, "orbits") wsl_task_dir = windows_path_to_wsl(task_dir, distro=runtime.distro) wsl_work_dir = windows_path_to_wsl(work_dir, distro=runtime.distro) wsl_output_dir = windows_path_to_wsl(export_dir, distro=runtime.distro) wsl_orbit_output_dir = windows_path_to_wsl(orbit_output_dir, distro=runtime.distro) emit_progress( "pair_started", pair_index=pair_index, pair_total=total_tasks, task_name=task_name, task_alias=task_alias, pair_key=pair_key, task_dir=task_dir, work_dir=work_dir, output_dir=run_dir, ) if not wsl_task_dir: pairs_failed += 1 error_text = f"Unable to convert task dir to WSL path: {task_dir}" emit_progress( "pair_finished", pair_index=pair_index, pair_total=total_tasks, task_name=task_name, task_alias=task_alias, pair_key=pair_key, success=False, returncode=-2, error=error_text, ) task_results.append( { "task_name": task_name, "task_alias": task_alias, "pair_key": pair_key, "run_key": run_key, "task_dir": task_dir, "run_dir": run_dir, "native_output_dir": native_output_dir, "work_dir": work_dir, "output_dir": run_dir, "export_dir": export_dir, "success": False, "returncode": -2, "error": error_text, "stdout_tail": "", "stderr_tail": "", "command": "", "wsl_task_dir": "", "wsl_work_dir": "", "wsl_output_dir": "", } ) continue if not wsl_work_dir or not wsl_output_dir or not wsl_orbit_output_dir: pairs_failed += 1 error_text = "Unable to convert ISCE2 work/output paths to WSL paths." emit_progress( "pair_finished", pair_index=pair_index, pair_total=total_tasks, task_name=task_name, task_alias=task_alias, pair_key=pair_key, success=False, returncode=-2, error=error_text, ) task_results.append( { "task_name": task_name, "task_alias": task_alias, "pair_key": pair_key, "run_key": run_key, "task_dir": task_dir, "run_dir": run_dir, "native_output_dir": native_output_dir, "work_dir": work_dir, "output_dir": run_dir, "export_dir": export_dir, "success": False, "returncode": -2, "error": error_text, "stdout_tail": "", "stderr_tail": "", "command": "", "wsl_task_dir": wsl_task_dir, "wsl_work_dir": wsl_work_dir, "wsl_output_dir": wsl_output_dir, } ) continue manifest_payload = self._build_lt1_manifest_payload( request=request, task_dir=task_dir, task_name=task_name, task_alias=task_alias, pair_key=pair_key, run_key=run_key, work_dir=work_dir, output_dir=export_dir, orbit_output_dir=orbit_output_dir, wsl_task_dir=wsl_task_dir, wsl_work_dir=wsl_work_dir, wsl_output_dir=wsl_output_dir, wsl_orbit_output_dir=wsl_orbit_output_dir, wsl_orbit_root=wsl_isce2_pool, wsl_dem=wsl_dem, force=force, target_grid_size_m=target_grid_size_m, bbox=bbox, coh_threshold=coh_threshold, bbox_margin=bbox_margin, wavelength=wavelength, orbit_margin_sec=orbit_margin_sec, full_geocode=full_geocode, resume_from=resume_from, pair_meta=pair_meta, ) broker_result = wsl_broker.run_manifest( runtime_id=runtime.runtime_id, operation="lt1_stripmap", payload=manifest_payload, job_id=f"{request.job_id}_{pair_key}", timeout_seconds=timeout, ) rc = broker_result.returncode stdout = broker_result.stdout stderr = broker_result.stderr command = _join_argv_for_log(list(broker_result.argv)) success = False layout_result: Dict[str, Any] = {} validation_result: Dict[str, Any] = {} error_text = stderr.strip() if stderr else "" if rc == 0: try: os.makedirs(run_dir, exist_ok=True) export_outputs = _find_isce2_export_outputs(export_dir) disp_path = export_outputs.get("disp", "") coh_path = export_outputs.get("coh", "") if not disp_path: raise FileNotFoundError( f"No ISCE2 displacement GeoTIFF found under export dir: {export_dir}" ) source_files = [disp_path] if coh_path: source_files.append(coh_path) validation_result = validate_isce2_result_files(disp_path, source_files) if not bool(validation_result.get("accepted")): issues = validation_result.get("issues") or [] issue_text = "; ".join(str(item) for item in issues[:3]) or "unknown validation error" raise RuntimeError(f"ISCE2 output validation failed: {issue_text}") layout_result = normalize_isce2_run_layout( run_dir, primary_file=str(validation_result.get("primary_file") or disp_path), source_files=list(validation_result.get("source_files") or source_files), rewrite_metadata=False, ) write_run_metadata( run_dir, { "run_key": run_key, "pair_key": pair_key, "task_name": task_name, "task_alias": task_alias, "engine_code": self.engine_code, "profile_code": request.profile, "source_root": source_root_override, "task_dir": os.path.normpath(task_dir), "work_dir": work_dir, "export_dir": export_dir, "output_dir": run_dir, "native_output_dir": layout_result["native_output_dir"], "orbit_output_dir": orbit_output_dir, "runtime_id": runtime.runtime_id, "manifest_path_windows": broker_result.manifest.manifest_path_windows, "manifest_path_wsl": broker_result.manifest.manifest_path_wsl, "started_at": run_started_at_text, "finished_at": datetime.utcnow().isoformat(timespec="seconds") + "Z", "primary_file": layout_result["primary_file"], "source_files": layout_result["source_files"], "acceptance": validation_result, "params": { "force": force, "target_grid_size_m": target_grid_size_m, "bbox": bbox, "coh_threshold": coh_threshold, "bbox_margin": bbox_margin, "wavelength": wavelength, "orbit_margin_sec": orbit_margin_sec, }, "master_path": pair_meta.get("master_path"), "slave_path": pair_meta.get("slave_path"), "master_satellite": pair_meta.get("master_satellite"), "slave_satellite": pair_meta.get("slave_satellite"), "master_imaging_date": pair_meta.get("master_imaging_date"), "slave_imaging_date": pair_meta.get("slave_imaging_date"), "master_imaging_mode": pair_meta.get("master_imaging_mode"), "slave_imaging_mode": pair_meta.get("slave_imaging_mode"), "master_polarization": pair_meta.get("master_polarization"), "slave_polarization": pair_meta.get("slave_polarization"), "time_baseline_days": pair_meta.get("time_baseline_days"), "spatial_baseline_meters": pair_meta.get("spatial_baseline_meters"), "scene_pair_uid": pair_meta.get("scene_pair_uid") or pair_meta.get("pair_uid"), "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"), }, ) output_dirs.append(run_dir) pairs_processed += 1 success = True except Exception as exc: error_text = str(exc) if stderr: stderr = stderr.rstrip() + "\n" + error_text else: stderr = error_text if not success: pairs_failed += 1 emit_progress( "pair_finished", pair_index=pair_index, pair_total=total_tasks, task_name=task_name, task_alias=task_alias, pair_key=pair_key, success=success, returncode=rc, error=error_text, ) task_results.append( { "task_name": task_name, "task_alias": task_alias, "pair_key": pair_key, "run_key": run_key, "task_dir": task_dir, "run_dir": run_dir, "native_output_dir": ( layout_result.get("native_output_dir") or native_output_dir ), "work_dir": work_dir, "output_dir": run_dir, "export_dir": export_dir, "primary_file": layout_result.get("primary_file", ""), "source_files": layout_result.get("source_files", []), "validation": validation_result, "wsl_task_dir": wsl_task_dir, "wsl_work_dir": wsl_work_dir, "wsl_output_dir": wsl_output_dir, "runtime_id": runtime.runtime_id, "command": command, "runner_argv": list(broker_result.argv), "manifest_path_windows": broker_result.manifest.manifest_path_windows, "manifest_path_wsl": broker_result.manifest.manifest_path_wsl, "success": success, "returncode": rc, "stdout_tail": stdout[-3000:] if stdout else "", "stderr_tail": stderr[-3000:] if stderr else "", "error": error_text, } ) invalid_candidates = validation.get("invalid_candidates", []) pairs_failed += len(invalid_candidates) overall_success = pairs_processed > 0 or (pairs_processed == 0 and pairs_failed == 0) failed_task_names = [ item["task_name"] for item in task_results if not item.get("success") ] + [item["name"] for item in invalid_candidates] error = None if not overall_success: if failed_task_names: error = f"All ISCE2 tasks failed: {', '.join(failed_task_names[:10])}" else: error = "ISCE2 run failed." last_task_result = task_results[-1] if task_results else {} return RunResult( success=overall_success, engine_code=self.engine_code, profile=request.profile, job_id=request.job_id, pairs_processed=pairs_processed, pairs_failed=pairs_failed, output_dirs=output_dirs, error=error, detail={ "mode": validation["mode"], "task_count": len(task_dirs), "selected_tasks": [item.get("task_alias") or item.get("task_name") for item in task_results], "invalid_candidates": invalid_candidates, "task_results": task_results, "run_key": run_key, "started_at": run_started_at_text, "force": force, "timeout_seconds": timeout, "target_grid_size_m": target_grid_size_m, "wavelength": wavelength, "orbit_margin_sec": orbit_margin_sec, "runtime_id": runtime.runtime_id, "command": last_task_result.get("command", ""), "runner_argv": last_task_result.get("runner_argv", []), "manifest_path_windows": last_task_result.get("manifest_path_windows", ""), "manifest_path_wsl": last_task_result.get("manifest_path_wsl", ""), "stdout_tail": last_task_result.get("stdout_tail", ""), "stderr_tail": last_task_result.get("stderr_tail", ""), "wsl_task_dir": last_task_result.get("wsl_task_dir", ""), "wsl_work_dir": last_task_result.get("wsl_work_dir", ""), "wsl_output_dir": last_task_result.get("wsl_output_dir", ""), "wsl_orbit_pool": wsl_isce2_pool, "wsl_dem": wsl_dem, "wsl_work_root": ( os.path.dirname(str(last_task_result.get("wsl_work_dir") or "").strip()) if str(last_task_result.get("wsl_work_dir") or "").strip() else (windows_path_to_wsl(self._work_root, distro=runtime.distro) if self._work_root else "") ), "wsl_output_root": ( os.path.dirname(str(last_task_result.get("wsl_output_dir") or "").strip()) if str(last_task_result.get("wsl_output_dir") or "").strip() else (windows_path_to_wsl(self._output_root, distro=runtime.distro) if self._output_root else "") ), }, )