""" Reality Gap Index (RGI) - The Intelligence Engine Quantifies the difference between declared/planned reality and observed reality """ from typing import Dict, List, Optional, Tuple from dataclasses import dataclass from enum import Enum from datetime import datetime class RealityLayer(str, Enum): """Seven reality layers""" PLANNED = "planned" # Layer 1: What was planned CONSTRUCTED = "constructed" # Layer 2: What was built MAINTAINED = "maintained" # Layer 3: What is maintained OBSERVED = "observed" # Layer 4: What is observed HUMAN_EXPERIENCE = "human_experience" # Layer 5: What people experience INFORMAL_ADAPTATION = "informal_adaptation" # Layer 6: Community adaptations HIDDEN_SYSTEMS = "hidden_systems" # Layer 7: Hidden systems (evidence-based only) class EvidenceType(str, Enum): """Types of evidence""" IMAGE = "image" SENSOR = "sensor" CROWD = "crowd" OFFICIAL = "official" CORROBORATED = "corroborated" class ConfidenceLevel(str, Enum): """Confidence levels""" HIGH = "high" MEDIUM = "medium" LOW = "low" UNKNOWN = "unknown" @dataclass class Evidence: """Evidence for a gap measurement""" type: EvidenceType description: str confidence: float # 0-1 source: str timestamp: Optional[str] = None def to_dict(self) -> Dict: return { "type": self.type.value, "description": self.description, "confidence": round(self.confidence, 2), "source": self.source, "timestamp": self.timestamp } @dataclass class GapScore: """Score for a specific gap dimension""" dimension: str score: float # 0-100 confidence: float # 0-1 evidence: List[Evidence] observations: List[str] def to_dict(self) -> Dict: return { "dimension": self.dimension, "score": round(self.score, 2), "confidence": round(self.confidence, 2), "evidence": [e.to_dict() for e in self.evidence], "observations": self.observations } @dataclass class RealityGapIndex: """Complete RGI result""" location: str timestamp: str overall_rgi: float # 0-100 # Subscores physical_gap: GapScore # PRG operational_gap: GapScore # ORG safety_gap: GapScore # SRG institutional_gap: GapScore # IRG economic_gap: GapScore # ERG maintenance_gap: GapScore # MRG human_experience_gap: GapScore # HEG # Reality layers layer_scores: Dict[str, float] # Contradiction analysis contradictions: List[Dict] contradiction_level: str # high, medium, low # Metadata observation_quality: float data_sources: List[str] def to_dict(self) -> Dict: return { "location": self.location, "timestamp": self.timestamp, "overall_rgi": round(self.overall_rgi, 2), "subscores": { "physical_reality_gap": self.physical_gap.to_dict(), "operational_reality_gap": self.operational_gap.to_dict(), "safety_reality_gap": self.safety_gap.to_dict(), "institutional_reality_gap": self.institutional_gap.to_dict(), "economic_reality_gap": self.economic_gap.to_dict(), "maintenance_reality_gap": self.maintenance_gap.to_dict(), "human_experience_gap": self.human_experience_gap.to_dict() }, "reality_layers": self.layer_scores, "contradictions": self.contradictions, "contradiction_level": self.contradiction_level, "observation_quality": round(self.observation_quality, 2), "data_sources": self.data_sources } class RealityGapAnalyzer: """Analyzes reality gaps from observations""" def __init__(self): self.dimension_weights = { "physical": 0.20, "operational": 0.15, "safety": 0.15, "institutional": 0.15, "economic": 0.10, "maintenance": 0.10, "human_experience": 0.15 } def analyze( self, location: str, observations: List[Dict], official_data: Optional[Dict] = None, planned_data: Optional[Dict] = None ) -> RealityGapIndex: """ Analyze reality gap for a location Args: location: Location identifier observations: IOM observations official_data: Official reports/data planned_data: Planning documents Returns: Complete RGI analysis """ # Analyze each dimension physical = self._analyze_physical_gap(observations, planned_data) operational = self._analyze_operational_gap(observations, official_data) safety = self._analyze_safety_gap(observations) institutional = self._analyze_institutional_gap(observations, official_data) economic = self._analyze_economic_gap(observations, official_data) maintenance = self._analyze_maintenance_gap(observations) human_exp = self._analyze_human_experience_gap(observations) # Calculate layer scores layer_scores = self._calculate_layer_scores( observations, official_data, planned_data ) # Calculate overall RGI overall = self._calculate_overall_rgi( physical, operational, safety, institutional, economic, maintenance, human_exp ) # Find contradictions contradictions = self._find_contradictions( observations, official_data, planned_data ) # Determine contradiction level contradiction_level = self._determine_contradiction_level( contradictions, overall ) # Calculate observation quality observation_quality = self._calculate_observation_quality(observations) # List data sources data_sources = self._list_data_sources(observations, official_data) return RealityGapIndex( location=location, timestamp=datetime.utcnow().isoformat(), overall_rgi=overall, physical_gap=physical, operational_gap=operational, safety_gap=safety, institutional_gap=institutional, economic_gap=economic, maintenance_gap=maintenance, human_experience_gap=human_exp, layer_scores=layer_scores, contradictions=contradictions, contradiction_level=contradiction_level, observation_quality=observation_quality, data_sources=data_sources ) def _analyze_physical_gap( self, observations: List[Dict], planned_data: Optional[Dict] ) -> GapScore: """Analyze Physical Reality Gap (PRG)""" evidence = [] observations_list = [] score = 0.0 # Compare with plans if planned_data: planned_buildings = planned_data.get("planned_buildings", 0) actual = len(observations) if planned_buildings > 0: ratio = actual / planned_buildings if ratio > 1.5: score += 30.0 observations_list.append(f"Significant overbuilding: {actual} vs {planned_buildings} planned") evidence.append(Evidence( type=EvidenceType.OFFICIAL, description=f"Planned {planned_buildings}, observed {actual}", confidence=0.8, source="planning_document" )) # Check for informal construction informal_count = 0 for obs in observations: findings = obs.get("findings", []) for finding in findings: code = finding.get("code", "") if code in ["2300", "3300", "4100", "4200"]: informal_count += 1 observations_list.append(f"Informal construction: {finding.get('type', 'unknown')}") if len(observations) > 0: informal_ratio = informal_count / len(observations) score += informal_ratio * 50.0 # Check for deterioration deteriorated = sum( 1 for obs in observations if obs.get("overall_condition", 3) >= 4 ) if len(observations) > 0: deterioration_ratio = deteriorated / len(observations) score += deterioration_ratio * 20.0 score = min(100.0, score) confidence = 0.7 if len(observations) > 5 else 0.5 return GapScore( dimension="Physical Reality Gap", score=score, confidence=confidence, evidence=evidence, observations=observations_list ) def _analyze_operational_gap( self, observations: List[Dict], official_data: Optional[Dict] ) -> GapScore: """Analyze Operational Reality Gap (ORG)""" evidence = [] observations_list = [] score = 0.0 # Check if systems actually work # School exists but can children reach it? # Hospital exists but can patients access it? # Look for accessibility indicators blocked = 0 for obs in observations: findings = obs.get("findings", []) for finding in findings: code = finding.get("code", "") if code in ["5100", "5200"]: # Blockages blocked += 1 observations_list.append("Access blocked or obstructed") if len(observations) > 0: score = (blocked / len(observations)) * 100.0 # Check for functional indicators non_functional = sum( 1 for obs in observations if obs.get("function_status", "operational") != "operational" ) if len(observations) > 0: score += (non_functional / len(observations)) * 50.0 score = min(100.0, score) return GapScore( dimension="Operational Reality Gap", score=score, confidence=0.6, evidence=evidence, observations=observations_list ) def _analyze_safety_gap(self, observations: List[Dict]) -> GapScore: """Analyze Safety Reality Gap (SRG)""" evidence = [] observations_list = [] score = 0.0 # Only score observable indicators safety_issues = { "1100": "Surface rust", "1200": "Deep corrosion", "1300": "Crack", "1400": "Impact damage", "1600": "Tilt/settlement", "1700": "Vibration", "6200": "Water damage", "6300": "Ice damage" } for obs in observations: findings = obs.get("findings", []) for finding in findings: code = finding.get("code", "") if code in safety_issues: score += 10.0 observations_list.append(f"Safety issue: {safety_issues[code]}") evidence.append(Evidence( type=EvidenceType.IMAGE, description=f"Observed {safety_issues[code]}", confidence=0.8, source="field_observation" )) score = min(100.0, score) return GapScore( dimension="Safety Reality Gap", score=score, confidence=0.75, evidence=evidence, observations=observations_list ) def _analyze_institutional_gap( self, observations: List[Dict], official_data: Optional[Dict] ) -> GapScore: """Analyze Institutional Reality Gap (IRG)""" evidence = [] observations_list = [] score = 0.0 # Look for informal commerce, settlements, encroachment informal_indicators = { "2100": "Dirt accumulation (high traffic)", "2400": "Graffiti", "4100": "Missing parts", "4200": "Broken parts", "5100": "Physical blockage" } for obs in observations: findings = obs.get("findings", []) for finding in findings: code = finding.get("code", "") if code in informal_indicators: score += 8.0 observations_list.append(f"Institutional gap: {informal_indicators[code]}") # Compare with official governance if official_data: official_control = official_data.get("institutional_control", "full") if official_control == "limited": score += 30.0 observations_list.append("Limited institutional control reported") elif official_control == "none": score += 60.0 observations_list.append("No institutional control reported") score = min(100.0, score) return GapScore( dimension="Institutional Reality Gap", score=score, confidence=0.6, evidence=evidence, observations=observations_list ) def _analyze_economic_gap( self, observations: List[Dict], official_data: Optional[Dict] ) -> GapScore: """Analyze Economic Reality Gap (ERG)""" evidence = [] observations_list = [] score = 0.0 # Look for contrast between investment and functionality conditions = [obs.get("overall_condition", 3) for obs in observations] if len(conditions) > 1: variance = max(conditions) - min(conditions) # High variance = economic contrast score = variance * 20.0 if variance >= 3: observations_list.append(f"High economic contrast: condition variance {variance}") evidence.append(Evidence( type=EvidenceType.IMAGE, description="Visible economic contrast in built environment", confidence=0.7, source="field_observation" )) # Check for luxury vs improvised luxury = sum(1 for c in conditions if c <= 2) improvised = sum(1 for c in conditions if c >= 4) if luxury > 0 and improvised > 0: score += 30.0 observations_list.append(f"Luxury ({luxury}) and improvised ({improvised}) coexist") score = min(100.0, score) return GapScore( dimension="Economic Reality Gap", score=score, confidence=0.65, evidence=evidence, observations=observations_list ) def _analyze_maintenance_gap(self, observations: List[Dict]) -> GapScore: """Analyze Maintenance Reality Gap (MRG)""" evidence = [] observations_list = [] score = 0.0 # Look for reactive vs systematic maintenance maintenance_indicators = { "1100": "Surface rust", "1200": "Deep corrosion", "2100": "Dirt accumulation", "2200": "Color change", "2300": "Surface damage", "6200": "Water damage" } for obs in observations: findings = obs.get("findings", []) for finding in findings: code = finding.get("code", "") if code in maintenance_indicators: score += 7.0 observations_list.append(f"Maintenance needed: {maintenance_indicators[code]}") # Check for temporary fixes temporary = sum( 1 for obs in observations if any(f.get("code", "") in ["4100", "4200", "4300"] for f in obs.get("findings", [])) ) if len(observations) > 0: score += (temporary / len(observations)) * 30.0 score = min(100.0, score) return GapScore( dimension="Maintenance Reality Gap", score=score, confidence=0.7, evidence=evidence, observations=observations_list ) def _analyze_human_experience_gap(self, observations: List[Dict]) -> GapScore: """Analyze Human Experience Gap (HEG)""" evidence = [] observations_list = [] score = 0.0 # Evaluate from citizen perspective # What does average person experience vs official reports? # Poor conditions affect human experience poor_conditions = sum( 1 for obs in observations if obs.get("overall_condition", 3) >= 4 ) if len(observations) > 0: score = (poor_conditions / len(observations)) * 60.0 # Accessibility issues accessibility = sum( 1 for obs in observations if any(f.get("code", "") in ["5100", "5200"] for f in obs.get("findings", [])) ) if len(observations) > 0: score += (accessibility / len(observations)) * 40.0 score = min(100.0, score) return GapScore( dimension="Human Experience Gap", score=score, confidence=0.6, evidence=evidence, observations=observations_list ) def _calculate_layer_scores( self, observations: List[Dict], official_data: Optional[Dict], planned_data: Optional[Dict] ) -> Dict[str, float]: """Calculate scores for each reality layer""" scores = {} # Layer 1: Planned scores["planned"] = 100.0 # Baseline # Layer 2: Constructed if planned_data: planned_count = planned_data.get("planned_buildings", len(observations)) actual = len(observations) if planned_count > 0: scores["constructed"] = min(100.0, (actual / planned_count) * 100.0) else: scores["constructed"] = 100.0 else: scores["constructed"] = 100.0 # Layer 3: Maintained good_condition = sum( 1 for obs in observations if obs.get("overall_condition", 3) <= 2 ) if len(observations) > 0: scores["maintained"] = (good_condition / len(observations)) * 100.0 else: scores["maintained"] = 100.0 # Layer 4: Observed scores["observed"] = 100.0 # Baseline # Layer 5: Human Experience accessible = sum( 1 for obs in observations if not any(f.get("code", "") in ["5100", "5200"] for f in obs.get("findings", [])) ) if len(observations) > 0: scores["human_experience"] = (accessible / len(observations)) * 100.0 else: scores["human_experience"] = 100.0 # Layer 6: Informal Adaptation informal = sum( 1 for obs in observations if any(f.get("code", "") in ["2100", "2300", "2400", "4100", "4200"] for f in obs.get("findings", [])) ) if len(observations) > 0: scores["informal_adaptation"] = (informal / len(observations)) * 100.0 else: scores["informal_adaptation"] = 0.0 # Layer 7: Hidden Systems (evidence-based only) scores["hidden_systems"] = 0.0 # Only with independent evidence return scores def _calculate_overall_rgi( self, physical: GapScore, operational: GapScore, safety: GapScore, institutional: GapScore, economic: GapScore, maintenance: GapScore, human_exp: GapScore ) -> float: """Calculate overall RGI score""" scores = { "physical": physical.score, "operational": operational.score, "safety": safety.score, "institutional": institutional.score, "economic": economic.score, "maintenance": maintenance.score, "human_experience": human_exp.score } # Weighted average overall = sum( scores[dim] * weight for dim, weight in self.dimension_weights.items() ) return min(100.0, overall) def _find_contradictions( self, observations: List[Dict], official_data: Optional[Dict], planned_data: Optional[Dict] ) -> List[Dict]: """Find contradictions between different reality layers""" contradictions = [] # Official vs Observed if official_data: official_condition = official_data.get("average_condition", 2) observed_conditions = [obs.get("overall_condition", 3) for obs in observations] if observed_conditions: avg_observed = sum(observed_conditions) / len(observed_conditions) if abs(official_condition - avg_observed) > 1.5: contradictions.append({ "type": "official_vs_observed", "description": f"Official condition {official_condition} vs observed {avg_observed:.1f}", "severity": "high" if abs(official_condition - avg_observed) > 2 else "medium" }) # Planned vs Constructed if planned_data: planned_count = planned_data.get("planned_buildings", 0) actual_count = len(observations) if planned_count > 0 and actual_count > planned_count * 1.3: contradictions.append({ "type": "planned_vs_constructed", "description": f"Planned {planned_count} buildings but observed {actual_count}", "severity": "high" }) # Wealth vs Implementation luxury_near_poor = False conditions = [obs.get("overall_condition", 3) for obs in observations] if conditions: if max(conditions) - min(conditions) >= 3: luxury_near_poor = True contradictions.append({ "type": "wealth_vs_implementation", "description": "Luxury and poor conditions coexist", "severity": "medium" }) return contradictions def _determine_contradiction_level( self, contradictions: List[Dict], overall_rgi: float ) -> str: """Determine overall contradiction level""" high_count = sum(1 for c in contradictions if c.get("severity") == "high") if high_count >= 2 or overall_rgi > 70: return "high" elif high_count >= 1 or overall_rgi > 40: return "medium" else: return "low" def _calculate_observation_quality(self, observations: List[Dict]) -> float: """Calculate observation quality score""" if not observations: return 0.0 # Factors: count, diversity, recency count_score = min(1.0, len(observations) / 10) # Diversity of object types types = set(obs.get("goid", "").split("-")[0] for obs in observations if obs.get("goid")) diversity_score = min(1.0, len(types) / 3) return (count_score + diversity_score) / 2 def _list_data_sources( self, observations: List[Dict], official_data: Optional[Dict] ) -> List[str]: """List all data sources used""" sources = ["field_observations"] if official_data: sources.append("official_data") # Check for images has_images = any( obs.get("media") for obs in observations ) if has_images: sources.append("image_evidence") return sources # Example analysis def example_bangkok_silom_rgi(): """Example RGI analysis for Bangkok Silom""" analyzer = RealityGapAnalyzer() observations = [ { "goid": "BYG-FAC-WIN-GLA-001", "overall_condition": 4, "findings": [{"code": "2100"}, {"code": "2300"}], "function_status": "operational" }, { "goid": "BYG-FAC-WIN-GLA-002", "overall_condition": 5, "findings": [{"code": "4100"}, {"code": "4200"}], "function_status": "degraded" }, { "goid": "COM-DIS-SGN-001", "overall_condition": 3, "findings": [{"code": "2400"}, {"code": "2100"}], "function_status": "operational" }, { "goid": "ENE-EVC-CHA-001", "overall_condition": 4, "findings": [{"code": "6200"}], "function_status": "degraded" }, { "goid": "BYG-FAC-WIN-GLA-003", "overall_condition": 2, "findings": [{"code": "2200"}], "function_status": "operational" } ] official_data = { "average_condition": 2.5, "institutional_control": "limited" } planned_data = { "planned_buildings": 3 } result = analyzer.analyze("Bangkok Silom", observations, official_data, planned_data) print("=== Reality Gap Index: Bangkok Silom ===") print(f"Overall RGI: {result.overall_rgi:.1f}/100") print(f"Contradiction Level: {result.contradiction_level}") print(f"Observation Quality: {result.observation_quality:.2f}") print(f"Data Sources: {', '.join(result.data_sources)}") print("\nSubscores:") print(f" Physical Reality Gap: {result.physical_gap.score:.1f}") print(f" Operational Reality Gap: {result.operational_gap.score:.1f}") print(f" Safety Reality Gap: {result.safety_gap.score:.1f}") print(f" Institutional Reality Gap: {result.institutional_gap.score:.1f}") print(f" Economic Reality Gap: {result.economic_gap.score:.1f}") print(f" Maintenance Reality Gap: {result.maintenance_gap.score:.1f}") print(f" Human Experience Gap: {result.human_experience_gap.score:.1f}") print("\nReality Layers:") for layer, score in result.layer_scores.items(): print(f" {layer}: {score:.1f}%") print("\nContradictions:") for c in result.contradictions: print(f" [{c['severity'].upper()}] {c['type']}: {c['description']}") return result def example_stockholm_rgi(): """Example RGI analysis for Stockholm""" analyzer = RealityGapAnalyzer() observations = [ { "goid": "BYG-FAC-WIN-GLA-001", "overall_condition": 2, "findings": [{"code": "2100"}], "function_status": "operational" }, { "goid": "TRN-ROD-SUR-001", "overall_condition": 2, "findings": [], "function_status": "operational" }, { "goid": "BEL-STR-LED-001", "overall_condition": 2, "findings": [], "function_status": "operational" } ] official_data = { "average_condition": 2.0, "institutional_control": "full" } planned_data = { "planned_buildings": 3 } result = analyzer.analyze("Stockholm Inner City", observations, official_data, planned_data) print("\n=== Reality Gap Index: Stockholm ===") print(f"Overall RGI: {result.overall_rgi:.1f}/100") print(f"Contradiction Level: {result.contradiction_level}") return result if __name__ == '__main__': example_bangkok_silom_rgi() example_stockholm_rgi()