Syncing DVIR Severity Scores to Samsara and Geotab via Webhook

Once a DVIR is scored, the severity number only changes fleet behavior if the telematics platform that dispatchers actually watch reflects it. This page is the concrete implementation for pushing a DvirComplianceEvent into Samsara and Geotab: mapping the internal severity score to each vendor’s REST shape, flagging an out-of-service hold when the score crosses the critical band, and acknowledging the delivery without double-applying it on a retry. Getting the mapping wrong has a regulatory edge — a critical defect that scores 82 but lands in the telematics platform as an ordinary maintenance note means a vehicle meeting the DOT Out-of-Service Criteria stays dispatchable, and the carrier is operating a known-defective unit against 49 CFR § 396.11.

This guide sits under the webhook event delivery layer, which owns the outbox, signing, and retry machinery. Here the focus is narrow: the per-vendor adapter functions that translate one canonical event into a Samsara request and a Geotab request, and the gotchas — authentication, rate limits, and vehicle-id mapping — that separate a working integration from one that silently drops holds. The severity score itself is produced upstream by the severity scoring engine and carried unchanged; this code never recomputes it.

• Python 3.10+ — the adapters use match/case and X | Y unions.
• httpx — async POST with per-request timeouts, reused from the delivery layer.
• The DvirComplianceEvent contract from the platform integration architecture: event_id, vehicle_vin, severity_score (0–100), compliance_action, and audit_hash.
• A Samsara API token (Bearer) scoped to write vehicle tags/attributes, and a Geotab database, username, and session/API key.
• A vehicle-id crosswalk mapping your internal vehicle_vin to each vendor’s opaque vehicle id (Samsara vehicle.id, Geotab Device.id). Never assume the VIN is the vendor key.

The critical banding is fixed and identical to the rest of the pipeline: 0–34 minor, 35–69 major, 70–100 critical. A severity_score of 70 or above, or any defect meeting the DOT Out-of-Service Criteria, is an out-of-service hold and must be flagged as such to the vendor — a mapping no adapter may relax.

Neither vendor keys vehicles by VIN in its write APIs, so resolve the internal vehicle_vin to the vendor’s opaque id before building any payload. Cache the crosswalk and fail the delivery loudly if a VIN is unmapped — a silent skip here drops the hold.

class VehicleNotMapped(Exception):
    """No vendor vehicle id for this VIN. Never silently skip an OOS hold."""


def resolve_vehicle_id(vin: str, crosswalk: dict[str, dict[str, str]], vendor: str) -> str:
    entry = crosswalk.get(vin)
    if entry is None or vendor not in entry:
        raise VehicleNotMapped(f"VIN {vin} has no {vendor} vehicle id")
    return entry[vendor]

Compute the hold decision from the canonical compliance_action (or the score) exactly once, so both adapters agree. The critical floor is 70; a defect meeting the DOT Out-of-Service Criteria arrives already carrying OOS_HOLD.

def is_oos_hold(event) -> bool:
    """Critical band (score >= 70) or an explicit OOS action means an
    out-of-service hold under 49 CFR 396.11. The floor is immovable."""
    return event.compliance_action == "OOS_HOLD" or event.severity_score >= 70

Samsara authenticates with an Authorization: Bearer <token> header. Write the severity as a vehicle attribute and, on a hold, apply an out-of-service tag so it surfaces on the dispatch board. Send the event_id as the Idempotency-Key so a retried POST is deduplicated by Samsara rather than double-applied.

import httpx

SAMSARA_BASE = "https://api.samsara.com"


async def push_samsara(event, token: str, crosswalk) -> httpx.Response:
    vehicle_id = resolve_vehicle_id(event.vehicle_vin, crosswalk, "samsara")
    body = {
        "attributes": [
            {"name": "dvir_severity_score", "value": str(event.severity_score)},
            {"name": "dvir_event_id", "value": str(event.event_id)},
        ],
        "outOfService": is_oos_hold(event),          # OOS hold flag, 49 CFR 396.11
    }
    headers = {
        "Authorization": f"Bearer {token}",
        "Idempotency-Key": str(event.event_id),        # dedup on retry
        "Content-Type": "application/json",
    }
    async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, read=10.0)) as client:
        resp = await client.patch(
            f"{SAMSARA_BASE}/fleet/vehicles/{vehicle_id}",
            json=body, headers=headers,
        )
    resp.raise_for_status()   # 2xx acks; 4xx/5xx raise into the retry path
    return resp

Geotab uses a JSON-RPC style API: you call a method (Set, Add, Get) against an authenticated session. Represent the DVIR defect as a DVIRDefect with RepairStatus of Unrepaired on a hold, referencing the Device resolved from the crosswalk. Geotab does not take a Bearer token — it takes credentials (database, userName, and a session id or password) inside the request body.

GEOTAB_BASE = "https://my.geotab.com/apiv1"


