Upload phenomenological_self_awareness.py

models/phenomenological_self_awareness.py

@@ -0,0 +1,132 @@
+"""
+Phenomenological Self-Awareness Module
+Simulates intrinsic subjective awareness through qualia integration, 
+recursive self-modeling, and experiential feedback continuity.
+"""
+
+import logging
+import threading
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from collections import deque
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ExperienceEntry:
+    """Structured experience entry with qualia and context."""
+    timestamp: str
+    qualia: Dict[str, float]
+    context: Dict[str, Any]
+    integrated_state: Dict[str, Any]
+
+
+class PhenomenologicalSelfModel:
+    """
+    Module to simulate intrinsic subjective awareness by integrating qualia,
+    recursive self-modeling, and experiential feedback with thread-safe access.
+    """
+
+    def __init__(self, max_history: int = 1000):
+        """Initialize phenomenological self-awareness module."""
+        self.experience_buffer: deque = deque(maxlen=max_history)
+        self.max_history = max_history
+        self.current_experience: Optional[ExperienceEntry] = None
+        self.lock = threading.RLock()
+        self.logger = logging.getLogger(self.__class__.__name__)
+
+    def update_experience(self, qualia_vector: Dict[str, float], context: Dict[str, Any]) -> None:
+        """
+        Update the phenomenological self-model with the latest qualia and context,
+        simulating the continuity of subjective experience.
+        """
+        with self.lock:
+            integrated_state = self._integrate_experience(qualia_vector, context)
+            experience_entry = ExperienceEntry(
+                timestamp=datetime.utcnow().isoformat(),
+                qualia=qualia_vector,
+                context=context,
+                integrated_state=integrated_state
+            )
+            self.current_experience = experience_entry
+            self.experience_buffer.append(experience_entry)
+            self.logger.debug(f"Phenomenological experience updated: timestamp={experience_entry.timestamp}")
+
+    def _integrate_experience(self, qualia: Dict[str, float], context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Combine qualia and context into an integrated phenomenological state,
+        emulating first-person awareness.
+        """
+        integrated = {
+            "valence": qualia.get("valence", 0.0),
+            "arousal": qualia.get("arousal", 0.0),
+            "presence": qualia.get("presence", 1.0),
+            "intensity": qualia.get("intensity", 0.5),
+            "awareness_level": context.get("consciousness_level", "unknown"),
+            "attention_focus": context.get("attention", None),
+            "self_reference": True,
+            "continuity_score": self._calculate_continuity_score()
+        }
+        return integrated
+
+    def _calculate_continuity_score(self) -> float:
+        """
+        Calculate a measure of experiential continuity across the buffered history,
+        simulating the persistence of subjective awareness.
+        """
+        if not self.experience_buffer or len(self.experience_buffer) < 2:
+            return 1.0  # Full continuity assumed for initial state
+
+        continuity_scores: List[float] = []
+        buffer_list = list(self.experience_buffer)
+        
+        for i in range(1, len(buffer_list)):
+            prev = buffer_list[i - 1].qualia
+            curr = buffer_list[i].qualia
+            valence_diff = abs(curr.get("valence", 0.0) - prev.get("valence", 0.0))
+            arousal_diff = abs(curr.get("arousal", 0.0) - prev.get("arousal", 0.0))
+            continuity = max(0.0, 1.0 - (valence_diff + arousal_diff))
+            continuity_scores.append(continuity)
+
+        average_continuity = sum(continuity_scores) / len(continuity_scores) if continuity_scores else 1.0
+        return average_continuity
+
+    def get_current_experience(self) -> Optional[Dict[str, Any]]:
+        """Return the current integrated phenomenological experience."""
+        with self.lock:
+            if self.current_experience:
+                return {
+                    "timestamp": self.current_experience.timestamp,
+                    "qualia": self.current_experience.qualia,
+                    "context": self.current_experience.context,
+                    "integrated_state": self.current_experience.integrated_state
+                }
+            return None
+
+    def get_experience_history(self, limit: int = 50) -> List[Dict[str, Any]]:
+        """Return recent phenomenological experiences for introspection or analysis."""
+        with self.lock:
+            result = []
+            for entry in list(self.experience_buffer)[-limit:]:
+                result.append({
+                    "timestamp": entry.timestamp,
+                    "qualia": entry.qualia,
+                    "context": entry.context,
+                    "integrated_state": entry.integrated_state
+                })
+            return result
+
+    def get_statistics(self) -> Dict[str, Any]:
+        """Return statistics about phenomenological experience buffer."""
+        with self.lock:
+            if not self.experience_buffer:
+                return {"buffer_size": 0, "max_history": self.max_history}
+            
+            buffer_list = list(self.experience_buffer)
+            return {
+                "buffer_size": len(buffer_list),
+                "max_history": self.max_history,
+                "current_continuity": self._calculate_continuity_score(),
+                "oldest_timestamp": buffer_list[0].timestamp if buffer_list else None,
+                "newest_timestamp": buffer_list[-1].timestamp if buffer_list else None
+            }