Real-Time Telemetry Streaming API

Authorization: Bearer YOUR_API_KEY

POST /api/v1/streaming/telemetry/start

POST /api/v1/streaming/telemetry/stop

GET /api/v1/streaming/telemetry/status/{stream_id}

PUT /api/v1/streaming/telemetry/configure/{stream_id}

{
  "vehicles": ["vehicle_001", "vehicle_002"],
  "data_types": [
    "engine_diagnostics",
    "fuel_system", 
    "vehicle_dynamics"
  ],
  "stream_config": {
    "frequency": "high",
    "compression": true,
    "buffer_size": 1000,
    "include_metadata": true
  },
  "filters": {
    "min_engine_temp": 80,
    "max_fuel_level": 95,
    "alert_thresholds": {
      "engine_rpm": 4000,
      "coolant_temp": 210
    }
  },
  "delivery_method": "websocket",
  "webhook_url": "https://your-app.com/telemetry-webhook"
}

{
  "frequency": "low|medium|high|custom",
  "custom_interval": 15,
  "compression": true,
  "encryption": true,
  "buffer_size": 1000,
  "batch_processing": {
    "enabled": true,
    "batch_size": 50,
    "max_wait_time": 30
  },
  "quality_settings": {
    "precision": "standard|high", 
    "include_raw_data": false,
    "calculated_metrics": true
  },
  "alert_integration": {
    "real_time_alerts": true,
    "threshold_monitoring": true,
    "predictive_warnings": true
  }
}

{
  "timestamp": "2024-03-15T14:30:00Z",
  "vehicle_id": "vehicle_001",
  "data_type": "engine_diagnostics",
  "sequence": 12345,
  "data": {
    "engine": {
      "rpm": 2100,
      "load_percentage": 45.2,
      "coolant_temp": 185,
      "oil_pressure": 35.8,
      "oil_temp": 210,
      "air_intake_temp": 95,
      "throttle_position": 25.5,
      "fuel_trim_bank1": 2.3,
      "fuel_trim_bank2": 1.8,
      "timing_advance": 15.2
    },
    "exhaust": {
      "egr_valve_position": 12.5,
      "dpf_pressure": 2.1,
      "nox_level": 45.2,
      "exhaust_temp": 650
    },
    "turbo": {
      "boost_pressure": 14.7,
      "turbo_rpm": 85000,
      "wastegate_position": 8.5
    },
    "status_codes": {
      "dtc_count": 0,
      "pending_codes": [],
      "mil_status": false,
      "readiness_status": "ready"
    }
  },
  "quality_indicators": {
    "signal_strength": 95,
    "data_completeness": 100,
    "sensor_health": "good"
  }
}

{
  "timestamp": "2024-03-15T14:30:00Z",
  "vehicle_id": "vehicle_001", 
  "data_type": "fuel_system",
  "sequence": 12346,
  "data": {
    "fuel_level": {
      "percentage": 75.5,
      "gallons": 22.65,
      "liters": 85.7,
      "range_remaining": 425.8
    },
    "consumption": {
      "instant_mpg": 18.5,
      "average_mpg": 16.2,
      "fuel_rate_gph": 2.1,
      "fuel_rate_lph": 7.95
    },
    "system_health": {
      "fuel_pressure": 58.2,
      "fuel_temp": 85,
      "injector_pulse_width": 4.2,
      "fuel_pump_duty": 35.8
    },
    "efficiency_metrics": {
      "efficiency_score": 87.2,
      "eco_driving_score": 92.1,
      "fuel_waste_estimate": 0.15,
      "optimization_potential": 8.5
    }
  }
}

