Automate Predictive Maintenance 🚀

from datetime import datetime, timedelta

class SensorCalibrationManager:
    def __init__(self):
        self.calibration_db = {}  # sensor_id -> {date, baseline_offset}
    
    def adjust_reading(self, sensor_id: str, raw_value: float) -> float:
        """Adjust sensor reading for calibration drift"""
        calibration = self.calibration_db.get(sensor_id)
        
        if not calibration:
            return raw_value
        
        # Calculate drift based on time since last calibration
        days_since_cal = (datetime.now() - calibration['date']).days
        
        # Assume 0.5% drift per month for vibration sensors
        drift_factor = 1 + (0.005 * days_since_cal / 30)
        
        # Adjust reading
        adjusted = raw_value / drift_factor
        
        # Flag if calibration is overdue (>6 months)
        if days_since_cal > 180:
            self.flag_calibration_needed(sensor_id)
        
        return adjusted
    
    def update_calibration(self, sensor_id: str, baseline_offset: float):
        """Record new calibration after maintenance"""
        self.calibration_db[sensor_id] = {
            'date': datetime.now(),
            'baseline_offset': baseline_offset
        }

# Usage in monitoring pipeline
cal_manager = SensorCalibrationManager()

raw_vibration = 8.2  # mm/s from sensor
adjusted_vibration = cal_manager.adjust_reading('VIB-047-01', raw_vibration)

# Use adjusted value for analysis
sensor_data['vibration'] = adjusted_vibration

interface ProductionOrder {
  order_id: string;
  machine_id: string;
  value: number;
  due_date: Date;
  completion_percent: number;
}

class MaintenanceScheduleOptimizer {
  async findOptimalMaintenanceWindow(
    machineId: string,
    urgency: 'critical' | 'high' | 'medium',
    estimatedDuration: number
  ): Promise<Date> {
    // Get active orders on this machine
    const orders = await this.getActiveOrders(machineId);
    
    // Calculate revenue at risk for each potential window
    const windows = this.generatePotentialWindows(urgency);
    const scored = windows.map(window => {
      const impactedOrders = orders.filter(order => 
        this.isOrderImpacted(order, window, estimatedDuration)
      );
      
      const revenueAtRisk = impactedOrders.reduce((sum, order) => 
        sum + (order.value * (1 - order.completion_percent)), 0
      );
      
      const lateDeliveryPenalty = impactedOrders
        .filter(order => this.wouldBeLate(order, window, estimatedDuration))
        .reduce((sum, order) => sum + order.value * 0.1, 0); // 10% penalty
      
      return {
        window,
        revenueAtRisk,
        lateDeliveryPenalty,
        totalCost: revenueAtRisk + lateDeliveryPenalty
      };
    });
    
    // Sort by lowest cost
    scored.sort((a, b) => a.totalCost - b.totalCost);
    
    // If critical, may need to accept some revenue loss
    if (urgency === 'critical' && scored[0].totalCost > 50000) {
      await this.notifyProductionManager({
        machineId,
        optimalWindow: scored[0].window,
        estimatedCost: scored[0].totalCost,
        reason: 'Critical failure prevention'
      });
    }
    
    return scored[0].window;
  }
  
  private generatePotentialWindows(urgency: string): Date[] {
    const now = new Date();
    const windows: Date[] = [];
    
    // Critical: next 24 hours
    if (urgency === 'critical') {
      for (let i = 0; i < 24; i += 2) {
        windows.push(new Date(now.getTime() + i * 3600000));
      }
    } else {
      // High/Medium: next 7 days, prefer nights/weekends
      for (let day = 0; day < 7; day++) {
        const date = new Date(now.getTime() + day * 86400000);
        // Add night shift (11pm)
        date.setHours(23, 0, 0, 0);
        windows.push(new Date(date));
        
        // Add weekend days
        if (date.getDay() === 0 || date.getDay() === 6) {
          date.setHours(8, 0, 0, 0);
          windows.push(new Date(date));
        }
      }
    }
    
    return windows;
  }
}

