convert-models.py

#!/usr/bin/env python3
"""
Convert JavaScript Rule-Based Models to Neural Networks for Hailo-8
This script converts your HVAC diagnostic models to neural networks
"""

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import json
import os
from typing import Dict, List, Tuple
import onnx
import torch.onnx

class HVACDataGenerator:
    """Generate synthetic training data based on rule-based logic"""
    
    def __init__(self, model_name: str):
        self.model_name = model_name
        self.ranges = self.get_sensor_ranges()
    
    def get_sensor_ranges(self) -> Dict:
        """Define normal and abnormal ranges for sensors"""
        return {
            'temperature': {'min': 32, 'max': 122, 'normal': (68, 76)},
            'humidity': {'min': 0, 'max': 100, 'normal': (30, 60)},
            'pressure': {'min': 10, 'max': 20, 'normal': (14, 15.5)},
            'flow_rate': {'min': 0, 'max': 1000, 'normal': (400, 600)},
            'vibration': {'min': 0, 'max': 1, 'normal': (0, 0.1)},
            'current': {'min': 0, 'max': 30, 'normal': (12, 18)},
            'voltage': {'min': 400, 'max': 500, 'normal': (460, 490)},
            'power_factor': {'min': 0, 'max': 1, 'normal': (0.85, 0.98)}
        }
    
    def generate_samples(self, num_samples: int = 10000) -> Tuple[np.ndarray, np.ndarray]:
        """Generate training samples with labels"""
        X = []
        y = []
        
        for _ in range(num_samples):
            # Generate mixture of normal and abnormal samples
            is_normal = np.random.random() > 0.3  # 70% normal, 30% abnormal
            
            sample = []
            issues = []
            
            for sensor, config in self.ranges.items():
                if is_normal:
                    # Generate within normal range
                    value = np.random.uniform(config['normal'][0], config['normal'][1])
                else:
                    # Sometimes generate abnormal values
                    if np.random.random() > 0.7:
                        # Abnormal value
                        if np.random.random() > 0.5:
                            value = np.random.uniform(config['min'], config['normal'][0])
                        else:
                            value = np.random.uniform(config['normal'][1], config['max'])
                        issues.append(sensor)
                    else:
                        value = np.random.uniform(config['normal'][0], config['normal'][1])
                
                # Normalize to 0-1 range
                normalized = (value - config['min']) / (config['max'] - config['min'])
                sample.append(normalized)
            
            X.append(sample)
            
            # Create labels based on model type
            if self.model_name == 'Aquilo':  # Electrical
                label = self.label_electrical(sample, issues)
            elif self.model_name == 'Boreas':  # Airflow
                label = self.label_airflow(sample, issues)
            elif self.model_name == 'Vulcan':  # Thermal
                label = self.label_thermal(sample, issues)
            elif self.model_name == 'Naiad':  # Fluid
                label = self.label_fluid(sample, issues)
            elif self.model_name == 'Gaia':  # Environmental
                label = self.label_environmental(sample, issues)
            elif self.model_name == 'Zephyrus':  # Air quality
                label = self.label_air_quality(sample, issues)
            else:  # Colossus - master aggregator
                label = self.label_master(sample, issues)
            
            y.append(label)
        
        return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)
    
    def label_electrical(self, sample: List, issues: List) -> List:
        """Generate labels for electrical diagnostics"""
        # [normal, voltage_issue, current_issue, power_factor_issue, critical]
        labels = [0, 0, 0, 0, 0]
        
        if not issues:
            labels[0] = 1  # Normal
        else:
            if 'voltage' in issues:
                labels[1] = 1
            if 'current' in issues:
                labels[2] = 1
            if 'power_factor' in issues:
                labels[3] = 1
            if len(issues) > 2:
                labels[4] = 1  # Critical
        
        return labels
    
    def label_airflow(self, sample: List, issues: List) -> List:
        """Generate labels for airflow diagnostics"""
        # [normal, low_flow, high_flow, obstruction, maintenance_needed]
        labels = [0, 0, 0, 0, 0]
        
        if not issues:
            labels[0] = 1
        else:
            if 'flow_rate' in issues:
                if sample[3] < 0.4:  # Low flow
                    labels[1] = 1
                else:
                    labels[2] = 1  # High flow
            if 'pressure' in issues:
                labels[3] = 1  # Possible obstruction
            if 'vibration' in issues:
                labels[4] = 1  # Maintenance needed
        
        return labels
    
    def label_thermal(self, sample: List, issues: List) -> List:
        """Generate labels for thermal diagnostics"""
        # [normal, overheating, overcooling, efficiency_loss, critical]
        labels = [0, 0, 0, 0, 0]
        
        if not issues:
            labels[0] = 1
        else:
            if 'temperature' in issues:
                if sample[0] > 0.7:
                    labels[1] = 1  # Overheating
                else:
                    labels[2] = 1  # Overcooling
            if 'power_factor' in issues or 'current' in issues:
                labels[3] = 1  # Efficiency loss
            if len(issues) > 2 or sample[0] > 0.9:
                labels[4] = 1  # Critical
        
        return labels
    
    def label_fluid(self, sample: List, issues: List) -> List:
        """Generate labels for fluid dynamics"""
        return [1, 0, 0, 0, 0] if not issues else [0, 1, 0, 0, 0]
    
    def label_environmental(self, sample: List, issues: List) -> List:
        """Generate labels for environmental monitoring"""
        return [1, 0, 0, 0, 0] if not issues else [0, 0, 1, 0, 0]
    
    def label_air_quality(self, sample: List, issues: List) -> List:
        """Generate labels for air quality"""
        return [1, 0, 0, 0, 0] if not issues else [0, 0, 0, 1, 0]
    
    def label_master(self, sample: List, issues: List) -> List:
        """Generate labels for master aggregator"""
        # [health_score, maintenance_urgency, system_efficiency, failure_risk, action_required]
        health = 1.0 - (len(issues) * 0.2)
        urgency = min(1.0, len(issues) * 0.3)
        efficiency = 1.0 if not issues else 0.7 - (len(issues) * 0.1)
        risk = min(1.0, len(issues) * 0.25)
        action = 1.0 if len(issues) > 1 else 0.0
        
        return [health, urgency, efficiency, risk, action]


