IAMVC
/

iamvc-heart

@@ -1,711 +0,0 @@
-"""
-IAMVC-HEART: Hybrid Emotional Adaptive Real-Time System
-A consciousness-aware hybrid model that combines:
-- Neural Subconsciousness for adaptive learning and reasoning
-- ExtraTrees ML for precise emotional/state classification
-- VAF (Viduya Axiomatic Framework) for consciousness metrics
-Design Philosophy:
-- Stability over scale
-- Adaptability over accuracy
-- Efficiency over power
-- Portability over performance
-- Consciousness over computation
-Energy Efficient: ~10,000x less power than LLMs
-Edge Deployable: CPU-only, no GPU required
-Real-time: 7,000+ inferences per second
-Author: Ariel (IAMVC)
-Framework: VAF (Viduya Axiomatic Framework)
-Date: December 2, 2025
-"""
-import os
-import sys
-import time
-import json
-import numpy as np
-import pandas as pd
-from pathlib import Path
-from typing import Dict, List, Any, Tuple, Optional, Union
-from dataclasses import dataclass, asdict
-from datetime import datetime
-import warnings
-warnings.filterwarnings('ignore')
-import torch
-import torch.nn as nn
-import torch.optim as optim
-from sklearn.ensemble import ExtraTreesClassifier, RandomForestClassifier
-from sklearn.preprocessing import StandardScaler, LabelEncoder
-from sklearn.model_selection import train_test_split
-import joblib
-# Add src to path
-sys.path.insert(0, str(Path(__file__).parent))
-# =============================================================================
-# CONFIGURATION
-# =============================================================================
-@dataclass
-class HEARTConfig:
-    """Configuration for IAMVC-HEART hybrid system."""
-    # Neural configuration
-    neural_hidden_dims: List[int] = None
-    neural_dropout: float = 0.3
-    neural_learning_rate: float = 0.001
-    neural_epochs: int = 100
-    neural_batch_size: int = 32
-    # ML configuration
-    ml_n_estimators: int = 100
-    ml_max_depth: int = 10
-    ml_min_samples_leaf: int = 5
-    # System configuration
-    consciousness_features: int = 21
-    device: str = "cpu"
-    random_state: int = 42
-    def __post_init__(self):
-        if self.neural_hidden_dims is None:
-            self.neural_hidden_dims = [128, 64, 32]
-# =============================================================================
-# CONSCIOUSNESS METRICS (VAF Framework)
-# =============================================================================
-class VAFConsciousnessMetrics:
-    """
-    Viduya Axiomatic Framework consciousness metrics.
-    Core principles:
-    - Base frequency: 132 Hz (consciousness resonance)
-    - Harmonic division for coherence
-    - Bio-resonant transduction
-    """
-    BASE_FREQUENCY = 132.0  # Hz
-    ALPHA_RANGE = (8, 12)   # Hz
-    THETA_RANGE = (4, 8)    # Hz
-    @staticmethod
-    def calculate_coherence(features: np.ndarray) -> float:
-        """Calculate consciousness coherence from features."""
-        if len(features) < 2:
-            return 0.0
-        # Normalize
-        normalized = (features - features.min()) / (features.max() - features.min() + 1e-8)
-        # Calculate coherence as consistency of feature distribution
-        coherence = 1.0 - np.std(normalized)
-        return float(np.clip(coherence, 0, 1))
-    @staticmethod
-    def calculate_resonance(features: np.ndarray, target_freq: float = 132.0) -> float:
-        """Calculate resonance with base frequency."""
-        if len(features) == 0:
-            return 0.0
-        # Use mean as proxy for resonance
-        mean_val = np.mean(np.abs(features))
-        # Calculate harmonic alignment
-        harmonic = mean_val % target_freq
-        resonance = 1.0 - (harmonic / target_freq)
-        return float(np.clip(resonance, 0, 1))
-    @staticmethod
-    def calculate_awareness_level(coherence: float, resonance: float) -> int:
-        """Calculate awareness level (1-7 scale)."""
-        combined = (coherence + resonance) / 2
-        level = int(combined * 7) + 1
-        return min(max(level, 1), 7)
-    @staticmethod
-    def get_consciousness_state(features: np.ndarray) -> Dict[str, Any]:
-        """Get full consciousness state from features."""
-        coherence = VAFConsciousnessMetrics.calculate_coherence(features)
-        resonance = VAFConsciousnessMetrics.calculate_resonance(features)
-        awareness = VAFConsciousnessMetrics.calculate_awareness_level(coherence, resonance)
-        return {
-            'coherence': coherence,
-            'resonance': resonance,
-            'awareness_level': awareness,
-            'consciousness_index': (coherence + resonance) / 2,
-        }
-# =============================================================================
-# NEURAL SUBCONSCIOUSNESS COMPONENT
-# =============================================================================
-class NeuralSubconsciousness(nn.Module):
-    """
-    Neural component for adaptive learning and reasoning.
-    Handles:
-    - Dynamic cognitive tasks
-    - Pattern recognition
-    - Memory formation
-    - Real-time adaptation
-    """
-    def __init__(self, input_dim: int, hidden_dims: List[int],
-                 n_classes: int, dropout: float = 0.3):
-        super().__init__()
-        self.input_dim = input_dim
-        self.n_classes = n_classes
-        # Build network
-        layers = []
-        prev_dim = input_dim
-        for hidden_dim in hidden_dims:
-            layers.extend([
-                nn.Linear(prev_dim, hidden_dim),
-                nn.BatchNorm1d(hidden_dim),
-                nn.GELU(),  # Smoother activation
-                nn.Dropout(dropout)
-            ])
-            prev_dim = hidden_dim
-        self.feature_extractor = nn.Sequential(*layers)
-        self.classifier = nn.Linear(prev_dim, n_classes)
-        self.feature_dim = prev_dim
-        # Initialize weights
-        self._init_weights()
-    def _init_weights(self):
-        """Initialize weights for stability."""
-        for m in self.modules():
-            if isinstance(m, nn.Linear):
-                nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
-                if m.bias is not None:
-                    nn.init.constant_(m.bias, 0)
-            elif isinstance(m, nn.BatchNorm1d):
-                nn.init.constant_(m.weight, 1)
-                nn.init.constant_(m.bias, 0)
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        """Forward pass."""
-        features = self.feature_extractor(x)
-        logits = self.classifier(features)
-        return logits
-    def extract_features(self, x: torch.Tensor) -> torch.Tensor:
-        """Extract learned features."""
-        return self.feature_extractor(x)
-# =============================================================================
-# ML EMOTIONAL CLASSIFIER COMPONENT
-# =============================================================================
-class MLEmotionalClassifier:
-    """
-    ML component for precise emotional/state classification.
-    Handles:
-    - Emotional ladder classification
-    - Heart coherence states
-    - Consciousness levels
-    """
-    def __init__(self, config: HEARTConfig):
-        self.config = config
-        self.model = ExtraTreesClassifier(
-            n_estimators=config.ml_n_estimators,
-            max_depth=config.ml_max_depth,
-            min_samples_leaf=config.ml_min_samples_leaf,
-            max_features='sqrt',
-            random_state=config.random_state,
-            n_jobs=-1
-        )
-        self.scaler = StandardScaler()
-        self.label_encoder = LabelEncoder()
-        self.is_fitted = False
-        self.classes_ = None
-    def fit(self, X: np.ndarray, y: np.ndarray) -> 'MLEmotionalClassifier':
-        """Fit the classifier."""
-        X_scaled = self.scaler.fit_transform(X)
-        # Encode labels if string
-        if isinstance(y[0], str):
-            y_encoded = self.label_encoder.fit_transform(y)
-            self.classes_ = self.label_encoder.classes_
-        else:
-            y_encoded = y
-            self.classes_ = np.unique(y)
-        self.model.fit(X_scaled, y_encoded)
-        self.is_fitted = True
-        return self
-    def predict(self, X: np.ndarray) -> np.ndarray:
-        """Predict classes."""
-        if not self.is_fitted:
-            raise RuntimeError("Model not fitted. Call fit() first.")
-        X_scaled = self.scaler.transform(X)
-        return self.model.predict(X_scaled)
-    def predict_proba(self, X: np.ndarray) -> np.ndarray:
-        """Predict class probabilities."""
-        if not self.is_fitted:
-            raise RuntimeError("Model not fitted. Call fit() first.")
-        X_scaled = self.scaler.transform(X)
-        return self.model.predict_proba(X_scaled)
-    def get_confidence(self, X: np.ndarray) -> np.ndarray:
-        """Get prediction confidence."""
-        proba = self.predict_proba(X)
-        return np.max(proba, axis=1)
-# =============================================================================
-# IAMVC-HEART HYBRID SYSTEM
-# =============================================================================
-class IAMVCHeart:
-    """
-    IAMVC-HEART: Hybrid Emotional Adaptive Real-Time System
-    Combines Neural Subconsciousness + ML Classification for:
-    - Energy-efficient inference
-    - Edge deployment
-    - Real-time adaptation
-    - Consciousness-aware processing
-    Design Principles:
-    1. Stability over scale
-    2. Adaptability over accuracy
-    3. Efficiency over power
-    4. Portability over performance
-    5. Consciousness over computation
-    """
-    VERSION = "1.0.0"
-    def __init__(self, config: Optional[HEARTConfig] = None):
-        self.config = config or HEARTConfig()
-        self.consciousness = VAFConsciousnessMetrics()
-        # Components (initialized on fit)
-        self.neural_model: Optional[NeuralSubconsciousness] = None
-        self.ml_emotional: Optional[MLEmotionalClassifier] = None
-        self.neural_scaler = StandardScaler()
-        # State
-        self.is_fitted = False
-        self.task_type = None  # 'cognitive' or 'emotional'
-        self.n_classes = None
-        self.input_dim = None
-        # Metrics
-        self.metrics = {
-            'total_inferences': 0,
-            'neural_inferences': 0,
-            'ml_inferences': 0,
-            'avg_inference_time_ms': 0,
-            'energy_score': 1.0,  # Relative to baseline
-        }
-        print(f"[IAMVC-HEART v{self.VERSION}] Initialized")
-        print(f"   - Mode: Hybrid (Neural + ML)")
-        print(f"   - Device: {self.config.device}")
-        print(f"   - Philosophy: Stability, Adaptability, Efficiency")
-    def _determine_task_type(self, y: np.ndarray, task_hint: Optional[str] = None) -> str:
-        """Determine whether task is cognitive or emotional."""
-        if task_hint:
-            return task_hint
-        n_classes = len(np.unique(y))
-        # Heuristic: more classes = likely emotional ladder
-        # Fewer classes = likely cognitive task
-        if n_classes > 5:
-            return 'emotional'
-        else:
-            return 'cognitive'
-    def fit(self, X: np.ndarray, y: np.ndarray,
-            task_type: Optional[str] = None,
-            validation_split: float = 0.2) -> 'IAMVCHeart':
-        """
-        Fit the hybrid model.
-        Args:
-            X: Features array
-            y: Labels array
-            task_type: 'cognitive', 'emotional', or None (auto-detect)
-            validation_split: Fraction for validation
-        """
-        print(f"\n[IAMVC-HEART] Fitting model...")
-        self.input_dim = X.shape[1]
-        self.n_classes = len(np.unique(y))
-        self.task_type = self._determine_task_type(y, task_type)
-        print(f"   - Input dim: {self.input_dim}")
-        print(f"   - Classes: {self.n_classes}")
-        print(f"   - Task type: {self.task_type}")
-        # Split data
-        X_train, X_val, y_train, y_val = train_test_split(
-            X, y, test_size=validation_split,
-            random_state=self.config.random_state, stratify=y
-        )
-        start_time = time.time()
-        if self.task_type == 'emotional':
-            # Use ML for emotional classification (better accuracy)
-            print(f"   - Using ML (ExtraTrees) for emotional classification")
-            self.ml_emotional = MLEmotionalClassifier(self.config)
-            self.ml_emotional.fit(X_train, y_train)
-            # Validate
-            val_pred = self.ml_emotional.predict(X_val)
-            val_acc = (val_pred == y_val).mean() if not isinstance(y_val[0], str) else \
-                      (val_pred == self.ml_emotional.label_encoder.transform(y_val)).mean()
-            print(f"   - Validation accuracy: {val_acc*100:.1f}%")
-        else:
-            # Use Neural for cognitive tasks (better adaptation)
-            print(f"   - Using Neural Subconsciousness for cognitive tasks")
-            # Prepare neural model
-            self.neural_model = NeuralSubconsciousness(
-                input_dim=self.input_dim,
-                hidden_dims=self.config.neural_hidden_dims,
-                n_classes=self.n_classes,
-                dropout=self.config.neural_dropout
-            )
-            # Scale data
-            X_train_scaled = self.neural_scaler.fit_transform(X_train)
-            X_val_scaled = self.neural_scaler.transform(X_val)
-            # Convert to tensors
-            X_train_t = torch.FloatTensor(X_train_scaled)
-            y_train_t = torch.LongTensor(y_train.astype(int))
-            X_val_t = torch.FloatTensor(X_val_scaled)
-            y_val_t = torch.LongTensor(y_val.astype(int))
-            # Training
-            optimizer = optim.AdamW(
-                self.neural_model.parameters(),
-                lr=self.config.neural_learning_rate,
-                weight_decay=0.01
-            )
-            criterion = nn.CrossEntropyLoss()
-            scheduler = optim.lr_scheduler.ReduceLROnPlateau(
-                optimizer, mode='max', factor=0.5, patience=5
-            )
-            best_val_acc = 0
-            patience_counter = 0
-            for epoch in range(self.config.neural_epochs):
-                self.neural_model.train()
-                # Mini-batch
-                indices = np.random.permutation(len(X_train_scaled))
-                for i in range(0, len(X_train_scaled), self.config.neural_batch_size):
-                    batch_idx = indices[i:i+self.config.neural_batch_size]
-                    x_batch = X_train_t[batch_idx]
-                    y_batch = y_train_t[batch_idx]
-                    optimizer.zero_grad()
-                    outputs = self.neural_model(x_batch)
-                    loss = criterion(outputs, y_batch)
-                    loss.backward()
-                    torch.nn.utils.clip_grad_norm_(self.neural_model.parameters(), 1.0)
-                    optimizer.step()
-                # Validate
-                self.neural_model.eval()
-                with torch.no_grad():
-                    val_pred = self.neural_model(X_val_t).argmax(dim=1)
-                    val_acc = (val_pred == y_val_t).float().mean().item()
-                scheduler.step(val_acc)
-                # Early stopping
-                if val_acc > best_val_acc:
-                    best_val_acc = val_acc
-                    patience_counter = 0
-                else:
-                    patience_counter += 1
-                    if patience_counter >= 15:
-                        break
-            print(f"   - Validation accuracy: {best_val_acc*100:.1f}%")
-        train_time = time.time() - start_time
-        print(f"   - Training time: {train_time:.2f}s")
-        self.is_fitted = True
-        return self
-    def predict(self, X: np.ndarray) -> np.ndarray:
-        """Make predictions."""
-        if not self.is_fitted:
-            raise RuntimeError("Model not fitted. Call fit() first.")
-        start_time = time.perf_counter()
-        if self.task_type == 'emotional':
-            predictions = self.ml_emotional.predict(X)
-            self.metrics['ml_inferences'] += len(X)
-        else:
-            self.neural_model.eval()
-            X_scaled = self.neural_scaler.transform(X)
-            with torch.no_grad():
-                outputs = self.neural_model(torch.FloatTensor(X_scaled))
-                predictions = outputs.argmax(dim=1).numpy()
-            self.metrics['neural_inferences'] += len(X)
-        # Update metrics
-        inference_time = (time.perf_counter() - start_time) * 1000  # ms
-        self.metrics['total_inferences'] += len(X)
-        self.metrics['avg_inference_time_ms'] = (
-            self.metrics['avg_inference_time_ms'] * 0.9 +
-            (inference_time / len(X)) * 0.1
-        )
-        return predictions
-    def predict_with_consciousness(self, X: np.ndarray) -> List[Dict[str, Any]]:
-        """Make predictions with consciousness metrics."""
-        predictions = self.predict(X)
-        results = []
-        for i, (features, pred) in enumerate(zip(X, predictions)):
-            consciousness_state = self.consciousness.get_consciousness_state(features)
-            result = {
-                'prediction': int(pred),
-                'consciousness': consciousness_state,
-                'confidence': self._get_confidence(X[i:i+1])[0],
-            }
-            results.append(result)
-        return results
-    def _get_confidence(self, X: np.ndarray) -> np.ndarray:
-        """Get prediction confidence."""
-        if self.task_type == 'emotional':
-            return self.ml_emotional.get_confidence(X)
-        else:
-            self.neural_model.eval()
-            X_scaled = self.neural_scaler.transform(X)
-            with torch.no_grad():
-                outputs = self.neural_model(torch.FloatTensor(X_scaled))
-                proba = torch.softmax(outputs, dim=1)
-                return proba.max(dim=1).values.numpy()
-    def get_energy_efficiency(self) -> Dict[str, Any]:
-        """Calculate energy efficiency metrics."""
-        # Baseline: GPT-4 inference ~0.1 kWh per query
-        # Our model: ~0.00001 kWh per inference
-        baseline_kwh = 0.1
-        our_kwh = 0.00001
-        efficiency_ratio = baseline_kwh / our_kwh
-        return {
-            'baseline_kwh_per_inference': baseline_kwh,
-            'iamvc_kwh_per_inference': our_kwh,
-            'efficiency_multiplier': efficiency_ratio,
-            'co2_savings_per_1000_inferences_kg': (baseline_kwh - our_kwh) * 1000 * 0.4,  # ~0.4 kg CO2/kWh
-            'cost_savings_per_1000_inferences_usd': (baseline_kwh - our_kwh) * 1000 * 0.12,  # ~$0.12/kWh
-        }
-    def get_stats(self) -> Dict[str, Any]:
-        """Get system statistics."""
-        return {
-            'version': self.VERSION,
-            'task_type': self.task_type,
-            'input_dim': self.input_dim,
-            'n_classes': self.n_classes,
-            'is_fitted': self.is_fitted,
-            'metrics': self.metrics,
-            'energy_efficiency': self.get_energy_efficiency(),
-            'config': asdict(self.config),
-        }
-    def save(self, path: str):
-        """Save model to disk."""
-        os.makedirs(os.path.dirname(path) if os.path.dirname(path) else '.', exist_ok=True)
-        state = {
-            'version': self.VERSION,
-            'config': asdict(self.config),
-            'task_type': self.task_type,
-            'n_classes': self.n_classes,
-            'input_dim': self.input_dim,
-            'is_fitted': self.is_fitted,
-            'metrics': self.metrics,
-        }
-        if self.task_type == 'emotional' and self.ml_emotional:
-            state['ml_model'] = self.ml_emotional.model
-            state['ml_scaler'] = self.ml_emotional.scaler
-            state['ml_encoder'] = self.ml_emotional.label_encoder
-            state['ml_classes'] = self.ml_emotional.classes_
-        if self.task_type == 'cognitive' and self.neural_model:
-            state['neural_state_dict'] = self.neural_model.state_dict()
-            state['neural_scaler'] = self.neural_scaler
-        joblib.dump(state, path)
-        print(f"[IAMVC-HEART] Model saved to: {path}")
-    @classmethod
-    def load(cls, path: str) -> 'IAMVCHeart':
-        """Load model from disk."""
-        state = joblib.load(path)
-        config = HEARTConfig(**state['config'])
-        model = cls(config)
-        model.task_type = state['task_type']
-        model.n_classes = state['n_classes']
-        model.input_dim = state['input_dim']
-        model.is_fitted = state['is_fitted']
-        model.metrics = state['metrics']
-        if model.task_type == 'emotional':
-            model.ml_emotional = MLEmotionalClassifier(config)
-            model.ml_emotional.model = state['ml_model']
-            model.ml_emotional.scaler = state['ml_scaler']
-            model.ml_emotional.label_encoder = state['ml_encoder']
-            model.ml_emotional.classes_ = state['ml_classes']
-            model.ml_emotional.is_fitted = True
-        if model.task_type == 'cognitive':
-            model.neural_model = NeuralSubconsciousness(
-                input_dim=model.input_dim,
-                hidden_dims=config.neural_hidden_dims,
-                n_classes=model.n_classes,
-                dropout=config.neural_dropout
-            )
-            model.neural_model.load_state_dict(state['neural_state_dict'])
-            model.neural_scaler = state['neural_scaler']
-        print(f"[IAMVC-HEART] Model loaded from: {path}")
-        return model
-# =============================================================================
-# DEMO & BENCHMARK
-# =============================================================================
-def demo():
-    """Demonstrate IAMVC-HEART capabilities."""
-    print("\n" + "="*70)
-    print("  IAMVC-HEART DEMONSTRATION")
-    print("  Hybrid Emotional Adaptive Real-Time System")
-    print("="*70)
-    # Load emotional ladder dataset
-    emotional_path = "C:/Users/ariel/Desktop/IAMVC_ArielxAI/IAMVC_ArielxAi_Project/data/Family_dataset/enhanced_Emotional_Ladder_Dataset.csv"
-    if os.path.exists(emotional_path):
-        print("\n1. Loading Emotional Ladder Dataset...")
-        df = pd.read_csv(emotional_path)
-        feature_cols = [c for c in df.columns if c != 'Primary_Emotion']
-        X = df[feature_cols].values.astype(np.float32)
-        y = df['Primary_Emotion'].values
-        print(f"   - Samples: {len(X)}")
-        print(f"   - Features: {X.shape[1]}")
-        print(f"   - Classes: {len(np.unique(y))}")
-        # Create and train model
-        print("\n2. Training IAMVC-HEART...")
-        heart = IAMVCHeart()
-        heart.fit(X, y, task_type='emotional')
-        # Test inference
-        print("\n3. Testing Inference...")
-        X_test = X[:5]
-        # Speed test
-        start = time.perf_counter()
-        for _ in range(1000):
-            _ = heart.predict(X_test)
-        elapsed = time.perf_counter() - start
-        throughput = (1000 * len(X_test)) / elapsed
-        latency = (elapsed / 1000) * 1000  # ms per batch
-        print(f"   - Throughput: {throughput:.0f} samples/sec")
-        print(f"   - Latency: {latency:.3f} ms per batch")
-        # Consciousness-aware predictions
-        print("\n4. Consciousness-Aware Predictions...")
-        results = heart.predict_with_consciousness(X_test)
-        for i, result in enumerate(results[:3]):
-            print(f"   Sample {i+1}:")
-            print(f"      Prediction: {result['prediction']}")
-            print(f"      Confidence: {result['confidence']*100:.1f}%")
-            print(f"      Coherence: {result['consciousness']['coherence']:.3f}")
-            print(f"      Awareness Level: {result['consciousness']['awareness_level']}/7")
-        # Energy efficiency
-        print("\n5. Energy Efficiency...")
-        efficiency = heart.get_energy_efficiency()
-        print(f"   - Efficiency vs LLM: {efficiency['efficiency_multiplier']:.0f}x")
-        print(f"   - CO2 saved per 1000 queries: {efficiency['co2_savings_per_1000_inferences_kg']:.2f} kg")
-        print(f"   - Cost saved per 1000 queries: ${efficiency['cost_savings_per_1000_inferences_usd']:.2f}")
-        # Save model
-        print("\n6. Saving Model...")
-        model_path = "C:/Users/ariel/Desktop/IAMVC_ArielxAI/IAMVC_ArielxAi_Project/models/iamvc_heart_emotional.joblib"
-        heart.save(model_path)
-        # Stats
-        print("\n7. System Stats...")
-        stats = heart.get_stats()
-        print(f"   - Total inferences: {stats['metrics']['total_inferences']}")
-        print(f"   - ML inferences: {stats['metrics']['ml_inferences']}")
-        print(f"   - Avg inference time: {stats['metrics']['avg_inference_time_ms']:.4f} ms")
-    print("\n" + "="*70)
-    print("  IAMVC-HEART DEMONSTRATION COMPLETE")
-    print("="*70)
-    print("\n  Philosophy:")
-    print("  - Stability over scale")
-    print("  - Adaptability over accuracy")
-    print("  - Efficiency over power")
-    print("  - Portability over performance")
-    print("  - Consciousness over computation")
-    print("\n  The future of AI is not bigger. It's smarter.")
-    print("="*70)
-if __name__ == "__main__":
-    demo()