Checkpoint production workflow updates

This commit is contained in:
2026-06-30 15:25:29 +08:00
parent 19ae3ec37f
commit 9c80b95385
66 changed files with 7639 additions and 267 deletions
+47 -3
View File
@@ -252,6 +252,14 @@ class Settings(BaseSettings):
SAR_ANALYSIS_WORK_ROOT: str = ""
SAR_ANALYSIS_NODATA_VALUE: float = -9999.0
SAR_ANALYSIS_OUTPUT_COG: bool = True
SAR_ANALYSIS_DEM_PATH: str = ""
SAR_ANALYSIS_TARGET_GRID_SIZE_M: float = 30.0
SAR_ANALYSIS_DEM_RESOLUTION_M: float = 30.0
SAR_ANALYSIS_RANGE_LOOKS: int = 6
SAR_ANALYSIS_AZIMUTH_LOOKS: int = 5
SAR_ANALYSIS_SPECKLE_FILTER_ENABLED: bool = True
SAR_ANALYSIS_SPECKLE_FILTER_METHOD: str = "lee"
SAR_ANALYSIS_SPECKLE_FILTER_SIZE: int = 5
SRTM_DEM_DIR: str = ""
GF3_GEO_DEM_PATH: str = ""
@@ -318,7 +326,7 @@ class Settings(BaseSettings):
ISCE2_WSL_DISTRO: str = "Ubuntu-24.04"
ISCE2_PYTHON: str = "/home/administrator/miniconda3/envs/isce2/bin/python"
ISCE2_PROFILE: str = "lt1_stripmap"
ISCE2_DEM_PATH: str = "D:\\SRTM30m\\SRTMDEM_RSP_SARscape.wgs84"
ISCE2_DEM_PATH: str = "D:\\DEM\\SRTMDEM_RSP_SARscape.wgs84"
ISCE2_WORK_ROOT: str = ""
ISCE2_OUTPUT_ROOT: str = ""
ISCE2_PER_TASK_TIMEOUT_SECONDS: int = 43200
@@ -355,7 +363,7 @@ 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_DEM_RESOLUTION_M: float = 90.0
PYINT_OPENTOPO_DEM_TYPE: str = "SRTMGL1"
PYINT_OPENTOPO_API_KEY: str = ""
PYINT_DEM_STRICT: bool = True
@@ -423,6 +431,7 @@ class Settings(BaseSettings):
JOB_WORKER_STALE_RECOVER_INTERVAL: float = 15.0
JOB_WORKER_STALE_RUNNING_SECONDS: int = 300
JOB_WORKER_HEARTBEAT_INTERVAL: float = 5.0
JOB_WORKER_CONCURRENCY: int = 1
JOB_WORKER_ALLOWED_TYPES: str = ""
TIMESERIES_ENABLED: bool = False
@@ -500,6 +509,39 @@ class Settings(BaseSettings):
"SAR_ANALYSIS_NODATA_VALUE",
float(self.SAR_ANALYSIS_NODATA_VALUE if self.SAR_ANALYSIS_NODATA_VALUE is not None else -9999.0),
)
if not self.SAR_ANALYSIS_DEM_PATH:
object.__setattr__(
self,
"SAR_ANALYSIS_DEM_PATH",
self.GAMMA_SBAS_DEM_PATH
or self.PYINT_PREPARED_DEM_PATH
or self.ISCE2_DEM_PATH
or self.IDL_DINSAR_DEM_BASE_FILE,
)
object.__setattr__(
self,
"SAR_ANALYSIS_TARGET_GRID_SIZE_M",
max(1.0, float(self.SAR_ANALYSIS_TARGET_GRID_SIZE_M or 30.0)),
)
object.__setattr__(
self,
"SAR_ANALYSIS_DEM_RESOLUTION_M",
max(1.0, float(self.SAR_ANALYSIS_DEM_RESOLUTION_M or self.SAR_ANALYSIS_TARGET_GRID_SIZE_M or 30.0)),
)
object.__setattr__(self, "SAR_ANALYSIS_RANGE_LOOKS", max(1, int(self.SAR_ANALYSIS_RANGE_LOOKS or 6)))
object.__setattr__(self, "SAR_ANALYSIS_AZIMUTH_LOOKS", max(1, int(self.SAR_ANALYSIS_AZIMUTH_LOOKS or 5)))
filter_method = str(self.SAR_ANALYSIS_SPECKLE_FILTER_METHOD or "lee").strip().lower()
if filter_method in {"", "0", "false", "none", "off", "disabled", "no"}:
filter_method = "none"
elif filter_method not in {"lee"}:
filter_method = "lee"
if not self.SAR_ANALYSIS_SPECKLE_FILTER_ENABLED:
filter_method = "none"
filter_size = max(3, min(99, int(self.SAR_ANALYSIS_SPECKLE_FILTER_SIZE or 5)))
if filter_size % 2 == 0:
filter_size += 1
object.__setattr__(self, "SAR_ANALYSIS_SPECKLE_FILTER_METHOD", filter_method)
object.__setattr__(self, "SAR_ANALYSIS_SPECKLE_FILTER_SIZE", filter_size)
if not self.SRTM_DEM_DIR:
object.__setattr__(self, "SRTM_DEM_DIR", os.path.join(backend_dir, "dem_data"))
object.__setattr__(self, "ASSET_SCAN_PARSE_WORKERS", max(1, int(self.ASSET_SCAN_PARSE_WORKERS or 1)))
@@ -514,6 +556,7 @@ class Settings(BaseSettings):
max(60, int(self.ASSET_SCAN_PARSE_TIMEOUT_SECONDS or 600)),
)
object.__setattr__(self, "ASSET_SCAN_DB_BATCH_SIZE", max(1, int(self.ASSET_SCAN_DB_BATCH_SIZE or 1)))
object.__setattr__(self, "JOB_WORKER_CONCURRENCY", max(1, int(self.JOB_WORKER_CONCURRENCY or 1)))
if not self.GF3_ARCHIVE_SOURCE_DIRS:
object.__setattr__(
self,
@@ -725,7 +768,7 @@ 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_DEM_RESOLUTION_M", max(0.1, float(self.PYINT_DEM_RESOLUTION_M or 90.0)))
object.__setattr__(
self,
"PYINT_UNWRAP_COH_THRESHOLD",
@@ -1291,6 +1334,7 @@ def validate_runtime_config() -> dict[str, Any]:
_check_path(label="WATER_RESULTS_DIR", value=settings.WATER_RESULTS_DIR, errors=errors, warnings=warnings, expect_file=False)
_check_path(label="SAR_ANALYSIS_READY_ROOT", value=settings.SAR_ANALYSIS_READY_ROOT, errors=errors, warnings=warnings, expect_file=False)
_check_path(label="SAR_ANALYSIS_WORK_ROOT", value=settings.SAR_ANALYSIS_WORK_ROOT, errors=errors, warnings=warnings, expect_file=False)
_check_path(label="SAR_ANALYSIS_DEM_PATH", value=settings.SAR_ANALYSIS_DEM_PATH, errors=errors, warnings=warnings, expect_file=True)
_check_path(label="MONITOR_ORBIT_DIR", value=settings.MONITOR_ORBIT_DIR, errors=errors, warnings=warnings, expect_file=False)
for label, value in (
("SOURCE_PRODUCT_DIRS", settings.SOURCE_PRODUCT_DIRS),
+14
View File
@@ -305,6 +305,10 @@ def _normalize_source_bundle_archive_path(source_path: str) -> str:
def _safe_archive_member_name(member_name: str, archive_path: str) -> str:
name = str(member_name or "").replace("\\", "/").strip("/")
while name.startswith("./"):
name = name[2:]
if name in {"", "."}:
return ""
if not name or name.startswith("../") or "/../" in f"/{name}/":
raise ValueError(f"Unsafe archive member path in {archive_path}: {member_name}")
if os.path.isabs(name) or os.path.splitdrive(name)[0]:
@@ -318,6 +322,11 @@ def _extract_archive_to_dir(archive_path: str, dest_dir: str) -> int:
with zipfile.ZipFile(archive_path) as zip_obj:
for info in zip_obj.infolist():
rel_name = _safe_archive_member_name(info.filename, archive_path)
if not rel_name:
if info.is_dir():
os.makedirs(dest_dir, exist_ok=True)
continue
raise ValueError(f"Unsafe ZIP member path: {info.filename}")
dest_path = os.path.abspath(os.path.join(dest_dir, rel_name))
if not dest_path.startswith(os.path.abspath(dest_dir) + os.sep):
raise ValueError(f"Unsafe ZIP member path: {info.filename}")
@@ -335,6 +344,11 @@ def _extract_archive_to_dir(archive_path: str, dest_dir: str) -> int:
with tarfile.open(archive_path, "r:*") as tar_obj:
for member in tar_obj:
rel_name = _safe_archive_member_name(member.name, archive_path)
if not rel_name:
if member.isdir():
os.makedirs(dest_dir, exist_ok=True)
continue
raise ValueError(f"Unsafe TAR member path: {member.name}")
dest_path = os.path.abspath(os.path.join(dest_dir, rel_name))
if not dest_path.startswith(os.path.abspath(dest_dir) + os.sep):
raise ValueError(f"Unsafe TAR member path: {member.name}")
@@ -31,12 +31,12 @@ PathTransform = Callable[[str | Path], Path]
DEFAULT_WINDOWS_DEM_CANDIDATES = (
r"D:\SRTM30m\SRTMDEM_RSP_SARscape.wgs84",
r"D:\SRTM30m\SRTMDEM_RSP_SARscape",
r"D:\DEM\SRTMDEM_RSP_SARscape.wgs84",
r"D:\DEM\SRTMDEM_RSP_SARscape",
)
DEFAULT_WSL_DEM_CANDIDATES = (
"/mnt/d/SRTM30m/SRTMDEM_RSP_SARscape.wgs84",
"/mnt/d/SRTM30m/SRTMDEM_RSP_SARscape",
"/mnt/d/DEM/SRTMDEM_RSP_SARscape.wgs84",
"/mnt/d/DEM/SRTMDEM_RSP_SARscape",
)
DEFAULT_WINDOWS_ORBIT_POOL_CANDIDATES = (r"D:\orbit_pools\isce2",)
DEM_SIDECAR_PROPERTY_NAMES = ("file_name", "metadata_location", "extra_file_name")
+4
View File
@@ -231,6 +231,10 @@ class RadarData(BaseModel):
stack_selection_mode: Optional[str] = None
stack_network_edge_count: Optional[int] = None
stack_network_warnings: Optional[List[str]] = None
lt1_image_produced: bool = False
lt1_image_product: Optional[Dict[str, Any]] = None
lt1_landsar_produced: bool = False
lt1_landsar_product: Optional[Dict[str, Any]] = None
model_config = ConfigDict(from_attributes=True)
@@ -2,7 +2,7 @@
"""Single-scene Gamma preprocessing to analysis-ready GeoTIFF.
The script is intentionally narrower than the full PyINT DInSAR pipeline:
LT source product -> Gamma SLC -> multilook amplitude -> geocode -> GeoTIFF.
LT source product -> Gamma SLC -> multilook amplitude -> geocode -> speckle-filtered dB GeoTIFF.
It is executed inside WSL by backend.app.services.lt_gamma_scene_service.
"""
from __future__ import annotations
@@ -27,6 +27,10 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--pyint-home", required=True)
parser.add_argument("--dem-root", required=True)
parser.add_argument("--prepared-dem-path", default="")
parser.add_argument("--dem-resolution-m", type=float, default=30.0)
parser.add_argument("--target-grid-size-m", type=float, default=30.0)
parser.add_argument("--dem-lat-ovr", type=float, default=0.0)
parser.add_argument("--dem-lon-ovr", type=float, default=0.0)
parser.add_argument("--project-name", required=True)
parser.add_argument("--date", required=True)
parser.add_argument("--satellite-family", default="LT1")
@@ -35,9 +39,175 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--geo-interp", default="1")
parser.add_argument("--nodata-value", type=float, default=-9999.0)
parser.add_argument("--to-db", action="store_true")
parser.add_argument("--speckle-filter-method", default="lee")
parser.add_argument("--speckle-filter-size", type=int, default=5)
parser.add_argument("--speckle-filter-enl", type=float, default=0.0)
return parser.parse_args()
def clamp_float(value: float, minimum: float, maximum: float) -> float:
if not math.isfinite(value):
return minimum
return min(maximum, max(minimum, float(value)))
def format_gamma_number(value: float) -> str:
text = f"{float(value):.6f}".rstrip("0").rstrip(".")
return text or "0"
def calculate_dem_oversampling(
*,
dem_resolution_m: float,
target_grid_size_m: float,
dem_lat_ovr: float,
dem_lon_ovr: float,
) -> dict[str, Any]:
dem_resolution = float(dem_resolution_m or 30.0)
target_grid = float(target_grid_size_m or 30.0)
if dem_resolution <= 0:
dem_resolution = 30.0
if target_grid <= 0:
target_grid = dem_resolution
derived = dem_resolution / target_grid
lat_factor = clamp_float(float(dem_lat_ovr or derived), 0.25, 16.0)
lon_factor = clamp_float(float(dem_lon_ovr or derived), 0.25, 16.0)
actual_grid = dem_resolution / ((lat_factor + lon_factor) / 2.0)
return {
"dem_resolution_m": dem_resolution,
"target_grid_size_m": target_grid,
"derived_oversampling": derived,
"dem_lat_ovr": lat_factor,
"dem_lon_ovr": lon_factor,
"actual_grid_size_m": actual_grid,
}
def meters_per_degree_lon(latitude_deg: float) -> float:
latitude_rad = math.radians(float(latitude_deg))
return max(1.0, 111_320.0 * math.cos(latitude_rad))
def inspect_prepared_dem_path(path_text: str) -> dict[str, str]:
text = str(path_text or "").strip()
if not text:
return {"kind": "", "direct_dem_path": "", "source_dem_path": ""}
path = Path(text)
try:
resolved = path.resolve()
except Exception:
resolved = path
if resolved.is_file() and Path(str(resolved) + ".par").is_file():
return {"kind": "gamma_ready", "direct_dem_path": str(resolved), "source_dem_path": ""}
if resolved.is_file():
return {"kind": "source_dem", "direct_dem_path": "", "source_dem_path": str(resolved)}
return {"kind": "", "direct_dem_path": "", "source_dem_path": str(resolved)}
def read_slc_bbox(
pyint_home: Path,
slc_par: Path,
env: dict[str, str],
*,
margin_deg: float = 0.1,
) -> tuple[float, float, float, float]:
result = subprocess.run(
["SLC_corners", str(slc_par)],
cwd=str(pyint_home),
env=env,
text=True,
capture_output=True,
check=False,
)
if result.returncode != 0:
detail = (result.stderr or result.stdout or "").strip()
raise RuntimeError(f"SLC_corners failed rc={result.returncode}: {detail}")
lines = result.stdout.splitlines()
if len(lines) < 10:
raise RuntimeError(f"Unexpected SLC_corners output for {slc_par}")
lat_line = lines[8].rstrip()
lon_line = lines[9].rstrip()
min_lat = float(lat_line.split(":")[1].split(" max. ")[0])
max_lat = float(lat_line.split(":")[2])
min_lon = float(lon_line.split(":")[1].split(" max. ")[0])
max_lon = float(lon_line.split(":")[2])
margin = max(0.0, float(margin_deg or 0.0))
return min_lon - margin, min_lat - margin, max_lon + margin, max_lat + margin
def build_gamma_dem_from_source(
*,
source_dem: Path,
target_base: Path,
slc_par: Path,
pyint_home: Path,
log_dir: Path,
env: dict[str, str],
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
west, south, east, north = read_slc_bbox(pyint_home, slc_par, env)
log_dir.mkdir(parents=True, exist_ok=True)
source_open = Path(str(source_dem) + ".vrt") if Path(str(source_dem) + ".vrt").is_file() else source_dem
clipped_tif = target_base.with_suffix(".prepared_source_clip.tif")
clipped_aux = Path(str(clipped_tif) + ".aux.xml")
commands: list[dict[str, Any]] = []
commands.append(run_logged(
[
"gdal_translate",
"-projwin",
str(west),
str(north),
str(east),
str(south),
"-of",
"GTiff",
str(source_open),
str(clipped_tif),
],
cwd=target_base.parent,
env=env,
log_dir=log_dir,
stage="clip_prepared_dem",
))
commands.append(run_logged(
[
"makedem.py",
"-d",
str(clipped_tif),
"-p",
"gamma",
"-o",
str(target_base),
],
cwd=target_base.parent,
env=env,
log_dir=log_dir,
stage="convert_prepared_dem",
))
for path in (clipped_tif, clipped_aux):
try:
if path.exists():
path.unlink()
except OSError:
pass
dem_path = Path(str(target_base) + ".dem")
dem_par_path = Path(str(target_base) + ".dem.par")
if not dem_path.is_file() or not dem_par_path.is_file():
raise RuntimeError(f"Prepared source DEM conversion did not create Gamma DEM: {dem_path}")
return (
{
"kind": "source_dem_converted",
"source_dem_path": str(source_dem),
"source_open_path": str(source_open),
"gamma_dem_path": str(dem_path),
"bbox": {"west": west, "south": south, "east": east, "north": north},
},
commands,
)
def run_logged(command: list[str], *, cwd: Path, env: dict[str, str], log_dir: Path, stage: str) -> dict[str, Any]:
log_dir.mkdir(parents=True, exist_ok=True)
stdout_path = log_dir / f"{stage}.stdout.log"
@@ -74,6 +244,48 @@ def read_gamma_par(path: Path, key: str) -> str:
raise KeyError(f"Cannot read {key} from {path}")
def calculate_dem_oversampling_from_gamma_dem(
*,
dem_par_path: Path,
target_grid_size_m: float,
explicit_dem_lat_ovr: float,
explicit_dem_lon_ovr: float,
) -> dict[str, Any]:
target_grid = max(1.0, float(target_grid_size_m or 30.0))
post_lat_deg = abs(float(read_gamma_par(dem_par_path, "post_lat")))
post_lon_deg = abs(float(read_gamma_par(dem_par_path, "post_lon")))
corner_lat = float(read_gamma_par(dem_par_path, "corner_lat"))
nlines = int(float(read_gamma_par(dem_par_path, "nlines")))
center_lat = corner_lat - (post_lat_deg * max(0, nlines - 1) / 2.0)
lat_spacing_m = post_lat_deg * 111_320.0
lon_spacing_m = post_lon_deg * meters_per_degree_lon(center_lat)
derived_lat = lat_spacing_m / target_grid
derived_lon = lon_spacing_m / target_grid
lat_factor = clamp_float(float(explicit_dem_lat_ovr or derived_lat), 0.25, 16.0)
lon_factor = clamp_float(float(explicit_dem_lon_ovr or derived_lon), 0.25, 16.0)
actual_lat_m = lat_spacing_m / lat_factor
actual_lon_m = lon_spacing_m / lon_factor
return {
"dem_resolution_m": (lat_spacing_m + lon_spacing_m) / 2.0,
"target_grid_size_m": target_grid,
"derived_oversampling": (derived_lat + derived_lon) / 2.0,
"derived_dem_lat_ovr": derived_lat,
"derived_dem_lon_ovr": derived_lon,
"dem_lat_ovr": lat_factor,
"dem_lon_ovr": lon_factor,
"actual_grid_size_m": (actual_lat_m + actual_lon_m) / 2.0,
"actual_lat_grid_size_m": actual_lat_m,
"actual_lon_grid_size_m": actual_lon_m,
"source_dem_post_lat_deg": post_lat_deg,
"source_dem_post_lon_deg": post_lon_deg,
"source_dem_lat_spacing_m": lat_spacing_m,
"source_dem_lon_spacing_m": lon_spacing_m,
"source_dem_center_lat": center_lat,
"source_dem_par_path": str(dem_par_path),
}
def discover_lt_inputs(source_path: Path, date: str) -> list[Path]:
patterns = [f"LT1*{date}*.tar.gz", f"LT1*{date}*.tiff", f"LT1*{date}*.tif"]
if source_path.is_file():
@@ -119,15 +331,18 @@ def write_template(
range_looks: int,
azimuth_looks: int,
geo_interp: str,
prepared_dem_path: str,
dem_path: str,
prepared_dem_source: str,
dem_oversampling: dict[str, Any],
) -> None:
lines = [
"satelite = LT",
f"masterDate = {date}",
f"range_looks = {range_looks}",
f"azimuth_looks = {azimuth_looks}",
"dem_lat_ovr = 0.5",
"dem_lon_ovr = 0.5",
f"target_grid_size_m = {format_gamma_number(float(dem_oversampling.get('target_grid_size_m') or 0.0))}",
f"dem_lat_ovr = {format_gamma_number(float(dem_oversampling.get('dem_lat_ovr') or 1.0))}",
f"dem_lon_ovr = {format_gamma_number(float(dem_oversampling.get('dem_lon_ovr') or 1.0))}",
"Simphase_rpos = -",
"Simphase_azpos = -",
"Simphase_rwin = 256",
@@ -135,24 +350,153 @@ def write_template(
"Simphase_thresh = -",
f"geo_interp = {geo_interp}",
]
dem = str(prepared_dem_path or "").strip()
dem = str(dem_path or "").strip()
if dem and Path(dem).is_file() and Path(dem + ".par").is_file():
lines.append(f"DEM = {dem}")
source = str(prepared_dem_source or "").strip()
if source:
lines.append(f"prepared_dem_source = {source}")
template_path.parent.mkdir(parents=True, exist_ok=True)
template_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def convert_to_db_geotiff(source_tif: Path, target_tif: Path, nodata_value: float) -> dict[str, Any]:
def normalize_speckle_filter_method(method: str) -> str:
text = str(method or "").strip().lower()
if text in {"", "0", "false", "none", "off", "disabled", "no"}:
return "none"
if text in {"lee", "lee_filter"}:
return "lee"
raise ValueError(f"Unsupported speckle filter method: {method}")
def normalize_speckle_filter_size(size: int | float | str) -> int:
try:
value = int(float(size or 5))
except Exception:
value = 5
value = max(3, min(99, value))
if value % 2 == 0:
value += 1
return value
def moving_sum_axis(values: Any, size: int, axis: int) -> Any:
import numpy as np
radius = size // 2
pad_width = [(0, 0)] * values.ndim
pad_width[axis] = (radius, size - 1 - radius)
padded = np.pad(values, pad_width, mode="edge")
cumulative = np.cumsum(padded, axis=axis, dtype="float64")
zero_shape = list(cumulative.shape)
zero_shape[axis] = 1
cumulative = np.concatenate([np.zeros(zero_shape, dtype="float64"), cumulative], axis=axis)
length = values.shape[axis]
start = np.arange(0, length)
end = np.arange(size, size + length)
return np.take(cumulative, end, axis=axis) - np.take(cumulative, start, axis=axis)
def box_sum(values: Any, size: int) -> Any:
return moving_sum_axis(moving_sum_axis(values, size, axis=0), size, axis=1)
def local_power_stats(data: Any, valid: Any, window_size: int) -> tuple[Any, Any, Any]:
import numpy as np
values = np.where(valid, data, 0.0).astype("float64", copy=False)
weights = valid.astype("float64", copy=False)
count = box_sum(weights, window_size)
power_sum = box_sum(values, window_size)
power_sq_sum = box_sum(values * values, window_size)
mean = np.divide(power_sum, count, out=np.zeros_like(power_sum), where=count > 0)
mean_sq = np.divide(power_sq_sum, count, out=np.zeros_like(power_sq_sum), where=count > 0)
variance = np.maximum(mean_sq - mean * mean, 0.0)
valid_fraction = count / float(window_size * window_size)
return mean, variance, valid_fraction
def apply_speckle_filter_power(
data: Any,
invalid: Any,
*,
method: str,
window_size: int,
equivalent_number_of_looks: float = 0.0,
) -> tuple[Any, dict[str, Any]]:
import numpy as np
normalized_method = normalize_speckle_filter_method(method)
normalized_size = normalize_speckle_filter_size(window_size)
record: dict[str, Any] = {
"enabled": normalized_method != "none",
"method": normalized_method,
"window_size": normalized_size,
"equivalent_number_of_looks": float(equivalent_number_of_looks or 0.0),
"domain": "linear_power",
}
if normalized_method == "none":
return data, record
valid = ~invalid
valid_count = int(np.count_nonzero(valid))
record["valid_pixels"] = valid_count
if valid_count == 0:
record["enabled"] = False
record["warning"] = "no valid positive pixels to filter"
return data, record
local_mean, local_variance, valid_fraction = local_power_stats(data, valid, normalized_size)
stats_mask = valid & np.isfinite(local_variance) & np.isfinite(local_mean) & (valid_fraction > 0.0)
enl = float(equivalent_number_of_looks or 0.0)
if math.isfinite(enl) and enl > 0:
noise_variance = np.maximum((local_mean * local_mean) / enl, 0.0)
weight = np.divide(
np.maximum(local_variance - noise_variance, 0.0),
local_variance,
out=np.zeros_like(local_variance),
where=local_variance > 0,
)
record["noise_variance_model"] = "local_mean_squared_over_enl"
else:
noise_samples = local_variance[stats_mask]
global_noise_variance = float(np.nanmedian(noise_samples)) if noise_samples.size else 0.0
if not math.isfinite(global_noise_variance) or global_noise_variance <= 0:
record["enabled"] = False
record["warning"] = "local variance estimate is zero; kept unfiltered power values"
return data, record
noise_variance = global_noise_variance
weight = np.divide(
local_variance,
local_variance + noise_variance,
out=np.zeros_like(local_variance),
where=(local_variance + noise_variance) > 0,
)
record["noise_variance"] = global_noise_variance
record["noise_variance_model"] = "global_median_local_variance"
filtered = local_mean + weight * (data.astype("float64", copy=False) - local_mean)
filtered = np.where(np.isfinite(filtered) & (filtered > 0), filtered, data)
output = data.astype("float32", copy=True)
output[stats_mask] = filtered[stats_mask].astype("float32")
return output, record
def convert_to_db_geotiff(
source_tif: Path,
target_tif: Path,
nodata_value: float,
*,
speckle_filter_method: str = "none",
speckle_filter_size: int = 5,
speckle_filter_enl: float = 0.0,
) -> dict[str, Any]:
filter_method = normalize_speckle_filter_method(speckle_filter_method)
filter_size = normalize_speckle_filter_size(speckle_filter_size)
try:
import numpy as np
import rasterio
except Exception as exc:
shutil.copy2(source_tif, target_tif)
return {
"target": str(target_tif),
"backscatter_unit": "gamma_mli_power",
"warning": f"rasterio/numpy unavailable; kept power values: {exc}",
}
raise RuntimeError(f"rasterio/numpy unavailable; cannot create filtered dB GeoTIFF: {exc}") from exc
with rasterio.open(source_tif) as src:
data = src.read(1).astype("float32")
@@ -163,14 +507,23 @@ def convert_to_db_geotiff(source_tif: Path, target_tif: Path, nodata_value: floa
if src_nodata is not None:
invalid |= data == src_nodata
invalid |= data <= 0
filtered_data, speckle_filter = apply_speckle_filter_power(
data,
invalid,
method=filter_method,
window_size=filter_size,
equivalent_number_of_looks=float(speckle_filter_enl or 0.0),
)
invalid |= ~np.isfinite(filtered_data)
invalid |= filtered_data <= 0
db_data = np.full(data.shape, nodata_value, dtype="float32")
db_data[~invalid] = (10.0 * np.log10(data[~invalid])).astype("float32")
db_data[~invalid] = (10.0 * np.log10(filtered_data[~invalid])).astype("float32")
profile.update(dtype="float32", count=1, nodata=nodata_value, compress="deflate")
target_tif.parent.mkdir(parents=True, exist_ok=True)
with rasterio.open(target_tif, "w", **profile) as dst:
dst.write(db_data, 1)
return {"target": str(target_tif), "backscatter_unit": "gamma_mli_db"}
return {"target": str(target_tif), "backscatter_unit": "gamma_mli_db", "speckle_filter": speckle_filter}
def main() -> int:
@@ -201,6 +554,18 @@ def main() -> int:
env["PATH"] = f"{pyint_home / 'pyint'}:{env.get('PATH', '')}"
staged_inputs = stage_lt_inputs(source_path, download_dir, date)
dem_oversampling = calculate_dem_oversampling(
dem_resolution_m=float(args.dem_resolution_m or 30.0),
target_grid_size_m=float(args.target_grid_size_m or 30.0),
dem_lat_ovr=float(args.dem_lat_ovr or 0.0),
dem_lon_ovr=float(args.dem_lon_ovr or 0.0),
)
prepared_dem = inspect_prepared_dem_path(args.prepared_dem_path)
if not prepared_dem.get("kind"):
raise RuntimeError(f"A prepared DEM is required for LT analysis GeoTIFF production: {args.prepared_dem_path}")
dem_path = prepared_dem.get("direct_dem_path") or ""
prepared_dem_conversion: dict[str, Any] | None = None
template_path = template_dir / f"{project_name}.template"
write_template(
template_path=template_path,
@@ -208,7 +573,9 @@ def main() -> int:
range_looks=max(1, int(args.range_looks)),
azimuth_looks=max(1, int(args.azimuth_looks)),
geo_interp=str(args.geo_interp or "1"),
prepared_dem_path=args.prepared_dem_path,
dem_path=dem_path,
prepared_dem_source=str(prepared_dem.get("source_dem_path") or ""),
dem_oversampling=dem_oversampling,
)
commands: list[dict[str, Any]] = []
@@ -221,6 +588,44 @@ def main() -> int:
stage="down2slc_lt1",
)
)
if prepared_dem.get("kind") == "source_dem":
slc_par = project_dir / "SLC" / date / f"{date}.slc.par"
if not slc_par.is_file():
raise FileNotFoundError(f"Gamma SLC parameter file missing before DEM conversion: {slc_par}")
dem_target_base = dem_root / project_name / project_name
dem_target_base.parent.mkdir(parents=True, exist_ok=True)
prepared_dem_conversion, dem_commands = build_gamma_dem_from_source(
source_dem=Path(str(prepared_dem.get("source_dem_path"))),
target_base=dem_target_base,
slc_par=slc_par,
pyint_home=pyint_home,
log_dir=log_dir,
env=env,
)
commands.extend(dem_commands)
dem_path = str(Path(str(dem_target_base) + ".dem"))
dem_par_path = Path(str(dem_path) + ".par") if dem_path else Path()
if dem_path and dem_par_path.is_file():
dem_oversampling = calculate_dem_oversampling_from_gamma_dem(
dem_par_path=dem_par_path,
target_grid_size_m=float(args.target_grid_size_m or 30.0),
explicit_dem_lat_ovr=float(args.dem_lat_ovr or 0.0),
explicit_dem_lon_ovr=float(args.dem_lon_ovr or 0.0),
)
write_template(
template_path=template_path,
date=date,
range_looks=max(1, int(args.range_looks)),
azimuth_looks=max(1, int(args.azimuth_looks)),
geo_interp=str(args.geo_interp or "1"),
dem_path=dem_path,
prepared_dem_source=str(prepared_dem.get("source_dem_path") or ""),
dem_oversampling=dem_oversampling,
)
commands.append(
run_logged(
[sys.executable, str(pyint_home / "pyint" / "generate_rdc_dem.py"), project_name],
@@ -286,9 +691,25 @@ def main() -> int:
raise RuntimeError(f"data2geotiff did not create output: {power_tif}")
final_tif = output_dir / "analysis_ready.tif"
conversion = convert_to_db_geotiff(power_tif, final_tif, float(args.nodata_value)) if args.to_db else {
speckle_filter_config = {
"method": normalize_speckle_filter_method(args.speckle_filter_method),
"window_size": normalize_speckle_filter_size(args.speckle_filter_size),
}
conversion = convert_to_db_geotiff(
power_tif,
final_tif,
float(args.nodata_value),
speckle_filter_method=args.speckle_filter_method,
speckle_filter_size=args.speckle_filter_size,
speckle_filter_enl=float(args.speckle_filter_enl or (range_looks * max(1, int(args.azimuth_looks)))),
) if args.to_db else {
"target": str(final_tif),
"backscatter_unit": "gamma_mli_power",
"speckle_filter": {
"enabled": False,
**speckle_filter_config,
"warning": "not applied because --to-db was disabled",
},
}
if not args.to_db:
shutil.copy2(power_tif, final_tif)
@@ -313,6 +734,24 @@ def main() -> int:
"geo_amp": str(geo_amp),
},
"looks": {"range": range_looks, "azimuth": max(1, int(args.azimuth_looks))},
"speckle_filter": conversion.get("speckle_filter"),
"processing_steps": {
"multilook": {
"enabled": True,
"range_looks": range_looks,
"azimuth_looks": max(1, int(args.azimuth_looks)),
},
"geocode": {"enabled": True, "interpolation": str(args.geo_interp or "1")},
"speckle_filter": conversion.get("speckle_filter"),
"db_conversion": {"enabled": bool(args.to_db), "unit": conversion.get("backscatter_unit")},
},
"dem": {
"prepared_dem_path": str(args.prepared_dem_path or "").strip(),
"prepared_dem_kind": prepared_dem.get("kind"),
"gamma_dem_path": dem_path,
"conversion": prepared_dem_conversion,
"oversampling": dem_oversampling,
},
"commands": commands,
"conversion": conversion,
}
+2
View File
@@ -14,6 +14,7 @@ from . import (
hazard,
health,
idl,
landsar_lt1_production,
license,
logs,
monitor,
@@ -55,6 +56,7 @@ def include_all_routers(router: APIRouter) -> None:
router.include_router(dinsar.router)
router.include_router(dinsar_products.router)
router.include_router(dinsar_production.router)
router.include_router(landsar_lt1_production.router)
router.include_router(sbas_insar_production.router)
router.include_router(sbas_insar_products.router)
router.include_router(timeseries_production.router)
+23 -79
View File
@@ -16,7 +16,6 @@ from .dependencies import _require_admin, _get_current_user, _validate_export_pa
from ..models import AuthUserORM
from ..services import envi_service
from ..services.job_queue_service import job_queue_service
from ..services.result_catalog_service import result_catalog_service
from ..services.task_service import task_service
router = APIRouter()
@@ -61,38 +60,6 @@ class SarscapeSbasInspectRequest(BaseModel):
timeout_seconds: Optional[int] = Field(default=120, ge=10, le=600)
def _normalize_existing_dir(path: Optional[str]) -> Optional[str]:
text = str(path or "").strip()
if not text:
return None
normalized = os.path.normpath(os.path.abspath(text))
if not os.path.isdir(normalized):
return None
return normalized
def _dedupe_publish_roots(*paths: Optional[str]) -> list[str]:
ordered: list[str] = []
for raw_path in paths:
normalized = _normalize_existing_dir(raw_path)
if not normalized:
continue
if any(
normalized == existing or normalized.startswith(existing + os.sep)
for existing in ordered
):
continue
ordered = [
existing
for existing in ordered
if not existing.startswith(normalized + os.sep)
]
ordered.append(normalized)
return ordered
# ---------------------------------------------------------------------------
# Job queue helper
# ---------------------------------------------------------------------------
@@ -292,7 +259,7 @@ async def get_task_overview_endpoint(
return result
@router.post("/idl/extract-disp")
@router.post("/idl/extract-disp", status_code=202)
async def extract_disp_endpoint(
request: ExtractDispRequest,
admin_user: AuthUserORM = Depends(_require_admin),
@@ -303,52 +270,29 @@ async def extract_disp_endpoint(
if request.dest_dir:
_validate_export_path(request.dest_dir, "dest_dir")
try:
result = await asyncio.to_thread(
envi_service.extract_disp_results, request.root_dir, request.dest_dir
payload = {
"root_dir": request.root_dir,
"dest_dir": request.dest_dir,
}
task_id = await task_service.create_task(
"EXTRACT_DINSAR_PRODUCTS",
"D-InSAR 结果提取与登记",
params=payload,
db=db,
)
job_id = await job_queue_service.create_job(
"EXTRACT_DINSAR_PRODUCTS",
payload=payload,
task_id=task_id,
db=db,
)
await db.commit()
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
raise HTTPException(status_code=409, detail=str(exc)) from exc
publish_roots = _dedupe_publish_roots(result.get("target_dir"))
catalog_status: Dict[str, Any] = {
"attempted": False,
"status": "skipped",
"source_directories": publish_roots,
"message": "catalog publish skipped",
return {
"queued": True,
"task_id": task_id,
"job_id": job_id,
"message": "D-InSAR 结果提取与登记任务已入队",
}
if publish_roots:
try:
catalog_status["attempted"] = True
publish_result = await result_catalog_service.publish_from_sources(
db,
publish_roots,
)
rebuild_result = None
if int(publish_result.get("processed", 0) or 0) > 0:
rebuild_result = await result_catalog_service.rebuild_catalog(
db,
full_rebuild=True,
)
catalog_status = {
"attempted": True,
"status": "ok",
"source_directories": publish_roots,
"publish": publish_result,
"rebuild": rebuild_result,
"message": (
"catalog published and rebuilt"
if rebuild_result is not None
else "catalog publish finished with no rebuild needed"
),
}
except Exception as exc:
await db.rollback()
catalog_status = {
"attempted": True,
"status": "error",
"source_directories": publish_roots,
"message": str(exc),
}
result["catalog"] = catalog_status
return result
@@ -0,0 +1,485 @@
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from ..database import get_db
from ..models import AuthUserORM, RadarDataORM, SARSceneGeoORM
from ..services.job_handlers import JOB_TYPE_SAR_SCENE_PREPROCESS
from ..services.job_queue_service import job_queue_service
from ..services.landsar_lt1_production_service import landsar_lt1_production_service
from ..services.task_service import task_service
from ..utils import normalize_satellite_family
from .dependencies import _add_operation_audit_log, _get_current_user, _require_admin
router = APIRouter()
STATIC_ASSET_CACHE_HEADERS = {"Cache-Control": "public, max-age=31536000, immutable"}
class LandsarLt1ImageProductionRequest(BaseModel):
source_asset_ids: List[int] = Field(default_factory=list)
radar_data_ids: List[int] = Field(default_factory=list)
mode: str = "scene"
task_name: Optional[str] = None
@field_validator("mode")
@classmethod
def _validate_mode(cls, value):
mode = str(value or "scene").strip().lower()
if mode == "stack":
mode = "batch"
if mode not in {"scene", "batch"}:
raise ValueError("mode must be scene or batch")
return mode
def _dedupe_positive_ids(values: List[int]) -> List[int]:
result: List[int] = []
for value in values or []:
try:
parsed = int(value)
except (TypeError, ValueError):
continue
if parsed > 0 and parsed not in result:
result.append(parsed)
return result
def _scene_product_marker(scene: SARSceneGeoORM) -> Dict[str, Any]:
return {
"scene_id": scene.id,
"radar_data_id": scene.radar_data_id,
"product_id": f"sar_scene_geo:{scene.id}",
"product_family": "lt1_analysis_ready_geotiff",
"engine_code": scene.analysis_engine,
"profile_code": scene.analysis_profile,
"analysis_tif_path": scene.analysis_tif_path,
"analysis_dir": scene.analysis_dir,
"analysis_preview_path": scene.analysis_preview_path,
"status": scene.status,
"published_at": scene.updated_at.isoformat() if scene.updated_at else None,
}
def _scene_asset_items(scene: SARSceneGeoORM) -> List[Dict[str, Any]]:
candidates = [
(1, "analysis_tif", "analysis_ready.tif", scene.analysis_tif_path, "image/tiff", True),
(2, "preview", "preview.png", scene.analysis_preview_path, "image/png", False),
]
metadata = scene.analysis_metadata_json if isinstance(scene.analysis_metadata_json, dict) else {}
manifest_path = str(metadata.get("manifest_path") or "").strip()
if manifest_path:
candidates.append((3, "manifest", "manifest.json", manifest_path, "application/json", False))
if scene.analysis_dir:
quality_path = os.path.join(scene.analysis_dir, "quality.json")
candidates.append((4, "quality", "quality.json", quality_path, "application/json", False))
assets: List[Dict[str, Any]] = []
for asset_id, role, name, path, media_type, primary in candidates:
if not path:
continue
assets.append(
{
"id": asset_id,
"role": role,
"name": name,
"relative_path": os.path.basename(path),
"absolute_path": path,
"format": os.path.splitext(path)[1].lower().lstrip(".") or None,
"media_type": media_type,
"is_required": primary,
"is_primary": primary,
"exists": os.path.isfile(path),
"file_size": os.path.getsize(path) if os.path.isfile(path) else None,
}
)
return assets
async def _resolve_lt1_radars_for_request(
db: AsyncSession,
request: LandsarLt1ImageProductionRequest,
) -> List[RadarDataORM]:
source_asset_ids = _dedupe_positive_ids(request.source_asset_ids)
radar_data_ids = _dedupe_positive_ids(request.radar_data_ids)
filters = []
if radar_data_ids:
filters.append(RadarDataORM.id.in_(radar_data_ids))
if source_asset_ids:
filters.append(RadarDataORM.source_product_ref_id.in_(source_asset_ids))
if not filters:
return []
result = await db.execute(select(RadarDataORM).where(*([filters[0]] if len(filters) == 1 else [filters[0] | filters[1]])))
radars = list(result.scalars().all())
unique: Dict[int, RadarDataORM] = {}
for radar in radars:
if not radar.id:
continue
family = normalize_satellite_family(radar.satellite_family or radar.satellite)
if str(family or "").upper() != "LT1":
continue
unique[int(radar.id)] = radar
return [unique[key] for key in sorted(unique.keys())]
async def _produced_radars_for_request(
db: AsyncSession,
request: LandsarLt1ImageProductionRequest,
) -> Dict[int, dict]:
radars = await _resolve_lt1_radars_for_request(db, request)
radar_ids = [int(item.id) for item in radars if item.id]
if not radar_ids:
return {}
result = await db.execute(
select(SARSceneGeoORM).where(
SARSceneGeoORM.radar_data_id.in_(radar_ids),
SARSceneGeoORM.status == "DONE",
SARSceneGeoORM.analysis_tif_path.isnot(None),
SARSceneGeoORM.analysis_engine == "lt_gamma",
SARSceneGeoORM.analysis_profile == "lt1_gamma_geocoded_mli",
)
)
return {int(scene.radar_data_id): _scene_product_marker(scene) for scene in result.scalars().all()}
async def _active_radars_for_request(
db: AsyncSession,
request: LandsarLt1ImageProductionRequest,
) -> Dict[int, dict]:
radars = await _resolve_lt1_radars_for_request(db, request)
radar_ids = [int(item.id) for item in radars if item.id]
if not radar_ids:
return {}
result = await db.execute(
select(SARSceneGeoORM).where(
SARSceneGeoORM.radar_data_id.in_(radar_ids),
SARSceneGeoORM.status.in_(["PENDING", "RUNNING"]),
)
)
return {int(scene.radar_data_id): _scene_product_marker(scene) for scene in result.scalars().all()}
def _already_produced_blocker(produced: Dict[int, dict]) -> str:
first_id = sorted(produced.keys())[0]
marker = produced[first_id] or {}
product_id = marker.get("product_id") or "unknown"
return f"Radar data {first_id} already has an analysis-ready GeoTIFF: {product_id}"
def _active_blocker(active: Dict[int, dict]) -> str:
first_id = sorted(active.keys())[0]
marker = active[first_id] or {}
return f"Radar data {first_id} already has an active GeoTIFF production task (scene_id={marker.get('scene_id')})."
@router.get("/landsar-lt1-production/capabilities")
async def get_landsar_lt1_capabilities(
current_user: AuthUserORM = Depends(_get_current_user),
):
_ = current_user
legacy = landsar_lt1_production_service.check_capabilities()
return {
"catalog_name": "sar_scene_geo",
"supported_profiles": ["lt1_gamma_geocoded_mli"],
"engine": "lt_gamma",
"available": True,
"status": "configured",
"message": "LT-1 image production uses the existing Gamma single-scene pipeline: multilook, geocode, and analysis-ready GeoTIFF registration.",
"legacy_landsar_import": legacy,
}
@router.post("/landsar-lt1-production/preview")
async def preview_landsar_lt1_production(
request: LandsarLt1ImageProductionRequest,
current_user: AuthUserORM = Depends(_get_current_user),
db: AsyncSession = Depends(get_db),
):
_ = current_user
blockers: List[str] = []
warnings: List[str] = []
radars = await _resolve_lt1_radars_for_request(db, request)
if not radars:
blockers.append("No LT-1 radar records were resolved from the selected source assets.")
if request.mode == "scene" and len(radars) != 1:
blockers.append("Scene mode requires exactly one LT-1 source asset.")
if request.mode == "batch" and len(radars) < 1:
blockers.append("Batch mode requires at least one LT-1 source asset.")
produced = await _produced_radars_for_request(db, request)
if produced:
blockers.append(_already_produced_blocker(produced))
active = await _active_radars_for_request(db, request)
if active:
blockers.append(_active_blocker(active))
if request.mode == "batch":
warnings.append("Batch mode submits one independent geocoded GeoTIFF task per scene; it does not build a D-InSAR stack.")
preview = {
"allow_submit": not blockers,
"blockers": blockers,
"warnings": warnings,
"mode": request.mode,
"profile_code": "lt1_gamma_geocoded_mli",
"engine": "lt_gamma",
"scene_count": len(radars),
"source_asset_count": len(_dedupe_positive_ids(request.source_asset_ids)),
"radar_data_count": len(radars),
"produced_radars": produced,
"active_radars": active,
"scenes": [
{
"radar_data_id": radar.id,
"source_asset_id": radar.source_product_ref_id,
"satellite": radar.satellite,
"imaging_date": radar.imaging_date,
"imaging_mode": radar.imaging_mode,
"polarization": radar.polarization,
"file_path": radar.file_path,
}
for radar in radars
],
}
return preview
@router.post("/landsar-lt1-production/run", status_code=202)
async def queue_landsar_lt1_production(
request: LandsarLt1ImageProductionRequest,
http_request: Request,
db: AsyncSession = Depends(get_db),
admin_user: AuthUserORM = Depends(_require_admin),
):
_ = admin_user
preview = await preview_landsar_lt1_production(request, current_user=admin_user, db=db)
if preview.get("blockers"):
raise HTTPException(status_code=400, detail={"blockers": preview.get("blockers")})
queued: List[Dict[str, Any]] = []
radars = await _resolve_lt1_radars_for_request(db, request)
for radar in radars:
result = await db.execute(
select(SARSceneGeoORM)
.where(SARSceneGeoORM.radar_data_id == int(radar.id))
.with_for_update(skip_locked=True)
)
scene = result.scalar_one_or_none()
if scene and scene.status in ("PENDING", "RUNNING"):
raise HTTPException(status_code=409, detail=f"Radar data {radar.id} already has an active GeoTIFF production task.")
if scene and scene.status == "DONE" and scene.analysis_tif_path:
raise HTTPException(status_code=409, detail=f"Radar data {radar.id} already has an analysis-ready GeoTIFF.")
if not scene:
scene = SARSceneGeoORM(radar_data_id=int(radar.id), status="PENDING")
db.add(scene)
await db.flush()
else:
scene.status = "PENDING"
scene.error_msg = None
await db.flush()
scene_id = int(scene.id)
await db.commit()
payload = {
"scene_id": scene_id,
"radar_data_id": int(radar.id),
"engine": "lt_gamma",
"source_asset_id": radar.source_product_ref_id,
"requested_from": "landsar_lt1_production",
}
task_label = request.task_name or radar.product_unique_id or radar.unique_id or f"radar_id={radar.id}"
task_type = f"LT1_SCENE_GEOTIFF_{scene_id}"
try:
task_id = await task_service.create_task(
task_type,
f"LT-1 geocoded GeoTIFF: {task_label}",
params=payload,
)
job_id = await job_queue_service.create_job(
JOB_TYPE_SAR_SCENE_PREPROCESS,
payload=payload,
task_id=task_id,
max_attempts=3,
)
except Exception as exc:
failed_scene = await db.get(SARSceneGeoORM, scene_id)
if failed_scene and failed_scene.status == "PENDING":
failed_scene.status = "FAILED"
failed_scene.error_msg = "Job queue failed"
await db.commit()
raise HTTPException(status_code=409 if "conflict" in str(exc).lower() else 400, detail=str(exc)) from exc
queued.append(
{
"task_id": task_id,
"job_id": job_id,
"scene_id": scene_id,
"radar_data_id": int(radar.id),
"source_asset_id": radar.source_product_ref_id,
}
)
await _add_operation_audit_log(
db,
request=http_request,
action="lt1_geotiff_production_queued",
resource="landsar-lt1-production/run",
detail={
"queued": queued,
"mode": request.mode,
"scene_count": len(queued),
},
)
await db.commit()
return {
"message": "LT-1 geocoded GeoTIFF production job queued.",
"task_id": queued[0]["task_id"] if len(queued) == 1 else None,
"job_id": queued[0]["job_id"] if len(queued) == 1 else None,
"queued": queued,
"preview": preview,
}
@router.get("/landsar-lt1-production/products")
async def list_landsar_lt1_products(
limit: int = 100,
offset: int = 0,
status: Optional[str] = None,
query: Optional[str] = None,
current_user: AuthUserORM = Depends(_get_current_user),
db: AsyncSession = Depends(get_db),
):
_ = current_user
safe_limit = max(1, min(500, int(limit or 100)))
safe_offset = max(0, int(offset or 0))
filters = [
SARSceneGeoORM.analysis_engine == "lt_gamma",
SARSceneGeoORM.analysis_profile == "lt1_gamma_geocoded_mli",
]
if status:
filters.append(SARSceneGeoORM.status == str(status).strip().upper())
if query:
like = f"%{str(query).strip()}%"
filters.append(RadarDataORM.product_unique_id.ilike(like) | RadarDataORM.unique_id.ilike(like) | RadarDataORM.file_path.ilike(like))
total_result = await db.execute(
select(func.count(SARSceneGeoORM.id))
.join(RadarDataORM, SARSceneGeoORM.radar_data_id == RadarDataORM.id)
.where(*filters)
)
total = int(total_result.scalar_one() or 0)
result = await db.execute(
select(SARSceneGeoORM, RadarDataORM)
.join(RadarDataORM, SARSceneGeoORM.radar_data_id == RadarDataORM.id)
.where(*filters)
.order_by(SARSceneGeoORM.updated_at.desc().nullslast(), SARSceneGeoORM.id.desc())
.limit(safe_limit)
.offset(safe_offset)
)
items = []
for scene, radar in result.all():
marker = _scene_product_marker(scene)
items.append(
{
"id": scene.id,
"product_id": marker["product_id"],
"catalog_name": "sar_scene_geo",
"product_family": "lt1_analysis_ready_geotiff",
"product_type": "analysis_ready_geotiff",
"display_name": radar.product_unique_id or radar.unique_id or f"radar_id={radar.id}",
"task_name": "",
"profile_code": scene.analysis_profile,
"engine_code": scene.analysis_engine,
"status": scene.status,
"health_status": "OK" if scene.status == "DONE" and scene.analysis_tif_path else "PENDING",
"publish_dir": scene.analysis_dir,
"manifest_path": (scene.analysis_metadata_json or {}).get("manifest_path") if isinstance(scene.analysis_metadata_json, dict) else None,
"native_output_dir": scene.analysis_dir,
"primary_asset_path": scene.analysis_tif_path,
"summary": {
"scene_count": 1,
"radar_data_id": radar.id,
"source_asset_ids": [radar.source_product_ref_id] if radar.source_product_ref_id else [],
"imaging_date": radar.imaging_date,
"polarization": radar.polarization,
"pixel_size_m": scene.pixel_size_m,
"backscatter_unit": scene.analysis_backscatter_unit,
},
"tags": {"engine": scene.analysis_engine, "profile": scene.analysis_profile},
"produced_at": scene.updated_at.isoformat() if scene.updated_at else None,
"published_at": scene.updated_at.isoformat() if scene.updated_at else None,
"registered_at": scene.created_at.isoformat() if scene.created_at else None,
}
)
return {"total": total, "limit": safe_limit, "offset": safe_offset, "items": items}
@router.get("/landsar-lt1-production/products/{product_db_id}")
async def get_landsar_lt1_product_detail(
product_db_id: int,
current_user: AuthUserORM = Depends(_get_current_user),
db: AsyncSession = Depends(get_db),
):
_ = current_user
result = await db.execute(
select(SARSceneGeoORM, RadarDataORM)
.join(RadarDataORM, SARSceneGeoORM.radar_data_id == RadarDataORM.id)
.where(
SARSceneGeoORM.id == product_db_id,
SARSceneGeoORM.analysis_engine == "lt_gamma",
SARSceneGeoORM.analysis_profile == "lt1_gamma_geocoded_mli",
)
)
row = result.first()
if row is None:
raise HTTPException(status_code=404, detail="LT-1 geocoded GeoTIFF product not found")
scene, radar = row
marker = _scene_product_marker(scene)
detail = {
"id": scene.id,
"product_id": marker["product_id"],
"catalog_name": "sar_scene_geo",
"product_family": "lt1_analysis_ready_geotiff",
"product_type": "analysis_ready_geotiff",
"display_name": radar.product_unique_id or radar.unique_id or f"radar_id={radar.id}",
"profile_code": scene.analysis_profile,
"engine_code": scene.analysis_engine,
"status": scene.status,
"publish_dir": scene.analysis_dir,
"primary_asset_path": scene.analysis_tif_path,
"summary": {
"scene_count": 1,
"radar_data_id": radar.id,
"source_asset_ids": [radar.source_product_ref_id] if radar.source_product_ref_id else [],
"imaging_date": radar.imaging_date,
"polarization": radar.polarization,
"pixel_size_m": scene.pixel_size_m,
"backscatter_unit": scene.analysis_backscatter_unit,
},
"assets": _scene_asset_items(scene),
}
return detail
@router.get("/landsar-lt1-production/products/{product_db_id}/assets/{asset_id}")
async def get_landsar_lt1_product_asset(
product_db_id: int,
asset_id: int,
current_user: AuthUserORM = Depends(_get_current_user),
db: AsyncSession = Depends(get_db),
):
_ = current_user
scene = await db.get(SARSceneGeoORM, product_db_id)
if scene is None or scene.analysis_engine != "lt_gamma" or scene.analysis_profile != "lt1_gamma_geocoded_mli":
raise HTTPException(status_code=404, detail="LT-1 geocoded GeoTIFF product not found")
asset = next((item for item in _scene_asset_items(scene) if int(item["id"]) == int(asset_id)), None)
if asset is None:
raise HTTPException(status_code=404, detail="LT-1 geocoded GeoTIFF asset not found")
absolute_path = str(asset.get("absolute_path") or "")
if not absolute_path or not os.path.isfile(absolute_path):
raise HTTPException(status_code=404, detail="Asset file not found")
return FileResponse(
absolute_path,
media_type=str(asset.get("media_type") or "application/octet-stream"),
filename=str(asset.get("name") or os.path.basename(absolute_path)),
headers=STATIC_ASSET_CACHE_HEADERS,
)
+52 -6
View File
@@ -26,7 +26,10 @@ from ..models import (
TimeseriesStackPlanORM,
)
from ..services.pairing_cache_service import pairing_cache_service
from ..services.job_handlers import JOB_TYPE_PAIRING_CACHE_REBUILD
from ..services.job_queue_service import job_queue_service
from ..services.spatial_service import spatial_service
from ..services.task_service import task_service
from .dependencies import (
_parse_aoi_from_files,
_parse_aoi_geojson_form_value,
@@ -38,6 +41,49 @@ logger = logging.getLogger(__name__)
router = APIRouter()
async def _queue_pairing_cache_job(
*,
db: AsyncSession,
mode: str,
force_full: bool = False,
) -> Dict[str, object]:
full_rebuild = mode in {"full", "full_rebuild"} or force_full
task_name = (
"D-InSAR pairing cache full rebuild"
if full_rebuild
else "D-InSAR pairing cache dirty reconcile"
)
payload = {
"mode": "full_rebuild" if full_rebuild else "auto_reconcile",
"force_full": bool(full_rebuild),
}
try:
task_id = await task_service.create_task(
JOB_TYPE_PAIRING_CACHE_REBUILD,
task_name,
params=payload,
db=db,
)
job_id = await job_queue_service.create_job(
JOB_TYPE_PAIRING_CACHE_REBUILD,
payload=payload,
max_attempts=1,
task_id=task_id,
db=db,
)
await db.commit()
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
return {
"ok": True,
"queued": True,
"mode": payload["mode"],
"task_id": task_id,
"job_id": job_id,
"message": f"{task_name} queued",
}
def get_pairing_request_from_form(
time_baseline_min: int = Form(1),
time_baseline_max: int = Form(30),
@@ -130,26 +176,26 @@ async def get_pairing_health_endpoint(
return await pairing_cache_service.get_admin_summary(db)
@router.post("/pairing/rebuild-cache")
@router.post("/pairing/rebuild-cache", status_code=202)
async def rebuild_pairing_cache_endpoint(
db: AsyncSession = Depends(get_db),
current_user: AuthUserORM = Depends(_require_admin),
):
_ = current_user
return await pairing_cache_service.rebuild_metric_cache(db, commit=True)
return await _queue_pairing_cache_job(db=db, mode="full_rebuild", force_full=True)
@router.post("/pairing/reconcile-dirty")
@router.post("/pairing/reconcile-dirty", status_code=202)
async def reconcile_dirty_pairing_endpoint(
force_full: bool = Query(False),
db: AsyncSession = Depends(get_db),
current_user: AuthUserORM = Depends(_require_admin),
):
_ = current_user
return await pairing_cache_service.reconcile_dirty_scenes(
db,
return await _queue_pairing_cache_job(
db=db,
mode="full_rebuild" if force_full else "auto_reconcile",
force_full=force_full,
commit=True,
)
+84 -14
View File
@@ -23,6 +23,7 @@ from ..models import (
RadarDataORM,
RadarDataPage,
RadarPreviewStatusInfo,
SARSceneGeoORM,
ScanRequest,
)
from ..services.data_service import data_service
@@ -154,6 +155,51 @@ def _normalize_list_pagination(limit: int, offset: int) -> Tuple[int, int]:
return safe_limit, safe_offset
def _lt1_image_marker(scene: SARSceneGeoORM) -> Dict[str, Any]:
return {
"scene_id": scene.id,
"radar_data_id": scene.radar_data_id,
"product_id": f"sar_scene_geo:{scene.id}",
"product_family": "lt1_analysis_ready_geotiff",
"engine_code": scene.analysis_engine,
"profile_code": scene.analysis_profile,
"analysis_tif_path": scene.analysis_tif_path,
"analysis_dir": scene.analysis_dir,
"analysis_preview_path": scene.analysis_preview_path,
"status": scene.status,
"published_at": scene.updated_at.isoformat() if scene.updated_at else None,
}
async def _decorate_lt1_landsar_status(db: AsyncSession, items: List[RadarDataORM]) -> List[RadarData]:
payloads = [RadarData.model_validate(item) for item in items]
radar_ids = [
int(item.id)
for item in items
if item.id and str(item.satellite_family or item.satellite or "").upper().replace("-", "") in {"LT1", "LT"}
]
if not radar_ids:
return payloads
result = await db.execute(
select(SARSceneGeoORM).where(
SARSceneGeoORM.radar_data_id.in_(radar_ids),
SARSceneGeoORM.status == "DONE",
SARSceneGeoORM.analysis_tif_path.isnot(None),
SARSceneGeoORM.analysis_engine == "lt_gamma",
SARSceneGeoORM.analysis_profile == "lt1_gamma_geocoded_mli",
)
)
produced = {int(scene.radar_data_id): _lt1_image_marker(scene) for scene in result.scalars().all()}
for payload in payloads:
marker = produced.get(int(payload.id or 0))
if marker:
payload.lt1_image_produced = True
payload.lt1_image_product = marker
payload.lt1_landsar_produced = True
payload.lt1_landsar_product = marker
return payloads
def _normalize_optional_text(value: Optional[str]) -> Optional[str]:
if value is None:
return None
@@ -256,6 +302,24 @@ def _build_radar_preview_status(
)
def _build_cached_radar_preview_status(record: RadarDataORM) -> RadarPreviewStatusInfo:
raw_cache_path, geo_cache_path = _radar_preview_paths(record)
preview_cache_path = str(record.preview_cache_path or "")
has_geo_cache = (
os.path.exists(geo_cache_path)
or bool(preview_cache_path and os.path.exists(preview_cache_path))
)
has_raw_cache = os.path.exists(raw_cache_path)
cached_status = str(record.preview_cache_status or "NONE").upper()
source_found = cached_status in {"READY", "FAILED"} or has_geo_cache or has_raw_cache
return _build_radar_preview_status(
record=record,
source_found=source_found,
has_geo_cache=has_geo_cache,
has_raw_cache=has_raw_cache,
)
async def _build_radar_preview_cache(
record: RadarDataORM,
db: AsyncSession,
@@ -510,10 +574,13 @@ async def search_radar_data_endpoint(
limit: int = Form(500),
offset: int = Form(0),
satellite: Optional[str] = Form(None),
satellite_family: Optional[str] = Form(None),
source_format: Optional[str] = Form(None),
satellite_mode: Optional[str] = Form(None),
receiving_station: Optional[str] = Form(None),
imaging_mode: Optional[str] = Form(None),
orbit_circle: Optional[str] = Form(None),
relative_orbit: Optional[str] = Form(None),
acquisition_time_utc: Optional[str] = Form(None),
product_type: Optional[str] = Form(None),
polarization: Optional[str] = Form(None),
@@ -536,10 +603,13 @@ async def search_radar_data_endpoint(
n_satellite_list: Optional[List[str]] = None
if n_satellite_raw and "," in n_satellite_raw:
n_satellite_list = [s.strip() for s in n_satellite_raw.split(",") if s.strip()]
n_satellite_family = _normalize_optional_text(satellite_family)
n_source_format = _normalize_optional_text(source_format)
n_satellite_mode = _normalize_optional_text(satellite_mode)
n_receiving_station = _normalize_optional_text(receiving_station)
n_imaging_mode = _normalize_optional_text(imaging_mode)
n_orbit_circle = _normalize_optional_text(orbit_circle)
n_relative_orbit = _normalize_optional_text(relative_orbit)
n_acquisition_time = _normalize_optional_text(acquisition_time_utc)
n_product_type = _normalize_optional_text(product_type)
n_polarization = _normalize_optional_text(polarization)
@@ -584,6 +654,10 @@ async def search_radar_data_endpoint(
filters.append(RadarDataORM.satellite.in_(n_satellite_list))
elif n_satellite_raw:
filters.append(RadarDataORM.satellite.ilike(f"%{n_satellite_raw}%"))
if n_satellite_family:
filters.append(func.upper(RadarDataORM.satellite_family) == n_satellite_family.upper())
if n_source_format:
filters.append(func.upper(RadarDataORM.source_format) == n_source_format.upper())
if n_satellite_mode:
filters.append(RadarDataORM.satellite_mode.ilike(f"%{n_satellite_mode}%"))
if n_receiving_station:
@@ -592,6 +666,8 @@ async def search_radar_data_endpoint(
filters.append(RadarDataORM.imaging_mode.ilike(f"%{n_imaging_mode}%"))
if n_orbit_circle:
filters.append(RadarDataORM.orbit_circle.ilike(f"%{n_orbit_circle}%"))
if n_relative_orbit:
filters.append(RadarDataORM.relative_orbit.ilike(f"%{n_relative_orbit}%"))
if n_acquisition_time:
filters.append(RadarDataORM.acquisition_time_utc.ilike(f"%{n_acquisition_time}%"))
if n_product_type:
@@ -606,10 +682,11 @@ async def search_radar_data_endpoint(
filters.append(RadarDataORM.orbit_direction.ilike(f"%{n_orbit_direction}%"))
if has_orbit_data is not None:
filters.append(RadarDataORM.has_orbit_data == has_orbit_data)
normalized_imaging_date = func.replace(RadarDataORM.imaging_date, "-", "")
if n_date_from:
filters.append(RadarDataORM.imaging_date >= n_date_from)
filters.append(normalized_imaging_date >= n_date_from.replace("-", ""))
if n_date_to:
filters.append(RadarDataORM.imaging_date <= n_date_to)
filters.append(normalized_imaging_date <= n_date_to.replace("-", ""))
if resolved_aoi_wkt:
aoi_geom = func.ST_GeomFromText(resolved_aoi_wkt, 4326)
filters.append(ST_Intersects(RadarDataORM.geom, aoi_geom))
@@ -630,8 +707,9 @@ async def search_radar_data_endpoint(
result = await db.execute(data_stmt)
items = result.scalars().all()
decorated_items = await _decorate_lt1_landsar_status(db, items)
return RadarDataSearchPageResponse(
items=items,
items=decorated_items,
total=total,
limit=limit,
offset=offset,
@@ -663,8 +741,9 @@ async def get_all_data_endpoint(
.limit(limit)
)
items = result.scalars().all()
decorated_items = await _decorate_lt1_landsar_status(db, items)
return RadarDataPage(
items=items,
items=decorated_items,
total=total,
limit=limit,
offset=offset,
@@ -724,16 +803,7 @@ async def get_radar_preview_status_endpoint(data_id: int, db: AsyncSession = Dep
if _is_gf3_native_preview_record(record):
return _build_gf3_native_preview_status(record)
raw_cache_path, geo_cache_path = _radar_preview_paths(record)
has_geo_cache = os.path.exists(geo_cache_path)
has_raw_cache = os.path.exists(raw_cache_path)
source_found = bool(await asyncio.to_thread(data_service.find_radar_preview_source, record.file_path))
return _build_radar_preview_status(
record=record,
source_found=source_found,
has_geo_cache=has_geo_cache,
has_raw_cache=has_raw_cache,
)
return _build_cached_radar_preview_status(record)
@router.post("/radar-data/{data_id}/rebuild-preview-cache", response_model=RadarPreviewStatusInfo)
+264 -1
View File
@@ -2,16 +2,19 @@ from __future__ import annotations
import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from sqlalchemy import case, func, select
from .. import database
from ..auth_service import SESSION_COOKIE_NAME, get_user_by_session_token
from ..auth_utils import verify_password
from ..models import AuthUserORM, TaskInfo
from ..config import settings
from ..models import AuthUserORM, SystemJobORM, SystemTaskORM, SystemWorkerHeartbeatORM, TaskInfo
from ..services.dinsar_production_service import dinsar_production_service
from ..services.task_service import (
TASK_ACTIVE_DEFAULT_LIMIT,
@@ -47,6 +50,73 @@ def _split_csv_param(raw: Optional[str]) -> List[str]:
return values
def _dt(value):
return value.isoformat() if value else None
def _task_payload(task: SystemTaskORM) -> dict:
return TaskInfo.model_validate(task).model_dump(mode="json")
def _worker_note(worker: SystemWorkerHeartbeatORM) -> dict:
try:
parsed = json.loads(str(worker.note or "") or "{}")
return parsed if isinstance(parsed, dict) else {}
except Exception:
return {}
def _worker_concurrency(worker: SystemWorkerHeartbeatORM) -> int:
note = _worker_note(worker)
try:
return max(1, int(note.get("concurrency") or 1))
except (TypeError, ValueError):
return 1
def _job_payload(job: SystemJobORM, task_by_id: dict[str, SystemTaskORM]) -> dict:
task = task_by_id.get(str(job.task_id or ""))
return {
"job_id": job.job_id,
"job_type": job.job_type,
"status": job.status,
"priority": int(job.priority or 0),
"attempts": int(job.attempts or 0),
"max_attempts": int(job.max_attempts or 0),
"task_id": job.task_id,
"task_type": task.task_type if task else None,
"task_name": task.task_name if task else None,
"task_status": task.status if task else None,
"task_progress": int(task.progress or 0) if task else None,
"task_message": task.message if task else None,
"workflow_run_id": job.workflow_run_id,
"workflow_step_id": job.workflow_step_id,
"locked_by": job.locked_by,
"locked_at": _dt(job.locked_at),
"heartbeat_at": _dt(job.heartbeat_at),
"next_run_at": _dt(job.next_run_at),
"created_at": _dt(job.created_at),
"started_at": _dt(job.started_at),
"finished_at": _dt(job.finished_at),
"last_error": job.last_error,
}
def _worker_payload(worker: SystemWorkerHeartbeatORM, active_job_count: int, concurrency: int) -> dict:
note = _worker_note(worker)
return {
"worker_id": worker.worker_id,
"hostname": worker.hostname,
"pid": worker.pid,
"note": worker.note,
"concurrency": concurrency,
"allowed_job_types": note.get("allowed_job_types") if isinstance(note.get("allowed_job_types"), list) else [],
"started_at": _dt(worker.started_at),
"last_seen": _dt(worker.last_seen),
"active_job_count": active_job_count,
}
@router.get("/tasks/active", response_model=List[TaskInfo])
async def get_active_tasks(limit: int = TASK_ACTIVE_DEFAULT_LIMIT, offset: int = 0):
safe_limit = min(TASK_ACTIVE_MAX_LIMIT, max(1, int(limit or TASK_ACTIVE_DEFAULT_LIMIT)))
@@ -73,6 +143,199 @@ async def get_recent_tasks(
return [TaskInfo.model_validate(task) for task in orm_tasks]
@router.get("/tasks/runtime-summary")
async def get_task_runtime_summary(limit: int = TASK_ACTIVE_DEFAULT_LIMIT, offset: int = 0):
safe_limit = min(TASK_ACTIVE_MAX_LIMIT, max(1, int(limit or TASK_ACTIVE_DEFAULT_LIMIT)))
safe_offset = min(TASK_QUERY_MAX_OFFSET, max(0, int(offset or 0)))
active_job_statuses = ["READY", "RETRY", "RUNNING"]
scan_job_types = {
"SCAN_DATA",
"SCAN_DINSAR",
"SCAN_ASSET_INVENTORY",
"AUDIT_SOURCE_ARCHIVE_INTEGRITY",
"GF3_SARSCAPE_SYNC",
"GF3_QUICKLOOK_WEBP",
}
worker_timeout = max(5, int(getattr(settings, "JOB_WORKER_HEALTH_TIMEOUT", 60) or 60))
worker_threshold = datetime.utcnow() - timedelta(seconds=worker_timeout)
configured_concurrency = max(1, int(getattr(settings, "JOB_WORKER_CONCURRENCY", 1) or 1))
async with _new_session() as db:
active_tasks = await task_service.get_active_tasks(limit=safe_limit, offset=safe_offset, db=db)
workers_result = await db.execute(
select(SystemWorkerHeartbeatORM)
.where(SystemWorkerHeartbeatORM.last_seen >= worker_threshold)
.order_by(SystemWorkerHeartbeatORM.last_seen.desc())
)
active_workers = workers_result.scalars().all()
active_worker_ids = {str(worker.worker_id) for worker in active_workers}
running_by_worker_result = await db.execute(
select(SystemJobORM.locked_by, func.count(SystemJobORM.id))
.where(SystemJobORM.status == "RUNNING")
.group_by(SystemJobORM.locked_by)
)
running_by_worker = {
str(worker_id or ""): int(count or 0)
for worker_id, count in running_by_worker_result.all()
}
status_counts_result = await db.execute(
select(SystemJobORM.status, func.count(SystemJobORM.id))
.where(SystemJobORM.status.in_(active_job_statuses))
.group_by(SystemJobORM.status)
)
job_status_counts = {
"READY": 0,
"RETRY": 0,
"RUNNING": 0,
}
for status, count in status_counts_result.all():
job_status_counts[str(status or "").upper()] = int(count or 0)
status_rank = case(
(SystemJobORM.status == "RUNNING", 0),
(SystemJobORM.status == "RETRY", 1),
else_=2,
)
jobs_result = await db.execute(
select(SystemJobORM)
.where(SystemJobORM.status.in_(active_job_statuses))
.order_by(status_rank, SystemJobORM.priority.desc(), SystemJobORM.id.asc())
.offset(safe_offset)
.limit(safe_limit)
)
active_jobs = jobs_result.scalars().all()
task_ids = {
str(task.task_id)
for task in active_tasks
if task.task_id
}
task_ids.update(
str(job.task_id)
for job in active_jobs
if job.task_id
)
task_by_id: dict[str, SystemTaskORM] = {}
if task_ids:
task_result = await db.execute(
select(SystemTaskORM).where(SystemTaskORM.task_id.in_(sorted(task_ids)))
)
task_by_id = {
str(task.task_id): task
for task in task_result.scalars().all()
if task.task_id
}
worker_concurrency_by_id = {
str(worker.worker_id): _worker_concurrency(worker)
for worker in active_workers
}
total_slots = sum(worker_concurrency_by_id.values())
busy_slots = sum(
count
for worker_id, count in running_by_worker.items()
if worker_id in active_worker_ids
)
running_count = int(job_status_counts.get("RUNNING") or 0)
stale_running_count = max(0, running_count - busy_slots)
queued_count = int(job_status_counts.get("READY") or 0) + int(job_status_counts.get("RETRY") or 0)
task_items = [_task_payload(task) for task in active_tasks]
job_items = [_job_payload(job, task_by_id) for job in active_jobs]
scan_jobs = [
item for item in job_items
if str(item.get("job_type") or "").upper() in scan_job_types
]
scan_tasks = [
item for item in task_items
if str(item.get("task_type") or "").upper() in scan_job_types
]
return {
"timestamp": datetime.utcnow().isoformat() + "Z",
"worker": {
"ok": len(active_workers) > 0,
"worker_count": len(active_workers),
"configured_concurrency": configured_concurrency,
"total_slots": total_slots,
"busy_slots": busy_slots,
"idle_slots": max(0, total_slots - busy_slots),
"timeout_seconds": worker_timeout,
"stale_running_job_count": stale_running_count,
"workers": [
_worker_payload(
worker,
running_by_worker.get(str(worker.worker_id), 0),
worker_concurrency_by_id.get(str(worker.worker_id), 1),
)
for worker in active_workers
],
},
"jobs": {
"active_count": running_count + queued_count,
"running_count": running_count,
"queued_count": queued_count,
"ready_count": int(job_status_counts.get("READY") or 0),
"retry_count": int(job_status_counts.get("RETRY") or 0),
"items": job_items,
},
"tasks": {
"active_count": len(task_items),
"running_count": sum(1 for item in task_items if item.get("status") == "RUNNING"),
"pending_count": sum(1 for item in task_items if item.get("status") == "PENDING"),
"items": task_items,
},
"scan": {
"active_task_count": len(scan_tasks),
"active_job_count": len(scan_jobs),
"running_job_count": sum(1 for item in scan_jobs if item.get("status") == "RUNNING"),
"queued_job_count": sum(1 for item in scan_jobs if item.get("status") in {"READY", "RETRY"}),
"tasks": scan_tasks,
"jobs": scan_jobs,
},
}
@router.get("/tasks/runtime-summary/stream")
async def stream_task_runtime_summary(request: Request):
token = request.cookies.get(SESSION_COOKIE_NAME)
if not token:
raise HTTPException(status_code=401, detail="Authentication required.")
async with _new_session() as db:
user = await get_user_by_session_token(db, token)
if not user:
raise HTTPException(status_code=401, detail="Authentication required.")
async def event_generator():
while True:
if await request.is_disconnected():
break
try:
summary = await get_task_runtime_summary(
limit=TASK_ACTIVE_MAX_LIMIT,
offset=0,
)
yield f"data: {json.dumps(summary)}\n\n"
except Exception:
yield "data: {}\n\n"
await asyncio.sleep(3)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
@router.get("/tasks/active/stream")
async def stream_active_tasks(request: Request):
token = request.cookies.get(SESSION_COOKIE_NAME)
+71 -17
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import gzip
import hashlib
import logging
import math
import multiprocessing as mp
import os
@@ -48,6 +49,8 @@ from ..utils import (
)
from .pairing_state_service import pairing_state_service
from .data_service import DataService
logger = logging.getLogger(__name__)
from .image_service import image_service
from .orbit_converter import sync_orbit_pools
from .task_service import task_service
@@ -2599,6 +2602,53 @@ def _image_data_format_for_source(row: Dict[str, Any]) -> str:
return "DIRECTORY"
PAIRING_RELEVANT_RADAR_FIELDS = {
"satellite",
"satellite_family",
"imaging_date",
"imaging_mode",
"orbit_direction",
"polarization",
"look_direction",
"relative_orbit",
"insar_source_ready",
"file_path",
"coverage_polygon",
"min_lon",
"min_lat",
"max_lon",
"max_lat",
"scene_center_lon",
"scene_center_lat",
}
def _normalize_pairing_compare_value(value: Any) -> Any:
if isinstance(value, str):
text = value.strip()
return text or None
if isinstance(value, bool):
return bool(value)
if isinstance(value, (int, float)):
return round(float(value), 9)
if isinstance(value, list):
return [_normalize_pairing_compare_value(item) for item in value]
if isinstance(value, tuple):
return [_normalize_pairing_compare_value(item) for item in value]
return value
def _radar_pairing_fields_changed(existing: RadarDataORM, new_values: Dict[str, Any]) -> bool:
for field in PAIRING_RELEVANT_RADAR_FIELDS:
if field not in new_values:
continue
before = _normalize_pairing_compare_value(getattr(existing, field, None))
after = _normalize_pairing_compare_value(new_values.get(field))
if before != after:
return True
return False
class AssetInventoryService:
async def _progress(self, task_id: Optional[str], message: str, progress: int) -> None:
if not task_id:
@@ -2784,6 +2834,13 @@ class AssetInventoryService:
raw_cache_path = DataService.get_radar_raw_cache_path(unique_id, record.file_path)
geo_cache_path = DataService.get_radar_geo_cache_path(unique_id, record.file_path)
product_name = os.path.basename(str(record.file_path or ""))
if task_id:
progress = progress_start + int(index / max(1, total) * max(1, progress_end - progress_start))
await task_service.update_task(
task_id,
message=f"Building archive preview cache ({index}/{total}): {product_name}",
progress=min(progress_end, progress),
)
preview_source = await asyncio.to_thread(DataService.find_radar_preview_source, record.file_path)
if not preview_source:
@@ -4675,6 +4732,7 @@ class AssetInventoryService:
profile_inputs.append((row, asset_id, int(scene.id)))
else:
before_orbit_id = existing.selected_orbit_asset_id
pairing_fields_changed = _radar_pairing_fields_changed(existing, radar_values)
for key, value in radar_values.items():
setattr(existing, key, value)
if not existing.orbit_binding_status:
@@ -4682,16 +4740,15 @@ class AssetInventoryService:
db.add(existing)
if existing.id is not None:
profile_inputs.append((row, asset_id, int(existing.id)))
if existing.id is not None and before_orbit_id != existing.selected_orbit_asset_id:
if existing.id is not None and (
pairing_fields_changed or before_orbit_id != existing.selected_orbit_asset_id
):
dirty_scene_ids.append(int(existing.id))
await db.flush()
if profile_inputs:
await self._upsert_geometry_profiles(db, profile_inputs)
await self._attach_radar_ids_to_metadata_documents(db, profile_inputs)
for _, _, radar_id in profile_inputs:
if radar_id not in dirty_scene_ids:
dirty_scene_ids.append(radar_id)
if dirty_scene_ids:
await pairing_state_service.mark_scenes_dirty(db, scene_ids=dirty_scene_ids, reason="asset_inventory_source_update", commit=False)
@@ -4988,7 +5045,6 @@ class AssetInventoryService:
matched = 0
missing = 0
candidate_count = 0
dirty_scene_ids: List[int] = []
for scene in scenes:
candidates = await self._find_orbit_candidates(db, scene)
if not candidates:
@@ -5018,8 +5074,6 @@ class AssetInventoryService:
},
)
)
if scene.id is not None:
dirty_scene_ids.append(int(scene.id))
continue
candidate_count += len(candidates)
@@ -5049,8 +5103,6 @@ class AssetInventoryService:
scene.orbit_binding_reason = selected[2]
db.add(scene)
matched += 1
if scene.id is not None:
dirty_scene_ids.append(int(scene.id))
if len(candidates) > 1 and abs(float(candidates[0][1]) - float(candidates[1][1])) < 0.001:
db.add(
@@ -5068,13 +5120,8 @@ class AssetInventoryService:
)
)
if dirty_scene_ids:
await pairing_state_service.mark_scenes_dirty(
db,
scene_ids=sorted(set(dirty_scene_ids)),
reason="asset_inventory_orbit_binding",
commit=False,
)
# Pairing queries join radar_data and filter has_orbit_data live; orbit rebinding
# does not change the cached geometric/time pairing metrics.
return {
"scene_count": len(scenes),
"matched_count": matched,
@@ -5279,8 +5326,15 @@ class AssetInventoryService:
.limit(safe_limit)
)
).scalars().all()
items = [self._source_asset_payload(row) for row in rows]
try:
from .landsar_lt1_production_service import landsar_lt1_production_service
await landsar_lt1_production_service.decorate_source_asset_payloads(db, items)
except Exception:
logger.debug("Failed to decorate LT-1 LandSAR production status", exc_info=True)
return {
"items": [self._source_asset_payload(row) for row in rows],
"items": items,
"total": total,
"limit": safe_limit,
"offset": safe_offset,
+10 -6
View File
@@ -288,6 +288,8 @@ def _radar_archive_expected_preview_rank(archive_path: str, member_name: str) ->
return None
member = str(member_name or "").replace("\\", "/").strip("/")
while member.startswith("./"):
member = member[2:].strip("/")
member_lower = member.lower()
product_lower = product_stem.lower()
expected_names = [
@@ -315,7 +317,10 @@ def _radar_archive_expected_preview_rank(archive_path: str, member_name: str) ->
def _radar_archive_preview_score(member_name: str, size_bytes: int = 0) -> Optional[Tuple[int, int, int, int, str]]:
lower_name = str(member_name or "").replace("\\", "/").lower()
normalized_name = str(member_name or "").replace("\\", "/").strip("/")
while normalized_name.startswith("./"):
normalized_name = normalized_name[2:].strip("/")
lower_name = normalized_name.lower()
base_name = os.path.basename(lower_name)
if not base_name.endswith(_RADAR_PREVIEW_EXTENSIONS):
return None
@@ -1024,7 +1029,6 @@ class DataService:
# 3. 更新缺失精轨的现有记录
update_progress("正在关联精轨数据...", 85)
updated_orbits = 0
updated_orbit_scene_ids = set()
if orbit_files_map:
stmt_select = select(RadarDataORM).where(RadarDataORM.has_orbit_data == False)
result = await db.execute(stmt_select)
@@ -1037,8 +1041,6 @@ class DataService:
record.orbit_file_path = orbit_files_map[key]
db.add(record)
updated_orbits += 1
if record.id is not None:
updated_orbit_scene_ids.add(int(record.id))
for data_type, root_path, mtime in scan_state_updates:
await DataService._upsert_scan_state(db, data_type, root_path, mtime)
@@ -1046,7 +1048,9 @@ class DataService:
await db.commit()
pairing_dirty_summary: Dict[str, Any] = {}
dirty_scene_ids = set(updated_orbit_scene_ids)
# Orbit availability is filtered live by pairing queries; it is not part of
# pairing_metric_cache, so orbit-only updates should not dirty the cache.
dirty_scene_ids = set()
processed_unique_ids = [key for key in radar_cache_candidates.keys() if key]
for chunk in _chunked(processed_unique_ids, 500):
id_result = await db.execute(
@@ -1061,7 +1065,7 @@ class DataService:
reason="radar_scan",
commit=True,
)
elif processed_scenes > 0 or updated_orbits > 0:
elif processed_scenes > 0:
pairing_dirty_summary = await pairing_state_service.mark_global_dirty(
db,
reason="radar_scan",
@@ -1007,6 +1007,7 @@ class DinsarProductionService:
"root_dir": run.source_root,
"publish_root_dir": run.publish_root_dir,
"message": run.latest_message,
"summary_json": run.summary_json if isinstance(run.summary_json, dict) else {},
"total_items": run.total_items,
"completed_items": run.completed_items,
"failed_items": run.failed_items,
+533 -1
View File
@@ -41,7 +41,7 @@ from .dinsar_result_layout_service import (
)
from .dinsar_scan_service import dinsar_scan_service
from .engine_lock_service import engine_lock_service
from .envi_service import build_envi_runner_command, get_envi_runner_cwd, get_envi_runner_env
from .envi_service import build_envi_runner_command, extract_disp_results, get_envi_runner_cwd, get_envi_runner_env
from .psinsar_catalog_service import psinsar_catalog_service
from .result_catalog_service import result_catalog_service
from .sbas_insar_catalog_service import sbas_insar_catalog_service
@@ -101,10 +101,13 @@ JOB_TYPE_ISCE2_RUN = "ISCE2_RUN"
JOB_TYPE_PYINT_RUN = "PYINT_RUN"
JOB_TYPE_LANDSAR_RUN = "LANDSAR_RUN"
JOB_TYPE_LANDSAR_CLUSTER_ITEM = "LANDSAR_CLUSTER_ITEM"
JOB_TYPE_LANDSAR_LT1_IMPORT = "LANDSAR_LT1_IMPORT"
JOB_TYPE_EXTRACT_DINSAR_PRODUCTS = "EXTRACT_DINSAR_PRODUCTS"
JOB_TYPE_PUBLISH_DINSAR_PRODUCTS = "PUBLISH_DINSAR_PRODUCTS"
JOB_TYPE_REBUILD_DINSAR_CATALOG = "REBUILD_DINSAR_CATALOG"
JOB_TYPE_REBUILD_PSINSAR_CATALOG = "REBUILD_PSINSAR_CATALOG"
JOB_TYPE_REBUILD_SBAS_INSAR_CATALOG = "REBUILD_SBAS_INSAR_CATALOG"
JOB_TYPE_PAIRING_CACHE_REBUILD = "PAIRING_CACHE_REBUILD"
JOB_TYPE_SCAN_ASSET_INVENTORY = "SCAN_ASSET_INVENTORY"
JOB_TYPE_AUDIT_SOURCE_ARCHIVE_INTEGRITY = "AUDIT_SOURCE_ARCHIVE_INTEGRITY"
JOB_TYPE_SBAS_COREGISTRATION = "SBAS_COREGISTRATION"
@@ -206,6 +209,142 @@ def _dedupe_existing_dirs(paths: Any) -> List[str]:
return ordered
def _compact_failure_text(value: Any, *, max_length: int = 700) -> str:
text = re.sub(r"\s+", " ", str(value or "")).strip()
if not text:
return "Unknown error"
if len(text) <= max_length:
return text
return text[: max(0, max_length - 3)].rstrip() + "..."
def _classify_dinsar_failure(error_message: Any) -> str:
text = str(error_message or "")
lowered = text.lower()
if "3221225477" in text or "status_access_violation" in lowered:
return "LandSAR access violation during coherence mask/phase unwrapping"
if "not enough gcps" in lowered or "space insar calibration failed" in lowered:
return "Insufficient GCPs for baseline/calibration"
if (
"no enough points" in lowered
or "not enough points" in lowered
or "geo_extract_gcp" in lowered
or "无满足snr" in lowered
or "离散采样点" in text
):
return "Insufficient tie/GCP points for DEM/geocoding"
if "dem/sub-terrain" in lowered or "subterrain" in lowered or "sub-terrain" in lowered:
return "DEM/sub-terrain processing failed"
if "相干性掩膜" in text and "相位解缠" in text:
return "Coherence mask/phase unwrapping failed"
if "timeout" in lowered or "timed out" in lowered or "超时" in text:
return "Processing timeout"
if "publish" in lowered:
return "Result catalog publish failed"
compact = _compact_failure_text(text, max_length=120)
return compact if compact != "Unknown error" else "Unclassified D-InSAR failure"
async def _build_dinsar_failure_summary(db, run) -> Dict[str, Any]:
result = await db.execute(
select(DinsarProductionRunItemORM)
.where(
DinsarProductionRunItemORM.run_id == run.run_id,
DinsarProductionRunItemORM.status == "FAILED",
)
.order_by(
DinsarProductionRunItemORM.order_index.asc().nullslast(),
DinsarProductionRunItemORM.id.asc(),
)
)
failed_items = result.scalars().all()
details: List[Dict[str, Any]] = []
grouped: Dict[str, Dict[str, Any]] = {}
for item in failed_items:
label = str(item.task_alias or item.task_name or f"item-{item.id}").strip()
reason = _classify_dinsar_failure(item.last_error)
compact_error = _compact_failure_text(item.last_error)
detail = {
"id": item.id,
"order_index": item.order_index,
"task_name": item.task_name,
"task_alias": item.task_alias,
"reason": reason,
"error": compact_error,
"source_task_dir": item.source_task_dir,
"latest_output_dir": item.latest_output_dir,
"latest_log_path": item.latest_log_path,
}
details.append(detail)
group = grouped.setdefault(reason, {"reason": reason, "count": 0, "items": []})
group["count"] += 1
if len(group["items"]) < 25:
group["items"].append(label)
groups = sorted(grouped.values(), key=lambda item: (-int(item["count"]), str(item["reason"])))
return {
"failed_count": len(details),
"groups": groups,
"items": details,
}
def _chunk_log_lines(lines: List[str], *, max_chars: int = 3500) -> List[str]:
chunks: List[str] = []
current: List[str] = []
current_len = 0
for line in lines:
line_len = len(line) + 1
if current and current_len + line_len > max_chars:
chunks.append("\n".join(current))
current = []
current_len = 0
current.append(line)
current_len += line_len
if current:
chunks.append("\n".join(current))
return chunks
async def _log_dinsar_failure_summary(
*,
task_id: str,
run_id: str,
engine_title: str,
run,
failure_summary: Dict[str, Any],
run_log,
) -> None:
if not failure_summary or int(failure_summary.get("failed_count") or 0) <= 0:
return
lines = [
(
f"{engine_title} D-InSAR failure summary: "
f"completed={run.completed_items} failed={run.failed_items} total={run.total_items}"
),
"Failure groups:",
]
for group in failure_summary.get("groups") or []:
items = ", ".join(str(item) for item in (group.get("items") or []))
omitted = int(group.get("count") or 0) - len(group.get("items") or [])
suffix = f" (+{omitted} more)" if omitted > 0 else ""
lines.append(f"- {group.get('reason')}: {group.get('count')} item(s): {items}{suffix}")
lines.append("Failed items:")
for item in failure_summary.get("items") or []:
order_index = item.get("order_index")
order_text = f"{order_index}/{run.total_items}" if order_index else f"id={item.get('id')}"
label = item.get("task_alias") or item.get("task_name") or f"item-{item.get('id')}"
lines.append(f"- [{order_text}] {label}: {item.get('reason')} | {item.get('error')}")
for chunk in _chunk_log_lines(lines):
await task_service.add_log(task_id, "WARNING", chunk)
run_log(run_id, f"[failure-summary]\n{chunk}")
async def _run_scan_data_custom(task_id: str, payload: Dict[str, Any]) -> None:
from ..database import AsyncSessionLocal
@@ -2086,6 +2225,7 @@ async def _run_dinsar_production_controller(job: SystemJobORM) -> None:
f"failed={run.failed_items} total={run.total_items}"
)
failure_summary = await _build_dinsar_failure_summary(db, run)
summary_payload = {
"workflow": workflow,
"engine_code": run.engine_code,
@@ -2098,7 +2238,17 @@ async def _run_dinsar_production_controller(job: SystemJobORM) -> None:
"publish": publish_result,
"publish_error": publish_error,
"published_output_dirs": publish_dirs,
"failure_summary": failure_summary,
}
if failure_summary.get("failed_count"):
await _log_dinsar_failure_summary(
task_id=job.task_id,
run_id=run.run_id,
engine_title="ENVI",
run=run,
failure_summary=failure_summary,
run_log=run_log,
)
await dinsar_production_service.finalize_run(
run,
db=db,
@@ -3061,6 +3211,7 @@ async def _run_wsl_dinsar_production_controller(
f"failed={run.failed_items} total={run.total_items}"
)
failure_summary = await _build_dinsar_failure_summary(db, run)
summary_payload = {
"workflow": f"dinsar_{engine_code}",
"engine_code": run.engine_code,
@@ -3074,7 +3225,17 @@ async def _run_wsl_dinsar_production_controller(
"rebuild": rebuild_result,
"publish_error": publish_error,
"published_output_dirs": publish_dirs,
"failure_summary": failure_summary,
}
if failure_summary.get("failed_count"):
await _log_dinsar_failure_summary(
task_id=job.task_id,
run_id=run.run_id,
engine_title=engine_title,
run=run,
failure_summary=failure_summary,
run_log=run_log,
)
await dinsar_production_service.finalize_run(
run,
db=db,
@@ -3266,6 +3427,211 @@ async def _handle_landsar_run(job: SystemJobORM) -> None:
)
async def _handle_landsar_lt1_import(job: SystemJobORM) -> None:
from .landsar_lt1_production_service import landsar_lt1_production_service
from .asset_inventory_service import asset_inventory_service
payload = dict(job.payload or {})
task_id = job.task_id
if not task_id:
raise ValueError("LANDSAR_LT1_IMPORT job missing task_id")
loop = asyncio.get_running_loop()
def _progress(event: Dict[str, Any]) -> None:
message = str(event.get("message") or event.get("event") or "").strip()
if not message:
return
progress = event.get("progress")
async def _write() -> None:
try:
await task_service.add_log(task_id, "INFO", message)
if progress is not None:
await task_service.update_task(task_id, progress=int(progress), message=message)
except Exception:
logger.debug("Failed to write LandSAR LT-1 progress", exc_info=True)
asyncio.run_coroutine_threadsafe(_write(), loop)
await task_service.start_task(task_id, message="LandSAR LT-1 import started")
await task_service.update_task(task_id, progress=5, message="Checking LandSAR LT-1 runtime")
try:
source_asset_ids = _dedupe_positive_ints(payload.get("source_asset_ids"))
radar_data_ids = _dedupe_positive_ints(payload.get("radar_data_ids"))
if source_asset_ids or radar_data_ids:
await task_service.update_task(task_id, progress=8, message="Preparing LT-1 source assets")
prepared = await _prepare_landsar_lt1_source_assets(
task_id,
payload,
source_asset_ids=source_asset_ids,
radar_data_ids=radar_data_ids,
landsar_lt1_production_service=landsar_lt1_production_service,
asset_inventory_service=asset_inventory_service,
)
payload = {
**payload,
"source_asset_ids": prepared["source_asset_ids"],
"radar_data_ids": prepared["radar_data_ids"],
"__prepared_scene_dirs": prepared["scene_dirs"],
"__materialized": prepared["materialized"],
"__materialize_task_root": prepared["task_root"],
}
async with _local_engine_lock("landsar"):
result = await asyncio.to_thread(
landsar_lt1_production_service.run_import,
payload,
progress_callback=_progress,
)
async with AsyncSessionLocal() as db:
catalog_result = await landsar_lt1_production_service.register_manifest(
db,
result["manifest_path"],
)
await task_service.add_log(
task_id,
"INFO",
(
"LandSAR LT-1 product registered: "
f"product_id={catalog_result.get('product_id')}, "
f"assets={catalog_result.get('asset_count')}"
),
)
await task_service.update_task(
task_id,
status="COMPLETED",
progress=100,
message=(
"LandSAR LT-1 import completed: "
f"product_id={catalog_result.get('product_id')}, "
f"Input_Data={result.get('input_data_dir')}"
),
)
except Exception as exc:
await task_service.add_log(task_id, "ERROR", f"LandSAR LT-1 import failed: {exc}")
await task_service.update_task(
task_id,
status="FAILED",
progress=100,
message=f"LandSAR LT-1 import failed: {exc}",
)
raise
def _dedupe_positive_ints(values: Any) -> List[int]:
result: List[int] = []
if not isinstance(values, list):
return result
for value in values:
try:
parsed = int(value)
except (TypeError, ValueError):
continue
if parsed > 0 and parsed not in result:
result.append(parsed)
return result
async def _prepare_landsar_lt1_source_assets(
task_id: str,
payload: Dict[str, Any],
*,
source_asset_ids: List[int],
radar_data_ids: List[int],
landsar_lt1_production_service: Any,
asset_inventory_service: Any,
) -> Dict[str, Any]:
from ..models import SourceProductAssetORM
scene_dirs = _dedupe_existing_dirs(payload.get("scene_dirs") or [])
radar_ids: List[int] = []
async with AsyncSessionLocal() as db:
if radar_data_ids:
rows = (
await db.execute(select(RadarDataORM).where(RadarDataORM.id.in_(radar_data_ids)))
).scalars().all()
radar_by_id = {int(row.id): row for row in rows}
for radar_id in radar_data_ids:
radar = radar_by_id.get(int(radar_id))
if radar is None:
raise ValueError(f"Radar data not found: {radar_id}")
radar_ids.append(int(radar.id))
if radar.source_product_ref_id and int(radar.source_product_ref_id) not in source_asset_ids:
source_asset_ids.append(int(radar.source_product_ref_id))
elif radar.file_path and os.path.isdir(radar.file_path):
scene_dirs.append(os.path.normpath(os.path.abspath(radar.file_path)))
else:
raise ValueError(f"Radar data {radar_id} has no source asset or scene directory.")
produced = await landsar_lt1_production_service.find_produced_source_asset_map(db, source_asset_ids)
if produced:
first_id = sorted(produced.keys())[0]
product_id = (produced[first_id] or {}).get("product_id")
raise ValueError(f"Source asset {first_id} already has a LandSAR LT-1 product: {product_id}")
assets = []
if source_asset_ids:
assets = (
await db.execute(select(SourceProductAssetORM).where(SourceProductAssetORM.id.in_(source_asset_ids)))
).scalars().all()
asset_by_id = {int(asset.id): asset for asset in assets}
task_root = os.path.join(
os.path.normpath(os.path.abspath(settings.LANDSAR_WORK_ROOT)),
"lt1_import_tasks",
task_id,
"scenes",
)
os.makedirs(task_root, exist_ok=True)
materialized: List[Dict[str, Any]] = []
overwrite = bool(payload.get("materialize_overwrite", False))
for asset_id in source_asset_ids:
asset = asset_by_id.get(int(asset_id))
if asset is None:
raise ValueError(f"Source asset not found: {asset_id}")
if str(asset.satellite_family or "").upper() != "LT1":
raise ValueError(f"Source asset {asset_id} is not LT-1.")
result = await asyncio.to_thread(
asset_inventory_service.materialize_source_asset,
asset,
target_root=task_root,
overwrite=overwrite,
)
target_dir = os.path.normpath(os.path.abspath(str(result.get("safe_dir") or result.get("target_dir") or "")))
if not target_dir or not os.path.isdir(target_dir):
raise ValueError(f"Source asset {asset_id} materialize did not produce a directory.")
if target_dir not in scene_dirs:
scene_dirs.append(target_dir)
materialized.append(
{
"source_asset_id": int(asset.id),
"asset_uid": asset.asset_uid,
"logical_product_uid": asset.logical_product_uid,
"source_path": asset.file_path,
"scene_dir": target_dir,
"status": result.get("status"),
"member_count": result.get("member_count"),
}
)
await task_service.add_log(
task_id,
"INFO",
f"Prepared LT-1 source asset {asset.id}: {target_dir} ({result.get('status')})",
)
scene_dirs = list(dict.fromkeys(scene_dirs))
if not scene_dirs:
raise ValueError("No LT-1 scene directories were prepared.")
return {
"scene_dirs": scene_dirs,
"source_asset_ids": list(dict.fromkeys(source_asset_ids)),
"radar_data_ids": radar_ids,
"materialized": materialized,
"task_root": task_root,
}
async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
payload = job.payload or {}
production_run_id = str(payload.get("production_run_id") or "").strip()
@@ -4636,6 +5002,85 @@ async def _handle_gf3_sarscape_clean(job: SystemJobORM) -> None:
await task_service.update_task(job.task_id, status=status, progress=100, message=message)
async def _handle_extract_dinsar_products(job: SystemJobORM) -> None:
if not job.task_id:
raise ValueError("EXTRACT_DINSAR_PRODUCTS requires task_id for progress tracking.")
payload = job.payload or {}
root_dir = str(payload.get("root_dir") or "").strip()
dest_dir = payload.get("dest_dir") or None
if not root_dir:
raise ValueError("EXTRACT_DINSAR_PRODUCTS requires root_dir payload.")
await task_service.start_task(job.task_id, message="正在提取 D-InSAR 位移结果...")
result = await asyncio.to_thread(
extract_disp_results,
root_dir,
dest_dir,
)
await task_service.update_task(
job.task_id,
progress=45,
message=(
"D-InSAR 位移结果提取完成: "
f"processed={result.get('processed', 0)}, "
f"copied={result.get('copied', 0)}, "
f"overwritten={result.get('overwritten', 0)}, "
f"failed={result.get('failed', 0)}"
),
)
target_dir = result.get("target_dir")
publish_roots = [target_dir] if target_dir and os.path.isdir(str(target_dir)) else []
publish_result = None
rebuild_result = None
if publish_roots:
async with AsyncSessionLocal() as db:
await task_service.update_task(
job.task_id,
progress=60,
message="正在发布 D-InSAR 标准结果包...",
)
publish_result = await result_catalog_service.publish_from_sources(
db,
publish_roots,
)
if int(publish_result.get("processed", 0) or 0) > 0:
await task_service.update_task(
job.task_id,
progress=80,
message="正在重建 D-InSAR 结果目录索引...",
)
rebuild_result = await result_catalog_service.rebuild_catalog(
db,
full_rebuild=True,
)
failed = int(result.get("failed", 0) or 0)
publish_failed = int((publish_result or {}).get("failed", 0) or 0)
rebuild_failed = int((rebuild_result or {}).get("failed", 0) or 0)
status = "FAILED" if (failed or publish_failed or rebuild_failed) else "COMPLETED"
message = (
"D-InSAR 结果提取与登记完成: "
f"提取 {int(result.get('processed', 0) or 0)} 项, "
f"复制 {int(result.get('copied', 0) or 0)} 个, "
f"覆盖 {int(result.get('overwritten', 0) or 0)}"
)
if publish_result is not None:
message += f", 发布 {int(publish_result.get('processed', 0) or 0)}"
if rebuild_result is not None:
message += f", 入库 {int(rebuild_result.get('registered', 0) or 0)}"
if status == "FAILED":
message += f", 失败 {failed + publish_failed + rebuild_failed}"
await task_service.update_task(
job.task_id,
status=status,
progress=100,
message=message,
)
async def _handle_publish_dinsar_products_clean(job: SystemJobORM) -> None:
if not job.task_id:
raise ValueError("PUBLISH_DINSAR_PRODUCTS requires task_id for progress tracking.")
@@ -5669,13 +6114,99 @@ async def _handle_sbas_landsar_workflow(job: SystemJobORM) -> None:
)
async def _handle_pairing_cache_rebuild(job: SystemJobORM) -> None:
if not job.task_id:
raise ValueError("PAIRING_CACHE_REBUILD requires task_id for progress tracking.")
from .pairing_cache_service import pairing_cache_service
payload = job.payload or {}
mode = str(payload.get("mode") or "auto_reconcile").strip().lower()
force_full = bool(payload.get("force_full", False))
full_rebuild = mode in {"full", "full_rebuild"} or force_full
action_label = "full rebuild" if full_rebuild else "dirty reconcile"
await task_service.start_task(
job.task_id,
message=f"D-InSAR pairing cache {action_label} started",
)
await task_service.update_task(
job.task_id,
progress=5,
message=f"D-InSAR pairing cache {action_label} is running",
)
progress_state: Dict[str, Any] = {
"progress": 5,
"message": f"D-InSAR pairing cache {action_label} is running",
}
async def _report_progress(message: str, progress: int) -> None:
safe_progress = max(5, min(95, int(progress)))
progress_state["progress"] = safe_progress
progress_state["message"] = message
await task_service.update_task(
job.task_id,
progress=safe_progress,
message=message,
)
async def _keepalive() -> None:
while True:
await asyncio.sleep(60)
message = str(progress_state.get("message") or f"D-InSAR pairing cache {action_label} is still running")
progress = int(progress_state.get("progress") or 5)
await task_service.update_task(
job.task_id,
progress=progress,
message=f"{message} (still running)",
)
keepalive_task = asyncio.create_task(_keepalive())
try:
async with AsyncSessionLocal() as db:
if full_rebuild:
result = await pairing_cache_service.rebuild_metric_cache(
db,
commit=True,
progress_callback=_report_progress,
)
else:
result = await pairing_cache_service.reconcile_dirty_scenes(
db,
force_full=False,
commit=True,
progress_callback=_report_progress,
)
finally:
keepalive_task.cancel()
try:
await keepalive_task
except asyncio.CancelledError:
pass
await task_service.update_task(
job.task_id,
status="COMPLETED",
progress=100,
message=(
"D-InSAR pairing cache completed: "
f"mode={result.get('mode')}, "
f"scenes={result.get('scene_count', 0)}, "
f"pairs={result.get('pair_count', 0)}, "
f"dirty={result.get('dirty_scene_count', 0)}"
),
)
_HANDLERS = {
JOB_TYPE_SCAN_DATA: _handle_scan_data,
JOB_TYPE_SCAN_ASSET_INVENTORY: _handle_scan_asset_inventory,
JOB_TYPE_AUDIT_SOURCE_ARCHIVE_INTEGRITY: _handle_archive_integrity_audit,
JOB_TYPE_SCAN_DINSAR: _handle_scan_dinsar,
JOB_TYPE_EXTRACT_DINSAR_PRODUCTS: _handle_extract_dinsar_products,
JOB_TYPE_PUBLISH_DINSAR_PRODUCTS: _handle_publish_dinsar_products_clean,
JOB_TYPE_REBUILD_DINSAR_CATALOG: _handle_rebuild_dinsar_catalog_clean,
JOB_TYPE_PAIRING_CACHE_REBUILD: _handle_pairing_cache_rebuild,
JOB_TYPE_TIMESERIES_PREPARE: _handle_timeseries_prepare,
JOB_TYPE_TIMESERIES_STACK_PREP: _handle_timeseries_stack_prep,
JOB_TYPE_TIMESERIES_MATERIALIZE: _handle_timeseries_materialize,
@@ -5701,6 +6232,7 @@ _HANDLERS = {
JOB_TYPE_ISCE2_RUN: _handle_isce2_run,
JOB_TYPE_PYINT_RUN: _handle_pyint_run,
JOB_TYPE_LANDSAR_RUN: _handle_landsar_run,
JOB_TYPE_LANDSAR_LT1_IMPORT: _handle_landsar_lt1_import,
JOB_TYPE_LANDSAR_CLUSTER_ITEM: _handle_landsar_cluster_item,
JOB_TYPE_WATER_GEOCODE: _handle_water_geocode,
JOB_TYPE_SAR_SCENE_PREPROCESS: _handle_sar_scene_preprocess,
+5 -2
View File
@@ -299,10 +299,11 @@ class JobQueueService:
)
stale_jobs = result.scalars().all()
if not stale_jobs:
return {"recovered": 0, "failed": 0}
return {"recovered": 0, "failed": 0, "failed_task_ids": []}
recovered = 0
failed = 0
failed_task_ids = []
for job in stale_jobs:
attempts = int(job.attempts or 0) + 1
if attempts < int(job.max_attempts or 1):
@@ -315,6 +316,8 @@ class JobQueueService:
next_run_at = None
failed += 1
finished_at = now
if job.task_id:
failed_task_ids.append(str(job.task_id))
await db.execute(
update(SystemJobORM)
@@ -331,7 +334,7 @@ class JobQueueService:
)
)
await db.commit()
return {"recovered": recovered, "failed": failed}
return {"recovered": recovered, "failed": failed, "failed_task_ids": failed_task_ids}
finally:
if gen_db:
await db.close()
+28 -2
View File
@@ -1,4 +1,5 @@
import asyncio
import json
import os
import socket
import uuid
@@ -50,20 +51,35 @@ def _new_session():
return database.AsyncSessionLocal()
async def _touch_worker(worker_id: str) -> None:
async def _touch_worker(
worker_id: str,
*,
concurrency: int,
allowed_job_types: Optional[Set[str]],
) -> None:
host = socket.gethostname()
pid = os.getpid()
note = json.dumps(
{
"concurrency": max(1, int(concurrency or 1)),
"allowed_job_types": sorted(allowed_job_types or []),
},
ensure_ascii=False,
separators=(",", ":"),
)
async with _new_session() as db:
stmt = pg_insert(SystemWorkerHeartbeatORM).values(
worker_id=worker_id,
hostname=host,
pid=pid,
note=note,
)
stmt = stmt.on_conflict_do_update(
index_elements=["worker_id"],
set_={
"hostname": host,
"pid": pid,
"note": note,
"last_seen": func.now(),
},
)
@@ -172,7 +188,11 @@ async def run_worker_loop(
now = time.monotonic()
if now - last_heartbeat >= heartbeat_interval:
try:
await _touch_worker(worker_id)
await _touch_worker(
worker_id,
concurrency=concurrency,
allowed_job_types=allowed_job_types,
)
except Exception as exc:
print(f"[WARN] worker cleanup: {exc}")
last_heartbeat = now
@@ -184,6 +204,12 @@ async def run_worker_loop(
f"[*] Recovered stale jobs: retry={recovered.get('recovered', 0)} "
f"failed={recovered.get('failed', 0)}"
)
for task_id in recovered.get("failed_task_ids", []) or []:
await task_service.update_task(
task_id,
status="FAILED",
message="后台任务心跳超时,任务已被标记为失败",
)
except Exception as exc:
print(f"[WARN] recover_stale: {exc}")
last_recover = now
File diff suppressed because it is too large Load Diff
+20 -3
View File
@@ -88,6 +88,15 @@ def run_lt_gamma_scene_preprocess(
if not runner.is_file():
raise FileNotFoundError(f"Gamma scene runner not found: {runner}")
analysis_dem_path = (
settings.SAR_ANALYSIS_DEM_PATH
or settings.GAMMA_SBAS_DEM_PATH
or settings.PYINT_PREPARED_DEM_PATH
or _prepared_dem_path()
)
if not analysis_dem_path:
raise RuntimeError("SAR_ANALYSIS_DEM_PATH is not configured for LT analysis GeoTIFF production")
args = [
pyint_python,
to_wsl_path(str(runner)),
@@ -102,7 +111,11 @@ def run_lt_gamma_scene_preprocess(
"--dem-root",
to_wsl_path(str(settings.PYINT_DEM_ROOT)),
"--prepared-dem-path",
to_wsl_path(_prepared_dem_path()),
to_wsl_path(str(analysis_dem_path)),
"--dem-resolution-m",
str(float(settings.SAR_ANALYSIS_DEM_RESOLUTION_M or 30.0)),
"--target-grid-size-m",
str(float(settings.SAR_ANALYSIS_TARGET_GRID_SIZE_M or 30.0)),
"--project-name",
run_name,
"--date",
@@ -110,9 +123,13 @@ def run_lt_gamma_scene_preprocess(
"--satellite-family",
"LT1",
"--range-looks",
str(DEFAULT_RANGE_LOOKS),
str(int(settings.SAR_ANALYSIS_RANGE_LOOKS or DEFAULT_RANGE_LOOKS)),
"--azimuth-looks",
str(DEFAULT_AZIMUTH_LOOKS),
str(int(settings.SAR_ANALYSIS_AZIMUTH_LOOKS or DEFAULT_AZIMUTH_LOOKS)),
"--speckle-filter-method",
str(settings.SAR_ANALYSIS_SPECKLE_FILTER_METHOD or "none"),
"--speckle-filter-size",
str(int(settings.SAR_ANALYSIS_SPECKLE_FILTER_SIZE or 5)),
"--geo-interp",
str(settings.PYINT_GEO_INTERP or "1"),
"--nodata-value",
+77 -4
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional, Sequence
from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence
from sqlalchemy import delete, func, or_, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession
@@ -380,6 +380,15 @@ def _incremental_insert_sql() -> str:
class PairingCacheService:
async def _notify_progress(
self,
progress_callback: Optional[Callable[[str, int], Awaitable[None]]],
message: str,
progress: int,
) -> None:
if progress_callback is not None:
await progress_callback(message, progress)
async def _get_state_row(self, db: AsyncSession) -> PairingCacheStateORM:
payload = await pairing_state_service.ensure_pairing_cache_state(db, commit=False)
result = await db.execute(
@@ -501,11 +510,24 @@ class PairingCacheService:
db: AsyncSession,
*,
commit: bool = True,
progress_callback: Optional[Callable[[str, int], Awaitable[None]]] = None,
) -> Dict[str, Any]:
await pairing_state_service.ensure_pairing_cache_state(db, commit=False)
await self._set_state_rebuilding(db)
if commit:
await db.commit()
try:
await self._notify_progress(
progress_callback,
"Pairing cache full rebuild: clearing old metric rows",
10,
)
delete_result = await db.execute(delete(PairingMetricCacheORM))
await self._notify_progress(
progress_callback,
"Pairing cache full rebuild: computing spatial/temporal metrics",
20,
)
await db.execute(
text(_full_rebuild_insert_sql()),
{
@@ -513,7 +535,17 @@ class PairingCacheService:
"orientation_rule_version": pairing_state_service.orientation_rule_version,
},
)
await self._notify_progress(
progress_callback,
"Pairing cache full rebuild: resolving dirty scene markers",
80,
)
resolved_dirty = await self._resolve_dirty_rows(db)
await self._notify_progress(
progress_callback,
"Pairing cache full rebuild: finalizing state",
90,
)
summary = await self._finalize_state_success(db, full_rebuild=True)
if commit:
await db.commit()
@@ -539,6 +571,7 @@ class PairingCacheService:
*,
force_full: bool = False,
commit: bool = True,
progress_callback: Optional[Callable[[str, int], Awaitable[None]]] = None,
) -> Dict[str, Any]:
await pairing_state_service.ensure_pairing_cache_state(db, commit=False)
dirty_result = await db.execute(
@@ -557,12 +590,21 @@ class PairingCacheService:
pair_count=pair_count,
force_full=force_full,
):
result = await self.rebuild_metric_cache(db, commit=commit)
result = await self.rebuild_metric_cache(
db,
commit=commit,
progress_callback=progress_callback,
)
result["trigger_dirty_scene_count"] = dirty_scene_count
result["forced"] = force_full
return result
if dirty_scene_count == 0:
await self._notify_progress(
progress_callback,
"Pairing cache reconcile: no dirty scenes, refreshing state",
80,
)
summary = await self._finalize_state_success(db, full_rebuild=False)
if commit:
await db.commit()
@@ -582,13 +624,24 @@ class PairingCacheService:
pair_count=pair_count,
force_full=force_full,
):
result = await self.rebuild_metric_cache(db, commit=commit)
result = await self.rebuild_metric_cache(
db,
commit=commit,
progress_callback=progress_callback,
)
result["trigger_dirty_scene_count"] = dirty_scene_count
result["forced"] = force_full
return result
await self._set_state_rebuilding(db)
if commit:
await db.commit()
try:
await self._notify_progress(
progress_callback,
f"Pairing cache incremental reconcile: deleting stale rows for {dirty_scene_count} dirty scenes",
20,
)
delete_result = await db.execute(
delete(PairingMetricCacheORM).where(
or_(
@@ -600,7 +653,7 @@ class PairingCacheService:
insert_attempts = 0
insert_sql = text(_incremental_insert_sql())
for dirty_scene_id in dirty_scene_ids:
for index, dirty_scene_id in enumerate(dirty_scene_ids, start=1):
insert_result = await db.execute(
insert_sql,
{
@@ -610,8 +663,28 @@ class PairingCacheService:
},
)
insert_attempts += int(insert_result.rowcount or 0)
if index == 1 or index == dirty_scene_count or index % 25 == 0:
progress = 20 + int(index / max(1, dirty_scene_count) * 55)
await self._notify_progress(
progress_callback,
(
"Pairing cache incremental reconcile: "
f"processed {index}/{dirty_scene_count} dirty scenes"
),
progress,
)
await self._notify_progress(
progress_callback,
"Pairing cache incremental reconcile: resolving dirty scene markers",
80,
)
resolved_dirty = await self._resolve_dirty_rows(db, scene_ids=dirty_scene_ids)
await self._notify_progress(
progress_callback,
"Pairing cache incremental reconcile: finalizing state",
90,
)
summary = await self._finalize_state_success(db, full_rebuild=False)
if commit:
await db.commit()
@@ -207,15 +207,20 @@ def _raster_quality(path: Path) -> dict[str, Any]:
return quality
def _is_geographic_crs(crs_text: str) -> bool:
text = str(crs_text or "").upper()
return "4326" in text or "GEOGCS" in text or 'UNIT["DEGREE"' in text or "UNIT['DEGREE'" in text
def _pixel_size_m_from_quality(quality: dict[str, Any]) -> float | None:
try:
transform = quality.get("transform") or []
xres = abs(float(transform[0]))
yres = abs(float(transform[4]))
crs = str(quality.get("crs") or "").upper()
crs = str(quality.get("crs") or "")
if not xres or not yres:
return None
if crs and "4326" not in crs:
if crs and not _is_geographic_crs(crs):
return round((xres + yres) / 2.0, 3)
bounds = quality.get("bounds") or {}
lat = (float(bounds.get("bottom", 0.0)) + float(bounds.get("top", 0.0))) / 2.0