Python SDK

The official Python SDK for Vehk Axle — v2.0 (vehk-axle): telemetry, ML predictions, GPS tracking, and device APIs over /api/external/*.

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}%")

# GPS track (device_id = registered device ID or IMEI)
client.tracking.send(
    device_id="862061048760001",
    latitude=12.9716,
    longitude=77.5946,
    speed=45.2,
    ignition=True,
)

Client configuration

from vehk_axle import VehkClient

client = VehkClient(
    api_key="vehk_your_api_key",
    # Default production host matches SDK wheel; override for self-hosted:
    base_url="https://vehk-api.bravedune-34ccdec3.centralindia.azurecontainerapps.io",
    timeout=60,                       # Request timeout in seconds
    retries=3,                        # Number of retry attempts
    retry_delay=1.0                   # Delay between retries
)

Authentication uses header X-API-Key. Ensure the key includes scopes for each operation (e.g. telemetry:write, tracking:write, predict, vehicles:read, devices:read). Register devices in the dashboard before calling tracking; see Device Configurator.

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")

Tracking API

GPS points go to POST /api/external/track (and batch: /api/external/track/batch). The device_id must match a device already registered in Axle (ID or IMEI).

Send one track point

result = client.tracking.send(
    device_id="862061048760001",
    latitude=12.9716,
    longitude=77.5946,
    speed=45.2,
    heading=90.0,
    timestamp="2026-03-26T10:00:00Z",
    ignition=True,
    battery_voltage=12.6,
    satellites=8,
    event="periodic",  # e.g. sos, tamper, overspeed, geofence
)

Optional extra accepts a dict for protocol-specific fields:

client.tracking.send(
    device_id="dev-1",
    latitude=13.0,
    longitude=77.6,
    extra={"raw": "..."},
)

Batch track (up to 100)

client.tracking.send_batch([
    {"device_id": "dev-1", "latitude": 12.97, "longitude": 77.59, "speed": 45},
    {"device_id": "dev-2", "latitude": 13.01, "longitude": 77.61, "speed": 30},
])

Latest locations

all_locs = client.tracking.locations()  # GET .../devices/locations
one = client.tracking.location("862061048760001")  # GET .../devices/{id}/location

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}")

Devices API

List tenant devices (GET /api/external/devices). Optional status filter (e.g. active, pending).

devices = client.devices.list()
pending = client.devices.list(status="pending")

for d in devices:
    print(d.get("id"), d.get("name"), d.get("status"))

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}")
    if e.action_required:
        print(f"   Action: {e.action_required}")  # e.g. subscribe, upgrade plan

Common statuses

401 — invalid or missing API key (VehkAuthenticationError).
403 — missing scope, subscription required, or quota exceeded (VehkAPIError; check action_required and message).
429 — rate limit (VehkRateLimitError; retry_after when present).

Production APIs enforce billing: active/trialing subscription expected for External API. Predictions may return quota errors when token limits are reached.

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

TrackingData / DeviceLocation

Import structured helpers from vehk_axle:

Name	Role
`TrackingData`	Dataclass for building track payloads (`to_dict()` for batch items)
`DeviceLocation`	Shape of latest location rows (when you normalize API responses yourself)

The high-level API is client.tracking.send, send_batch, locations, location.

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()

Installation​

Quick Start​

Client configuration​

Telemetry API​

Send Single Telemetry​

Send Batch Telemetry​

Tracking API​

Send one track point​

Batch track (up to 100)​

Latest locations​

Predictions API​

Get Prediction​

List Recent Predictions​

Vehicles API​

List All Vehicles​

Devices API​

Error handling​

Data models​

Prediction​

Vehicle​

TrackingData / DeviceLocation​

Context manager​

Health check​

Supported sensor fields​

Full example​

Links​