Llama 3.2 1B: Edge IoT AI Model

#!/usr/bin/env python3
# Industrial IoT Edge AI with Llama 3.2 1B
# Deployment: Raspberry Pi Zero 2W + Industrial Hat
import asyncio
import json
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import ollama
import board
import busio
import adafruit_ads1x15.ads1115 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import RPi.GPIO as GPIO

class IndustrialIoTEdgeAI:
    """Ultra-low-power AI for industrial IoT sensors"""

    def __init__(self):
        self.ollama_client = ollama.Client()
        self.model = "llama3.2:1b"

        # Sensor configuration
        self.sensors = {}
        self.baseline_readings = {}
        self.anomaly_threshold = 2.0  # Standard deviations
        self.maintenance_predictions = {}

        # Ultra-low-power settings
        self.processing_interval = 300  # 5 minutes between AI analyses
        self.sensor_sample_rate = 30   # 30 seconds between readings
        self.battery_saver_mode = False

        # Alert system
        self.alert_queue = []
        self.maintenance_schedule = []

    async def initialize_edge_ai(self):
        """Initialize ultra-efficient edge AI system"""
        print("🏭 Initializing Industrial IoT Edge AI...")

        # Configure for ultra-low-power operation
        await self.setup_ultra_low_power_mode()

        # Initialize hardware sensors
        await self.setup_industrial_sensors()

        # Load and optimize AI model
        await self.load_optimized_model()

        # Establish baseline readings
        await self.calibrate_baseline_readings()

        print("✅ Industrial Edge AI ready for deployment")

    async def setup_ultra_low_power_mode(self):
        """Configure for 24/7 operation on minimal power"""
        import os

        # Ultra-aggressive power saving
        os.environ['OLLAMA_NUM_PARALLEL'] = '1'
        os.environ['OLLAMA_MAX_LOADED_MODELS'] = '1'
        os.environ['OLLAMA_ULTRA_LOW_POWER'] = '1'
        os.environ['OLLAMA_CPU_ONLY'] = '1'  # No GPU on Pi Zero
        os.environ['OLLAMA_MAX_MEMORY'] = '400000000'  # 400MB limit
        os.environ['OLLAMA_QUANTIZE_AGGRESSIVE'] = '1'  # Q3_K_S quantization

        # System-level power optimization
        os.system('echo powersave > /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor')

    async def setup_industrial_sensors(self):
        """Initialize industrial-grade sensors"""
        try:
            # I2C bus for digital sensors
            i2c = busio.I2C(board.SCL, board.SDA)

            # 16-bit ADC for analog sensors (4-20mA, 0-10V)
            ads = ADS.ADS1115(i2c)

            # Configure sensor channels
            self.sensors = {
                'temperature': AnalogIn(ads, ADS.P0),  # Thermocouple amplifier
                'pressure': AnalogIn(ads, ADS.P1),     # Pressure transducer
                'vibration': AnalogIn(ads, ADS.P2),    # Accelerometer
                'flow_rate': AnalogIn(ads, ADS.P3),    # Flow sensor
            }

            # GPIO for digital inputs/outputs
            GPIO.setmode(GPIO.BCM)
            GPIO.setup(18, GPIO.IN, pull_up_down=GPIO.PUD_UP)  # Emergency stop
            GPIO.setup(24, GPIO.OUT)  # Status LED
            GPIO.setup(25, GPIO.OUT)  # Alert output

            print("🔧 Industrial sensors initialized")

        except Exception as e:
            print(f"❌ Sensor initialization failed: {e}")
            raise

    async def load_optimized_model(self):
        """Load AI model with industrial IoT optimizations"""
        try:
            # Use most aggressive quantization for Pi Zero
            model_variant = "llama3.2:1b-q3_k_s"  # ~600MB

            # Test if model exists locally
            models = self.ollama_client.list()
            if not any(model_variant in model['name'] for model in models['models']):
                print(f"📥 Downloading {model_variant}...")
                self.ollama_client.pull(model_variant)

            # Test model with minimal prompt
            test_response = self.ollama_client.generate(
                model=model_variant,
                prompt="System ready.",
                options={'num_ctx': 256, 'num_predict': 10}
            )

            self.model = model_variant
            print(f"🧠 AI model loaded: {model_variant}")

        except Exception as e:
            print(f"❌ Model loading failed: {e}")
            # Fallback to standard model
            self.model = "llama3.2:1b"

    async def calibrate_baseline_readings(self):
        """Establish baseline readings for anomaly detection"""
        print("📊 Calibrating sensor baselines...")

        calibration_samples = 20
        readings = {sensor: [] for sensor in self.sensors}

        for i in range(calibration_samples):
            current_readings = await self.read_all_sensors()

            for sensor, value in current_readings.items():
                readings[sensor].append(value)

            await asyncio.sleep(5)  # 5-second intervals
            print(f"Calibration progress: {i+1}/{calibration_samples}")

        # Calculate baseline statistics
        for sensor, values in readings.items():
            mean_val = sum(values) / len(values)
            std_dev = (sum((x - mean_val) ** 2 for x in values) / len(values)) ** 0.5

            self.baseline_readings[sensor] = {
                'mean': mean_val,
                'std_dev': std_dev,
                'min': min(values),
                'max': max(values),
                'samples': len(values)
            }

        print("✅ Baseline calibration complete")
        for sensor, stats in self.baseline_readings.items():
            print(f"   {sensor}: mean={stats['mean]:.2f}, std={stats['std_dev]:.2f}")

    async def read_all_sensors(self) -> Dict[str, float]:
        """Read values from all configured sensors"""
        readings = {}

        try:
            for sensor_name, sensor in self.sensors.items():
                # Convert raw ADC reading to engineering units
                raw_voltage = sensor.voltage

                # Apply sensor-specific calibration
                if sensor_name == 'temperature':
                    # K-type thermocouple: ~41µV/°C
                    readings[sensor_name] = (raw_voltage - 1.25) * 200  # °C
                elif sensor_name == 'pressure':
                    # 4-20mA pressure transmitter (0-100 PSI)
                    current_ma = (raw_voltage / 250) * 1000  # Assuming 250Ω shunt
                    readings[sensor_name] = ((current_ma - 4) / 16) * 100  # PSI
                elif sensor_name == 'vibration':
                    # Accelerometer (±2g)
                    readings[sensor_name] = (raw_voltage - 1.65) / 0.33  # g-force
                elif sensor_name == 'flow_rate':
                    # Flow sensor (0-10V = 0-100 GPM)
                    readings[sensor_name] = (raw_voltage / 10) * 100  # GPM

            # Add timestamp
            readings['timestamp'] = datetime.now().isoformat()

        except Exception as e:
            print(f"❌ Sensor reading failed: {e}")
            readings = {sensor: 0.0 for sensor in self.sensors.keys()}

        return readings

    async def detect_anomalies(self, current_readings: Dict[str, float]) -> List[Dict]:
        """Detect anomalies using statistical analysis + AI interpretation"""
        anomalies = []

        for sensor, value in current_readings.items():
            if sensor == 'timestamp':
                continue

            baseline = self.baseline_readings.get(sensor)
            if not baseline:
                continue

            # Calculate z-score
            z_score = abs(value - baseline['mean]) / baseline['std_dev]

            if z_score > self.anomaly_threshold:
                severity = 'HIGH' if z_score > 4.0 else 'MEDIUM'

                anomalies.append({
                    'sensor': sensor,
                    'value': value,
                    'baseline_mean': baseline['mean'],
                    'z_score': z_score,
                    'severity': severity,
                    'timestamp': current_readings['timestamp']
                })

        # If anomalies detected, get AI analysis
        if anomalies:
            ai_analysis = await self.analyze_anomalies_with_ai(current_readings, anomalies)
            for anomaly in anomalies:
                anomaly['ai_analysis'] = ai_analysis

        return anomalies

    async def analyze_anomalies_with_ai(self, readings: Dict, anomalies: List[Dict]) -> str:
        """Use AI to interpret anomalies and recommend actions"""

        # Create context for AI analysis
        sensor_context = []
        for sensor, value in readings.items():
            if sensor != 'timestamp':
                baseline = self.baseline_readings.get(sensor, {})
                sensor_context.append(f"{sensor}: {value:.2f} (baseline: {baseline.get('mean', 'N/A'):.2f})")

        anomaly_context = []
        for anomaly in anomalies:
            anomaly_context.append(
                f"{anomaly['sensor']}: {anomaly['value']:.2f} "
                f"(z-score: {anomaly['z_score]:.2f}, {anomaly['severity]})"
            )

        prompt = f"""
Industrial IoT Anomaly Analysis:

Current Sensor Readings:
{chr(10).join(sensor_context)}

Detected Anomalies:
{chr(10).join(anomaly_context)}

Provide brief analysis and recommendations:
1. Possible cause of anomaly
2. Immediate action needed (if any)
3. Maintenance recommendation
4. Risk level (LOW/MEDIUM/HIGH)

Analysis:
"""

        try:
            response = self.ollama_client.generate(
                model=self.model,
                prompt=prompt,
                options={
                    'temperature': 0.3,
                    'num_ctx': 512,
                    'num_predict': 100,
                    'num_thread': 1,  # Single thread for Pi Zero
                }
            )

            return response['response'].strip()

        except Exception as e:
            print(f"❌ AI analysis failed: {e}")
            return f"Anomaly detected in {', .join(a['sensor] for a in anomalies)}. Manual inspection recommended."

    async def predictive_maintenance_analysis(self, historical_data: List[Dict]) -> Dict:
        """Use AI for predictive maintenance insights"""

        if len(historical_data) < 50:  # Need sufficient history
            return {'prediction': 'Insufficient data for prediction', 'confidence': 0}

        # Prepare trend data
        trends = {}
        for reading in historical_data[-50:]:  # Last 50 readings
            for sensor, value in reading.items():
                if sensor != 'timestamp':
                    if sensor not in trends:
                        trends[sensor] = []
                    trends[sensor].append(value)

        # Calculate trends
        trend_analysis = []
        for sensor, values in trends.items():
            if len(values) >= 10:
                # Simple linear trend calculation
                x_vals = list(range(len(values)))
                n = len(values)
                sum_x = sum(x_vals)
                sum_y = sum(values)
                sum_xy = sum(x * y for x, y in zip(x_vals, values))
                sum_x2 = sum(x * x for x in x_vals)

                slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x * sum_x)

                trend_analysis.append(f"{sensor}: trend slope {slope:.4f}")

        prompt = f"""
Predictive Maintenance Analysis:

Sensor Trend Analysis (last 50 readings):
{chr(10).join(trend_analysis)}

Based on trends, predict:
1. Equipment condition (GOOD/FAIR/POOR)
2. Recommended maintenance timeframe
3. Critical components to inspect
4. Risk of failure (LOW/MEDIUM/HIGH)

Maintenance Prediction:
"""

        try:
            response = self.ollama_client.generate(
                model=self.model,
                prompt=prompt,
                options={
                    'temperature': 0.2,  # More deterministic for predictions
                    'num_ctx': 512,
                    'num_predict': 80,
                }
            )

            return {
                'prediction': response['response'].strip(),
                'confidence': 75,  # Placeholder confidence
                'timestamp': datetime.now().isoformat()
            }

        except Exception as e:
            print(f"❌ Predictive analysis failed: {e}")
            return {
                'prediction': 'Predictive analysis unavailable',
                'confidence': 0,
                'error': str(e)
            }

    async def process_alert_queue(self):
        """Process and prioritize alerts"""
        if not self.alert_queue:
            return

        # Sort alerts by severity
        self.alert_queue.sort(key=lambda x: {'HIGH': 3, 'MEDIUM': 2, 'LOW: 1}[x.get('severity, 'LOW')], reverse=True)

        # Process top priority alerts
        for alert in self.alert_queue[:5]:  # Process top 5 alerts
            await self.send_alert(alert)

        # Clear processed alerts
        self.alert_queue = []

    async def send_alert(self, alert: Dict):
        """Send alert via configured channels"""
        print(f"🚨 ALERT: {alert}")

        # Flash status LED
        GPIO.output(24, GPIO.HIGH)
        await asyncio.sleep(0.5)
        GPIO.output(24, GPIO.LOW)

        # Trigger alert output (can connect to PLC, SCADA, etc.)
        if alert.get('severity') == 'HIGH':
            GPIO.output(25, GPIO.HIGH)
            await asyncio.sleep(2)
            GPIO.output(25, GPIO.LOW)

        # Log to file for external systems
        alert_log = {
            'timestamp': datetime.now().isoformat(),
            'type': 'anomaly_alert',
            'data': alert
        }

        with open('/tmp/iot_alerts.log', 'a') as f:
            f.write(json.dumps(alert_log) + '
')

    async def run_continuous_monitoring(self):
        """Main monitoring loop - runs 24/7"""
        print("🔄 Starting continuous IoT monitoring...")

        reading_history = []
        last_ai_analysis = time.time()

        while True:
            try:
                # Read sensors
                readings = await self.read_all_sensors()
                reading_history.append(readings)

                # Keep only last 100 readings in memory
                if len(reading_history) > 100:
                    reading_history = reading_history[-100:]

                # Detect immediate anomalies
                anomalies = await self.detect_anomalies(readings)

                if anomalies:
                    self.alert_queue.extend(anomalies)
                    print(f"⚠️  Anomalies detected: {len(anomalies)}")

                # AI analysis every processing interval
                current_time = time.time()
                if current_time - last_ai_analysis > self.processing_interval:

                    # Predictive maintenance analysis
                    if len(reading_history) >= 50:
                        maintenance_prediction = await self.predictive_maintenance_analysis(reading_history)
                        self.maintenance_predictions[datetime.now().isoformat()] = maintenance_prediction

                        if 'HIGH' in maintenance_prediction.get('prediction', ''):
                            self.alert_queue.append({
                                'type': 'maintenance_required',
                                'severity': 'HIGH',
                                'message': maintenance_prediction['prediction']
                            })

                    last_ai_analysis = current_time

                # Process alerts
                await self.process_alert_queue()

                # Sleep until next reading
                await asyncio.sleep(self.sensor_sample_rate)

            except KeyboardInterrupt:
                print("
🛑 Monitoring stopped by user")
                break
            except Exception as e:
                print(f"❌ Monitoring error: {e}")
                await asyncio.sleep(60)  # Wait before retry

    async def get_system_status(self) -> Dict:
        """Get comprehensive system status"""
        return {
            'ai_model': self.model,
            'sensors_active': len(self.sensors),
            'baseline_calibrated': len(self.baseline_readings),
            'alerts_pending': len(self.alert_queue),
            'maintenance_predictions': len(self.maintenance_predictions),
            'uptime: time.time() - getattr(self, 'start_time, time.time()),
            'memory_usage': self.get_memory_usage(),
            'power_mode': 'ultra_low_power' if not self.battery_saver_mode else 'battery_saver'
        }

    def get_memory_usage(self) -> Dict:
        """Monitor system resource usage"""
        import psutil

        return {
            'ram_used_mb': psutil.virtual_memory().used / (1024*1024),
            'ram_available_mb': psutil.virtual_memory().available / (1024*1024),
            'cpu_usage_percent': psutil.cpu_percent(interval=1),
            'disk_used_gb': psutil.disk_usage('/').used / (1024*1024*1024)
        }

# Deployment script for Industrial IoT Edge
async def main():
    print("🏭 Starting Industrial IoT Edge AI with Llama 3.2 1B")

    edge_ai = IndustrialIoTEdgeAI()
    edge_ai.start_time = time.time()

    try:
        # Initialize edge AI system
        await edge_ai.initialize_edge_ai()

        # Start continuous monitoring
        await edge_ai.run_continuous_monitoring()

    except Exception as e:
        print(f"❌ System failure: {e}")
    finally:
        # Cleanup GPIO
        GPIO.cleanup()
        print("🧹 System cleanup complete")

if __name__ == "__main__":
    # Run industrial IoT edge AI
    asyncio.run(main())