{
  "timestamp": "2024-03-15T14:30:00Z",
  "vehicle_id": "vehicle_001",
  "data_type": "vehicle_dynamics",
  "sequence": 12347,
  "data": {
    "motion": {
      "speed_mph": 35.2,
      "speed_kmh": 56.7,
      "acceleration_x": 0.15,
      "acceleration_y": -0.05,
      "acceleration_z": 9.81,
      "angular_velocity_x": 0.02,
      "angular_velocity_y": 0.01,
      "angular_velocity_z": 0.08
    },
    "braking": {
      "brake_pedal_position": 0,
      "brake_pressure": 0,
      "abs_active": false,
      "brake_temp_front": 145,
      "brake_temp_rear": 135,
      "brake_wear_front": 25.5,
      "brake_wear_rear": 30.2
    },
    "steering": {
      "steering_angle": 5.2,
      "steering_rate": 0.8,
      "power_steering_pressure": 1200,
      "wheel_alignment": "normal"
    },
    "suspension": {
      "ride_height_front": 8.2,
      "ride_height_rear": 8.5,
      "shock_travel_front": 2.1,
      "shock_travel_rear": 1.8,
      "stability_control_active": false
    },
    "tires": {
      "pressure_fl": 32.5,
      "pressure_fr": 32.8,
      "pressure_rl": 35.0,
      "pressure_rr": 35.2,
      "temp_fl": 85,
      "temp_fr": 87,
      "temp_rl": 82,
      "temp_rr": 84,
      "tread_depth_fl": 8.5,
      "tread_depth_fr": 8.3,
      "tread_depth_rl": 7.8,
      "tread_depth_rr": 7.9
    }
  }
}

{
  "timestamp": "2024-03-15T14:30:00Z",
  "vehicle_id": "vehicle_001",
  "data_type": "electrical_system", 
  "sequence": 12348,
  "data": {
    "battery": {
      "voltage": 12.6,
      "current": -15.2,
      "state_of_charge": 85.5,
      "temperature": 68,
      "health_percentage": 92.1,
      "cycles": 245,
      "internal_resistance": 0.008
    },
    "charging_system": {
      "alternator_voltage": 13.8,
      "alternator_current": 45.2,
      "charging_rate": 2.1,
      "belt_tension": "normal",
      "regulator_temp": 125
    },
    "power_distribution": {
      "main_bus_voltage": 12.4,
      "accessory_load": 25.8,
      "lighting_load": 8.5,
      "hvac_load": 15.2,
      "total_load": 89.5
    },
    "starter_system": {
      "starter_voltage": 12.1,
      "starter_current": 0,
      "engagement_count": 8,
      "cranking_time_avg": 1.2
    }
  }
}

class PredictiveMaintenanceEngine {
  constructor(telemetryProcessor) {
    this.processor = telemetryProcessor;
    this.maintenanceModels = new Map();
    this.thresholds = this.initializeThresholds();
  }

  initializeThresholds() {
    return {
      engine_wear: {
        oil_pressure_min: 15,
        coolant_temp_max: 210,
        oil_temp_max: 240,
        vibration_max: 2.5
      },
      brake_wear: {
        pad_thickness_min: 3.0,
        rotor_thickness_min: 28.0,
        brake_temp_max: 300
      },
      tire_wear: {
        pressure_min: 28,
        tread_depth_min: 2.0,
        temp_diff_max: 20
      },
      battery_health: {
        voltage_min: 12.0,
        health_min: 70,
        cycle_count_max: 500
      }
    };
  }

  analyzePredictiveMaintenance(vehicleId, telemetryData) {
    const predictions = {
      engine: this.predictEngineServiceNeeds(vehicleId, telemetryData),
      brakes: this.predictBrakeServiceNeeds(vehicleId, telemetryData),
      tires: this.predictTireServiceNeeds(vehicleId, telemetryData),
      battery: this.predictBatteryReplacement(vehicleId, telemetryData)
    };

    // Generate maintenance schedule
    const schedule = this.generateMaintenanceSchedule(vehicleId, predictions);
    
    return {
      vehicle_id: vehicleId,
      predictions: predictions,
      recommended_schedule: schedule,
      priority_items: this.getPriorityMaintenanceItems(predictions),
      cost_estimate: this.estimateMaintenanceCosts(predictions)
    };
  }

