parallel_core/sensor_panels.py

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