"""Conservative local semantic interpretation of existing bi-temporal evidence. This module does not inspect model internals, alter masks, or invoke inference. It turns already-published change geometry, deterministic differences, and optional SVE scene-prior deltas into a structured intermediate representation and prose. """ from __future__ import annotations import math import os import re from dataclasses import dataclass from typing import Any, Dict, Iterable, Optional, Sequence from ..models import BuiltUpChangeAssessment, ChangeAnalysisResponse, SemanticChangeSummary, SemanticTransition SEMANTIC_INTERPRETER_VERSION = "bitemporal-semantic-1.0" def _float_env(name: str, default: float, minimum: float, maximum: float) -> float: try: return min(maximum, max(minimum, float(os.getenv(name, str(default))))) except (TypeError, ValueError): return default @dataclass(frozen=True) class SemanticInterpreterConfig: minimal_change_percent: float = 0.5 localized_change_percent: float = 2.0 noticeable_change_percent: float = 10.0 widespread_change_percent: float = 30.0 semantic_delta_threshold: float = 0.035 strong_semantic_delta: float = 0.08 disagreement_gap_percent: float = 10.0 disagreement_ratio: float = 4.0 DEFAULT_CONFIG = SemanticInterpreterConfig( semantic_delta_threshold=_float_env("SATQUERY_SEMANTIC_DELTA_THRESHOLD", 0.035, 0.001, 0.5), strong_semantic_delta=_float_env("SATQUERY_SEMANTIC_STRONG_DELTA", 0.08, 0.001, 1.0), disagreement_gap_percent=_float_env("SATQUERY_CHANGE_DISAGREEMENT_GAP", 10.0, 0.0, 100.0), disagreement_ratio=_float_env("SATQUERY_CHANGE_DISAGREEMENT_RATIO", 4.0, 1.0, 100.0), ) _FAMILIES: Dict[str, tuple[str, ...]] = { "built_up": ("urban or built-up area", "residential area", "industrial area"), "vegetation": ("vegetation", "forest", "grassland"), "water": ("inland water", "marine water", "wetland"), "infrastructure": ("road network", "airport", "harbor"), } def classify_change_intent(question: str) -> str: """Classify a bounded semantic intent without changing agent routing.""" normalized = re.sub(r"[^a-z0-9]+", " ", (question or "").lower()).strip() if not normalized: return "unknown" if any(term in normalized for term in ("what caused", "why did", "who caused", "reason for")): return "unknown" if "largest" in normalized and any(term in normalized for term in ("where", "location", "occur")): return "largest_change" if any(term in normalized for term in ("how much", "percentage", "percent", "amount of change")): return "amount_of_change" if any(term in normalized for term in ("built up", "builtup", "urban", "building", "construction", "developed")): return "built_up_change" if "structural" in normalized: return "structural_change" if any(term in normalized for term in ("vegetation", "forest", "trees", "green cover", "grassland")): return "vegetation_change" if any(term in normalized for term in ("water", "river", "lake", "coast", "shoreline", "wetland")): return "water_change" if any(term in normalized for term in ("road", "infrastructure", "airport", "harbor")): return "infrastructure_change" if any(term in normalized for term in ("no change", "unchanged", "remain the same", "stayed the same", "mostly similar")): return "no_change_check" if any(term in normalized for term in ("where", "location", "which part", "which area")) and "change" in normalized: return "change_location" if any(term in normalized for term in ("compare", "difference between", "how different")): return "general_comparison" if any(term in normalized for term in ("what changed", "describe the change", "change between", "changes between")): return "change_summary" return "unknown" def _requested_direction(question: str) -> Optional[str]: normalized = (question or "").lower() if any(term in normalized for term in ("increase", "increased", "expand", "expanded", "growth", "grew", "more")): return "increase" if any(term in normalized for term in ("decrease", "decreased", "shrink", "shrunk", "loss", "lost", "less", "reduced")): return "decrease" return None def _finite_number(value: Any) -> Optional[float]: try: parsed = float(value) except (TypeError, ValueError): return None return parsed if math.isfinite(parsed) else None def _location(box: Any, width: int, height: int) -> Optional[str]: try: x = (float(box.left) + float(box.right)) / (2.0 * max(width, 1)) y = (float(box.top) + float(box.bottom)) / (2.0 * max(height, 1)) except (AttributeError, TypeError, ValueError): return None horizontal = "left" if x < 1 / 3 else "right" if x > 2 / 3 else "central" vertical = "upper" if y < 1 / 3 else "lower" if y > 2 / 3 else "central" if horizontal == vertical == "central": return "center" if horizontal == "central": return f"{vertical}-central" if vertical == "central": return f"central-{horizontal}" return f"{vertical}-{horizontal}" def _magnitude(value: float, config: SemanticInterpreterConfig) -> tuple[str, str]: if value < config.minimal_change_percent: return "minimal", "very limited change" if value < config.localized_change_percent: return "localized", "localized change" if value < config.noticeable_change_percent: return "moderate", "noticeable but limited change" if value < config.widespread_change_percent: return "moderate", "substantial change" return "widespread", "widespread visual difference" def _prior_changes(change: ChangeAnalysisResponse) -> list[dict[str, Any]]: sve = getattr(change, "sve_result", None) comparison = getattr(sve, "semantic_comparison", None) if sve and getattr(sve, "available", False) else None if comparison is None or getattr(comparison, "status", "") != "supporting_evidence": return [] output = [] for item in getattr(comparison, "prior_changes", []) or []: if not isinstance(item, dict) or not isinstance(item.get("label"), str): continue difference = _finite_number(item.get("difference")) if difference is None: continue output.append({"label": item["label"], "difference": difference}) return output def _transition_for_family( family: str, changes: Sequence[dict[str, Any]], config: SemanticInterpreterConfig, ) -> Optional[SemanticTransition]: labels = set(_FAMILIES[family]) matching = [item for item in changes if item["label"] in labels and abs(item["difference"]) >= config.semantic_delta_threshold] if not matching: return None strongest = max(matching, key=lambda item: abs(item["difference"])) direction = "increase" if strongest["difference"] > 0 else "decrease" aligned = [item for item in matching if (item["difference"] > 0) == (strongest["difference"] > 0)] confidence = "strong" if len(aligned) >= 2 and abs(strongest["difference"]) >= config.strong_semantic_delta else "moderate" if len(aligned) >= 2 or abs(strongest["difference"]) >= config.strong_semantic_delta else "weak" transition_type = "infrastructure_change" if family == "infrastructure" else f"{family}_{direction}" evidence = [ f"SVE relative scene evidence for {item['label']} moved {'up' if item['difference'] > 0 else 'down'} between observations." for item in aligned[:3] ] return SemanticTransition(type=transition_type, confidence=confidence, evidence=evidence) def _claim_sentence(transition: SemanticTransition) -> str: prefix = {"strong": "indicates", "moderate": "appears to show", "weak": "may indicate"}[transition.confidence] sentences = { "built_up_increase": f"Scene-level evidence {prefix} stronger built-up characteristics in the later observation.", "built_up_decrease": f"Scene-level evidence {prefix} weaker built-up characteristics in the later observation.", "vegetation_increase": f"Scene-level evidence {prefix} more vegetation in the later observation.", "vegetation_decrease": f"Scene-level evidence {prefix} less vegetation in the later observation.", "water_increase": f"Scene-level evidence {prefix} greater water-related presence in the later observation.", "water_decrease": f"Scene-level evidence {prefix} reduced water-related presence in the later observation.", "infrastructure_change": f"Scene-level evidence {prefix} a change in infrastructure-related characteristics.", } return sentences.get(transition.type, "The semantic type of change remains uncertain.") def _family_for_intent(intent: str) -> Optional[str]: return { "built_up_change": "built_up", "vegetation_change": "vegetation", "water_change": "water", "infrastructure_change": "infrastructure", }.get(intent) def _stable_regions(changed_locations: Iterable[str], changed_percent: float) -> list[str]: locations = set(changed_locations) if changed_percent >= 30: return [] candidates = ["central", "left", "right", "upper", "lower"] blocked = set() for location in locations: if "left" in location: blocked.add("left") if "right" in location: blocked.add("right") if "upper" in location: blocked.add("upper") if "lower" in location: blocked.add("lower") if "center" in location or "central" in location: blocked.add("central") return [item for item in candidates if item not in blocked][:2] def interpret_change( question: str, change: ChangeAnalysisResponse, *, config: SemanticInterpreterConfig = DEFAULT_CONFIG, built_up_assessment: Optional[BuiltUpChangeAssessment] = None, ) -> SemanticChangeSummary: """Build structured facts first, then deterministic query-aware prose.""" intent = classify_change_intent(question) statistics = getattr(change, "statistics", None) engine = getattr(change, "change_engine", None) learned = bool(engine and getattr(engine, "mode", "") in {"hybrid", "ttp"}) fallback = bool(engine and getattr(engine, "fallback_used", False)) caveats = ["The binary change detector does not identify the real-world cause of change."] supporting: list[str] = [] if statistics is None: answer = "The image pair could not be compared reliably, so no change interpretation was generated." return SemanticChangeSummary( short_answer=answer, expanded_answer=answer, query_intent=intent, overall_change_level="minimal", evidence_strength="limited", caveats=caveats + ["Comparable-pixel change statistics were unavailable."], ) changed_percent = _finite_number(getattr(statistics, "percentage_changed", None)) changed_percent = min(100.0, max(0.0, changed_percent or 0.0)) level, magnitude_phrase = _magnitude(changed_percent, config) regions = list(getattr(statistics, "regions", None) or []) regions.sort(key=lambda item: getattr(item, "area_pixels", 0), reverse=True) locations = [value for value in (_location(getattr(region, "bounding_box", None), statistics.analysis_width, statistics.analysis_height) for region in regions) if value] dominant_location = locations[0] if locations else None changed_regions = list(dict.fromkeys(locations[:4])) stable_regions = _stable_regions(changed_regions, changed_percent) where = f"the {dominant_location} part of the scene" if dominant_location and dominant_location != "center" else "the center of the scene" if dominant_location else None if changed_percent <= 0 or not regions: supporting.append("No connected changed region was present in the authoritative change statistics.") else: supporting.append(f"The primary change evidence identifies {magnitude_phrase}.") if where: supporting.append(f"The largest connected region is centered in {where}.") deterministic = getattr(change, "deterministic_statistics", None) deterministic_percent = _finite_number(getattr(deterministic, "percentage_changed", None)) if deterministic else None disagreement = bool( learned and deterministic_percent is not None and deterministic_percent - changed_percent >= config.disagreement_gap_percent and deterministic_percent >= max(changed_percent, 0.01) * config.disagreement_ratio ) if disagreement: supporting.append("The deterministic analyzer reports much broader low-level visual difference than the learned mask.") changes = _prior_changes(change) transitions = [value for family in _FAMILIES if (value := _transition_for_family(family, changes, config))] # A semantic transition requires both spatial change evidence and compatible scene-level evidence. if not learned or changed_percent <= 0: transitions = [] requested_family = _family_for_intent(intent) selected = next((item for item in transitions if item.type.startswith(requested_family or "__none__")), None) if selected is None and intent in {"change_summary", "general_comparison", "unknown"} and transitions: order = {"strong": 2, "moderate": 1, "weak": 0} selected = max(transitions, key=lambda item: order[item.confidence]) if changes: caveats.append("SVE similarities are relative scene-level evidence, not calibrated probabilities or spatial segmentation.") else: caveats.append("No usable before/after scene-level semantic evidence was available, so no land-cover transition was assigned.") if fallback: caveats.append("The learned change specialist was unavailable; interpretation is limited to deterministic visual difference.") before_metadata = getattr(change, "before_metadata", None) if before_metadata is not None and not getattr(before_metadata, "is_georeferenced", False): caveats.append("Locations are coarse image directions, not geographic coordinates.") stable_summary = None if changed_percent < config.noticeable_change_percent: stable_summary = "Most of the scene remains comparatively stable outside the detected change areas." if stable_regions: stable_summary = f"The {', '.join(stable_regions)} parts show no comparable concentration of detected change." if changed_percent <= 0 or not regions: base = "The two observations are largely consistent, with no significant localized change detected." elif learned: base = f"The learned evidence identifies {magnitude_phrase}{f' concentrated in {where}' if where else ''}." else: base = f"The available visual-difference evidence identifies {magnitude_phrase}{f' concentrated in {where}' if where else ''}." semantic_sentence = _claim_sentence(selected) if selected else "The available evidence is not strong enough to determine a reliable land-cover transition." requested_direction = _requested_direction(question) if requested_family and selected: actual_direction = "increase" if selected.type.endswith("increase") else "decrease" if selected.type.endswith("decrease") else None if requested_direction and actual_direction and requested_direction != actual_direction: semantic_sentence = f"The available evidence does not support a {requested_direction}; it instead {('suggests' if selected.confidence != 'strong' else 'indicates')} a {actual_direction} in {requested_family.replace('_', '-')} characteristics." disagreement_sentence = "The images differ much more in overall appearance than in the learned structural mask; seasonal, illumination, sensor, or registration effects may contribute to the broader visual difference." if disagreement else None location_answer = f"The most significant detected change is in {where}." if where else "No dominant change location can be stated from the available geometry." if intent in {"largest_change", "change_location"}: short = location_answer elif intent == "amount_of_change": short = f"The primary evidence indicates {magnitude_phrase}, covering approximately {changed_percent:.1f}% of the analysis grid." elif intent == "no_change_check": short = "The observations are largely consistent, with no significant change detected." if changed_percent < config.minimal_change_percent else f"No. The evidence identifies {magnitude_phrase}{f' in {where}' if where else ''}." elif requested_family: short = semantic_sentence if where and selected: short = f"{short.rstrip('.')} The main detected change is concentrated in {where}." else: short = " ".join(item for item in (base, semantic_sentence if selected else None) if item) expanded_parts = [base] if selected: expanded_parts.append(semantic_sentence) supporting.extend(selected.evidence) elif intent not in {"amount_of_change", "largest_change", "change_location", "no_change_check"}: expanded_parts.append(semantic_sentence) if stable_summary: expanded_parts.append(stable_summary) if disagreement_sentence: expanded_parts.append(disagreement_sentence) expanded = " ".join(dict.fromkeys(expanded_parts)) evidence_strength = "limited" if learned: evidence_strength = "moderate" if selected and selected.confidence == "strong": evidence_strength = "strong" likely_type = selected.type if selected else "unknown_change" if changed_percent > 0 else None if built_up_assessment is not None and intent in {"built_up_change", "structural_change"}: assessment = built_up_assessment relevant = [region for region in assessment.regions if region.semantic_support_level == "supported"] or assessment.regions region_names = list(dict.fromkeys(region.relative_location for region in relevant)) where_text = " and ".join(region_names[:2]) if region_names else None strongest = assessment.regions[0] if assessment.regions else None if intent == "structural_change": short = ( f"The strongest structural change is in the {strongest.relative_location} part of the scene." if strongest else "No meaningful structural change region can be localized from the available evidence." ) likely_type = "structural_change" if strongest else None elif assessment.state == "INCREASE_SUPPORTED": short = f"Built-up expansion is suggested{f' in the {where_text} parts of the scene' if where_text else ''}." likely_type = "built_up_increase" elif assessment.state == "DECREASE_SUPPORTED": short = f"A decrease in built-up-like structural evidence is suggested{f' in the {where_text} parts of the scene' if where_text else ''}." likely_type = "built_up_decrease" elif assessment.state == "MIXED_CHANGE": short = f"Built-up-like evidence shows mixed directional change{f' across the {where_text} regions' if where_text else ''}." likely_type = "built_up_mixed_change" elif assessment.state == "NO_MEANINGFUL_EVIDENCE": short = "No meaningful evidence of built-up-region change was found." likely_type = None else: short = "Change is present, but the evidence is insufficient to establish whether built-up area increased or decreased." likely_type = "unknown_change" where_sentence = ( f"The largest relevant changed cluster is in the {strongest.relative_location} part of the image; up to {len(assessment.regions)} meaningful region{'s were' if len(assessment.regions) != 1 else ' was'} compared." if strongest else "No meaningful connected region passed the region-size filter." ) magnitude_sentence = f"The pattern is {assessment.magnitude}, based on changed-area share and region distribution." why_sentence = ( "This direction combines ChangerEx geometry, independent deterministic overlap, and before/after edge-texture structural evidence. " "It is a heuristic interpretation consistent with built-up terrain, not a ground-truth land-cover measurement." ) expanded = " ".join((short, where_sentence, magnitude_sentence, why_sentence)) dominant_location = strongest.relative_location if strongest else dominant_location changed_regions = [region.relative_location for region in assessment.regions] evidence_strength = assessment.confidence supporting = list(dict.fromkeys(supporting + assessment.confidence_factors)) caveats = list(dict.fromkeys(caveats + assessment.limitations)) return SemanticChangeSummary( short_answer=short, expanded_answer=expanded, query_intent=intent, overall_change_level=level, dominant_location=dominant_location, likely_change_type=likely_type, evidence_strength=evidence_strength, stable_area_summary=stable_summary, changed_regions=changed_regions, stable_regions=stable_regions, likely_transitions=transitions, supporting_facts=list(dict.fromkeys(supporting)), caveats=list(dict.fromkeys(caveats)), visual_structural_disagreement=disagreement, built_up_assessment=built_up_assessment, )