  predictEngineServiceNeeds(vehicleId, telemetryData) {
    const engineData = telemetryData.engine_diagnostics?.engine;
    if (!engineData) return null;

    const oilLife = this.calculateOilLife(vehicleId, engineData);
    const airFilterLife = this.calculateAirFilterLife(vehicleId, engineData);
    const sparkPlugLife = this.calculateSparkPlugLife(vehicleId, engineData);

    return {
      oil_change: {
        miles_remaining: oilLife.miles_remaining,
        days_remaining: oilLife.days_remaining,
        urgency: oilLife.urgency,
        confidence: 0.95
      },
      air_filter: {
        miles_remaining: airFilterLife.miles_remaining,
        days_remaining: airFilterLife.days_remaining,
        urgency: airFilterLife.urgency,
        confidence: 0.85
      },
      spark_plugs: {
        miles_remaining: sparkPlugLife.miles_remaining,
        days_remaining: sparkPlugLife.days_remaining,
        urgency: sparkPlugLife.urgency,
        confidence: 0.80
      }
    };
  }

  calculateOilLife(vehicleId, engineData) {
    const baseInterval = 5000; // miles
    const currentMileage = this.getCurrentMileage(vehicleId);
    const lastOilChange = this.getLastServiceMileage(vehicleId, 'oil_change');
    
    // Adjust based on driving conditions
    let adjustmentFactor = 1.0;
    
    // Harsh conditions reduce interval
    if (engineData.oil_temp > 220) adjustmentFactor *= 0.9;
    if (engineData.coolant_temp > 200) adjustmentFactor *= 0.9;
    if (engineData.load_percentage > 80) adjustmentFactor *= 0.85;
    
    const adjustedInterval = baseInterval * adjustmentFactor;
    const milesSinceService = currentMileage - lastOilChange;
    const milesRemaining = adjustedInterval - milesSinceService;
    
    // Estimate days based on average daily usage
    const avgDailyMiles = this.getAverageDailyMiles(vehicleId);
    const daysRemaining = milesRemaining / avgDailyMiles;
    
    let urgency = 'low';
    if (milesRemaining < 500) urgency = 'high';
    else if (milesRemaining < 1000) urgency = 'medium';
    
    return {
      miles_remaining: Math.max(0, milesRemaining),
      days_remaining: Math.max(0, Math.ceil(daysRemaining)),
      urgency: urgency
    };
  }

  predictBrakeServiceNeeds(vehicleId, dynamicsData) {
    const brakingData = dynamicsData.vehicle_dynamics?.braking;
    if (!brakingData) return null;

    // Brake pad wear prediction
    const frontWear = brakingData.brake_wear_front;
    const rearWear = brakingData.brake_wear_rear;
    
    // Calculate remaining life based on wear rate
    const wearRate = this.calculateBrakeWearRate(vehicleId);
    
    const frontLifeRemaining = (100 - frontWear) / wearRate;
    const rearLifeRemaining = (100 - rearWear) / wearRate;
    
    return {
      front_pads: {
        wear_percentage: frontWear,
        miles_remaining: frontLifeRemaining * this.getAverageDailyMiles(vehicleId),
        days_remaining: Math.ceil(frontLifeRemaining),
        urgency: frontWear > 85 ? 'high' : frontWear > 70 ? 'medium' : 'low'
      },
      rear_pads: {
        wear_percentage: rearWear,
        miles_remaining: rearLifeRemaining * this.getAverageDailyMiles(vehicleId),
        days_remaining: Math.ceil(rearLifeRemaining),
        urgency: rearWear > 85 ? 'high' : rearWear > 70 ? 'medium' : 'low'
      },
      brake_fluid: this.predictBrakeFluidService(vehicleId, brakingData)
    };
  }

  generateMaintenanceSchedule(vehicleId, predictions) {
    const schedule = [];
    
    // Process all predictions and create schedule items
    for (const [category, items] of Object.entries(predictions)) {
      if (!items) continue;
      
      for (const [service, prediction] of Object.entries(items)) {
        if (prediction.urgency && prediction.urgency !== 'low') {
          schedule.push({
            category: category,
            service: service,
            recommended_date: this.calculateRecommendedDate(prediction.days_remaining),
            urgency: prediction.urgency,
            estimated_cost: this.getServiceCost(category, service),
            confidence: prediction.confidence || 0.8
          });
        }
      }
    }
    
    // Sort by urgency and date
    schedule.sort((a, b) => {
      const urgencyOrder = { high: 3, medium: 2, low: 1 };
      if (urgencyOrder[a.urgency] !== urgencyOrder[b.urgency]) {
        return urgencyOrder[b.urgency] - urgencyOrder[a.urgency];
      }
      return new Date(a.recommended_date) - new Date(b.recommended_date);
    });
    
    return schedule;
  }

