Trip Export Report

Export historical trip data for multiple vehicles to a CSV file.

This recipe calls the legacy v1/fleet/trips endpoint for each vehicle in a list and writes the trip data to Trip_Export.csv.

import csv
import datetime
import json
import os
from datetime import datetime as DT

import requests

TRIPS_URL = "https://api.samsara.com/v1/fleet/trips"
API_TOKEN = os.environ["SAMSARA_API_TOKEN"]
CSV_NAME = "Trip_Export.csv"

trip_start_timestamp = "2022-10-24T18:53:48.231Z"
trip_end_timestamp = "2022-10-25T18:53:48.231Z"
vehicle_ids = []


def fetch_trip_data(vehicle_id, trip_start_ms, trip_end_ms):
    """
    Fetch trip data for one vehicle over a time range.
    """
    api_header = {
        "accept": "application/json",
        "authorization": "Bearer " + API_TOKEN,
    }
    querystring = {
        "vehicleId": str(vehicle_id),
        "startMs": str(trip_start_ms),
        "endMs": str(trip_end_ms),
    }

    response = requests.get(TRIPS_URL, headers=api_header, params=querystring)

    if response.status_code == 200:
        return json.loads(response.text)

    print(response.text)
    return None


def compute_trips(trip_data_list):
    """
    Generate a CSV file for trip data.
    """
    with open(CSV_NAME, "w", newline="") as csv_file:
        column_names = [
            "vehicleId",
            "startTimestamp",
            "startMs",
            "endTimestamp",
            "endMs",
            "startLocation",
            "endLocation",
            "startLatitude",
            "startLongitude",
            "endLatitude",
            "endLongitude",
            "distanceMeters",
            "fuelConsumedMl",
            "tollMeters",
            "driverId",
            "codriverIds",
            "startOdometer",
            "endOdometer",
            "assetIds",
        ]
        csv_dict_writer = csv.DictWriter(csv_file, fieldnames=column_names)
        csv_dict_writer.writeheader()

        for trip_data in trip_data_list:
            for trip in trip_data["trips"]:
                row = {}
                row["vehicleId"] = trip_data["vehicleId"]
                row["startTimestamp"] = DT.fromtimestamp(trip["startMs"] // 1000)
                row["startMs"] = trip["startMs"]

                if trip["endMs"] != 9223372036854775807:
                    row["endTimestamp"] = DT.fromtimestamp(trip["endMs"] // 1000)
                else:
                    row["endTimestamp"] = None

                row["endMs"] = trip["endMs"]
                row["startLocation"] = trip["startLocation"]
                row["endLocation"] = trip["endLocation"]
                row["startLatitude"] = trip["startCoordinates"]["latitude"]
                row["startLongitude"] = trip["startCoordinates"]["longitude"]
                row["endLatitude"] = trip["endCoordinates"]["latitude"]
                row["endLongitude"] = trip["endCoordinates"]["longitude"]
                row["distanceMeters"] = trip["distanceMeters"]
                row["fuelConsumedMl"] = trip["fuelConsumedMl"]
                row["tollMeters"] = trip["tollMeters"]
                row["driverId"] = trip["driverId"]
                row["codriverIds"] = trip["codriverIds"]
                row["startOdometer"] = trip["startOdometer"]
                row["endOdometer"] = trip["endOdometer"]
                row["assetIds"] = trip["assetIds"]

                csv_dict_writer.writerow(row)


def convert_to_ms(input_timestamp):
    """
    Convert an RFC 3339 timestamp to Unix time in milliseconds.
    """
    try:
        date = datetime.datetime.strptime(input_timestamp, "%Y-%m-%dT%H:%M:%S.%fZ")
    except ValueError:
        date = datetime.datetime.strptime(input_timestamp, "%Y-%m-%dT%H:%M:%SZ")

    timestamp = str((date - datetime.datetime(1970, 1, 1)).total_seconds() * 1000)
    return timestamp[:-2]


if __name__ == "__main__":
    consolidated_result = []

    for vehicle in vehicle_ids:
        start_ms = convert_to_ms(trip_start_timestamp)
        end_ms = convert_to_ms(trip_end_timestamp)
        trips_raw_data = fetch_trip_data(vehicle, start_ms, end_ms)

        if trips_raw_data:
            trips_raw_data["vehicleId"] = str(vehicle)
            consolidated_result.append(trips_raw_data)

    if consolidated_result:
        compute_trips(consolidated_result)

How it works

1. Install dependencies

   Run pip install requests to install the requests library.

2. Set up configuration

   Set SAMSARA_API_TOKEN, CSV_NAME, the trip start and end timestamps, and the vehicle_ids list.

3. Convert timestamps

   The trips endpoint accepts startMs and endMs as Unix time in milliseconds. The helper converts RFC 3339 timestamps to that format.

4. Fetch trips for each vehicle

   The script calls v1/fleet/trips once per vehicle ID.

5. Initialize CSV headers

   The output includes vehicle ID, trip start and end data, locations, distance, fuel, toll, driver, odometer, and asset fields.

6. Map each trip

   Each trip in the response is mapped to one row in the CSV file.

7. Write the CSV

   After all vehicles are processed, the script writes Trip_Export.csv.

For newer telematics workflows, see Telematics.