From 65a8cc4eac2424b2c7a8ea315ade88bb5c54523a Mon Sep 17 00:00:00 2001 From: Harmon Date: Mon, 15 Jun 2026 16:36:29 +0800 Subject: [PATCH] feat: stage GF3 archives locally before production --- .env.example | 1 + backend/app/config.py | 4 ++ backend/app/routers/monitor.py | 1 + .../gf3_sarscape_production_service.py | 56 ++++++++++++++++++- backend/app/services/job_handlers.py | 2 + ...SCAPE_NATIVE_TO_GEOTIFF_DESIGN_20260530.md | 10 ++++ 6 files changed, 72 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index 800d337..fd8b0fd 100644 --- a/.env.example +++ b/.env.example @@ -62,6 +62,7 @@ UNPACK_SOURCE_DIRS=D:\Archives TASK_POOL_ROOT=D:\Task_Pool DINSAR_TASK_POOL_ROOT=D:\Task_Pool\DInSAR SBAS_TASK_POOL_ROOT=D:\Task_Pool\SBAS +GF3_TASK_POOL_ROOT=D:\Task_Pool\GF3 SOURCE_PRODUCT_DIRS=D:\LuTan1_Image_Pool;D:\Sentinel1_Image_Pool_ZIP;\\DESKTOP-N16HJ84\InSAR_Storage_2\LuTan-1\Archive;\\DESKTOP-N16HJ84\InSAR_Storage_2\Sentinel-1\Archive SENTINEL1_STORAGE_DIRS=D:\Sentinel1_Image_Pool INSAR_STORAGE_DIRS=D:\LuTan1_Image_Pool diff --git a/backend/app/config.py b/backend/app/config.py index d470541..1984965 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -198,6 +198,7 @@ class Settings(BaseSettings): TASK_POOL_ROOT: str = "" DINSAR_TASK_POOL_ROOT: str = "" SBAS_TASK_POOL_ROOT: str = "" + GF3_TASK_POOL_ROOT: str = "" SOURCE_PRODUCT_DIRS: str = "" SENTINEL1_STORAGE_DIRS: str = "" ORBIT_SOURCE_DIRS: str = "" @@ -458,6 +459,8 @@ class Settings(BaseSettings): object.__setattr__(self, "DINSAR_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "DInSAR")) if not self.SBAS_TASK_POOL_ROOT: object.__setattr__(self, "SBAS_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "SBAS")) + if not self.GF3_TASK_POOL_ROOT: + object.__setattr__(self, "GF3_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "GF3")) if not self.SAR_ANALYSIS_READY_ROOT: object.__setattr__( self, @@ -1010,6 +1013,7 @@ class Settings(BaseSettings): os.makedirs(settings.TASK_POOL_ROOT, exist_ok=True) os.makedirs(settings.DINSAR_TASK_POOL_ROOT, exist_ok=True) os.makedirs(settings.SBAS_TASK_POOL_ROOT, exist_ok=True) + os.makedirs(settings.GF3_TASK_POOL_ROOT, exist_ok=True) for path in split_env_paths(settings.GF3_ARCHIVE_SOURCE_DIRS): os.makedirs(path, exist_ok=True) for path in split_env_paths(settings.GF3_SARSCAPE_NATIVE_DIRS): diff --git a/backend/app/routers/monitor.py b/backend/app/routers/monitor.py index 12a9ef3..ef78abb 100644 --- a/backend/app/routers/monitor.py +++ b/backend/app/routers/monitor.py @@ -277,6 +277,7 @@ async def run_gf3_sarscape_produce( "archive_exts": split_env_paths(settings.GF3_ARCHIVE_EXTS), "max_archives_per_run": int(options.max_archives_per_run or 0), "selected_dates": selected_dates, + "local_staging_root": settings.GF3_TASK_POOL_ROOT, "timeout_seconds": int(settings.GF3_SARSCAPE_PRODUCE_TIMEOUT_SECONDS or 0), "keep_extracted": bool(settings.GF3_SARSCAPE_KEEP_EXTRACTED), "auto_standardize": bool(auto_standardize), diff --git a/backend/app/services/gf3_sarscape_production_service.py b/backend/app/services/gf3_sarscape_production_service.py index be42782..3a58cc4 100644 --- a/backend/app/services/gf3_sarscape_production_service.py +++ b/backend/app/services/gf3_sarscape_production_service.py @@ -220,6 +220,39 @@ def _scene_name_from_input(path: Path) -> str: return path.stem +def _same_file_snapshot(src: Path, dst: Path) -> bool: + try: + src_stat = src.stat() + dst_stat = dst.stat() + except OSError: + return False + return int(src_stat.st_size) == int(dst_stat.st_size) and int(src_stat.st_mtime) == int(dst_stat.st_mtime) + + +def _copy_source_to_local_stage( + source_path: Path, + *, + staging_root: Path, + scene_name: str, + log_callback: LogCallback | None = None, +) -> Path: + source_path = source_path.resolve() + scene_stage = staging_root / _safe_slug(scene_name) / "source" + scene_stage.mkdir(parents=True, exist_ok=True) + target_path = scene_stage / source_path.name + if _same_file_snapshot(source_path, target_path): + _emit_log(log_callback, "INFO", f"GF3 SARscape using existing local staged archive: {target_path}") + return target_path + + tmp_path = target_path.with_name(f".{target_path.name}.copying") + if tmp_path.exists(): + tmp_path.unlink() + _emit_log(log_callback, "INFO", f"GF3 SARscape staging source archive locally: {source_path} -> {target_path}") + shutil.copy2(source_path, tmp_path) + os.replace(tmp_path, target_path) + return target_path + + def discover_gf3_sarscape_inputs( source_dirs: list[str] | tuple[str, ...] | None, *, @@ -452,6 +485,8 @@ def run_gf3_sarscape_production( archive_exts: list[str] | None = None, max_archives_per_run: int | None = None, selected_dates: list[str] | None = None, + task_id: str | None = None, + local_staging_root: str | None = None, timeout_seconds: int | None = None, keep_extracted: bool | None = None, log_callback: LogCallback | None = None, @@ -487,10 +522,17 @@ def run_gf3_sarscape_production( timeout = int(timeout_seconds or 0) keep = bool(settings.GF3_SARSCAPE_KEEP_EXTRACTED if keep_extracted is None else keep_extracted) storage_root = Path(os.path.normpath(settings.GF3_STORAGE_DIRS)).resolve() if settings.GF3_STORAGE_DIRS else None + staging_root_text = _clean_text(local_staging_root or getattr(settings, "GF3_TASK_POOL_ROOT", "")) + if not staging_root_text: + staging_root_text = os.path.join(settings.TASK_POOL_ROOT, "GF3") + run_stage_name = _safe_slug(task_id or datetime.now().strftime("%Y%m%dT%H%M%S"), default="manual") + staging_root = Path(os.path.normpath(staging_root_text)).resolve() / "SARscape" / run_stage_name + staging_root.mkdir(parents=True, exist_ok=True) _emit_log(log_callback, "INFO", f"GF3 SARscape source roots: {source_dirs}") _emit_log(log_callback, "INFO", f"GF3 SARscape native root: {native_root_path}") _emit_log(log_callback, "INFO", f"GF3 SARscape standardized root: {storage_root or '(not configured)'}") + _emit_log(log_callback, "INFO", f"GF3 SARscape local staging root: {staging_root}") _emit_log(log_callback, "INFO", f"GF3 SARscape wrapper: {wrapper_path}") _emit_log(log_callback, "INFO", f"GF3 SARscape DEM: {dem}") _emit_log(log_callback, "INFO", f"GF3 SARscape polarizations: {pol_text}") @@ -581,12 +623,19 @@ def run_gf3_sarscape_production( ) continue + local_input_path = _copy_source_to_local_stage( + input_path, + staging_root=staging_root, + scene_name=scene_name, + log_callback=log_callback, + ) + cmd = [ str(wrapper_path), "-config", str(config_path), "-input", - str(input_path), + str(local_input_path), "-output", str(native_root_path), "-dem", @@ -598,7 +647,7 @@ def run_gf3_sarscape_production( if idlrt is not None: cmd.extend(["-idlrt", str(idlrt)]) - _emit_log(log_callback, "INFO", f"GF3 SARscape processing {scene_name}: {input_path}") + _emit_log(log_callback, "INFO", f"GF3 SARscape processing {scene_name}: {local_input_path} (source {input_path})") started = time.monotonic() try: completed = _run_wrapper_command( @@ -629,6 +678,7 @@ def run_gf3_sarscape_production( { "scene_name": scene_name, "input_path": str(input_path), + "local_input_path": str(local_input_path), "scene_dir": str(scene_dir), "status": status, "returncode": int(completed.returncode), @@ -645,6 +695,7 @@ def run_gf3_sarscape_production( { "scene_name": scene_name, "input_path": str(input_path), + "local_input_path": str(local_input_path) if "local_input_path" in locals() else None, "scene_dir": str(scene_dir), "status": "failed", "error": f"timeout after {timeout}s", @@ -657,6 +708,7 @@ def run_gf3_sarscape_production( { "scene_name": scene_name, "input_path": str(input_path), + "local_input_path": str(local_input_path) if "local_input_path" in locals() else None, "scene_dir": str(scene_dir), "status": "failed", "error": str(exc), diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index 94edf67..a5f44d8 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -4052,6 +4052,8 @@ async def _handle_gf3_sarscape_produce(job: SystemJobORM) -> None: archive_exts=payload.get("archive_exts") or [], max_archives_per_run=payload.get("max_archives_per_run"), selected_dates=payload.get("selected_dates") or [], + task_id=job.task_id, + local_staging_root=payload.get("local_staging_root") or settings.GF3_TASK_POOL_ROOT, timeout_seconds=payload.get("timeout_seconds"), keep_extracted=payload.get("keep_extracted"), log_callback=_log_cb, diff --git a/docs/GF3_SARSCAPE_NATIVE_TO_GEOTIFF_DESIGN_20260530.md b/docs/GF3_SARSCAPE_NATIVE_TO_GEOTIFF_DESIGN_20260530.md index 34158ae..5544664 100644 --- a/docs/GF3_SARSCAPE_NATIVE_TO_GEOTIFF_DESIGN_20260530.md +++ b/docs/GF3_SARSCAPE_NATIVE_TO_GEOTIFF_DESIGN_20260530.md @@ -499,3 +499,13 @@ GF3 SARscape production now supports an optional scene-date filter. - `POST /api/monitor/gf3-sarscape-produce` accepts `selected_dates: ["YYYYMMDD"]`. When omitted or empty, production keeps the previous all-date behavior. - The frontend monitor panel exposes a date selector in the GF3 SARscape production section. Operators can select one image date before starting production, or leave it as all dates. - The existing duplicate-result preflight still runs after date filtering. A selected date will not reprocess scenes whose standardized L2 result or complete native `_geo` outputs already exist. + +## 2026-06-15 Local Task_Pool Staging + +GF3 SARscape production no longer passes UNC archives directly to `gf3wrapper.exe`. + +- Source archives may remain on UNC storage for source management. +- Before each wrapper run, the selected archive is copied to `GF3_TASK_POOL_ROOT\SARscape\\\source\`. +- The wrapper receives the local staged archive path as `-input`. +- Native SARscape output still goes to `GF3_SARSCAPE_NATIVE_DIRS`, then standardization writes durable L2 GeoTIFFs to `GF3_STORAGE_DIRS`. +- This avoids network extraction stalls and makes source staging part of the local Task_Pool cleanup domain.