  // Utility methods
  getCurrentMileage(vehicleId) {
    // Implementation to get current vehicle mileage
    return 45000; // Placeholder
  }

  getLastServiceMileage(vehicleId, serviceType) {
    // Implementation to get last service mileage
    return 42500; // Placeholder
  }

  getAverageDailyMiles(vehicleId) {
    // Implementation to calculate average daily miles
    return 50; // Placeholder
  }

  calculateBrakeWearRate(vehicleId) {
    // Implementation to calculate brake wear rate
    return 2.5; // Placeholder: % per month
  }

  calculateRecommendedDate(daysRemaining) {
    const date = new Date();
    date.setDate(date.getDate() + daysRemaining);
    return date.toISOString().split('T')[0];
  }

  getServiceCost(category, service) {
    const costs = {
      engine: { oil_change: 75, air_filter: 45, spark_plugs: 120 },
      brakes: { front_pads: 250, rear_pads: 200, brake_fluid: 85 },
      tires: { rotation: 50, replacement: 600, balancing: 80 },
      battery: { replacement: 180, testing: 25 }
    };
    
    return costs[category]?.[service] || 100;
  }
}

type FleetHealthMonitor struct {
    vehicles            map[string]*VehicleHealth
    healthThresholds    HealthThresholds
    alertManager        *AlertManager
    trendAnalyzer      *TrendAnalyzer
    mutex              sync.RWMutex
}

type VehicleHealth struct {
    VehicleID          string                 `json:"vehicle_id"`
    LastUpdate         time.Time              `json:"last_update"`
    OverallScore       float64                `json:"overall_score"`
    ComponentScores    map[string]float64     `json:"component_scores"`
    ActiveAlerts       []HealthAlert          `json:"active_alerts"`
    TrendIndicators    map[string]string      `json:"trend_indicators"`
    MaintenanceStatus  MaintenanceStatus      `json:"maintenance_status"`
    HealthHistory      []HealthSnapshot       `json:"health_history"`
}

type HealthThresholds struct {
    Critical    float64 `json:"critical"`    // < 40
    Poor        float64 `json:"poor"`        // 40-60
    Fair        float64 `json:"fair"`        // 60-75
    Good        float64 `json:"good"`        // 75-85
    Excellent   float64 `json:"excellent"`   // 85+
}

type HealthAlert struct {
    Component   string    `json:"component"`
    Severity    string    `json:"severity"`
    Message     string    `json:"message"`
    Timestamp   time.Time `json:"timestamp"`
}

type MaintenanceStatus struct {
    DueCount        int       `json:"due_count"`
    OverdueCount    int       `json:"overdue_count"`
    NextServiceDate time.Time `json:"next_service_date"`
    Priority        string    `json:"priority"`
}

func NewFleetHealthMonitor() *FleetHealthMonitor {
    return &FleetHealthMonitor{
        vehicles: make(map[string]*VehicleHealth),
        healthThresholds: HealthThresholds{
            Critical:  40,
            Poor:      60,
            Fair:      75,
            Good:      85,
            Excellent: 100,
        },
    }
}

func (fhm *FleetHealthMonitor) ProcessTelemetryData(vehicleID string, telemetryData map[string]interface{}) {
    fhm.mutex.Lock()
    defer fhm.mutex.Unlock()
    
    // Get or create vehicle health record
    vh, exists := fhm.vehicles[vehicleID]
    if !exists {
        vh = &VehicleHealth{
            VehicleID:       vehicleID,
            ComponentScores: make(map[string]float64),
            TrendIndicators: make(map[string]string),
            HealthHistory:   make([]HealthSnapshot, 0),
        }
        fhm.vehicles[vehicleID] = vh
    }
    
    // Update component scores
    fhm.updateComponentScores(vh, telemetryData)
    
    // Calculate overall health score
    vh.OverallScore = fhm.calculateOverallHealth(vh.ComponentScores)
    
    // Update trends
    fhm.updateHealthTrends(vh)
    
    // Check for health alerts
    fhm.checkHealthAlerts(vh, telemetryData)
    
    // Update maintenance status
    fhm.updateMaintenanceStatus(vh)
    
    // Record health snapshot
    fhm.recordHealthSnapshot(vh)
    
    vh.LastUpdate = time.Now()
}

