Spaces:
Running on Zero
Running on Zero
File size: 6,320 Bytes
2407511 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Authoritative lifecycle checks for SAR-generated optical-like evidence."""
from __future__ import annotations
from typing import Any, Mapping
from .models import AgentResponse, Confidence, ConfidenceLevel, EvidenceLifecycleState
GENERATED_EVIDENCE_TYPES = {
"generated_optical_like",
"generated_optical_like_representation",
"generated_representation_grounding",
}
def authoritative_translation_products_dict(response: Mapping[str, Any]) -> list[dict[str, Any]]:
analysis = response.get("sar_translated_optical_analysis")
primary = response.get("primary_image_metadata")
if not isinstance(analysis, Mapping) or not isinstance(primary, Mapping):
return []
if analysis.get("generation_state") != EvidenceLifecycleState.SUCCEEDED.value:
return []
products: list[dict[str, Any]] = []
for product in analysis.get("evidence_products") or []:
if not isinstance(product, dict):
continue
if (
product.get("evidence_id")
and product.get("status") == EvidenceLifecycleState.SUCCEEDED.value
and product.get("evidence_type") == "generated_optical_like"
and product.get("generator") == analysis.get("model")
and product.get("evidence_run_id") == response.get("request_id")
and product.get("source_observation_id") == primary.get("file_id")
and str(product.get("source_modality") or "").startswith("sar")
and product.get("reference")
):
products.append(dict(product))
return products
def normalize_translation_evidence(response: AgentResponse) -> AgentResponse:
"""Suppress mismatched/legacy products and prevent unsupported semantic claims."""
analysis = response.sar_translated_optical_analysis
if analysis is None:
return response
dumped = response.model_dump(mode="json")
valid_products = authoritative_translation_products_dict(dumped)
valid_ids = {item["evidence_id"] for item in valid_products if item.get("evidence_id")}
had_generated_artifact = bool(
analysis.generated_preview_url
or analysis.evidence_products
or any(item.evidence_type in GENERATED_EVIDENCE_TYPES or item.type in GENERATED_EVIDENCE_TYPES for item in response.evidence)
)
generated_evidence = [
item for item in response.evidence
if item.evidence_type not in GENERATED_EVIDENCE_TYPES and item.type not in GENERATED_EVIDENCE_TYPES
]
generated_evidence.extend(
item for item in analysis.evidence_products if item.evidence_id in valid_ids
)
generation_is_publishable = bool(valid_products)
semantic_is_usable = generation_is_publishable and analysis.semantic_comparison_state == EvidenceLifecycleState.SUCCEEDED
warning: str | None = None
if not generation_is_publishable:
warning = (
"Legacy generated evidence provenance unavailable; the generated evidence product was suppressed."
if had_generated_artifact and analysis.generation_state == EvidenceLifecycleState.NOT_REQUESTED
else "Pix2Pix translation evidence was unavailable; native SAR evidence was retained."
)
direct_answer = " ".join(analysis.native_sar_findings[:1])
direct_answer = (direct_answer + " " if direct_answer else "") + "The generated optical-like representation is unavailable, so no translated evidence was used."
analysis = analysis.model_copy(update={
"generated_preview_url": None,
"normalized_sar_preview_url": None,
"color_corrected_preview_url": None,
"evidence_products": [],
"optical_caption": None,
"optical_scene_priors": [],
"optical_grounding": None,
"optical_vqa": None,
"optical_specialists_executed": [],
"translated_findings": [],
"semantic_comparison_state": EvidenceLifecycleState.FAILED if analysis.semantic_comparison_state == EvidenceLifecycleState.FAILED else EvidenceLifecycleState.NOT_REQUESTED,
"agreement": "translation_unavailable",
"direct_answer": direct_answer,
})
elif not semantic_is_usable:
direct_answer = " ".join(analysis.native_sar_findings[:1])
direct_answer = (direct_answer + " " if direct_answer else "") + "Translation semantic comparison was unavailable, so no translation-derived claim was used."
analysis = analysis.model_copy(update={
"generated_preview_url": valid_products[0]["reference"],
"evidence_products": [item for item in analysis.evidence_products if item.evidence_id in valid_ids],
"optical_caption": None,
"optical_scene_priors": [],
"optical_grounding": None,
"optical_vqa": None,
"optical_specialists_executed": [],
"translated_findings": [],
"agreement": "supported_by_native_sar_only" if analysis.native_sar_findings else "insufficient_evidence",
"direct_answer": direct_answer,
})
else:
analysis = analysis.model_copy(update={
"generated_preview_url": valid_products[0]["reference"],
"evidence_products": [item for item in analysis.evidence_products if item.evidence_id in valid_ids],
})
warnings = list(dict.fromkeys(response.warnings + ([warning] if warning else [])))
confidence = analysis.confidence
if not semantic_is_usable:
confidence = Confidence(
level=ConfidenceLevel.LOW if analysis.native_sar_findings else ConfidenceLevel.UNAVAILABLE,
score=min(confidence.score, 0.4) if confidence.score is not None and analysis.native_sar_findings else None,
reason="Only native SAR evidence contributed to this answer; translation-derived semantics were not used.",
)
analysis = analysis.model_copy(update={"confidence": confidence})
return response.model_copy(update={
"answer": analysis.direct_answer if not semantic_is_usable else response.answer,
"confidence": confidence if not semantic_is_usable else response.confidence,
"evidence": generated_evidence,
"warnings": warnings,
"sar_translated_optical_analysis": analysis,
})
|