async def push_geotab(event, creds: dict, crosswalk) -> httpx.Response:
    device_id = resolve_vehicle_id(event.vehicle_vin, crosswalk, "geotab")
    repair_status = "Unrepaired" if is_oos_hold(event) else "Repaired"
    rpc = {
        "method": "Set",
        "params": {
            "typeName": "DVIRDefect",
            "entity": {
                "id": str(event.event_id),                 # external ref for dedup
                "device": {"id": device_id},
                "repairStatus": repair_status,
                "comment": f"DVIR severity {event.severity_score}/100",
            },
            "credentials": creds,   # {database, userName, sessionId}
        },
    }
    async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, read=10.0)) as client:
        resp = await client.post(GEOTAB_BASE, json=rpc)
    resp.raise_for_status()
    payload = resp.json()
    if "error" in payload:       # Geotab returns 200 with an error envelope
        raise httpx.HTTPStatusError(
            payload["error"]["message"], request=resp.request, response=resp,
        )
    return resp

Deliver to both vendors independently so one vendor’s outage never blocks the other, and record each acknowledgment against the event_id in the delivery log. Because the outbox retries the whole event, an adapter that already succeeded must be safe to call again — which the Idempotency-Key and the event_id external reference guarantee.

import asyncio


async def sync_all(event, samsara_token, geotab_creds, crosswalk) -> dict:
    results = await asyncio.gather(
        push_samsara(event, samsara_token, crosswalk),
        push_geotab(event, geotab_creds, crosswalk),
        return_exceptions=True,   # one vendor failing must not cancel the other
    )
    return {
        "samsara": _summarize(results[0]),
        "geotab": _summarize(results[1]),
    }

Prove three things: the threshold maps correctly at the boundary, a retry does not double-apply, and an unmapped VIN fails loudly rather than silently.

import pytest


@pytest.mark.parametrize("score,expect_hold", [
    (34, False),   # minor
    (69, False),   # major, not OOS
    (70, True),    # critical floor — OOS hold
    (95, True),    # critical
])
def test_oos_threshold(score, expect_hold, make_event):
    event = make_event(severity_score=score, compliance_action="REPAIR_QUEUE")
    assert is_oos_hold(event) is (expect_hold or score >= 70)


def test_explicit_oos_action_forces_hold(make_event):
    # A DOT_OOS_CRITERIA defect arrives as OOS_HOLD regardless of raw score.
    event = make_event(severity_score=40, compliance_action="OOS_HOLD")
    assert is_oos_hold(event) is True


def test_unmapped_vin_raises(make_event):
    with pytest.raises(VehicleNotMapped):
        resolve_vehicle_id("1FUJGLDR9CLBP8834", {}, "samsara")


@pytest.mark.asyncio
async def test_retry_is_idempotent(make_event, respx_mock):
    # Same event_id posted twice must carry the same Idempotency-Key,
    # so the vendor collapses the duplicate to one effect.
    event = make_event(severity_score=82, compliance_action="OOS_HOLD")
    route = respx_mock.patch(url__regex=r".*/fleet/vehicles/.*")
    route.respond(200, json={"data": {}})
    await push_samsara(event, "tok", {"...VIN...": {"samsara": "v1"}})
    await push_samsara(event, "tok", {"...VIN...": {"samsara": "v1"}})
    keys = {c.request.headers["Idempotency-Key"] for c in route.calls}
    assert keys == {str(event.event_id)}   # identical key on both attempts

• Assuming the VIN is the vendor key. Both Samsara and Geotab key vehicles by an opaque internal id, not the VIN. Resolve through the crosswalk and raise VehicleNotMapped on a miss — never let an unmapped VIN silently drop an out-of-service hold.
• Mismatched authentication. Samsara takes an Authorization: Bearer token; Geotab takes database credentials in the request body and returns HTTP 200 even on logical errors, wrapping the failure in an error envelope. Check the body, not just the status code, or a failed Geotab write looks like a success and the retry never fires.
• Rate limits differ per vendor. Samsara and Geotab enforce different request ceilings and both can answer 429. Honor Retry-After when present and let the delivery layer’s exponential backoff absorb the rest; do not tighten a per-vendor loop that hammers a throttled endpoint.
• Double-sending on retry. Because the outbox retries the whole event at-least-once, an adapter may be invoked again after it already succeeded. The Samsara Idempotency-Key and the Geotab event_id external reference make the repeat a no-op; the receiver-side dedup that guarantees this is detailed in idempotent webhook retry and signature verification.

• Webhook Event Delivery for DVIR Pipelines — the outbox, signing, and retry layer this adapter runs on.
• Idempotent Webhook Retry and Signature Verification — the dedup that makes double-sending on retry safe.
• Samsara, Geotab and Motive Platform Adapters — the vendor-neutral adapter interface this specializes.
• Severity Scoring Algorithms for DVIR Defects — where the score these adapters carry is computed.

This page sits under Webhook Event Delivery for DVIR Pipelines; for the full outbound architecture, see Telematics, ELD & Maintenance Platform Integration.