equipment-telematics

name	equipment-telematics
description	Integrate and analyze telematics data from heavy construction equipment. Track location, utilization, fuel consumption, maintenance needs, and operator behavior.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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

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 = 
    speed:  = 
    heading:  = 
    timestamp: datetime = field(default_factory=datetime.now)


 :
    equipment_id: 
    timestamp: datetime
    location: GPSLocation
    engine_hours: 
    fuel_level:   
    fuel_rate:    
    engine_rpm: 
    hydraulic_temp: 
    coolant_temp: 
    operating_status: OperatingStatus
    load_percentage:  = 
    operator_id:  = 


 :
    code: 
    description: 
    severity: FaultSeverity
    timestamp: datetime
    equipment_id: 
    resolved:  = 


 :
    : 
    name: 
    equipment_type: EquipmentType
    make: 
    model: 
    year: 
    serial_number: 
    hourly_rate:  = 
    fuel_capacity:  =   
    current_hours:  = 
    next_service_hours:  = 
    assigned_site:  = 
    assigned_operator:  = 


 :
    : 
    name: 
    center_lat: 
    center_lon: 
    radius_meters: 
    allowed_equipment: [] = field(default_factory=)


 :
    equipment_id: 
    period_start: datetime
    period_end: datetime
    total_hours: 
    operating_hours: 
    idle_hours: 
    off_hours: 
    utilization_pct: 
    idle_pct: 
    fuel_consumed: 
    fuel_efficiency:   
    cycles: 

 :
    

    
    SERVICE_INTERVALS = {
        EquipmentType.EXCAVATOR: ,
        EquipmentType.CRANE: ,
        EquipmentType.LOADER: ,
        EquipmentType.BULLDOZER: ,
        EquipmentType.DUMP_TRUCK: ,
    }

    
    TYPICAL_FUEL_RATES = {
        EquipmentType.EXCAVATOR: ,
        EquipmentType.CRANE: ,
        EquipmentType.LOADER: ,
        EquipmentType.BULLDOZER: ,
        EquipmentType.DUMP_TRUCK: ,
    }

     ():
        .fleet_name = fleet_name
        .equipment: [, Equipment] = {}
        .readings: [TelematicsReading] = []
        .faults: [FaultCode] = []
        .geofences: [, Geofence] = {}

     () -> Equipment:
        
        equipment = Equipment(
            =,
            name=name,
            equipment_type=equipment_type,
            make=make,
            model=model,
            year=year,
            serial_number=serial_number,
            hourly_rate=hourly_rate,
            fuel_capacity=fuel_capacity
        )
        .equipment[] = equipment
         equipment

     () -> Geofence:
        
        geofence = Geofence(
            =,
            name=name,
            center_lat=center_lat,
            center_lon=center_lon,
            radius_meters=radius_meters,
            allowed_equipment=allowed_equipment  []
        )
        .geofences[] = geofence
         geofence

     () -> TelematicsReading:
        
         equipment_id   .equipment:
             ValueError()

        
         engine_rpm == :
            status = OperatingStatus.OFF
         engine_rpm <   load_percentage < :
            status = OperatingStatus.IDLE
        :
            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
        )

        .readings.append(reading)

        
        equip = .equipment[equipment_id]
        equip.current_hours = engine_hours

        
        ._check_diagnostics(equipment_id, reading)
        ._check_geofence(equipment_id, location)

         reading

     ():
        
        equip = .equipment[equipment_id]

        
         reading.hydraulic_temp > :
            ._add_fault(equipment_id, ,
                          , FaultSeverity.WARNING)

         reading.coolant_temp > :
            ._add_fault(equipment_id, ,
                          , FaultSeverity.CRITICAL)

        
         reading.fuel_level < :
            ._add_fault(equipment_id, ,
                          , FaultSeverity.WARNING)

        
        service_interval = .SERVICE_INTERVALS.get(equip.equipment_type, )
        hours_to_service = equip.next_service_hours - reading.engine_hours

         hours_to_service < :
            ._add_fault(equipment_id, ,
                          , FaultSeverity.WARNING)
         hours_to_service < :
            ._add_fault(equipment_id, ,
                          , FaultSeverity.INFO)

     ():
        
         geofence  .geofences.values():
            
            distance = ._haversine_distance(
                location.latitude, location.longitude,
                geofence.center_lat, geofence.center_lon
            )

             distance > geofence.radius_meters:
                 ( geofence.allowed_equipment 
                    equipment_id  geofence.allowed_equipment):
                    ._add_fault(equipment_id, ,
                                  ,
                                  FaultSeverity.WARNING)

     () -> :
        
        R =   

        phi1 = math.radians(lat1)
        phi2 = math.radians(lat2)
        delta_phi = math.radians(lat2 - lat1)
        delta_lambda = math.radians(lon2 - lon1)

        a = (math.sin(delta_phi/)** +
             math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda/)**)
        c =  * math.atan2(math.sqrt(a), math.sqrt(-a))

         R * c

     ():
        
        
        existing = [f  f  .faults
                    f.equipment_id == equipment_id
                    f.code == code
                     f.resolved]
         existing:
            

        fault = FaultCode(
            code=code,
            description=description,
            severity=severity,
            timestamp=datetime.now(),
            equipment_id=equipment_id
        )
        .faults.append(fault)

     () -> :
        
         equipment_id   .equipment:
             ValueError()

        equip = .equipment[equipment_id]

        
        readings = [r  r  .readings  r.equipment_id == equipment_id]
          readings:
             {: equip, : }

        latest = (readings, key= r: r.timestamp)

        
        active_faults = [f  f  .faults
                         f.equipment_id == equipment_id   f.resolved]

         {
            : equip.,
            : equip.name,
            : equip.equipment_type.value,
            : latest.operating_status.value,
            : {
                : latest.location.latitude,
                : latest.location.longitude,
                : latest.location.speed
            },
            : latest.engine_hours,
            : latest.fuel_level,
            : latest.fuel_rate,
            : {
                : latest.hydraulic_temp,
                : latest.coolant_temp
            },
            : latest.operator_id,
            : (active_faults),
            : latest.timestamp
        }

     () -> UtilizationReport:
        
        readings = [r  r  .readings
                    r.equipment_id == equipment_id
                    start_date <= r.timestamp <= end_date]

          readings:
             

        readings.sort(key= r: r.timestamp)

        total_hours = (end_date - start_date).total_seconds() / 
        operating_hours = 
        idle_hours = 
        fuel_consumed = 

        
         i  (, (readings)):
            prev = readings[i-]
            curr = readings[i]

            interval_hours = (curr.timestamp - prev.timestamp).total_seconds() / 

             prev.operating_status == OperatingStatus.OPERATING:
                operating_hours += interval_hours
                fuel_consumed += prev.fuel_rate * interval_hours
             prev.operating_status == OperatingStatus.IDLE:
                idle_hours += interval_hours
                fuel_consumed += prev.fuel_rate * interval_hours *   

        off_hours = total_hours - operating_hours - idle_hours
        utilization_pct = (operating_hours / total_hours * )  total_hours >   
        idle_pct = (idle_hours / (operating_hours + idle_hours) * )  (operating_hours + idle_hours) >   
        fuel_efficiency = (fuel_consumed / operating_hours)  operating_hours >   

         UtilizationReport(
            equipment_id=equipment_id,
            period_start=start_date,
            period_end=end_date,
            total_hours=total_hours,
            operating_hours=operating_hours,
            idle_hours=idle_hours,
            off_hours=off_hours,
            utilization_pct=utilization_pct,
            idle_pct=idle_pct,
            fuel_consumed=fuel_consumed,
            fuel_efficiency=fuel_efficiency,
            cycles=  
        )

     () -> :
        
        summary = {
            : (.equipment),
            : {},
            : {},
            : ,
            : []
        }

         equip  .equipment.values():
            
            eq_type = equip.equipment_type.value
            summary[][eq_type] = summary[].get(eq_type, ) + 

            
            :
                status = .get_current_status(equip.)
                op_status = status.get(, )
                summary[][op_status] = summary[].get(op_status, ) + 

                
                service_interval = .SERVICE_INTERVALS.get(equip.equipment_type, )
                hours_to_service = equip.next_service_hours - equip.current_hours
                 hours_to_service < :
                    summary[].append({
                        : equip.name,
                        : hours_to_service
                    })
             Exception:
                summary[][] = summary[].get(, ) + 

        
        summary[] = ([f  f  .faults   f.resolved])

         summary

     () -> :
        
         equipment_id   .equipment:
             ValueError()

        equip = .equipment[equipment_id]

        
        week_ago = datetime.now() - timedelta(days=)
        recent_readings = [r  r  .readings
                          r.equipment_id == equipment_id
                          r.timestamp > week_ago]

         (recent_readings) < :
             {: }

        hours_start = (r.engine_hours  r  recent_readings)
        hours_end = (r.engine_hours  r  recent_readings)
        days = ((r.timestamp  r  recent_readings) -
                (r.timestamp  r  recent_readings)).days  

        daily_hours = (hours_end - hours_start) / days

        
        service_interval = .SERVICE_INTERVALS.get(equip.equipment_type, )
        hours_to_service = equip.next_service_hours - equip.current_hours

         daily_hours > :
            days_to_service = hours_to_service / daily_hours
            service_date = datetime.now() + timedelta(days=days_to_service)
        :
            service_date = 

         {
            : equipment_id,
            : equip.current_hours,
            : equip.next_service_hours,
            : hours_to_service,
            : daily_hours,
            : service_date,
            : ,
            : 
        }

     () -> :
        
        summary = .get_fleet_summary()

        lines = [
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            
        ]

         status, count  summary[].items():
            lines.append()

        
        lines.extend([
            ,
            ,
            ,
            ,
            
        ])

         equip  .equipment.values():
            :
                status = .get_current_status(equip.)
                status_icon =   status[] ==     status[] ==   
                lines.append(
                    
                    
                    
                )
             Exception:
                lines.append(
                    
                )

        
         summary[]:
            lines.extend([
                ,
                ,
                ,
                ,
                
            ])
             svc  summary[]:
                lines.append()

        
        active_faults = [f  f  .faults   f.resolved]
         active_faults:
            lines.extend([
                ,
                ,
                ,
                ,
                
            ])
             fault  active_faults[:]:
                sev_icon =   fault.severity == FaultSeverity.CRITICAL  
                equip = .equipment.get(fault.equipment_id)
                lines.append(
                    
                    
                )

         .join(lines)

