"""Metadata-based sensor and pair semantics. Appearance never establishes time.""" from __future__ import annotations from datetime import datetime, UTC from satquery_engine.schemas import RasterMetadata def acquisition_time(value): date=datetime.fromisoformat(value.replace("Z","+00:00")) return date.replace(tzinfo=UTC) if date.tzinfo is None else date.astimezone(UTC) def classify_configuration(assets: list[RasterMetadata], pair_type: str = "auto") -> tuple[str, str | None]: if len(assets) not in (1, 2): return "INVALID", "Please upload one or two supported images." if len(assets) == 1: if assets[0].modality == "optical" and assets[0].bands == 3: return "SINGLE_RGB", None return {"optical": "SINGLE_OPTICAL", "multispectral": "SINGLE_MULTISPECTRAL", "sar": "SINGLE_SAR"}.get(assets[0].modality, "UNKNOWN"), None a, b = assets cross = (a.modality == "sar") != (b.modality == "sar") and "unknown" not in (a.modality, b.modality) dated = False if a.acquisition_date and b.acquisition_date: try: dated = acquisition_time(a.acquisition_date) != acquisition_time(b.acquisition_date) except ValueError: pass if pair_type == "optical_sar" or cross: if not cross: return "PAIR_AMBIGUOUS", "I cannot verify which image is radar imagery from its metadata. Please supply imagery with sensor or polarization metadata." return "PAIR_MULTISENSOR_BITEMPORAL" if dated else "PAIR_OPTICAL_SAR", None if pair_type == "bi_temporal" or dated: return "PAIR_BITEMPORAL_SAR" if a.modality == b.modality == "sar" else "PAIR_BITEMPORAL_OPTICAL", None return "PAIR_AMBIGUOUS", "Are these images of the same place at different dates, in before-then-after order?" def policy_blockers(plan, assets, configuration): blockers = [] if plan.requires_pair and len(assets) != 2: blockers.append("Please upload two images to answer this comparison question.") if plan.requires_temporal_relationship and configuration not in {"PAIR_BITEMPORAL_OPTICAL", "PAIR_BITEMPORAL_SAR"}: blockers.append("A verified pair of images from different dates is required for change analysis.") if any(n.tool == "fusion" for n in plan.nodes) and configuration not in {"PAIR_OPTICAL_SAR", "PAIR_MULTISENSOR_BITEMPORAL"}: blockers.append("This request needs one optical image and one SAR image with identifiable sensor metadata.") if any(n.tool == "buildings" for n in plan.nodes) and any(a.modality not in {"optical", "multispectral"} for a in assets): blockers.append("The installed building specialist accepts optical RGB imagery; SAR building counting is unavailable.") return blockers def node_policy_error(node,assets): """Run before the node invokes any model; failed independent branches can abstain.""" if node.tool=="fusion": from satquery_engine.services.sar import croma_compatibility errors=croma_compatibility(assets) return " ".join(errors) if errors else None if node.tool in {"spectral","spectral_change"}: if node.tool == "spectral" and node.parameters["target"] in {"water", "built-up"} and not node.parameters.get("strict"): return None # Specialist performs its own sensor/radiometry preflight. from satquery_engine.services.spectral import TARGET_INDEX index=TARGET_INDEX[node.parameters["target"]] selected=assets if node.tool=="spectral_change" else assets[:1] for asset in selected: if asset.modality=="sar": return "Optical spectral indices cannot be calculated from SAR backscatter." if index not in asset.available_indices and (node.parameters.get("strict") or node.tool=="spectral_change" or asset.bands!=3): return f"I cannot calculate {index.upper()} because the required named spectral bands are missing." if node.tool=="buildings" and node.parameters["asset"]>=len(assets): return "Building comparison requires two images." return None