class HVACNeuralNetwork(nn.Module):
    """Neural network for HVAC diagnostics"""
    
    def __init__(self, input_size: int = 8, output_size: int = 5, hidden_sizes: List[int] = [64, 32, 16]):
        super().__init__()
        
        layers = []
        prev_size = input_size
        
        for hidden_size in hidden_sizes:
            layers.extend([
                nn.Linear(prev_size, hidden_size),
                nn.BatchNorm1d(hidden_size),
                nn.ReLU(),
                nn.Dropout(0.2)
            ])
            prev_size = hidden_size
        
        layers.append(nn.Linear(prev_size, output_size))
        layers.append(nn.Sigmoid())  # For multi-label classification
        
        self.model = nn.Sequential(*layers)
    
    def forward(self, x):
        return self.model(x)


def train_model(model_name: str, epochs: int = 100, batch_size: int = 32):
    """Train a neural network for a specific diagnostic model"""
    
    print(f"\n{'='*60}")
    print(f"Training Neural Network for {model_name}")
    print(f"{'='*60}")
    
    # Generate training data
    generator = HVACDataGenerator(model_name)
    X_train, y_train = generator.generate_samples(10000)
    X_val, y_val = generator.generate_samples(2000)
    
    # Create datasets
    train_dataset = torch.utils.data.TensorDataset(
        torch.FloatTensor(X_train),
        torch.FloatTensor(y_train)
    )
    val_dataset = torch.utils.data.TensorDataset(
        torch.FloatTensor(X_val),
        torch.FloatTensor(y_val)
    )
    
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size)
    
    # Create model
    model = HVACNeuralNetwork(input_size=8, output_size=y_train.shape[1])
    criterion = nn.BCELoss()  # Binary cross-entropy for multi-label
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=10)
    
    # Training loop
    best_val_loss = float('inf')
    
    for epoch in range(epochs):
        # Training
        model.train()
        train_loss = 0
        for X_batch, y_batch in train_loader:
            optimizer.zero_grad()
            outputs = model(X_batch)
            loss = criterion(outputs, y_batch)
            loss.backward()
            optimizer.step()
            train_loss += loss.item()
        
        # Validation
        model.eval()
        val_loss = 0
        with torch.no_grad():
            for X_batch, y_batch in val_loader:
                outputs = model(X_batch)
                loss = criterion(outputs, y_batch)
                val_loss += loss.item()
        
        avg_train_loss = train_loss / len(train_loader)
        avg_val_loss = val_loss / len(val_loader)
        
        scheduler.step(avg_val_loss)
        
        if avg_val_loss < best_val_loss:
            best_val_loss = avg_val_loss
            # Save best model
            torch.save(model.state_dict(), f'/home/Automata/mydata/neural-nexus/models/{model_name.lower()}_best.pth')
        
        if epoch % 10 == 0:
            print(f"Epoch {epoch}/{epochs} - Train Loss: {avg_train_loss:.4f}, Val Loss: {avg_val_loss:.4f}")
    
    print(f"✓ Training complete for {model_name}")
    print(f"  Best validation loss: {best_val_loss:.4f}")
    
    # Export to ONNX
    export_to_onnx(model, model_name)
    
    return model


