Harmon/insar-management-system-v2
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: sync repository with current workspace state

Browse Source
This commit is contained in:
Harmon
2026-05-07 11:16:40 +08:00
parent 39cf92044f
commit 69d15bffcb
71 changed files with 3382 additions and 6776 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/config.py
+81
View File
@@ -265,9 +265,26 @@ class Settings(BaseSettings):
PYINT_DEM_MODE: str = "local_fabdem"
PYINT_FABDEM_ROOT: str = ""
PYINT_PREPARED_DEM_PATH: str = ""
PYINT_DEM_RESOLUTION_M: float = 30.0
PYINT_OPENTOPO_DEM_TYPE: str = "SRTMGL1"
PYINT_OPENTOPO_API_KEY: str = ""
PYINT_DEM_STRICT: bool = True
PYINT_UNWRAP_COH_THRESHOLD: float = 0.05
PYINT_PRODUCT_COH_THRESHOLD: float = 0.20
PYINT_REFERENCE_MODE: str = "none"
PYINT_REFERENCE_COH_THRESHOLD: float = 0.30
PYINT_DERAMP_MODE: str = "none"
PYINT_DERAMP_COH_THRESHOLD: float = 0.30
PYINT_GAMMA_NODATA_VALUE: float = -9999.0
PYINT_GEO_INTERP: str = "1"
PYINT_ATMCOR_ENABLED: bool = False
PYINT_ATMCOR_USE_FOR_DISP: bool = False
PYINT_REFLATTEN_ENABLED: bool = True
PYINT_REFLATTEN_MODEL: str = "plane"
PYINT_REFLATTEN_COH_THRESHOLD: float = 0.70
PYINT_REFLATTEN_FALLBACK_COH_THRESHOLD: float = 0.20
PYINT_REFLATTEN_RANGE_STEP: int = 32
PYINT_REFLATTEN_AZIMUTH_STEP: int = 32
PYINT_ORBIT_POLICY: str = "require_txt"
PYINT_ORBIT_POOL_TXT: str = ""
PYINT_RECORD_INPUT_ASSETS: bool = True
@@ -491,6 +508,70 @@ class Settings(BaseSettings):
if pyint_dem_mode not in {"local_fabdem", "opentopo", "prepared_file"}:
pyint_dem_mode = "local_fabdem"
object.__setattr__(self, "PYINT_DEM_MODE", pyint_dem_mode)
object.__setattr__(self, "PYINT_DEM_RESOLUTION_M", max(0.1, float(self.PYINT_DEM_RESOLUTION_M or 30.0)))
object.__setattr__(
self,
"PYINT_UNWRAP_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_UNWRAP_COH_THRESHOLD or 0.05))),
)
object.__setattr__(
self,
"PYINT_PRODUCT_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_PRODUCT_COH_THRESHOLD or 0.20))),
)
pyint_reference_mode = str(self.PYINT_REFERENCE_MODE or "none").strip().lower() or "none"
if pyint_reference_mode not in {"none", "coh_median"}:
pyint_reference_mode = "none"
object.__setattr__(self, "PYINT_REFERENCE_MODE", pyint_reference_mode)
object.__setattr__(
self,
"PYINT_REFERENCE_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_REFERENCE_COH_THRESHOLD or 0.30))),
)
pyint_deramp_mode = str(self.PYINT_DERAMP_MODE or "none").strip().lower() or "none"
if pyint_deramp_mode not in {"none", "plane"}:
pyint_deramp_mode = "none"
object.__setattr__(self, "PYINT_DERAMP_MODE", pyint_deramp_mode)
object.__setattr__(
self,
"PYINT_DERAMP_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_DERAMP_COH_THRESHOLD or 0.30))),
)
object.__setattr__(
self,
"PYINT_GAMMA_NODATA_VALUE",
float(self.PYINT_GAMMA_NODATA_VALUE if self.PYINT_GAMMA_NODATA_VALUE is not None else -9999.0),
)
pyint_geo_interp = str(self.PYINT_GEO_INTERP or "0").strip()
if pyint_geo_interp not in {"0", "1"}:
pyint_geo_interp = "1"
object.__setattr__(self, "PYINT_GEO_INTERP", pyint_geo_interp)
pyint_reflatten_model = str(self.PYINT_REFLATTEN_MODEL or "plane").strip().lower() or "plane"
if pyint_reflatten_model in {"linear"}:
pyint_reflatten_model = "plane"
if pyint_reflatten_model not in {"plane", "quadratic"}:
pyint_reflatten_model = "plane"
object.__setattr__(self, "PYINT_REFLATTEN_MODEL", pyint_reflatten_model)
object.__setattr__(
self,
"PYINT_REFLATTEN_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_REFLATTEN_COH_THRESHOLD or 0.70))),
)
object.__setattr__(
self,
"PYINT_REFLATTEN_FALLBACK_COH_THRESHOLD",
min(1.0, max(0.0, float(self.PYINT_REFLATTEN_FALLBACK_COH_THRESHOLD or 0.20))),
)
object.__setattr__(
self,
"PYINT_REFLATTEN_RANGE_STEP",
max(1, int(self.PYINT_REFLATTEN_RANGE_STEP or 32)),
)
object.__setattr__(
self,
"PYINT_REFLATTEN_AZIMUTH_STEP",
max(1, int(self.PYINT_REFLATTEN_AZIMUTH_STEP or 32)),
)
if not self.PYINT_OPENTOPO_DEM_TYPE:
object.__setattr__(self, "PYINT_OPENTOPO_DEM_TYPE", "SRTMGL1")
pyint_orbit_policy = str(self.PYINT_ORBIT_POLICY or "require_txt").strip().lower() or "require_txt"
backend/app/dinsar_engines/pyint_engine.py
+628 -27
View File
@@ -8,7 +8,9 @@ from pathlib import Path
from typing import Any, Dict, List
from ..config import get_env_text, read_bool_env, settings
from ..services.dinsar_completion_files import repair_managed_completion_files
from ..services.dinsar_naming import write_run_metadata
from ..services.isce2_result_validator import validate_isce2_result_files
from ..services.pyint_input_assets_service import (
get_pyint_dem_summary,
get_pyint_orbit_context,
@@ -17,11 +19,33 @@ from ..services.pyint_input_assets_service import (
)
from ..services.pyint_service import (
DEFAULT_AZIMUTH_LOOKS,
DEFAULT_DEM_RESOLUTION_M,
DEFAULT_DERAMP_COH_THRESHOLD,
DEFAULT_DERAMP_MODE,
DEFAULT_ATMCOR_ENABLED,
DEFAULT_ATMCOR_USE_FOR_DISP,
DEFAULT_GEO_INTERP,
DEFAULT_PARALLEL_WORKERS,
DEFAULT_PRODUCT_COH_THRESHOLD,
DEFAULT_RANGE_LOOKS,
DEFAULT_REFLATTEN_AZIMUTH_STEP,
DEFAULT_REFLATTEN_COH_THRESHOLD,
DEFAULT_REFLATTEN_ENABLED,
DEFAULT_REFLATTEN_FALLBACK_COH_THRESHOLD,
DEFAULT_REFLATTEN_MODEL,
DEFAULT_REFLATTEN_RANGE_STEP,
DEFAULT_REFERENCE_COH_THRESHOLD,
DEFAULT_REFERENCE_MODE,
DEFAULT_TARGET_GRID_SIZE_M,
DEFAULT_UNWRAP_COH_THRESHOLD,
MAX_LOOKS,
MAX_PARALLEL_WORKERS,
REFLATTEN_MODEL_CHOICES,
TARGET_GRID_SIZE_MAX_M,
TARGET_GRID_SIZE_MIN_M,
build_project_name,
calculate_dem_oversampling,
calculate_looks_from_task_dir,
check_pyint_environment,
infer_scene_date_from_archives,
infer_task_identity,
@@ -31,10 +55,11 @@ from ..services.pyint_service import (
to_wsl_path,
validate_pyint_root_dir,
)
from ..services.wsl_service import run_wsl_command
from ..services.wsl_service import run_wsl_command_stream
from .base import DinsarEngine, EngineAvailability, EngineProfile, RunRequest, RunResult
RERUN_MODE_UNFINISHED_ONLY = "unfinished_only"
DEFAULT_COHERENCE_MASK_THRESHOLD = DEFAULT_PRODUCT_COH_THRESHOLD
def _read_env(name: str, default: str = "") -> str:
@@ -121,6 +146,85 @@ class PyintEngine(DinsarEngine):
def _dem_mode(self) -> str:
return str(getattr(settings, "PYINT_DEM_MODE", "local_fabdem") or "local_fabdem").strip().lower()
@property
def _dem_resolution_m(self) -> float:
return max(0.1, float(getattr(settings, "PYINT_DEM_RESOLUTION_M", DEFAULT_DEM_RESOLUTION_M) or DEFAULT_DEM_RESOLUTION_M))
@property
def _default_unwrap_coh_threshold(self) -> float:
return float(getattr(settings, "PYINT_UNWRAP_COH_THRESHOLD", DEFAULT_UNWRAP_COH_THRESHOLD) or DEFAULT_UNWRAP_COH_THRESHOLD)
@property
def _default_product_coh_threshold(self) -> float:
return float(getattr(settings, "PYINT_PRODUCT_COH_THRESHOLD", DEFAULT_PRODUCT_COH_THRESHOLD) or DEFAULT_PRODUCT_COH_THRESHOLD)
@property
def _default_reference_mode(self) -> str:
return str(getattr(settings, "PYINT_REFERENCE_MODE", DEFAULT_REFERENCE_MODE) or DEFAULT_REFERENCE_MODE).strip().lower()
@property
def _default_reference_coh_threshold(self) -> float:
return float(getattr(settings, "PYINT_REFERENCE_COH_THRESHOLD", DEFAULT_REFERENCE_COH_THRESHOLD) or DEFAULT_REFERENCE_COH_THRESHOLD)
@property
def _default_deramp_mode(self) -> str:
return str(getattr(settings, "PYINT_DERAMP_MODE", DEFAULT_DERAMP_MODE) or DEFAULT_DERAMP_MODE).strip().lower()
@property
def _default_deramp_coh_threshold(self) -> float:
return float(getattr(settings, "PYINT_DERAMP_COH_THRESHOLD", DEFAULT_DERAMP_COH_THRESHOLD) or DEFAULT_DERAMP_COH_THRESHOLD)
@property
def _gamma_nodata_value(self) -> float:
return float(getattr(settings, "PYINT_GAMMA_NODATA_VALUE", -9999.0) if getattr(settings, "PYINT_GAMMA_NODATA_VALUE", None) is not None else -9999.0)
@property
def _geo_interp(self) -> str:
value = str(getattr(settings, "PYINT_GEO_INTERP", DEFAULT_GEO_INTERP) or DEFAULT_GEO_INTERP).strip()
return value if value in {"0", "1"} else DEFAULT_GEO_INTERP
@property
def _atmcor_enabled(self) -> bool:
return bool(getattr(settings, "PYINT_ATMCOR_ENABLED", DEFAULT_ATMCOR_ENABLED))
@property
def _atmcor_use_for_disp(self) -> bool:
return bool(getattr(settings, "PYINT_ATMCOR_USE_FOR_DISP", DEFAULT_ATMCOR_USE_FOR_DISP))
@property
def _reflatten_enabled(self) -> bool:
return bool(getattr(settings, "PYINT_REFLATTEN_ENABLED", DEFAULT_REFLATTEN_ENABLED))
@property
def _reflatten_model(self) -> str:
value = str(getattr(settings, "PYINT_REFLATTEN_MODEL", DEFAULT_REFLATTEN_MODEL) or DEFAULT_REFLATTEN_MODEL).strip().lower()
if value == "linear":
value = "plane"
return value if value in REFLATTEN_MODEL_CHOICES else DEFAULT_REFLATTEN_MODEL
@property
def _reflatten_coh_threshold(self) -> float:
return float(getattr(settings, "PYINT_REFLATTEN_COH_THRESHOLD", DEFAULT_REFLATTEN_COH_THRESHOLD) or DEFAULT_REFLATTEN_COH_THRESHOLD)
@property
def _reflatten_fallback_coh_threshold(self) -> float:
return float(
getattr(
settings,
"PYINT_REFLATTEN_FALLBACK_COH_THRESHOLD",
DEFAULT_REFLATTEN_FALLBACK_COH_THRESHOLD,
)
or DEFAULT_REFLATTEN_FALLBACK_COH_THRESHOLD
)
@property
def _reflatten_range_step(self) -> int:
return max(1, int(getattr(settings, "PYINT_REFLATTEN_RANGE_STEP", DEFAULT_REFLATTEN_RANGE_STEP) or DEFAULT_REFLATTEN_RANGE_STEP))
@property
def _reflatten_azimuth_step(self) -> int:
return max(1, int(getattr(settings, "PYINT_REFLATTEN_AZIMUTH_STEP", DEFAULT_REFLATTEN_AZIMUTH_STEP) or DEFAULT_REFLATTEN_AZIMUTH_STEP))
@property
def _fabdem_root(self) -> str:
return _read_env("PYINT_FABDEM_ROOT", "")
@@ -197,25 +301,39 @@ class PyintEngine(DinsarEngine):
"label": "强制重跑",
"type": "boolean",
"default": False,
"section": "Execution",
"description": "删除当前 run_key 对应的工作区后重跑。",
},
"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,
"section": "Advanced",
"description": "可选。仅在未手动填写 looks 时用于估算多视数；不会重采样 DEM 或改写 Gamma 产品。",
"recommendation": "保持 0 使用显式或默认的 Gamma/PyINT looks。",
},
"range_looks": {
"label": "距离向多视",
"label": "距离向多视（手动覆盖）",
"type": "number",
"default": DEFAULT_RANGE_LOOKS,
"step": 1,
"min": 1,
"max": MAX_LOOKS,
"description": "PyINT 模板中的 range_looks。",
"section": "Execution",
"description": "PyINT/Gamma 模板中的 range_looks。",
},
"azimuth_looks": {
"label": "方位向多视",
"label": "方位向多视（手动覆盖）",
"type": "number",
"default": DEFAULT_AZIMUTH_LOOKS,
"step": 1,
"min": 1,
"max": MAX_LOOKS,
"description": "PyINT 模板中的 azimuth_looks。",
"section": "Execution",
"description": "PyINT/Gamma 模板中的 azimuth_looks。",
},
"parallel_workers": {
"label": "并行数",
@@ -224,18 +342,121 @@ class PyintEngine(DinsarEngine):
"step": 1,
"min": 1,
"max": MAX_PARALLEL_WORKERS,
"section": "Execution",
"description": "同步控制 raw2slc/coreg/diff/unwrap/geocode 的并行数。",
},
"coherence_mask_threshold": {
"label": "Coherence quality",
"type": "number",
"default": self._default_product_coh_threshold,
"step": 0.05,
"min": 0.0,
"max": 1.0,
"section": "Delivery",
"description": "Only used for quality support statistics. It is not applied as a Python product mask.",
"recommendation": "Use 0.20 by default for LT-1 single-pair reporting; raise it for stricter review maps.",
},
"unwrap_coh_threshold": {
"label": "Unwrap coherence",
"type": "number",
"default": self._default_unwrap_coh_threshold,
"step": 0.05,
"min": 0.0,
"max": 1.0,
"section": "Advanced",
"description": "Minimum coherence used by Gamma rascc_mask/mcf during unwrapping.",
"recommendation": "Use 0.05 for ENVI-like permissive LT-1 unwrapping; raise it only when low-coherence bridges cause unwrap artifacts.",
},
"geo_interp": {
"label": "Geocode interpolation",
"type": "select",
"default": self._geo_interp,
"enum": ["0", "1"],
"section": "Advanced",
"description": "Gamma geocode_back interpolation mode: 0 nearest, 1 bicubic spline.",
},
"atmcor": {
"label": "Gamma atmcor",
"type": "boolean",
"default": self._atmcor_enabled,
"section": "Advanced",
"description": "Run PyINT/Gamma atm_correction stage using atm_mod_2d/atm_sim_2d/sub_phase.",
},
"atmcor_use_for_disp": {
"label": "Use atmcor for disp",
"type": "boolean",
"default": self._atmcor_use_for_disp,
"section": "Advanced",
"description": "Use the Gamma atmospheric-corrected unwrapped phase as dispmap input when available.",
},
"reflatten": {
"label": "Gamma residual reflatten",
"type": "boolean",
"default": self._reflatten_enabled,
"section": "Gamma Refinement",
"description": "After unwrapping, fit and remove residual long-wavelength phase ramps with Gamma rascc_mask/quad_fit/quad_sub.",
"recommendation": "Keep enabled for LT-1 D-InSAR unless validating the raw PyINT/Gamma baseline.",
},
"reflatten_model": {
"label": "Reflatten model",
"type": "select",
"default": self._reflatten_model,
"enum": ["plane", "quadratic"],
"section": "Gamma Refinement",
"description": "Gamma quad_fit model used for residual phase trend removal.",
"recommendation": "plane is safer for single-pair production; use quadratic only when a clear curved residual ramp remains.",
},
"reflatten_coh_threshold": {
"label": "Reflatten coherence",
"type": "number",
"default": self._reflatten_coh_threshold,
"step": 0.05,
"min": 0.0,
"max": 1.0,
"section": "Gamma Refinement",
"description": "Coherence threshold used to build the fit mask.",
"recommendation": "Keep the primary fit conservative at 0.70; the backend can retry with a looser fallback.",
},
"reflatten_fallback_coh_threshold": {
"label": "Reflatten fallback coherence",
"type": "number",
"default": self._reflatten_fallback_coh_threshold,
"step": 0.05,
"min": 0.0,
"max": 1.0,
"section": "Gamma Refinement",
"description": "Fallback coherence threshold if the primary reflatten fit does not have enough usable samples.",
},
"reflatten_range_step": {
"label": "Reflatten range step",
"type": "number",
"default": self._reflatten_range_step,
"step": 1,
"min": 1,
"section": "Gamma Refinement",
"description": "Sampling step in range pixels for Gamma quad_fit control points.",
},
"reflatten_azimuth_step": {
"label": "Reflatten azimuth step",
"type": "number",
"default": self._reflatten_azimuth_step,
"step": 1,
"min": 1,
"section": "Gamma Refinement",
"description": "Sampling step in azimuth lines for Gamma quad_fit control points.",
},
"unwrap": {
"label": "执行解缠",
"type": "boolean",
"default": True,
"section": "Execution",
"description": "关闭后仅做到差分干涉图，不做解缠。",
},
"geocode": {
"label": "执行地理编码",
"type": "boolean",
"default": True,
"section": "Execution",
"description": "关闭后不导出地理编码结果。",
},
},
@@ -257,10 +478,36 @@ class PyintEngine(DinsarEngine):
return False
return bool(value)
for key in ("force", "unwrap", "geocode"):
for key in ("force", "unwrap", "geocode", "atmcor", "atmcor_use_for_disp", "reflatten"):
if key in normalized:
normalized[key] = _coerce_bool(normalized[key])
if "geo_interp" in normalized and normalized["geo_interp"] is not None:
value = str(normalized["geo_interp"] or "").strip()
if not value:
normalized.pop("geo_interp", None)
elif value not in {"0", "1"}:
raise ValueError("geo_interp must be 0 or 1.")
else:
normalized["geo_interp"] = value
if "target_grid_size_m" in normalized and str(normalized["target_grid_size_m"] or "").strip() == "":
normalized.pop("target_grid_size_m", None)
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
for key, maximum, label in (
("range_looks", MAX_LOOKS, "距离向多视"),
("azimuth_looks", MAX_LOOKS, "方位向多视"),
@@ -268,6 +515,9 @@ class PyintEngine(DinsarEngine):
):
if key not in normalized or normalized[key] is None:
continue
if str(normalized[key]).strip() == "":
normalized.pop(key, None)
continue
try:
parsed = int(normalized[key])
except (TypeError, ValueError) as exc:
@@ -276,6 +526,59 @@ class PyintEngine(DinsarEngine):
raise ValueError(f"{label}必须在 1 到 {maximum} 之间。")
normalized[key] = parsed
for mode_key, choices in (
("reference_mode", {"none", "coh_median"}),
("deramp_mode", {"none", "plane"}),
("reflatten_model", {"plane", "linear", "quadratic"}),
):
if mode_key not in normalized or normalized[mode_key] is None:
continue
value = str(normalized[mode_key] or "").strip().lower()
if not value:
normalized.pop(mode_key, None)
continue
if mode_key == "reflatten_model" and value == "linear":
value = "plane"
if value not in choices:
supported = ", ".join(sorted(choices))
raise ValueError(f"{mode_key} must be one of: {supported}.")
normalized[mode_key] = value
for threshold_key in (
"coherence_mask_threshold",
"unwrap_coh_threshold",
"reference_coh_threshold",
"deramp_coh_threshold",
"reflatten_coh_threshold",
"reflatten_fallback_coh_threshold",
):
if threshold_key not in normalized or normalized[threshold_key] is None:
continue
if str(normalized[threshold_key]).strip() == "":
normalized.pop(threshold_key, None)
continue
try:
parsed_threshold = float(normalized[threshold_key])
except (TypeError, ValueError) as exc:
raise ValueError(f"{threshold_key} must be a number.") from exc
if parsed_threshold < 0.0 or parsed_threshold > 1.0:
raise ValueError(f"{threshold_key} must be between 0.0 and 1.0.")
normalized[threshold_key] = parsed_threshold
for step_key in ("reflatten_range_step", "reflatten_azimuth_step"):
if step_key not in normalized or normalized[step_key] is None:
continue
if str(normalized[step_key]).strip() == "":
normalized.pop(step_key, None)
continue
try:
parsed_step = int(normalized[step_key])
except (TypeError, ValueError) as exc:
raise ValueError(f"{step_key} must be an integer.") from exc
if parsed_step < 1:
raise ValueError(f"{step_key} must be greater than or equal to 1.")
normalized[step_key] = parsed_step
return normalized
def _has_completed_task_result(self, task_dir: str, profile_code: str) -> bool:
@@ -413,7 +716,10 @@ class PyintEngine(DinsarEngine):
total_tasks = len(task_dirs)
run_started_at = datetime.utcnow()
run_started_at_text = run_started_at.isoformat(timespec="seconds") + "Z"
run_key = f"run_{run_started_at.strftime('%Y%m%dT%H%M%SZ')}_{self.engine_code}_{request.profile}"
managed_run_key = str(extra.get("__managed_run_key") or "").strip()
run_key = managed_run_key or f"run_{run_started_at.strftime('%Y%m%dT%H%M%SZ')}_{self.engine_code}_{request.profile}"
managed_run_dir_override = str(extra.get("__managed_run_dir") or "").strip()
managed_native_output_dir_override = str(extra.get("__managed_native_output_dir") or "").strip()
progress_callback = request.progress_callback
def emit_progress(event_type: str, **payload: Any) -> None:
@@ -426,9 +732,44 @@ class PyintEngine(DinsarEngine):
timeout = max(60, int(request.timeout_seconds or self.default_timeout_seconds))
force = bool(extra.get("force"))
range_looks = int(extra.get("range_looks", DEFAULT_RANGE_LOOKS))
azimuth_looks = int(extra.get("azimuth_looks", DEFAULT_AZIMUTH_LOOKS))
target_grid_size_m = int(extra.get("target_grid_size_m") or 0)
manual_range_looks = extra.get("range_looks")
manual_azimuth_looks = extra.get("azimuth_looks")
parallel_workers = int(extra.get("parallel_workers", DEFAULT_PARALLEL_WORKERS))
dem_resolution_m = self._dem_resolution_m
dem_oversampling = calculate_dem_oversampling(
dem_resolution_m=dem_resolution_m,
target_grid_size_m=target_grid_size_m,
)
dem_lat_ovr = float(dem_oversampling["oversampling"])
dem_lon_ovr = float(dem_oversampling["oversampling"])
unwrap_coh_threshold = float(extra.get("unwrap_coh_threshold", self._default_unwrap_coh_threshold))
coherence_mask_threshold = float(extra.get("coherence_mask_threshold", self._default_product_coh_threshold))
reference_mode = "none"
reference_coh_threshold = float(self._default_reference_coh_threshold)
deramp_mode = "none"
deramp_coh_threshold = float(self._default_deramp_coh_threshold)
gamma_nodata_value = self._gamma_nodata_value
geo_interp = str(extra.get("geo_interp", self._geo_interp) or self._geo_interp).strip()
if geo_interp not in {"0", "1"}:
geo_interp = DEFAULT_GEO_INTERP
atmcor = bool(extra.get("atmcor", self._atmcor_enabled))
atmcor_use_for_disp = bool(extra.get("atmcor_use_for_disp", self._atmcor_use_for_disp)) if atmcor else False
reflatten = bool(extra.get("reflatten", self._reflatten_enabled))
reflatten_model = str(extra.get("reflatten_model", self._reflatten_model) or self._reflatten_model).strip().lower()
if reflatten_model == "linear":
reflatten_model = "plane"
if reflatten_model not in {"plane", "quadratic"}:
reflatten_model = DEFAULT_REFLATTEN_MODEL
reflatten_coh_threshold = float(extra.get("reflatten_coh_threshold", self._reflatten_coh_threshold))
reflatten_fallback_coh_threshold = float(
extra.get(
"reflatten_fallback_coh_threshold",
self._reflatten_fallback_coh_threshold,
)
)
reflatten_range_step = int(extra.get("reflatten_range_step", self._reflatten_range_step))
reflatten_azimuth_step = int(extra.get("reflatten_azimuth_step", self._reflatten_azimuth_step))
unwrap = bool(extra.get("unwrap", True))
geocode = bool(extra.get("geocode", True))
@@ -443,6 +784,57 @@ class PyintEngine(DinsarEngine):
wsl_prepared_dem_path = to_wsl_path(prepared_dem_path) if prepared_dem_path else ""
shared_orbit_context = get_pyint_orbit_context()
def resolve_pair_looks(task_dir: str) -> Dict[str, Any]:
manual_range = int(manual_range_looks) if manual_range_looks is not None else None
manual_azimuth = int(manual_azimuth_looks) if manual_azimuth_looks is not None else None
calculation: Dict[str, Any] = {}
error_text = ""
if target_grid_size_m > 0 and (manual_range is None or manual_azimuth is None):
try:
calculation = calculate_looks_from_task_dir(
task_dir,
target_grid_size_m,
)
except Exception as exc:
error_text = str(exc)
calculation = {
"mode": "fallback_default",
"target_resolution_m": target_grid_size_m,
"error": error_text,
}
elif manual_range is None or manual_azimuth is None:
calculation = {
"mode": "gamma_default_looks",
"target_resolution_m": None,
}
range_looks = manual_range
if range_looks is None:
range_looks = int(calculation.get("range_looks") or DEFAULT_RANGE_LOOKS)
azimuth_looks = manual_azimuth
if azimuth_looks is None:
azimuth_looks = int(calculation.get("azimuth_looks") or DEFAULT_AZIMUTH_LOOKS)
if manual_range is not None or manual_azimuth is not None:
calculation = {
**calculation,
"mode": "manual_override" if calculation else "manual",
"manual_range_looks": manual_range,
"manual_azimuth_looks": manual_azimuth,
}
calculation["resolved_range_looks"] = int(range_looks)
calculation["resolved_azimuth_looks"] = int(azimuth_looks)
calculation["target_grid_size_m"] = int(target_grid_size_m)
return {
"range_looks": int(range_looks),
"azimuth_looks": int(azimuth_looks),
"calculation": calculation,
"error": error_text,
}
task_results: List[Dict[str, Any]] = []
output_dirs: List[str] = []
pairs_processed = 0
@@ -458,11 +850,19 @@ class PyintEngine(DinsarEngine):
slave_date = task_identity["slave_date"]
work_run_root = os.path.normpath(os.path.join(self._work_root, pair_key, run_key))
output_dir = os.path.normpath(os.path.join(self._output_root, pair_key, "runs", run_key, "native"))
run_dir = os.path.normpath(managed_run_dir_override) if managed_run_dir_override else os.path.normpath(
os.path.join(self._output_root, pair_key, "runs", run_key)
)
output_dir = (
os.path.normpath(managed_native_output_dir_override)
if managed_native_output_dir_override
else os.path.join(run_dir, "native")
)
template_root = os.path.normpath(os.path.join(self._template_root, pair_key, run_key))
project_name = build_project_name(pair_key, run_key)
project_dir = os.path.join(work_run_root, project_name)
input_assets_dir = os.path.join(work_run_root, "input_assets")
# Keep input assets outside the run root because the WSL pipeline may delete run_root on --force.
input_assets_dir = os.path.join(self._work_root, pair_key, "input_assets", run_key)
wsl_task_dir = to_wsl_path(task_dir)
wsl_project_dir = to_wsl_path(project_dir)
@@ -663,6 +1063,45 @@ class PyintEngine(DinsarEngine):
else ""
)
look_resolution = resolve_pair_looks(task_dir)
range_looks = int(look_resolution["range_looks"])
azimuth_looks = int(look_resolution["azimuth_looks"])
look_calculation = dict(look_resolution.get("calculation") or {})
look_message = (
f"PyINT looks resolved for {task_alias}: "
f"range={range_looks}, azimuth={azimuth_looks}, "
f"target_grid={target_grid_size_m or 'not_set'}m, mode={look_calculation.get('mode', 'unknown')}"
)
if look_resolution.get("error"):
look_message += f", fallback_reason={look_resolution['error']}"
emit_progress(
"log",
pair_index=pair_index,
pair_total=total_tasks,
task_name=task_name,
task_alias=task_alias,
pair_key=pair_key,
level="WARNING" if look_resolution.get("error") else "INFO",
source="looks",
message=look_message,
)
emit_progress(
"log",
pair_index=pair_index,
pair_total=total_tasks,
task_name=task_name,
task_alias=task_alias,
pair_key=pair_key,
level="INFO",
source="dem",
message=(
f"PyINT DEM oversampling for {task_alias}: "
f"dem_resolution={dem_resolution_m:g}m, target_grid={target_grid_size_m or 'not_set'}m, "
f"dem_lat_ovr={dem_lat_ovr:g}, dem_lon_ovr={dem_lon_ovr:g}, "
f"actual_grid={float(dem_oversampling.get('actual_grid_size_m') or 0.0):g}m"
),
)
cmd_parts = [
f"{quote_shell(self._python)} {quote_shell(self._pipeline_script)} {quote_shell(wsl_task_dir)}",
f"--project-dir {quote_shell(wsl_project_dir)}",
@@ -679,10 +1118,24 @@ class PyintEngine(DinsarEngine):
f"--orbit-policy {quote_shell(self._orbit_policy)}",
f"--range-looks {range_looks}",
f"--azimuth-looks {azimuth_looks}",
f"--dem-resolution-m {dem_resolution_m}",
f"--dem-lat-ovr {dem_lat_ovr}",
f"--dem-lon-ovr {dem_lon_ovr}",
f"--parallel-workers {parallel_workers}",
f"--master-date {quote_shell(master_date)}" if master_date else "",
f"--slave-date {quote_shell(slave_date)}" if slave_date else "",
f"--time-baseline-days {time_baseline_days}",
f"--target-grid-size-m {target_grid_size_m}",
f"--unwrap-coh-threshold {unwrap_coh_threshold}",
f"--coherence-mask-threshold {coherence_mask_threshold}",
f"--geo-interp {quote_shell(geo_interp)}",
f"--gamma-nodata-value {gamma_nodata_value}",
"--reflatten" if reflatten else "--no-reflatten",
f"--reflatten-model {quote_shell(reflatten_model)}",
f"--reflatten-coh-threshold {reflatten_coh_threshold}",
f"--reflatten-fallback-coh-threshold {reflatten_fallback_coh_threshold}",
f"--reflatten-range-step {reflatten_range_step}",
f"--reflatten-azimuth-step {reflatten_azimuth_step}",
f"--input-assets-dir {quote_shell(wsl_input_assets_dir)}" if wsl_input_assets_dir else "",
f"--input-assets-json {quote_shell(wsl_input_assets_json)}" if wsl_input_assets_json else "",
f"--lt1-precise-orbit-enabled {'true' if self._lt1_precise_orbit_enabled else 'false'}",
@@ -695,6 +1148,8 @@ class PyintEngine(DinsarEngine):
f"--lt1-precise-orbit-backup {'true' if self._lt1_precise_orbit_backup else 'false'}",
f"--lt1-precise-orbit-orb-filt-degree {self._lt1_precise_orbit_orb_filt_degree}",
"--unwrap" if unwrap else "--no-unwrap",
"--atmcor" if atmcor else "--no-atmcor",
"--atmcor-use-for-disp" if atmcor_use_for_disp else "--no-atmcor-use-for-disp",
"--geocode" if geocode else "--no-geocode",
]
if self._dem_mode == "local_fabdem" and wsl_fabdem_root:
@@ -712,19 +1167,61 @@ class PyintEngine(DinsarEngine):
cmd_parts.append("--force")
cmd = " ".join(part for part in cmd_parts if part)
rc, stdout, stderr = run_wsl_command(
def _emit_stream_log(level: str, source: str, text: str) -> None:
line = str(text or "").strip()
if not line:
return
max_len = 2000
if len(line) > max_len:
line = line[:max_len] + "...<truncated>"
emit_progress(
"log",
pair_index=pair_index,
pair_total=total_tasks,
task_name=task_name,
task_alias=task_alias,
pair_key=pair_key,
level=level,
source=source,
message=line,
)
rc, stdout, stderr = run_wsl_command_stream(
cmd,
distro=self._distro,
timeout=timeout,
stdout_callback=lambda line: _emit_stream_log("INFO", "stdout", line),
stderr_callback=lambda line: _emit_stream_log("WARNING", "stderr", line),
)
success = rc == 0
error_text = stderr.strip() if stderr else ""
validation_result: Dict[str, Any] = {}
completion_files_result: Dict[str, Any] = {}
primary_file = ""
source_files: List[str] = []
if success:
pairs_processed += 1
os.makedirs(output_dir, exist_ok=True)
write_run_metadata(
output_dir,
{
try:
os.makedirs(output_dir, exist_ok=True)
os.makedirs(run_dir, exist_ok=True)
standard_disp_path = os.path.join(run_dir, "assets", "disp", "disp.tif")
standard_coh_path = os.path.join(run_dir, "assets", "coh", "coh.tif")
if geocode:
validation_sources = [standard_disp_path]
if os.path.isfile(standard_coh_path):
validation_sources.append(standard_coh_path)
validation_result = validate_isce2_result_files(
standard_disp_path,
validation_sources,
)
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"PyINT standard GeoTIFF validation failed: {issue_text}")
primary_file = str(validation_result.get("primary_file") or standard_disp_path)
source_files = list(validation_result.get("source_files") or validation_sources)
run_metadata = {
"run_key": run_key,
"pair_key": pair_key,
"task_name": task_name,
@@ -734,15 +1231,50 @@ class PyintEngine(DinsarEngine):
"source_root": os.path.normpath(request.root_dir),
"task_dir": os.path.normpath(task_dir),
"work_dir": work_run_root,
"output_dir": output_dir,
"output_dir": run_dir,
"native_output_dir": output_dir,
"project_dir": project_dir,
"runtime_id": settings.PYINT_RUNTIME_ID,
"started_at": run_started_at_text,
"finished_at": datetime.utcnow().isoformat(timespec="seconds") + "Z",
"primary_file": primary_file,
"source_files": source_files,
"acceptance": validation_result,
"params": {
"force": force,
"target_grid_size_m": target_grid_size_m,
"dem_resolution_m": dem_resolution_m,
"dem_oversampling": dem_oversampling,
"dem_lat_ovr": dem_lat_ovr,
"dem_lon_ovr": dem_lon_ovr,
"range_looks": range_looks,
"azimuth_looks": azimuth_looks,
"manual_range_looks": manual_range_looks,
"manual_azimuth_looks": manual_azimuth_looks,
"look_calculation": look_calculation,
"parallel_workers": parallel_workers,
"unwrap_coh_threshold": unwrap_coh_threshold,
"coherence_quality_threshold": coherence_mask_threshold,
"reference_mode": reference_mode,
"reference_coh_threshold": reference_coh_threshold,
"deramp_mode": deramp_mode,
"deramp_coh_threshold": deramp_coh_threshold,
"gamma_nodata_value": gamma_nodata_value,
"geo_interp": geo_interp,
"atmcor": atmcor,
"atmcor_use_for_disp": atmcor_use_for_disp,
"reflatten": reflatten,
"reflatten_model": reflatten_model,
"reflatten_coh_threshold": reflatten_coh_threshold,
"reflatten_fallback_coh_threshold": reflatten_fallback_coh_threshold,
"reflatten_range_step": reflatten_range_step,
"reflatten_azimuth_step": reflatten_azimuth_step,
"gamma_native_export": {
"python_data_processing_applied": False,
"coherence_mask_applied": False,
"reference_applied": False,
"deramp_applied": False,
},
"unwrap": unwrap,
"geocode": geocode,
},
@@ -765,10 +1297,24 @@ class PyintEngine(DinsarEngine):
"policy_version": pair_meta.get("policy_version"),
"selection_strategy": pair_meta.get("selection_strategy"),
"input_assets": input_assets_summary,
},
)
output_dirs.append(output_dir)
else:
}
write_run_metadata(run_dir, run_metadata)
write_run_metadata(output_dir, run_metadata)
if geocode and primary_file:
completion_files_result = repair_managed_completion_files(
run_dir,
primary_file=primary_file,
source_files=source_files,
run_meta=run_metadata,
)
output_dirs.append(run_dir)
pairs_processed += 1
except Exception as exc:
success = False
error_text = str(exc)
stderr = (stderr.rstrip() + "\n" + error_text) if stderr else error_text
if not success:
pairs_failed += 1
emit_progress(
@@ -780,7 +1326,7 @@ class PyintEngine(DinsarEngine):
pair_key=pair_key,
success=success,
returncode=rc,
error=stderr.strip() if stderr else "",
error=error_text,
)
task_results.append(
{
@@ -791,13 +1337,45 @@ class PyintEngine(DinsarEngine):
"task_dir": task_dir,
"work_dir": work_run_root,
"project_dir": project_dir,
"output_dir": output_dir,
"run_dir": run_dir,
"output_dir": run_dir,
"native_output_dir": output_dir,
"primary_file": primary_file,
"source_files": source_files,
"acceptance": validation_result,
"completion_files": completion_files_result,
"target_grid_size_m": target_grid_size_m,
"dem_resolution_m": dem_resolution_m,
"dem_oversampling": dem_oversampling,
"dem_lat_ovr": dem_lat_ovr,
"dem_lon_ovr": dem_lon_ovr,
"range_looks": range_looks,
"azimuth_looks": azimuth_looks,
"manual_range_looks": manual_range_looks,
"manual_azimuth_looks": manual_azimuth_looks,
"look_calculation": look_calculation,
"unwrap_coh_threshold": unwrap_coh_threshold,
"coherence_quality_threshold": coherence_mask_threshold,
"reference_mode": reference_mode,
"reference_coh_threshold": reference_coh_threshold,
"deramp_mode": deramp_mode,
"deramp_coh_threshold": deramp_coh_threshold,
"gamma_nodata_value": gamma_nodata_value,
"geo_interp": geo_interp,
"atmcor": atmcor,
"atmcor_use_for_disp": atmcor_use_for_disp,
"gamma_native_export": {
"python_data_processing_applied": False,
"coherence_mask_applied": False,
"reference_applied": False,
"deramp_applied": False,
},
"command": cmd,
"success": success,
"returncode": rc,
"stdout_tail": stdout[-3000:] if stdout else "",
"stderr_tail": stderr[-3000:] if stderr else "",
"error": stderr.strip() if stderr else "",
"error": error_text,
"wsl_task_dir": wsl_task_dir,
"wsl_project_dir": wsl_project_dir,
"wsl_output_dir": wsl_output_dir,
@@ -849,9 +1427,32 @@ class PyintEngine(DinsarEngine):
"started_at": run_started_at_text,
"force": force,
"timeout_seconds": timeout,
"range_looks": range_looks,
"azimuth_looks": azimuth_looks,
"target_grid_size_m": target_grid_size_m,
"dem_resolution_m": dem_resolution_m,
"dem_oversampling": dem_oversampling,
"dem_lat_ovr": dem_lat_ovr,
"dem_lon_ovr": dem_lon_ovr,
"range_looks": last_task_result.get("range_looks"),
"azimuth_looks": last_task_result.get("azimuth_looks"),
"manual_range_looks": manual_range_looks,
"manual_azimuth_looks": manual_azimuth_looks,
"parallel_workers": parallel_workers,
"unwrap_coh_threshold": unwrap_coh_threshold,
"coherence_quality_threshold": coherence_mask_threshold,
"reference_mode": reference_mode,
"reference_coh_threshold": reference_coh_threshold,
"deramp_mode": deramp_mode,
"deramp_coh_threshold": deramp_coh_threshold,
"gamma_nodata_value": gamma_nodata_value,
"geo_interp": geo_interp,
"atmcor": atmcor,
"atmcor_use_for_disp": atmcor_use_for_disp,
"gamma_native_export": {
"python_data_processing_applied": False,
"coherence_mask_applied": False,
"reference_applied": False,
"deramp_applied": False,
},
"unwrap": unwrap,
"geocode": geocode,
"command": last_task_result.get("command", ""),
backend/app/pyint_pipeline/run_lt1_pyint_pipeline.py
+1817 -25
View File
File diff suppressed because it is too large Load Diff
backend/app/routers/dinsar_production.py
+1
View File
@@ -268,6 +268,7 @@ async def submit_run(
else:
job_type = JOB_TYPE_PYINT_RUN
max_attempts = PYINT_PRODUCTION_JOB_MAX_ATTEMPTS
create_managed_run = True
if validation_summary is not None:
validated_task_count = validation_summary.get("task_count", 0)
payload["extra"].update(
backend/app/services/dinsar_production_service.py
+116
View File
@@ -27,6 +27,7 @@ from .workflow_service import workflow_service
TASK_TYPE_DINSAR_PRODUCTION = "IDL_RUN_DINSAR"
TASK_TYPE_ISCE2_DINSAR_PRODUCTION = "ISCE2_RUN"
TASK_TYPE_PYINT_DINSAR_PRODUCTION = "PYINT_RUN"
RUN_STATUS_PENDING = "PENDING"
RUN_STATUS_RUNNING = "RUNNING"
RUN_STATUS_COMPLETED = "COMPLETED"
@@ -75,6 +76,8 @@ def _task_type_for_engine(engine_code: str) -> str:
return TASK_TYPE_DINSAR_PRODUCTION
if normalized == "isce2":
return TASK_TYPE_ISCE2_DINSAR_PRODUCTION
if normalized in {"pyint", "gamma"}:
return TASK_TYPE_PYINT_DINSAR_PRODUCTION
raise ValueError(f"Unsupported engine for D-InSAR production run: {engine_code}")
@@ -84,6 +87,8 @@ def _workflow_name_for_engine(engine_code: str) -> str:
return "dinsar_sarscape_production"
if normalized == "isce2":
return "dinsar_isce2_production"
if normalized in {"pyint", "gamma"}:
return "dinsar_pyint_gamma_production"
raise ValueError(f"Unsupported engine for D-InSAR production run: {engine_code}")
@@ -93,6 +98,8 @@ def _workflow_step_name_for_engine(engine_code: str) -> str:
return RUNS_STEP_NAME
if normalized == "isce2":
return "Execute ISCE2 D-InSAR items"
if normalized in {"pyint", "gamma"}:
return "Execute PyINT/Gamma D-InSAR items"
raise ValueError(f"Unsupported engine for D-InSAR production run: {engine_code}")
@@ -329,6 +336,82 @@ def _execution_dir(item: DinsarProductionRunItemORM, run_key: str) -> str:
return os.path.join(item.results_root_dir, "runs", run_key)
def _first_text(*values: Any) -> str:
for value in values:
text = str(value or "").strip()
if text:
return text
return ""
def _read_json_if_exists(path: str) -> Dict[str, Any]:
text = str(path or "").strip()
if not text or not os.path.isfile(text):
return {}
try:
with open(text, "r", encoding="utf-8") as fp:
payload = json.load(fp)
return payload if isinstance(payload, dict) else {}
except Exception:
return {}
def _maybe_join(base: str, *parts: str) -> str:
text = str(base or "").strip()
if not text:
return ""
return os.path.normpath(os.path.join(text, *parts))
def _build_output_paths(
*,
engine_code: str,
item: DinsarProductionRunItemORM,
run_key: str,
output_dir: str,
manifest_path: Optional[str] = None,
) -> Dict[str, Any]:
run_dir = os.path.normpath(str(output_dir or _execution_dir(item, run_key)))
native_dir = _maybe_join(run_dir, "native")
paths: Dict[str, Any] = {
"run_dir": run_dir,
"native_dir": native_dir,
"assets_dir": _maybe_join(run_dir, "assets"),
"quality_dir": _maybe_join(run_dir, "quality"),
"manifest_path": str(manifest_path or "").strip(),
}
if str(engine_code or "").strip().lower() in {"pyint", "gamma"}:
pair_key = _first_text(item.pair_key, os.path.basename(os.path.dirname(os.path.dirname(run_dir))))
project_name = f"{pair_key}_{run_key}" if pair_key and run_key else ""
work_root = _maybe_join(settings.PYINT_WORK_ROOT, pair_key, run_key)
project_dir = _maybe_join(work_root, project_name) if project_name else ""
summary_payload = _read_json_if_exists(_maybe_join(native_dir, "pyint_run_summary.json"))
summary_project_dir = _first_text(summary_payload.get("project_dir"))
project_dir = summary_project_dir or project_dir
master_date = _first_text(summary_payload.get("master_date"))
slave_date = _first_text(summary_payload.get("slave_date"))
pair_name = f"{master_date}-{slave_date}" if master_date and slave_date else ""
ifgrams_dir = _maybe_join(project_dir, "ifgrams", pair_name) if pair_name else _maybe_join(project_dir, "ifgrams")
paths.update(
{
"work_dir": work_root,
"project_dir": project_dir,
"ifgrams_dir": ifgrams_dir,
"reflatten_dir": _maybe_join(run_dir, "gamma_reflatten"),
"native_reflatten_dir": _maybe_join(native_dir, "reflatten"),
"pyint_summary_path": _maybe_join(native_dir, "pyint_run_summary.json"),
"stdout_log": _maybe_join(work_root, "pyint.stdout.log"),
"stderr_log": _maybe_join(work_root, "pyint.stderr.log"),
}
)
return paths
def _sanitize_pointer_fragment(value: str, default: str) -> str:
text = _SAFE_POINTER_RE.sub("_", str(value or "").strip()).strip("._")
return text or default
@@ -616,6 +699,7 @@ class DinsarProductionService:
)
result = await db.execute(stmt)
runs = result.scalars().all()
run_ids = [run.run_id for run in runs if run.run_id]
pending_reconcile = [
run
for run in runs
@@ -636,6 +720,15 @@ class DinsarProductionService:
) or changed
if changed:
await db.commit()
items_by_run_id: Dict[str, List[DinsarProductionRunItemORM]] = {}
if run_ids:
items_result = await db.execute(
select(DinsarProductionRunItemORM)
.where(DinsarProductionRunItemORM.run_id.in_(run_ids))
.order_by(DinsarProductionRunItemORM.order_index.asc(), DinsarProductionRunItemORM.id.asc())
)
for item in items_result.scalars().all():
items_by_run_id.setdefault(item.run_id, []).append(item)
return {
"runs": [
{
@@ -656,6 +749,29 @@ class DinsarProductionService:
"completed_items": run.completed_items,
"failed_items": run.failed_items,
"skipped_items": run.skipped_items,
"items": [
{
"task_name": item.task_name,
"task_alias": item.task_alias,
"pair_key": item.pair_key,
"status": item.status,
"current_step": item.current_step,
"latest_run_key": item.latest_run_key,
"latest_output_dir": item.latest_output_dir,
"latest_manifest_path": item.latest_manifest_path,
"last_error": item.last_error,
"paths": _build_output_paths(
engine_code=run.engine_code,
item=item,
run_key=str(item.latest_run_key or ""),
output_dir=str(item.latest_output_dir or _execution_dir(item, str(item.latest_run_key or ""))),
manifest_path=item.latest_manifest_path,
)
if item.latest_run_key
else {},
}
for item in items_by_run_id.get(run.run_id, [])[:5]
],
}
for run in runs
],
backend/app/services/job_handlers.py
+81 -1
View File
@@ -2058,6 +2058,25 @@ async def _handle_queued_engine_run(
pair_index = max(0, int(event.get("pair_index") or 0))
task_label = str(event.get("task_alias") or event.get("task_name") or "").strip()
if event_type == "log":
level = str(event.get("level") or "INFO").strip().upper()
if level not in {"DEBUG", "INFO", "WARNING", "ERROR"}:
level = "INFO"
source = str(event.get("source") or "").strip()
message = str(event.get("message") or "").strip()
if not message:
continue
label = task_label or str(progress_state.get("pair_label") or "").strip() or "pair"
prefix = f"{engine_title} {pair_index}/{pair_total} {label}"
if source:
prefix = f"{prefix} {source}"
await task_service.add_log(
job.task_id,
level,
f"{prefix}: {message}",
)
continue
if event_type == "pair_started":
progress = min(
90,
@@ -2496,7 +2515,22 @@ async def _run_wsl_dinsar_production_controller(
if event is None:
return
event_type = str(event.get("event") or "").strip().lower()
if event_type == "pair_started":
if event_type == "log":
level = str(event.get("level") or "INFO").strip().upper()
if level not in {"DEBUG", "INFO", "WARNING", "ERROR"}:
level = "INFO"
source = str(event.get("source") or "").strip()
message = str(event.get("message") or "").strip()
if message:
prefix = f"[{item_index}/{total_items}] {engine_title} {item_label}"
if source:
prefix = f"{prefix} {source}"
await task_service.add_log(
job.task_id,
level,
f"{prefix}: {message}",
)
elif event_type == "pair_started":
progress_state["message"] = (
f"[{engine_code}/{run.profile_code}] Running "
f"{item_index}/{total_items}: {item_label}"
@@ -2892,6 +2926,52 @@ async def _handle_isce2_run(job: SystemJobORM) -> None:
async def _handle_pyint_run(job: SystemJobORM) -> None:
production_run_id = str((job.payload or {}).get("production_run_id") or "").strip()
if production_run_id:
try:
await _run_wsl_dinsar_production_controller(
job,
engine_code="pyint",
engine_title="PyINT/Gamma",
fallback_timeout_seconds=settings.PYINT_DEFAULT_TIMEOUT_SECONDS,
)
except Exception as exc:
latest_message = f"PyINT/Gamma D-InSAR production controller failed: {exc}"
try:
async with AsyncSessionLocal() as db:
run = await dinsar_production_service.get_run(production_run_id, db)
if run is not None and str(run.status or "").strip().upper() not in {"COMPLETED", "FAILED", "CANCELLED"}:
summary_payload = dict(run.summary_json or {})
summary_payload["controller_error"] = str(exc)
await dinsar_production_service.finalize_run(
run,
db=db,
status="FAILED",
summary_payload=summary_payload,
latest_message=latest_message,
)
dinsar_production_service.append_run_log(
run.run_id,
f"[controller-failed] {exc}",
)
except Exception:
pass
try:
current_task = await task_service.get_task(job.task_id)
if current_task and current_task.status not in {"COMPLETED", "FAILED", "CANCELLED"}:
await task_service.add_log(job.task_id, "ERROR", latest_message)
await task_service.update_task(
job.task_id,
status="FAILED",
progress=100,
message=latest_message,
)
except Exception:
pass
raise
return
await _handle_queued_engine_run(
job,
engine_title="PyINT",
backend/app/services/pyint_input_assets_service.py
+1
View File
@@ -316,6 +316,7 @@ def get_pyint_dem_summary() -> Dict[str, Any]:
"hdr_exists": bool(prepared_dem_info.get("hdr_exists")),
"vrt_exists": bool(prepared_dem_info.get("vrt_exists")),
},
"configured_resolution_m": float(getattr(settings, "PYINT_DEM_RESOLUTION_M", 30.0) or 30.0),
"opentopo_dem_type": opentopo_dem_type,
"opentopo_api_key_configured": bool(opentopo_api_key),
"status": status,
backend/app/services/pyint_service.py
+270 -18
View File
@@ -4,6 +4,8 @@ from __future__ import annotations
import os
import re
import shlex
import math
import defusedxml.ElementTree as ET
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
@@ -11,12 +13,83 @@ from typing import Any, Dict, Iterable, List, Optional
from ..config import get_env_text, read_bool_env, settings
from .dinsar_naming import PAIR_META_FILENAME, build_fallback_pair_key, find_json_sidecar
from .wsl_service import run_wsl_command
from .wsl_service import run_wsl_exec
LT1_INPUT_GLOBS = ("LT1*.tar.gz", "LT1*.tiff")
DEFAULT_RANGE_LOOKS = 2
DEFAULT_AZIMUTH_LOOKS = 2
DEFAULT_DEM_RESOLUTION_M = 30.0
DEFAULT_UNWRAP_COH_THRESHOLD = 0.05
DEFAULT_PRODUCT_COH_THRESHOLD = 0.20
DEFAULT_REFERENCE_MODE = "none"
DEFAULT_REFERENCE_COH_THRESHOLD = 0.30
DEFAULT_DERAMP_MODE = "none"
DEFAULT_DERAMP_COH_THRESHOLD = 0.30
DEFAULT_GEO_INTERP = "1"
DEFAULT_ATMCOR_ENABLED = False
DEFAULT_ATMCOR_USE_FOR_DISP = False
DEFAULT_REFLATTEN_ENABLED = True
DEFAULT_REFLATTEN_MODEL = "plane"
DEFAULT_REFLATTEN_COH_THRESHOLD = 0.70
DEFAULT_REFLATTEN_FALLBACK_COH_THRESHOLD = 0.20
DEFAULT_REFLATTEN_RANGE_STEP = 32
DEFAULT_REFLATTEN_AZIMUTH_STEP = 32
DEM_OVERSAMPLING_MIN = 0.25
DEM_OVERSAMPLING_MAX = 16.0
REFERENCE_MODE_CHOICES = {"none", "coh_median"}
DERAMP_MODE_CHOICES = {"none", "plane"}
REFLATTEN_MODEL_CHOICES = {"plane", "linear", "quadratic"}
def _read_default_target_grid_size_m() -> int:
for name in ("PYINT_DEFAULT_TARGET_GRID_SIZE_M",):
text = str(get_env_text(name, "") or "").strip()
if not text:
continue
try:
value = float(text)
except (TypeError, ValueError):
continue
if value > 0:
return int(value)
return 0
def _read_float_env(names: Iterable[str], default: float) -> float:
for name in names:
text = str(get_env_text(name, "") or "").strip()
if not text:
continue
try:
value = float(text)
except (TypeError, ValueError):
continue
if math.isfinite(value):
return value
return float(default)
DEFAULT_TARGET_GRID_SIZE_M = _read_default_target_grid_size_m()
TARGET_GRID_SIZE_MIN_M = 0
TARGET_GRID_SIZE_MAX_M = 100
DEFAULT_DEM_RESOLUTION_M = _read_float_env(("PYINT_DEM_RESOLUTION_M",), DEFAULT_DEM_RESOLUTION_M)
DEFAULT_UNWRAP_COH_THRESHOLD = _read_float_env(
("PYINT_UNWRAP_COH_THRESHOLD",),
DEFAULT_UNWRAP_COH_THRESHOLD,
)
DEFAULT_PRODUCT_COH_THRESHOLD = _read_float_env(
("PYINT_PRODUCT_COH_THRESHOLD", "PYINT_COHERENCE_MASK_THRESHOLD"),
DEFAULT_PRODUCT_COH_THRESHOLD,
)
DEFAULT_REFERENCE_COH_THRESHOLD = _read_float_env(
("PYINT_REFERENCE_COH_THRESHOLD",),
DEFAULT_REFERENCE_COH_THRESHOLD,
)
DEFAULT_DERAMP_COH_THRESHOLD = _read_float_env(
("PYINT_DERAMP_COH_THRESHOLD",),
DEFAULT_DERAMP_COH_THRESHOLD,
)
DEFAULT_PARALLEL_WORKERS = 1
MAX_LOOKS = 32
MAX_PARALLEL_WORKERS = 16
@@ -76,6 +149,180 @@ def normalize_date_text(value: Any) -> str:
return ""
def _local_xml_tag_name(tag: Any) -> str:
text = str(tag or "")
return text.split("}")[-1] if "}" in text else text
def _read_xml_first_parameter(xml_file: str, names: Iterable[str]) -> Optional[str]:
path = os.path.normpath(str(xml_file or "").strip())
if not path or not os.path.isfile(path):
return None
wanted = {str(name or "").strip().lower() for name in names if str(name or "").strip()}
if not wanted:
return None
try:
tree = ET.parse(path)
root = tree.getroot()
except Exception:
return None
for elem in root.iter():
local_name = _local_xml_tag_name(elem.tag).lower()
if local_name in wanted and elem.text and str(elem.text).strip():
return str(elem.text).strip()
return None
def _read_scene_geometry_metadata(metadata_path: str) -> Dict[str, Any]:
source = os.path.normpath(str(metadata_path or "").strip())
range_spacing = _read_xml_first_parameter(
source,
("PixelSpacingRg", "columnSpacing", "slantRange", "range_pixel_spacing"),
)
azimuth_spacing = _read_xml_first_parameter(
source,
("PixelSpacingAz", "rowSpacing", "projectedSpacingAzimuth", "azimuth_pixel_spacing"),
)
incidence_angle = _read_xml_first_parameter(
source,
("IncidenceAngle", "incidence_angle"),
)
if not all((range_spacing, azimuth_spacing, incidence_angle)):
raise ValueError(f"Cannot read range/azimuth spacing and incidence angle from: {source}")
return {
"source": source,
"range_pixel_spacing_m": float(range_spacing),
"azimuth_pixel_spacing_m": float(azimuth_spacing),
"incidence_angle_deg": float(incidence_angle),
}
def _scene_geometry_metadata_candidates(directory: str, patterns: Iterable[str]) -> List[str]:
root = os.path.normpath(str(directory or "").strip())
if not root or not os.path.isdir(root):
return []
candidates: List[str] = []
for pattern in patterns:
candidates.extend(str(path) for path in Path(root).glob(pattern) if path.is_file())
return [
os.path.normpath(path)
for path in sorted(
set(candidates),
key=lambda item: (0 if item.lower().endswith(".sml") else 1, item.lower()),
)
]
def resolve_scene_geometry_metadata_files(scene_dir: str) -> List[str]:
return _scene_geometry_metadata_candidates(
scene_dir,
(
"*.sml",
"*.SML",
"*.meta.xml",
"*.META.XML",
),
)
def resolve_scene_geometry_metadata_file(scene_dir: str) -> str:
candidates = resolve_scene_geometry_metadata_files(scene_dir)
return candidates[0] if candidates else ""
def calculate_looks_from_scene_metadata(
*,
master_metadata: str,
slave_metadata: str,
target_resolution_m: float,
) -> Dict[str, Any]:
target_resolution = float(target_resolution_m)
if target_resolution <= 0:
raise ValueError("target_resolution_m must be greater than 0")
master = _read_scene_geometry_metadata(master_metadata)
slave = _read_scene_geometry_metadata(slave_metadata)
avg_azimuth = (
float(master["azimuth_pixel_spacing_m"]) + float(slave["azimuth_pixel_spacing_m"])
) / 2.0
master_ground_range = float(master["range_pixel_spacing_m"]) / math.sin(
math.radians(float(master["incidence_angle_deg"]))
)
slave_ground_range = float(slave["range_pixel_spacing_m"]) / math.sin(
math.radians(float(slave["incidence_angle_deg"]))
)
avg_ground_range = (master_ground_range + slave_ground_range) / 2.0
range_ratio = target_resolution / avg_ground_range
azimuth_ratio = target_resolution / avg_azimuth
range_looks = max(1, int(math.floor(range_ratio + 0.5)))
azimuth_looks = max(1, int(math.floor(azimuth_ratio + 0.5)))
return {
"mode": "target_grid_size",
"target_resolution_m": target_resolution,
"range_looks": range_looks,
"azimuth_looks": azimuth_looks,
"avg_ground_range_spacing_m": avg_ground_range,
"avg_azimuth_spacing_m": avg_azimuth,
"range_look_ratio": range_ratio,
"azimuth_look_ratio": azimuth_ratio,
"resolved_ground_range_spacing_m": avg_ground_range * range_looks,
"resolved_azimuth_spacing_m": avg_azimuth * azimuth_looks,
"master": master,
"slave": slave,
}
def calculate_looks_from_task_dir(task_dir: str, target_resolution_m: float) -> Dict[str, Any]:
task_root = os.path.normpath(str(task_dir or "").strip())
master_candidates = resolve_scene_geometry_metadata_files(os.path.join(task_root, "master"))
slave_candidates = resolve_scene_geometry_metadata_files(os.path.join(task_root, "slave"))
if not master_candidates or not slave_candidates:
raise ValueError(f"Cannot find SML/meta XML metadata under task: {task_root}")
errors: List[str] = []
for master_metadata in master_candidates:
for slave_metadata in slave_candidates:
try:
return calculate_looks_from_scene_metadata(
master_metadata=master_metadata,
slave_metadata=slave_metadata,
target_resolution_m=target_resolution_m,
)
except Exception as exc:
errors.append(f"{os.path.basename(master_metadata)} + {os.path.basename(slave_metadata)}: {exc}")
detail = "; ".join(errors[:3]) if errors else "unknown metadata parsing error"
raise ValueError(f"Cannot calculate looks from task metadata under {task_root}: {detail}")
def calculate_dem_oversampling(
*,
dem_resolution_m: float,
target_grid_size_m: float,
) -> Dict[str, Any]:
dem_resolution = float(dem_resolution_m or 0.0)
target_grid = float(target_grid_size_m or 0.0)
if not math.isfinite(dem_resolution) or dem_resolution <= 0:
dem_resolution = DEFAULT_DEM_RESOLUTION_M
raw_factor = dem_resolution / target_grid if math.isfinite(target_grid) and target_grid > 0 else None
oversampling = 1.0
actual_grid = dem_resolution / oversampling if oversampling > 0 else dem_resolution
mismatch_ratio = abs(actual_grid - target_grid) / target_grid if target_grid > 0 else None
return {
"mode": "gamma_dem_oversampling",
"dem_resolution_m": dem_resolution,
"target_grid_size_m": target_grid,
"raw_oversampling": raw_factor,
"oversampling": oversampling,
"actual_grid_size_m": actual_grid,
"mismatch_ratio": mismatch_ratio,
"min_oversampling": DEM_OVERSAMPLING_MIN,
"max_oversampling": DEM_OVERSAMPLING_MAX,
}
def slugify_text(value: Any, *, default: str = "item", max_len: int = 96) -> str:
text = _SAFE_TEXT_RE.sub("_", str(value or "").strip()).strip("._")
if not text:
@@ -278,7 +525,14 @@ def _gamma_prefix(gamma_env_script_wsl: str) -> str:
script = str(gamma_env_script_wsl or "").strip()
if not script:
return ""
return f". {quote_shell(script)} >/dev/null 2>&1 && "
return f". {quote_shell(script)} >/dev/null 2>&1 || exit 1; "
def _pyint_path_prefix(pyint_home_wsl: str) -> str:
home = str(pyint_home_wsl or "").strip().rstrip("/")
if not home:
return ""
return f"export PATH={quote_shell(home + '/pyint')}:\"$PATH\" && "
def check_pyint_environment(
@@ -320,7 +574,10 @@ def check_pyint_environment(
def add(name: str, ok: bool, detail: str = "", skipped: bool = False) -> None:
checks.append(PyintCheck(name=name, ok=ok, detail=detail, skipped=skipped))
rc, out, err = run_wsl_command("echo pyint_alive", distro=distro_value, timeout=15)
def run_check(command: str, timeout: int = 30):
return run_wsl_exec(["bash", "-lc", command], distro=distro_value, timeout=timeout)
rc, out, err = run_check("echo pyint_alive", timeout=15)
wsl_ok = rc == 0 and "pyint_alive" in out
add("WSL distro", wsl_ok, out or err or distro_value)
@@ -331,17 +588,15 @@ def check_pyint_environment(
message=f"WSL distro is unavailable: {distro_value}",
)
rc, out, err = run_wsl_command(
rc, out, err = run_check(
f"{quote_shell(python_value)} --version",
distro=distro_value,
timeout=15,
)
add("WSL Python", rc == 0, out or err or python_value)
if pyint_home_wsl:
rc, out, err = run_wsl_command(
rc, out, err = run_check(
f"test -d {quote_shell(pyint_home_wsl)} && echo ok",
distro=distro_value,
timeout=10,
)
add("PYINT_HOME", rc == 0 and "ok" in out, pyint_home_wsl or err)
@@ -349,9 +604,8 @@ def check_pyint_environment(
add("PYINT_HOME", False, "PYINT_HOME is empty")
if pyint_app_wsl:
rc, out, err = run_wsl_command(
rc, out, err = run_check(
f"test -f {quote_shell(pyint_app_wsl)} && echo ok",
distro=distro_value,
timeout=10,
)
add("pyintApp.py", rc == 0 and "ok" in out, pyint_app_wsl or err)
@@ -367,17 +621,15 @@ def check_pyint_environment(
if not path_text:
add(name, False, f"{name} is empty")
continue
rc, out, err = run_wsl_command(
rc, out, err = run_check(
f"test -d {quote_shell(path_text)} && test -w {quote_shell(path_text)} && echo ok",
distro=distro_value,
timeout=10,
)
add(name, rc == 0 and "ok" in out, path_text or err)
if gamma_env_wsl:
rc, out, err = run_wsl_command(
rc, out, err = run_check(
f"test -f {quote_shell(gamma_env_wsl)} && echo ok",
distro=distro_value,
timeout=10,
)
add("GAMMA env script", rc == 0 and "ok" in out, gamma_env_wsl or err)
@@ -385,16 +637,16 @@ def check_pyint_environment(
add("GAMMA env script", True, "Not configured; using current PATH", skipped=True)
gamma_prefix = _gamma_prefix(gamma_env_wsl)
pyint_prefix = _pyint_path_prefix(pyint_home_wsl)
for name, command_name in (
("GAMMA LT1 import", "LT1_import_SLC_from_zipfiles1"),
("GAMMA geocode_back", "geocode_back"),
):
rc, out, err = run_wsl_command(
gamma_prefix + f"command -v {quote_shell(command_name)}",
distro=distro_value,
rc, out, err = run_check(
gamma_prefix + pyint_prefix + f"command -v {quote_shell(command_name)} >/dev/null 2>&1 && echo ok",
timeout=10,
)
add(name, rc == 0 and bool(out.strip()), out or err or command_name)
add(name, rc == 0 and "ok" in out, out or err or command_name)
helper_path = (
Path(__file__).resolve().parent.parent
@@ -412,7 +664,7 @@ def check_pyint_environment(
+ gamma_prefix
+ f"{quote_shell(python_value)} {quote_shell(pyint_app_wsl)} -h >/dev/null"
)
rc, out, err = run_wsl_command(smoke_cmd, distro=distro_value, timeout=60)
rc, out, err = run_check(smoke_cmd, timeout=60)
add("PyINT smoke test", rc == 0, out or err or "pyintApp.py -h")
else:
add("PyINT smoke test", True, "Skipped", skipped=True)
backend/app/services/wsl_service.py
+103 -1
View File
@@ -11,8 +11,10 @@ from __future__ import annotations
import os
import shutil
import subprocess
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
# ---------------------------------------------------------------------------
@@ -101,6 +103,79 @@ def _run_windows_command(
)
def _run_windows_command_stream(
args: List[str],
timeout: int = 30,
env: Optional[Dict[str, str]] = None,
stdout_callback: Optional[Callable[[str], None]] = None,
stderr_callback: Optional[Callable[[str], None]] = None,
) -> Tuple[int, str, str]:
proc_env = os.environ.copy()
if env:
proc_env.update(env)
stdout_parts: List[str] = []
stderr_parts: List[str] = []
try:
proc = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=False,
env=proc_env,
)
except FileNotFoundError:
return -2, "", "wsl.exe not found"
except Exception as exc:
return -3, "", str(exc)
def _drain(stream: Any, parts: List[str], callback: Optional[Callable[[str], None]]) -> None:
for raw_line in iter(stream.readline, b""):
text = _decode_subprocess_output(raw_line)
if not text:
continue
parts.append(text)
if callback:
try:
callback(text)
except Exception:
pass
threads = [
threading.Thread(target=_drain, args=(proc.stdout, stdout_parts, stdout_callback), daemon=True),
threading.Thread(target=_drain, args=(proc.stderr, stderr_parts, stderr_callback), daemon=True),
]
for thread in threads:
thread.start()
timed_out = False
deadline = time.monotonic() + max(1, int(timeout or 30))
while proc.poll() is None:
if time.monotonic() >= deadline:
timed_out = True
try:
proc.kill()
except Exception:
pass
break
time.sleep(0.2)
try:
returncode = proc.wait(timeout=10)
except subprocess.TimeoutExpired:
returncode = -1
for thread in threads:
thread.join(timeout=5)
stdout = "\n".join(stdout_parts)
stderr = "\n".join(stderr_parts)
if timed_out:
timeout_text = f"command timed out ({timeout}s)"
stderr = f"{stderr}\n{timeout_text}".strip()
return -1, stdout, stderr
return returncode, stdout, stderr
def run_wsl_command(
cmd: str,
distro: Optional[str] = None,
@@ -127,6 +202,33 @@ def run_wsl_command(
return -3, "", str(exc)
def run_wsl_command_stream(
cmd: str,
distro: Optional[str] = None,
timeout: int = 30,
env: Optional[Dict[str, str]] = None,
stdout_callback: Optional[Callable[[str], None]] = None,
stderr_callback: Optional[Callable[[str], None]] = None,
) -> Tuple[int, str, str]:
"""Run a WSL bash command and stream decoded stdout/stderr lines to callbacks."""
wsl_exe = _find_wsl_executable()
if not wsl_exe:
return -2, "", "wsl.exe not found"
wsl_args = [wsl_exe]
if distro:
wsl_args += ["-d", distro]
wsl_args += ["bash", "-lc", cmd]
return _run_windows_command_stream(
wsl_args,
timeout=timeout,
env=env,
stdout_callback=stdout_callback,
stderr_callback=stderr_callback,
)
def run_wsl_exec(
argv: Sequence[str],
distro: Optional[str] = None,