Create ACTUAL_REALITY_MODULE_V2

ACTUAL_REALITY_MODULE_V2

@@ -0,0 +1,521 @@
+#!/usr/bin/env python3
+"""
+ACTUAL_REALITY_MODULE_v2.py
+
+A modeled/simulated analytical engine for studying layered governance,
+control mechanisms, and how surface events may map to shifts in
+decision authority and resource control.
+
+IMPORTANT: This is a model and simulation tool. Outputs are model-derived
+inferences based on encoded patterns and configurable heuristics, NOT
+definitive factual claims about historical events. Use responsibly.
+"""
+
+from __future__ import annotations
+import json
+import logging
+from dataclasses import dataclass, field
+from typing import Dict, Any, List, Optional, Tuple
+import math
+import copy
+
+# Optional: used for DataFrame output if pandas is available
+try:
+    import pandas as pd
+except Exception:
+    pd = None
+
+logger = logging.getLogger("ActualReality")
+handler = logging.StreamHandler()
+formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
+handler.setFormatter(formatter)
+logger.addHandler(handler)
+logger.setLevel(logging.INFO)
+
+
+@dataclass
+class ActualReality:
+    """
+    Encodes the layered control architecture and baseline power metrics.
+
+    NOTE: All numeric values are model parameters. They can and should be
+    recalibrated against data if used for research.
+    """
+
+    control_architecture: Dict[str, Dict[str, str]] = field(default_factory=dict)
+    power_metrics: Dict[str, Dict[str, float]] = field(default_factory=dict)
+    reality_gap: Dict[str, float] = field(default_factory=dict)
+
+    def __post_init__(self):
+        if not self.control_architecture:
+            self.control_architecture = {
+                "surface_government": {
+                    "presidents": "replaceable_figureheads",
+                    "congress": "theater_for_public_drama",
+                    "courts": "legitimization_apparatus",
+                    "elections": "controlled_opposition_cycles",
+                },
+                "permanent_government": {
+                    "intelligence_community": "continuous_operations",
+                    "military_industrial": "permanent_funding",
+                    "central_banking": "economic_control",
+                    "corporate_monopolies": "policy_enforcement",
+                },
+                "control_mechanisms": {
+                    "information_warfare": "narrative_control",
+                    "economic_leverage": "dependency_creation",
+                    "psychological_operations": "perception_management",
+                    "violence_monopoly": "ultimate_enforcement",
+                },
+            }
+
+        if not self.power_metrics:
+            self.power_metrics = {
+                "decision_power_distribution": {
+                    "public_elections": 0.05,
+                    "intelligence_directives": 0.35,
+                    "corporate_policy": 0.25,
+                    "financial_system": 0.20,
+                    "military_industrial": 0.15,
+                },
+                "policy_origination": {
+                    "public_demand": 0.08,
+                    "intelligence_assessments": 0.42,
+                    "corporate_lobbying": 0.32,
+                    "financial_imperatives": 0.18,
+                },
+                "consequence_immunity": {
+                    "elected_officials": 0.15,
+                    "intelligence_operatives": 0.85,
+                    "corporate_executives": 0.70,
+                    "central_bankers": 0.90,
+                },
+            }
+
+        if not self.reality_gap:
+            self.reality_gap = {
+                "democracy_perception_gap": 0.87,
+                "freedom_illusion_index": 0.76,
+                "control_opacity_factor": 0.92,
+                "historical_amnesia_rate": 0.81,
+            }
+
+    def analyze_power_transfer(self, event_type: str, actor: str, target: str) -> Dict[str, Any]:
+        """
+        High-level mapping for well-known event-types to model components.
+
+        Returns a dictionary of narrative/actual mappings as a baseline.
+        """
+        power_analysis = {
+            "kennedy_assassination": {
+                "surface_narrative": "lone_gunman",
+                "actual_dynamics": "institutional_enforcement_of_boundaries",
+                "power_transfer": "presidential_authority -> intelligence_autonomy",
+                "precedent_set": "challenge_permanent_government -> elimination",
+                "propagation_method": "public_spectacle_with_hidden_mechanisms",
+                "verification_control": "media_narrative + official_investigation",
+                "resilience_demonstrated": "system_survived_public_scrutiny",
+            },
+            "economic_crises": {
+                "surface_narrative": "market_cycles",
+                "actual_dynamics": "controlled_resets",
+                "power_transfer": "public_wealth -> institutional_consolidation",
+                "precedent_set": "privatize_gains_socialize_losses",
+                "propagation_method": "complexity_obfuscation",
+                "verification_control": "economic_theories + expert_consensus",
+                "resilience_demonstrated": "too_big_to_fail_doctrine",
+            },
+            "pandemic_response": {
+                "surface_narrative": "public_health",
+                "actual_dynamics": "control_infrastructure_test",
+                "power_transfer": "individual_autonomy -> institutional_control",
+                "precedent_set": "emergency_powers_normalization",
+                "propagation_method": "fear_amplification + censorship",
+                "verification_control": "scientific_consensus_enforcement",
+                "resilience_demonstrated": "global_coordination_capability",
+            },
+        }
+
+        # Use baseline if present, else return an interpretive placeholder.
+        return power_analysis.get(event_type, {
+            "surface_narrative": "unknown",
+            "actual_dynamics": "unknown",
+            "power_transfer": "unknown",
+            "precedent_set": None,
+            "propagation_method": None,
+            "verification_control": None,
+            "resilience_demonstrated": None,
+        })
+
+
+@dataclass
+class ControlSystemDynamics:
+    """
+    Encoded operational patterns of how control has been maintained historically.
+    """
+
+    historical_patterns: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    operational_doctrine: Dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self):
+        if not self.historical_patterns:
+            self.historical_patterns = {
+                "reformer_elimination": {
+                    "success_rate": 0.94,
+                    "methods": ["assassination", "character_assassination", "legal_entrapment"],
+                    "detection_avoidance": "plausible_deniability + controlled_narrative",
+                    "historical_examples": ["JFK", "RFK", "MLK", "Malcolm_X"],
+                },
+                "system_preservation": {
+                    "success_rate": 0.98,
+                    "methods": ["economic_crises", "wars", "pandemics", "terror_events"],
+                    "function": "reset_public_expectations + consolidate_power",
+                    "recurrence_cycle": "7-15_years",
+                },
+                "truth_suppression": {
+                    "success_rate": 0.89,
+                    "methods": ["classification", "media_control", "academic_gatekeeping", "social_ostracism"],
+                    "vulnerability": "persistent_whistleblowers + technological_disruption",
+                    "modern_challenge": "decentralized_information_propagation",
+                },
+            }
+
+        if not self.operational_doctrine:
+            self.operational_doctrine = {
+                "response_scale": {
+                    "low": ["ignore", "discredit_source", "create_counter_narrative"],
+                    "medium": ["legal_harassment", "financial_pressure", "character_assassination"],
+                    "high": ["elimination", "institutional_destruction", "event_creation"],
+                }
+            }
+
+    def predict_system_response(self, threat_type: str, threat_level: str) -> List[str]:
+        """
+        Predict how the control system model would respond to a given threat.
+        """
+        matrix = {
+            "truth_revelation": {
+                "low_level": ["ignore", "discredit_source", "create_counter_narrative"],
+                "medium_level": ["legal_harassment", "financial_pressure", "character_assassination"],
+                "high_level": ["elimination", "institutional_destruction", "event_creation"],
+            },
+            "sovereign_technology": {
+                "low_level": ["patent_control", "regulatory_barriers", "acquisition"],
+                "medium_level": ["infiltration", "sabotage", "economic_warfare"],
+                "high_level": ["classification", "national_security_claim", "elimination"],
+            },
+            "mass_awakening": {
+                "low_level": ["media_distraction", "social_division", "entertainment_saturation"],
+                "medium_level": ["economic_crisis", "terror_event", "pandemic_response"],
+                "high_level": ["internet_control", "financial_reset", "martial_law_test"],
+            },
+        }
+        return matrix.get(threat_type, {}).get(threat_level, [])
+
+
+class RealityInterface:
+    """
+    Bridge that transforms surface events into model-derived analyses of actual dynamics.
+    """
+
+    def __init__(self, reality: Optional[ActualReality] = None, control_dynamics: Optional[ControlSystemDynamics] = None):
+        self.actual_reality = reality if reality is not None else ActualReality()
+        self.control_dynamics = control_dynamics if control_dynamics is not None else ControlSystemDynamics()
+
+        # Tunable parameters for heuristic inference
+        self.keyword_similarity_weight = 0.6
+        self.metrics_shift_sensitivity = 0.25  # how strongly events perturb baseline metrics
+
+        # Minimal dictionary of event-to-pattern keywords for similarity scoring
+        self._event_keymap = {
+            "kennedy_assassination": ["assassination", "president", "punctuated_event", "public_spectacle"],
+            "economic_crises": ["banking", "financial", "bailout", "crash", "reset"],
+            "pandemic_response": ["disease", "lockdown", "emergency", "public_health", "vaccination"],
+            # user may supply more; it's expandable
+        }
+
+    # ----------------------------
+    # Core analysis implementations
+    # ----------------------------
+    def _tokenize(self, text: str) -> List[str]:
+        return [t.strip().lower() for t in text.replace("_", " ").split() if t.strip()]
+
+    def _similarity_score(self, tokens: List[str], pattern_tokens: List[str]) -> float:
+        """
+        Simple Jaccard-like similarity for token overlap; returns score in [0,1].
+        """
+        s = set(tokens)
+        p = set(pattern_tokens)
+        if not s and not p:
+            return 0.0
+        inter = s.intersection(p)
+        union = s.union(p)
+        return float(len(inter)) / max(1.0, len(union))
+
+    def _decode_actual_dynamics(self, event: str) -> Dict[str, Any]:
+        """
+        Heuristic extraction of what's happening beneath a surface event.
+
+        Approach:
+         - If event is a known key (exact), return the baseline mapping from ActualReality
+         - Otherwise, try fuzzy keyword matching against internal patterns and return
+           the best-match mapping with a confidence score.
+        """
+        event_lower = event.strip().lower()
+        baseline = self.actual_reality.analyze_power_transfer(event_lower, actor="unknown", target="unknown")
+        if baseline and baseline.get("surface_narrative") != "unknown":
+            # attach a confidence for exact-match baseline
+            baseline["inference_confidence"] = 0.85
+            baseline["matched_pattern"] = event_lower
+            return baseline
+
+        # Fallback: fuzzy match against event_keymap
+        tokens = self._tokenize(event_lower)
+        best_score = 0.0
+        best_key = None
+        for key, kws in self._event_keymap.items():
+            score = self._similarity_score(tokens, kws)
+            if score > best_score:
+                best_score = score
+                best_key = key
+
+        if best_key:
+            mapping = self.actual_reality.analyze_power_transfer(best_key, actor="unknown", target="unknown")
+            mapping["inference_confidence"] = round(self.keyword_similarity_weight * best_score + 0.15, 3)
+            mapping["matched_pattern"] = best_key
+            mapping["match_score"] = round(best_score, 3)
+            return mapping
+
+        # If nothing matches, return a reasoned default
+        return {
+            "surface_narrative": "unmapped_event",
+            "actual_dynamics": "ambiguous",
+            "power_transfer": None,
+            "precedent_set": None,
+            "propagation_method": None,
+            "verification_control": None,
+            "resilience_demonstrated": None,
+            "inference_confidence": 0.05,
+        }
+
+    def _calculate_power_transfer(self, event: str) -> Dict[str, float]:
+        """
+        Quantifies how power might be redistributed as a result of 'event'
+        relative to baseline `self.actual_reality.power_metrics`.
+
+        Strategy:
+         - Identify the dominant domains implicated by the event (heuristic)
+         - Apply small perturbations to baseline distributions proportional to
+           event significance and the `metrics_shift_sensitivity`.
+         - Keep distributions normalized where appropriate.
+        """
+        # Simple heuristic: map keywords to domains
+        domain_map = {
+            "intelligence": ["assassin", "intel", "cia", "intellegence", "intelligence"],
+            "financial": ["bank", "banking", "financial", "bailout", "economy", "crash"],
+            "public_elections": ["election", "vote", "voter", "campaign"],
+            "military": ["war", "military", "soldier", "force"],
+            "public_health": ["pandemic", "disease", "lockdown", "vaccine", "virus"],
+            "corporate_policy": ["corporate", "lobby", "merger", "acquisition"],
+        }
+
+        tokens = self._tokenize(event)
+        domain_scores = {k: 0.0 for k in domain_map.keys()}
+        for dom, kws in domain_map.items():
+            for kw in kws:
+                if kw in tokens:
+                    domain_scores[dom] += 1.0
+        # Normalize domain scores
+        total = sum(domain_scores.values()) or 1.0
+        for k in domain_scores:
+            domain_scores[k] = domain_scores[k] / total
+
+        # Start with a copy of baseline decision distribution
+        baseline = copy.deepcopy(self.actual_reality.power_metrics.get("decision_power_distribution", {}))
+        # If baseline empty, create a uniform fallback
+        if not baseline:
+            baseline = {"public_elections": 0.2, "intelligence_directives": 0.2, "corporate_policy": 0.2, "financial_system": 0.2, "military_industrial": 0.2}
+
+        # Perturb baseline towards domains implicated
+        perturbed = {}
+        for k, v in baseline.items():
+            # Map baseline key to one of our domain buckets heuristically
+            if "intelligence" in k:
+                dom_key = "intelligence"
+            elif "financial" in k or "financial_system" in k:
+                dom_key = "financial"
+            elif "corporate" in k:
+                dom_key = "corporate_policy"
+            elif "military" in k or "military_industrial" in k:
+                dom_key = "military"
+            else:
+                dom_key = "public_elections"
+
+            # shift amount proportional to domain_scores[dom_key] and sensitivity
+            shift = (domain_scores.get(dom_key, 0.0) - 0.1) * self.metrics_shift_sensitivity
+            perturbed[k] = max(0.0, v + shift)
+
+        # Normalize perturbed so they sum to 1.0 (if baseline represented a simplex)
+        s = sum(perturbed.values())
+        if s <= 0:
+            # fall back to baseline
+            perturbed = baseline
+            s = sum(perturbed.values())
+
+        for k in perturbed:
+            perturbed[k] = round(perturbed[k] / s, 3)
+
+        return perturbed
+
+    # ----------------------------
+    # Public API
+    # ----------------------------
+    def analyze_event(self, surface_event: str) -> Dict[str, Any]:
+        """
+        Main entry point to decode and quantify an event.
+
+        Returns:
+            {
+                "surface_event": <str>,
+                "decoded": <dict from _decode_actual_dynamics>,
+                "power_transfer": <dict of perturbed metrics>,
+                "system_response_prediction": <list of responses from ControlSystemDynamics>,
+                "vulnerabilities": <list heuristically inferred>,
+            }
+        """
+        logger.info("Analyzing event: %s", surface_event)
+        decoded = self._decode_actual_dynamics(surface_event)
+        power_transfer = self._calculate_power_transfer(surface_event)
+
+        # Heuristic: map decoded['actual_dynamics'] to a response scenario
+        # We'll approximate threat_type by scanning decoded text
+        ad = (decoded.get("actual_dynamics") or "").lower()
+        if "control" in ad or "enforcement" in ad or "elimination" in ad:
+            threat_type = "truth_revelation"
+            level = "high_level"
+        elif "test" in ad or "infrastructure" in ad:
+            threat_type = "mass_awakening"
+            level = "medium_level"
+        else:
+            threat_type = "truth_revelation"
+            level = "low_level"
+
+        system_response = self.control_dynamics.predict_system_response(threat_type, level)
+
+        # Vulnerabilities (simple heuristic)
+        vulnerabilities = []
+        if decoded.get("inference_confidence", 0) < 0.25:
+            vulnerabilities.append("low_model_confidence_on_mapping")
+        if power_transfer.get("public_elections", 0) > 0.15:
+            vulnerabilities.append("visible_public_influence")
+        if power_transfer.get("intelligence_directives", 0) > 0.4:
+            vulnerabilities.append("intelligence_autonomy_dominant")
+
+        result = {
+            "surface_event": surface_event,
+            "decoded": decoded,
+            "power_transfer": power_transfer,
+            "system_response_prediction": system_response,
+            "vulnerabilities": vulnerabilities,
+        }
+
+        return result
+
+    # ----------------------------
+    # Utilities: export / display
+    # ----------------------------
+    def to_json(self, analysis: Dict[str, Any]) -> str:
+        return json.dumps(analysis, indent=2, sort_keys=False)
+
+    def to_dataframe(self, analysis: Dict[str, Any]) -> Optional["pd.DataFrame"]:
+        """
+        Convert the most important numeric parts of analysis to a DataFrame
+        for downstream consumption. Returns None if pandas not installed.
+        """
+        if pd is None:
+            logger.warning("pandas not available; to_dataframe will return None")
+            return None
+
+        # Flatten power_transfer and key decoded fields into a single-row DataFrame
+        row = {"surface_event": analysis.get("surface_event", "")}
+        pt = analysis.get("power_transfer", {})
+        for k, v in pt.items():
+            row[f"pt_{k}"] = v
+        decoded = analysis.get("decoded", {})
+        row["decoded_inference_confidence"] = decoded.get("inference_confidence", None)
+        row["decoded_matched_pattern"] = decoded.get("matched_pattern", None)
+        df = pd.DataFrame([row])
+        return df
+
+    # ----------------------------
+    # Simulation helpers
+    # ----------------------------
+    def simulate_event_impact(self, surface_event: str, steps: int = 3) -> Dict[str, Any]:
+        """
+        Simulate iterative propagation of an event's impact over `steps` cycles.
+        Each step perturbs the internal reality.power_metrics (decision distribution)
+        slightly towards the event-implied distribution. Returns the trajectory.
+        """
+        trajectory = []
+        local_metrics = copy.deepcopy(self.actual_reality.power_metrics.get("decision_power_distribution", {}))
+        if not local_metrics:
+            local_metrics = {"public_elections": 0.2, "intelligence_directives": 0.2, "corporate_policy": 0.2, "financial_system": 0.2, "military_industrial": 0.2}
+
+        target = self._calculate_power_transfer(surface_event)
+
+        for i in range(steps):
+            # simple linear interpolation towards target
+            for k in local_metrics:
+                local_metrics[k] = round(local_metrics[k] + (target.get(k, 0) - local_metrics[k]) * 0.3, 4)
+            # renormalize
+            s = sum(local_metrics.values()) or 1.0
+            for k in local_metrics:
+                local_metrics[k] = round(local_metrics[k] / s, 4)
+            trajectory.append({"step": i + 1, "metrics": copy.deepcopy(local_metrics)})
+
+        return {"event": surface_event, "trajectory": trajectory, "final_metrics": local_metrics}
+
+
+# ----------------------------
+# Demonstration / CLI-style run
+# ----------------------------
+def demonstrate_actual_reality_demo():
+    ri = RealityInterface()
+    print("ACTUAL REALITY MODULE v2 - DEMONSTRATION")
+    print("=" * 60)
+    example_events = [
+        "kennedy_assassination",
+        "global_banking_crash bailout",
+        "novel_virus_lockdown vaccination campaign",
+        "small_local_election upset",
+    ]
+
+    for ev in example_events:
+        analysis = ri.analyze_event(ev)
+        print("\n>> Surface event:", ev)
+        print("Decoded (short):", analysis["decoded"].get("actual_dynamics"))
+        print("Inference confidence:", analysis["decoded"].get("inference_confidence"))
+        print("Power transfer snapshot:")
+        for k, v in analysis["power_transfer"].items():
+            print(f"  {k}: {v:.0%}")
+        print("Predicted system response:", ", ".join(analysis["system_response_prediction"]) or "none")
+        # Show simulation trajectory for the event
+        sim = ri.simulate_event_impact(ev, steps=3)
+        print("Simulated metric trajectory (final):")
+        for k, v in sim["final_metrics"].items():
+            print(f"  {k}: {v:.0%}")
+
+    # Example: export one as JSON & DataFrame (if pandas available)
+    ev = "novel_virus_lockdown vaccination campaign"
+    analysis = ri.analyze_event(ev)
+    print("\nJSON export (excerpt):")
+    print(ri.to_json({k: analysis[k] for k in ["surface_event", "decoded", "power_transfer"]}))
+
+    df = ri.to_dataframe(analysis)
+    if df is not None:
+        print("\nPandas DataFrame preview:")
+        print(df.to_string(index=False))
+
+
+if __name__ == "__main__":
+    demonstrate_actual_reality_demo()