| 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
telematics = EquipmentTelematics("Site A Fleet")
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
)
telematics.add_geofence(
"SITE-A", "Site A Boundary",
center_lat=40.7128, center_lon=-74.0060,
radius_meters=500
)
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"
)
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)
