Spaces:
Running on Zero
Running on Zero
| """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 | |
| 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, | |
| ) | |