"""Helpers for integrating the external PyINT workflow.""" from __future__ import annotations import os import re import shlex from dataclasses import dataclass, field from datetime import datetime from pathlib import Path 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 LT1_INPUT_GLOBS = ("LT1*.tar.gz", "LT1*.tiff") DEFAULT_RANGE_LOOKS = 2 DEFAULT_AZIMUTH_LOOKS = 2 DEFAULT_PARALLEL_WORKERS = 1 MAX_LOOKS = 32 MAX_PARALLEL_WORKERS = 16 _DATE_TOKEN_RE = re.compile(r"(20\d{6})") _SAFE_TEXT_RE = re.compile(r"[^0-9A-Za-z._-]+") @dataclass class PyintCheck: name: str ok: bool detail: str = "" skipped: bool = False @dataclass class PyintEnvironmentReport: overall_ok: bool checks: List[PyintCheck] = field(default_factory=list) message: str = "" 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 normalize_date_text(value: Any) -> str: text = str(value or "").strip() if not text: return "" match = _DATE_TOKEN_RE.search(re.sub(r"\D", "", text)) if match: return match.group(1) match = _DATE_TOKEN_RE.search(text) if match: return match.group(1) return "" 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: text = default return text[:max_len] def build_project_name(pair_key: str, run_key: str) -> str: return slugify_text(f"{pair_key}_{run_key}", default="pyint_project", max_len=120) def windows_path_to_wsl_mount(path: str) -> str: text = str(path or "").strip().strip('"').strip("'") if not text: return "" normalized = os.path.normpath(text) if normalized.startswith("/"): return normalized.replace("\\", "/") if normalized.startswith("\\\\"): return "" drive, tail = os.path.splitdrive(normalized) if not drive: return normalized.replace("\\", "/") drive_letter = drive.rstrip(":").lower() normalized_tail = tail.replace("\\", "/") return f"/mnt/{drive_letter}/{normalized_tail}" def to_wsl_path(path: str) -> str: return windows_path_to_wsl_mount(path) def quote_shell(value: str) -> str: return shlex.quote(str(value or "")) def discover_lt1_archives(task_dir: str) -> Dict[str, List[str]]: task_path = Path(os.path.normpath(os.path.abspath(str(task_dir or "").strip()))) result: Dict[str, List[str]] = {"master": [], "slave": []} for role in ("master", "slave"): role_dir = task_path / role if not role_dir.is_dir(): continue inputs = [] for pattern in LT1_INPUT_GLOBS: inputs.extend( str(path.resolve()) for path in role_dir.rglob(pattern) if path.is_file() ) result[role] = sorted(set(inputs)) return result def infer_scene_date_from_archives(paths: Iterable[str]) -> str: dates = { date_text for path in paths for date_text in [normalize_date_text(os.path.basename(path))] if date_text } if len(dates) == 1: return next(iter(dates)) return "" def infer_task_identity(task_dir: str) -> Dict[str, Any]: 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) master_date = normalize_date_text(pair_meta.get("master_imaging_date")) slave_date = normalize_date_text(pair_meta.get("slave_imaging_date")) return { "task_name": task_name, "task_alias": task_alias, "pair_key": pair_key, "pair_meta": pair_meta, "master_date": master_date, "slave_date": slave_date, } def build_template_text( *, project_name: str, master_date: str, range_looks: int, azimuth_looks: int, parallel_workers: int, unwrap: bool, geocode: bool, ) -> str: lines = [ f"# Auto-generated for {project_name}", "satelite=LT", f"masterDate={master_date}", f"range_looks={int(range_looks)}", f"azimuth_looks={int(azimuth_looks)}", "download_data=0", "raw2slc_all=1", f"raw2slc_all_parallel={int(parallel_workers)}", "coreg_all=1", f"coreg_all_parallel={int(parallel_workers)}", "select_pairs=0", "diff_all=1", f"diff_all_parallel={int(parallel_workers)}", f"unwrap_all={1 if unwrap else 0}", f"unwrap_all_parallel={int(parallel_workers)}", f"geocode_all={1 if geocode else 0}", f"geocode_all_parallel={int(parallel_workers)}", "geocode_products=hyp3,licsbas", ] return "\n".join(lines) + "\n" def validate_pyint_root_dir(root_dir: str, num_to_process: int = 0) -> 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"PyINT root_dir does not exist or is not a directory: {root_dir}") 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 _iter_child_dirs(directory: str): with os.scandir(directory) 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 if not _missing_task_subdirs(normalized_root): task_dirs = [normalized_root] invalid_candidates: List[Dict[str, Any]] = [] mode = "single_task_dir" else: task_dirs = [] invalid_candidates = [] for entry in _iter_child_dirs(normalized_root): if not entry.name.lower().startswith("task_"): continue missing = _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( "PyINT root_dir must be either a single task directory containing " "'master' and 'slave', or a parent directory containing valid Task_* subdirectories." f"{detail}" ) selected_count = int(num_to_process or 0) if selected_count > 0: task_dirs = task_dirs[:selected_count] return { "root_dir": normalized_root, "mode": mode, "task_dirs": task_dirs, "task_count": len(task_dirs), "invalid_candidates": invalid_candidates, } def resolve_time_baseline_days(master_date: str, slave_date: str, pair_meta: Dict[str, Any]) -> int: raw_days = pair_meta.get("time_baseline_days") try: if raw_days not in (None, ""): return int(raw_days) except (TypeError, ValueError): pass if not master_date or not slave_date: return 0 try: master_dt = datetime.strptime(master_date, "%Y%m%d") slave_dt = datetime.strptime(slave_date, "%Y%m%d") except ValueError: return 0 return (slave_dt - master_dt).days 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 && " def check_pyint_environment( *, enabled: Optional[bool] = None, distro: Optional[str] = None, python_cmd: Optional[str] = None, pyint_home: Optional[str] = None, pyint_app_script: Optional[str] = None, template_root: Optional[str] = None, work_root: Optional[str] = None, output_root: Optional[str] = None, dem_root: Optional[str] = None, gamma_env_script: Optional[str] = None, smoke_test: Optional[bool] = None, ) -> PyintEnvironmentReport: enabled_value = _read_bool_env("PYINT_ENABLED", False) if enabled is None else bool(enabled) if not enabled_value: return PyintEnvironmentReport( overall_ok=False, checks=[PyintCheck(name="PYINT_ENABLED", ok=False, detail="PYINT_ENABLED=false")], message="PyINT is disabled. Set PYINT_ENABLED=true to enable it.", ) distro_value = str(distro or _read_env("PYINT_WSL_DISTRO", settings.ISCE2_WSL_DISTRO)).strip() python_value = str(python_cmd or _read_env("PYINT_WSL_PYTHON", settings.ISCE2_PYTHON)).strip() pyint_home_wsl = to_wsl_path(str(pyint_home or _read_env("PYINT_HOME", ""))) pyint_app_wsl = to_wsl_path(str(pyint_app_script or _read_env("PYINT_APP_SCRIPT", ""))) template_root_wsl = to_wsl_path(str(template_root or _read_env("PYINT_TEMPLATE_ROOT", ""))) work_root_wsl = to_wsl_path(str(work_root or _read_env("PYINT_WORK_ROOT", ""))) output_root_wsl = to_wsl_path(str(output_root or _read_env("PYINT_OUTPUT_ROOT", ""))) dem_root_wsl = to_wsl_path(str(dem_root or _read_env("PYINT_DEM_ROOT", ""))) gamma_env_wsl = to_wsl_path(str(gamma_env_script or _read_env("PYINT_GAMMA_ENV_SCRIPT", ""))) smoke_enabled = _read_bool_env("PYINT_SMOKE_TEST_ENABLED", False) if smoke_test is None else bool(smoke_test) precise_orbit_enabled = _read_bool_env("PYINT_LT1_PRECISE_ORBIT_ENABLED", True) checks: List[PyintCheck] = [] 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) wsl_ok = rc == 0 and "pyint_alive" in out add("WSL distro", wsl_ok, out or err or distro_value) if not wsl_ok: return PyintEnvironmentReport( overall_ok=False, checks=checks, message=f"WSL distro is unavailable: {distro_value}", ) rc, out, err = run_wsl_command( 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( 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) else: add("PYINT_HOME", False, "PYINT_HOME is empty") if pyint_app_wsl: rc, out, err = run_wsl_command( 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) else: add("pyintApp.py", False, "PYINT_APP_SCRIPT is empty") for name, path_text in ( ("PYINT_TEMPLATE_ROOT", template_root_wsl), ("PYINT_WORK_ROOT", work_root_wsl), ("PYINT_OUTPUT_ROOT", output_root_wsl), ("PYINT_DEM_ROOT", dem_root_wsl), ): if not path_text: add(name, False, f"{name} is empty") continue rc, out, err = run_wsl_command( 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( 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) else: add("GAMMA env script", True, "Not configured; using current PATH", skipped=True) gamma_prefix = _gamma_prefix(gamma_env_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, timeout=10, ) add(name, rc == 0 and bool(out.strip()), out or err or command_name) helper_path = ( Path(__file__).resolve().parent.parent / "pyint_pipeline" / "apply_lt1_precise_orbit.py" ) if precise_orbit_enabled: add("LT1 precise orbit bridge helper", helper_path.is_file(), str(helper_path)) else: add("LT1 precise orbit bridge helper", True, "Skipped", skipped=True) if smoke_enabled: smoke_cmd = ( f"export PYTHONPATH={quote_shell(pyint_home_wsl)}:$PYTHONPATH && " + 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) add("PyINT smoke test", rc == 0, out or err or "pyintApp.py -h") else: add("PyINT smoke test", True, "Skipped", skipped=True) required_checks = [check for check in checks if not check.skipped] overall_ok = all(check.ok for check in required_checks) failed_names = [check.name for check in required_checks if not check.ok] message = "All PyINT checks passed." if overall_ok else f"Failed checks: {', '.join(failed_names)}" return PyintEnvironmentReport(overall_ok=overall_ok, checks=checks, message=message)