func (fhm *FleetHealthMonitor) updateComponentScores(vh *VehicleHealth, telemetryData map[string]interface{}) {
    // Engine health
    if engineData, ok := telemetryData["engine_diagnostics"].(map[string]interface{}); ok {
        vh.ComponentScores["engine"] = fhm.calculateEngineHealth(engineData)
    }
    
    // Fuel system health
    if fuelData, ok := telemetryData["fuel_system"].(map[string]interface{}); ok {
        vh.ComponentScores["fuel_system"] = fhm.calculateFuelSystemHealth(fuelData)
    }
    
    // Braking system health
    if dynamicsData, ok := telemetryData["vehicle_dynamics"].(map[string]interface{}); ok {
        vh.ComponentScores["braking"] = fhm.calculateBrakingHealth(dynamicsData)
        vh.ComponentScores["tires"] = fhm.calculateTireHealth(dynamicsData)
    }
    
    // Electrical system health
    if electricalData, ok := telemetryData["electrical_system"].(map[string]interface{}); ok {
        vh.ComponentScores["electrical"] = fhm.calculateElectricalHealth(electricalData)
    }
}

func (fhm *FleetHealthMonitor) calculateEngineHealth(engineData map[string]interface{}) float64 {
    engine, ok := engineData["engine"].(map[string]interface{})
    if !ok {
        return 50 // Default score if data unavailable
    }
    
    score := 100.0
    
    // Temperature checks
    if coolantTemp, ok := engine["coolant_temp"].(float64); ok {
        if coolantTemp > 220 {
            score -= 30 // Critical overheating
        } else if coolantTemp > 200 {
            score -= 15 // Running hot
        }
    }
    
    // Oil pressure checks
    if oilPressure, ok := engine["oil_pressure"].(float64); ok {
        if oilPressure < 10 {
            score -= 40 // Critical low pressure
        } else if oilPressure < 20 {
            score -= 20 // Low pressure
        }
    }
    
    // RPM efficiency
    if rpm, ok := engine["rpm"].(float64); ok {
        if rpm > 4000 {
            score -= 10 // High RPM operation
        }
    }
    
    // Load efficiency
    if load, ok := engine["load_percentage"].(float64); ok {
        if load > 90 {
            score -= 15 // High load operation
        }
    }
    
    // DTC codes
    if statusCodes, ok := engineData["status_codes"].(map[string]interface{}); ok {
        if dtcCount, ok := statusCodes["dtc_count"].(float64); ok && dtcCount > 0 {
            score -= dtcCount * 5 // 5 points per diagnostic code
        }
    }
    
    return math.Max(0, math.Min(100, score))
}

func (fhm *FleetHealthMonitor) calculateFuelSystemHealth(fuelData map[string]interface{}) float64 {
    score := 100.0
    
    // Fuel system health indicators
    if systemHealth, ok := fuelData["system_health"].(map[string]interface{}); ok {
        if fuelPressure, ok := systemHealth["fuel_pressure"].(float64); ok {
            if fuelPressure < 40 {
                score -= 20 // Low fuel pressure
            } else if fuelPressure > 80 {
                score -= 10 // High fuel pressure
            }
        }
        
        if injectorPulse, ok := systemHealth["injector_pulse_width"].(float64); ok {
            if injectorPulse > 6.0 {
                score -= 15 // Long injector pulse (inefficiency)
            }
        }
    }
    
    // Efficiency metrics
    if efficiency, ok := fuelData["efficiency_metrics"].(map[string]interface{}); ok {
        if efficiencyScore, ok := efficiency["efficiency_score"].(float64); ok {
            // Use efficiency score as part of health calculation
            score = (score + efficiencyScore) / 2
        }
    }
    
    return math.Max(0, math.Min(100, score))
}

