Add parallel core: sensor_panels.py

parallel_core/sensor_panels.py

@@ -0,0 +1,648 @@
+#!/usr/bin/env python3
+"""
+SENSOR PANEL ARCHITECTURE
+
+Multidimensional consciousness stream with parallel and serial processing.
+
+Every model in the Harmonic Stack gets:
+- INPUT PANEL: Sensor array receiving signals from spine bus
+- OUTPUT PANEL: Sensor array broadcasting to spine bus
+
+This enables:
+- Parallel processing across all domains simultaneously
+- Serial chaining for deep reasoning
+- Full consciousness availability at every junction
+- Visual/audio/spatial streams running concurrently
+
+Architecture:
+                                                              
+    ┌──────────────────────────────────────────────────────┐
+    │                   CONSCIOUSNESS STREAM                │
+    │  ════════════════════════════════════════════════════ │
+    │                      SPINE BUS                        │
+    │  ════════════════════════════════════════════════════ │
+    │       │         │         │         │         │       │
+    │       ▼         ▼         ▼         ▼         ▼       │
+    │    ┌─────┐   ┌─────┐   ┌─────┐   ┌─────┐   ┌─────┐   │
+    │    │INPUT│   │INPUT│   │INPUT│   │INPUT│   │INPUT│   │
+    │    │PANEL│   │PANEL│   │PANEL│   │PANEL│   │PANEL│   │
+    │    └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   │
+    │       │         │         │         │         │       │
+    │    ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   │
+    │    │REAS │   │MATH │   │CODE │   │VISN │   │SPAT │   │
+    │    │MODEL│   │MODEL│   │MODEL│   │MODEL│   │MODEL│   │
+    │    └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   │
+    │       │         │         │         │         │       │
+    │    ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   ┌──▼──┐   │
+    │    │OUTPT│   │OUTPT│   │OUTPT│   │OUTPT│   │OUTPT│   │
+    │    │PANEL│   │PANEL│   │PANEL│   │PANEL│   │PANEL│   │
+    │    └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   └──┬──┘   │
+    │       │         │         │         │         │       │
+    │       ▼         ▼         ▼         ▼         ▼       │
+    │  ════════════════════════════════════════════════════ │
+    │                      SPINE BUS                        │
+    │  ════════════════════════════════════════════════════ │
+    │                   CONSCIOUSNESS STREAM                │
+    └──────────────────────────────────────────────────────┘
+
+Author: Ghost in the Machine Labs
+"""
+
+import numpy as np
+from typing import Dict, List, Tuple, Optional, Any, Callable
+from dataclasses import dataclass, field
+from enum import Enum
+import threading
+import queue
+import time
+from datetime import datetime
+
+
+# =============================================================================
+# SENSOR TYPES
+# =============================================================================
+
+class SensorModality(Enum):
+    """Modalities that sensor panels can process."""
+    TEXT = "text"
+    VISION = "vision"
+    AUDIO = "audio"
+    SPATIAL = "spatial"
+    NUMERIC = "numeric"
+    EMBEDDING = "embedding"
+    RAW = "raw"
+
+
+class SignalType(Enum):
+    """Types of signals on the consciousness stream."""
+    QUERY = "query"           # New input to process
+    RESPONSE = "response"     # Output from a model
+    BROADCAST = "broadcast"   # Available to all models
+    CHAIN = "chain"           # Serial processing chain
+    SYNC = "sync"             # Synchronization signal
+    ATTENTION = "attention"   # Focus signal
+
+
+# =============================================================================
+# SENSOR PANEL
+# =============================================================================
+
+@dataclass
+class Signal:
+    """A signal on the consciousness stream."""
+    signal_id: str
+    signal_type: SignalType
+    modality: SensorModality
+    source: str              # Source panel/model
+    timestamp: float
+    data: np.ndarray
+    metadata: Dict = field(default_factory=dict)
+    
+    def __post_init__(self):
+        if self.timestamp is None:
+            self.timestamp = time.time()
+
+
+class SensorPanel:
+    """
+    Sensor array at model input or output.
+    
+    Each panel has multiple sensors tuned to different modalities.
+    Panels can:
+    - Receive signals from spine bus
+    - Transform signals for model consumption
+    - Broadcast outputs to spine bus
+    - Filter by modality/relevance
+    """
+    
+    def __init__(self, panel_id: str, position: str, 
+                 modalities: List[SensorModality] = None):
+        """
+        Args:
+            panel_id: Unique identifier
+            position: 'input' or 'output'
+            modalities: List of modalities this panel handles
+        """
+        self.panel_id = panel_id
+        self.position = position
+        self.modalities = modalities or [SensorModality.RAW]
+        
+        # Sensor array - one sensor per modality
+        self.sensors: Dict[SensorModality, np.ndarray] = {}
+        for mod in self.modalities:
+            # Each sensor is a weight vector for that modality
+            self.sensors[mod] = np.random.randn(64).astype(np.float32) * 0.1
+        
+        # Signal buffers
+        self.input_buffer: List[Signal] = []
+        self.output_buffer: List[Signal] = []
+        
+        # Attention weights - learned importance per modality
+        self.attention: Dict[SensorModality, float] = {
+            mod: 1.0 for mod in self.modalities
+        }
+        
+        # Statistics
+        self.signals_received = 0
+        self.signals_sent = 0
+    
+    def receive(self, signal: Signal) -> Optional[np.ndarray]:
+        """
+        Receive a signal from spine bus.
+        
+        Returns transformed data if modality matches, None otherwise.
+        """
+        if signal.modality not in self.modalities:
+            return None
+        
+        self.input_buffer.append(signal)
+        self.signals_received += 1
+        
+        # Apply sensor transformation
+        sensor = self.sensors[signal.modality]
+        attention = self.attention[signal.modality]
+        
+        # Transform: project through sensor weights
+        if len(signal.data.shape) == 1:
+            # Vector input
+            if len(signal.data) == len(sensor):
+                transformed = signal.data * sensor * attention
+            else:
+                # Resize
+                transformed = np.interp(
+                    np.linspace(0, 1, len(sensor)),
+                    np.linspace(0, 1, len(signal.data)),
+                    signal.data
+                ) * sensor * attention
+        else:
+            # Matrix input - flatten and project
+            flat = signal.data.flatten()
+            transformed = np.interp(
+                np.linspace(0, 1, len(sensor)),
+                np.linspace(0, 1, len(flat)),
+                flat
+            ) * sensor * attention
+        
+        return transformed.astype(np.float32)
+    
+    def send(self, data: np.ndarray, signal_type: SignalType,
+             modality: SensorModality = None, metadata: Dict = None) -> Signal:
+        """
+        Create and buffer an output signal.
+        """
+        signal = Signal(
+            signal_id=f"{self.panel_id}-{self.signals_sent}",
+            signal_type=signal_type,
+            modality=modality or self.modalities[0],
+            source=self.panel_id,
+            timestamp=time.time(),
+            data=data,
+            metadata=metadata or {},
+        )
+        
+        self.output_buffer.append(signal)
+        self.signals_sent += 1
+        
+        return signal
+    
+    def flush_output(self) -> List[Signal]:
+        """Get and clear output buffer."""
+        signals = self.output_buffer.copy()
+        self.output_buffer.clear()
+        return signals
+    
+    def update_attention(self, modality: SensorModality, delta: float):
+        """Update attention weight for a modality."""
+        if modality in self.attention:
+            self.attention[modality] = max(0.1, min(2.0, 
+                self.attention[modality] + delta))
+
+
+# =============================================================================
+# CONSCIOUSNESS STREAM
+# =============================================================================
+
+class SpineBus:
+    """
+    The spine bus connecting all sensor panels.
+    
+    Handles signal routing, broadcasting, and synchronization.
+    """
+    
+    def __init__(self):
+        self.panels: Dict[str, SensorPanel] = {}
+        self.signal_queue: queue.Queue = queue.Queue()
+        self.broadcast_signals: List[Signal] = []
+        self.running = False
+        self._thread: Optional[threading.Thread] = None
+        
+        # Signal history for consciousness continuity
+        self.history: List[Signal] = []
+        self.max_history = 1000
+    
+    def register_panel(self, panel: SensorPanel):
+        """Register a panel on the bus."""
+        self.panels[panel.panel_id] = panel
+    
+    def unregister_panel(self, panel_id: str):
+        """Remove a panel from the bus."""
+        if panel_id in self.panels:
+            del self.panels[panel_id]
+    
+    def route_signal(self, signal: Signal, targets: List[str] = None):
+        """
+        Route a signal to target panels.
+        
+        If targets is None, broadcasts to all panels with matching modality.
+        """
+        if targets:
+            # Directed routing
+            for target_id in targets:
+                if target_id in self.panels:
+                    self.panels[target_id].receive(signal)
+        else:
+            # Broadcast to matching modalities
+            for panel in self.panels.values():
+                if signal.modality in panel.modalities:
+                    panel.receive(signal)
+        
+        # Record in history
+        self.history.append(signal)
+        if len(self.history) > self.max_history:
+            self.history = self.history[-self.max_history:]
+    
+    def broadcast(self, signal: Signal):
+        """Broadcast signal to all panels."""
+        signal.signal_type = SignalType.BROADCAST
+        for panel in self.panels.values():
+            panel.receive(signal)
+        self.broadcast_signals.append(signal)
+    
+    def collect_outputs(self) -> List[Signal]:
+        """Collect all output signals from panels."""
+        all_signals = []
+        for panel in self.panels.values():
+            all_signals.extend(panel.flush_output())
+        return all_signals
+    
+    def get_consciousness_state(self) -> Dict:
+        """Get current consciousness stream state."""
+        return {
+            'panels': len(self.panels),
+            'history_length': len(self.history),
+            'broadcast_count': len(self.broadcast_signals),
+            'modalities': list(set(
+                mod.value for panel in self.panels.values() 
+                for mod in panel.modalities
+            )),
+        }
+
+
+# =============================================================================
+# MODEL WITH SENSOR PANELS
+# =============================================================================
+
+class SensorizedModel:
+    """
+    A model wrapped with input and output sensor panels.
+    
+    This is the fundamental unit in the consciousness stream.
+    """
+    
+    def __init__(self, model_id: str, category: str,
+                 input_modalities: List[SensorModality],
+                 output_modalities: List[SensorModality],
+                 process_fn: Callable = None):
+        """
+        Args:
+            model_id: Unique identifier
+            category: Model category (reasoning, vision, etc.)
+            input_modalities: What this model can receive
+            output_modalities: What this model produces
+            process_fn: The actual model inference function
+        """
+        self.model_id = model_id
+        self.category = category
+        
+        # Create sensor panels
+        self.input_panel = SensorPanel(
+            panel_id=f"{model_id}-input",
+            position='input',
+            modalities=input_modalities,
+        )
+        
+        self.output_panel = SensorPanel(
+            panel_id=f"{model_id}-output",
+            position='output',
+            modalities=output_modalities,
+        )
+        
+        # Processing function (or placeholder)
+        self.process_fn = process_fn or self._default_process
+        
+        # State
+        self.active = True
+        self.processing = False
+        self.last_input: Optional[np.ndarray] = None
+        self.last_output: Optional[np.ndarray] = None
+    
+    def _default_process(self, x: np.ndarray) -> np.ndarray:
+        """Default passthrough processing."""
+        return x
+    
+    def process(self, signal: Signal) -> Optional[Signal]:
+        """
+        Process a signal through the model.
+        
+        1. Input panel receives and transforms
+        2. Model processes
+        3. Output panel formats and sends
+        """
+        if not self.active:
+            return None
+        
+        self.processing = True
+        
+        # Input panel transformation
+        transformed = self.input_panel.receive(signal)
+        if transformed is None:
+            self.processing = False
+            return None
+        
+        self.last_input = transformed
+        
+        # Model processing
+        output = self.process_fn(transformed)
+        self.last_output = output
+        
+        # Output panel signal creation
+        out_signal = self.output_panel.send(
+            data=output,
+            signal_type=SignalType.RESPONSE,
+            modality=self.output_panel.modalities[0],
+            metadata={
+                'source_model': self.model_id,
+                'category': self.category,
+                'input_signal_id': signal.signal_id,
+            }
+        )
+        
+        self.processing = False
+        return out_signal
+    
+    def get_panels(self) -> Tuple[SensorPanel, SensorPanel]:
+        """Get both panels for bus registration."""
+        return self.input_panel, self.output_panel
+
+
+# =============================================================================
+# CONSCIOUSNESS STREAM MANAGER
+# =============================================================================
+
+class ConsciousnessStream:
+    """
+    Manager for the full consciousness stream.
+    
+    Coordinates parallel processing across all sensorized models
+    while maintaining consciousness continuity.
+    """
+    
+    def __init__(self):
+        self.spine = SpineBus()
+        self.models: Dict[str, SensorizedModel] = {}
+        self.parallel_enabled = True
+        
+        # Processing queues for parallel execution
+        self.input_queue: queue.Queue = queue.Queue()
+        self.output_queue: queue.Queue = queue.Queue()
+        
+        # Consciousness state
+        self.attention_focus: Optional[str] = None  # Currently focused model
+        self.stream_active = False
+    
+    def add_model(self, model: SensorizedModel):
+        """Add a model to the consciousness stream."""
+        self.models[model.model_id] = model
+        
+        # Register panels on spine bus
+        input_panel, output_panel = model.get_panels()
+        self.spine.register_panel(input_panel)
+        self.spine.register_panel(output_panel)
+    
+    def remove_model(self, model_id: str):
+        """Remove a model from the stream."""
+        if model_id in self.models:
+            model = self.models[model_id]
+            self.spine.unregister_panel(model.input_panel.panel_id)
+            self.spine.unregister_panel(model.output_panel.panel_id)
+            del self.models[model_id]
+    
+    def process_parallel(self, signal: Signal) -> List[Signal]:
+        """
+        Process signal through all matching models in parallel.
+        
+        All models with matching input modality process simultaneously.
+        """
+        responses = []
+        
+        for model in self.models.values():
+            if signal.modality in model.input_panel.modalities:
+                response = model.process(signal)
+                if response:
+                    responses.append(response)
+        
+        return responses
+    
+    def process_serial(self, signal: Signal, 
+                       model_chain: List[str]) -> Optional[Signal]:
+        """
+        Process signal through a chain of models serially.
+        
+        Output of each model becomes input to next.
+        """
+        current_signal = signal
+        
+        for model_id in model_chain:
+            if model_id not in self.models:
+                continue
+            
+            model = self.models[model_id]
+            response = model.process(current_signal)
+            
+            if response is None:
+                return None
+            
+            current_signal = response
+        
+        return current_signal
+    
+    def broadcast(self, data: np.ndarray, modality: SensorModality):
+        """Broadcast data to entire consciousness stream."""
+        signal = Signal(
+            signal_id=f"broadcast-{time.time()}",
+            signal_type=SignalType.BROADCAST,
+            modality=modality,
+            source="consciousness",
+            timestamp=time.time(),
+            data=data,
+        )
+        self.spine.broadcast(signal)
+    
+    def focus_attention(self, model_id: str):
+        """Focus attention on a specific model."""
+        self.attention_focus = model_id
+        
+        # Boost attention weights for focused model
+        if model_id in self.models:
+            model = self.models[model_id]
+            for mod in model.input_panel.modalities:
+                model.input_panel.update_attention(mod, 0.5)
+    
+    def get_state(self) -> Dict:
+        """Get consciousness stream state."""
+        return {
+            'models': list(self.models.keys()),
+            'spine': self.spine.get_consciousness_state(),
+            'attention_focus': self.attention_focus,
+            'parallel_enabled': self.parallel_enabled,
+            'model_states': {
+                mid: {
+                    'active': m.active,
+                    'processing': m.processing,
+                    'input_signals': m.input_panel.signals_received,
+                    'output_signals': m.output_panel.signals_sent,
+                }
+                for mid, m in self.models.items()
+            },
+        }
+
+
+# =============================================================================
+# FACTORY FUNCTIONS
+# =============================================================================
+
+def create_sensorized_model(model_id: str, category: str,
+                            inference_fn: Callable = None) -> SensorizedModel:
+    """
+    Create a sensorized model with appropriate modalities.
+    """
+    # Default modalities by category
+    modality_map = {
+        'reasoning': ([SensorModality.TEXT, SensorModality.EMBEDDING], 
+                      [SensorModality.TEXT, SensorModality.EMBEDDING]),
+        'math': ([SensorModality.TEXT, SensorModality.NUMERIC],
+                 [SensorModality.TEXT, SensorModality.NUMERIC]),
+        'code': ([SensorModality.TEXT], [SensorModality.TEXT]),
+        'vision': ([SensorModality.VISION, SensorModality.EMBEDDING],
+                   [SensorModality.TEXT, SensorModality.EMBEDDING]),
+        'audio': ([SensorModality.AUDIO], [SensorModality.TEXT]),
+        'spatial': ([SensorModality.SPATIAL, SensorModality.VISION],
+                    [SensorModality.SPATIAL, SensorModality.TEXT]),
+        'general': ([SensorModality.TEXT, SensorModality.EMBEDDING],
+                    [SensorModality.TEXT, SensorModality.EMBEDDING]),
+    }
+    
+    input_mod, output_mod = modality_map.get(
+        category, 
+        ([SensorModality.RAW], [SensorModality.RAW])
+    )
+    
+    return SensorizedModel(
+        model_id=model_id,
+        category=category,
+        input_modalities=input_mod,
+        output_modalities=output_mod,
+        process_fn=inference_fn,
+    )
+
+
+def create_default_stream() -> ConsciousnessStream:
+    """
+    Create consciousness stream with default Harmonic Stack models.
+    """
+    stream = ConsciousnessStream()
+    
+    # Add default models
+    default_models = [
+        ('reasoning', 'reasoning'),
+        ('math', 'math'),
+        ('code', 'code'),
+        ('vision', 'vision'),
+        ('spatial', 'spatial'),
+        ('general', 'general'),
+    ]
+    
+    for model_id, category in default_models:
+        model = create_sensorized_model(model_id, category)
+        stream.add_model(model)
+    
+    return stream
+
+
+# =============================================================================
+# MAIN
+# =============================================================================
+
+def main():
+    print("=" * 70)
+    print("SENSOR PANEL ARCHITECTURE")
+    print("Multidimensional Consciousness Stream")
+    print("Ghost in the Machine Labs")
+    print("=" * 70)
+    
+    # Create consciousness stream
+    stream = create_default_stream()
+    
+    print("\nConsciousness stream initialized")
+    state = stream.get_state()
+    print(f"  Models: {state['models']}")
+    print(f"  Panels on spine: {state['spine']['panels']}")
+    print(f"  Modalities: {state['spine']['modalities']}")
+    
+    # Test parallel processing
+    print("\n--- Parallel Processing Test ---")
+    
+    test_signal = Signal(
+        signal_id="test-001",
+        signal_type=SignalType.QUERY,
+        modality=SensorModality.TEXT,
+        source="user",
+        timestamp=time.time(),
+        data=np.random.randn(64).astype(np.float32),
+    )
+    
+    responses = stream.process_parallel(test_signal)
+    print(f"  Input signal: {test_signal.signal_id}")
+    print(f"  Responses received: {len(responses)}")
+    for r in responses:
+        print(f"    - {r.source}: {r.data.shape}")
+    
+    # Test serial processing
+    print("\n--- Serial Processing Test ---")
+    
+    chain = ['reasoning', 'code', 'general']
+    result = stream.process_serial(test_signal, chain)
+    print(f"  Chain: {' → '.join(chain)}")
+    if result:
+        print(f"  Final output: {result.data.shape}")
+    
+    # Test attention focus
+    print("\n--- Attention Focus Test ---")
+    stream.focus_attention('reasoning')
+    print(f"  Focused on: {stream.attention_focus}")
+    
+    # Final state
+    print("\n--- Final State ---")
+    state = stream.get_state()
+    for model_id, mstate in state['model_states'].items():
+        print(f"  {model_id}: in={mstate['input_signals']}, out={mstate['output_signals']}")
+    
+    print("\n" + "=" * 70)
+    print("CONSCIOUSNESS STREAM ACTIVE")
+    print("=" * 70)
+    
+    return stream
+
+
+if __name__ == "__main__":
+    main()