def export_to_onnx(model: nn.Module, model_name: str):
    """Export PyTorch model to ONNX format"""
    
    model.eval()
    dummy_input = torch.randn(1, 8)
    
    onnx_path = f'/home/Automata/mydata/neural-nexus/models/{model_name.lower()}.onnx'
    
    torch.onnx.export(
        model,
        dummy_input,
        onnx_path,
        export_params=True,
        opset_version=12,
        do_constant_folding=True,
        input_names=['sensor_input'],
        output_names=['diagnostic_output'],
        dynamic_axes={
            'sensor_input': {0: 'batch_size'},
            'diagnostic_output': {0: 'batch_size'}
        }
    )
    
    print(f"✓ Exported to ONNX: {onnx_path}")
    
    # Verify ONNX model
    onnx_model = onnx.load(onnx_path)
    onnx.checker.check_model(onnx_model)
    print(f"✓ ONNX model verified")


def compile_to_hef(model_name: str):
    """Compile ONNX model to Hailo HEF format"""
    
    print(f"\n{'='*60}")
    print(f"Compiling {model_name} to Hailo HEF format")
    print(f"{'='*60}")
    
    # This would use the Hailo Dataflow Compiler
    # For now, we'll create a placeholder script
    
    compile_script = f"""#!/bin/bash
# Hailo Model Compilation Script for {model_name}

ONNX_PATH="/home/Automata/mydata/neural-nexus/models/{model_name.lower()}.onnx"
HEF_PATH="/home/Automata/mydata/neural-nexus/models/{model_name.lower()}.hef"

# Step 1: Parse ONNX model
hailo parse --onnx $ONNX_PATH --name {model_name.lower()} --output-path /tmp/

# Step 2: Optimize for Hailo-8
hailo optimize /tmp/{model_name.lower()}.har --output-path /tmp/

# Step 3: Compile to HEF
hailo compile /tmp/{model_name.lower()}_optimized.har \\
    --device-id hailo8 \\
    --output-path $HEF_PATH

echo "✓ Compiled {model_name} to HEF format: $HEF_PATH"
"""
    
    script_path = f'/home/Automata/mydata/neural-nexus/compile_{model_name.lower()}.sh'
    with open(script_path, 'w') as f:
        f.write(compile_script)
    
    os.chmod(script_path, 0o755)
    print(f"✓ Created compilation script: {script_path}")


def main():
    """Main conversion pipeline"""
    
    print("="*60)
    print("HVAC Diagnostic Models - Neural Network Conversion")
    print("Converting JavaScript rule-based models to Hailo-8 format")
    print("="*60)
    
    # Create models directory
    os.makedirs('/home/Automata/mydata/neural-nexus/models', exist_ok=True)
    
    # List of models to convert
    models = [
        'Aquilo',     # Electrical diagnostics
        'Boreas',     # Airflow analysis
        'Vulcan',     # Thermal analysis
        'Naiad',      # Fluid dynamics
        'Gaia',       # Environmental monitoring
        'Zephyrus',   # Air quality
        'Colossus'    # Master aggregator
    ]
    
    # Train and convert each model
    for model_name in models:
        try:
            # Train neural network
            trained_model = train_model(model_name, epochs=50)
            
            # Create compilation script
            compile_to_hef(model_name)
            
        except Exception as e:
            print(f"✗ Error converting {model_name}: {e}")
    
    print("\n" + "="*60)
    print("CONVERSION COMPLETE")
    print("="*60)
    print("\nNext Steps:")
    print("1. Review the generated ONNX models")
    print("2. Run the compilation scripts to generate HEF files")
    print("3. Test the models with the Hailo-8 accelerator")
    print("\nModels saved to: /home/Automata/mydata/neural-nexus/models/")


if __name__ == "__main__":
    main()