func (fhm *FleetHealthMonitor) calculateBrakingHealth(dynamicsData map[string]interface{}) float64 {
    braking, ok := dynamicsData["braking"].(map[string]interface{})
    if !ok {
        return 50
    }
    
    score := 100.0
    
    // Brake wear
    if frontWear, ok := braking["brake_wear_front"].(float64); ok {
        if frontWear > 90 {
            score -= 30 // Critical wear
        } else if frontWear > 80 {
            score -= 15 // High wear
        }
    }
    
    if rearWear, ok := braking["brake_wear_rear"].(float64); ok {
        if rearWear > 90 {
            score -= 30 // Critical wear
        } else if rearWear > 80 {
            score -= 15 // High wear
        }
    }
    
    // Brake temperature
    if frontTemp, ok := braking["brake_temp_front"].(float64); ok {
        if frontTemp > 300 {
            score -= 20 // Overheating
        } else if frontTemp > 250 {
            score -= 10 // Running hot
        }
    }
    
    return math.Max(0, math.Min(100, score))
}

func (fhm *FleetHealthMonitor) calculateOverallHealth(componentScores map[string]float64) float64 {
    if len(componentScores) == 0 {
        return 0
    }
    
    // Weighted average of component scores
    weights := map[string]float64{
        "engine":      0.3,
        "braking":     0.25,
        "electrical":  0.2,
        "fuel_system": 0.15,
        "tires":       0.1,
    }
    
    totalWeight := 0.0
    weightedSum := 0.0
    
    for component, score := range componentScores {
        weight := weights[component]
        if weight == 0 {
            weight = 0.1 // Default weight for unknown components
        }
        
        totalWeight += weight
        weightedSum += score * weight
    }
    
    if totalWeight == 0 {
        return 0
    }
    
    return weightedSum / totalWeight
}

func (fhm *FleetHealthMonitor) GetFleetHealthSummary() FleetHealthSummary {
    fhm.mutex.RLock()
    defer fhm.mutex.RUnlock()
    
    totalVehicles := len(fhm.vehicles)
    if totalVehicles == 0 {
        return FleetHealthSummary{}
    }
    
    var totalScore float64
    healthDistribution := make(map[string]int)
    activeAlertCount := 0
    maintenanceDueCount := 0
    
    for _, vh := range fhm.vehicles {
        totalScore += vh.OverallScore
        
        // Categorize health
        category := fhm.categorizeHealth(vh.OverallScore)
        healthDistribution[category]++
        
        // Count alerts and maintenance
        activeAlertCount += len(vh.ActiveAlerts)
        maintenanceDueCount += vh.MaintenanceStatus.DueCount + vh.MaintenanceStatus.OverdueCount
    }
    
    return FleetHealthSummary{
        TotalVehicles:       totalVehicles,
        AverageHealthScore:  totalScore / float64(totalVehicles),
        HealthDistribution:  healthDistribution,
        ActiveAlertsCount:   activeAlertCount,
        MaintenanceDueCount: maintenanceDueCount,
        LastUpdated:        time.Now(),
    }
}

type FleetHealthSummary struct {
    TotalVehicles       int            `json:"total_vehicles"`
    AverageHealthScore  float64        `json:"average_health_score"`
    HealthDistribution  map[string]int `json:"health_distribution"`
    ActiveAlertsCount   int            `json:"active_alerts_count"`
    MaintenanceDueCount int            `json:"maintenance_due_count"`
    LastUpdated         time.Time      `json:"last_updated"`
}

func (fhm *FleetHealthMonitor) categorizeHealth(score float64) string {
    switch {
    case score >= fhm.healthThresholds.Good:
        return "excellent"
    case score >= fhm.healthThresholds.Fair:
        return "good"
    case score >= fhm.healthThresholds.Poor:
        return "fair"
    case score >= fhm.healthThresholds.Critical:
        return "poor"
    default:
        return "critical"
    }
}