// Usage
const optimizer = new MaintenanceScheduleOptimizer();
const optimalTime = await optimizer.findOptimalMaintenanceWindow(
  'CNC-047',
  'high',
  4  // 4 hour maintenance window
);

console.log(`Schedule maintenance for: ${optimalTime}`);

import requests
from datetime import datetime

class EnvironmentalNormalizer:
    def __init__(self, weather_api_key: str, factory_location: dict):
        self.api_key = weather_api_key
        self.location = factory_location
        self.baseline_temp = 20  # Celsius
        self.baseline_humidity = 50  # %
    
    def get_current_conditions(self) -> dict:
        """Get current weather from OpenWeather API"""
        response = requests.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={
                'lat': self.location['lat'],
                'lon': self.location['lon'],
                'appid': self.api_key,
                'units': 'metric'
            }
        )
        data = response.json()
        
        return {
            'ambient_temp': data['main']['temp'],
            'humidity': data['main']['humidity'],
            'timestamp': datetime.now()
        }
    
    def normalize_temperature_reading(self, machine_temp: float, ambient_temp: float) -> float:
        """Adjust machine temperature for ambient conditions"""
        # Machine temp typically runs 40-50°C above ambient
        expected_delta = 45
        
        # Calculate what temp would be at baseline ambient
        ambient_delta = ambient_temp - self.baseline_temp
        normalized = machine_temp - ambient_delta
        
        return normalized
    
    def normalize_electrical_reading(self, amperage: float, humidity: float) -> float:
        """Adjust electrical readings for humidity"""
        # High humidity increases resistance, lowers current draw
        # Normalize to baseline 50% humidity
        humidity_factor = 1 + ((humidity - self.baseline_humidity) * 0.002)
        normalized = amperage * humidity_factor
        
        return normalized
    
    def apply_normalizations(self, sensor_data: dict) -> dict:
        """Apply all environmental normalizations"""
        conditions = self.get_current_conditions()
        
        normalized = sensor_data.copy()
        
        # Normalize temperature
        if 'temperature' in sensor_data:
            normalized['temperature'] = self.normalize_temperature_reading(
                sensor_data['temperature'],
                conditions['ambient_temp']
            )
        
        # Normalize electrical readings
        if 'amperage' in sensor_data:
            normalized['amperage'] = self.normalize_electrical_reading(
                sensor_data['amperage'],
                conditions['humidity']
            )
        
        # Store ambient conditions for reference
        normalized['ambient_conditions'] = conditions
        
        return normalized

# Usage
normalizer = EnvironmentalNormalizer(
    weather_api_key='your-key',
    factory_location={'lat': 42.3601, 'lon': -71.0589}  # Boston
)

# Before analysis, normalize readings
raw_sensor_data = {
    'machine_id': 'CNC-047',
    'temperature': 78,  # Celsius
    'amperage': 42,
    'vibration': 8.2
}

normalized_data = normalizer.apply_normalizations(raw_sensor_data)

# Use normalized data for failure prediction
result = analyze_machine(normalized_data)

interface SparePart {
  part_number: string;
  description: string;
  quantity_on_hand: number;
  lead_time_days: number;
  cost: number;
  supplier: string;
}

class SparePartsManager {
  private inventoryApi: string;
  private purchasingApi: string;

  async checkAvailability(requiredParts: string[]): Promise<{
    available: string[];
    needsOrder: string[];
    canMeetTimeline: boolean;
  }> {
    const inventory = await this.getInventoryLevels(requiredParts);
    
    const available: string[] = [];
    const needsOrder: string[] = [];
    
    for (const part of requiredParts) {
      const stock = inventory.find(i => i.part_number === part);
      
      if (!stock || stock.quantity_on_hand === 0) {
        needsOrder.push(part);
      } else {
        available.push(part);
      }
    }
    
    // Check if we can get needed parts in time
    const canMeetTimeline = await this.canDeliverInTime(needsOrder, 48);
    
    return { available, needsOrder, canMeetTimeline };
  }

