Python SDK
The official Python SDK for Vehk Axle Fleet Management Platform.
Installation
pip install vehk-axle
Requirements: Python 3.8+
Quick Start
from vehk_axle import VehkClient
# Initialize client
client = VehkClient(api_key="vehk_your_api_key")
# Send telemetry
client.telemetry.send(
vehicle_id="VH-001",
sensor_data={"speed": 65, "rpm": 2500}
)
# Get prediction
prediction = client.predict("VH-001", {"speed": 65, "rpm": 2500})
print(f"Health: {prediction.health_score}%")
Client Configuration
from vehk_axle import VehkClient
client = VehkClient(
api_key="vehk_your_api_key",
base_url="https://api.vehk.in", # Custom URL (optional)
timeout=60, # Request timeout in seconds
retries=3, # Number of retry attempts
retry_delay=1.0 # Delay between retries
)
Telemetry API
Send Single Telemetry
response = client.telemetry.send(
vehicle_id="TRUCK-001",
sensor_data={
"speed": 65,
"rpm": 2500,
"coolant_temp": 92,
"fuel_level": 45,
"battery_voltage": 12.6
},
timestamp="2026-01-14T10:00:00Z", # Optional
location={"lat": 12.97, "lng": 77.59} # Optional
)
print(response["telemetry_id"]) # Unique ID for this data point
Send Batch Telemetry
Send up to 100 records at once for better performance:
from vehk_axle import TelemetryData
batch_result = client.telemetry.send_batch([
{"vehicle_id": "TRUCK-001", "sensor_data": {"speed": 65, "rpm": 2500}},
{"vehicle_id": "TRUCK-002", "sensor_data": {"speed": 70, "rpm": 2800}},
{"vehicle_id": "TRUCK-003", "sensor_data": {"speed": 55, "rpm": 2200}},
])
print(f"Processed: {batch_result.records_processed}")
print(f"Time: {batch_result.processing_time_ms}ms")
Predictions API
Get Prediction
prediction = client.predictions.get(
vehicle_id="TRUCK-001",
sensor_data={
"speed": 65,
"rpm": 2500,
"coolant_temp": 92
}
)
# Access prediction data
print(f"Health Score: {prediction.health_score}%")
print(f"Health Status: {prediction.health_status.value}") # healthy/warning/critical
print(f"Risk Level: {prediction.risk_level.value}") # low/medium/high
# Check if critical
if prediction.is_critical:
print("⚠️ Vehicle needs immediate attention!")
# Get maintenance recommendations
for alert in prediction.maintenance_alerts:
print(f"- {alert.type}: {alert.description} (Urgency: {alert.urgency})")
# Get AI recommendations
for rec in prediction.recommendations:
print(f"💡 {rec}")
List Recent Predictions
# Get all recent predictions
predictions = client.predictions.list(limit=50)
# Filter by vehicle
vehicle_predictions = client.predictions.list(vehicle_id="TRUCK-001", limit=10)
for p in predictions:
print(f"{p.vehicle_id}: {p.health_score}%")
Vehicles API
List All Vehicles
vehicles = client.vehicles.list()
for vehicle in vehicles:
print(f"{vehicle.vehicle_id}: {vehicle.make} {vehicle.model}")
print(f" Health Score: {vehicle.health_score}%")
print(f" VIN: {vehicle.vin}")
Error Handling
from vehk_axle import (
VehkClient,
VehkAuthenticationError,
VehkRateLimitError,
VehkConnectionError,
VehkAPIError,
)
try:
prediction = client.predict("VH-001", {"speed": 65})
except VehkAuthenticationError:
print("❌ Invalid API key")
except VehkRateLimitError as e:
print(f"⏳ Rate limited. Retry after {e.retry_after} seconds")
except VehkConnectionError:
print("🔌 Network error - check your connection")
except VehkAPIError as e:
print(f"⚠️ API Error [{e.status_code}]: {e.message}")
Data Models
Prediction
| Property | Type | Description |
|---|---|---|
health_score | float | Overall health score (0-100) |
health_status | HealthStatus | critical, warning, or healthy |
risk_level | RiskLevel | high, medium, low, or none |
components | dict | Component-level health scores |
maintenance_alerts | list | Pending maintenance items |
recommendations | list[str] | AI-generated recommendations |
is_critical | bool | True if immediate attention needed |
needs_maintenance | bool | True if maintenance is pending |
Vehicle
| Property | Type | Description |
|---|---|---|
vehicle_id | str | Unique vehicle identifier |
vin | str | Vehicle Identification Number |
make | str | Manufacturer name |
model | str | Model name |
year | int | Manufacturing year |
health_score | float | Current health score |
Context Manager
# Automatically close the session when done
with VehkClient(api_key="vehk_your_key") as client:
prediction = client.predict("VH-001", {"speed": 65})
print(prediction.health_score)
Health Check
# Check if API is reachable (no auth required)
health = client.health_check()
print(f"API Status: {health['status']}")
Supported Sensor Fields
| Field | Description | Unit |
|---|---|---|
speed | Vehicle speed | km/h |
rpm | Engine RPM | RPM |
coolant_temp | Coolant temperature | °C |
fuel_level | Fuel level | % |
battery_voltage | Battery voltage | V |
oil_pressure | Oil pressure | PSI |
throttle | Throttle position | % |
intake_air_temp | Intake air temperature | °C |
ambient_temp | Ambient temperature | °C |
odometer | Odometer reading | km |
Full Example
from vehk_axle import VehkClient
import time
def monitor_fleet():
client = VehkClient(api_key="vehk_your_key")
# Get all vehicles
vehicles = client.vehicles.list()
print(f"Monitoring {len(vehicles)} vehicles...")
for vehicle in vehicles:
# Simulate sensor reading
sensor_data = {
"speed": 65,
"rpm": 2500,
"coolant_temp": 92,
"fuel_level": 45
}
# Get prediction
prediction = client.predict(vehicle.vehicle_id, sensor_data)
# Alert on critical vehicles
if prediction.is_critical:
print(f"🚨 CRITICAL: {vehicle.vehicle_id}")
print(f" Health: {prediction.health_score}%")
for alert in prediction.maintenance_alerts:
print(f" - {alert.description}")
elif prediction.health_score < 70:
print(f"⚠️ WARNING: {vehicle.vehicle_id} - Health: {prediction.health_score}%")
else:
print(f"✅ OK: {vehicle.vehicle_id} - Health: {prediction.health_score}%")
if __name__ == "__main__":
monitor_fleet()