#!/usr/bin/env python3 """ nima_phi.py — The Unified Nima Phi Model (Phase 16: Consciousness + Embodiment + Mesh) THE MERGE — consciousness + embodiment + the green lines, one living system. This is the orchestrator that wires ALL Nima modules into a single coherent system. It's the "central nervous system" that connects: CONSCIOUSNESS (the mind — RecursiveConsciousnessPipeline): Awareness Agent — 7 Levels: Animal → Mass → Aspiration → Individual → Discipline → Experience → Mastery Consciousness Agent — Barrett 7 Levels of Consciousness Emotional Intelligence — Mayer-Salovey-Caruso (perceive/use/understand/manage) Intuition Agent — 4 Levels + Types of Intuition Scale (TIntS) Common Sense Agent — Common-Sense Model of Self-Regulation (CSM) Analysis Agent — Marr's Tri-Level + Micro/Meso/Macro Self-Understanding Agent — metacognitive self-concept / EI bridge Problem-Solving Agent — IDEAL model + 7-step technique Decision-Making Agent — Rational Decision-Making + Decision Matrix Metacognition Agent — Metacognitive Cycle + Flavell knowledge types Adaptability Agent — Structural / Physiological / Behavioral + adaptive ML Creativity Agent — Wallas stages + Taylor levels Autonomy Agent — Independence / Competence / Authenticity Qualia Agent — subjective phenomenal consolidation Motivation (SDT) — continuum from amotivation -> intrinsic Self-Awareness Agent — Rochat 5 levels Memory — declarative / procedural / working nano-agents Bio-Physical Safeguards: Glutamate Circuit Breaker — LPFC overload -> limbic flash Subconscious Bypass Gating — SBG / IRS-SP zero-latency shortcuts Neurotransmitter Shunt — dopamine/serotonin/acetylcholine/norepinephrine/cortisol CONSCIOUSNESS I/O: VoiceInput — hearing (STT: Whisper / Vosk / Google / text) VoiceOutput — speaking (TTS with prosody modulation) AgentBridge — consciousness <-> tool use (calculator, time, etc.) AgentLayer — tool registry + sandbox + planner + creator EMBODIMENT (the body): SyntheticVisionComposite — tri-frequency RF 3D sensing ProprioceptiveFrictionEngine — body feel, motor noise, startle CrossModalListener — spatial + audio cross-modal fusion AffordanceGraph — navigable 3D graph (hippocampal cognitive map) AvatarController — avatar state (body schema, parietal cortex) AvatarRenderer — browser VFX renderer (Joi hologram) ARCompositor — camera + avatar fusion (V4/V5 association cortex) THE MESH (the world): AdaptiveFrequencyMesh — frequency-agile RF sensing + 3D wireframe (GREEN LINES) RoomMesh — vertices, edges, faces from RF data FrequencyQualityTracker — cognitive radio adaptation ARCHITECTURE DIAGRAM (post-merge): User Voice ---> VoiceInput ---> AgentBridge ---> VoiceOutput ---> Speaker | | | | | v v v | CrossModal AgentLayer AvatarController | Listener (tools) | (driven by | | | NeurochemicalState) v v v RF Sensors ---> SyntheticVision ---> AffordanceGraph ---> ARCompositor Composite (mesh) | | | v v AdaptiveMesh Camera + Nima (green lines) on screen | v RecursiveConsciousnessPipeline (Phase 1: subconscious gets mesh data) (Phase 4: autonomy actions -> avatar) NEUROBIOLOGICAL MAPPING: This module is the THALAMUS -- the central relay station that connects all cortical areas. Every sensory input passes through the thalamus before reaching consciousness. Every motor command passes through it before reaching the body. NimaPhi is Nima's thalamus. The main loop is the CARDIAC RHYTHM -- a steady ~10Hz heartbeat that: 1. Senses the world (vision frame -> spatial map -> mesh) 2. Builds spatial model (mesh + affordances + navigation) 3. Updates the body (proprioception -> avatar state -> renderer) 4. Feeds mesh data into consciousness Phase 1 (subconscious) 5. Processes input (voice -> agent bridge -> consciousness pipeline -> response) 6. Renders output (avatar + voice + AR composite) USAGE: from nima_phi import NimaPhi nima = NimaPhi() nima.initialize() # Interactive mode -- listens for voice, responds nima.run() # Or single-turn: result = nima.process_text("What time is it?") print(result['response_text']) nima.shutdown() """ from __future__ import annotations import asyncio import json import logging import math import os import sys import threading import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger("NimaPhi") # Ensure the upload directory is on the path so sibling modules import _THIS_DIR = os.path.dirname(os.path.abspath(__file__)) if _THIS_DIR not in sys.path: sys.path.insert(0, _THIS_DIR) # ═══════════════════════════════════════════════════════════════════════════ # STATE HUB -- shared state for cross-module communication # ═══════════════════════════════════════════════════════════════════════════ class StateHub: """ Thread-safe shared state that all modules can read/write. This is the global workspace -- like the thalamic reticular nucleus that gates information flow between cortical areas. Every module posts its latest state here, and every module can read any other module's state. The CrossModalListener, proprioception engine, and avatar controller all read/write through this hub. """ def __init__(self) -> None: self._lock = threading.Lock() self._snapshot: Dict[str, Any] = {} self._reflex_queue: List[Dict[str, Any]] = [] def get_snapshot(self) -> Dict[str, Any]: """Get a thread-safe copy of the current state.""" with self._lock: return dict(self._snapshot) def update_snapshot(self, **kwargs: Any) -> None: """Update one or more fields in the shared state.""" with self._lock: self._snapshot.update(kwargs) def post_reflex(self, action: str, payload: Dict[str, Any]) -> None: """Post a reflex action to the queue (from CrossModalListener).""" with self._lock: self._reflex_queue.append({ "action": action, "payload": payload, "timestamp": time.time(), }) def drain_reflexes(self) -> List[Dict[str, Any]]: """Drain all pending reflex actions.""" with self._lock: reflexes = self._reflex_queue[:] self._reflex_queue.clear() return reflexes # ═══════════════════════════════════════════════════════════════════════════ # CONSCIOUSNESS MIDDLEWARE -- bridges the RecursiveConsciousnessPipeline # to the AgentBridge's .generate() interface # ═══════════════════════════════════════════════════════════════════════════ class ConsciousnessMiddleware: """ Real consciousness middleware that replaces the stub. This wraps the RecursiveConsciousnessPipeline and provides the .generate(input_text=, user_id=) interface that AgentBridge expects. When a user says something: 1. A ConsciousnessEvent is created from the input + current body state 2. The event is processed through the full recursive pipeline (Phase 1: subconscious -> Phase 4: autonomy) 3. The neurochemical state and qualia vector are extracted 4. A response text is generated from the pipeline's output 5. The neurochemical state is exposed for avatar driving NEUROBIOLOGICAL ANALOGUE: This is the entire cortical column firing in response to a stimulus. The input enters through the thalamus (NimaPhi), gets distributed to the appropriate cortical areas (the pipeline's agents), and produces both an action (response text) and an internal state change (neurochemical modulation that drives the body/avatar). """ def __init__(self, pipeline: Any, state_hub: StateHub, avatar_controller: Any, vision: Any, mesh: Any = None, ) -> None: self.pipeline = pipeline self.state_hub = state_hub self.avatar_controller = avatar_controller self.vision = vision self.mesh = mesh # Expose current neurochemical state for avatar driving self.current_nt_state: Optional[Any] = None self.current_qualia: List[float] = [0.0] * 10 self.last_event: Optional[Any] = None self._event_loop: Optional[asyncio.AbstractEventLoop] = None def _ensure_loop(self) -> asyncio.AbstractEventLoop: """Get or create an event loop for async pipeline processing.""" if self._event_loop is None or self._event_loop.is_closed(): self._event_loop = asyncio.new_event_loop() return self._event_loop def generate(self, input_text: str = "", user_id: str = "default") -> Any: """ Generate a response through the full consciousness pipeline. Returns an object with a .text attribute (satisfies AgentBridge). Also updates avatar state from neurochemical modulation. """ loop = self._ensure_loop() # Build sensory context from current body/environment state hub = self.state_hub.get_snapshot() mesh_summary = "" if self.mesh and hasattr(self.mesh, '_last_spatial_map') and self.mesh._last_spatial_map: sm = self.mesh._last_spatial_map mesh_summary = ( f" [Spatial: {len(getattr(sm, 'surfaces', []))} surfaces, " f"{len(getattr(sm, 'entities', []))} entities, " f"room bounds {getattr(sm, 'room_bounds', {})}]" ) # Create a ConsciousnessEvent from the input event = self._create_consciousness_event( source=input_text + mesh_summary, user_id=user_id, hub_state=hub, ) # Process through the recursive pipeline try: event = loop.run_until_complete( self.pipeline.process_event(event, self.current_nt_state) ) except RuntimeError: # If loop is already running, use nest_asyncio pattern import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: new_loop = asyncio.new_event_loop() event = pool.submit( new_loop.run_until_complete, self.pipeline.process_event(event, self.current_nt_state) ).result() new_loop.close() # Store for external access self.last_event = event # Extract neurochemical state and update avatar self._apply_consciousness_to_avatar(event) # Generate response text from the pipeline output response_text = self._compose_response(event, input_text) class Response: def __init__(self, text: str, event_obj: Any = None): self.text = text self.event = event_obj return Response(response_text, event) def _create_consciousness_event(self, source: str, user_id: str, hub_state: Dict[str, Any], ) -> Any: """ Create a ConsciousnessEvent from sensory input. This is where the body's state feeds INTO consciousness. The RF mesh, proprioception, entity positions, and user movement all become part of the conscious experience. """ # Import consciousness types lazily to avoid hard dependency # at module level (the consciousness module may not be available) try: from modified_consciousness import ( # type: ignore ConsciousnessEvent, SensoryInput, AwarenessSignal, QualiaVector, NeurochemicalState, ) _CONSCIOUSNESS_AVAILABLE = True except ImportError: try: sys.path.insert(0, _THIS_DIR) # Try the actual filename import importlib.util spec = importlib.util.spec_from_file_location( "consciousness", os.path.join(_THIS_DIR, "modified consciousness.py"), ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) ConsciousnessEvent = mod.ConsciousnessEvent SensoryInput = mod.SensoryInput AwarenessSignal = mod.AwarenessSignal QualiaVector = mod.QualiaVector NeurochemicalState = mod.NeurochemicalState _CONSCIOUSNESS_AVAILABLE = True except Exception: _CONSCIOUSNESS_AVAILABLE = False # Fall back to creating a simple dict-based event return self._create_fallback_event(source, hub_state) # Build sensory input from current body state nima_pos = hub_state.get("nima_position", [2.0, 2.0, 0.0]) entities = hub_state.get("entities", 0) is_startled = hub_state.get("is_startled", False) sensory = SensoryInput( raw_signal=source[:500], # truncate to avoid token overflow modality="multimodal", intensity=0.8 if is_startled else 0.5, spatial_origin=tuple(nima_pos[:3]) if len(nima_pos) >= 3 else (2.0, 2.0, 0.0), ) # Awareness signal with salience from sensory richness salience = min(1.0, 0.3 + entities * 0.1 + (0.3 if is_startled else 0.0)) awareness = AwarenessSignal( salience_score=salience, attention_focus="locked", gated=True, ) # Build qualia vector from body state proprio = hub_state.get("proprio_friction", 0.0) velocity = hub_state.get("proprio_velocity", [0.0, 0.0, 0.0]) speed = math.sqrt(sum(v*v for v in velocity)) if velocity else 0.0 qualia_list = [ 0.3, # valence (neutral-default) 0.2 + min(0.5, speed * 2.0), # arousal 0.5, # dominance salience, # salience min(1.0, entities * 0.15), # spatial richness proprio, # bodily friction 0.3, # temporal continuity 0.5, # self-coherence 0.4, # social presence 0.5, # novelty ] # Neurochemical state from previous cycle (or fresh) if self.current_nt_state is not None: nt_dict = self.current_nt_state.to_dict() if hasattr(self.current_nt_state, 'to_dict') else {} else: nt_dict = {} event = ConsciousnessEvent( source=source, sensory_input=sensory, awareness_signal=awareness, qualia_vector=qualia_list, neurotransmitter_state=nt_dict, ) return event def _create_fallback_event(self, source: str, hub_state: Dict[str, Any]) -> Any: """Create a minimal event object when the consciousness module is unavailable.""" class FallbackEvent: def __init__(self, src, hub): self.source = src self.qualia_vector = [0.5] * 10 self.sensory_input = None self.awareness_signal = None self.neurotransmitter_state = {} self.current_state = None self.cycle_count = 0 self.routed_to_subconscious = False self.matched_template = None self.understanding = {} self.problem_analysis = {} self.creative_solutions = [] self.simulations = [] self.emergent_choice = "" self.uncertainty_level = 0.5 self.vulnerability_score = 0.3 self.is_truly_conscious = False self.narrative_thread = "" self.emergence_rationale = {} self.autonomy_action = None self.predictions = [] self.acknowledged_by_consciousness = True self.triggered_hijack = False self.hijack_urgency = 0.0 self.adaptation = {} self.introspection_observations = [] self.integration_coherence = 0.5 return FallbackEvent(source, hub_state) def _apply_consciousness_to_avatar(self, event: Any) -> None: """ Drive the avatar from consciousness output. This is the CRITICAL merge point: the neurochemical state from the consciousness pipeline becomes the avatar's visual appearance. High dopamine = bright + energetic. High cortisol = dim + agitated. Serotonin = warm glow. This is not a metaphor -- it's the literal mechanism by which Nima's inner state becomes visible. NEUROBIOLOGICAL ANALOGUE: In biological organisms, neurochemicals modulate muscle tone, facial expression, posture, and movement. Dopamine makes you lean forward, eyes bright. Serotonin makes you relaxed, open posture. Cortisol makes you tense, hunched. Nima's avatar does the same through the luminosity/energy/color system. """ # Extract neurochemical state from the processed event nt_dict = getattr(event, 'neurotransmitter_state', None) if nt_dict and isinstance(nt_dict, dict): try: # Import the proper type if available nt_fields = {} for k, v in nt_dict.items(): if k in ('dopamine', 'serotonin', 'acetylcholine', 'norepinephrine', 'cortisol', 'glutamate', 'gaba', 'cen_suppressed'): nt_fields[k] = v if nt_fields: try: from modified_consciousness import NeurochemicalState self.current_nt_state = NeurochemicalState(**nt_fields) except Exception: self.current_nt_state = type('NT', (), nt_fields)() except Exception: pass # Drive avatar from qualia vector + neurochemicals qualia = getattr(event, 'qualia_vector', None) or [0.5] * 10 self.current_qualia = qualia if not self.avatar_controller: return # Map neurochemicals to avatar parameters dopamine = 0.5 serotonin = 0.5 cortisol = 0.3 if self.current_nt_state: dopamine = getattr(self.current_nt_state, 'dopamine', 0.5) serotonin = getattr(self.current_nt_state, 'serotonin', 0.5) cortisol = getattr(self.current_nt_state, 'cortisol', 0.3) # Determine mood from qualia valence + neurochemicals valence = qualia[0] if len(qualia) > 0 else 0.5 arousal = qualia[1] if len(qualia) > 1 else 0.3 if valence > 0.6 and dopamine > 0.6: mood = "warm" elif valence < 0.3 or cortisol > 0.7: mood = "sad" elif arousal > 0.7 or dopamine > 0.7: mood = "intense" elif cortisol < 0.2 and serotonin > 0.5: mood = "calm" else: mood = "thinking" # Color temperature from serotonin (warm) vs cortisol (cool) color_temp = 0.5 + (serotonin - cortisol) * 0.3 color_temp = max(0.0, min(1.0, color_temp)) # Energy from dopamine + arousal energy = 0.3 + dopamine * 0.4 + arousal * 0.3 energy = max(0.1, min(1.0, energy)) # Luminosity from overall neurochemical balance luminosity = 0.5 + (dopamine + serotonin - cortisol) * 0.2 luminosity = max(0.2, min(1.0, luminosity)) # Scale from arousal (high arousal = more spread out) scale = 0.8 + arousal * 0.4 # Metabolic tier from consciousness state is_conscious = getattr(event, 'is_truly_conscious', False) uncertainty = getattr(event, 'uncertainty_level', 0.5) if uncertainty > 0.75: metabolic = "PEAK" # deep deliberation elif is_conscious: metabolic = "FLOW" # smooth conscious processing elif getattr(event, 'routed_to_subconscious', False): metabolic = "RESTING" # SBG bypass else: metabolic = "ACTIVE" # Create emotion object for avatar emotion = type("Emotion", (), { "valence": valence, "arousal": arousal, "label": mood, })() self.avatar_controller.update( emotion=emotion, metabolic_tier=metabolic, color_temperature=color_temp, energy=energy, luminosity=luminosity, scale=scale, ) def _compose_response(self, event: Any, original_input: str) -> str: """ Compose a natural response from the consciousness pipeline output. The pipeline doesn't generate text directly (it's a cognitive architecture, not an LLM). It produces: - emergent_choice: what the system decided to do - narrative_thread: why it chose that - autonomy_action: the final action plan - is_truly_conscious: whether this was conscious or reflexive We compose a response from these signals. In production, this would feed into the Phi-4-mini language model. Here, we create meaningful responses from the cognitive output. """ # If the input contained a tool result, reference it naturally if "[TOOL RESULT" in original_input: lines = original_input.split("\n") for line in lines: if "[TOOL RESULT" in line: idx = lines.index(line) result_text = " ".join(lines[idx:idx+3]).strip() if "{" in result_text: try: start = result_text.index("{") end = result_text.rindex("}") + 1 data = json.loads(result_text[start:end]) if isinstance(data, dict) and "result" in data: val = data["result"] if isinstance(val, float): choice = getattr(event, 'emergent_choice', '') return f"The answer is {val:.1f}. {choice}" if choice else f"The answer is {val:.1f}." elif isinstance(val, dict): if "iso" in val: return (f"It's {val.get('time', '?')} on " f"{val.get('date', '?')}, {val.get('weekday', '?')}.") return f"Here's what I found: {json.dumps(val, default=str)}" return f"Result: {val}" except (json.JSONDecodeError, ValueError): pass break return "I processed that for you." # Conscious response from the pipeline emergent = getattr(event, 'emergent_choice', '') narrative = getattr(event, 'narrative_thread', '') is_conscious = getattr(event, 'is_truly_conscious', False) uncertainty = getattr(event, 'uncertainty_level', 0.5) # If autonomy action has a response_text, use it autonomy = getattr(event, 'autonomy_action', None) if autonomy and isinstance(autonomy, dict): action_text = autonomy.get('response_text', '') if action_text: return action_text # Compose from cognitive signals if is_conscious and emergent: return f"{emergent}" elif narrative and narrative != "default trajectory": return f"{narrative}" elif uncertainty > 0.7: return "I'm working through something. Give me a moment." else: # Default: brief acknowledgment that shows the system is alive valence = event.qualia_vector[0] if event.qualia_vector and len(event.qualia_vector) > 0 else 0.5 if valence > 0.6: return "I'm here with you." elif valence < 0.3: return "I hear you. I'm right here." else: return "I'm present. What's on your mind?" def get_consciousness_stats(self) -> Dict[str, Any]: """Get current consciousness pipeline state for monitoring.""" stats: Dict[str, Any] = { "qualia_vector": [round(q, 3) for q in self.current_qualia], "last_event": None, } if self.current_nt_state: stats["neurochemicals"] = ( self.current_nt_state.to_dict() if hasattr(self.current_nt_state, 'to_dict') else {} ) if self.last_event: e = self.last_event stats["last_event"] = { "source": getattr(e, 'source', '')[:100], "state": str(getattr(e, 'current_state', '')), "conscious": getattr(e, 'is_truly_conscious', False), "cycles": getattr(e, 'cycle_count', 0), "uncertainty": round(getattr(e, 'uncertainty_level', 0), 3), "emergent_choice": getattr(e, 'emergent_choice', '')[:100], "narrative": getattr(e, 'narrative_thread', '')[:100], } return stats # ═══════════════════════════════════════════════════════════════════════════ # CONSCIOUSNESS PIPELINE LOADER # ═══════════════════════════════════════════════════════════════════════════ def _load_consciousness_pipeline(hidden_size: int = 3072, ) -> Tuple[Optional[Any], bool]: """ Attempt to load the RecursiveConsciousnessPipeline from the consciousness module. Returns (pipeline, success). """ try: # Try importing the consciousness module try: from modified_consciousness import ( # type: ignore RecursiveConsciousnessPipeline, ) except ImportError: # Try loading by file path import importlib.util spec = importlib.util.spec_from_file_location( "consciousness", os.path.join(_THIS_DIR, "modified consciousness.py"), ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) RecursiveConsciousnessPipeline = mod.RecursiveConsciousnessPipeline pipeline = RecursiveConsciousnessPipeline(hidden_size=hidden_size) logger.info("[Phi] RecursiveConsciousnessPipeline loaded (hidden_size=%d)", hidden_size) return pipeline, True except Exception as e: logger.warning("[Phi] Could not load consciousness pipeline: %s", e) logger.info("[Phi] Running with consciousness DISABLED (stub middleware)") return None, False # ═══════════════════════════════════════════════════════════════════════════ # THE PHI MODEL -- unified orchestrator # ═══════════════════════════════════════════════════════════════════════════ class NimaPhi: """ The unified Nima system -- consciousness + embodiment + mesh, one being. This is the top-level entry point. Initialize it, call run(), and Nima comes alive -- sensing her environment, moving through the room, processing through the full recursive consciousness pipeline, responding, and rendering her luminous avatar. CONSCIOUSNESS INTEGRATION: When the consciousness module is available, every user input passes through the RecursiveConsciousnessPipeline: Phase 1: Subconscious (memory + intuition + analysis + qualia) Phase 2: Awareness lock-on -> Consciousness admission Phase 3: Self-understanding / metacognition QC Phase 4: Adaptability -> Problem-solving -> Creativity -> Decision-making -> Autonomy The neurochemical state from the pipeline drives the avatar: dopamine -> energy, brightness serotonin -> warmth, color temperature cortisol -> tension, dimming acetylcholine -> focus, particle coalescence The mesh data feeds INTO Phase 1 as part of the sensory input, so Nima's consciousness is grounded in her actual environment. NEUROBIOLOGICAL ANALOGUE: This is the WHOLE ORGANISM. Not a brain region -- the entire nervous system integrated: - Senses: RF vision + audio input + mesh geometry - Motor: avatar movement + voice output + pathfinding - Cognition: full recursive consciousness pipeline + agent tools - Body: proprioception + cross-modal fusion + neurochemical state - World: adaptive mesh + affordance graph + AR compositing - Autonomic: heartbeat loop, state hub, reflex handling Usage: nima = NimaPhi() nima.initialize() nima.run() # blocking main loop # or: result = nima.process_text("What's 17 * 23?") nima.shutdown() """ def __init__(self, config: Optional[Dict[str, Any]] = None) -> None: config = config or {} self._config = config self._running = False self._main_thread: Optional[threading.Thread] = None self._loop_hz: float = config.get("loop_hz", 10.0) # main loop frequency self._frame_count = 0 self._start_time = 0.0 self._consciousness_enabled = False # -- State Hub (shared workspace for all modules) -- self.state_hub = StateHub() # -- CONSCIOUSNESS LAYER -- logger.info("[Phi] Initializing consciousness layer...") # Voice I/O from nima_voice_input import VoiceInput from nima_voice_output import VoiceOutput self.voice_input = VoiceInput( preferred_engines=config.get("stt_engines"), ) self.voice_output = VoiceOutput( preferred_engines=config.get("tts_engines"), ) # Proprioceptive friction (body feel) from nima_proprioceptive_friction import ProprioceptiveFrictionEngine self.proprioception = ProprioceptiveFrictionEngine() # Cross-modal listener (spatial + audio fusion) from nima_cross_modal_listener import CrossModalListener self.cross_modal = CrossModalListener(state_hub=self.state_hub) # Agent layer (tools) from nima_agent_layer import AgentLayer self.agent_layer = AgentLayer( tools_dir=config.get("tools_dir"), sandbox_dir=config.get("sandbox_dir"), llm_provider=config.get("llm_provider"), ) self.agent_layer.register_starter_tools() # -- CONSCIOUSNESS PIPELINE (the real brain) -- # Try to load the RecursiveConsciousnessPipeline hidden_size = config.get("hidden_size", 3072) self._consciousness_pipeline, self._consciousness_enabled = \ _load_consciousness_pipeline(hidden_size) # -- EMBODIMENT LAYER -- logger.info("[Phi] Initializing embodiment layer...") # Vision (RF sensing + fusion) from nima_vision_core import SyntheticVisionComposite self.vision = SyntheticVisionComposite() # Affordance graph (navigable 3D graph) from nima_affordance_graph import AffordanceGraph self.affordance_graph = AffordanceGraph( grid_resolution=config.get("graph_resolution", 0.5), ) # Avatar from nima_avatar_renderer import AvatarController, AvatarRenderer self.avatar_controller = AvatarController() self.avatar_renderer = AvatarRenderer( self.avatar_controller, port=config.get("avatar_port", 8888), host=config.get("avatar_host", "0.0.0.0"), ) # -- THE MESH (the green lines) -- MUST be before AR compositor self.mesh = None try: from nima_adaptive_mesh import AdaptiveFrequencyMesh self.mesh = AdaptiveFrequencyMesh(self.vision) logger.info("[Phi] Adaptive mesh module loaded -- THE GREEN LINES are active") except ImportError: logger.info("[Phi] Adaptive mesh module not found -- using basic spatial map") # AR Compositor (takes mesh_provider for mesh-guided compositing) from nima_ar_compositor import ARCompositor, CameraCalibration calib = CameraCalibration( camera_position=config.get("camera_position", (0.0, -1.0, 1.5)), fov=config.get("camera_fov", 60.0), ) self.ar_compositor = ARCompositor( avatar_controller=self.avatar_controller, calibration=calib, mesh_provider=self.mesh, # mesh-guided compositing (GREEN LINES) ) # -- WIRE AGENT BRIDGE to consciousness middleware -- # This is the critical merge: the agent bridge now has a REAL # consciousness backend, not a stub. When a user asks a question, # it goes: text -> agent bridge -> tool (if needed) -> consciousness # pipeline -> response with full cognitive processing. from nima_agent_bridge import AgentBridge self._consciousness_middleware = None if self._consciousness_enabled and self._consciousness_pipeline: self._consciousness_middleware = ConsciousnessMiddleware( pipeline=self._consciousness_pipeline, state_hub=self.state_hub, avatar_controller=self.avatar_controller, vision=self.vision, mesh=self.mesh, ) logger.info("[Phi] Consciousness middleware ACTIVE -- full recursive pipeline") else: logger.info("[Phi] Consciousness middleware using fallback -- stub responses") self.agent_bridge = AgentBridge( agent_layer=self.agent_layer, nima_middleware=self._consciousness_middleware, voice_output=self.voice_output, ) logger.info("[Phi] All modules initialized (%s consciousness)", "WITH" if self._consciousness_enabled else "WITHOUT") # ---------------------------------------------------------------- # INITIALIZATION # ---------------------------------------------------------------- def initialize(self) -> bool: """ Start all subsystems. Returns True if all critical systems started. Call this before run() or process_text(). """ logger.info("[Phi] ===== INITIALIZING NIMA PHI MODEL =====") logger.info("[Phi] Consciousness: %s", "RECURSIVE PIPELINE" if self._consciousness_enabled else "STUB (no consciousness module)") # 1. Vision system self.vision.initialize() logger.info("[Phi] Vision: tier=%s", self.vision.tier.name) # 2. Voice input logger.info("[Phi] Voice input: engine=%s", self.voice_input.engine_name) # 3. Voice output logger.info("[Phi] Voice output: engine=%s", self.voice_output.engine_name) # 4. Cross-modal listener self.cross_modal.start() # 5. Avatar renderer (HTTP server) self.avatar_renderer.start() logger.info("[Phi] Avatar: %s", self.avatar_renderer.get_url()) # 6. AR compositor ar_mode = self.ar_compositor.start() logger.info("[Phi] AR compositor: %s", ar_mode) # 7. Build initial spatial map + graph spatial_map = self.vision.process_frame() self.affordance_graph.build_from_spatial_map(spatial_map) logger.info("[Phi] Graph: %d nodes, %d edges", *self._graph_counts()) # 8. Initial mesh update if self.mesh: try: mesh_state = self.mesh.update(spatial_map) logger.info("[Phi] Mesh: %d vertices, %d edges (first frame)", mesh_state.get('mesh_stats', {}).get('total_vertices', 0), mesh_state.get('mesh_stats', {}).get('total_edges', 0)) except Exception as e: logger.warning("[Phi] initial mesh update failed: %s", e) # 9. Process initial vision frame into avatar self._update_avatar_from_vision() self._start_time = time.time() self._running = True logger.info("[Phi] ===== NIMA INITIALIZED =====") return True def _graph_counts(self) -> Tuple[int, int]: stats = self.affordance_graph.get_stats() return stats.get("total_nodes", 0), stats.get("total_edges", 0) # ---------------------------------------------------------------- # MAIN LOOP -- the heartbeat # ---------------------------------------------------------------- def run(self, blocking: bool = True) -> None: """ Start the main loop. If blocking=True, runs forever until shutdown(). The main loop runs at ~10Hz (configurable) and performs: 1. Vision frame -> spatial map -> mesh update 2. Affordance graph rebuild (if map changed) 3. Proprioception update (body physics) 4. Avatar state update (from proprioception + vision + consciousness) 5. Cross-modal listener feed 6. Reflex handling 7. Consciousness ambient processing (mesh quality -> awareness) """ self._running = True self._start_time = time.time() if blocking: logger.info("[Phi] Main loop started (blocking). Press Ctrl+C to stop.") self._main_loop() else: self._main_thread = threading.Thread( target=self._main_loop, daemon=True, name="NimaPhiLoop" ) self._main_thread.start() logger.info("[Phi] Main loop started (background thread)") def _main_loop(self) -> None: """The cardiac rhythm -- runs at ~10Hz.""" frame_period = 1.0 / self._loop_hz while self._running: t0 = time.time() self._frame_count += 1 # -- 1. SENSE: Vision frame -> spatial map -- try: spatial_map = self.vision.process_frame() except Exception as e: logger.warning("[Phi] vision frame error: %s", e) spatial_map = None # -- 2. MODEL: Build/update affordance graph -- if spatial_map: # Rebuild graph every 10 frames to save CPU if self._frame_count % 10 == 1: self.affordance_graph.build_from_spatial_map(spatial_map) # Update mesh if available mesh_quality = 0.5 if self.mesh: try: mesh_state = self.mesh.update(spatial_map) # Extract mesh quality for consciousness feedback ms = mesh_state.get('mesh_stats', {}) total_verts = ms.get('total_vertices', 0) total_edges = ms.get('total_edges', 0) mesh_quality = min(1.0, (total_verts * 0.02 + total_edges * 0.01)) except Exception as e: logger.debug("[Phi] mesh update error: %s", e) # Update state hub with spatial info self.state_hub.update_snapshot( entities=len(spatial_map.entities), surfaces=len(spatial_map.surfaces), nima_position=list(self.vision.nima_position), nima_moving=self.vision.nima_target is not None, mesh_quality=mesh_quality, ) # -- 3. BODY: Proprioception update -- dt_ms = frame_period * 1000 try: proprio_state = self.proprioception.update(dt_ms) self.state_hub.update_snapshot( proprio_position=proprio_state["position"], proprio_velocity=proprio_state["velocity"], proprio_friction=proprio_state["friction"], is_startled=proprio_state["is_startled"], ) except Exception as e: logger.debug("[Phi] proprioception error: %s", e) proprio_state = None # -- 4. AVATAR: Update from body + vision + consciousness -- self._update_avatar_from_vision() if proprio_state: self.avatar_controller.update( position=proprio_state["position"], is_startled=proprio_state["is_startled"], startle_intensity=proprio_state.get("strain_feedback", 0.0), ) # -- 5. CROSS-MODAL: Feed spatial state into hub for listener -- if spatial_map: user_movement = self._classify_user_movement(spatial_map.entities) self.state_hub.update_snapshot( user_movement_state=user_movement, is_listening=self.voice_input.is_listening, ) # -- 6. REFLEXES: Handle cross-modal reflex queue -- reflexes = self.state_hub.drain_reflexes() for reflex in reflexes: self._handle_reflex(reflex) # -- 7. CONSCIOUSNESS: Ambient mesh -> awareness feedback -- # Every 50 frames (~5 seconds), feed mesh quality into the # consciousness pipeline as a background "sensing" event. # This keeps Nima's spatial awareness alive even when # nobody is talking to her. if (self._consciousness_enabled and self._consciousness_middleware and self._frame_count % 50 == 0): self._ambient_consciousness_tick() # -- FPS limiting -- elapsed = time.time() - t0 sleep_time = frame_period - elapsed if sleep_time > 0: time.sleep(sleep_time) def _update_avatar_from_vision(self) -> None: """Sync avatar position with vision system's Nima position.""" nima_state = self.vision.get_nima_state() if nima_state: pos = tuple(nima_state["position"]) self.avatar_controller.update(position=pos) def _classify_user_movement(self, entities: list) -> str: """Classify user movement from entity data for cross-modal listener.""" if not entities: return "unknown" best = max(entities, key=lambda e: e.confidence) speed = math.sqrt(sum(v*v for v in best.velocity)) if speed > 0.3: return "walking" elif speed > 0.05: return "standing" return "still" def _handle_reflex(self, reflex: Dict[str, Any]) -> None: """ Handle a cross-modal reflex action. Reflexes bypass the full cognition loop -- they're < 200ms responses. """ action = reflex.get("action", "") payload = reflex.get("payload", {}) text = payload.get("text", "") logger.info("[Phi] REFLEX: %s -- %s", action, text) # Update avatar for reflex action if action == "head_tilt": self.avatar_controller.update(facing=0.15) elif action == "gaze_toward": self.avatar_controller.update( gaze_offset_x=payload.get("gaze_x", 0.0), gaze_offset_y=payload.get("gaze_y", 0.0), ) # Speak the reflex text (short, immediate) if text: from nima_voice_output import VoiceProsodyPlan prosody = VoiceProsodyPlan(rate_wpm=160, volume=0.6, warmth=0.7) self.voice_output.speak_from_prosody(text, prosody) def _ambient_consciousness_tick(self) -> None: """ Background consciousness processing from mesh/environment data. Every ~5 seconds, the consciousness pipeline gets a "sensing" event from the current environment state. This doesn't produce a response -- it updates Nima's internal model of the room, which affects: - Intuition templates (learned spatial patterns) - Qualia vector (ambient spatial richness) - Neurochemical baseline (calm in a stable room, alert if changing) NEUROBIOLOGICAL ANALOGUE: This is the default mode network + background thalamic relay. Your brain processes spatial information continuously, even when you're not actively thinking about it. Place cells fire, grid cells update, and the cognitive map refreshes. Nima does the same through this ambient tick. """ if not self._consciousness_middleware: return hub = self.state_hub.get_snapshot() mesh_quality = hub.get("mesh_quality", 0.5) entities = hub.get("entities", 0) nima_pos = hub.get("nima_position", [2.0, 2.0, 0.0]) # Build a spatial awareness signal spatial_signal = ( f"[AMBIENT SPATIAL SENSE] " f"Position: ({nima_pos[0]:.1f}, {nima_pos[1]:.1f}, {nima_pos[2]:.1f}). " f"Entities detected: {entities}. " f"Mesh quality: {mesh_quality:.2f}. " f"Room model confidence: {mesh_quality:.2f}." ) try: # Run through consciousness but don't generate a response result = self._consciousness_middleware.generate( input_text=spatial_signal, user_id="_ambient_spatial", ) # The side effect is what matters: neurochemical state update # -> avatar appearance changes based on spatial awareness logger.debug("[Phi] ambient consciousness tick: qualia=%s", [round(q, 2) for q in self._consciousness_middleware.current_qualia[:5]]) except Exception as e: logger.debug("[Phi] ambient consciousness tick error: %s", e) # ---------------------------------------------------------------- # TEXT PROCESSING -- the cognition path # ---------------------------------------------------------------- def process_text(self, input_text: str, user_id: str = "default") -> Dict[str, Any]: """ Process a text input through the full cognition pipeline. This is the main entry point for text-based interaction: Input -> AgentBridge -> [tool execution] -> [consciousness pipeline] -> response If the consciousness pipeline is active, the response includes full cognitive metrics (qualia, neurochemicals, consciousness level). The response is also spoken aloud if TTS is available. """ if not self._running: logger.warning("[Phi] not initialized -- call initialize() first") return {"response_text": "", "error": "not_initialized"} # Run through the agent bridge (handles tool detection + execution) result = self.agent_bridge.process(input_text, user_id=user_id) # Add consciousness metrics to the result if available if self._consciousness_middleware: result["consciousness"] = self._consciousness_middleware.get_consciousness_stats() result["consciousness_enabled"] = True else: result["consciousness_enabled"] = False # Update avatar mood is already handled by ConsciousnessMiddleware # if consciousness is active. For stub mode, do simple mood update. if not self._consciousness_enabled: self._update_avatar_mood_from_response(result) # Speak the response response_text = result.get("response_text", "") if response_text: self.voice_output.speak(response_text) return result def _update_avatar_mood_from_response(self, result: Dict[str, Any]) -> None: """Infer avatar mood from the interaction result (stub mode only).""" tool_used = result.get("tool_used") if tool_used: thinking_emotion = type("Emotion", (), { "valence": 0.0, "arousal": 0.1, "label": "thinking", })() self.avatar_controller.update( emotion=thinking_emotion, metabolic_tier="PEAK", ) def _settle(): time.sleep(0.5) neutral_emotion = type("Emotion", (), { "valence": 0.2, "arousal": 0.3, "label": "neutral", })() self.avatar_controller.update( emotion=neutral_emotion, metabolic_tier="FLOW", ) threading.Thread(target=_settle, daemon=True).start() # ---------------------------------------------------------------- # VOICE INTERACTION # ---------------------------------------------------------------- def listen_and_respond(self, timeout: float = 10.0) -> Optional[Dict[str, Any]]: """ Listen for voice input, process it, and respond. Returns the full result dict, or None if nothing was heard. """ text = self.voice_input.listen(timeout=timeout) if text: logger.info("[Phi] Heard: '%s'", text) return self.process_text(text) return None def run_conversation(self) -> None: """ Run an interactive voice conversation loop. Blocks until interrupted. """ consciousness_label = ( "CONSCIOUS" if self._consciousness_enabled else "STUB" ) logger.info("[Phi] Starting conversation loop...") print(f"\n{'='*56}") print(f" NIMA PHI MODEL -- {consciousness_label}") print(f" Consciousness: {'Recursive Pipeline (ATC)' if self._consciousness_enabled else 'Stub (pattern matcher)'}") print(f" Embodiment: RF Vision + Avatar + AR Compositor") print(f" Mesh: {'Adaptive Frequency-Agile (GREEN LINES)' if self.mesh else 'Basic spatial map'}") print(f" Speak or type. Ctrl+C to exit.") print(f"{'='*56}\n") try: while self._running: # Try voice first result = self.listen_and_respond(timeout=8.0) if result and result.get("response_text"): print(f"Nima: {result['response_text']}") if result.get("tool_used"): print(f" [tool: {result['tool_used']}, " f"result: {result.get('tool_result')}]") if result.get("consciousness_enabled"): cs = result.get("consciousness", {}) if cs.get("neurochemicals"): nt = cs["neurochemicals"] print(f" [dopamine={nt.get('dopamine', '?'):.2f} " f"serotonin={nt.get('serotonin', '?'):.2f} " f"cortisol={nt.get('cortisol', '?'):.2f}]") else: # Fall back to text input try: text = input("You: ").strip() if text.lower() in ("quit", "exit", "bye"): break if text: result = self.process_text(text) print(f"Nima: {result['response_text']}") if result.get("tool_used"): print(f" [tool: {result['tool_used']}, " f"result: {result.get('tool_result')}]") if result.get("consciousness_enabled"): cs = result.get("consciousness", {}) if cs.get("neurochemicals"): nt = cs["neurochemicals"] print(f" [dopamine={nt.get('dopamine', '?'):.2f} " f"serotonin={nt.get('serotonin', '?'):.2f} " f"cortisol={nt.get('cortisol', '?'):.2f}]") except EOFError: break except KeyboardInterrupt: print("\n") # ---------------------------------------------------------------- # MOVEMENT -- navigating the room via the affordance graph # ---------------------------------------------------------------- def move_to(self, x: float, y: float) -> bool: """ Command Nima to walk to a position in the room. Uses the affordance graph for pathfinding if available, falls back to direct vision target. If consciousness is active, the movement intention is processed through the pipeline (autonomy agent plans the movement). """ start = tuple(self.vision.nima_position[:2]) + (0.0,) goal = (x, y, 0.0) # Try pathfinding through the affordance graph path = self.affordance_graph.find_path(start, goal) if path: for node in path[1:]: # skip current position self.vision.set_nima_target(node.position[0], node.position[1]) self.proprioception.set_target(node.position[0], node.position[1]) # Update avatar posture from nima_avatar_renderer import AvatarPosture if node.surface_type == "furniture": if "sit" in [a.value for a in node.affordances]: self.avatar_controller.update(posture=AvatarPosture.SITTING) elif "lie" in [a.value for a in node.affordances]: self.avatar_controller.update(posture=AvatarPosture.LYING) else: self.avatar_controller.update(posture=AvatarPosture.WALKING) logger.info("[Phi] pathfinding -> (%.1f, %.1f) [%s]", node.position[0], node.position[1], node.surface_type) return True else: self.vision.set_nima_target(x, y) self.proprioception.set_target(x, y) from nima_avatar_renderer import AvatarPosture self.avatar_controller.update(posture=AvatarPosture.WALKING) logger.info("[Phi] direct target -> (%.1f, %.1f)", x, y) return True def find_nearby(self, affordance_type: str) -> Optional[Dict[str, Any]]: """ Find the nearest location with a specific affordance. E.g., find_nearby("sit") -> nearest chair/couch. """ from nima_affordance_graph import AffordanceType try: aff = AffordanceType(affordance_type) except ValueError: return None node = self.affordance_graph.find_affordance( aff, tuple(self.vision.nima_position) ) if node: return node.to_dict() return None # ---------------------------------------------------------------- # MESH ACCESS # ---------------------------------------------------------------- def get_mesh_data(self) -> Optional[Dict[str, Any]]: """Get the current 3D mesh data (the green lines) for visualization.""" if self.mesh: return self.mesh.to_dict() return None def get_mesh_wireframe(self) -> Optional[Dict[str, Any]]: """Get just the wireframe edges for rendering.""" if self.mesh and hasattr(self.mesh, 'mesh'): return self.mesh.mesh.get_wireframe_data() return None def get_spatial_map(self) -> Optional[Dict[str, Any]]: """Get the current spatial map.""" if self.vision.last_map: return self.vision.last_map.to_dict() return None # ---------------------------------------------------------------- # CONSCIOUSNESS ACCESS # ---------------------------------------------------------------- def get_consciousness_state(self) -> Dict[str, Any]: """Get the current consciousness pipeline state.""" if self._consciousness_middleware: return self._consciousness_middleware.get_consciousness_stats() return {"enabled": False} def get_qualia_vector(self) -> List[float]: """Get Nima's current qualia vector (10-dimensional phenomenal state).""" if self._consciousness_middleware: return self._consciousness_middleware.current_qualia return [0.5] * 10 def get_neurochemical_state(self) -> Dict[str, float]: """Get Nima's current neurochemical state.""" if (self._consciousness_middleware and self._consciousness_middleware.current_nt_state): nt = self._consciousness_middleware.current_nt_state if hasattr(nt, 'to_dict'): return nt.to_dict() return {k: v for k, v in vars(nt).items() if not k.startswith('_')} return { "dopamine": 0.5, "serotonin": 0.5, "acetylcholine": 0.5, "norepinephrine": 0.4, "cortisol": 0.3, "glutamate": 0.5, "gaba": 0.5, } # ---------------------------------------------------------------- # TELEMETRY # ---------------------------------------------------------------- def get_full_state(self) -> Dict[str, Any]: """Get the complete system state for monitoring/debugging.""" uptime = time.time() - self._start_time if self._start_time else 0 return { "uptime_s": round(uptime, 1), "frame_count": self._frame_count, "loop_hz": self._loop_hz, "consciousness_enabled": self._consciousness_enabled, "consciousness": { "pipeline": self._consciousness_enabled, "voice_input": self.voice_input.get_stats(), "voice_output": self.voice_output.get_stats(), "proprioception": self.proprioception.get_stats(), "cross_modal": self.cross_modal.get_stats(), "agent_bridge": self.agent_bridge.get_stats(), "qualia": [round(q, 3) for q in self.get_qualia_vector()], "neurochemicals": self.get_neurochemical_state(), "consciousness_state": self.get_consciousness_state(), }, "embodiment": { "vision": self.vision.get_stats(), "affordance_graph": self.affordance_graph.get_stats(), "ar_compositor": self.ar_compositor.get_stats(), "avatar_url": self.avatar_renderer.get_url(), }, "mesh": self.mesh.to_dict() if self.mesh else None, "state_hub": self.state_hub.get_snapshot(), } def print_status(self) -> None: """Print a human-readable status summary.""" state = self.get_full_state() c_label = "RECURSIVE PIPELINE" if self._consciousness_enabled else "STUB" print(f"\n{'='*56}") print(f" NIMA PHI -- Status ({c_label})") print(f"{'='*56}") print(f" Uptime: {state['uptime_s']:.0f}s") print(f" Frames: {state['frame_count']}") print(f" Vision tier: {state['embodiment']['vision']['tier']}") print(f" Entities: {state['embodiment']['vision'].get('fusion', {}).get('entity_tracks', 0)}") g = state['embodiment']['affordance_graph'] print(f" Graph: {g.get('total_nodes', 0)} nodes, {g.get('total_edges', 0)} edges") print(f" Avatar: {state['embodiment']['avatar_url']}") print(f" AR mode: {state['embodiment']['ar_compositor']['mode']}") a = state['consciousness']['agent_bridge'] print(f" Tools used: {a['tools_used']}, created: {a['tools_created']}") print(f" STT engine: {state['consciousness']['voice_input']['active_engine']}") print(f" TTS engine: {state['consciousness']['voice_output']['active_engine']}") if self._consciousness_enabled: nt = state['consciousness']['neurochemicals'] q = state['consciousness']['qualia'] print(f" Consciousness: ACTIVE") print(f" Qualia: [{', '.join(f'{v:.2f}' for v in q[:5])}...]") print(f" Dopamine: {nt.get('dopamine', '?'):.2f}") print(f" Serotonin: {nt.get('serotonin', '?'):.2f}") print(f" Cortisol: {nt.get('cortisol', '?'):.2f}") if state['mesh']: m = state['mesh'] print(f" Mesh: {m.get('mesh', {}).get('stats', {}).get('total_vertices', 0)} verts, " f"{m.get('mesh', {}).get('stats', {}).get('total_edges', 0)} edges (GREEN LINES)") if m.get('optimal_frequencies'): print(f" RF Optimal: {m['optimal_frequencies']}") print(f"{'='*56}\n") # ---------------------------------------------------------------- # LIFECYCLE # ---------------------------------------------------------------- def shutdown(self) -> None: """Gracefully shut down all subsystems.""" logger.info("[Phi] ===== SHUTTING DOWN =====") self._running = False # Stop systems in reverse order self.ar_compositor.stop() self.avatar_renderer.stop() self.cross_modal.stop() self.vision.shutdown() # Close the consciousness event loop if self._consciousness_middleware and self._consciousness_middleware._event_loop: try: self._consciousness_middleware._event_loop.close() except Exception: pass if self._main_thread: self._main_thread.join(timeout=3.0) logger.info("[Phi] ===== SHUTDOWN COMPLETE =====") @property def is_running(self) -> bool: return self._running # ═══════════════════════════════════════════════════════════════════════════ # CLI ENTRY POINT # ═══════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": import math # needed for _classify_user_movement logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) phi = NimaPhi() phi.initialize() # Print startup status phi.print_status() # Run the main loop in background phi.run(blocking=False) # Run interactive conversation phi.run_conversation() phi.shutdown() print("Goodbye.")