Compare commits

..
2 Commits
8 changed files with 117 additions and 19 deletions
+1
View File
@@ -62,6 +62,7 @@ UNPACK_SOURCE_DIRS=D:\Archives
TASK_POOL_ROOT=D:\Task_Pool TASK_POOL_ROOT=D:\Task_Pool
DINSAR_TASK_POOL_ROOT=D:\Task_Pool\DInSAR DINSAR_TASK_POOL_ROOT=D:\Task_Pool\DInSAR
SBAS_TASK_POOL_ROOT=D:\Task_Pool\SBAS 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 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 SENTINEL1_STORAGE_DIRS=D:\Sentinel1_Image_Pool
INSAR_STORAGE_DIRS=D:\LuTan1_Image_Pool INSAR_STORAGE_DIRS=D:\LuTan1_Image_Pool
+4
View File
@@ -198,6 +198,7 @@ class Settings(BaseSettings):
TASK_POOL_ROOT: str = "" TASK_POOL_ROOT: str = ""
DINSAR_TASK_POOL_ROOT: str = "" DINSAR_TASK_POOL_ROOT: str = ""
SBAS_TASK_POOL_ROOT: str = "" SBAS_TASK_POOL_ROOT: str = ""
GF3_TASK_POOL_ROOT: str = ""
SOURCE_PRODUCT_DIRS: str = "" SOURCE_PRODUCT_DIRS: str = ""
SENTINEL1_STORAGE_DIRS: str = "" SENTINEL1_STORAGE_DIRS: str = ""
ORBIT_SOURCE_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")) object.__setattr__(self, "DINSAR_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "DInSAR"))
if not self.SBAS_TASK_POOL_ROOT: if not self.SBAS_TASK_POOL_ROOT:
object.__setattr__(self, "SBAS_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "SBAS")) 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: if not self.SAR_ANALYSIS_READY_ROOT:
object.__setattr__( object.__setattr__(
self, self,
@@ -1010,6 +1013,7 @@ class Settings(BaseSettings):
os.makedirs(settings.TASK_POOL_ROOT, exist_ok=True) os.makedirs(settings.TASK_POOL_ROOT, exist_ok=True)
os.makedirs(settings.DINSAR_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.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): for path in split_env_paths(settings.GF3_ARCHIVE_SOURCE_DIRS):
os.makedirs(path, exist_ok=True) os.makedirs(path, exist_ok=True)
for path in split_env_paths(settings.GF3_SARSCAPE_NATIVE_DIRS): for path in split_env_paths(settings.GF3_SARSCAPE_NATIVE_DIRS):
+1
View File
@@ -277,6 +277,7 @@ async def run_gf3_sarscape_produce(
"archive_exts": split_env_paths(settings.GF3_ARCHIVE_EXTS), "archive_exts": split_env_paths(settings.GF3_ARCHIVE_EXTS),
"max_archives_per_run": int(options.max_archives_per_run or 0), "max_archives_per_run": int(options.max_archives_per_run or 0),
"selected_dates": selected_dates, "selected_dates": selected_dates,
"local_staging_root": settings.GF3_TASK_POOL_ROOT,
"timeout_seconds": int(settings.GF3_SARSCAPE_PRODUCE_TIMEOUT_SECONDS or 0), "timeout_seconds": int(settings.GF3_SARSCAPE_PRODUCE_TIMEOUT_SECONDS or 0),
"keep_extracted": bool(settings.GF3_SARSCAPE_KEEP_EXTRACTED), "keep_extracted": bool(settings.GF3_SARSCAPE_KEEP_EXTRACTED),
"auto_standardize": bool(auto_standardize), "auto_standardize": bool(auto_standardize),
@@ -220,6 +220,39 @@ def _scene_name_from_input(path: Path) -> str:
return path.stem 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( def discover_gf3_sarscape_inputs(
source_dirs: list[str] | tuple[str, ...] | None, source_dirs: list[str] | tuple[str, ...] | None,
*, *,
@@ -452,6 +485,8 @@ def run_gf3_sarscape_production(
archive_exts: list[str] | None = None, archive_exts: list[str] | None = None,
max_archives_per_run: int | None = None, max_archives_per_run: int | None = None,
selected_dates: list[str] | None = None, selected_dates: list[str] | None = None,
task_id: str | None = None,
local_staging_root: str | None = None,
timeout_seconds: int | None = None, timeout_seconds: int | None = None,
keep_extracted: bool | None = None, keep_extracted: bool | None = None,
log_callback: LogCallback | None = None, log_callback: LogCallback | None = None,
@@ -487,10 +522,17 @@ def run_gf3_sarscape_production(
timeout = int(timeout_seconds or 0) timeout = int(timeout_seconds or 0)
keep = bool(settings.GF3_SARSCAPE_KEEP_EXTRACTED if keep_extracted is None else keep_extracted) 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 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 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 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 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 wrapper: {wrapper_path}")
_emit_log(log_callback, "INFO", f"GF3 SARscape DEM: {dem}") _emit_log(log_callback, "INFO", f"GF3 SARscape DEM: {dem}")
_emit_log(log_callback, "INFO", f"GF3 SARscape polarizations: {pol_text}") _emit_log(log_callback, "INFO", f"GF3 SARscape polarizations: {pol_text}")
@@ -581,12 +623,19 @@ def run_gf3_sarscape_production(
) )
continue continue
local_input_path = _copy_source_to_local_stage(
input_path,
staging_root=staging_root,
scene_name=scene_name,
log_callback=log_callback,
)
cmd = [ cmd = [
str(wrapper_path), str(wrapper_path),
"-config", "-config",
str(config_path), str(config_path),
"-input", "-input",
str(input_path), str(local_input_path),
"-output", "-output",
str(native_root_path), str(native_root_path),
"-dem", "-dem",
@@ -598,7 +647,7 @@ def run_gf3_sarscape_production(
if idlrt is not None: if idlrt is not None:
cmd.extend(["-idlrt", str(idlrt)]) 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() started = time.monotonic()
try: try:
completed = _run_wrapper_command( completed = _run_wrapper_command(
@@ -629,6 +678,7 @@ def run_gf3_sarscape_production(
{ {
"scene_name": scene_name, "scene_name": scene_name,
"input_path": str(input_path), "input_path": str(input_path),
"local_input_path": str(local_input_path),
"scene_dir": str(scene_dir), "scene_dir": str(scene_dir),
"status": status, "status": status,
"returncode": int(completed.returncode), "returncode": int(completed.returncode),
@@ -645,6 +695,7 @@ def run_gf3_sarscape_production(
{ {
"scene_name": scene_name, "scene_name": scene_name,
"input_path": str(input_path), "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), "scene_dir": str(scene_dir),
"status": "failed", "status": "failed",
"error": f"timeout after {timeout}s", "error": f"timeout after {timeout}s",
@@ -657,6 +708,7 @@ def run_gf3_sarscape_production(
{ {
"scene_name": scene_name, "scene_name": scene_name,
"input_path": str(input_path), "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), "scene_dir": str(scene_dir),
"status": "failed", "status": "failed",
"error": str(exc), "error": str(exc),
@@ -46,6 +46,19 @@ def _db_now() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None, microsecond=0) return datetime.now(timezone.utc).replace(tzinfo=None, microsecond=0)
def _date_to_naive_utc(value: Any) -> datetime | None:
text = str(value or "").strip()
if not text:
return None
digits = "".join(ch for ch in text if ch.isdigit())
if len(digits) != 8:
return None
try:
return datetime.strptime(digits, "%Y%m%d")
except ValueError:
return None
def _path_kind(path: str) -> str: def _path_kind(path: str) -> str:
text = str(path or "").strip() text = str(path or "").strip()
if text.startswith("\\\\"): if text.startswith("\\\\"):
@@ -534,6 +547,8 @@ async def _upsert_source_product_asset(
return None return None
standard_dir = Path(standard_dir_text) standard_dir = Path(standard_dir_text)
metadata = scene_manifest.get("metadata") or {} metadata = scene_manifest.get("metadata") or {}
imaging_date = str(metadata.get("imaging_date") or "").strip() or None
acquisition_start = _date_to_naive_utc(imaging_date)
scene_name = scene_manifest.get("scene_name") or standard_dir.name scene_name = scene_manifest.get("scene_name") or standard_dir.name
now = _db_now() now = _db_now()
root = await _find_managed_root_for_path(db, standard_dir_text) root = await _find_managed_root_for_path(db, standard_dir_text)
@@ -565,9 +580,9 @@ async def _upsert_source_product_asset(
"absolute_orbit": metadata.get("absolute_orbit") or metadata.get("orbit_circle"), "absolute_orbit": metadata.get("absolute_orbit") or metadata.get("orbit_circle"),
"relative_orbit": metadata.get("relative_orbit"), "relative_orbit": metadata.get("relative_orbit"),
"orbit_direction": metadata.get("orbit_direction"), "orbit_direction": metadata.get("orbit_direction"),
"acquisition_start_time_utc": None, "acquisition_start_time_utc": acquisition_start,
"acquisition_stop_time_utc": None, "acquisition_stop_time_utc": None,
"imaging_date": metadata.get("imaging_date"), "imaging_date": imaging_date,
"root_ref_id": root.id if root else None, "root_ref_id": root.id if root else None,
"root_path": root.path if root else str(standard_dir.parent), "root_path": root.path if root else str(standard_dir.parent),
"file_path": standard_dir_text, "file_path": standard_dir_text,
@@ -625,6 +640,8 @@ async def _upsert_radar_data(
center_lon, center_lat = _scene_center_from_polygon(polygon) center_lon, center_lat = _scene_center_from_polygon(polygon)
geom = _geom_from_polygon(polygon) geom = _geom_from_polygon(polygon)
metadata = scene_manifest.get("metadata") or {} metadata = scene_manifest.get("metadata") or {}
imaging_date = str(metadata.get("imaging_date") or "").strip() or None
acquisition_start = _date_to_naive_utc(imaging_date)
radar_metadata = _metadata_for_radar(scene_manifest, standard_manifest) radar_metadata = _metadata_for_radar(scene_manifest, standard_manifest)
scene_name = scene_manifest.get("scene_name") or Path(str(scene_manifest.get("native_dir") or "")).name scene_name = scene_manifest.get("scene_name") or Path(str(scene_manifest.get("native_dir") or "")).name
unique_id = f"gf3_sarscape:{scene_name}" unique_id = f"gf3_sarscape:{scene_name}"
@@ -634,7 +651,7 @@ async def _upsert_radar_data(
"unique_id": unique_id, "unique_id": unique_id,
"satellite": "GF3", "satellite": "GF3",
"satellite_family": "GF3", "satellite_family": "GF3",
"imaging_date": metadata.get("imaging_date"), "imaging_date": imaging_date,
"imaging_mode": metadata.get("imaging_mode"), "imaging_mode": metadata.get("imaging_mode"),
"polarization": ",".join( "polarization": ",".join(
pol pol
@@ -644,8 +661,14 @@ async def _upsert_radar_data(
or metadata.get("polarization"), or metadata.get("polarization"),
"scene_center_lon": metadata.get("scene_center_lon") if metadata.get("scene_center_lon") is not None else center_lon, "scene_center_lon": metadata.get("scene_center_lon") if metadata.get("scene_center_lon") is not None else center_lon,
"scene_center_lat": metadata.get("scene_center_lat") if metadata.get("scene_center_lat") is not None else center_lat, "scene_center_lat": metadata.get("scene_center_lat") if metadata.get("scene_center_lat") is not None else center_lat,
"acquisition_time_utc": acquisition_start.isoformat() if acquisition_start else None,
"product_level": "L2", "product_level": "L2",
"product_unique_id": metadata.get("product_unique_id"), "product_unique_id": metadata.get("product_unique_id") or scene_name,
"source_product_token": scene_name,
"acquisition_start_time_utc": acquisition_start,
"acquisition_stop_time_utc": None,
"absolute_orbit": metadata.get("absolute_orbit") or metadata.get("orbit_circle"),
"relative_orbit": metadata.get("relative_orbit"),
"source_format": SOURCE_ASSET_FORMAT, "source_format": SOURCE_ASSET_FORMAT,
"source_product_ref_id": source_product_ref_id, "source_product_ref_id": source_product_ref_id,
"image_data_format": "GEOTIFF", "image_data_format": "GEOTIFF",
+2
View File
@@ -4052,6 +4052,8 @@ async def _handle_gf3_sarscape_produce(job: SystemJobORM) -> None:
archive_exts=payload.get("archive_exts") or [], archive_exts=payload.get("archive_exts") or [],
max_archives_per_run=payload.get("max_archives_per_run"), max_archives_per_run=payload.get("max_archives_per_run"),
selected_dates=payload.get("selected_dates") or [], 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"), timeout_seconds=payload.get("timeout_seconds"),
keep_extracted=payload.get("keep_extracted"), keep_extracted=payload.get("keep_extracted"),
log_callback=_log_cb, log_callback=_log_cb,
@@ -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. - `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 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. - 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\<task_id>\<scene>\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.
+17 -12
View File
@@ -796,23 +796,28 @@ const DataMonitorPanel = ({ apiEndpoint, onTaskStart, readOnly = false, enabled
return; return;
} }
setGf3ScanLoading(true); setGf3ScanLoading(true);
setGf3Message('GF3 扫描启动中...'); setGf3Message('GF3 资产扫描启动中...');
try { try {
const res = await fetch(`${apiEndpoint}/monitor/run-now?target=gf3`, { const inventoryStatus = await getAssetInventoryStatus();
method: 'POST', const gf3SourcePathSet = new Set(config.gf3_archive_source_dirs.map(normalizeComparePath));
credentials: 'include', const rootIds = (inventoryStatus?.states || [])
.filter((item) => item?.inventory_type === 'source_product' && gf3SourcePathSet.has(normalizeComparePath(item?.root_path)))
.map((item) => item.root_ref_id)
.filter((value, index, array) => value && array.indexOf(value) === index);
const data = await scanAssetInventory({
inventory_types: ['source_product'],
root_ids: rootIds,
bind_orbits: false,
}); });
const data = await parseJsonSafe(res, {}); setGf3Message(data.message || 'GF3 资产扫描任务已启动');
if (res.ok) {
setGf3Message(data.message || 'GF3 扫描任务已启动');
if (onTaskStart) { if (onTaskStart) {
onTaskStart(data.task_id, '已触发 GF3 手动扫描...'); onTaskStart(data.task_id, '已触发 GF3 资产扫描...', {
} nonBlocking: true,
} else { taskType: 'SCAN_ASSET_INVENTORY',
setGf3Message(`失败:${data.detail || '未知错误'}`); });
} }
} catch (err) { } catch (err) {
setGf3Message(`失败:${err.message || '未知错误'}`); setGf3Message(`失败:${err?.response?.data?.detail || err.message || '未知错误'}`);
} finally { } finally {
setGf3ScanLoading(false); setGf3ScanLoading(false);
} }