  async autoOrderParts(
    parts: string[],
    urgency: 'critical' | 'high' | 'medium',
    requiredBy: Date
  ): Promise<{
    orders: any[];
    totalCost: number;
    estimatedDelivery: Date;
  }> {
    const orders = [];
    let totalCost = 0;
    
    for (const partNumber of parts) {
      const partInfo = await this.getPartInfo(partNumber);
      
      // Check multiple suppliers for fastest delivery
      const suppliers = await this.getSuppliers(partNumber);
      const sorted = suppliers.sort((a, b) => {
        // For critical, prioritize speed over cost
        if (urgency === 'critical') {
          return a.lead_time_days - b.lead_time_days;
        }
        // For others, balance cost and speed
        return (a.cost * a.lead_time_days) - (b.cost * b.lead_time_days);
      });
      
      const bestSupplier = sorted[0];
      
      // Place order
      const order = await fetch(this.purchasingApi, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          part_number: partNumber,
          quantity: 1,
          supplier: bestSupplier.supplier,
          urgency: urgency,
          required_by: requiredBy.toISOString(),
          reason: 'Predictive maintenance - equipment failure prevention',
          expedite: urgency === 'critical'
        })
      });
      
      const orderResult = await order.json();
      orders.push(orderResult);
      totalCost += bestSupplier.cost;
    }
    
    // Find latest delivery date
    const deliveryDates = orders.map(o => new Date(o.estimated_delivery));
    const estimatedDelivery = new Date(Math.max(...deliveryDates.map(d => d.getTime())));
    
    return { orders, totalCost, estimatedDelivery };
  }

  async getInventoryLevels(partNumbers: string[]): Promise<SparePart[]> {
    const response = await fetch(
      `${this.inventoryApi}/parts?ids=${partNumbers.join(',')}`
    );
    return response.json();
  }

  async canDeliverInTime(parts: string[], hoursNeeded: number): Promise<boolean> {
    if (parts.length === 0) return true;
    
    const inventory = await this.getInventoryLevels(parts);
    const maxLeadTime = Math.max(...inventory.map(p => p.lead_time_days));
    
    return (maxLeadTime * 24) <= hoursNeeded;
  }
}

// Usage in maintenance workflow
const partsManager = new SparePartsManager();

// When failure is predicted
const prediction = {
  machine_id: 'CNC-047',
  time_until_failure_hours: 48,
  required_parts: ['BEARING-6205-2RS', 'SEAL-47-62-8'],
  urgency: 'critical'
};

// Check if parts are available
const availability = await partsManager.checkAvailability(prediction.required_parts);

if (!availability.canMeetTimeline) {
  console.log('WARNING: Parts cannot be delivered in time!');
  console.log('Needed:', availability.needsOrder);
  
  // Escalate to production manager
  await notifyProductionManager({
    issue: 'parts_unavailable',
    machine: prediction.machine_id,
    timeline: prediction.time_until_failure_hours
  });
} else if (availability.needsOrder.length > 0) {
  // Auto-order needed parts
  const requiredBy = new Date(Date.now() + prediction.time_until_failure_hours * 3600000);
  const orderResult = await partsManager.autoOrderParts(
    availability.needsOrder,
    prediction.urgency,
    requiredBy
  );
  
  console.log(`Parts ordered. Total cost: $${orderResult.totalCost}`);
  console.log(`Delivery: ${orderResult.estimatedDelivery}`);
}

import networkx as nx
from typing import Dict, List
import anthropic

