equipment-telematics

What this skill does

# Equipment Telematics

## Overview

Integrate telematics data from heavy construction equipment (excavators, cranes, loaders, trucks) to monitor utilization, track location, analyze fuel efficiency, predict maintenance needs, and ensure safe operation.

## Telematics Data Flow

```
┌─────────────────────────────────────────────────────────────────┐
│                  EQUIPMENT TELEMATICS                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  EQUIPMENT                  TELEMATICS              ANALYTICS   │
│  ─────────                  ──────────              ─────────   │
│                                                                  │
│  🚜 Excavator  ────┐       📍 Location              📊 Utilization│
│  🏗️ Crane      ────┼──────→ 🔧 Engine Hours ────────→ ⛽ Fuel      │
│  🚛 Truck      ────┤       ⛽ Fuel Level             🔧 Maintenance│
│  🚧 Loader     ────┘       ⚡ Performance            👷 Operator   │
│                                                                  │
│  METRICS TRACKED:                                               │
│  • GPS location and geofencing                                  │
│  • Engine hours and idle time                                   │
│  • Fuel consumption rate                                        │
│  • Load cycles and productivity                                 │
│  • Fault codes and diagnostics                                  │
│  • Operator behavior and safety                                 │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
```

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import statistics
import math

class EquipmentType(Enum):
    EXCAVATOR = "excavator"
    CRANE = "crane"
    LOADER = "loader"
    BULLDOZER = "bulldozer"
    DUMP_TRUCK = "dump_truck"
    CONCRETE_MIXER = "concrete_mixer"
    FORKLIFT = "forklift"
    COMPACTOR = "compactor"
    GRADER = "grader"
    TELEHANDLER = "telehandler"

class OperatingStatus(Enum):
    OPERATING = "operating"
    IDLE = "idle"
    OFF = "off"
    MAINTENANCE = "maintenance"
    FAULT = "fault"

class FaultSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    SHUTDOWN = "shutdown"

@dataclass
class GPSLocation:
    latitude: float
    longitude: float
    altitude: float = 0.0
    speed: float = 0.0
    heading: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class TelematicsReading:
    equipment_id: str
    timestamp: datetime
    location: GPSLocation
    engine_hours: float
    fuel_level: float  # Percentage
    fuel_rate: float   # L/hr
    engine_rpm: int
    hydraulic_temp: float
    coolant_temp: float
    operating_status: OperatingStatus
    load_percentage: float = 0.0
    operator_id: str = ""

@dataclass
class FaultCode:
    code: str
    description: str
    severity: FaultSeverity
    timestamp: datetime
    equipment_id: str
    resolved: bool = False

@dataclass
class Equipment:
    id: str
    name: str
    equipment_type: EquipmentType
    make: str
    model: str
    year: int
    serial_number: str
    hourly_rate: float = 0.0
    fuel_capacity: float = 0.0  # Liters
    current_hours: float = 0.0
    next_service_hours: float = 0.0
    assigned_site: str = ""
    assigned_operator: str = ""

@dataclass
class Geofence:
    id: str
    name: str
    center_lat: float
    center_lon: float
    radius_meters: float
    allowed_equipment: List[str] = field(default_factory=list)

@dataclass
class UtilizationReport:
    equipment_id: str
    period_start: datetime
    period_end: datetime
    total_hours: float
    operating_hours: float
    idle_hours: float
    off_hours: float
    utilization_pct: float
    idle_pct: float
    fuel_consumed: float
    fuel_efficiency: float  # L/operating hour
    cycles: int

class EquipmentTelematics:
    """Integrate and analyze equipment telematics data."""

    # Maintenance intervals by type (hours)
    SERVICE_INTERVALS = {
        EquipmentType.EXCAVATOR: 250,
        EquipmentType.CRANE: 200,
        EquipmentType.LOADER: 250,
        EquipmentType.BULLDOZER: 250,
        EquipmentType.DUMP_TRUCK: 300,
    }

    # Typical fuel rates (L/hr)
    TYPICAL_FUEL_RATES = {
        EquipmentType.EXCAVATOR: 15,
        EquipmentType.CRANE: 12,
        EquipmentType.LOADER: 18,
        EquipmentType.BULLDOZER: 25,
        EquipmentType.DUMP_TRUCK: 20,
    }

    def __init__(self, fleet_name: str):
        self.fleet_name = fleet_name
        self.equipment: Dict[str, Equipment] = {}
        self.readings: List[TelematicsReading] = []
        self.faults: List[FaultCode] = []
        self.geofences: Dict[str, Geofence] = {}

    def register_equipment(self, id: str, name: str, equipment_type: EquipmentType,
                          make: str, model: str, year: int, serial_number: str,
                          hourly_rate: float = 0, fuel_capacity: float = 0) -> Equipment:
        """Register equipment in fleet."""
        equipment = Equipment(
            id=id,
            name=name,
            equipment_type=equipment_type,
            make=make,
            model=model,
            year=year,
            serial_number=serial_number,
            hourly_rate=hourly_rate,
            fuel_capacity=fuel_capacity
        )
        self.equipment[id] = equipment
        return equipment

    def add_geofence(self, id: str, name: str, center_lat: float,
                    center_lon: float, radius_meters: float,
                    allowed_equipment: List[str] = None) -> Geofence:
        """Add geofence boundary."""
        geofence = Geofence(
            id=id,
            name=name,
            center_lat=center_lat,
            center_lon=center_lon,
            radius_meters=radius_meters,
            allowed_equipment=allowed_equipment or []
        )
        self.geofences[id] = geofence
        return geofence

    def ingest_reading(self, equipment_id: str, location: GPSLocation,
                      engine_hours: float, fuel_level: float, fuel_rate: float,
                      engine_rpm: int, hydraulic_temp: float, coolant_temp: float,
                      load_percentage: float = 0, operator_id: str = "") -> TelematicsReading:
        """Ingest telematics reading from equipment."""
        if equipment_id not in self.equipment:
            raise ValueError(f"Unknown equipment: {equipment_id}")

        # Determine operating status
        if engine_rpm == 0:
            status = OperatingStatus.OFF
        elif engine_rpm < 800 or load_percentage < 10:
            status = OperatingStatus.IDLE
        else:
            status = OperatingStatus.OPERATING

        reading = TelematicsReading(
            equipment_id=equipment_id,
            timestamp=location.timestamp,
            location=location,
            engine_hours=engine_hours,
            fuel_level=fuel_level,
            fuel_rate=fuel_rate,
            engine_rpm=engine_rpm,
            hydraulic_temp=hydraulic_temp,
            coolant_temp=coolant_temp,
            operating_status=status,
            load_percentage=load_percentage,
            operator_id=operator_id
        )

        self.readings.append(reading)

        # Update equipment status
        equip = self.equipment[equipment_id]
        equip.current_hours = engine_hours

        # Check for issues
        self._check_diagnostics(equipment_id, reading)
        self._check_geofence(equipment_id, location)

        return reading

    def _check_diagnostics(self, equipment_id: str, reading: TelematicsReading):
        """Check for diagnostic issues."""
        equip = self.equipment[equipment_id]

        # High tempe