ACTUAL_REALITY_MODULE_V2

#!/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()