Quick Start

from datetime import datetime, timedelta

# Initialize telematics system
telematics = EquipmentTelematics("Site A Fleet")

# Register equipment
telematics.register_equipment(
    "EX-001", "Excavator #1", EquipmentType.EXCAVATOR,
    make="Caterpillar", model="320", year=2022,
    serial_number="CAT320X12345",
    hourly_rate=150, fuel_capacity=400
)

telematics.register_equipment(
    "CR-001", "Tower Crane #1", EquipmentType.CRANE,
    make="Liebherr", model="200EC-H", year=2021,
    serial_number="LH200EC54321",
    hourly_rate=200, fuel_capacity=300
)

# Add geofence for site boundary
telematics.add_geofence(
    "SITE-A", "Site A Boundary",
    center_lat=40.7128, center_lon=-74.0060,
    radius_meters=500
)

# Ingest telematics reading
location = GPSLocation(
    latitude=40.7128, longitude=-74.0059,
    speed=5.0, timestamp=datetime.now()
)

telematics.ingest_reading(
    "EX-001", location,
    engine_hours=1250.5,
    fuel_level=65.0,
    fuel_rate=18.5,
    engine_rpm=1800,
    hydraulic_temp=75.0,
    coolant_temp=85.0,
    load_percentage=75,
    operator_id="OP-101"
)

# Get current status
status = telematics.get_current_status("EX-001")
()
()
()


util = telematics.calculate_utilization(
    ,
    datetime.now() - timedelta(days=),
    datetime.now()
)
 util:
    ()
    ()


maintenance = telematics.predict_maintenance()
()


summary = telematics.get_fleet_summary()
()


(telematics.generate_report())

Requirements

pip install (no external dependencies)