class CascadingFailureDetector:
    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
        self.production_graph = self.build_production_graph()
    
    def build_production_graph(self) -> nx.DiGraph:
        """Build directed graph of production line dependencies"""
        G = nx.DiGraph()
        
        # Add machines as nodes
        machines = [
            {'id': 'CNC-047', 'type': 'milling', 'capacity': 100},
            {'id': 'CNC-048', 'type': 'milling', 'capacity': 100},
            {'id': 'LATHE-12', 'type': 'lathe', 'capacity': 80},
            {'id': 'GRIND-05', 'type': 'grinder', 'capacity': 120},
        ]
        
        for machine in machines:
            G.add_node(machine['id'], **machine)
        
        # Add edges for material flow
        G.add_edge('CNC-047', 'LATHE-12', flow_rate=50)
        G.add_edge('CNC-048', 'LATHE-12', flow_rate=50)
        G.add_edge('LATHE-12', 'GRIND-05', flow_rate=80)
        
        return G
    
    async def analyze_cascade_risk(
        self,
        failing_machine: str,
        time_until_failure: float
    ) -> Dict:
        """Analyze risk of cascading failures"""
        
        # Find downstream machines
        downstream = list(nx.descendants(self.production_graph, failing_machine))
        
        # Calculate load redistribution
        failing_capacity = self.production_graph.nodes[failing_machine]['capacity']
        
        impact_analysis = []
        for machine in downstream:
            # Calculate extra load this machine will handle
            current_load = self.get_current_load(machine)
            extra_load = self.calculate_extra_load(failing_machine, machine)
            new_load = current_load + extra_load
            
            capacity = self.production_graph.nodes[machine]['capacity']
            utilization = (new_load / capacity) * 100
            
            # Get current health status
            health = await self.get_machine_health(machine)
            
            impact_analysis.append({
                'machine_id': machine,
                'current_load': current_load,
                'extra_load': extra_load,
                'new_utilization': utilization,
                'current_health': health,
                'cascade_risk': 'high' if utilization > 90 and health['status'] != 'good' else 'medium'
            })
        
        # Use LLM to analyze complex cascade scenarios
        cascade_prompt = f"""Analyze cascading failure risk in production line.

Failing Machine: {failing_machine}
Time Until Failure: {time_until_failure} hours

Downstream Impact:
{json.dumps(impact_analysis, indent=2)}

Production Line Graph:
{self.graph_to_text(self.production_graph)}

Analyze:
1. Which machines are at highest risk of cascading failure?
2. What's the timeline for each potential cascade?
3. What preventive actions should be taken?
4. What's the total production impact if cascades occur?

Format as JSON with: at_risk_machines[], timeline, preventive_actions[], production_impact"""

        response = await self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=3072,
            messages=[{"role": "user", "content": cascade_prompt}]
        )
        
        cascade_analysis = json.loads(response.content[0].text)
        cascade_analysis['downstream_machines'] = impact_analysis
        
        return cascade_analysis
    
    def calculate_extra_load(self, failing_machine: str, downstream_machine: str) -> float:
        """Calculate extra load on downstream machine when upstream fails"""
        # If there are parallel machines, load redistributes
        upstream_machines = list(self.production_graph.predecessors(downstream_machine))
        
        if failing_machine not in upstream_machines:
            return 0
        
        # Get flow from failing machine
        failing_flow = self.production_graph.edges[failing_machine, downstream_machine]['flow_rate']
        
        # Redistribute to other upstream machines
        active_upstream = [m for m in upstream_machines if m != failing_machine]
        
        if len(active_upstream) == 0:
            return 0  # Downstream will stop
        
        # Distribute load evenly (in reality, would check capacity)
        extra_per_machine = failing_flow / len(active_upstream)
        
        return extra_per_machine
    
    def get_current_load(self, machine_id: str) -> float:
        # Query real-time production data
        pass
    
    async def get_machine_health(self, machine_id: str) -> dict:
        # Get latest health status from monitoring system
        pass
    
    def graph_to_text(self, graph: nx.DiGraph) -> str:
        # Convert graph to text representation for LLM
        lines = []
        for edge in graph.edges(data=True):
            lines.append(f"{edge[0]} -> {edge[1]} (flow: {edge[2]['flow_rate']} units/hr)")
        return '\n'.join(lines)

# Usage
detector = CascadingFailureDetector()

# When a failure is predicted
prediction = {
    'machine_id': 'CNC-047',
    'time_until_failure_hours': 48
}

# Analyze cascade risk
cascade_risk = await detector.analyze_cascade_risk(
    prediction['machine_id'],
    prediction['time_until_failure_hours']
)

if cascade_risk['at_risk_machines']:
    print(f"WARNING: {len(cascade_risk['at_risk_machines'])} machines at risk of cascade failure")
    print(f"Preventive actions: {cascade_risk['preventive_actions']}")
    
    # Alert production manager
    await send_cascade_alert(cascade_risk)